diff --git a/internal/adapters/seriatim/fake.go b/internal/adapters/seriatim/fake.go index b143b24..59ad493 100644 --- a/internal/adapters/seriatim/fake.go +++ b/internal/adapters/seriatim/fake.go @@ -74,6 +74,11 @@ func (f *FakeRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, er } func materializePlaceholders(req MergeRequest) error { + if req.OutputMergedTranscriptPath != "" { + if err := subprocess.WriteFileAtomic(req.OutputMergedTranscriptPath, []byte(`{"schema":"seriatim.intermediate.v1","segments":[]}`), 0o644); err != nil { + return fmt.Errorf("write merged transcript %q: %w", req.OutputMergedTranscriptPath, err) + } + } if req.GeneratedConfigPath != "" { payload := map[string]any{ "schema": "seriatim.generated.v1", diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index 8d85dce..511d950 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -3,6 +3,7 @@ package app import ( "bytes" "context" + "fmt" "net/http" "net/http/httptest" "os" @@ -195,6 +196,8 @@ func writeValidConfigFiles(t *testing.T, workspaceRoot string, transcribeURL ... if len(transcribeURL) > 0 && strings.TrimSpace(transcribeURL[0]) != "" { url = transcribeURL[0] } + seriatimBinary := writeSeriatimAppTestWrapper(t) + t.Setenv("GO_WANT_APP_SERIATIM_HELPER", "1") pipelineYAML := `workspace: root: ` + workspaceRoot + ` @@ -207,7 +210,7 @@ whisperx: retry_delay: 1ms concurrency: 1 seriatim: - binary: seriatim + binary: ` + seriatimBinary + ` timeout: 10m output_schema: seriatim-intermediate coalesce_gap: 3.0 @@ -237,7 +240,7 @@ inputs: t.Fatalf("write session config: %v", err) } - mustWriteTestFile(t, filepath.Join(dir, "speakers.yml"), "alice: alice.flac\n") + mustWriteTestFile(t, filepath.Join(dir, "speakers.yml"), "match:\n - speaker: Alice\n match: [\"alice\"]\n") mustWriteTestFile(t, filepath.Join(dir, "autocorrect.yml"), "[]\n") mustWriteTestFile(t, filepath.Join(dir, "glossary.yml"), "[]\n") mustWriteTestFile(t, filepath.Join(dir, "audio", "alice.flac"), "audio-bytes") @@ -269,3 +272,74 @@ func mustWriteTestFile(t *testing.T, path, contents string) { t.Fatalf("write %q: %v", path, err) } } + +func writeSeriatimAppTestWrapper(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(), "seriatim-helper-wrapper.sh") + content := "#!/bin/sh\nexec \"" + exe + "\" -test.run=TestSeriatimAppHelper -- \"$@\"\n" + if err := os.WriteFile(path, []byte(content), 0o755); err != nil { + t.Fatalf("WriteFile(%q): %v", path, err) + } + return path +} + +func TestSeriatimAppHelper(t *testing.T) { + if os.Getenv("GO_WANT_APP_SERIATIM_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) + } + mergeArgs := args[start:] + + outputPath := appSeriatimFlagValue(mergeArgs, "--output-file") + reportPath := appSeriatimFlagValue(mergeArgs, "--report-file") + if strings.TrimSpace(outputPath) == "" { + _, _ = os.Stderr.WriteString("missing --output-file\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":"seriatim-intermediate","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) + } + } + _, _ = os.Stdout.WriteString("seriatim helper stdout\n") + _, _ = os.Stderr.WriteString("seriatim 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 { + return args[i+1] + } + } + return "" +} diff --git a/internal/app/resume_run_stage_test.go b/internal/app/resume_run_stage_test.go index 6630cf7..b7b85b3 100644 --- a/internal/app/resume_run_stage_test.go +++ b/internal/app/resume_run_stage_test.go @@ -25,6 +25,10 @@ func TestResumeStartsAfterCompletedStages(t *testing.T) { if err := store.Save(context.Background(), manifestPath, m); err != nil { t.Fatalf("save manifest: %v", err) } + workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + 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") var out bytes.Buffer err := Resume(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath}, &out) diff --git a/internal/app/runner.go b/internal/app/runner.go index 8e114a4..64ca76a 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -59,7 +59,11 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage env.WhisperX = client } if env.Seriatim == nil { - env.Seriatim = &seriatim.NoopRunner{} + runner, err := buildDefaultSeriatimRunner(env.Config) + if err != nil { + return nil, fmt.Errorf("initialize seriatim runner: %w", err) + } + env.Seriatim = runner } if env.Audita == nil { env.Audita = &audita.NoopRunner{} @@ -179,6 +183,40 @@ func buildDefaultWhisperXClient(cfg *config.Config) (whisperx.Client, error) { return client, nil } +func buildDefaultSeriatimRunner(cfg *config.Config) (seriatim.Runner, error) { + if cfg == nil || cfg.Pipeline == nil { + return &seriatim.NoopRunner{}, nil + } + + s := cfg.Pipeline.Seriatim + if strings.TrimSpace(s.Binary) == "" || strings.TrimSpace(s.Timeout) == "" || strings.TrimSpace(s.OutputSchema) == "" { + // Compatibility fallback for tests or internal call paths that bypass config validation/defaults. + return &seriatim.NoopRunner{}, nil + } + + report := false + if s.Report != nil { + report = *s.Report + } + runner, err := seriatim.NewSubprocessRunnerFromConfigValues( + s.Binary, + s.Timeout, + s.OutputSchema, + s.CoalesceGap, + report, + seriatim.EnvConfig{ + OverlapWordRunGap: s.Env.OverlapWordRunGap, + OverlapWordRunReorderWindow: s.Env.OverlapWordRunReorderWindow, + BackchannelMaxDuration: s.Env.BackchannelMaxDuration, + FillerMaxDuration: s.Env.FillerMaxDuration, + }, + ) + if err != nil { + return nil, fmt.Errorf("from pipeline.seriatim: %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 94631f2..364c431 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -84,6 +84,21 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { } continue } + if name == "merge" { + if sr.Metadata == nil || sr.Metadata["stage"] != "merge" { + t.Fatalf("merge metadata missing stage=merge: %#v", sr.Metadata) + } + if len(sr.Outputs) == 0 { + t.Fatalf("merge outputs missing") + } + if len(sr.Logs) == 0 { + t.Fatalf("merge logs missing") + } + if len(sr.GeneratedConfigs) == 0 { + t.Fatalf("merge generated configs missing") + } + continue + } if sr.Metadata == nil || sr.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", name) } @@ -286,6 +301,25 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { t.Fatalf("seed manifest: %v", err) } } + if tc.name == "merge" { + paths, ensureErr := artifactStore.EnsureLayout(cfg.Session.SessionID) + if ensureErr != nil { + t.Fatalf("EnsureLayout() error = %v", ensureErr) + } + rawPath := filepath.Join(paths.TranscriptsRawDir, "alice.json") + if err := os.MkdirAll(filepath.Dir(rawPath), 0o755); err != nil { + t.Fatalf("mkdir raw dir: %v", err) + } + if err := os.WriteFile(rawPath, []byte(`{"segments":[]}`), 0o644); err != nil { + t.Fatalf("write raw transcript: %v", err) + } + if err := os.WriteFile(filepath.Join(paths.InputsDir, "speakers.yml"), []byte("match: []\n"), 0o644); err != nil { + t.Fatalf("write speakers: %v", err) + } + if err := os.WriteFile(filepath.Join(paths.InputsDir, "autocorrect.yml"), []byte("rules: []\n"), 0o644); err != nil { + t.Fatalf("write autocorrect: %v", err) + } + } _, runErr := executeStages(context.Background(), cfg, []stage.Stage{selected}, RunOptions{Env: tc.env}) if runErr == nil { diff --git a/internal/stage/merge.go b/internal/stage/merge.go new file mode 100644 index 0000000..005aad6 --- /dev/null +++ b/internal/stage/merge.go @@ -0,0 +1,247 @@ +package stage + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +type mergeStage struct{} + +func (mergeStage) Name() string { return "merge" } + +func (mergeStage) Declares() IODecl { + return IODecl{ + Inputs: []artifacts.Ref{ + {Kind: "transcript_raw", Category: "transcripts", RelativePath: "transcripts/raw/*.json"}, + {Kind: "speakers", Category: "inputs", RelativePath: "inputs/speakers.yml"}, + {Kind: "autocorrect", Category: "inputs", RelativePath: "inputs/autocorrect.yml"}, + }, + Outputs: []artifacts.Ref{ + {Kind: "transcript_merged", Category: "transcripts", RelativePath: "transcripts/merged.json"}, + {Kind: "seriatim_report", Category: "artifacts", RelativePath: "artifacts/seriatim.report.json"}, + }, + } +} + +func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { + if env == nil || env.Config == nil { + return nil, fmt.Errorf("merge: stage environment config is required") + } + if env.ArtifactStore == nil { + return nil, fmt.Errorf("merge: artifact store is required") + } + if env.Config.Pipeline == nil || env.Config.Session == nil { + return nil, fmt.Errorf("merge: resolved config must include pipeline and session") + } + if env.Seriatim == nil { + return nil, fmt.Errorf("merge: seriatim 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("merge: session id is required") + } + + paths := env.ArtifactStore.SessionPaths(sessionID) + + inputs, err := discoverRawTranscripts(m, paths) + if err != nil { + return nil, fmt.Errorf("merge: resolve raw transcripts: %w", err) + } + if len(inputs) == 0 { + return nil, fmt.Errorf("merge: no raw transcript inputs found") + } + for _, in := range inputs { + if err := validateTranscriptJSONFile(in); err != nil { + return nil, fmt.Errorf("merge: input transcript %q invalid: %w", in, err) + } + } + + speakersPath := filepath.Join(paths.InputsDir, "speakers.yml") + if err := requireFile(speakersPath, "speakers.yml"); err != nil { + return nil, fmt.Errorf("merge: %w", err) + } + autocorrectPath := filepath.Join(paths.InputsDir, "autocorrect.yml") + if err := requireFile(autocorrectPath, "autocorrect.yml"); err != nil { + return nil, fmt.Errorf("merge: %w", err) + } + + mergedPath := filepath.Join(paths.TranscriptsDir, "merged.json") + reportPath := filepath.Join(paths.ArtifactsDir, "seriatim.report.json") + stdoutPath := filepath.Join(paths.LogsDir, "seriatim.stdout.log") + stderrPath := filepath.Join(paths.LogsDir, "seriatim.stderr.log") + genCfgPath := filepath.Join(paths.ConfigDir, "seriatim.generated.yml") + + reportEnabled := env.Config.Pipeline.Seriatim.Report != nil && *env.Config.Pipeline.Seriatim.Report + req := seriatim.MergeRequest{ + GeneratedConfigPath: genCfgPath, + InputTranscriptPaths: inputs, + OutputMergedTranscriptPath: mergedPath, + ReportPath: "", + SpeakersPath: speakersPath, + AutocorrectPath: autocorrectPath, + StdoutLogPath: stdoutPath, + StderrLogPath: stderrPath, + } + if reportEnabled { + req.ReportPath = reportPath + } + + res, err := env.Seriatim.Run(ctx, req) + if err != nil { + return nil, fmt.Errorf("merge: seriatim merge failed: %w", err) + } + + finalMergedPath := mergedPath + if strings.TrimSpace(res.MergedTranscriptPath) != "" { + finalMergedPath = res.MergedTranscriptPath + } + if err := validateTranscriptJSONFile(finalMergedPath); err != nil { + return nil, fmt.Errorf("merge: merged transcript %q invalid: %w", finalMergedPath, err) + } + + finalReportPath := req.ReportPath + if strings.TrimSpace(res.ReportPath) != "" { + finalReportPath = res.ReportPath + } + if reportEnabled { + if err := validateTranscriptJSONFile(finalReportPath); err != nil { + return nil, fmt.Errorf("merge: report %q invalid: %w", finalReportPath, err) + } + } + + outputs := []artifacts.Ref{{ + Kind: "transcript_merged", + Category: "transcripts", + SessionID: sessionID, + AbsolutePath: finalMergedPath, + }} + if reportEnabled { + outputs = append(outputs, artifacts.Ref{ + Kind: "seriatim_report", + Category: "artifacts", + SessionID: sessionID, + AbsolutePath: finalReportPath, + }) + } + + coalesceGap := any(nil) + if env.Config.Pipeline.Seriatim.CoalesceGap != nil { + coalesceGap = *env.Config.Pipeline.Seriatim.CoalesceGap + } + + meta := map[string]any{ + "stage": "merge", + "input_transcripts_count": len(inputs), + "input_transcript_paths": inputs, + "output_schema": env.Config.Pipeline.Seriatim.OutputSchema, + "coalesce_gap": coalesceGap, + "report_enabled": reportEnabled, + "output_path": finalMergedPath, + "report_path": finalReportPath, + "timeout": env.Config.Pipeline.Seriatim.Timeout, + "binary": env.Config.Pipeline.Seriatim.Binary, + "adapter_duration_ms": res.Duration.Milliseconds(), + "adapter_exit_code": res.ExitCode, + "adapter_invoked_binary": res.InvokedBinary, + "adapter_output_schema": res.OutputSchema, + "generated_config_path": genCfgPath, + "stdout_log_path": stdoutPath, + "stderr_log_path": stderrPath, + "adapter_report_path": res.ReportPath, + "adapter_merged_out_path": res.MergedTranscriptPath, + "adapter_generated_config": res.GeneratedConfigPath, + } + if res.Metadata != nil { + meta["adapter_metadata"] = res.Metadata + } + + return &StageResult{ + Outputs: outputs, + Logs: []string{stdoutPath, stderrPath}, + GeneratedConfigs: []string{genCfgPath}, + Metadata: meta, + }, nil +} + +func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) { + fromManifest := make([]string, 0) + if m != nil && m.Stages != nil { + if tr := m.Stages["transcribe"]; tr != nil { + for _, out := range tr.Outputs { + if out.Kind != "transcript_raw" { + continue + } + p := strings.TrimSpace(out.LocalPath) + if p == "" { + continue + } + if !filepath.IsAbs(p) { + p = filepath.Join(paths.Root, p) + } + fromManifest = append(fromManifest, filepath.Clean(p)) + } + } + } + + if len(fromManifest) > 0 { + deduped := dedupeAndSortPaths(fromManifest) + return deduped, nil + } + + entries, err := os.ReadDir(paths.TranscriptsRawDir) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, fmt.Errorf("read raw transcript directory %q: %w", paths.TranscriptsRawDir, err) + } + + out := make([]string, 0, len(entries)) + for _, entry := range entries { + if entry.IsDir() { + continue + } + p := filepath.Join(paths.TranscriptsRawDir, entry.Name()) + if !strings.EqualFold(filepath.Ext(p), ".json") { + continue + } + out = append(out, p) + } + return dedupeAndSortPaths(out), nil +} + +func dedupeAndSortPaths(paths []string) []string { + if len(paths) == 0 { + return nil + } + set := make(map[string]struct{}, len(paths)) + out := make([]string, 0, len(paths)) + for _, p := range paths { + clean := filepath.Clean(strings.TrimSpace(p)) + if clean == "" { + continue + } + if _, exists := set[clean]; exists { + continue + } + set[clean] = struct{}{} + out = append(out, clean) + } + sort.Strings(out) + return out +} diff --git a/internal/stage/merge_test.go b/internal/stage/merge_test.go new file mode 100644 index 0000000..5b3bb7e --- /dev/null +++ b/internal/stage/merge_test.go @@ -0,0 +1,197 @@ +package stage + +import ( + "context" + "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 TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) { + env, m := setupMergeEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + + inA := filepath.Join(paths.TranscriptsRawDir, "alice.json") + inB := filepath.Join(paths.TranscriptsRawDir, "bob.json") + writeFile(t, inA, `{"segments":[{"speaker":"alice","text":"hello","start":0,"end":1}]}`) + writeFile(t, inB, `{"segments":[{"speaker":"bob","text":"hi","start":1,"end":2}]}`) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match:\n - speaker: Alice\n match: [\"alice\"]\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + + m.MarkStageSucceeded("transcribe", time.Now().UTC(), []manifest.ArtifactRecord{ + {Kind: "transcript_raw", LocalPath: inA}, + {Kind: "transcript_raw", LocalPath: inB}, + }) + + fake := &seriatim.FakeRunner{} + env.Seriatim = fake + + result, err := (mergeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("merge.Run() error = %v", err) + } + if result == nil { + t.Fatal("merge result is nil") + } + if len(fake.Requests) != 1 { + t.Fatalf("fake requests = %d, want 1", len(fake.Requests)) + } + + req := fake.Requests[0] + if req.SpeakersPath != filepath.Join(paths.InputsDir, "speakers.yml") { + t.Fatalf("speakers path = %q", req.SpeakersPath) + } + if req.AutocorrectPath != filepath.Join(paths.InputsDir, "autocorrect.yml") { + t.Fatalf("autocorrect path = %q", req.AutocorrectPath) + } + if len(req.InputTranscriptPaths) != 2 { + t.Fatalf("input transcripts = %#v, want 2", req.InputTranscriptPaths) + } + + if len(result.Outputs) != 2 { + t.Fatalf("outputs len = %d, want 2", len(result.Outputs)) + } + if result.Outputs[0].Kind != "transcript_merged" { + t.Fatalf("output[0] kind = %q, want transcript_merged", result.Outputs[0].Kind) + } + if result.Outputs[1].Kind != "seriatim_report" { + t.Fatalf("output[1] kind = %q, want seriatim_report", result.Outputs[1].Kind) + } + if len(result.Logs) != 2 { + t.Fatalf("logs = %#v, want 2 paths", result.Logs) + } + if len(result.GeneratedConfigs) != 1 { + t.Fatalf("generated configs = %#v, want 1 path", result.GeneratedConfigs) + } + + meta := result.Metadata + if meta["stage"] != "merge" { + t.Fatalf("metadata stage = %#v, want merge", meta["stage"]) + } + if meta["output_schema"] != "seriatim-intermediate" { + t.Fatalf("metadata output_schema = %#v", meta["output_schema"]) + } + if meta["report_enabled"] != true { + t.Fatalf("metadata report_enabled = %#v, want true", meta["report_enabled"]) + } + if meta["input_transcripts_count"] != 2 { + t.Fatalf("metadata input_transcripts_count = %#v, want 2", meta["input_transcripts_count"]) + } +} + +func TestMergeStageFailsWhenNoRawTranscripts(t *testing.T) { + env, m := setupMergeEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + env.Seriatim = &seriatim.FakeRunner{} + + _, err := (mergeStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "no raw transcript inputs found") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestMergeStageFailsOnInvalidInputJSON(t *testing.T) { + env, m := setupMergeEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), "not-json") + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + env.Seriatim = &seriatim.FakeRunner{} + + _, err := (mergeStage{}).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 TestMergeStageFailsWhenAdapterFails(t *testing.T) { + env, m := setupMergeEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + env.Seriatim = &seriatim.FakeRunner{Err: context.DeadlineExceeded} + + _, err := (mergeStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "seriatim merge failed") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testing.T) { + env, m := setupMergeEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + fake := &seriatim.FakeRunner{} + env.Seriatim = fake + + _, err := (mergeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("merge.Run() error = %v", err) + } + if len(fake.Requests) != 1 || len(fake.Requests[0].InputTranscriptPaths) != 1 { + t.Fatalf("fallback inputs = %#v", fake.Requests) + } +} + +func setupMergeEnv(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") + + coalesce := 3.0 + 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: &coalesce, + Report: &report, + }, + }, + Session: &config.SessionConfig{ + SessionID: "2026-05-03", + Inputs: config.SessionInputsConfig{ + SpeakersFile: "./speakers.yml", + AutocorrectFile: "./autocorrect.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()) +} diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index 6aa4ab9..8d29aff 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -8,7 +8,6 @@ import ( "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/seriatim" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/manifest" @@ -53,23 +52,6 @@ func (s placeholderStage) Run(ctx context.Context, env *Env, m *manifest.Manifes paths := env.ArtifactStore.SessionPaths(sessionID) switch s.name { - case "merge": - if env.Seriatim != nil { - req := seriatim.MergeRequest{ - GeneratedConfigPath: filepath.Join(paths.ConfigDir, "seriatim.generated.yml"), - InputTranscriptPaths: []string{filepath.Join(paths.TranscriptsNormalizedDir, "placeholder-speaker.json")}, - OutputMergedTranscriptPath: filepath.Join(paths.TranscriptsDir, "merged.json"), - StdoutLogPath: filepath.Join(paths.LogsDir, "seriatim.stdout.log"), - StderrLogPath: filepath.Join(paths.LogsDir, "seriatim.stderr.log"), - } - resp, err := env.Seriatim.Run(ctx, req) - if err != nil { - return nil, fmt.Errorf("placeholder merge adapter call failed: %w", err) - } - result.Outputs = append(result.Outputs, artifacts.Ref{Kind: "transcript_merged", Category: "transcripts", SessionID: sessionID, AbsolutePath: resp.MergedTranscriptPath}) - result.Logs = append(result.Logs, req.StdoutLogPath, req.StderrLogPath) - result.GeneratedConfigs = append(result.GeneratedConfigs, req.GeneratedConfigPath) - } case "polish": if env.Audita != nil { req := audita.PolishRequest{ @@ -147,7 +129,7 @@ func All() []Stage { prepareStage{}, transcribeStage{}, placeholderStage{name: "normalize"}, - placeholderStage{name: "merge"}, + mergeStage{}, placeholderStage{name: "polish"}, placeholderStage{name: "analyze"}, placeholderStage{name: "archive"}, diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 90a7976..2bc7151 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -93,6 +93,15 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } continue } + if s.Name() == "merge" { + if result.Metadata["stage"] != "merge" { + t.Fatalf("merge metadata = %#v, want stage=merge", result.Metadata) + } + if len(result.Outputs) == 0 { + t.Fatalf("merge outputs = %#v, want non-empty", result.Outputs) + } + continue + } if result.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", s.Name()) } @@ -124,7 +133,6 @@ func TestPlaceholderAdapterErrorPropagation(t *testing.T) { env *Env wantErr string }{ - {stageName: "merge", env: &Env{Seriatim: &seriatim.FakeRunner{Err: errors.New("serr")}}, wantErr: "merge"}, {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"}, diff --git a/reference/seriatim/README.md b/reference/seriatim/README.md new file mode 100644 index 0000000..acb3bbc --- /dev/null +++ b/reference/seriatim/README.md @@ -0,0 +1,403 @@ +# seriatim + +`seriatim` merges per-speaker WhisperX-style JSON transcripts into a single JSON transcript that preserves speaker identity and chronological order. + +The current implementation supports the `merge` command. It reads one or more input JSON files, optionally maps each input file to a canonical speaker using `speakers.yml`, sorts all segments by timestamp, detects and resolves overlaps when word-level timing is available, assigns consecutive numeric `id` values, and writes a merged JSON artifact. + +## Usage + +Run from source: + +```sh +go run ./cmd/seriatim merge \ + --input-file samples/raw/2026-04-19-Eric_Rakestraw.json \ + --input-file samples/raw/2026-04-19-Mike_Brown.json \ + --output-file merged.json +``` + +Optional report output: + +```sh +go run ./cmd/seriatim merge \ + --input-file eric.json \ + --input-file mike.json \ + --output-file merged.json \ + --report-file report.json +``` + +## CLI + +```text +seriatim merge [flags] +``` + +Global flags: + +| Flag | Description | +| --- | --- | +| `--help` | Show command help. | +| `--version` | Show application version. Local builds default to `dev`; release builds inject the release version. | + +`merge` flags: + +| Flag | Required | Default | Description | +| --- | --- | --- | --- | +| `--input-file` | Yes | none | Input transcript JSON file. Repeat once per speaker/input file. | +| `--output-file` | Yes | none | Merged transcript JSON output path. | +| `--report-file` | No | none | Optional report JSON output path. | +| `--speakers` | No | none | Speaker map YAML file. When omitted, input file basenames are used as speaker labels. | +| `--autocorrect` | No | none | Autocorrect rules YAML file. When omitted, the default `autocorrect` module leaves text unchanged. | +| `--input-reader` | No | `json-files` | Input reader module. | +| `--output-modules` | No | `json` | Comma-separated output modules. | +| `--output-schema` | No | `seriatim-intermediate` | JSON output contract. Allowed values are `seriatim-minimal`, `seriatim-intermediate`, and `seriatim-full`. If omitted, the runtime default is used; consumers that depend on a specific shape should set this explicitly. | +| `--preprocessing-modules` | No | `validate-raw,normalize-speakers,trim-text` | Comma-separated preprocessing modules, evaluated in order. | +| `--postprocessing-modules` | No | `detect-overlaps,resolve-overlaps,backchannel,filler,resolve-danglers,coalesce,detect-overlaps,autocorrect,assign-ids,validate-output` | Comma-separated postprocessing modules, evaluated in order. | +| `--coalesce-gap` | No | `3.0` | Maximum same-speaker gap in seconds for `coalesce`; also used as the `resolve-overlaps` context window. Must be a non-negative float. | + +Environment variables: + +| Environment Variable | Default | Description | +| --- | --- | --- | +| `SERIATIM_OUTPUT_SCHEMA` | `seriatim-intermediate` | Output schema used when `--output-schema` is not explicitly provided. Allowed values are `seriatim-minimal`, `seriatim-intermediate`, and `seriatim-full`. The CLI flag takes precedence. | +| `SERIATIM_OVERLAP_WORD_RUN_GAP` | `1.0` | Maximum gap in seconds between adjacent timed words when `resolve-overlaps` builds word-run replacement segments. Must be a positive float. | +| `SERIATIM_OVERLAP_WORD_RUN_REORDER_WINDOW` | `1.0` | Near-start window in seconds for ordering replacement word runs shortest-first. Must be a positive float. | +| `SERIATIM_BACKCHANNEL_MAX_DURATION` | `2.0` | Maximum duration in seconds for `backchannel` classification. Must be a positive float. | +| `SERIATIM_FILLER_MAX_DURATION` | `1.25` | Maximum duration in seconds for `filler` classification. Must be a positive float. | + +## Input JSON Format + +Each input file must be valid JSON with a top-level `segments` array. The current parser accepts the WhisperX segment subset needed for merging: + +```json +{ + "segments": [ + { + "start": 1.25, + "end": 3.5, + "text": "Hello there.", + "words": [ + {"word": "Hello", "start": 1.25, "end": 1.55, "score": 0.98}, + {"word": "there.", "start": 1.7, "end": 2.0} + ] + } + ] +} +``` + +Required segment fields: + +- `start`: number, must be `>= 0`. +- `end`: number, must be `>= start`. +- `text`: string. + +Optional word fields: + +- `words`: array of word timing objects. +- `words[].word`: string. +- `words[].start`: optional number, must be `>= 0` when present. +- `words[].end`: optional number, must be `>= start` when present with `start`. +- `words[].score`: optional number. +- `words[].speaker`: optional raw speaker label string. + +Word-level timing is preserved internally for overlap resolution. If a word is missing `start` or `end`, seriatim keeps the word text, emits a warning in the optional report, and does not use that word as a timing anchor. Word timing is not emitted in the final JSON artifact. + +## Speaker Map Format + +`speakers.yml` maps input files to canonical speaker names using ordered substring rules: + +This file is optional. If `--speakers` is omitted, `seriatim` uses each input file basename as the segment speaker label. + +```yaml +match: + - speaker: "Eric Rakestraw" + match: + - "Eric_Rakestraw" + - "Eric" + + - speaker: "Mike Brown" + match: + - "Mike_Brown" + - "mb" +``` + +For each `--input-file`, `seriatim` takes the file basename and evaluates the rules in order. The first rule with a matching substring wins, and no later rules are evaluated. + +For example, this input: + +```text +samples/raw/2026-04-19-Eric_Rakestraw.json +``` + +matches this rule because the basename contains `Eric_Rakestraw`: + +```yaml +- speaker: "Eric Rakestraw" + match: + - "Eric_Rakestraw" +``` + +Important details: + +- Matching is against the input file basename, not the full path. +- Matching is case-insensitive. +- Rules are evaluated from first to last. +- Each rule must have a non-empty `speaker`. +- Each rule must have at least one non-empty `match` string. +- Duplicate speaker names are invalid. +- Every input file must match at least one rule or the command fails. + +Deprecated old format: + +```yaml +inputs: + eric.json: + speaker: "Eric Rakestraw" +``` + +The old `inputs:` direct mapping format is no longer supported. + +## Output JSON Format + +`--output-modules json` controls the writer. `--output-schema` controls the JSON contract that writer serializes. + +The named schemas are stable public contracts. If a consumer depends on a specific shape, it should request that schema explicitly at runtime. The runtime default selection may change in a future release. + +The `seriatim-intermediate` schema is the current default selection when neither `--output-schema` nor `SERIATIM_OUTPUT_SCHEMA` is set. It stays close to the minimal schema, but adds optional `categories` on each segment: + +```json +{ + "metadata": { + "application": "seriatim", + "version": "dev", + "output_schema": "seriatim-intermediate" + }, + "segments": [ + { + "id": 1, + "start": 1.25, + "end": 3.5, + "speaker": "Eric Rakestraw", + "text": "Hello there.", + "categories": ["backchannel"] + } + ] +} +``` + +The `seriatim-full` schema uses the full seriatim envelope: + +```json +{ + "metadata": { + "application": "seriatim", + "version": "dev", + "input_reader": "json-files", + "input_files": ["eric.json", "mike.json"], + "preprocessing_modules": ["validate-raw", "normalize-speakers", "trim-text"], + "postprocessing_modules": ["detect-overlaps", "resolve-overlaps", "backchannel", "filler", "resolve-danglers", "coalesce", "detect-overlaps", "autocorrect", "assign-ids", "validate-output"], + "output_modules": ["json"] + }, + "segments": [ + { + "id": 1, + "source": "eric.json", + "source_segment_index": 0, + "speaker": "Eric Rakestraw", + "start": 1.25, + "end": 3.5, + "text": "Hello there.", + "overlap_group_id": 1 + }, + { + "id": 2, + "source": "eric.json", + "source_ref": "word-run:1:1:1", + "derived_from": ["eric.json#0"], + "speaker": "Eric Rakestraw", + "start": 2.0, + "end": 2.5, + "text": "Resolved word run", + "categories": ["backchannel"] + } + ], + "overlap_groups": [ + { + "id": 1, + "start": 1.25, + "end": 4.0, + "segments": ["eric.json#0", "mike.json#0"], + "speakers": ["Eric Rakestraw", "Mike Brown"], + "class": "unknown", + "resolution": "unresolved" + } + ] +} +``` + +The `seriatim-minimal` schema emits minimal metadata and compact ordered segments: + +```json +{ + "metadata": { + "application": "seriatim", + "version": "dev", + "output_schema": "seriatim-minimal" + }, + "segments": [ + { + "id": 1, + "start": 1.25, + "end": 3.5, + "speaker": "Eric Rakestraw", + "text": "Hello there." + } + ] +} +``` + +Minimal output intentionally omits categories, overlap groups, source/provenance fields, and pipeline configuration metadata. + +Intermediate output intentionally omits overlap groups and source/provenance fields, but keeps optional `categories` and minimal metadata. + +Segments are sorted deterministically by: + +```text +(start, end, source, source_segment_index/source_ref, speaker) +``` + +Final segment IDs are assigned after sorting and start at `1`. + +The public Go output contract is available from: + +```go +import "gitea.maximumdirect.net/eric/seriatim/schema" +``` + +The same package embeds machine-readable JSON Schemas in `schema/full-output.schema.json`, `schema/intermediate-output.schema.json`, and `schema/minimal-output.schema.json`. The default `validate-output` postprocessor validates the selected output shape and verifies final segment IDs are present, sequential, and start at `1`. + +## Overlap Detection + +The default postprocessing pipeline detects overlapping segment groups. + +Overlap behavior: + +- A strict timing overlap is required: `next.start < current_group_end`. +- Segments that only touch at a boundary are not grouped. +- Groups require at least two distinct speakers. +- Transitive overlaps are grouped together. +- Segments in detected groups receive `overlap_group_id`. +- `overlap_groups[].segments` contains stable references in `source#source_segment_index` format. +- `class` is currently `unknown`. +- `resolution` is `unresolved` until `resolve-overlaps` replaces the group. + +## Overlap Resolution + +The default postprocessing pipeline runs `detect-overlaps`, then `resolve-overlaps`, then `backchannel`, then `filler`, then `resolve-danglers`, then `coalesce`, then a second `detect-overlaps` pass. + +For each detected overlap group, `resolve-overlaps` uses preserved WhisperX word timing to build smaller word-run replacement segments: + +- The resolution window expands the detected overlap group by `--coalesce-gap` seconds on both sides. +- Nearby same-speaker context segments are included when they intersect the expanded window and their start or end is within `--coalesce-gap` of the original overlap boundary. +- Once a segment is selected for replacement, all timed words from that segment participate in word-run construction; the window controls segment selection, not per-word clipping. +- Context segments that are part of another detected overlap group are not pulled into the current group. +- Untimed words are included in replacement text in original word order when nearby timed words create a replacement run. +- Untimed words do not affect replacement segment start/end times or word-run gap splitting. +- Words for the same speaker are merged into one run when the gap between adjacent words is no greater than `SERIATIM_OVERLAP_WORD_RUN_GAP`. +- The default word-run gap is `1.0` seconds. +- Set `SERIATIM_OVERLAP_WORD_RUN_GAP` to a positive number of seconds to override the default. +- Near-start replacement word runs are reordered so shorter segments come first when adjacent starts are within `SERIATIM_OVERLAP_WORD_RUN_REORDER_WINDOW`. +- The default word-run reorder window is `1.0` seconds. +- Set `SERIATIM_OVERLAP_WORD_RUN_REORDER_WINDOW` to a positive number of seconds to override the default. +- Replacement segment text is built by joining word text with single spaces. +- Replacement segments include `source_ref` and `derived_from`. +- Replacement segments omit `source_segment_index` because they are derived from one or more original segments. +- Resolved overlap groups are removed before the second detection pass. +- Replacement segments are left without `overlap_group_id` until the second detection pass annotates any remaining overlap. +- If a speaker has no usable word timing in a group, that speaker's original segment is kept. +- If no speakers in a group have usable word timing, the original group and annotations remain unchanged. + +## Backchannels + +The default pipeline runs `backchannel` before `coalesce`. It tags short acknowledgement segments with: + +```json +"categories": ["backchannel"] +``` + +Backchannel matching is case-insensitive, ignores punctuation for matching and word-count purposes, trims surrounding whitespace, and requires a matching acknowledgement phrase, no more than three whitespace-delimited words, and duration no greater than `SERIATIM_BACKCHANNEL_MAX_DURATION` seconds. The default maximum duration is `2.0` seconds. + +## Fillers + +The default pipeline runs `filler` after `backchannel` and before `coalesce`. It tags short filler utterances with: + +```json +"categories": ["filler"] +``` + +Filler matching is case-insensitive, ignores punctuation for matching and word-count purposes, trims surrounding whitespace, and requires only filler tokens such as `um`, `uh`, `er`, `erm`, `ah`, `eh`, `hmm`, `mm`, or repeated combinations of those tokens. Matching segments must contain no more than three whitespace-delimited words and have duration no greater than `SERIATIM_FILLER_MAX_DURATION` seconds. The default maximum duration is `1.25` seconds. + +## Dangler Resolution + +The default pipeline runs `resolve-danglers` before `coalesce` and before the second overlap detection pass. It repairs short derived fragments when they share provenance with a nearby segment: + +- Dangling-end fragments have no more than two words and end in punctuation. +- Dangling-start fragments have no more than two words. +- Matching uses same-speaker segments with any shared `derived_from` value. +- Merged segments use `source_ref` values such as `resolve-danglers:1`, keep the target segment's transcript position, and union `derived_from`. + +## Coalescing + +The default pipeline runs `coalesce` after `resolve-danglers` and before the second overlap detection pass. It merges adjacent same-speaker segments in the transcript's current order when `next.start - current.end <= --coalesce-gap`. + +Coalesced segments use `source_ref` values such as `coalesce:1`, include `derived_from`, and omit `source_segment_index`. + +Different-speaker backchannel and filler segments do not block coalescing of surrounding same-speaker segments. Same-speaker backchannel and filler segments are merged normally when they are within `--coalesce-gap`. When same-speaker segments are coalesced, any `backchannel` or `filler` category from the merged inputs is dropped from the coalesced segment. + +## Autocorrect + +Autocorrect is included in the default postprocessing pipeline. If `--autocorrect` is omitted, the module leaves transcript text unchanged and records a skip event in the optional report. + +Enable corrections by passing `--autocorrect`: + +```sh +go run ./cmd/seriatim merge \ + --input-file input.json \ + --autocorrect autocorrect.yml \ + --output-file merged.json +``` + +`autocorrect.yml` format: + +```yaml +autocorrect: + - target: "Hrank" + match: + - "hrank" + - "Frank" + + - target: "Mike Brown" + match: + - "Mike Pat" +``` + +Matching behavior: + +- Matching is case-sensitive. +- Matches apply only to whole tokens, not substrings inside larger words. +- Punctuation and whitespace can surround a match. +- Multi-word and hyphenated matches are supported. +- Duplicate match strings are invalid, including duplicates across separate rules. + +## Current Limitations + +- Only JSON input is supported. +- Overlap resolution depends on WhisperX word timing; groups without usable word timing remain unresolved. +- Alternate output formats are not implemented yet. + +## Release Builds + +Local builds record version metadata as `dev`. Release builds should inject the release version with `ldflags`: + +```sh +go build -ldflags "-X gitea.maximumdirect.net/eric/seriatim/internal/buildinfo.Version=v1.0.0" ./cmd/seriatim +```