443 lines
17 KiB
Go
443 lines
17 KiB
Go
package cli
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/config"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/diagnostics"
|
|
coreio "gitea.maximumdirect.net/eric/audita/internal/core/io"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/outputschema"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/reporting"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/schema"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/llm"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/modules"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/outcomes"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/processreport"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/runner"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput"
|
|
)
|
|
|
|
func runProcess(ctx context.Context, args []string, stdout, stderr io.Writer) int {
|
|
input, help, err := parseConfigInput("process", args, stderr)
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err))
|
|
return 2
|
|
}
|
|
if help {
|
|
writeProcessUsage(stdout)
|
|
return 0
|
|
}
|
|
selected, err := loadConfigInput(input, "process")
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err))
|
|
return 2
|
|
}
|
|
cfg := selected.Result.Config
|
|
reportDir, err := selectedReportDir(cfg, input.OutputPath)
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
|
|
return 2
|
|
}
|
|
if cfg.Diagnostics.Debug && strings.TrimSpace(cfg.Diagnostics.WorkDir) == "" {
|
|
fmt.Fprintln(stderr, "audita process: debug work directory is required")
|
|
return 2
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", err)
|
|
return 1
|
|
}
|
|
runtime, err := llm.NewRuntime(ctx, cfg, selected.Path, nil)
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
|
|
if ctx.Err() != nil || llm.IsAssetError(err) {
|
|
return 1
|
|
}
|
|
return 2
|
|
}
|
|
specs, err := contracts.ResolveModuleRunSpecs(cfg.Pipeline.Modules)
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
|
|
return 2
|
|
}
|
|
runID, err := generateRunID()
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: run ID: %s\n", llm.RedactError(err, cfg))
|
|
return 1
|
|
}
|
|
started := time.Now().UTC()
|
|
state := &processState{cfg: cfg, runtime: runtime, specs: specs, runID: runID, started: started,
|
|
transcriptPath: input.TranscriptPath, glossaryPath: input.GlossaryPath, outputPath: input.OutputPath,
|
|
selected: selected, overlay: input.Overlay, stderr: stderr, stdout: stdout, ctx: ctx}
|
|
protected := []string{input.GlossaryPath}
|
|
for _, source := range selected.Result.Sources() {
|
|
if source.Path != "" {
|
|
protected = append(protected, source.Path)
|
|
}
|
|
}
|
|
protected = append(protected, runtime.ProfilePaths...)
|
|
session, err := diagnostics.PrepareOutput(diagnostics.OutputSelection{InputTranscriptPath: input.TranscriptPath, TranscriptPath: input.OutputPath,
|
|
ReportDir: reportDir, ReportingEnabled: cfg.Reporting.Enabled, ProtectedPaths: protected})
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
|
|
return 2
|
|
}
|
|
state.session = session
|
|
initial := state.snapshot(reporting.StatusInProgress, nil, nil)
|
|
initialBundle, err := state.bundle(initial)
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
|
|
return 1
|
|
}
|
|
initFailure := []processreport.RunFailure{{Phase: "initialization", Reason: outcomes.ReasonInternalFailure}}
|
|
failedInit, err := state.bundle(state.snapshot(reporting.StatusFailed, nil, initFailure))
|
|
if err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
|
|
return 1
|
|
}
|
|
if err := session.Initialize(initialBundle, failedInit); err != nil {
|
|
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
|
|
return 1
|
|
}
|
|
state.startDebug()
|
|
return state.execute()
|
|
}
|
|
|
|
type processState struct {
|
|
cfg config.Config
|
|
runtime *llm.Runtime
|
|
specs []contracts.ModuleRunSpec
|
|
runID string
|
|
started time.Time
|
|
transcriptPath string
|
|
glossaryPath string
|
|
outputPath string
|
|
selected selectedConfig
|
|
overlay string
|
|
ctx context.Context
|
|
stdout io.Writer
|
|
stderr io.Writer
|
|
session *diagnostics.OutputSession
|
|
debugDir *diagnostics.RunDirectory
|
|
debugWarnings []string
|
|
debugMu sync.Mutex
|
|
runOutput *runner.RunOutput
|
|
}
|
|
|
|
func generateRunID() (string, error) {
|
|
var random [12]byte
|
|
if _, err := rand.Read(random[:]); err != nil {
|
|
return "", err
|
|
}
|
|
return "run-" + hex.EncodeToString(random[:]), nil
|
|
}
|
|
|
|
func (s *processState) snapshot(status string, completed *time.Time, failures []processreport.RunFailure) reporting.Snapshot {
|
|
return processreport.Build(processreport.BuildInput{RunID: s.runID, Status: status, StartedAt: s.started,
|
|
CompletedAt: completed, TranscriptPath: s.transcriptPath, GlossaryPath: s.glossaryPath,
|
|
OutputPath: s.outputPath, ModuleSpecs: s.specs, RunOutput: s.runOutput, RunFailures: failures})
|
|
}
|
|
|
|
func (s *processState) bundle(snapshot reporting.Snapshot) (diagnostics.ReportBundle, error) {
|
|
if !s.cfg.Reporting.Enabled {
|
|
return diagnostics.ReportBundle{}, nil
|
|
}
|
|
secrets := llm.ConfiguredSecrets(&s.cfg)
|
|
errorsYAML, err := reporting.YAML(snapshot.ErrorsDocument(), secrets...)
|
|
if err != nil {
|
|
return diagnostics.ReportBundle{}, err
|
|
}
|
|
recoveriesYAML, err := reporting.YAML(snapshot.RecoveriesDocument(), secrets...)
|
|
if err != nil {
|
|
return diagnostics.ReportBundle{}, err
|
|
}
|
|
return diagnostics.ReportBundle{Errors: errorsYAML, Recoveries: recoveriesYAML,
|
|
Markdown: reporting.Markdown(snapshot, secrets...)}, nil
|
|
}
|
|
|
|
func (s *processState) debugWrite(label string, write func(*diagnostics.RunDirectory) error) {
|
|
if s.debugDir == nil {
|
|
return
|
|
}
|
|
if err := write(s.debugDir); err != nil {
|
|
s.reportDiagnosticError(fmt.Errorf("%s: %w", label, err))
|
|
}
|
|
}
|
|
|
|
func (s *processState) startDebug() {
|
|
if !s.cfg.Diagnostics.Debug {
|
|
return
|
|
}
|
|
dir, err := diagnostics.NewRunDirectory(s.cfg.Diagnostics.WorkDir, s.runID, llm.ConfiguredSecrets(&s.cfg)...)
|
|
if err != nil {
|
|
s.reportDiagnosticError(fmt.Errorf("directory: %w", err))
|
|
return
|
|
}
|
|
s.debugDir = dir
|
|
s.debugWrite("invocation", func(d *diagnostics.RunDirectory) error {
|
|
return d.WriteInvocationMetadata(diagnostics.InvocationMetadata{
|
|
Operation: "process", TranscriptPath: s.transcriptPath, GlossaryPath: s.glossaryPath,
|
|
OutputPath: s.outputPath, ConfigPath: s.selected.Path, ConfigSource: s.selected.Source,
|
|
Overlay: s.overlay, ConfigVersion: &s.cfg.Version, TranscriptDescription: s.cfg.Context.Description,
|
|
Modules: append([]string(nil), s.cfg.Pipeline.Modules...), RunID: s.runID, StartedAt: s.started,
|
|
})
|
|
})
|
|
var effective bytes.Buffer
|
|
if err := s.selected.Result.WriteEffective(&effective); err == nil {
|
|
s.debugWrite("effective settings", func(d *diagnostics.RunDirectory) error { return d.WriteEffectiveConfig(effective.Bytes()) })
|
|
} else {
|
|
s.reportDiagnosticError(fmt.Errorf("effective settings: %w", err))
|
|
}
|
|
if origins, err := llm.ExplanationYAML(s.selected.Result); err == nil {
|
|
s.debugWrite("configuration origins", func(d *diagnostics.RunDirectory) error {
|
|
return d.WriteYAMLArtifact(diagnostics.ArtifactConfigOrigins, origins)
|
|
})
|
|
} else {
|
|
s.reportDiagnosticError(fmt.Errorf("configuration origins: %w", err))
|
|
}
|
|
if profiles, err := s.runtime.SelectedProfilesYAML(); err == nil {
|
|
s.debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error {
|
|
return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles)
|
|
})
|
|
} else {
|
|
s.reportDiagnosticError(fmt.Errorf("selected profiles: %w", err))
|
|
}
|
|
}
|
|
|
|
func (s *processState) debugPath() string {
|
|
if s.debugDir == nil {
|
|
return ""
|
|
}
|
|
return s.debugDir.Path()
|
|
}
|
|
|
|
func (s *processState) execute() int {
|
|
transcriptBytes, err := coreio.ReadRequiredFile(s.transcriptPath, "transcript")
|
|
if err != nil {
|
|
return s.fail("input", outcomes.ReasonInternalFailure, err)
|
|
}
|
|
glossaryBytes, err := coreio.ReadRequiredFile(s.glossaryPath, "glossary")
|
|
if err != nil {
|
|
return s.fail("input", outcomes.ReasonInternalFailure, err)
|
|
}
|
|
source, err := schema.ParseSourceTranscriptJSON(transcriptBytes)
|
|
if err != nil {
|
|
return s.fail("input", outcomes.ReasonInputInvalid, err)
|
|
}
|
|
glossary, err := schema.ParseGlossaryYAML(glossaryBytes)
|
|
if err != nil {
|
|
return s.fail("input", outcomes.ReasonInputInvalid, err)
|
|
}
|
|
s.debugWrite("source transcript", func(d *diagnostics.RunDirectory) error { return d.WriteSourceTranscript(source, transcriptBytes) })
|
|
normalizer := normalization.NewNormalizer(normalization.NormalizationConfig{
|
|
MaxSegmentGap: s.cfg.Normalization.MaxSegmentGap, EllipsisGap: s.cfg.Normalization.EllipsisGap,
|
|
MaxSegmentDuration: s.cfg.Normalization.MaxSegmentDuration, MaxSegmentTokens: s.cfg.Normalization.MaxSegmentTokens,
|
|
})
|
|
normalized, normSummary := normalizer.Normalize(sourceToCanonicalTranscript(source))
|
|
s.debugWrite("normalized transcript", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizedTranscript(normalized) })
|
|
s.debugWrite("normalization summary", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizationSummary(normSummary) })
|
|
if err := s.ctx.Err(); err != nil {
|
|
return s.fail("processing", outcomes.ReasonCanceled, err)
|
|
}
|
|
observations := llm.NewObservationCollector()
|
|
proposalClient, err := s.runtime.NewRoleClient("proposal", observations.Hooks())
|
|
if err != nil {
|
|
return s.fail("processing", outcomes.ReasonSetupFailure, err)
|
|
}
|
|
validationClient, err := s.runtime.NewRoleClient("validation", observations.Hooks())
|
|
if err != nil {
|
|
return s.fail("processing", outcomes.ReasonSetupFailure, err)
|
|
}
|
|
factory := modules.NewFactory(modules.Dependencies{Config: &s.cfg, Glossary: glossary, DiagnosticsDir: s.debugPath()})
|
|
result, runErr := runner.New(factory).Run(s.ctx, runner.RunInput{
|
|
Config: &s.cfg, Transcript: normalized, Glossary: glossary, ModuleSpecs: s.specs,
|
|
ProposalLLMClient: proposalClient, ProposalDiagnosticsDir: s.debugPath(),
|
|
ValidationLLMClient: validationClient, ValidationDiagnosticsDir: s.debugPath(),
|
|
CaptureCandidates: s.debugDir != nil,
|
|
ReportDiagnosticError: s.reportDiagnosticError,
|
|
})
|
|
result.Observations = observations.Snapshot()
|
|
result.ObservationsAvailable = true
|
|
result.LLMSummary = llm.Summarize(result.Observations)
|
|
s.runOutput = &result
|
|
s.writeExecutionEvidence()
|
|
if len(result.Sections) > 0 || runErr == nil {
|
|
s.writeChunkSummary(result.Sections)
|
|
}
|
|
if runErr != nil {
|
|
return s.fail("processing", fatalReason(s.ctx, runErr), runErr)
|
|
}
|
|
encoder, err := outputschema.Resolve(s.cfg.Output.Schema)
|
|
if err != nil {
|
|
return s.fail("encoding", outcomes.ReasonSetupFailure, err)
|
|
}
|
|
outputBytes, err := encoder.Encoder(result.FinalTranscript)
|
|
if err != nil {
|
|
return s.fail("encoding", outcomes.ReasonInternalFailure, err)
|
|
}
|
|
return s.finish(outputBytes)
|
|
}
|
|
|
|
func (s *processState) writeChunkSummary(inventory []contracts.SectionMetadata) {
|
|
if s.debugDir == nil {
|
|
return
|
|
}
|
|
sections := make([]chunking.Section, len(inventory))
|
|
for i, metadata := range inventory {
|
|
sections[i] = chunking.Section{Index: metadata.Index, StartSegmentID: metadata.StartSegmentID,
|
|
EndSegmentID: metadata.EndSegmentID, EstimatedTokens: metadata.EstimatedTokens,
|
|
Segments: make([]schema.Segment, metadata.SegmentCount)}
|
|
}
|
|
config := chunking.ChunkingConfig{TargetSections: s.cfg.Chunking.TargetSections,
|
|
MaxSectionTokens: s.cfg.Chunking.MaxSectionTokens, MinSectionTokens: s.cfg.Chunking.MinSectionTokens}
|
|
summary := chunking.ComputeDetailedSummary(sections, config)
|
|
s.debugWrite("chunking summary", func(d *diagnostics.RunDirectory) error { return d.WriteChunkingSummary(&summary) })
|
|
}
|
|
|
|
func (s *processState) writeExecutionEvidence() {
|
|
if s.debugDir == nil || s.runOutput == nil {
|
|
return
|
|
}
|
|
output := s.runOutput
|
|
s.debugWrite("model execution", func(d *diagnostics.RunDirectory) error {
|
|
return d.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(output.Observations))
|
|
})
|
|
s.debugWrite("utilization", func(d *diagnostics.RunDirectory) error {
|
|
return d.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, processreport.BuildUtilization(output, s.cfg.Backends))
|
|
})
|
|
s.debugWrite("correction ledger", func(d *diagnostics.RunDirectory) error {
|
|
return d.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{
|
|
RunDirectoryPath: d.Path(), RunOutput: output}))
|
|
})
|
|
}
|
|
|
|
func fatalReason(ctx context.Context, err error) outcomes.ReasonCode {
|
|
if ctx.Err() != nil || errors.Is(err, context.Canceled) {
|
|
return outcomes.ReasonCanceled
|
|
}
|
|
var failure *structuredoutput.Failure
|
|
if errors.As(err, &failure) && failure.Reason != "" {
|
|
return failure.Reason
|
|
}
|
|
return outcomes.ReasonInternalFailure
|
|
}
|
|
|
|
func (s *processState) fail(phase string, reason outcomes.ReasonCode, cause error) int {
|
|
s.debugWrite("error log", func(d *diagnostics.RunDirectory) error {
|
|
return d.WriteErrorLog(phase + ": " + llm.RedactError(cause, s.cfg))
|
|
})
|
|
failures := []processreport.RunFailure{{Phase: phase, Reason: reason}}
|
|
completed := time.Now().UTC()
|
|
base, err := s.bundle(s.snapshot(reporting.StatusFailed, &completed, failures))
|
|
if err != nil {
|
|
fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(err, s.cfg))
|
|
}
|
|
writeFailures := s.session.Finalize(nil, base, func(extra []diagnostics.FileFailure) diagnostics.ReportBundle {
|
|
all := append(append([]processreport.RunFailure(nil), failures...), mapWriteFailures(extra)...)
|
|
bundle, renderErr := s.bundle(s.snapshot(reporting.StatusFailed, &completed, all))
|
|
if renderErr != nil {
|
|
return base
|
|
}
|
|
return bundle
|
|
})
|
|
fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(cause, s.cfg))
|
|
s.printFailures(writeFailures)
|
|
s.printDebugWarnings()
|
|
if s.debugDir != nil {
|
|
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.safePath(s.debugDir.Path()))
|
|
}
|
|
return 1
|
|
}
|
|
|
|
func (s *processState) finish(outputBytes []byte) int {
|
|
completed := time.Now().UTC()
|
|
snapshot := s.snapshot(reporting.StatusCompleted, &completed, nil)
|
|
complete, err := s.bundle(snapshot)
|
|
if err != nil {
|
|
return s.fail("report_write", outcomes.ReasonInternalFailure, err)
|
|
}
|
|
if s.session.TranscriptPath() == "" {
|
|
if _, err := s.stdout.Write(outputBytes); err != nil {
|
|
return s.fail("transcript_write", outcomes.ReasonInternalFailure, err)
|
|
}
|
|
outputBytes = nil
|
|
}
|
|
writeFailures := s.session.Finalize(outputBytes, complete, func(extra []diagnostics.FileFailure) diagnostics.ReportBundle {
|
|
failed, renderErr := s.bundle(s.snapshot(reporting.StatusFailed, &completed, mapWriteFailures(extra)))
|
|
if renderErr != nil {
|
|
return diagnostics.ReportBundle{}
|
|
}
|
|
return failed
|
|
})
|
|
if len(writeFailures) > 0 {
|
|
s.printFailures(writeFailures)
|
|
s.printDebugWarnings()
|
|
if s.debugDir != nil {
|
|
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.safePath(s.debugDir.Path()))
|
|
}
|
|
return 1
|
|
}
|
|
fmt.Fprintln(s.stderr, reporting.CompletionLine(snapshot))
|
|
for _, path := range s.session.ReportPaths() {
|
|
fmt.Fprintf(s.stderr, "Report: %s\n", s.safePath(path))
|
|
}
|
|
if s.debugDir != nil {
|
|
fmt.Fprintf(s.stderr, "Debug: %s\n", s.safePath(s.debugDir.Path()))
|
|
}
|
|
s.printDebugWarnings()
|
|
return 0
|
|
}
|
|
|
|
func mapWriteFailures(failures []diagnostics.FileFailure) []processreport.RunFailure {
|
|
out := make([]processreport.RunFailure, len(failures))
|
|
for i, failure := range failures {
|
|
phase := "report_write"
|
|
if failure.Kind == "transcript write" {
|
|
phase = "transcript_write"
|
|
}
|
|
out[i] = processreport.RunFailure{Phase: phase, Reason: outcomes.ReasonInternalFailure, OutputPath: failure.Path}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (s *processState) printFailures(failures []diagnostics.FileFailure) {
|
|
for _, failure := range failures {
|
|
fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(failure, s.cfg))
|
|
}
|
|
}
|
|
|
|
func (s *processState) printDebugWarnings() {
|
|
s.debugMu.Lock()
|
|
defer s.debugMu.Unlock()
|
|
for _, warning := range s.debugWarnings {
|
|
fmt.Fprintf(s.stderr, "audita process: %s\n", warning)
|
|
}
|
|
}
|
|
|
|
// safePath sanitizes presentation only; filesystem destinations remain literal.
|
|
func (s *processState) safePath(path string) string {
|
|
return redaction.New(llm.ConfiguredSecrets(&s.cfg)).String(path)
|
|
}
|
|
|
|
func (s *processState) reportDiagnosticError(err error) {
|
|
if err == nil {
|
|
return
|
|
}
|
|
message := "debug " + llm.RedactError(err, s.cfg)
|
|
s.debugMu.Lock()
|
|
defer s.debugMu.Unlock()
|
|
s.debugWarnings = append(s.debugWarnings, message)
|
|
}
|