diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index ba9de53..b84f093 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -197,7 +197,10 @@ func writeValidConfigFiles(t *testing.T, workspaceRoot string, transcribeURL ... url = transcribeURL[0] } seriatimBinary := writeSeriatimAppTestWrapper(t) + auditaBinary := writeAuditaAppTestWrapper(t) t.Setenv("GO_WANT_APP_SERIATIM_HELPER", "1") + t.Setenv("GO_WANT_APP_AUDITA_HELPER", "1") + t.Setenv("AUDITA_LLM_API_KEY", "test-audita-key") pipelineYAML := `workspace: root: ` + workspaceRoot + ` @@ -216,7 +219,7 @@ seriatim: coalesce_gap: 3.0 report: true audita: - binary: audita + binary: ` + auditaBinary + ` analyzer: timeout: 20m artifacts: @@ -335,6 +338,75 @@ func TestSeriatimAppHelper(t *testing.T) { os.Exit(0) } +func writeAuditaAppTestWrapper(t *testing.T) string { + t.Helper() + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable() error = %v", err) + } + path := filepath.Join(t.TempDir(), "audita-helper-wrapper.sh") + content := "#!/bin/sh\nexec \"" + exe + "\" -test.run=TestAuditaAppHelper -- \"$@\"\n" + if err := os.WriteFile(path, []byte(content), 0o755); err != nil { + t.Fatalf("WriteFile(%q): %v", path, err) + } + return path +} + +func TestAuditaAppHelper(t *testing.T) { + if os.Getenv("GO_WANT_APP_AUDITA_HELPER") != "1" { + return + } + + args := os.Args + start := -1 + for i := range args { + if args[i] == "--" { + start = i + 1 + break + } + } + if start < 0 || start >= len(args) { + _, _ = os.Stderr.WriteString("missing -- args separator\n") + os.Exit(2) + } + processArgs := args[start:] + + outputPath := appSeriatimFlagValue(processArgs, "--output") + reportPath := appSeriatimFlagValue(processArgs, "--report-json") + workDir := appSeriatimFlagValue(processArgs, "--work-dir") + if strings.TrimSpace(outputPath) == "" { + _, _ = os.Stderr.WriteString("missing --output\n") + os.Exit(2) + } + if err := os.MkdirAll(filepath.Dir(outputPath), 0o755); err != nil { + _, _ = os.Stderr.WriteString(fmt.Sprintf("mkdir output dir: %v\n", err)) + os.Exit(2) + } + if err := os.WriteFile(outputPath, []byte(`{"schema":"audita.processed.v1","segments":[]}`), 0o644); err != nil { + _, _ = os.Stderr.WriteString(fmt.Sprintf("write output: %v\n", err)) + os.Exit(2) + } + if strings.TrimSpace(reportPath) != "" { + if err := os.MkdirAll(filepath.Dir(reportPath), 0o755); err != nil { + _, _ = os.Stderr.WriteString(fmt.Sprintf("mkdir report dir: %v\n", err)) + os.Exit(2) + } + if err := os.WriteFile(reportPath, []byte(`{"report":true}`), 0o644); err != nil { + _, _ = os.Stderr.WriteString(fmt.Sprintf("write report: %v\n", err)) + os.Exit(2) + } + } + if strings.TrimSpace(workDir) != "" { + if err := os.MkdirAll(workDir, 0o755); err != nil { + _, _ = os.Stderr.WriteString(fmt.Sprintf("mkdir work dir: %v\n", err)) + os.Exit(2) + } + } + _, _ = os.Stdout.WriteString("audita helper stdout\n") + _, _ = os.Stderr.WriteString("audita helper stderr\n") + os.Exit(0) +} + func appSeriatimFlagValue(args []string, name string) string { for i := 0; i < len(args)-1; i++ { if args[i] == name { diff --git a/internal/app/resume_run_stage_test.go b/internal/app/resume_run_stage_test.go index b7b85b3..2adbe87 100644 --- a/internal/app/resume_run_stage_test.go +++ b/internal/app/resume_run_stage_test.go @@ -29,6 +29,7 @@ func TestResumeStartsAfterCompletedStages(t *testing.T) { mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "raw", "alice.json"), `{"segments":[]}`) mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "speakers.yml"), "match:\n - speaker: Alice\n match: [\"alice\"]\n") mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "autocorrect.yml"), "[]\n") + mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "glossary.yml"), "terms: []\n") var out bytes.Buffer err := Resume(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath}, &out) @@ -105,6 +106,9 @@ func TestRunStageExecutesOnlySelectedStage(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "merged.json"), `{"segments":[]}`) + mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "glossary.yml"), "terms: []\n") var out bytes.Buffer err := RunStage(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath, "polish"}, &out) @@ -132,6 +136,9 @@ func TestRunStageSkipAndForce(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "merged.json"), `{"segments":[]}`) + mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "glossary.yml"), "terms: []\n") store := &manifest.LocalStore{} m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) diff --git a/internal/app/runner.go b/internal/app/runner.go index 64ca76a..f87400d 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -66,7 +66,11 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage env.Seriatim = runner } if env.Audita == nil { - env.Audita = &audita.NoopRunner{} + runner, err := buildDefaultAuditaRunner(env.Config) + if err != nil { + return nil, fmt.Errorf("initialize audita runner: %w", err) + } + env.Audita = runner } if env.Analyzer == nil { env.Analyzer = &analyzer.NoopRunner{} @@ -217,6 +221,40 @@ func buildDefaultSeriatimRunner(cfg *config.Config) (seriatim.Runner, error) { return runner, nil } +func buildDefaultAuditaRunner(cfg *config.Config) (audita.Runner, error) { + if cfg == nil || cfg.Pipeline == nil { + return &audita.NoopRunner{}, nil + } + + a := cfg.Pipeline.Audita + if strings.TrimSpace(a.Binary) == "" || strings.TrimSpace(a.Timeout) == "" || strings.TrimSpace(a.LLMAPIKeyEnv) == "" || len(a.Modules) == 0 || strings.TrimSpace(a.BaseURL) == "" || strings.TrimSpace(a.Model) == "" { + // Compatibility fallback for tests or internal call paths that bypass config validation/defaults. + return &audita.NoopRunner{}, nil + } + + report := false + if a.Report != nil { + report = *a.Report + } + + runner, err := audita.NewSubprocessRunnerFromConfigValues( + a.Binary, + a.Timeout, + a.LLMAPIKeyEnv, + append([]string(nil), a.Modules...), + a.BaseURL, + a.Model, + a.LLMConcurrency, + a.ValidationModel, + a.ValidationLLMConcurrency, + report, + ) + if err != nil { + return nil, fmt.Errorf("from pipeline.audita: %w", err) + } + return runner, nil +} + func loadOrCreateManifest(ctx context.Context, store manifest.Store, path, sessionID string) (*manifest.Manifest, error) { exists, err := fileExists(path) if err != nil { diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 364c431..9ef3c03 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -99,6 +99,21 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { } continue } + if name == "polish" { + if sr.Metadata == nil || sr.Metadata["stage"] != "polish" { + t.Fatalf("polish metadata missing stage=polish: %#v", sr.Metadata) + } + if len(sr.Outputs) == 0 { + t.Fatalf("polish outputs missing") + } + if len(sr.Logs) == 0 { + t.Fatalf("polish logs missing") + } + if len(sr.GeneratedConfigs) == 0 { + t.Fatalf("polish generated configs missing") + } + continue + } if sr.Metadata == nil || sr.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", name) } @@ -320,6 +335,18 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { t.Fatalf("write autocorrect: %v", err) } } + if tc.name == "polish" { + paths, ensureErr := artifactStore.EnsureLayout(cfg.Session.SessionID) + if ensureErr != nil { + t.Fatalf("EnsureLayout() error = %v", ensureErr) + } + if err := os.WriteFile(filepath.Join(paths.TranscriptsDir, "merged.json"), []byte(`{"segments":[]}`), 0o644); err != nil { + t.Fatalf("write merged transcript: %v", err) + } + if err := os.WriteFile(filepath.Join(paths.InputsDir, "glossary.yml"), []byte("terms: []\n"), 0o644); err != nil { + t.Fatalf("write glossary: %v", err) + } + } _, runErr := executeStages(context.Background(), cfg, []stage.Stage{selected}, RunOptions{Env: tc.env}) if runErr == nil { diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index 8d29aff..16b35b2 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -6,7 +6,6 @@ import ( "path/filepath" "gitea.maximumdirect.net/eric/narratio/internal/adapters/analyzer" - "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" @@ -52,23 +51,6 @@ func (s placeholderStage) Run(ctx context.Context, env *Env, m *manifest.Manifes paths := env.ArtifactStore.SessionPaths(sessionID) switch s.name { - case "polish": - if env.Audita != nil { - req := audita.PolishRequest{ - GeneratedConfigPath: filepath.Join(paths.ConfigDir, "audita.generated.yml"), - MergedTranscriptPath: filepath.Join(paths.TranscriptsDir, "merged.json"), - OutputProcessedPath: filepath.Join(paths.TranscriptsDir, "processed.json"), - StdoutLogPath: filepath.Join(paths.LogsDir, "audita.stdout.log"), - StderrLogPath: filepath.Join(paths.LogsDir, "audita.stderr.log"), - } - resp, err := env.Audita.Run(ctx, req) - if err != nil { - return nil, fmt.Errorf("placeholder polish adapter call failed: %w", err) - } - result.Outputs = append(result.Outputs, artifacts.Ref{Kind: "transcript_processed", Category: "transcripts", SessionID: sessionID, AbsolutePath: resp.ProcessedTranscriptPath}) - result.Logs = append(result.Logs, req.StdoutLogPath, req.StderrLogPath) - result.GeneratedConfigs = append(result.GeneratedConfigs, req.GeneratedConfigPath) - } case "analyze": if env.Analyzer != nil { req := analyzer.AnalyzeRequest{ @@ -130,7 +112,7 @@ func All() []Stage { transcribeStage{}, placeholderStage{name: "normalize"}, mergeStage{}, - placeholderStage{name: "polish"}, + polishStage{}, placeholderStage{name: "analyze"}, placeholderStage{name: "archive"}, placeholderStage{name: "notify"}, diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 2bc7151..8f0bc3d 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -102,6 +102,15 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } continue } + if s.Name() == "polish" { + if result.Metadata["stage"] != "polish" { + t.Fatalf("polish metadata = %#v, want stage=polish", result.Metadata) + } + if len(result.Outputs) == 0 { + t.Fatalf("polish outputs = %#v, want non-empty", result.Outputs) + } + continue + } if result.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", s.Name()) } @@ -133,7 +142,6 @@ func TestPlaceholderAdapterErrorPropagation(t *testing.T) { env *Env wantErr string }{ - {stageName: "polish", env: &Env{Audita: &audita.FakeRunner{Err: errors.New("aerr")}}, wantErr: "polish"}, {stageName: "analyze", env: &Env{Analyzer: &analyzer.FakeRunner{Err: errors.New("anerr")}}, wantErr: "analyze"}, {stageName: "archive", env: &Env{Storage: &storage.FakeBackend{Err: errors.New("sterr")}}, wantErr: "archive"}, {stageName: "notify", env: &Env{Notifier: ¬ify.FakeSender{Err: errors.New("nerr")}}, wantErr: "notify"}, diff --git a/internal/stage/polish.go b/internal/stage/polish.go new file mode 100644 index 0000000..60de37e --- /dev/null +++ b/internal/stage/polish.go @@ -0,0 +1,260 @@ +package stage + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +type polishStage struct{} + +func (polishStage) Name() string { return "polish" } + +func (polishStage) Declares() IODecl { + return IODecl{ + Inputs: []artifacts.Ref{ + {Kind: "transcript_merged", Category: "transcripts", RelativePath: "transcripts/merged.json"}, + {Kind: "glossary", Category: "inputs", RelativePath: "inputs/glossary.yml"}, + }, + Outputs: []artifacts.Ref{ + {Kind: "transcript_processed", Category: "transcripts", RelativePath: "transcripts/processed.json"}, + {Kind: "audita_report", Category: "artifacts", RelativePath: "artifacts/audita.report.json"}, + }, + } +} + +func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { + if env == nil || env.Config == nil { + return nil, fmt.Errorf("polish: stage environment config is required") + } + if env.ArtifactStore == nil { + return nil, fmt.Errorf("polish: artifact store is required") + } + if env.Config.Pipeline == nil || env.Config.Session == nil { + return nil, fmt.Errorf("polish: resolved config must include pipeline and session") + } + if env.Audita == nil { + return nil, fmt.Errorf("polish: audita adapter is required") + } + + var sessionID string + if m != nil { + sessionID = strings.TrimSpace(m.SessionID) + } + if sessionID == "" { + sessionID = strings.TrimSpace(env.Config.Session.SessionID) + } + if sessionID == "" { + return nil, fmt.Errorf("polish: session id is required") + } + + paths := env.ArtifactStore.SessionPaths(sessionID) + mergedPath, source, err := discoverMergedTranscript(m, paths) + if err != nil { + return nil, fmt.Errorf("polish: resolve merged transcript: %w", err) + } + if mergedPath == "" { + return nil, fmt.Errorf("polish: merged transcript input is required") + } + if err := validateTranscriptJSONFile(mergedPath); err != nil { + return nil, fmt.Errorf("polish: merged transcript %q invalid: %w", mergedPath, err) + } + + glossaryPath := filepath.Join(paths.InputsDir, "glossary.yml") + if err := requireFile(glossaryPath, "glossary.yml"); err != nil { + return nil, fmt.Errorf("polish: %w", err) + } + + processedPath := filepath.Join(paths.TranscriptsDir, "processed.json") + reportPath := filepath.Join(paths.ArtifactsDir, "audita.report.json") + workDir := filepath.Join(paths.ArtifactsDir, "audita-work") + stdoutPath := filepath.Join(paths.LogsDir, "audita.stdout.log") + stderrPath := filepath.Join(paths.LogsDir, "audita.stderr.log") + generatedConfigPath := filepath.Join(paths.ConfigDir, "audita.generated.yml") + + reportEnabled := env.Config.Pipeline.Audita.Report != nil && *env.Config.Pipeline.Audita.Report + req := audita.PolishRequest{ + GeneratedConfigPath: generatedConfigPath, + MergedTranscriptPath: mergedPath, + OutputProcessedPath: processedPath, + GlossaryPath: glossaryPath, + ReportPath: "", + WorkDir: workDir, + Modules: append([]string(nil), env.Config.Pipeline.Audita.Modules...), + BaseURL: env.Config.Pipeline.Audita.BaseURL, + Model: env.Config.Pipeline.Audita.Model, + ValidationModel: env.Config.Pipeline.Audita.ValidationModel, + ValidationLLMConcurrency: env.Config.Pipeline.Audita.ValidationLLMConcurrency, + StdoutLogPath: stdoutPath, + StderrLogPath: stderrPath, + } + if reportEnabled { + req.ReportPath = reportPath + } + + res, err := env.Audita.Run(ctx, req) + if err != nil { + return nil, fmt.Errorf("polish: audita polish failed: %w", err) + } + + finalProcessedPath := processedPath + if strings.TrimSpace(res.ProcessedTranscriptPath) != "" { + finalProcessedPath = res.ProcessedTranscriptPath + } + if err := validateProcessedTranscriptOutput(finalProcessedPath); err != nil { + return nil, fmt.Errorf("polish: processed transcript %q invalid: %w", finalProcessedPath, err) + } + + finalReportPath := req.ReportPath + if strings.TrimSpace(res.ReportPath) != "" { + finalReportPath = res.ReportPath + } + if reportEnabled { + if err := validateTranscriptJSONFile(finalReportPath); err != nil { + return nil, fmt.Errorf("polish: report %q invalid: %w", finalReportPath, err) + } + } + + outputs := []artifacts.Ref{{ + Kind: "transcript_processed", + Category: "transcripts", + SessionID: sessionID, + AbsolutePath: finalProcessedPath, + }} + if reportEnabled { + outputs = append(outputs, artifacts.Ref{ + Kind: "audita_report", + Category: "artifacts", + SessionID: sessionID, + AbsolutePath: finalReportPath, + }) + } + + var validationConcurrency any + if env.Config.Pipeline.Audita.ValidationLLMConcurrency != nil { + validationConcurrency = *env.Config.Pipeline.Audita.ValidationLLMConcurrency + } + var llmConcurrency any + if env.Config.Pipeline.Audita.LLMConcurrency != nil { + llmConcurrency = *env.Config.Pipeline.Audita.LLMConcurrency + } + + meta := map[string]any{ + "stage": "polish", + "merged_transcript_path": mergedPath, + "merged_transcript_source": source, + "glossary_path": glossaryPath, + "output_path": finalProcessedPath, + "report_path": finalReportPath, + "audita_work_dir": workDir, + "report_enabled": reportEnabled, + "modules": append([]string(nil), req.Modules...), + "base_url": req.BaseURL, + "model": req.Model, + "validation_model": req.ValidationModel, + "llm_concurrency": llmConcurrency, + "validation_llm_concurrency": validationConcurrency, + "llm_api_key_env": env.Config.Pipeline.Audita.LLMAPIKeyEnv, + "timeout": env.Config.Pipeline.Audita.Timeout, + "binary": env.Config.Pipeline.Audita.Binary, + "generated_config_path": generatedConfigPath, + "stdout_log_path": stdoutPath, + "stderr_log_path": stderrPath, + "adapter_duration_ms": res.Duration.Milliseconds(), + "adapter_exit_code": res.ExitCode, + "adapter_invoked_binary": res.InvokedBinary, + "adapter_processed_output_path": res.ProcessedTranscriptPath, + "adapter_report_path": res.ReportPath, + "adapter_generated_config_path": res.GeneratedConfigPath, + "adapter_work_dir": res.WorkDir, + "adapter_stdout_log_path": res.StdoutLogPath, + "adapter_stderr_log_path": res.StderrLogPath, + "credential_env_var": env.Config.Pipeline.Audita.LLMAPIKeyEnv, + "credential_present": false, + "primary_llm_concurrency_via_env": false, + } + if res.Metadata != nil { + meta["adapter_metadata"] = res.Metadata + if value, ok := res.Metadata["credential_present"]; ok { + meta["credential_present"] = value + } + if value, ok := res.Metadata["credential_env_var"]; ok { + meta["credential_env_var"] = value + } + if value, ok := res.Metadata["primary_llm_concurrency_via_env"]; ok { + meta["primary_llm_concurrency_via_env"] = value + } + } + + return &StageResult{ + Outputs: outputs, + Logs: []string{stdoutPath, stderrPath}, + GeneratedConfigs: []string{generatedConfigPath}, + Metadata: meta, + }, nil +} + +func discoverMergedTranscript(m *manifest.Manifest, paths artifacts.SessionPaths) (string, string, error) { + candidates := make([]string, 0) + if m != nil && m.Stages != nil { + if sr := m.Stages["merge"]; sr != nil { + for _, out := range sr.Outputs { + if out.Kind != "transcript_merged" { + continue + } + p := strings.TrimSpace(out.LocalPath) + if p == "" { + continue + } + resolved := artifacts.ResolveSessionLocalPathForRead(paths, p) + candidates = append(candidates, filepath.Clean(resolved)) + } + } + } + + deduped := dedupeAndSortPaths(candidates) + for _, p := range deduped { + if info, err := os.Stat(p); err == nil && !info.IsDir() { + return p, "manifest.merge.outputs", nil + } + } + + fallback := filepath.Join(paths.TranscriptsDir, "merged.json") + if info, err := os.Stat(fallback); err == nil && !info.IsDir() { + return filepath.Clean(fallback), "fallback.transcripts_dir", nil + } + if len(deduped) > 0 { + return deduped[0], "manifest.merge.outputs", nil + } + return "", "", nil +} + +func validateProcessedTranscriptOutput(path string) error { + if err := requireFile(path, "processed transcript"); err != nil { + return err + } + data, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read processed transcript: %w", err) + } + + var payload map[string]any + if err := json.Unmarshal(data, &payload); err != nil { + return fmt.Errorf("decode json: %w", err) + } + segments, ok := payload["segments"] + if !ok { + return fmt.Errorf("top-level segments is required") + } + if _, ok := segments.([]any); !ok { + return fmt.Errorf("top-level segments must be an array") + } + return nil +} diff --git a/internal/stage/polish_test.go b/internal/stage/polish_test.go new file mode 100644 index 0000000..efc2910 --- /dev/null +++ b/internal/stage/polish_test.go @@ -0,0 +1,261 @@ +package stage + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestPolishStagePolishesMergedTranscriptAndRecordsMetadata(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + + mergedPath := filepath.Join(paths.TranscriptsDir, "merged.json") + writeFile(t, mergedPath, `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + m.MarkStageSucceeded("merge", time.Now().UTC(), []manifest.ArtifactRecord{ + {Kind: "transcript_merged", LocalPath: mergedPath}, + }) + + fake := &audita.FakeRunner{} + env.Audita = fake + + result, err := (polishStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("polish.Run() error = %v", err) + } + if result == nil { + t.Fatal("polish result is nil") + } + if len(fake.Requests) != 1 { + t.Fatalf("audita requests = %d, want 1", len(fake.Requests)) + } + req := fake.Requests[0] + if req.MergedTranscriptPath != mergedPath { + t.Fatalf("merged path = %q, want %q", req.MergedTranscriptPath, mergedPath) + } + if req.GlossaryPath != filepath.Join(paths.InputsDir, "glossary.yml") { + t.Fatalf("glossary path = %q", req.GlossaryPath) + } + if req.WorkDir != filepath.Join(paths.ArtifactsDir, "audita-work") { + t.Fatalf("work dir = %q", req.WorkDir) + } + if strings.Join(req.Modules, ",") != "glossary,homophones,grammar" { + t.Fatalf("modules = %#v", req.Modules) + } + if req.BaseURL != "https://openrouter.ai/api/v1" { + t.Fatalf("base url = %q", req.BaseURL) + } + if req.Model != "openrouter/google/gemma-4-31b-it" { + t.Fatalf("model = %q", req.Model) + } + if req.ValidationModel != "openrouter/google/gemma-4-31b-it" { + t.Fatalf("validation model = %q", req.ValidationModel) + } + if req.ValidationLLMConcurrency == nil || *req.ValidationLLMConcurrency != 2 { + t.Fatalf("validation llm concurrency = %#v, want 2", req.ValidationLLMConcurrency) + } + + if len(result.Outputs) != 2 { + t.Fatalf("outputs len = %d, want 2", len(result.Outputs)) + } + if result.Outputs[0].Kind != "transcript_processed" { + t.Fatalf("output[0] kind = %q, want transcript_processed", result.Outputs[0].Kind) + } + if result.Outputs[1].Kind != "audita_report" { + t.Fatalf("output[1] kind = %q, want audita_report", result.Outputs[1].Kind) + } + if len(result.Logs) != 2 { + t.Fatalf("logs = %#v, want 2", result.Logs) + } + if len(result.GeneratedConfigs) != 1 { + t.Fatalf("generated configs = %#v, want 1", result.GeneratedConfigs) + } + + if result.Metadata["stage"] != "polish" { + t.Fatalf("metadata stage = %#v, want polish", result.Metadata["stage"]) + } + if result.Metadata["report_enabled"] != true { + t.Fatalf("metadata report_enabled = %#v, want true", result.Metadata["report_enabled"]) + } + if result.Metadata["audita_work_dir"] != filepath.Join(paths.ArtifactsDir, "audita-work") { + t.Fatalf("metadata audita_work_dir = %#v", result.Metadata["audita_work_dir"]) + } +} + +func TestPolishStageFallsBackToMergedTranscriptPath(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + mergedPath := filepath.Join(paths.TranscriptsDir, "merged.json") + writeFile(t, mergedPath, `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + + fake := &audita.FakeRunner{} + env.Audita = fake + + _, err := (polishStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("polish.Run() error = %v", err) + } + if len(fake.Requests) != 1 { + t.Fatalf("requests = %d, want 1", len(fake.Requests)) + } + if fake.Requests[0].MergedTranscriptPath != mergedPath { + t.Fatalf("merged path = %q, want %q", fake.Requests[0].MergedTranscriptPath, mergedPath) + } +} + +func TestPolishStageFailsWhenMergedTranscriptMissing(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + env.Audita = &audita.FakeRunner{} + + _, err := (polishStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "merged transcript input is required") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestPolishStageFailsWhenMergedTranscriptInvalidJSON(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), "not-json") + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + env.Audita = &audita.FakeRunner{} + + _, err := (polishStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(strings.ToLower(err.Error()), "invalid") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestPolishStageFailsWhenGlossaryMissing(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) + env.Audita = &audita.FakeRunner{} + + _, err := (polishStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "glossary.yml") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestPolishStageFailsWhenAdapterFails(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + env.Audita = &audita.FakeRunner{Err: errors.New("audita failed")} + + _, err := (polishStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "audita polish failed") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestPolishStageFailsWhenProcessedOutputInvalid(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + badPath := filepath.Join(paths.TranscriptsDir, "processed.invalid.json") + writeFile(t, badPath, `{"schema":"audita.processed.v1","segments":"wrong-type"}`) + env.Audita = &audita.FakeRunner{Result: audita.PolishResult{ProcessedTranscriptPath: badPath}} + + _, err := (polishStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "processed transcript") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestPolishStageFailsWhenReportInvalid(t *testing.T) { + env, m := setupPolishEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + badReport := filepath.Join(paths.ArtifactsDir, "bad.report.json") + writeFile(t, badReport, "not-json") + env.Audita = &audita.FakeRunner{Result: audita.PolishResult{ReportPath: badReport}} + + _, err := (polishStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "report") { + t.Fatalf("error = %q", err.Error()) + } +} + +func setupPolishEnv(t *testing.T) (*Env, *manifest.Manifest) { + t.Helper() + workspace := t.TempDir() + cfgDir := t.TempDir() + sessionPath := filepath.Join(cfgDir, "session.yml") + pipelinePath := filepath.Join(cfgDir, "pipeline.yml") + writeFile(t, sessionPath, "session_id: 2026-05-03\n") + writeFile(t, pipelinePath, "workspace:\n root: "+workspace+"\n") + + report := true + llmConcurrency := 1 + validationLLMConcurrency := 2 + + cfg := &config.Config{ + PipelinePath: pipelinePath, + SessionPath: sessionPath, + Pipeline: &config.PipelineConfig{ + Workspace: config.WorkspaceConfig{Root: workspace}, + Audita: config.AuditaConfig{ + Binary: "audita", + Timeout: "3h", + LLMAPIKeyEnv: "AUDITA_LLM_API_KEY", + Modules: []string{"glossary", "homophones", "grammar"}, + BaseURL: "https://openrouter.ai/api/v1", + Model: "openrouter/google/gemma-4-31b-it", + ValidationModel: "openrouter/google/gemma-4-31b-it", + LLMConcurrency: &llmConcurrency, + ValidationLLMConcurrency: &validationLLMConcurrency, + Report: &report, + }, + }, + Session: &config.SessionConfig{ + SessionID: "2026-05-03", + Inputs: config.SessionInputsConfig{ + GlossaryFile: "./glossary.yml", + }, + }, + } + + store := artifacts.NewLocalStore(workspace) + if _, err := store.EnsureLayout("2026-05-03"); err != nil { + t.Fatalf("EnsureLayout() error = %v", err) + } + return &Env{ + Config: cfg, + ArtifactStore: store, + }, manifest.New("2026-05-03", time.Now().UTC()) +}