From 39da1cebf503643c7467884da9191c1b8af15c26 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 8 May 2026 17:08:44 +0000 Subject: [PATCH] Implement transcript trim stage --- internal/app/resume_run_stage_test.go | 8 +- internal/app/runner_test.go | 12 + internal/stage/placeholders.go | 2 +- internal/stage/placeholders_test.go | 12 + internal/stage/trim.go | 384 ++++++++++++++++++++++++ internal/stage/trim_test.go | 405 ++++++++++++++++++++++++++ 6 files changed, 819 insertions(+), 4 deletions(-) create mode 100644 internal/stage/trim.go create mode 100644 internal/stage/trim_test.go diff --git a/internal/app/resume_run_stage_test.go b/internal/app/resume_run_stage_test.go index 0f7a6ea..94c737e 100644 --- a/internal/app/resume_run_stage_test.go +++ b/internal/app/resume_run_stage_test.go @@ -166,10 +166,12 @@ func TestRunStageSkipAndForce(t *testing.T) { } } -func TestRunStageTrimPlaceholderExecutes(t *testing.T) { +func TestRunStageTrimExecutes(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) var out bytes.Buffer err := RunStage(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath, "trim"}, &out) @@ -188,7 +190,7 @@ func TestRunStageTrimPlaceholderExecutes(t *testing.T) { if m.Stages["trim"] == nil || m.Stages["trim"].Status != manifest.StatusSucceeded { t.Fatalf("trim stage = %#v, want succeeded", m.Stages["trim"]) } - if m.Stages["trim"].Metadata == nil || m.Stages["trim"].Metadata["placeholder"] != true { - t.Fatalf("trim stage metadata = %#v, want placeholder=true", m.Stages["trim"].Metadata) + if m.Stages["trim"].Metadata == nil || m.Stages["trim"].Metadata["trim_action"] != "copy_disabled" { + t.Fatalf("trim stage metadata = %#v, want trim_action=copy_disabled", m.Stages["trim"].Metadata) } } diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 2249a06..06fabad 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -123,6 +123,18 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { } continue } + if name == "trim" { + if sr.Metadata == nil || sr.Metadata["stage"] != "trim" { + t.Fatalf("trim metadata missing stage=trim: %#v", sr.Metadata) + } + if sr.Metadata["trim_action"] != "copy_disabled" { + t.Fatalf("trim metadata missing trim_action=copy_disabled: %#v", sr.Metadata) + } + if len(sr.Outputs) == 0 { + t.Fatalf("trim outputs missing") + } + continue + } if sr.Metadata == nil || sr.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", name) } diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index f60e643..9f01ae7 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -92,7 +92,7 @@ func All() []Stage { transcribeStage{}, mergeStage{}, polishStage{}, - placeholderStage{name: "trim"}, + trimStage{}, analyzeStage{}, placeholderStage{name: "archive"}, placeholderStage{name: "notify"}, diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index f3d789c..d618a48 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -120,6 +120,18 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } continue } + if s.Name() == "trim" { + if result.Metadata["stage"] != "trim" { + t.Fatalf("trim metadata = %#v, want stage=trim", result.Metadata) + } + if result.Metadata["trim_action"] != "copy_disabled" { + t.Fatalf("trim metadata = %#v, want trim_action=copy_disabled", result.Metadata) + } + if len(result.Outputs) == 0 || result.Outputs[0].Kind != "transcript_trimmed" { + t.Fatalf("trim outputs = %#v, want transcript_trimmed output", result.Outputs) + } + continue + } if result.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", s.Name()) } diff --git a/internal/stage/trim.go b/internal/stage/trim.go new file mode 100644 index 0000000..c02d096 --- /dev/null +++ b/internal/stage/trim.go @@ -0,0 +1,384 @@ +package stage + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" + "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/contracts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +type trimStage struct{} + +func (trimStage) Name() string { return "trim" } + +func (trimStage) Declares() IODecl { + return IODecl{ + Inputs: []artifacts.Ref{ + {Kind: "transcript_processed", Category: "transcripts", RelativePath: "transcripts/processed.json"}, + }, + Outputs: []artifacts.Ref{ + {Kind: "transcript_trimmed", Category: "transcripts", RelativePath: "transcripts/trimmed.json"}, + {Kind: "session_bounds", Category: "artifacts", RelativePath: "artifacts/session_bounds.json"}, + }, + } +} + +func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { + if env == nil || env.Config == nil { + return nil, fmt.Errorf("trim: stage environment config is required") + } + if env.ArtifactStore == nil { + return nil, fmt.Errorf("trim: artifact store is required") + } + if env.Config.Pipeline == nil || env.Config.Session == nil { + return nil, fmt.Errorf("trim: resolved config must include pipeline and session") + } + + 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("trim: session id is required") + } + + paths := env.ArtifactStore.SessionPaths(sessionID) + processedPath, processedSource, err := discoverProcessedTranscript(m, paths) + if err != nil { + return nil, fmt.Errorf("trim: resolve processed transcript: %w", err) + } + if processedPath == "" { + return nil, fmt.Errorf("trim: processed transcript input is required") + } + if err := validateProcessedTranscriptOutput(processedPath); err != nil { + return nil, fmt.Errorf("trim: processed transcript %q invalid: %w", processedPath, err) + } + + trimCfg := env.Config.Pipeline.Trim + enabled := trimCfg != nil && trimCfg.Enabled + + trimmedPath, err := resolveTrimmedOutputPath(paths, trimCfg) + if err != nil { + return nil, fmt.Errorf("trim: resolve trimmed output path: %w", err) + } + + logPaths := []string{} + generatedConfigs := []string{} + metadata := map[string]any{ + "stage": "trim", + "trim_enabled": enabled, + "processed_transcript_path": processedPath, + "processed_transcript_source": processedSource, + "trimmed_output_path": trimmedPath, + } + + if !enabled { + if err := copyTranscript(env.ArtifactStore, processedPath, trimmedPath); err != nil { + return nil, fmt.Errorf("trim: copy processed transcript to trimmed output: %w", err) + } + if err := validateProcessedTranscriptOutput(trimmedPath); err != nil { + return nil, fmt.Errorf("trim: copied trimmed transcript %q invalid: %w", trimmedPath, err) + } + metadata["trim_action"] = "copy_disabled" + return &StageResult{ + Outputs: []artifacts.Ref{{ + Kind: "transcript_trimmed", + Category: "transcripts", + SessionID: sessionID, + AbsolutePath: trimmedPath, + }}, + Metadata: metadata, + }, nil + } + + if env.Scriptorium == nil { + return nil, fmt.Errorf("trim: scriptorium adapter is required when trim is enabled") + } + if env.Seriatim == nil { + return nil, fmt.Errorf("trim: seriatim adapter is required when trim is enabled") + } + if env.Config.Pipeline.Scriptorium == nil { + return nil, fmt.Errorf("trim: pipeline.scriptorium is required when trim is enabled") + } + + boundsCfg := trimCfg.Bounds + boundsOutputPath, err := resolveScriptoriumOutputPath(paths, boundsCfg.OutputPath) + if err != nil { + return nil, fmt.Errorf("trim: resolve bounds output path: %w", err) + } + boundsStdoutLogPath := filepath.Join(paths.LogsDir, "scriptorium.bounds.stdout.log") + boundsStderrLogPath := filepath.Join(paths.LogsDir, "scriptorium.bounds.stderr.log") + boundsGeneratedConfigPath := filepath.Join(paths.ConfigDir, "scriptorium.bounds.generated.yml") + boundsTimeout, err := resolveScriptoriumTimeout(env.Config.Pipeline.Scriptorium.Timeout, boundsCfg.Timeout) + if err != nil { + return nil, fmt.Errorf("trim: resolve bounds timeout: %w", err) + } + + inputPaths := map[string]string{ + boundsCfg.TranscriptInputName: processedPath, + } + vars := map[string]string{} + + metadata["bounds_prompt_id"] = boundsCfg.PromptID + metadata["bounds_profile_id"] = boundsCfg.ProfileID + metadata["bounds_output_path"] = boundsOutputPath + metadata["bounds_timeout"] = boundsTimeout.String() + metadata["bounds_input_name"] = boundsCfg.TranscriptInputName + metadata["bounds_input_path"] = processedPath + metadata["bounds_render_debug_enabled"] = boundsCfg.RenderDebug + + renderOutputPath := "" + if boundsCfg.RenderDebug { + renderOutputPath, err = resolveScriptoriumOutputPath(paths, boundsCfg.RenderOutputPath) + if err != nil { + return nil, fmt.Errorf("trim: resolve bounds render output path: %w", err) + } + renderStdoutLogPath := filepath.Join(paths.LogsDir, "scriptorium.bounds.render.stdout.log") + renderStderrLogPath := filepath.Join(paths.LogsDir, "scriptorium.bounds.render.stderr.log") + renderGeneratedConfigPath := filepath.Join(paths.ConfigDir, "scriptorium.bounds.render.generated.yml") + + renderReq := scriptorium.RenderArtifactRequest{ + Binary: env.Config.Pipeline.Scriptorium.Binary, + ConfigPath: env.Config.Pipeline.Scriptorium.ConfigPath, + PromptID: boundsCfg.PromptID, + ProfileID: boundsCfg.ProfileID, + InputPaths: inputPaths, + Vars: vars, + OutputPath: renderOutputPath, + StdoutLogPath: renderStdoutLogPath, + StderrLogPath: renderStderrLogPath, + GeneratedConfigPath: renderGeneratedConfigPath, + Timeout: boundsTimeout, + } + renderRes, renderErr := env.Scriptorium.RenderArtifact(ctx, renderReq) + if renderErr != nil { + return nil, fmt.Errorf("trim: scriptorium bounds render failed: %w", renderErr) + } + if renderRes.ValidationFailed { + return nil, fmt.Errorf("trim: scriptorium bounds render returned validation_failed=true") + } + finalRenderOutputPath := coalesceString(renderRes.OutputPath, renderReq.OutputPath) + if err := requireNonEmptyFile(finalRenderOutputPath, "bounds render output"); err != nil { + return nil, fmt.Errorf("trim: %w", err) + } + if err := validateJSONFile(finalRenderOutputPath); err != nil { + return nil, fmt.Errorf("trim: bounds render diagnostics %q invalid json: %w", finalRenderOutputPath, err) + } + + logPaths = append(logPaths, renderStdoutLogPath, renderStderrLogPath) + generatedConfigs = append(generatedConfigs, renderGeneratedConfigPath) + metadata["bounds_render_output_path"] = finalRenderOutputPath + metadata["bounds_render_stdout_log_path"] = renderStdoutLogPath + metadata["bounds_render_stderr_log_path"] = renderStderrLogPath + metadata["bounds_render_generated_config_path"] = renderGeneratedConfigPath + metadata["bounds_render_adapter_exit_code"] = renderRes.ExitCode + metadata["bounds_render_adapter_duration_ms"] = renderRes.Duration.Milliseconds() + metadata["bounds_render_adapter_command_mode"] = renderRes.CommandMode + if renderRes.Metadata != nil { + metadata["bounds_render_adapter_metadata"] = renderRes.Metadata + } + } + + boundsReq := scriptorium.RunArtifactRequest{ + Binary: env.Config.Pipeline.Scriptorium.Binary, + ConfigPath: env.Config.Pipeline.Scriptorium.ConfigPath, + PromptID: boundsCfg.PromptID, + ProfileID: boundsCfg.ProfileID, + InputPaths: inputPaths, + Vars: vars, + OutputPath: boundsOutputPath, + StdoutLogPath: boundsStdoutLogPath, + StderrLogPath: boundsStderrLogPath, + GeneratedConfigPath: boundsGeneratedConfigPath, + Timeout: boundsTimeout, + } + + boundsRes, boundsRunErr := env.Scriptorium.RunArtifact(ctx, boundsReq) + if boundsRunErr != nil { + if boundsRes.ValidationFailed { + return nil, fmt.Errorf( + "trim: scriptorium bounds validation failed (prompt_id=%q, output_path=%q, exit_code=%d, stdout_log=%q, stderr_log=%q): %w", + boundsReq.PromptID, + coalesceString(boundsRes.OutputPath, boundsReq.OutputPath), + boundsRes.ExitCode, + coalesceString(boundsRes.StdoutLogPath, boundsReq.StdoutLogPath), + coalesceString(boundsRes.StderrLogPath, boundsReq.StderrLogPath), + boundsRunErr, + ) + } + return nil, fmt.Errorf("trim: scriptorium bounds generation failed: %w", boundsRunErr) + } + if boundsRes.ValidationFailed { + return nil, fmt.Errorf("trim: scriptorium bounds run returned validation_failed=true") + } + + finalBoundsOutputPath := coalesceString(boundsRes.OutputPath, boundsReq.OutputPath) + if err := requireNonEmptyFile(finalBoundsOutputPath, "session bounds output"); err != nil { + return nil, fmt.Errorf("trim: %w", err) + } + + boundsPayload, err := contracts.ParseSessionBoundsFile(finalBoundsOutputPath) + if err != nil { + return nil, fmt.Errorf("trim: parse bounds output %q: %w", finalBoundsOutputPath, err) + } + if err := contracts.ValidateSessionBoundsAgainstTranscript(boundsPayload, processedPath); err != nil { + return nil, fmt.Errorf("trim: validate bounds output %q against transcript %q: %w", finalBoundsOutputPath, processedPath, err) + } + + keepSelector, copyUnchanged, err := contracts.BuildSeriatimKeepSelector(boundsPayload) + if err != nil { + return nil, fmt.Errorf("trim: build keep selector from bounds output: %w", err) + } + trimAction := strings.ToLower(strings.TrimSpace(boundsPayload.TrimAction)) + if trimAction == "" { + trimAction = "trim" + } + + logPaths = append(logPaths, boundsStdoutLogPath, boundsStderrLogPath) + generatedConfigs = append(generatedConfigs, boundsGeneratedConfigPath) + metadata["bounds_confidence"] = boundsPayload.Confidence + metadata["trim_action"] = trimAction + metadata["start_segment_id"] = boundsPayload.StartSegmentID + metadata["end_segment_id"] = boundsPayload.EndSegmentID + metadata["warnings"] = boundsPayload.Warnings + metadata["keep_selector"] = keepSelector + metadata["bounds_output_path"] = finalBoundsOutputPath + metadata["bounds_stdout_log_path"] = boundsStdoutLogPath + metadata["bounds_stderr_log_path"] = boundsStderrLogPath + metadata["bounds_generated_config_path"] = boundsGeneratedConfigPath + metadata["bounds_adapter_exit_code"] = boundsRes.ExitCode + metadata["bounds_adapter_duration_ms"] = boundsRes.Duration.Milliseconds() + metadata["bounds_adapter_command_mode"] = boundsRes.CommandMode + metadata["bounds_adapter_prompt_id"] = boundsRes.PromptID + metadata["bounds_adapter_profile_id"] = boundsRes.ProfileID + metadata["bounds_adapter_output_path"] = boundsRes.OutputPath + metadata["bounds_adapter_generated_config"] = boundsRes.GeneratedConfigPath + metadata["bounds_adapter_stdout_log_path"] = boundsRes.StdoutLogPath + metadata["bounds_adapter_stderr_log_path"] = boundsRes.StderrLogPath + if boundsRes.Metadata != nil { + metadata["bounds_adapter_metadata"] = boundsRes.Metadata + } + + if copyUnchanged { + if err := copyTranscript(env.ArtifactStore, processedPath, trimmedPath); err != nil { + return nil, fmt.Errorf("trim: copy processed transcript to trimmed output: %w", err) + } + } else { + trimStdoutLogPath := filepath.Join(paths.LogsDir, "seriatim.trim.stdout.log") + trimStderrLogPath := filepath.Join(paths.LogsDir, "seriatim.trim.stderr.log") + trimGeneratedConfigPath := filepath.Join(paths.ConfigDir, "seriatim.trim.generated.yml") + trimTimeout, err := resolveTrimSeriatimTimeout(env.Config.Pipeline.Seriatim.Timeout) + if err != nil { + return nil, fmt.Errorf("trim: resolve seriatim timeout: %w", err) + } + trimReq := seriatim.TrimRequest{ + Binary: env.Config.Pipeline.Seriatim.Binary, + InputTranscriptPath: processedPath, + OutputTrimmedPath: trimmedPath, + KeepSelector: keepSelector, + StdoutLogPath: trimStdoutLogPath, + StderrLogPath: trimStderrLogPath, + GeneratedConfigPath: trimGeneratedConfigPath, + Timeout: trimTimeout, + } + trimRes, trimErr := env.Seriatim.Trim(ctx, trimReq) + if trimErr != nil { + return nil, fmt.Errorf("trim: seriatim trim failed: %w", trimErr) + } + logPaths = append(logPaths, trimStdoutLogPath, trimStderrLogPath) + generatedConfigs = append(generatedConfigs, trimGeneratedConfigPath) + metadata["seriatim_trim_stdout_log_path"] = trimStdoutLogPath + metadata["seriatim_trim_stderr_log_path"] = trimStderrLogPath + metadata["seriatim_trim_generated_config_path"] = trimGeneratedConfigPath + metadata["seriatim_trim_exit_code"] = trimRes.ExitCode + metadata["seriatim_trim_duration_ms"] = trimRes.Duration.Milliseconds() + metadata["seriatim_trim_invoked_binary"] = trimRes.InvokedBinary + metadata["seriatim_trim_keep_selector"] = trimRes.KeepSelector + if trimRes.Metadata != nil { + metadata["seriatim_trim_metadata"] = trimRes.Metadata + } + } + + if err := validateProcessedTranscriptOutput(trimmedPath); err != nil { + return nil, fmt.Errorf("trim: trimmed transcript %q invalid: %w", trimmedPath, err) + } + + outputs := []artifacts.Ref{ + { + Kind: "transcript_trimmed", + Category: "transcripts", + SessionID: sessionID, + AbsolutePath: trimmedPath, + }, + { + Kind: "session_bounds", + Category: "artifacts", + SessionID: sessionID, + AbsolutePath: finalBoundsOutputPath, + }, + } + if renderOutputPath != "" { + outputs = append(outputs, artifacts.Ref{ + Kind: "session_bounds_render", + Category: "artifacts", + SessionID: sessionID, + AbsolutePath: renderOutputPath, + }) + } + + return &StageResult{ + Outputs: outputs, + Logs: logPaths, + GeneratedConfigs: generatedConfigs, + Metadata: metadata, + }, nil +} + +func resolveTrimmedOutputPath(paths artifacts.SessionPaths, cfg *config.TrimConfig) (string, error) { + configured := "" + if cfg != nil { + configured = strings.TrimSpace(cfg.OutputPath) + } + if configured == "" { + return filepath.Join(paths.TranscriptsDir, "trimmed.json"), nil + } + return resolveScriptoriumOutputPath(paths, configured) +} + +func copyTranscript(store artifacts.Store, src, dst string) error { + data, err := os.ReadFile(src) + if err != nil { + return fmt.Errorf("read source transcript: %w", err) + } + if err := store.WriteFileAtomic(dst, data, 0o644); err != nil { + return fmt.Errorf("write destination transcript: %w", err) + } + return nil +} + +func resolveTrimSeriatimTimeout(raw string) (time.Duration, error) { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return 0, nil + } + d, err := time.ParseDuration(trimmed) + if err != nil { + return 0, fmt.Errorf("parse duration %q: %w", raw, err) + } + if d <= 0 { + return 0, fmt.Errorf("duration must be > 0") + } + return d, nil +} diff --git a/internal/stage/trim_test.go b/internal/stage/trim_test.go new file mode 100644 index 0000000..76a025e --- /dev/null +++ b/internal/stage/trim_test.go @@ -0,0 +1,405 @@ +package stage + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" + "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 TestTrimStageConsumesProcessedAndProducesTrimmedTranscript(t *testing.T) { + env, m, scr, ser := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + processed := filepath.Join(paths.TranscriptsDir, "processed.json") + writeFile(t, processed, `{"segments":[{"id":10},{"id":868}]}`) + m.MarkStageSucceeded("polish", time.Now().UTC(), []manifest.ArtifactRecord{ + {Kind: "transcript_processed", LocalPath: processed}, + }) + scr.BoundsBody = `{"confidence":"high","trim_action":"trim","start_segment_id":10,"end_segment_id":868,"warnings":[]}` + + result, err := (trimStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("trim.Run() error = %v", err) + } + + if len(scr.RunRequests) != 1 { + t.Fatalf("scriptorium run requests = %d, want 1", len(scr.RunRequests)) + } + if len(ser.TrimRequests) != 1 { + t.Fatalf("seriatim trim requests = %d, want 1", len(ser.TrimRequests)) + } + trimReq := ser.TrimRequests[0] + if trimReq.InputTranscriptPath != processed { + t.Fatalf("trim input = %q, want %q", trimReq.InputTranscriptPath, processed) + } + if trimReq.KeepSelector != "10-868" { + t.Fatalf("keep selector = %q, want %q", trimReq.KeepSelector, "10-868") + } + if trimReq.OutputTrimmedPath != filepath.Join(paths.TranscriptsDir, "trimmed.json") { + t.Fatalf("trim output = %q", trimReq.OutputTrimmedPath) + } + + if result == nil { + t.Fatal("trim result is nil") + } + if len(result.Outputs) < 2 { + t.Fatalf("outputs = %#v, want trimmed+bounds outputs", result.Outputs) + } + if result.Outputs[0].Kind != "transcript_trimmed" { + t.Fatalf("output[0] kind = %q, want transcript_trimmed", result.Outputs[0].Kind) + } + if result.Outputs[1].Kind != "session_bounds" { + t.Fatalf("output[1] kind = %q, want session_bounds", result.Outputs[1].Kind) + } + if result.Metadata["trimmed_output_path"] != filepath.Join(paths.TranscriptsDir, "trimmed.json") { + t.Fatalf("metadata trimmed_output_path = %#v", result.Metadata["trimmed_output_path"]) + } + if result.Metadata["keep_selector"] != "10-868" { + t.Fatalf("metadata keep_selector = %#v, want 10-868", result.Metadata["keep_selector"]) + } +} + +func TestTrimStageUsesConfiguredScriptoriumInputName(t *testing.T) { + env, m, scr, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + processed := filepath.Join(paths.TranscriptsDir, "processed.json") + writeFile(t, processed, `{"segments":[{"id":10},{"id":11}]}`) + scr.BoundsBody = `{"trim_action":"trim","start_segment_id":10,"end_segment_id":11}` + + cfg := env.Config.Pipeline.Trim + cfg.Bounds.TranscriptInputName = "transcript_body" + env.Config.Pipeline.Trim = cfg + + _, err := (trimStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("trim.Run() error = %v", err) + } + if len(scr.RunRequests) != 1 { + t.Fatalf("scriptorium run requests = %d, want 1", len(scr.RunRequests)) + } + req := scr.RunRequests[0] + if req.InputPaths["transcript_body"] != processed { + t.Fatalf("configured input path = %q, want %q", req.InputPaths["transcript_body"], processed) + } +} + +func TestTrimStageRecordsLogAndGeneratedConfigRefs(t *testing.T) { + env, m, scr, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) + scr.BoundsBody = `{"trim_action":"trim","start_segment_id":1,"end_segment_id":2}` + + result, err := (trimStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("trim.Run() error = %v", err) + } + logs := strings.Join(result.Logs, "\n") + if !strings.Contains(logs, "scriptorium.bounds.stdout.log") || !strings.Contains(logs, "seriatim.trim.stdout.log") { + t.Fatalf("logs = %#v, want scriptorium+seriatim logs", result.Logs) + } + generated := strings.Join(result.GeneratedConfigs, "\n") + if !strings.Contains(generated, "scriptorium.bounds.generated.yml") || !strings.Contains(generated, "seriatim.trim.generated.yml") { + t.Fatalf("generated configs = %#v, want scriptorium+seriatim generated configs", result.GeneratedConfigs) + } +} + +func TestTrimStageFailsWhenProcessedTranscriptMissing(t *testing.T) { + env, m, _, _ := setupTrimEnv(t) + _, err := (trimStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "processed transcript input is required") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestTrimStageFailsWhenProcessedTranscriptInvalidJSON(t *testing.T) { + env, m, _, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), "not-json") + _, err := (trimStage{}).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 TestTrimStageFailsWhenProcessedTranscriptMissingSegmentsArray(t *testing.T) { + env, m, _, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"schema":"audita.processed.v1"}`) + _, err := (trimStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "segments") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestTrimStageFailsWhenBoundsOutputInvalidJSON(t *testing.T) { + env, m, scr, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) + scr.BoundsBody = "not-json" + _, err := (trimStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "parse bounds output") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestTrimStageFailsWhenBoundsRangeIsDescending(t *testing.T) { + env, m, scr, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) + scr.BoundsBody = `{"trim_action":"trim","start_segment_id":2,"end_segment_id":1}` + _, err := (trimStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "<=") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestTrimStageFailsWhenBoundsIDsMissingFromTranscript(t *testing.T) { + env, m, scr, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":20},{"id":21}]}`) + scr.BoundsBody = `{"trim_action":"trim","start_segment_id":10,"end_segment_id":21}` + _, err := (trimStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "does not exist in transcript segments") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestTrimStageFailsWhenScriptoriumAdapterFails(t *testing.T) { + env, m, scr, _ := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) + scr.RunErr = errors.New("bounds failed") + _, err := (trimStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "scriptorium bounds generation failed") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestTrimStageFailsWhenSeriatimTrimAdapterFails(t *testing.T) { + env, m, scr, ser := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) + scr.BoundsBody = `{"trim_action":"trim","start_segment_id":1,"end_segment_id":2}` + ser.TrimErr = errors.New("trim failed") + _, err := (trimStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "seriatim trim failed") { + t.Fatalf("error = %q", err.Error()) + } +} + +func TestTrimStageDisabledCopiesProcessedTranscript(t *testing.T) { + env, m, scr, ser := setupTrimEnv(t) + paths := env.ArtifactStore.SessionPaths(m.SessionID) + processed := filepath.Join(paths.TranscriptsDir, "processed.json") + processedBody := `{"segments":[{"id":1,"text":"alpha"},{"id":2,"text":"beta"}]}` + writeFile(t, processed, processedBody) + + disabled := *env.Config.Pipeline.Trim + disabled.Enabled = false + env.Config.Pipeline.Trim = &disabled + + result, err := (trimStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("trim.Run() error = %v", err) + } + if len(scr.RunRequests) != 0 || len(ser.TrimRequests) != 0 { + t.Fatalf("scriptorium/seriatim should not be called when disabled; run=%d trim=%d", len(scr.RunRequests), len(ser.TrimRequests)) + } + if result.Metadata["trim_action"] != "copy_disabled" { + t.Fatalf("metadata trim_action = %#v, want copy_disabled", result.Metadata["trim_action"]) + } + trimmedPath := filepath.Join(paths.TranscriptsDir, "trimmed.json") + data, err := os.ReadFile(trimmedPath) + if err != nil { + t.Fatalf("read trimmed transcript: %v", err) + } + if string(data) != processedBody { + t.Fatalf("trimmed body = %q, want copied processed body", string(data)) + } +} + +type boundsScriptoriumRunner struct { + RunRequests []scriptorium.RunArtifactRequest + RenderRequests []scriptorium.RenderArtifactRequest + RunErr error + RenderErr error + BoundsBody string + RenderBody string +} + +func (r *boundsScriptoriumRunner) RunArtifact(_ context.Context, req scriptorium.RunArtifactRequest) (scriptorium.ArtifactResult, error) { + r.RunRequests = append(r.RunRequests, req) + if r.RunErr != nil { + return scriptorium.ArtifactResult{}, r.RunErr + } + body := strings.TrimSpace(r.BoundsBody) + if body == "" { + body = `{"trim_action":"trim","start_segment_id":1,"end_segment_id":1,"warnings":[]}` + } + if err := writeTrimFixtureFile(req.OutputPath, body); err != nil { + return scriptorium.ArtifactResult{}, err + } + if err := writeTrimFixtureFile(req.StdoutLogPath, "bounds stdout\n"); err != nil { + return scriptorium.ArtifactResult{}, err + } + if err := writeTrimFixtureFile(req.StderrLogPath, "bounds stderr\n"); err != nil { + return scriptorium.ArtifactResult{}, err + } + if req.GeneratedConfigPath != "" { + if err := writeTrimFixtureFile(req.GeneratedConfigPath, "schema: scriptorium.generated.v1\n"); err != nil { + return scriptorium.ArtifactResult{}, err + } + } + return scriptorium.ArtifactResult{ + OutputPath: req.OutputPath, + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, + GeneratedConfigPath: req.GeneratedConfigPath, + ExitCode: 0, + Duration: 35 * time.Millisecond, + CommandMode: scriptorium.CommandModeRun, + PromptID: req.PromptID, + ProfileID: req.ProfileID, + Metadata: map[string]any{"fake": true}, + }, nil +} + +func (r *boundsScriptoriumRunner) RenderArtifact(_ context.Context, req scriptorium.RenderArtifactRequest) (scriptorium.ArtifactResult, error) { + r.RenderRequests = append(r.RenderRequests, req) + if r.RenderErr != nil { + return scriptorium.ArtifactResult{}, r.RenderErr + } + body := strings.TrimSpace(r.RenderBody) + if body == "" { + body = `{"rendered":true}` + } + if err := writeTrimFixtureFile(req.OutputPath, body); err != nil { + return scriptorium.ArtifactResult{}, err + } + if err := writeTrimFixtureFile(req.StdoutLogPath, "render stdout\n"); err != nil { + return scriptorium.ArtifactResult{}, err + } + if err := writeTrimFixtureFile(req.StderrLogPath, "render stderr\n"); err != nil { + return scriptorium.ArtifactResult{}, err + } + if req.GeneratedConfigPath != "" { + if err := writeTrimFixtureFile(req.GeneratedConfigPath, "schema: scriptorium.generated.v1\n"); err != nil { + return scriptorium.ArtifactResult{}, err + } + } + return scriptorium.ArtifactResult{ + OutputPath: req.OutputPath, + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, + GeneratedConfigPath: req.GeneratedConfigPath, + ExitCode: 0, + Duration: 20 * time.Millisecond, + CommandMode: scriptorium.CommandModeRender, + PromptID: req.PromptID, + ProfileID: req.ProfileID, + Metadata: map[string]any{"fake": true}, + }, nil +} + +func setupTrimEnv(t *testing.T) (*Env, *manifest.Manifest, *boundsScriptoriumRunner, *seriatim.FakeRunner) { + 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") + + seriatimReport := false + 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: func() *float64 { v := 3.0; return &v }(), + Report: &seriatimReport, + }, + Trim: &config.TrimConfig{ + Enabled: true, + OutputPath: "transcripts/trimmed.json", + Bounds: config.TrimBoundsConfig{ + PromptID: "dnd_session.bounds", + ProfileID: "", + TranscriptInputName: "transcript", + OutputPath: "artifacts/session_bounds.json", + Timeout: "10m", + RenderDebug: false, + RenderOutputPath: "artifacts/session_bounds.render.json", + }, + Seriatim: config.TrimSeriatimConfig{ + Report: &seriatimReport, + }, + }, + Scriptorium: &config.ScriptoriumConfig{ + Binary: "scriptorium", + Timeout: "10m", + }, + }, + Session: &config.SessionConfig{ + SessionID: "2026-05-03", + }, + } + + store := artifacts.NewLocalStore(workspace) + if _, err := store.EnsureLayout("2026-05-03"); err != nil { + t.Fatalf("EnsureLayout() error = %v", err) + } + + scr := &boundsScriptoriumRunner{} + ser := &seriatim.FakeRunner{} + return &Env{ + Config: cfg, + ArtifactStore: store, + Scriptorium: scr, + Seriatim: ser, + }, manifest.New("2026-05-03", time.Now().UTC()), scr, ser +} + +func writeTrimFixtureFile(path string, contents string) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + return os.WriteFile(path, []byte(contents), 0o644) +}