Updated the merge stage to normalize the per-speaker transcripts before merging them

internal/stage/merge.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
+	"time"
 
 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
@@ -86,10 +87,15 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
 	stderrPath := filepath.Join(paths.LogsDir, "seriatim.stderr.log")
 	genCfgPath := filepath.Join(paths.ConfigDir, "seriatim.generated.yml")
 
+	normalizedInputs, normalizeLogs, normalizeConfigs, normalizeMeta, err := normalizeMergeInputs(ctx, env, inputs, paths)
+	if err != nil {
+		return nil, err
+	}
+
 	reportEnabled := env.Config.Pipeline.Seriatim.Report != nil && *env.Config.Pipeline.Seriatim.Report
 	req := seriatim.MergeRequest{
 		GeneratedConfigPath:        genCfgPath,
-		InputTranscriptPaths:       inputs,
+		InputTranscriptPaths:       normalizedInputs,
 		OutputMergedTranscriptPath: mergedPath,
 		ReportPath:                 "",
 		SpeakersPath:               speakersPath,
@@ -146,8 +152,11 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
 
 	meta := map[string]any{
 		"stage":                    "merge",
-		"input_transcripts_count":  len(inputs),
+		"input_transcripts_count":  len(normalizedInputs),
 		"input_transcript_paths":   inputs,
+		"normalized_inputs_count":  len(normalizedInputs),
+		"normalized_input_paths":   normalizedInputs,
+		"normalize_inputs":         normalizeMeta,
 		"output_schema":            env.Config.Pipeline.Seriatim.OutputSchema,
 		"coalesce_gap":             coalesceGap,
 		"report_enabled":           reportEnabled,
@@ -172,12 +181,93 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
 
 	return &StageResult{
 		Outputs:          outputs,
-		Logs:             []string{stdoutPath, stderrPath},
+		Logs:             append(normalizeLogs, stdoutPath, stderrPath),
-		GeneratedConfigs: []string{genCfgPath},
+		GeneratedConfigs: append(normalizeConfigs, genCfgPath),
 		Metadata:         meta,
 	}, nil
 }
 
+type normalizeMergeInputMeta struct {
+	InputPath         string `json:"input_path"`
+	OutputPath        string `json:"output_path"`
+	StdoutLogPath     string `json:"stdout_log_path"`
+	StderrLogPath     string `json:"stderr_log_path"`
+	GeneratedConfig   string `json:"generated_config_path"`
+	DurationMs        int64  `json:"duration_ms"`
+	ExitCode          int    `json:"exit_code"`
+	InvokedBinary     string `json:"invoked_binary"`
+	OutputSchema      string `json:"output_schema"`
+	AdapterReportPath string `json:"adapter_report_path,omitempty"`
+	AdapterOutputPath string `json:"adapter_output_path,omitempty"`
+}
+
+func normalizeMergeInputs(ctx context.Context, env *Env, rawInputs []string, paths artifacts.SessionPaths) ([]string, []string, []string, []normalizeMergeInputMeta, error) {
+	normalizedInputs := make([]string, 0, len(rawInputs))
+	logs := make([]string, 0, len(rawInputs)*2)
+	configs := make([]string, 0, len(rawInputs))
+	meta := make([]normalizeMergeInputMeta, 0, len(rawInputs))
+
+	var timeout time.Duration
+	timeoutRaw := strings.TrimSpace(env.Config.Pipeline.Seriatim.Timeout)
+	if timeoutRaw != "" {
+		parsed, err := time.ParseDuration(timeoutRaw)
+		if err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("merge: parse seriatim timeout %q: %w", env.Config.Pipeline.Seriatim.Timeout, err)
+		}
+		timeout = parsed
+	}
+	for _, input := range rawInputs {
+		base := strings.TrimSuffix(filepath.Base(input), filepath.Ext(input))
+		outPath := filepath.Join(paths.TranscriptsRawDir, "normalized", base+".normalized.json")
+		stdoutPath := filepath.Join(paths.LogsDir, "seriatim.normalize."+base+".stdout.log")
+		stderrPath := filepath.Join(paths.LogsDir, "seriatim.normalize."+base+".stderr.log")
+		cfgPath := filepath.Join(paths.ConfigDir, "seriatim.normalize."+base+".generated.yml")
+
+		req := seriatim.NormalizeRequest{
+			Binary:               env.Config.Pipeline.Seriatim.Binary,
+			InputTranscriptPath:  input,
+			OutputNormalizedPath: outPath,
+			OutputSchema:         env.Config.Pipeline.Seriatim.OutputSchema,
+			ReportPath:           "",
+			StdoutLogPath:        stdoutPath,
+			StderrLogPath:        stderrPath,
+			GeneratedConfigPath:  cfgPath,
+			Timeout:              timeout,
+		}
+		res, err := env.Seriatim.Normalize(ctx, req)
+		if err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("merge: normalize input %q failed: %w", input, err)
+		}
+
+		finalOutputPath := outPath
+		if strings.TrimSpace(res.OutputNormalizedPath) != "" {
+			finalOutputPath = strings.TrimSpace(res.OutputNormalizedPath)
+		}
+		if err := validateTranscriptJSONFile(finalOutputPath); err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("merge: normalized transcript %q invalid (input %q): %w", finalOutputPath, input, err)
+		}
+
+		normalizedInputs = append(normalizedInputs, finalOutputPath)
+		logs = append(logs, stdoutPath, stderrPath)
+		configs = append(configs, cfgPath)
+		meta = append(meta, normalizeMergeInputMeta{
+			InputPath:         input,
+			OutputPath:        finalOutputPath,
+			StdoutLogPath:     stdoutPath,
+			StderrLogPath:     stderrPath,
+			GeneratedConfig:   cfgPath,
+			DurationMs:        res.Duration.Milliseconds(),
+			ExitCode:          res.ExitCode,
+			InvokedBinary:     res.InvokedBinary,
+			OutputSchema:      res.OutputSchema,
+			AdapterReportPath: res.ReportPath,
+			AdapterOutputPath: res.OutputNormalizedPath,
+		})
+	}
+
+	return normalizedInputs, logs, configs, meta, nil
+}
+
 func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) {
 	fromManifest := make([]string, 0)
 	if m != nil && m.Stages != nil {

internal/stage/merge_test.go

@@ -54,6 +54,17 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
 	if len(req.InputTranscriptPaths) != 2 {
 		t.Fatalf("input transcripts = %#v, want 2", req.InputTranscriptPaths)
 	}
+	if len(fake.NormalizeRequests) != 2 {
+		t.Fatalf("normalize requests = %#v, want 2", fake.NormalizeRequests)
+	}
+	if fake.NormalizeRequests[0].InputTranscriptPath != inA || fake.NormalizeRequests[1].InputTranscriptPath != inB {
+		t.Fatalf("normalize request inputs = %#v", fake.NormalizeRequests)
+	}
+	for _, mergeIn := range req.InputTranscriptPaths {
+		if !strings.Contains(mergeIn, filepath.Join("transcripts", "raw", "normalized")) {
+			t.Fatalf("merge input path = %q, want normalized input path", mergeIn)
+		}
+	}
 
 	if len(result.Outputs) != 2 {
 		t.Fatalf("outputs len = %d, want 2", len(result.Outputs))
@@ -64,11 +75,11 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
 	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 {
+	if len(result.Logs) != 6 {
-		t.Fatalf("logs = %#v, want 2 paths", result.Logs)
+		t.Fatalf("logs = %#v, want 6 paths (4 normalize + 2 merge)", result.Logs)
 	}
-	if len(result.GeneratedConfigs) != 1 {
+	if len(result.GeneratedConfigs) != 3 {
-		t.Fatalf("generated configs = %#v, want 1 path", result.GeneratedConfigs)
+		t.Fatalf("generated configs = %#v, want 3 paths (2 normalize + 1 merge)", result.GeneratedConfigs)
 	}
 
 	meta := result.Metadata
@@ -84,6 +95,12 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
 	if meta["input_transcripts_count"] != 2 {
 		t.Fatalf("metadata input_transcripts_count = %#v, want 2", meta["input_transcripts_count"])
 	}
+	if meta["normalized_inputs_count"] != 2 {
+		t.Fatalf("metadata normalized_inputs_count = %#v, want 2", meta["normalized_inputs_count"])
+	}
+	if _, ok := meta["normalize_inputs"]; !ok {
+		t.Fatalf("metadata normalize_inputs missing: %#v", meta)
+	}
 }
 
 func TestMergeStageFailsWhenNoRawTranscripts(t *testing.T) {
@@ -152,6 +169,9 @@ func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testin
 	if len(fake.Requests) != 1 || len(fake.Requests[0].InputTranscriptPaths) != 1 {
 		t.Fatalf("fallback inputs = %#v", fake.Requests)
 	}
+	if len(fake.NormalizeRequests) != 1 {
+		t.Fatalf("normalize requests = %#v, want 1", fake.NormalizeRequests)
+	}
 }
 
 func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t *testing.T) {
@@ -180,8 +200,51 @@ func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t
 	if len(got) != 1 {
 		t.Fatalf("input transcript paths = %#v, want len 1", got)
 	}
-	if got[0] != filepath.Clean(rawPath) {
+	if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) {
-		t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath))
+		t.Fatalf("resolved transcript path = %q, want normalized path under transcripts/raw/normalized", got[0])
+	}
+}
+
+func TestMergeStageFailsWhenNormalizeAdapterFails(t *testing.T) {
+	env, m := setupMergeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	in := filepath.Join(paths.TranscriptsRawDir, "alice.json")
+	writeFile(t, in, `{"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{NormalizeErr: context.DeadlineExceeded}
+
+	_, err := (mergeStage{}).Run(context.Background(), env, m)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "normalize input") {
+		t.Fatalf("error = %q", err.Error())
+	}
+}
+
+func TestMergeStageFailsWhenNormalizedOutputInvalid(t *testing.T) {
+	env, m := setupMergeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	in := filepath.Join(paths.TranscriptsRawDir, "alice.json")
+	writeFile(t, in, `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
+	writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
+
+	badNormalized := filepath.Join(paths.ArtifactsDir, "bad.normalized.json")
+	writeFile(t, badNormalized, "not-json")
+	env.Seriatim = &seriatim.FakeRunner{
+		NormalizeResult: seriatim.NormalizeResult{
+			OutputNormalizedPath: badNormalized,
+		},
+	}
+
+	_, err := (mergeStage{}).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())
 	}
 }
 
@@ -211,8 +274,8 @@ func TestMergeStageResolvesSessionRelativeManifestOutputs(t *testing.T) {
 	if len(got) != 1 {
 		t.Fatalf("input transcript paths = %#v, want len 1", got)
 	}
-	if got[0] != filepath.Clean(rawPath) {
+	if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) {
-		t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath))
+		t.Fatalf("resolved transcript path = %q, want normalized path under transcripts/raw/normalized", got[0])
 	}
 }