eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Implement real Seriatim merge stage

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-03 22:17:17 -05:00
parent a155af700a
commit 793ed48d13
10 changed files with 1015 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
internal/stage/merge.go Normal file
View File

@@ -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
}

197
internal/stage/merge_test.go Normal file
View File

@@ -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())
}

20
internal/stage/placeholders.go
View File

@@ -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"},

10
internal/stage/placeholders_test.go
View File

@@ -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"},