Implement real normalize stage

2026-05-10 21:40:31 +00:00
parent 5e2234e418
commit 9392da2615
5 changed files with 378 additions and 28 deletions
--- a/README.md
+++ b/README.md
@@ -9,8 +9,7 @@ Implemented now:
 - strict config loading/validation (`pipeline.yml` and `session.yml`)
 - local workspace/session layout, locking, and manifest persistence
 - resumable stage control (`run`, `plan`, `resume`, `run-stage`, `status`)
- real `prepare`, `transcribe`, `merge`, `polish`, `trim`, and `analyze` stages
- placeholder `normalize` stage
+- real `prepare`, `transcribe`, `merge`, `polish`, `normalize`, `trim`, and `analyze` stages
 - real WhisperX, Seriatim, and Audita adapters
 - real Scriptorium subprocess adapter
 - optional Scriptorium render diagnostics (`render_debug`)
@@ -47,7 +46,7 @@ YAML decoding is strict (`KnownFields(true)`), so unknown fields fail fast.

 - `transcripts/merged.json`: canonical deterministic merged transcript from Seriatim merge
 - `transcripts/processed.json`: full polished transcript from Audita polish
- `transcripts/normalized.json`: placeholder normalized transcript (Seriatim schema pass-through for now)
+- `transcripts/normalized.json`: Seriatim-normalized transcript from the normalize stage
 - `transcripts/trimmed.json`: gameplay-only polished transcript from trim stage

 ## Normalize Configuration
--- a/architecture.md
+++ b/architecture.md
@@ -18,7 +18,7 @@ Implemented:
 - real Seriatim subprocess adapter
 - real Audita subprocess adapter
 - real Scriptorium subprocess adapter
- placeholder `normalize` stage producing `transcripts/normalized.json`
+- real `normalize` stage producing `transcripts/normalized.json`
 - real `trim` stage producing `transcripts/trimmed.json`
 - real `analyze` stage for initial `session_recap` generation
 - optional Scriptorium render diagnostics (`render_debug`) before production run
@@ -176,7 +176,7 @@ Narratio currently produces and uses four transcript tiers:

 - `transcripts/merged.json`: canonical deterministic merged transcript from Seriatim merge
 - `transcripts/processed.json`: full polished transcript from Audita polish (includes pre/post-game content)
- `transcripts/normalized.json`: placeholder normalized transcript (Seriatim-compatible schema pass-through for now)
+- `transcripts/normalized.json`: normalized transcript generated by Seriatim normalize
 - `transcripts/trimmed.json`: gameplay-only polished transcript from trim stage

 Trim reads `transcripts/normalized.json`, validates bounds IDs against that same transcript ID space, and writes `transcripts/trimmed.json`.
--- a/internal/stage/normalize.go
+++ b/internal/stage/normalize.go
@@ -7,7 +7,9 @@ import (
 	"path/filepath"
 	"strings"

+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
 	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
 )

@@ -22,11 +24,12 @@ func (normalizeStage) Declares() IODecl {
 		},
 		Outputs: []artifacts.Ref{
 			{Kind: "transcript_normalized", Category: "transcripts", RelativePath: "transcripts/normalized.json"},
+			{Kind: "seriatim_normalize_report", Category: "artifacts", RelativePath: "artifacts/seriatim.normalize.report.json"},
 		},
 	}
 }

