From 9392da261528e96d82b73c60e1ac9292b8352e1a Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 10 May 2026 21:40:31 +0000 Subject: [PATCH] Implement real normalize stage --- README.md | 5 +- architecture.md | 4 +- internal/stage/normalize.go | 131 +++++++++++--- internal/stage/normalize_test.go | 256 ++++++++++++++++++++++++++++ internal/stage/placeholders_test.go | 10 +- 5 files changed, 378 insertions(+), 28 deletions(-) create mode 100644 internal/stage/normalize_test.go diff --git a/README.md b/README.md index c2c0bdd..08524a3 100644 --- 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 diff --git a/architecture.md b/architecture.md index 1ee8fc6..9b538cc 100644 --- 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`. diff --git a/internal/stage/normalize.go b/internal/stage/normalize.go index 5a7c4ec..99b2767 100644 --- 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 { diff --git a/internal/stage/normalize_test.go b/internal/stage/normalize_test.go new file mode 100644 index 0000000..a3ca1d3 --- /dev/null +++ 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 +} diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 7c7d8c1..229568d 100644 --- 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" {