Implement real Audita polish stage

2026-05-04 08:05:27 -05:00
parent 28c7ab3287
commit de11651528
8 changed files with 677 additions and 22 deletions
--- 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 {
--- 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))
--- 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 {
--- 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 {
--- 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"},
--- 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: &notify.FakeSender{Err: errors.New("nerr")}}, wantErr: "notify"},
--- a/internal/stage/polish.go
+++ 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
+}
--- a/internal/stage/polish_test.go
+++ 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())
+}