package cli import ( "bytes" "context" "crypto/rand" "encoding/hex" "errors" "fmt" "io" "path/filepath" "strings" "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/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.TranscriptPath, input.GlossaryPath} for _, source := range selected.Result.Sources() { if source.Path != "" { protected = append(protected, source.Path) } } protected = append(protected, selectedProfilePaths(cfg, selected.Path)...) session, err := diagnostics.PrepareOutput(diagnostics.OutputSelection{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 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 selectedProfilePaths(cfg config.Config, configPath string) []string { dir := cfg.ProfilesDir if !filepath.IsAbs(dir) { dir = filepath.Join(filepath.Dir(configPath), dir) } ids := []string{cfg.LLM.ProposalProfile, cfg.LLM.ValidationProfile} paths := make([]string, 0, len(ids)*2) for _, id := range ids { paths = append(paths, filepath.Join(dir, id+".yaml"), filepath.Join(dir, id+".yml")) } return paths } 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.debugWarnings = append(s.debugWarnings, "debug "+label+": "+llm.RedactError(err, s.cfg)) } } 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.debugWarnings = append(s.debugWarnings, "debug directory: "+llm.RedactError(err, s.cfg)) 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.debugWarnings = append(s.debugWarnings, "debug effective settings: "+llm.RedactError(err, s.cfg)) } 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) }) } if profiles, err := s.runtime.SelectedProfilesYAML(); err == nil { s.debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error { return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles) }) } } 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, }) 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.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.outputPath == "" { 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.debugDir.Path()) } return 1 } fmt.Fprintln(s.stderr, reporting.CompletionLine(snapshot)) for _, path := range s.session.ReportPaths() { fmt.Fprintf(s.stderr, "Report: %s\n", path) } if s.debugDir != nil { fmt.Fprintf(s.stderr, "Debug: %s\n", 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() { for _, warning := range s.debugWarnings { fmt.Fprintf(s.stderr, "audita process: %s\n", warning) } }