-func (normalizeStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
+func (normalizeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
 	if env == nil || env.Config == nil {
 		return nil, fmt.Errorf("normalize: stage environment config is required")
 	}
@@ -36,6 +39,9 @@ func (normalizeStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*S
 	if env.Config.Pipeline == nil || env.Config.Session == nil {
 		return nil, fmt.Errorf("normalize: resolved config must include pipeline and session")
 	}
+	if env.Seriatim == nil {
+		return nil, fmt.Errorf("normalize: seriatim adapter is required")
+	}

 	var sessionID string
 	if m != nil {
@@ -60,33 +66,116 @@ func (normalizeStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*S
 		return nil, fmt.Errorf("normalize: processed transcript %q invalid: %w", processedPath, err)
 	}

-	normalizedPath := filepath.Join(paths.TranscriptsDir, "normalized.json")
-	if err := copyTranscript(env.ArtifactStore, processedPath, normalizedPath); err != nil {
-		return nil, fmt.Errorf("normalize: copy processed transcript to normalized output: %w", err)
+	normalizeCfg := normalizeConfigOrDefault(env.Config.Pipeline.Normalize)
+	normalizedPath, err := resolveScriptoriumOutputPath(paths, normalizeCfg.OutputPath)
+	if err != nil {
+		return nil, fmt.Errorf("normalize: resolve normalized output path: %w", err)
 	}
-	if err := validateProcessedTranscriptOutput(normalizedPath); err != nil {
-		return nil, fmt.Errorf("normalize: normalized transcript %q invalid: %w", normalizedPath, err)
+	reportEnabled := normalizeCfg.Report != nil && *normalizeCfg.Report
+	reportPath := ""
+	if reportEnabled {
+		reportPath = filepath.Join(paths.ArtifactsDir, "seriatim.normalize.report.json")
+	}
+	stdoutPath := filepath.Join(paths.LogsDir, "seriatim.normalize.stdout.log")
+	stderrPath := filepath.Join(paths.LogsDir, "seriatim.normalize.stderr.log")
+	generatedConfigPath := filepath.Join(paths.ConfigDir, "seriatim.normalize.generated.yml")
+	timeout, err := resolveTrimSeriatimTimeout(env.Config.Pipeline.Seriatim.Timeout)
+	if err != nil {
+		return nil, fmt.Errorf("normalize: resolve seriatim timeout: %w", err)
+	}
+
+	req := seriatim.NormalizeRequest{
+		Binary:               env.Config.Pipeline.Seriatim.Binary,
+		InputTranscriptPath:  processedPath,
+		OutputNormalizedPath: normalizedPath,
+		OutputSchema:         normalizeCfg.OutputSchema,
+		ReportPath:           reportPath,
+		StdoutLogPath:        stdoutPath,
+		StderrLogPath:        stderrPath,
+		GeneratedConfigPath:  generatedConfigPath,
+		Timeout:              timeout,
+	}
+
+	res, err := env.Seriatim.Normalize(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("normalize: seriatim normalize failed: %w", err)
+	}
+
+	finalNormalizedPath := coalesceString(res.OutputNormalizedPath, req.OutputNormalizedPath)
+	if err := validateProcessedTranscriptOutput(finalNormalizedPath); err != nil {
+		return nil, fmt.Errorf("normalize: normalized transcript %q invalid: %w", finalNormalizedPath, err)
+	}
+
+	finalReportPath := coalesceString(res.ReportPath, req.ReportPath)
+	if reportEnabled {
+		if err := validateJSONFile(finalReportPath); err != nil {
+			return nil, fmt.Errorf("normalize: report %q invalid: %w", finalReportPath, err)
+		}
+	}
+
+	outputs := []artifacts.Ref{{
+		Kind:         "transcript_normalized",
+		Category:     "transcripts",
+		SessionID:    sessionID,
+		AbsolutePath: finalNormalizedPath,
+	}}
+	if reportEnabled {
+		outputs = append(outputs, artifacts.Ref{
+			Kind:         "seriatim_normalize_report",
+			Category:     "artifacts",
+			SessionID:    sessionID,
+			AbsolutePath: finalReportPath,
+		})
+	}
+
+	meta := map[string]any{
+		"stage":                        "normalize",
+		"processed_transcript_path":    processedPath,
+		"processed_transcript_source":  processedSource,
+		"normalized_transcript_path":   finalNormalizedPath,
+		"normalized_transcript_source": "stage.normalize.output",
+		"output_schema":                normalizeCfg.OutputSchema,
+		"report_enabled":               reportEnabled,
+		"report_path":                  finalReportPath,
+		"timeout":                      env.Config.Pipeline.Seriatim.Timeout,
+		"binary":                       env.Config.Pipeline.Seriatim.Binary,
+		"stdout_log_path":              stdoutPath,
+		"stderr_log_path":              stderrPath,
+		"generated_config_path":        generatedConfigPath,
+		"adapter_duration_ms":          res.Duration.Milliseconds(),
+		"adapter_exit_code":            res.ExitCode,
+		"adapter_invoked_binary":       res.InvokedBinary,
+		"adapter_output_schema":        res.OutputSchema,
+		"adapter_output_path":          res.OutputNormalizedPath,
+		"adapter_report_path":          res.ReportPath,
+		"adapter_generated_config":     res.GeneratedConfigPath,
+		"adapter_stdout_log_path":      res.StdoutLogPath,
+		"adapter_stderr_log_path":      res.StderrLogPath,
+	}
+	if res.Metadata != nil {
+		meta["adapter_metadata"] = res.Metadata
 	}

 	return &StageResult{
-		Outputs: []artifacts.Ref{{
-			Kind:         "transcript_normalized",
-			Category:     "transcripts",
-			SessionID:    sessionID,
-			AbsolutePath: normalizedPath,
-		}},
-		Metadata: map[string]any{
-			"stage":                        "normalize",
-			"placeholder":                  true,
-			"normalize_action":             "copy_placeholder",
-			"processed_transcript_path":    processedPath,
-			"processed_transcript_source":  processedSource,
-			"normalized_transcript_path":   normalizedPath,
-			"normalized_transcript_source": "stage.normalize.output",
-		},
+		Outputs:          outputs,
+		Logs:             []string{stdoutPath, stderrPath},
+		GeneratedConfigs: []string{generatedConfigPath},
+		Metadata:         meta,
 	}, nil
 }

+func normalizeConfigOrDefault(cfg *config.NormalizeConfig) *config.NormalizeConfig {
+	if cfg != nil {
+		return cfg
+	}
+	report := true
+	return &config.NormalizeConfig{
+		OutputPath:   "transcripts/normalized.json",
+		OutputSchema: "seriatim-intermediate",
+		Report:       &report,
+	}
+}
+
 func discoverNormalizedTranscript(m *manifest.Manifest, paths artifacts.SessionPaths) (string, string, error) {
 	candidates := []string{}
 	if m != nil && m.Stages != nil {
--- a/internal/stage/normalize_test.go
+++ b/internal/stage/normalize_test.go
@@ -0,0 +1,256 @@
+package stage
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+func TestNormalizeStageConsumesProcessedTranscriptFromManifest(t *testing.T) {
+	env, m, ser := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	manifestProcessed := filepath.Join(paths.ArtifactsDir, "processed.from-manifest.json")
+	writeFile(t, manifestProcessed, `{"segments":[{"id":10}]}`)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":99}]}`)
+	m.MarkStageSucceeded("polish", time.Now().UTC(), []manifest.ArtifactRecord{{
+		Kind:      "transcript_processed",
+		LocalPath: manifestProcessed,
+	}})
+
+	result, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("normalize.Run() error = %v", err)
+	}
+	if len(ser.NormalizeRequests) != 1 {
+		t.Fatalf("normalize requests = %d, want 1", len(ser.NormalizeRequests))
+	}
+	req := ser.NormalizeRequests[0]
+	if req.InputTranscriptPath != manifestProcessed {
+		t.Fatalf("input transcript = %q, want %q", req.InputTranscriptPath, manifestProcessed)
+	}
+	if result.Metadata["processed_transcript_source"] != "manifest.polish.outputs" {
+		t.Fatalf("processed source = %#v, want manifest.polish.outputs", result.Metadata["processed_transcript_source"])
+	}
+}
+
+func TestNormalizeStageFallsBackToProcessedTranscriptPath(t *testing.T) {
+	env, m, ser := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	fallback := filepath.Join(paths.TranscriptsDir, "processed.json")
+	writeFile(t, fallback, `{"segments":[{"id":1}]}`)
+
+	_, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("normalize.Run() error = %v", err)
+	}
+	if len(ser.NormalizeRequests) != 1 {
+		t.Fatalf("normalize requests = %d, want 1", len(ser.NormalizeRequests))
+	}
+	if ser.NormalizeRequests[0].InputTranscriptPath != fallback {
+		t.Fatalf("input transcript = %q, want %q", ser.NormalizeRequests[0].InputTranscriptPath, fallback)
+	}
+}
+
+func TestNormalizeStageFailsWhenProcessedTranscriptMissing(t *testing.T) {
+	env, m, _ := setupNormalizeEnv(t)
+	_, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "processed transcript input is required") {
+		t.Fatalf("error = %q", err.Error())
+	}
+}
+
+func TestNormalizeStageFailsWhenProcessedTranscriptInvalidJSON(t *testing.T) {
+	env, m, _ := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), "not-json")
+
+	_, err := (normalizeStage{}).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 TestNormalizeStageFailsWhenProcessedTranscriptMissingSegments(t *testing.T) {
+	env, m, _ := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"schema":"audita.processed.v1"}`)
+
+	_, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "segments") {
+		t.Fatalf("error = %q", err.Error())
+	}
+}
+
+func TestNormalizeStagePassesConfiguredOutputSchemaToAdapter(t *testing.T) {
+	env, m, ser := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`)
+	env.Config.Pipeline.Normalize.OutputSchema = "seriatim-full"
+
+	result, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("normalize.Run() error = %v", err)
+	}
+	if len(ser.NormalizeRequests) != 1 {
+		t.Fatalf("normalize requests = %d, want 1", len(ser.NormalizeRequests))
+	}
+	if ser.NormalizeRequests[0].OutputSchema != "seriatim-full" {
+		t.Fatalf("output schema = %q, want seriatim-full", ser.NormalizeRequests[0].OutputSchema)
+	}
+	if result.Metadata["output_schema"] != "seriatim-full" {
+		t.Fatalf("metadata output_schema = %#v, want seriatim-full", result.Metadata["output_schema"])
+	}
+}
+
+func TestNormalizeStageRecordsNormalizedTranscriptOutputKind(t *testing.T) {
+	env, m, _ := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`)
+
+	result, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("normalize.Run() error = %v", err)
+	}
+	if len(result.Outputs) == 0 {
+		t.Fatalf("outputs = %#v, want at least transcript_normalized", result.Outputs)
+	}
+	if result.Outputs[0].Kind != "transcript_normalized" {
+		t.Fatalf("outputs[0].kind = %q, want transcript_normalized", result.Outputs[0].Kind)
+	}
+}
+
+func TestNormalizeStageRecordsReportLogAndGeneratedConfigRefs(t *testing.T) {
+	env, m, _ := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`)
+	report := true
+	env.Config.Pipeline.Normalize.Report = &report
+
+	result, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("normalize.Run() error = %v", err)
+	}
+	if len(result.Outputs) < 2 {
+		t.Fatalf("outputs = %#v, want transcript_normalized + report", result.Outputs)
+	}
+	if result.Outputs[1].Kind != "seriatim_normalize_report" {
+		t.Fatalf("outputs[1].kind = %q, want seriatim_normalize_report", result.Outputs[1].Kind)
+	}
+	logs := strings.Join(result.Logs, "\n")
+	if !strings.Contains(logs, "seriatim.normalize.stdout.log") || !strings.Contains(logs, "seriatim.normalize.stderr.log") {
+		t.Fatalf("logs = %#v, want normalize stdout/stderr", result.Logs)
+	}
+	if len(result.GeneratedConfigs) != 1 || !strings.Contains(result.GeneratedConfigs[0], "seriatim.normalize.generated.yml") {
+		t.Fatalf("generated configs = %#v, want seriatim.normalize.generated.yml", result.GeneratedConfigs)
+	}
+}
+
+func TestNormalizeStageFailsWhenAdapterReturnsError(t *testing.T) {
+	env, m, ser := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`)
+	ser.NormalizeErr = errors.New("normalize failed")
+
+	_, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "seriatim normalize failed") {
+		t.Fatalf("error = %q", err.Error())
+	}
+}
+
+func TestNormalizeStageFailsWhenNormalizedOutputInvalid(t *testing.T) {
+	env, m, ser := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`)
+	badOutput := filepath.Join(paths.TranscriptsDir, "normalized.bad.json")
+	writeFile(t, badOutput, "not-json")
+	ser.NormalizeResult = seriatim.NormalizeResult{OutputNormalizedPath: badOutput}
+
+	_, err := (normalizeStage{}).Run(context.Background(), env, m)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "normalized transcript") {
+		t.Fatalf("error = %q", err.Error())
+	}
+}
+
+func TestNormalizeStageReportEnabledFailsWhenReportMissing(t *testing.T) {
+	env, m, ser := setupNormalizeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`)
+	report := true
+	env.Config.Pipeline.Normalize.Report = &report
+	ser.NormalizeResult = seriatim.NormalizeResult{ReportPath: filepath.Join(paths.ArtifactsDir, "missing-report.json")}
+
+	_, err := (normalizeStage{}).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 setupNormalizeEnv(t *testing.T) (*Env, *manifest.Manifest, *seriatim.FakeRunner) {
+	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
+	cfg := &config.Config{
+		PipelinePath: pipelinePath,
+		SessionPath:  sessionPath,
+		Pipeline: &config.PipelineConfig{
+			Workspace: config.WorkspaceConfig{Root: workspace},
+			Seriatim: config.SeriatimConfig{
+				Binary:       "seriatim",
+				Timeout:      "10m",
+				OutputSchema: "seriatim-intermediate",
+				CoalesceGap:  func() *float64 { v := 3.0; return &v }(),
+				Report:       &report,
+			},
+			Normalize: &config.NormalizeConfig{
+				OutputPath:   "transcripts/normalized.json",
+				OutputSchema: "seriatim-intermediate",
+				Report:       &report,
+			},
+		},
+		Session: &config.SessionConfig{SessionID: "2026-05-03"},
+	}
+
+	store := artifacts.NewLocalStore(workspace)
+	if _, err := store.EnsureLayout("2026-05-03"); err != nil {
+		t.Fatalf("EnsureLayout() error = %v", err)
+	}
+
+	ser := &seriatim.FakeRunner{}
+	return &Env{
+		Config:        cfg,
+		ArtifactStore: store,
+		Seriatim:      ser,
+	}, manifest.New("2026-05-03", time.Now().UTC()), ser
+}
--- a/internal/stage/placeholders_test.go
+++ b/internal/stage/placeholders_test.go
@@ -115,12 +115,18 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
 			if result.Metadata["stage"] != "normalize" {
 				t.Fatalf("normalize metadata = %#v, want stage=normalize", result.Metadata)
 			}
-			if result.Metadata["normalize_action"] != "copy_placeholder" {
-				t.Fatalf("normalize metadata = %#v, want normalize_action=copy_placeholder", result.Metadata)
+			if result.Metadata["output_schema"] == nil {
+				t.Fatalf("normalize metadata = %#v, want output_schema metadata", result.Metadata)
 			}
 			if len(result.Outputs) == 0 || result.Outputs[0].Kind != "transcript_normalized" {
 				t.Fatalf("normalize outputs = %#v, want transcript_normalized output", result.Outputs)
 			}
+			if len(result.Logs) != 2 {
+				t.Fatalf("normalize logs = %#v, want stdout+stderr", result.Logs)
+			}
+			if len(result.GeneratedConfigs) != 1 {
+				t.Fatalf("normalize generated configs = %#v, want one path", result.GeneratedConfigs)
+			}
 			continue
 		}
 		if s.Name() == "analyze" {