diff --git a/docs/architecture.md b/docs/architecture.md index 7d9d632..f772b2a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -17,14 +17,15 @@ Implemented today: - Per-run diagnostics directory creation plus Phase 6 process-level artifacts. - Process report JSON output with diagnostics artifact references. - Framework foundation packages for contracts and proposal application. +- Production runner orchestration package with deterministic sequential module execution. +- Module-level report structures with applied/skipped change records. Not implemented in CLI runtime path today: - Real module execution pipeline (`glossary`, `homophones`, `spoken_word`, `grammar`). - Structured LLM proposal generation. - Validator chain execution. -- Pipeline runner orchestration. - Production LLM scheduler behavior. -- End-to-end transcript polishing beyond normalization. +- End-to-end transcript polishing with real module behavior. ## Actual Go package layout @@ -73,6 +74,9 @@ internal/framework/proposals/ policy.go preview.go apply.go + +internal/framework/runner/ + runner.go ``` ## Current CLI behavior @@ -94,14 +98,15 @@ Current runtime flow (`internal/cli/run.go`): 9. Write normalized transcript and normalization summary artifacts. 10. Chunk normalized transcript and compute chunk summaries. 11. Write chunking summary artifact. -12. Output normalized transcript to `--output` file or stdout. -13. Build process report (`phase` currently set to `phase3-chunking`). -14. Optionally write `--report-json`; always write run-dir `report.json`. -15. Apply work-dir retention. +12. Optionally execute runner modules sequentially when an injected module registry/factory is available (used by deterministic tests today). +13. Output working transcript to `--output` file or stdout. +14. Build process report (`phase` currently set to `phase7-runner`). +15. Optionally write `--report-json`; always write run-dir `report.json`. +16. Apply work-dir retention. Important behavior details: -- Glossary is validated but not yet used for correction logic. -- No module execution occurs despite `--modules` config support. +- Glossary is validated but not yet used for real correction module logic. +- Default production CLI behavior remains deterministic normalization/chunking output because no real module implementations are registered yet. - No LLM calls occur. - Success path is generally quiet on stderr. - Source IDs are preserved into a canonical transcript before normalization; normalization then reassigns output IDs sequentially from `1`. @@ -187,7 +192,7 @@ Current behavior details: `internal/framework/contracts` provides interfaces and run-spec metadata scaffolding, including deterministic repeated module instance naming (`ResolveModuleRunSpecs`). -These are framework primitives only; they are not yet wired into CLI runtime module execution. +These primitives are wired into the production runner and report model. Real module implementations are still pending. ## Reports and diagnostics (implemented) Current per-run artifacts include: @@ -214,6 +219,10 @@ Current process reports include diagnostics metadata references for: - redacted effective-config artifact path; - error-log artifact path on failure. +Current process reports also include: +- module-level results (when runner modules execute), including applied/skipped proposal changes; +- run-level module summary totals and failed module instance metadata. + Retention modes implemented in `ApplyRetention`: - `always`: keep all run directories. - `never`: keep successful run directories. @@ -221,7 +230,7 @@ Retention modes implemented in `ApplyRetention`: - failed runs are always retained. Current runtime note: -- module execution is not implemented yet, so successful runs generally have no skipped corrections and `auto` typically removes clean successful run directories. +- real module execution is not implemented yet, so normal successful runs generally have no skipped corrections and `auto` typically removes clean successful run directories. Intentionally deferred to module/LLM phases: - module prompt/response diagnostics artifacts are not produced yet because module execution and LLM calls are not in the runtime path. @@ -236,8 +245,10 @@ Implemented tests currently cover: - deterministic chunking and summaries (`internal/core/chunking/*_test.go`) - proposal preview/apply semantics (`internal/framework/proposals/*_test.go`) - contracts/foundation composition tests (`internal/framework/contracts/*_test.go`) +- runner sequencing and failure behavior with deterministic fake modules (`internal/framework/runner/*_test.go`) +- CLI runner integration through injected fake module factories (`internal/cli/run_test.go`) -Not covered yet (because not implemented): end-to-end module runner behavior, validator runtime flow, and real LLM integration. +Not covered yet (because not implemented): validator runtime flow with approvals/rejections and real LLM integration. ## Intended final architecture (not yet implemented) The intended end-state still matches the rewrite plan: @@ -247,6 +258,6 @@ The intended end-state still matches the rewrite plan: - deterministic and LLM validators - validator cardinality enforcement in pipeline execution - proposal application integrated per module stage -- richer run reporting with module-level applied/skipped changes +- prompt/response diagnostics for LLM/module stages Until those phases are implemented, documentation and external descriptions should treat the current Go CLI as deterministic preprocessing/reporting infrastructure, not a full LLM transcript polisher. diff --git a/docs/rewrite-notes.md b/docs/rewrite-notes.md index eb7135d..b2e74e8 100644 --- a/docs/rewrite-notes.md +++ b/docs/rewrite-notes.md @@ -44,10 +44,12 @@ Implemented: - Process report output through `--report-json` and run-dir `report.json`. - Report-level diagnostics artifact references. - Framework foundation packages for contracts and proposal preview/apply semantics. +- Production runner orchestration over a mutable working transcript. +- Module-level report structures and run-level module summaries. +- CLI runner integration point via injectable module factory/registry (used by deterministic tests). - Broad deterministic and CLI/subprocess test coverage for implemented phases through `go test ./...`. Not yet implemented in runtime pipeline: -- Module runner orchestration. - Real correction modules. - Validator-chain execution. - Structured LLM client integration. @@ -135,53 +137,33 @@ Implemented: Intentionally deferred: - Module prompt/response diagnostics artifacts are not produced yet because module execution and LLM calls are not implemented in the runtime path. +### Phase 7: Pipeline runner with deterministic test modules + +Completed. + +Implemented: +- `internal/framework/runner` production package with sequential module orchestration. +- Deterministic module run-spec resolution and repeated instance naming (`glossary_1`, `glossary_2`, etc.). +- Mutable working transcript handoff across module instances. +- Proposal application through `internal/framework/proposals`. +- Per-module applied/skipped change capture and module status/timing metadata. +- Partial-progress return on module failure, with pipeline stop on first failure. +- Process report support for module-level results and run-level module summaries. +- CLI runtime integration point via injectable module factory/registry, exercised by deterministic fake-module tests. + +Not implemented in Phase 7 (by design): +- Real `glossary`, `homophones`, `spoken_word`, `grammar` production modules. +- Validators (Phase 8). +- Structured LLM calls or scheduler behavior. +- Prompt/response diagnostics. +- End-to-end transcript polishing. + +Current runtime behavior note: +- Default user-facing CLI behavior remains deterministic normalization/chunking output unless test-only module injection is used during tests. + ## Remaining work plan -## Phase 7: Pipeline runner with deterministic test modules - -### Purpose - -Introduce the production pipeline runner that executes configured module instances sequentially over a mutable working transcript, but without real LLM calls or real correction modules yet. - -This phase closes the largest architectural gap: the runtime path currently normalizes and chunks, but does not execute module stages. - -### Scope - -Implement: -- `internal/framework/runner` or equivalent production runner package. -- Module run-spec resolution from configured module keys. -- Sequential execution of module instances. -- Mutable working transcript handoff from one module to the next. -- Integration with existing proposal application package. -- Module-level report structures. -- Applied/skipped changes recorded per module instance. -- Test-only deterministic modules or fake modules to exercise runner behavior. -- CLI wiring from `internal/cli/run.go` into the runner. - -Do not implement: -- Real `glossary`, `homophones`, `spoken_word`, or `grammar` modules. -- LLM calls. -- Validator chains. -- LLM scheduler behavior. -- Prompt/response diagnostics. - -### Expected behavior at end of phase - -The CLI should still not perform real transcript polishing by default unless a test-only module sequence is explicitly used in tests. The production path may resolve configured module names, but unsupported real modules should fail cleanly or remain gated until implemented in later phases. - -The runner itself should be production-quality and tested with deterministic fake modules. - -### Definition of done - -- Configured module keys can be resolved into deterministic run specs. -- Repeated module names are resolved consistently, such as `glossary_1` and `glossary_2`. -- Runner executes module instances sequentially. -- Later modules see transcript changes from earlier modules. -- Applied and skipped changes are recorded per module instance. -- Process report includes module-level report structures. -- Existing stdout/stderr behavior remains unchanged. -- Diagnostics retention behavior remains unchanged. -- `go test ./...` passes. +Next recommended phase: **Phase 8 (runtime validator framework and deterministic validators)**. ## Phase 8: Runtime validator framework and deterministic validators @@ -716,4 +698,4 @@ Before accepting a phase implementation, verify: - Are fake LLM tests used instead of requiring real credentials? - Are new public behaviors documented? - Does the code follow the architecture in `docs/architecture.md`? -- Does the implementation move Audita closer to Python feature parity? \ No newline at end of file +- Does the implementation move Audita closer to Python feature parity? diff --git a/internal/cli/run.go b/internal/cli/run.go index ac63999..b2cac4d 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -1,6 +1,7 @@ package cli import ( + "context" "errors" "flag" "fmt" @@ -16,6 +17,8 @@ import ( "gitea.maximumdirect.net/eric/audita/internal/core/normalization" "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/runner" ) type processInvocation struct { @@ -26,15 +29,17 @@ type processInvocation struct { Config config.Config } -var processRunner = func(inv processInvocation, stdout io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *diagnostics.RunDirectory, error) { +var processModuleFactory runner.ModuleFactory + +var processRunner = func(inv processInvocation, stdout io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) { runDir, err := diagnostics.NewRunDirectory(inv.Config.WorkDir, string(inv.Config.WorkDirRetention)) if err != nil { - return nil, nil, nil, fmt.Errorf("run_dir_creation: %w", err) + return nil, nil, nil, nil, fmt.Errorf("run_dir_creation: %w", err) } - fail := func(phase string, err error) (*normalization.NormalizationSummary, *chunking.Summary, *diagnostics.RunDirectory, error) { + fail := func(phase string, err error, runOutput *runner.RunOutput) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) { _ = runDir.WriteErrorLog(fmt.Sprintf("%s: %v", phase, err)) - return nil, nil, runDir, fmt.Errorf("%s: %w", phase, err) + return nil, nil, runOutput, runDir, fmt.Errorf("%s: %w", phase, err) } if err := runDir.WriteInvocationMetadata(diagnostics.InvocationMetadata{ @@ -53,21 +58,22 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio transcriptBytes, err := coreio.ReadRequiredFile(inv.TranscriptPath, "transcript") if err != nil { - return fail("transcript_read", err) + return fail("transcript_read", err, nil) } glossaryBytes, err := coreio.ReadRequiredFile(inv.GlossaryPath, "glossary") if err != nil { - return fail("glossary_read", err) + return fail("glossary_read", err, nil) } sourceTranscript, err := schema.ParseSourceTranscriptJSON(transcriptBytes) if err != nil { - return fail("transcript_schema", err) + return fail("transcript_schema", err, nil) } - if _, err := schema.ParseGlossaryYAML(glossaryBytes); err != nil { - return fail("glossary_schema", err) + glossary, err := schema.ParseGlossaryYAML(glossaryBytes) + if err != nil { + return fail("glossary_schema", err, nil) } if err := runDir.WriteSourceTranscript(sourceTranscript, transcriptBytes); err != nil { @@ -100,7 +106,7 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio sections, chunkErr := chunker.ChunkTranscript(normalizedTranscript) if chunkErr != nil { - return fail("chunking", chunkErr) + return fail("chunking", chunkErr, nil) } chunkConfig := chunking.ChunkingConfig{ @@ -115,23 +121,44 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio _ = runDir.WriteErrorLog(fmt.Sprintf("chunking_summary: %v", err)) } - outputBytes, err := schema.TranscriptToJSON(normalizedTranscript) + workingTranscript := normalizedTranscript + var runOutput *runner.RunOutput + if processModuleFactory != nil { + moduleSpecs, err := contracts.ResolveModuleRunSpecs(inv.Config.Modules) + if err != nil { + return fail("runner_setup", err, nil) + } + + runnerResult, runErr := runner.New(processModuleFactory).Run(context.Background(), runner.RunInput{ + Config: &inv.Config, + Transcript: normalizedTranscript, + Glossary: glossary, + ModuleSpecs: moduleSpecs, + }) + runOutput = &runnerResult + if runErr != nil { + return fail("runner_execution", runErr, runOutput) + } + workingTranscript = runnerResult.FinalTranscript + } + + outputBytes, err := schema.TranscriptToJSON(workingTranscript) if err != nil { - return fail("serialization", err) + return fail("serialization", err, runOutput) } if strings.TrimSpace(inv.OutputPath) != "" { if err := coreio.WriteFile(inv.OutputPath, outputBytes); err != nil { - return fail("output_write", err) + return fail("output_write", err, runOutput) } - return normSummary, &chunkSummary, runDir, nil + return normSummary, &chunkSummary, runOutput, runDir, nil } if _, err := stdout.Write(outputBytes); err != nil { - return fail("stdout_write", err) + return fail("stdout_write", err, runOutput) } - return normSummary, &chunkSummary, runDir, nil + return normSummary, &chunkSummary, runOutput, runDir, nil } func sourceToCanonicalTranscript(source *schema.SourceTranscript) *schema.Transcript { @@ -292,12 +319,12 @@ func runProcess(args []string, stdout, stderr io.Writer) int { Config: cfg, } - normSummary, chunkSummary, runDir, runErr := processRunner(inv, stdout) + normSummary, chunkSummary, runOutput, runDir, runErr := processRunner(inv, stdout) completedAt := time.Now().UTC() if runErr != nil { errorPhase, errorMessage := extractErrorPhase(runErr) - report := buildProcessReport("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil) + report := buildProcessReport("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil, runOutput) if strings.TrimSpace(inv.ReportJSONPath) != "" { if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil { @@ -316,7 +343,7 @@ func runProcess(args []string, stdout, stderr io.Writer) int { return 1 } - report := buildProcessReport("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary) + report := buildProcessReport("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary, runOutput) if strings.TrimSpace(inv.ReportJSONPath) != "" { if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil { @@ -331,11 +358,21 @@ func runProcess(args []string, stdout, stderr io.Writer) int { } } + hasSkippedCorrections := false + if runOutput != nil { + for _, mr := range runOutput.ModuleResults { + if len(mr.SkippedChanges) > 0 { + hasSkippedCorrections = true + break + } + } + } + if runDir != nil { _ = runDir.WriteReport(report) if err := runDir.ApplyRetention(diagnostics.RetentionDecisionInput{ RunSucceeded: true, - HasSkippedCorrections: false, // Hook for future module-level skipped-correction reporting. + HasSkippedCorrections: hasSkippedCorrections, }); err != nil { fmt.Fprintf(stderr, "audita process: failed to apply work-dir retention: %v\n", err) return 1 @@ -356,9 +393,9 @@ func extractErrorPhase(err error) (phase string, message string) { return "", msg } -func buildProcessReport(status string, inv processInvocation, runDir *diagnostics.RunDirectory, startedAt, completedAt time.Time, errorMessage string, errorPhase string, normalizationSummary *normalization.NormalizationSummary, chunkingSummary *chunking.Summary) reporting.ProcessReport { +func buildProcessReport(status string, inv processInvocation, runDir *diagnostics.RunDirectory, startedAt, completedAt time.Time, errorMessage string, errorPhase string, normalizationSummary *normalization.NormalizationSummary, chunkingSummary *chunking.Summary, runOutput *runner.RunOutput) reporting.ProcessReport { report := reporting.ProcessReport{ - Phase: "phase3-chunking", + Phase: "phase7-runner", Status: status, Operation: "process", TranscriptPath: inv.TranscriptPath, @@ -409,9 +446,42 @@ func buildProcessReport(status string, inv processInvocation, runDir *diagnostic MinSectionTokens: chunkingSummary.MinSectionTokens, } } + report.ModulesSummary, report.ModuleResults = buildModuleReporting(runOutput) return report } +func buildModuleReporting(runOutput *runner.RunOutput) (*reporting.ModulesSummary, []reporting.ModuleReport) { + if runOutput == nil || len(runOutput.ModuleResults) == 0 { + return nil, nil + } + + moduleReports := make([]reporting.ModuleReport, 0, len(runOutput.ModuleResults)) + summary := &reporting.ModulesSummary{ModuleCount: len(runOutput.ModuleResults)} + for _, r := range runOutput.ModuleResults { + startedAt := r.StartedAt + completedAt := r.CompletedAt + moduleReports = append(moduleReports, reporting.ModuleReport{ + ModuleKey: r.ModuleKey, + ModuleInstance: r.ModuleInstance, + ReplacementPolicy: string(r.ReplacementPolicy), + Status: r.Status, + ProposalCount: r.ProposalCount, + AppliedChanges: r.AppliedChanges, + SkippedChanges: r.SkippedChanges, + ErrorMessage: r.ErrorMessage, + StartedAt: &startedAt, + CompletedAt: &completedAt, + }) + summary.TotalAppliedChanges += len(r.AppliedChanges) + summary.TotalSkippedChanges += len(r.SkippedChanges) + if r.Status == runner.ModuleStatusFailed && summary.FailedModuleInstance == "" { + summary.FailedModuleInstance = r.ModuleInstance + } + } + + return summary, moduleReports +} + type processFlags struct { glossaryPath *string outputPath *string diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index f5994cc..1edc869 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -2,7 +2,9 @@ package cli import ( "bytes" + "context" "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -11,6 +13,8 @@ import ( "gitea.maximumdirect.net/eric/audita/internal/core/normalization" "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/proposals" ) func TestRunRootHelp(t *testing.T) { @@ -609,8 +613,188 @@ func TestRunProcessReportJSONIncludesChunkingSummary(t *testing.T) { if report.Chunking.MaxSectionTokens == 0 { t.Errorf("expected max_section_tokens in report") } - if report.Phase != "phase3-chunking" { - t.Errorf("expected phase 'phase3-chunking', got %q", report.Phase) + if report.Phase != "phase7-runner" { + t.Errorf("expected phase 'phase7-runner', got %q", report.Phase) + } +} + +type fakeModuleFactory struct { + modules map[string]contracts.TranscriptModule +} + +func (f fakeModuleFactory) ModuleForSpec(spec contracts.ModuleRunSpec) (contracts.TranscriptModule, error) { + m, ok := f.modules[spec.InstanceName] + if !ok { + return nil, errors.New("module not registered") + } + return m, nil +} + +type fakeModule struct { + key string + policy proposals.ReplacementPolicy + proposeF func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) +} + +func (m fakeModule) Key() string { return m.key } +func (m fakeModule) ReplacementPolicy() proposals.ReplacementPolicy { return m.policy } +func (m fakeModule) Validators() []contracts.Validator { return nil } +func (m fakeModule) Propose(ctx context.Context, req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + if m.proposeF == nil { + return nil, nil + } + return m.proposeF(req) +} + +func TestRunProcessInjectedFactoryExecutesRunnerAndReportsModules(t *testing.T) { + processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{ + "m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{ + {TargetSegmentID: 1, OriginalText: "Hello", CorrectedText: "Hi", Confidence: 1}, + }, nil + }}, + "m2": fakeModule{key: "m2", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + if req.WorkingTranscript.Segments[0].Text != "Hi world" { + t.Fatalf("expected module 2 to see module 1 changes, got %q", req.WorkingTranscript.Segments[0].Text) + } + return []proposals.CorrectionProposal{ + {TargetSegmentID: 1, OriginalText: "Hi", CorrectedText: "Hey", Confidence: 1}, + {TargetSegmentID: 1, OriginalText: "missing", CorrectedText: "noop", Confidence: 1}, + }, nil + }}, + }} + t.Cleanup(func() { processModuleFactory = nil }) + + var stdout, stderr bytes.Buffer + workDir := t.TempDir() + reportPath := filepath.Join(t.TempDir(), "report.json") + outputPath := filepath.Join(t.TempDir(), "out.json") + transcriptPath := writeFile(t, "transcript.json", `[ + {"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"Hello world"} +]`) + + exitCode := Run([]string{ + "process", transcriptPath, + "--glossary", fixturePath("tiny_glossary.yaml"), + "--modules", "m1,m2", + "--output", outputPath, + "--report-json", reportPath, + "--work-dir", workDir, + "--work-dir-retention", "always", + }, &stdout, &stderr) + if exitCode != 0 { + t.Fatalf("expected success, got exit %d stderr=%q", exitCode, stderr.String()) + } + if stdout.Len() != 0 { + t.Fatalf("expected empty stdout with --output, got %q", stdout.String()) + } + + out := readFile(t, outputPath) + parsed, err := schema.ParseTranscriptJSON(out) + if err != nil { + t.Fatalf("parse output transcript: %v", err) + } + if parsed.Segments[0].Text != "Hey world" { + t.Fatalf("expected runner-applied output text, got %q", parsed.Segments[0].Text) + } + + report := readProcessReport(t, reportPath) + if report.ModulesSummary == nil || report.ModulesSummary.ModuleCount != 2 { + t.Fatalf("expected module summary for 2 modules, got %+v", report.ModulesSummary) + } + if report.ModulesSummary.TotalAppliedChanges != 2 || report.ModulesSummary.TotalSkippedChanges != 1 { + t.Fatalf("unexpected module totals: %+v", report.ModulesSummary) + } + if len(report.ModuleResults) != 2 { + t.Fatalf("expected 2 module results, got %d", len(report.ModuleResults)) + } + if len(report.ModuleResults[1].SkippedChanges) != 1 { + t.Fatalf("expected skipped change for module 2, got %+v", report.ModuleResults[1].SkippedChanges) + } + + runDirReport := readProcessReport(t, filepath.Join(onlyRunDir(t, workDir), "report.json")) + if len(runDirReport.ModuleResults) != 2 { + t.Fatalf("expected module results in run-dir report") + } +} + +func TestRunProcessInjectedFactorySkippedKeepsAutoRetention(t *testing.T) { + processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{ + "m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{ + {TargetSegmentID: 1, OriginalText: "word", CorrectedText: "term", Confidence: 1}, + }, nil + }}, + }} + t.Cleanup(func() { processModuleFactory = nil }) + + var stdout, stderr bytes.Buffer + workDir := t.TempDir() + transcriptPath := writeFile(t, "transcript.json", `[ + {"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"word word"} +]`) + exitCode := Run([]string{ + "process", transcriptPath, + "--glossary", fixturePath("tiny_glossary.yaml"), + "--modules", "m1", + "--work-dir", workDir, + "--work-dir-retention", "auto", + }, &stdout, &stderr) + if exitCode != 0 { + t.Fatalf("expected success, got %d stderr=%q", exitCode, stderr.String()) + } + if _, err := os.Stat(onlyRunDir(t, workDir)); err != nil { + t.Fatalf("expected run dir retained for skipped corrections under auto: %v", err) + } +} + +func TestRunProcessInjectedFactoryFailureWritesFailedReport(t *testing.T) { + processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{ + "m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return nil, errors.New("test failure") + }}, + }} + t.Cleanup(func() { processModuleFactory = nil }) + + var stdout, stderr bytes.Buffer + workDir := t.TempDir() + reportPath := filepath.Join(t.TempDir(), "report.json") + transcriptPath := writeFile(t, "transcript.json", `[ + {"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"Hello"} +]`) + + exitCode := Run([]string{ + "process", transcriptPath, + "--glossary", fixturePath("tiny_glossary.yaml"), + "--modules", "m1", + "--work-dir", workDir, + "--work-dir-retention", "always", + "--report-json", reportPath, + }, &stdout, &stderr) + if exitCode == 0 { + t.Fatal("expected failure exit code") + } + if stdout.Len() != 0 { + t.Fatalf("expected empty stdout on failure, got %q", stdout.String()) + } + if !strings.Contains(stderr.String(), "runner_execution") { + t.Fatalf("expected runner_execution failure on stderr, got %q", stderr.String()) + } + + report := readProcessReport(t, reportPath) + if report.Status != "failed" { + t.Fatalf("expected failed report status, got %q", report.Status) + } + if report.ModulesSummary == nil || report.ModulesSummary.FailedModuleInstance == "" { + t.Fatalf("expected failed module summary, got %+v", report.ModulesSummary) + } + if len(report.ModuleResults) != 1 || report.ModuleResults[0].Status != "failed" { + t.Fatalf("expected failed module result, got %+v", report.ModuleResults) + } + + runPath := onlyRunDir(t, workDir) + if _, err := os.Stat(filepath.Join(runPath, "error.log")); err != nil { + t.Fatalf("expected error.log for failed runner module: %v", err) } } diff --git a/internal/core/reporting/report.go b/internal/core/reporting/report.go index 5e0a712..718b959 100644 --- a/internal/core/reporting/report.go +++ b/internal/core/reporting/report.go @@ -5,6 +5,8 @@ import ( "fmt" "os" "time" + + "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" ) type ProcessReport struct { @@ -26,6 +28,28 @@ type ProcessReport struct { NormalizationSkipped NormalizationSkipped `json:"normalization_skipped,omitempty"` Chunking *ChunkingSummary `json:"chunking,omitempty"` Diagnostics *DiagnosticsMetadata `json:"diagnostics,omitempty"` + ModulesSummary *ModulesSummary `json:"modules_summary,omitempty"` + ModuleResults []ModuleReport `json:"module_results,omitempty"` +} + +type ModuleReport struct { + ModuleKey string `json:"module_key"` + ModuleInstance string `json:"module_instance"` + ReplacementPolicy string `json:"replacement_policy,omitempty"` + Status string `json:"status"` + ProposalCount int `json:"proposal_count"` + AppliedChanges []proposals.AppliedChange `json:"applied_changes,omitempty"` + SkippedChanges []proposals.SkippedChange `json:"skipped_changes,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` +} + +type ModulesSummary struct { + ModuleCount int `json:"module_count"` + TotalAppliedChanges int `json:"total_applied_changes"` + TotalSkippedChanges int `json:"total_skipped_changes"` + FailedModuleInstance string `json:"failed_module_instance,omitempty"` } type DiagnosticsMetadata struct { diff --git a/internal/core/reporting/report_test.go b/internal/core/reporting/report_test.go new file mode 100644 index 0000000..d18d5fc --- /dev/null +++ b/internal/core/reporting/report_test.go @@ -0,0 +1,91 @@ +package reporting + +import ( + "encoding/json" + "testing" + "time" + + "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" +) + +func TestProcessReportModuleResultsJSONSuccessAndSkipped(t *testing.T) { + now := time.Now().UTC() + report := ProcessReport{ + Phase: "phase7-runner", + Status: "success", + ModuleResults: []ModuleReport{ + { + ModuleKey: "grammar", + ModuleInstance: "grammar_1", + ReplacementPolicy: "require_unique", + Status: "success", + ProposalCount: 2, + AppliedChanges: []proposals.AppliedChange{{ + ProposalIndex: 0, + ModuleKey: "grammar", + }}, + SkippedChanges: []proposals.SkippedChange{{ + ProposalIndex: 1, + ModuleKey: "grammar", + SkipReason: proposals.SkipReasonMissingOriginalText, + }}, + StartedAt: &now, + CompletedAt: &now, + }, + }, + ModulesSummary: &ModulesSummary{ModuleCount: 1, TotalAppliedChanges: 1, TotalSkippedChanges: 1}, + } + + b, err := json.Marshal(report) + if err != nil { + t.Fatalf("marshal report: %v", err) + } + if !json.Valid(b) { + t.Fatalf("expected valid json") + } + + var parsed ProcessReport + if err := json.Unmarshal(b, &parsed); err != nil { + t.Fatalf("unmarshal report: %v", err) + } + if len(parsed.ModuleResults) != 1 { + t.Fatalf("expected one module result, got %d", len(parsed.ModuleResults)) + } + if parsed.ModulesSummary == nil || parsed.ModulesSummary.TotalSkippedChanges != 1 { + t.Fatalf("unexpected module summary: %+v", parsed.ModulesSummary) + } +} + +func TestProcessReportModuleResultsJSONFailedModule(t *testing.T) { + now := time.Now().UTC() + report := ProcessReport{ + Phase: "phase7-runner", + Status: "failed", + ModuleResults: []ModuleReport{ + { + ModuleKey: "grammar", + ModuleInstance: "grammar_1", + Status: "failed", + ErrorMessage: "boom", + StartedAt: &now, + CompletedAt: &now, + }, + }, + ModulesSummary: &ModulesSummary{ModuleCount: 1, FailedModuleInstance: "grammar_1"}, + } + + b, err := json.Marshal(report) + if err != nil { + t.Fatalf("marshal report: %v", err) + } + var decoded map[string]any + if err := json.Unmarshal(b, &decoded); err != nil { + t.Fatalf("unmarshal map: %v", err) + } + if _, ok := decoded["module_results"]; !ok { + t.Fatalf("expected module_results field") + } + if _, ok := decoded["modules_summary"]; !ok { + t.Fatalf("expected modules_summary field") + } +} diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go new file mode 100644 index 0000000..d8b8d4a --- /dev/null +++ b/internal/framework/runner/runner.go @@ -0,0 +1,163 @@ +package runner + +import ( + "context" + "fmt" + "time" + + "gitea.maximumdirect.net/eric/audita/internal/core/config" + "gitea.maximumdirect.net/eric/audita/internal/core/schema" + "gitea.maximumdirect.net/eric/audita/internal/framework/contracts" + "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" +) + +const ( + ModuleStatusSuccess = "success" + ModuleStatusFailed = "failed" +) + +// ModuleFactory resolves one module instance for one run spec. +type ModuleFactory interface { + ModuleForSpec(spec contracts.ModuleRunSpec) (contracts.TranscriptModule, error) +} + +// Runner executes module instances sequentially against a working transcript. +type Runner struct { + factory ModuleFactory +} + +// ModuleResult captures deterministic per-module execution output. +type ModuleResult struct { + ModuleKey string `json:"module_key"` + ModuleInstance string `json:"module_instance"` + ReplacementPolicy proposals.ReplacementPolicy `json:"replacement_policy"` + Status string `json:"status"` + ProposalCount int `json:"proposal_count"` + AppliedChanges []proposals.AppliedChange `json:"applied_changes,omitempty"` + SkippedChanges []proposals.SkippedChange `json:"skipped_changes,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` + StartedAt time.Time `json:"started_at"` + CompletedAt time.Time `json:"completed_at"` +} + +// RunInput is the deterministic runner input. +type RunInput struct { + Config *config.Config + Transcript *schema.Transcript + Glossary *schema.Glossary + ModuleSpecs []contracts.ModuleRunSpec +} + +// RunOutput is the deterministic runner output. +type RunOutput struct { + FinalTranscript *schema.Transcript `json:"-"` + ModuleResults []ModuleResult `json:"module_results"` +} + +func New(factory ModuleFactory) *Runner { + return &Runner{factory: factory} +} + +func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { + if r == nil || r.factory == nil { + return RunOutput{}, fmt.Errorf("runner module factory is required") + } + + working := cloneTranscript(input.Transcript) + results := make([]ModuleResult, 0, len(input.ModuleSpecs)) + + for _, spec := range input.ModuleSpecs { + startedAt := time.Now().UTC() + module, err := r.factory.ModuleForSpec(spec) + if err != nil { + failed := ModuleResult{ + ModuleKey: spec.ModuleKey, + ModuleInstance: spec.InstanceName, + Status: ModuleStatusFailed, + ErrorMessage: err.Error(), + StartedAt: startedAt, + CompletedAt: time.Now().UTC(), + } + results = append(results, failed) + return RunOutput{FinalTranscript: working, ModuleResults: results}, fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err) + } + + policy := module.ReplacementPolicy() + proposed, err := module.Propose(ctx, contracts.ProposalRequest{ + ExecutionContext: contracts.ExecutionContext{ + Config: input.Config, + WorkingTranscript: working, + Glossary: input.Glossary, + }, + RunSpec: contracts.ModuleRunSpec{ + ModuleKey: spec.ModuleKey, + InstanceName: spec.InstanceName, + ReplacementPolicy: policy, + }, + }) + if err != nil { + failed := ModuleResult{ + ModuleKey: spec.ModuleKey, + ModuleInstance: spec.InstanceName, + ReplacementPolicy: policy, + Status: ModuleStatusFailed, + ErrorMessage: err.Error(), + StartedAt: startedAt, + CompletedAt: time.Now().UTC(), + } + results = append(results, failed) + return RunOutput{FinalTranscript: working, ModuleResults: results}, fmt.Errorf("module %q failed: %w", spec.InstanceName, err) + } + + enriched := make([]proposals.EnrichedCorrectionProposal, 0, len(proposed)) + for i, p := range proposed { + enriched = append(enriched, proposals.EnrichedCorrectionProposal{ + CorrectionProposal: p, + ProposalMetadata: proposals.ProposalMetadata{ + ProposalIndex: i, + ModuleKey: spec.ModuleKey, + ModuleInstance: spec.InstanceName, + }, + }) + } + + applyResult := proposals.ApplyProposals(working, enriched, policy) + working = applyResult.Transcript + + results = append(results, ModuleResult{ + ModuleKey: spec.ModuleKey, + ModuleInstance: spec.InstanceName, + ReplacementPolicy: policy, + Status: ModuleStatusSuccess, + ProposalCount: len(enriched), + AppliedChanges: applyResult.Applied, + SkippedChanges: applyResult.Skipped, + StartedAt: startedAt, + CompletedAt: time.Now().UTC(), + }) + } + + return RunOutput{FinalTranscript: working, ModuleResults: results}, nil +} + +func cloneTranscript(t *schema.Transcript) *schema.Transcript { + if t == nil { + return &schema.Transcript{} + } + segments := make([]schema.Segment, len(t.Segments)) + for i, s := range t.Segments { + var categories []string + if s.Categories != nil { + categories = append([]string(nil), s.Categories...) + } + segments[i] = schema.Segment{ + ID: s.ID, + Speaker: s.Speaker, + Start: s.Start, + End: s.End, + Text: s.Text, + Categories: categories, + } + } + return &schema.Transcript{Segments: segments} +} diff --git a/internal/framework/runner/runner_test.go b/internal/framework/runner/runner_test.go new file mode 100644 index 0000000..eef43ed --- /dev/null +++ b/internal/framework/runner/runner_test.go @@ -0,0 +1,152 @@ +package runner + +import ( + "context" + "errors" + "testing" + + "gitea.maximumdirect.net/eric/audita/internal/core/config" + "gitea.maximumdirect.net/eric/audita/internal/core/schema" + "gitea.maximumdirect.net/eric/audita/internal/framework/contracts" + "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" +) + +type fakeFactory struct { + modules map[string]contracts.TranscriptModule +} + +func (f fakeFactory) ModuleForSpec(spec contracts.ModuleRunSpec) (contracts.TranscriptModule, error) { + m, ok := f.modules[spec.InstanceName] + if !ok { + return nil, errors.New("module not registered") + } + return m, nil +} + +type fakeModule struct { + key string + policy proposals.ReplacementPolicy + proposeF func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) +} + +func (m fakeModule) Key() string { return m.key } +func (m fakeModule) ReplacementPolicy() proposals.ReplacementPolicy { return m.policy } +func (m fakeModule) Validators() []contracts.Validator { return nil } +func (m fakeModule) Propose(ctx context.Context, req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + if m.proposeF == nil { + return nil, nil + } + return m.proposeF(req) +} + +func TestRunnerOneModuleAppliesProposal(t *testing.T) { + transcript := &schema.Transcript{Segments: []schema.Segment{{ID: 1, Speaker: "A", Start: 0, End: 1, Text: "teh cat"}}} + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ + "grammar": fakeModule{key: "grammar", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 0.9}}, nil + }}, + }}) + + out, err := r.Run(context.Background(), RunInput{Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "grammar", InstanceName: "grammar"}}}) + if err != nil { + t.Fatalf("Run error: %v", err) + } + if out.FinalTranscript.Segments[0].Text != "the cat" { + t.Fatalf("expected corrected text, got %q", out.FinalTranscript.Segments[0].Text) + } + if len(out.ModuleResults) != 1 || len(out.ModuleResults[0].AppliedChanges) != 1 { + t.Fatalf("expected one module with one applied change, got %+v", out.ModuleResults) + } +} + +func TestRunnerModulesRunSequentially(t *testing.T) { + transcript := &schema.Transcript{Segments: []schema.Segment{{ID: 1, Speaker: "A", Start: 0, End: 1, Text: "teh cat"}}} + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ + "m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 1}}, nil + }}, + "m2": fakeModule{key: "m2", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + if req.WorkingTranscript.Segments[0].Text != "the cat" { + t.Fatalf("second module did not see first module changes: %q", req.WorkingTranscript.Segments[0].Text) + } + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: 1}}, nil + }}, + }}) + + out, err := r.Run(context.Background(), RunInput{Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m1", InstanceName: "m1"}, {ModuleKey: "m2", InstanceName: "m2"}}}) + if err != nil { + t.Fatalf("Run error: %v", err) + } + if out.FinalTranscript.Segments[0].Text != "the dog" { + t.Fatalf("expected sequential updates, got %q", out.FinalTranscript.Segments[0].Text) + } +} + +func TestRunnerSkippedRecorded(t *testing.T) { + transcript := &schema.Transcript{Segments: []schema.Segment{{ID: 1, Speaker: "A", Start: 0, End: 1, Text: "word word"}}} + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ + "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "word", CorrectedText: "term", Confidence: 1}}, nil + }}, + }}) + out, err := r.Run(context.Background(), RunInput{Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}}) + if err != nil { + t.Fatalf("Run error: %v", err) + } + if len(out.ModuleResults[0].SkippedChanges) != 1 { + t.Fatalf("expected one skipped change, got %+v", out.ModuleResults[0].SkippedChanges) + } +} + +func TestRunnerFailureReturnsPartialProgress(t *testing.T) { + transcript := &schema.Transcript{Segments: []schema.Segment{{ID: 1, Speaker: "A", Start: 0, End: 1, Text: "teh cat"}}} + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ + "m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 1}}, nil + }}, + "m2": fakeModule{key: "m2", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return nil, errors.New("boom") + }}, + }}) + + out, err := r.Run(context.Background(), RunInput{Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m1", InstanceName: "m1"}, {ModuleKey: "m2", InstanceName: "m2"}}}) + if err == nil { + t.Fatal("expected error") + } + if out.FinalTranscript.Segments[0].Text != "the cat" { + t.Fatalf("expected partial transcript progress preserved, got %q", out.FinalTranscript.Segments[0].Text) + } + if len(out.ModuleResults) != 2 || out.ModuleResults[1].Status != ModuleStatusFailed { + t.Fatalf("expected second module failed in results, got %+v", out.ModuleResults) + } +} + +func TestRunnerDoesNotMutateInputTranscript(t *testing.T) { + transcript := &schema.Transcript{Segments: []schema.Segment{{ID: 1, Speaker: "A", Start: 0, End: 1, Text: "teh"}}} + before := transcript.Segments[0].Text + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ + "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 1}}, nil + }}, + }}) + + _, err := r.Run(context.Background(), RunInput{Config: ptrConfig(config.Default()), Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}}) + if err != nil { + t.Fatalf("Run error: %v", err) + } + if transcript.Segments[0].Text != before { + t.Fatalf("expected input transcript unchanged, got %q", transcript.Segments[0].Text) + } +} + +func TestRunnerRepeatedModuleInstanceNames(t *testing.T) { + specs, err := contracts.ResolveModuleRunSpecs([]string{"glossary", "glossary"}) + if err != nil { + t.Fatalf("ResolveModuleRunSpecs error: %v", err) + } + if specs[0].InstanceName != "glossary_1" || specs[1].InstanceName != "glossary_2" { + t.Fatalf("unexpected instance names: %+v", specs) + } +} + +func ptrConfig(c config.Config) *config.Config { return &c }