diff --git a/internal/stage/merge.go b/internal/stage/merge.go index 05d9032..4e591ff 100644 --- a/internal/stage/merge.go +++ b/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}, - GeneratedConfigs: []string{genCfgPath}, + Logs: append(normalizeLogs, stdoutPath, stderrPath), + 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 { diff --git a/internal/stage/merge_test.go b/internal/stage/merge_test.go index 8c7e647..26d0e97 100644 --- a/internal/stage/merge_test.go +++ b/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 { - t.Fatalf("logs = %#v, want 2 paths", result.Logs) + if len(result.Logs) != 6 { + t.Fatalf("logs = %#v, want 6 paths (4 normalize + 2 merge)", result.Logs) } - if len(result.GeneratedConfigs) != 1 { - t.Fatalf("generated configs = %#v, want 1 path", result.GeneratedConfigs) + if len(result.GeneratedConfigs) != 3 { + 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) { - t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath)) + if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) { + 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) { - t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath)) + if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) { + t.Fatalf("resolved transcript path = %q, want normalized path under transcripts/raw/normalized", got[0]) } }