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 } p = artifacts.ResolveSessionLocalPathForRead(paths, 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 }