diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 59ead6f..961fdaf 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -344,6 +344,12 @@ retain already published output and primary failure details. ## Stage 8: Wire processing, common preflight, and reporting modes +**Completed.** Processing now acquires typed inputs, prepares outputs, and +finalizes through CommandKit. The `executePipeline` handoff is intentionally +implemented with the ordered runner work below; until then the application and +unconverted tests do not build. The handler was checked with a controlled runner +boundary; later readiness checks and assembled validation remain assigned below. + Own `internal/cli/process.go` and the new `internal/framework/preflight`. Use the [named preflight boundary](glossary-stages.md#empty-collections-and-audita-preflight--resolved) and the [CommandKit reporting decisions](commandkit.md#debug-reporting-and-artifact-capture--resolved). diff --git a/internal/cli/process.go b/internal/cli/process.go index fcabf3f..0b811bf 100644 --- a/internal/cli/process.go +++ b/internal/cli/process.go @@ -1,357 +1,81 @@ package cli import ( - "bytes" - "context" - "crypto/rand" - "encoding/hex" - "errors" "fmt" - "io" - "strings" "sync" - "time" - "gitea.maximumdirect.net/eric/audita/v3/internal/core/chunking" - "gitea.maximumdirect.net/eric/audita/v3/internal/core/config" - "gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics" - coreio "gitea.maximumdirect.net/eric/audita/v3/internal/core/io" "gitea.maximumdirect.net/eric/audita/v3/internal/core/normalization" "gitea.maximumdirect.net/eric/audita/v3/internal/core/outputschema" - "gitea.maximumdirect.net/eric/audita/v3/internal/core/redaction" "gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting" "gitea.maximumdirect.net/eric/audita/v3/internal/core/schema" - "gitea.maximumdirect.net/eric/audita/v3/internal/framework/contracts" "gitea.maximumdirect.net/eric/audita/v3/internal/framework/llm" - "gitea.maximumdirect.net/eric/audita/v3/internal/framework/modules" - "gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes" - "gitea.maximumdirect.net/eric/audita/v3/internal/framework/processreport" - "gitea.maximumdirect.net/eric/audita/v3/internal/framework/runner" - "gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput" + commandkit "go.maximumdirect.net/commandkit" ) -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 +// processResult carries the runner's transcript and ordinary report facts. +type processResult struct { + Transcript *schema.Transcript + Snapshot reporting.Snapshot } -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...) +func runProcess(inv commandkit.Invocation, input processInput) commandkit.Outcome { + loaded, err := loadProcessInputs(inv.Context, input, inv.ReportingMode()) if err != nil { - return diagnostics.ReportBundle{}, err + if inv.Context.Err() != nil || llm.IsAssetError(err) { + return commandkit.Failure("Could not prepare processing.").WithCause(err) + } + return commandkit.Usage(err.Error()).WithCause(err) } - recoveriesYAML, err := reporting.YAML(snapshot.RecoveriesDocument(), secrets...) + prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output()) if err != nil { - return diagnostics.ReportBundle{}, err + return commandkit.Failure(loaded.Runtime.RedactError(inv.Context, err)).WithCause(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 + defer prepared.Prepared.Output.Close() + var warningsMu sync.Mutex + warn := func(err error) { + if err == nil { + return + } + warningsMu.Lock() + prepared.Prepared.Warnings = append(prepared.Prepared.Warnings, err) + warningsMu.Unlock() } - if err := write(s.debugDir); err != nil { - s.reportDiagnosticError(fmt.Errorf("%s: %w", label, err)) + debug := prepared.Prepared.Debug + if debug != nil { + warn(debug.WriteSourceTranscript(loaded.Source, loaded.RawTranscript)) } -} - -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, - }) + cfg := &loaded.Configuration.Config + normalizer := normalization.NewNormalizer(normalization.NormalizationConfig{ + MaxSegmentGap: cfg.Normalization.MaxSegmentGap, EllipsisGap: cfg.Normalization.EllipsisGap, + MaxSegmentDuration: cfg.Normalization.MaxSegmentDuration, MaxSegmentTokens: cfg.Normalization.MaxSegmentTokens, }) - 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)) + normalized, summary := normalizer.Normalize(sourceToCanonicalTranscript(loaded.Source)) + if debug != nil { + warn(debug.WriteNormalizedTranscript(normalized)) + warn(debug.WriteNormalizationSummary(summary)) } - 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)) + result, runErr := executePipeline(inv.Context, loaded, prepared, normalized, warn) + if runErr != nil { + return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, runErr) } - 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)) + encoder, err := outputschema.Resolve(cfg.Output.Schema) + if err != nil { + return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, err) } + if result.Transcript == nil { + return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, fmt.Errorf("processing returned no transcript")) + } + payload, err := encoder.Encoder(result.Transcript) + if err != nil { + return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, err) + } + return finishProcess(inv, input, loaded, prepared, result.Snapshot, payload, nil) } -func (s *processState) debugPath() string { - if s.debugDir == nil { +// DebugPath returns the retained artifact directory, if its preparation succeeded. +func (p *preparedProcess) DebugPath() string { + if p == nil || p.Prepared == nil || p.Prepared.Debug == 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) + return p.Prepared.Debug.Path() } diff --git a/internal/cli/process_command.go b/internal/cli/process_command.go index d0efa18..34711fb 100644 --- a/internal/cli/process_command.go +++ b/internal/cli/process_command.go @@ -79,7 +79,7 @@ func processCommand() commandkit.Command { in.Overrides = append(in.Overrides, ck.Override{Path: ck.Keys("reporting", "enabled"), Value: v.Value, Flag: "--reporting"}) } })) - return commandkit.NewCommand(commandkit.CommandSpec[processInput]{Name: "process", Summary: "Polish a transcript.", Flags: flags, + return commandkit.NewCommand(commandkit.CommandSpec[processInput]{Name: "process", Summary: "Polish a transcript.", Flags: flags, Run: runProcess, Arguments: []commandkit.Argument[processInput]{commandkit.ValueArgument(commandkit.ArgumentMeta{Name: "transcript", Usage: "Transcript JSON file."}, true, stringValue, func(in *processInput, v commandkit.Supplied[string]) { in.Transcript = v.Value })}, }) } diff --git a/internal/cli/process_command_integration_test.go b/internal/cli/process_command_integration_test.go new file mode 100644 index 0000000..86b33ca --- /dev/null +++ b/internal/cli/process_command_integration_test.go @@ -0,0 +1,58 @@ +package cli + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strings" + "testing" + + commandkit "go.maximumdirect.net/commandkit" +) + +func TestProcessUsesResolvedReportingMode(t *testing.T) { + dir := t.TempDir() + for name, contents := range map[string]string{ + "config.yml": "version: 3\npipeline:\n stages: []\nreporting:\n enabled: false\n", + "transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`, + "blocked": "file", + } { + if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil { + t.Fatal(err) + } + } + app, err := buildApplication() + if err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + name string + flags []string + status int + stdout bool + info bool + warning bool + }{ + {"ordinary", nil, 0, true, false, false}, + {"verbose", []string{"--verbose"}, 0, true, true, false}, + {"quiet stdout", []string{"--quiet"}, 2, false, false, false}, + {"quiet file", []string{"--quiet", "--output", filepath.Join(dir, "quiet.json")}, 0, false, false, false}, + {"verbose after quiet", []string{"--quiet", "--verbose"}, 0, true, true, false}, + {"quiet after verbose", []string{"--verbose", "--quiet", "--output", filepath.Join(dir, "ordered.json")}, 0, false, false, false}, + {"debug disabled", []string{"--debug=false", "--work-dir", filepath.Join(dir, "blocked")}, 0, true, false, false}, + {"debug failure", []string{"--debug", "--work-dir", filepath.Join(dir, "blocked")}, 0, true, true, true}, + {"debug quiet", []string{"--debug", "--quiet", "--output", filepath.Join(dir, "debug-quiet.json"), "--work-dir", filepath.Join(dir, "blocked")}, 0, false, false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + var out, stderr bytes.Buffer + args := []string{"process", "--config", filepath.Join(dir, "config.yml")} + args = append(args, tc.flags...) + args = append(args, filepath.Join(dir, "transcript.json")) + result := app.Execute(context.Background(), args, commandkit.Streams{Out: &out, Err: &stderr}) + if result.Status != tc.status || (out.Len() > 0) != tc.stdout || strings.Contains(stderr.String(), "Completed:") != tc.info || strings.Contains(stderr.String(), "Debug artifact:") != tc.warning { + t.Fatalf("status=%d out=%q err=%q", result.Status, out.String(), stderr.String()) + } + }) + } +} diff --git a/internal/cli/process_inputs.go b/internal/cli/process_inputs.go new file mode 100644 index 0000000..b32959e --- /dev/null +++ b/internal/cli/process_inputs.go @@ -0,0 +1,139 @@ +package cli + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "io" + "path/filepath" + "time" + + "gitea.maximumdirect.net/eric/audita/v3/internal/core/config" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics" + coreio "gitea.maximumdirect.net/eric/audita/v3/internal/core/io" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/schema" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/llm" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/preflight" + "go.maximumdirect.net/ckconfig" + commandkit "go.maximumdirect.net/commandkit" + ck "go.maximumdirect.net/configkit" +) + +type processInputs struct { + Configuration *ck.Result[config.Config] + Runtime *llm.Runtime + Source *schema.SourceTranscript + RawTranscript []byte + ProtectedPaths []string +} + +func loadProcessInputs(ctx context.Context, input processInput, mode commandkit.ReportingMode) (*processInputs, error) { + if mode == commandkit.ReportingModeQuiet && input.Output.Value == "" { + return nil, fmt.Errorf("quiet processing requires an output file or /dev/null") + } + selection := ckconfig.Selection{Config: input.Config, Overlay: input.Overlay, Overrides: input.Overrides} + loaded, err := loadApplication(ctx, selection, struct{}{}, nil) + if err != nil { + return nil, err + } + runtime, err := llm.NewRuntime(ctx, loaded, nil) + if err != nil { + return nil, err + } + raw, err := coreio.ReadRequiredFile(input.Transcript, "transcript") + if err != nil { + return nil, err + } + source, err := schema.ParseSourceTranscriptJSON(raw) + if err != nil { + return nil, err + } + protected := append([]string(nil), runtime.SourcePaths...) + for _, entry := range loaded.Sources() { + if entry.Kind == ck.SourceKindFile && entry.Path != "" { + protected = append(protected, entry.Path) + } + } + return &processInputs{Configuration: loaded, Runtime: runtime, Source: source, RawTranscript: raw, ProtectedPaths: protected}, nil +} + +// preparedProcess groups the output resources and the facts needed for reports. +type preparedProcess struct { + Prepared *preflight.Prepared + RunID string + Started time.Time +} + +func prepareProcess(ctx context.Context, input processInput, loaded *processInputs, stdout io.Writer) (*preparedProcess, error) { + cfg := &loaded.Configuration.Config + reportDir := "" + reportDirSet := false + for _, override := range input.Overrides { + if override.Path.String() == "/reporting/stdout_dir" { + reportDirSet = true + } + } + if cfg.Reporting.Enabled { + reportDir = cfg.Reporting.StdoutDir + if input.Output.Value != "" && !reportDirSet { + reportDir = filepath.Dir(input.Output.Value) + } + } + runID, err := generateRunID() + if err != nil { + return nil, err + } + started := time.Now().UTC() + snapshot := reporting.Snapshot{RunID: runID, Status: reporting.StatusInProgress, StartedAt: started, TranscriptPath: input.Transcript, OutputPath: input.Output.Value} + for _, item := range cfg.Pipeline.Stages { + snapshot.StageNames = append(snapshot.StageNames, item.Module) + } + secrets := loaded.Runtime.ConfiguredSecrets(ctx) + initial, err := renderReportBundle(snapshot, secrets...) + if err != nil { + return nil, err + } + snapshot.Status = reporting.StatusFailed + failed, err := renderReportBundle(snapshot, secrets...) + if err != nil { + return nil, err + } + var debug *preflight.DebugInput + var warnings []error + if input.Debug.Value { + var effective bytes.Buffer + if err := loaded.Configuration.WriteEffective(&effective); err != nil { + warnings = append(warnings, err) + } + debug = &preflight.DebugInput{WorkDir: cfg.Diagnostics.WorkDir, RunID: runID, Secrets: secrets, EffectiveConfig: effective.Bytes(), Metadata: diagnostics.InvocationMetadata{Operation: "process", TranscriptPath: input.Transcript, OutputPath: input.Output.Value, Overlay: input.Overlay.Value, ConfigVersion: &cfg.Version, RunID: runID, StartedAt: started}} + } + prepared, err := preflight.Prepare(ctx, preflight.Input{Config: cfg, Transcript: loaded.Source, Output: diagnostics.OutputSelection{InputTranscriptPath: input.Transcript, TranscriptPath: input.Output.Value, ReportDir: reportDir, ReportingEnabled: cfg.Reporting.Enabled, ProtectedPaths: loaded.ProtectedPaths, Stdout: stdout}, InitialReports: initial, FailedReports: failed, Debug: debug}) + if err != nil { + return nil, err + } + prepared.Warnings = append(warnings, prepared.Warnings...) + return &preparedProcess{Prepared: prepared, RunID: runID, Started: started}, nil +} + +func generateRunID() (string, error) { + var random [12]byte + if _, err := rand.Read(random[:]); err != nil { + return "", err + } + return "run-" + hex.EncodeToString(random[:]), nil +} + +func renderReportBundle(value reporting.Snapshot, secrets ...string) (diagnostics.ReportBundle, error) { + errorsYAML, err := reporting.YAML(value.ErrorsDocument(), secrets...) + if err != nil { + return diagnostics.ReportBundle{}, err + } + recoveriesYAML, err := reporting.YAML(value.RecoveriesDocument(), secrets...) + if err != nil { + return diagnostics.ReportBundle{}, err + } + return diagnostics.ReportBundle{Errors: errorsYAML, Recoveries: recoveriesYAML, Markdown: reporting.Markdown(value, secrets...)}, nil +} diff --git a/internal/cli/process_inputs_test.go b/internal/cli/process_inputs_test.go new file mode 100644 index 0000000..5a6d789 --- /dev/null +++ b/internal/cli/process_inputs_test.go @@ -0,0 +1,78 @@ +package cli + +import ( + "context" + "io" + "os" + "path/filepath" + "slices" + "strings" + "testing" + + commandkit "go.maximumdirect.net/commandkit" + ck "go.maximumdirect.net/configkit" +) + +func TestQuietDestinationCheckPrecedesInputAcquisition(t *testing.T) { + _, err := loadProcessInputs(context.Background(), processInput{Transcript: "missing.json"}, commandkit.ReportingModeQuiet) + if err == nil || !strings.Contains(err.Error(), "/dev/null") { + t.Fatalf("error=%v", err) + } +} + +func TestProcessAcquisitionCapturesConsumedFiles(t *testing.T) { + dir := t.TempDir() + for name, contents := range map[string]string{ + "config.yml": "version: 3\npipeline:\n stages: []\n", + "transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`, + "prompts.yml": "prompts:\n - id: extra\n messages:\n - role: user\n content_file: text.txt\n output:\n format: text\n validation_mode: none\n", + "text.txt": "example", + } { + if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil { + t.Fatal(err) + } + } + input := processInput{Config: commandkit.Supplied[string]{Value: filepath.Join(dir, "config.yml"), Set: true}, Transcript: filepath.Join(dir, "transcript.json")} + result, err := loadProcessInputs(context.Background(), input, commandkit.ReportingModeOrdinary) + if err != nil { + t.Fatal(err) + } + for _, name := range []string{"config.yml", "prompts.yml", "text.txt"} { + if !slices.Contains(result.ProtectedPaths, filepath.Join(dir, name)) { + t.Errorf("missing %s in %v", name, result.ProtectedPaths) + } + } +} + +func TestPrepareProcessInitializesReportsAndToleratesDebugLoss(t *testing.T) { + dir := t.TempDir() + for name, contents := range map[string]string{ + "config.yml": "version: 3\npipeline:\n stages: []\n", + "transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`, + "blocked": "not a directory", + } { + if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil { + t.Fatal(err) + } + } + input := processInput{Config: commandkit.Supplied[string]{Value: filepath.Join(dir, "config.yml"), Set: true}, Transcript: filepath.Join(dir, "transcript.json"), Output: commandkit.Supplied[string]{Value: filepath.Join(dir, "output.json"), Set: true}, Debug: commandkit.Supplied[bool]{Value: true, Set: true}, Overrides: []ck.Override{{Path: ck.Keys("diagnostics", "work_dir"), Value: filepath.Join(dir, "blocked"), Flag: "--work-dir"}}} + loaded, err := loadProcessInputs(context.Background(), input, commandkit.ReportingModeOrdinary) + if err != nil { + t.Fatal(err) + } + prepared, err := prepareProcess(context.Background(), input, loaded, io.Discard) + if err != nil { + t.Fatal(err) + } + defer prepared.Prepared.Output.Close() + if len(prepared.Prepared.Warnings) == 0 { + t.Fatal("debug loss was not reported") + } + report, err := os.ReadFile(filepath.Join(dir, "errors.yml")) + if err != nil || !strings.Contains(string(report), "in_progress") { + t.Fatalf("report=%q err=%v", report, err) + } + if _, err := os.Stat(input.Output.Value); !os.IsNotExist(err) { + t.Fatalf("output published early: %v", err) + } +} diff --git a/internal/cli/process_output.go b/internal/cli/process_output.go new file mode 100644 index 0000000..56d243d --- /dev/null +++ b/internal/cli/process_output.go @@ -0,0 +1,65 @@ +package cli + +import ( + "fmt" + "time" + + "gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/redaction" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting" + commandkit "go.maximumdirect.net/commandkit" +) + +// finishProcess delivers the transcript and reports before emitting success information. +func finishProcess(inv commandkit.Invocation, input processInput, loaded *processInputs, prepared *preparedProcess, snapshot reporting.Snapshot, transcript []byte, primary error) commandkit.Outcome { + reporter := inv.Reporter() + secrets := loaded.Runtime.ConfiguredSecrets(inv.Context) + redact := func(err error) string { return loaded.Runtime.RedactError(inv.Context, err) } + for _, warning := range prepared.Prepared.Warnings { + _ = reporter.Warn("Debug artifact: " + redact(warning)) + } + completed := time.Now().UTC() + snapshot.RunID, snapshot.StartedAt, snapshot.CompletedAt = prepared.RunID, prepared.Started, &completed + snapshot.TranscriptPath, snapshot.OutputPath = input.Transcript, input.Output.Value + if primary != nil { + snapshot.Status = reporting.StatusFailed + if len(snapshot.Errors) == 0 { + snapshot.Errors = []reporting.ErrorRecord{{Scope: "run", Reason: "processing_failure", Message: "Processing failed"}} + } + } else { + snapshot.Status = reporting.StatusCompleted + } + reports, err := renderReportBundle(snapshot, secrets...) + if err != nil { + if primary == nil { + primary = err + } + snapshot.Status = reporting.StatusFailed + } + failed := func(failures []diagnostics.FileFailure) diagnostics.ReportBundle { + copySnapshot := snapshot + copySnapshot.Status = reporting.StatusFailed + copySnapshot.Errors = append([]reporting.ErrorRecord(nil), snapshot.Errors...) + for _, failure := range failures { + copySnapshot.Errors = append(copySnapshot.Errors, reporting.ErrorRecord{Scope: "run", Reason: "output_failure", Message: "Required output delivery failed", OutputPath: failure.Path}) + } + result, bundleErr := renderReportBundle(copySnapshot, secrets...) + if bundleErr != nil { + return reports + } + return result + } + result := prepared.Prepared.Output.Finish(primary, transcript, reports, failed) + if result != nil { + return commandkit.Failure(redact(result)).WithCause(result) + } + safe := redaction.New(secrets) + _ = reporter.AtEnd().Info(reporting.CompletionLine(snapshot)) + for _, path := range prepared.Prepared.Output.ReportPaths() { + _ = reporter.AtEnd().Info(fmt.Sprintf("Report: %s", safe.String(path))) + } + if prepared.Prepared.Debug != nil { + _ = reporter.AtEnd().Info(fmt.Sprintf("Diagnostics: %s", safe.String(prepared.Prepared.Debug.Path()))) + } + return commandkit.Success() +} diff --git a/internal/cli/process_output_test.go b/internal/cli/process_output_test.go new file mode 100644 index 0000000..9255add --- /dev/null +++ b/internal/cli/process_output_test.go @@ -0,0 +1,112 @@ +package cli + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting" + commandkit "go.maximumdirect.net/commandkit" +) + +func TestPreparedDeliveryFollowsReportingMode(t *testing.T) { + dir := t.TempDir() + for name, contents := range map[string]string{ + "config.yml": "version: 3\npipeline:\n stages: []\nreporting:\n enabled: false\n", + "transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`, + } { + if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil { + t.Fatal(err) + } + } + for _, tc := range []struct { + name string + flags []string + file bool + wantInfo bool + }{ + {"ordinary", nil, false, false}, {"verbose", []string{"--verbose"}, false, true}, {"quiet file", []string{"--quiet"}, true, false}, + } { + t.Run(tc.name, func(t *testing.T) { + input := processInput{Config: commandkit.Supplied[string]{Value: filepath.Join(dir, "config.yml"), Set: true}, Transcript: filepath.Join(dir, "transcript.json")} + if tc.file { + input.Output = commandkit.Supplied[string]{Value: filepath.Join(dir, "result.json"), Set: true} + } + cmd := commandkit.NewCommand(commandkit.CommandSpec[struct{}]{Name: "deliver", Run: func(inv commandkit.Invocation, _ struct{}) commandkit.Outcome { + loaded, err := loadProcessInputs(inv.Context, input, inv.ReportingMode()) + if err != nil { + return commandkit.Usage(err.Error()) + } + prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output()) + if err != nil { + return commandkit.Failure(err.Error()) + } + return finishProcess(inv, input, loaded, prepared, reporting.Snapshot{}, []byte("transcript bytes"), nil) + }}) + app, err := commandkit.New(commandkit.Identity{Name: "test"}, cmd) + if err != nil { + t.Fatal(err) + } + var out, stderr bytes.Buffer + args := append([]string{"deliver"}, tc.flags...) + outcome := app.Execute(context.Background(), args, commandkit.Streams{Out: &out, Err: &stderr}) + if outcome.Status != 0 { + t.Fatalf("outcome=%+v err=%s", outcome, stderr.String()) + } + if tc.file { + data, err := os.ReadFile(input.Output.Value) + if err != nil || string(data) != "transcript bytes" { + t.Fatalf("file=%q err=%v", data, err) + } + if out.Len() != 0 { + t.Fatalf("stdout=%q", out.String()) + } + } else if out.String() != "transcript bytes" { + t.Fatalf("stdout=%q", out.String()) + } + if strings.Contains(stderr.String(), "Completed:") != tc.wantInfo { + t.Fatalf("stderr=%q", stderr.String()) + } + }) + } +} + +func TestDeliveryFailureSuppressesCompletionSummary(t *testing.T) { + dir := t.TempDir() + for name, contents := range map[string]string{ + "config.yml": "version: 3\npipeline:\n stages: []\nreporting:\n enabled: false\n", + "transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`, + } { + if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil { + t.Fatal(err) + } + } + input := processInput{Config: commandkit.Supplied[string]{Value: filepath.Join(dir, "config.yml"), Set: true}, Transcript: filepath.Join(dir, "transcript.json")} + cmd := commandkit.NewCommand(commandkit.CommandSpec[struct{}]{Name: "deliver", Run: func(inv commandkit.Invocation, _ struct{}) commandkit.Outcome { + loaded, err := loadProcessInputs(inv.Context, input, inv.ReportingMode()) + if err != nil { + return commandkit.Failure(err.Error()) + } + prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output()) + if err != nil { + return commandkit.Failure(err.Error()) + } + return finishProcess(inv, input, loaded, prepared, reporting.Snapshot{}, []byte("transcript bytes"), nil) + }}) + app, err := commandkit.New(commandkit.Identity{Name: "test"}, cmd) + if err != nil { + t.Fatal(err) + } + var stderr bytes.Buffer + outcome := app.Execute(context.Background(), []string{"deliver", "--verbose"}, commandkit.Streams{Out: shortProcessWriter{}, Err: &stderr}) + if outcome.Status == 0 || strings.Contains(stderr.String(), "Completed:") { + t.Fatalf("status=%d stderr=%q", outcome.Status, stderr.String()) + } +} + +type shortProcessWriter struct{} + +func (shortProcessWriter) Write(data []byte) (int, error) { return len(data) / 2, nil } diff --git a/internal/framework/preflight/preflight.go b/internal/framework/preflight/preflight.go new file mode 100644 index 0000000..54182a7 --- /dev/null +++ b/internal/framework/preflight/preflight.go @@ -0,0 +1,76 @@ +// Package preflight checks resolved processing inputs and prepares output resources. +package preflight + +import ( + "context" + "fmt" + + "gitea.maximumdirect.net/eric/audita/v3/internal/core/config" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/schema" +) + +// Input contains already acquired application and transcript inputs. +type Input struct { + Config *config.Config + Transcript *schema.SourceTranscript + Output diagnostics.OutputSelection + InitialReports diagnostics.ReportBundle + FailedReports diagnostics.ReportBundle + Debug *DebugInput +} + +type DebugInput struct { + WorkDir string + RunID string + Metadata diagnostics.InvocationMetadata + EffectiveConfig []byte + Secrets []string +} + +type Prepared struct { + Output *diagnostics.OutputSession + Debug *diagnostics.RunDirectory + Warnings []error +} + +// Prepare validates invocation inputs and collisions before acquiring outputs. +func Prepare(ctx context.Context, input Input) (*Prepared, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if input.Config == nil { + return nil, fmt.Errorf("application configuration is required") + } + if input.Transcript == nil { + return nil, fmt.Errorf("transcript is required") + } + session, err := diagnostics.SelectOutput(input.Output) + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if err := session.Prepare(); err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + _ = session.Close() + return nil, err + } + if err := session.Initialize(input.InitialReports, input.FailedReports); err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, session.Finish(err, nil, input.FailedReports, func([]diagnostics.FileFailure) diagnostics.ReportBundle { return input.FailedReports }) + } + result := &Prepared{Output: session} + if input.Debug != nil { + result.Debug, result.Warnings = session.PrepareDebug(input.Debug.WorkDir, input.Debug.RunID, input.Debug.Metadata, input.Debug.EffectiveConfig, input.Debug.Secrets...) + } + if err := ctx.Err(); err != nil { + return nil, session.Finish(err, nil, input.FailedReports, func([]diagnostics.FileFailure) diagnostics.ReportBundle { return input.FailedReports }) + } + return result, nil +} diff --git a/internal/framework/preflight/preflight_test.go b/internal/framework/preflight/preflight_test.go new file mode 100644 index 0000000..1dade07 --- /dev/null +++ b/internal/framework/preflight/preflight_test.go @@ -0,0 +1,48 @@ +package preflight + +import ( + "context" + "os" + "path/filepath" + "testing" + + "gitea.maximumdirect.net/eric/audita/v3/internal/core/config" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/schema" +) + +func TestReadinessAndCollisionPrecedeOutputPreparation(t *testing.T) { + dir := t.TempDir() + output := filepath.Join(dir, "output.json") + if _, err := Prepare(context.Background(), Input{Config: &config.Config{}, Output: diagnostics.OutputSelection{TranscriptPath: output}}); err == nil { + t.Fatal("missing transcript accepted") + } + if _, err := os.Stat(output); !os.IsNotExist(err) { + t.Fatalf("output created: %v", err) + } + if _, err := Prepare(context.Background(), Input{Config: &config.Config{}, Transcript: &schema.SourceTranscript{}, Output: diagnostics.OutputSelection{TranscriptPath: output, ProtectedPaths: []string{output}}}); err == nil { + t.Fatal("collision accepted") + } + entries, err := os.ReadDir(dir) + if err != nil || len(entries) != 0 { + t.Fatalf("prepared before check: %v %v", entries, err) + } +} + +func TestOptionalDebugFailureDoesNotPreventPreparation(t *testing.T) { + dir := t.TempDir() + blocked := filepath.Join(dir, "file") + if err := os.WriteFile(blocked, []byte("x"), 0600); err != nil { + t.Fatal(err) + } + result, err := Prepare(context.Background(), Input{Config: &config.Config{}, Transcript: &schema.SourceTranscript{}, Output: diagnostics.OutputSelection{TranscriptPath: filepath.Join(dir, "output")}, Debug: &DebugInput{WorkDir: blocked, RunID: "run"}}) + if err != nil { + t.Fatal(err) + } + if len(result.Warnings) == 0 || result.Debug != nil { + t.Fatalf("debug=%v warnings=%v", result.Debug, result.Warnings) + } + if err := result.Output.Close(); err != nil { + t.Fatal(err) + } +}