From cb525c0f72900ca0921e2c817262ed9c609c5cc5 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 18 May 2026 00:55:35 +0000 Subject: [PATCH] Implemented run-local stage execution + immediate promotion for core output-producing stages --- internal/app/runner_test.go | 82 ++++++++++++++++++++++ internal/stage/analyze.go | 41 ++++++++--- internal/stage/merge.go | 80 ++++++++++++++++----- internal/stage/normalize.go | 102 +++++++++++++++++---------- internal/stage/polish.go | 63 +++++++++++++---- internal/stage/run_local.go | 132 +++++++++++++++++++++++++++++++++++ internal/stage/transcribe.go | 23 +++++- internal/stage/trim.go | 90 +++++++++++++++++------- 8 files changed, 508 insertions(+), 105 deletions(-) create mode 100644 internal/stage/run_local.go diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 711c6ca..f6619de 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -398,6 +398,88 @@ func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) { } } +func TestExecuteStagesRunLocalArtifactsAndCanonicalPromotion(t *testing.T) { + cfg := testConfig(t) + stages := []stage.Stage{ + BuildFullPlan()[0], // prepare + BuildFullPlan()[1], // transcribe + BuildFullPlan()[2], // merge + BuildFullPlan()[3], // polish + BuildFullPlan()[4], // normalize + BuildFullPlan()[5], // trim + } + + summary, err := executeStages(context.Background(), cfg, stages, RunOptions{}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if summary.RunID == "" { + t.Fatal("run id must be set") + } + + runRoot := artifacts.SessionRunRootForCampaign( + cfg.Pipeline.Workspace.Root, + cfg.Session.Campaign, + cfg.Session.SessionID, + summary.RunID, + ) + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + + runLocalChecks := []string{ + filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "alice.json"), + filepath.Join(runRoot, "merge", "logs", "seriatim.stdout.log"), + filepath.Join(runRoot, "polish", "config", "audita.generated.yml"), + filepath.Join(runRoot, "normalize", "logs", "seriatim.normalize.stdout.log"), + filepath.Join(runRoot, "trim", "outputs", "transcripts", "trimmed.json"), + } + for _, p := range runLocalChecks { + if _, statErr := os.Stat(p); statErr != nil { + t.Fatalf("run-local artifact missing at %q: %v", p, statErr) + } + } + + canonicalChecks := []string{ + filepath.Join(paths.TranscriptsRawDir, "alice.json"), + filepath.Join(paths.TranscriptsDir, "merged.json"), + filepath.Join(paths.TranscriptsDir, "processed.json"), + filepath.Join(paths.TranscriptsDir, "normalized.json"), + filepath.Join(paths.TranscriptsDir, "trimmed.json"), + } + for _, p := range canonicalChecks { + if _, statErr := os.Stat(p); statErr != nil { + t.Fatalf("canonical promoted artifact missing at %q: %v", p, statErr) + } + } + + store := &manifest.LocalStore{} + sessionManifest, err := store.Load(context.Background(), summary.ManifestPath) + if err != nil { + t.Fatalf("Load manifest error = %v", err) + } + if got := sessionManifest.Stages["trim"]; got == nil || len(got.Outputs) == 0 { + t.Fatalf("trim stage outputs missing in session manifest: %#v", got) + } + for _, out := range sessionManifest.Stages["trim"].Outputs { + if strings.Contains(out.LocalPath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { + t.Fatalf("session manifest output should be canonical, got run-local path %q", out.LocalPath) + } + } + + runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath) + if err != nil { + t.Fatalf("LoadRun() error = %v", err) + } + mergeStage := runManifest.Stages["merge"] + if mergeStage == nil || len(mergeStage.Logs) == 0 { + t.Fatalf("merge logs missing in run manifest: %#v", mergeStage) + } + for _, logPath := range mergeStage.Logs { + if !strings.Contains(logPath, filepath.Join("runs", summary.RunID, "merge", "logs")) { + t.Fatalf("run manifest merge log path = %q, want run-local merge logs path", logPath) + } + } +} + func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { cases := []struct { name string diff --git a/internal/stage/analyze.go b/internal/stage/analyze.go index 4d427a0..68ab6b5 100644 --- a/internal/stage/analyze.go +++ b/internal/stage/analyze.go @@ -59,6 +59,10 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S } paths := sessionPathsForEnv(env, sessionID) + runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "analyze") + if err != nil { + return nil, fmt.Errorf("analyze: resolve run-stage layout: %w", err) + } if env.Config.Pipeline.Scriptorium == nil { return &StageResult{ Metadata: map[string]any{ @@ -130,13 +134,22 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("analyze: resolve vars: %w", err) } - outputPath, err := resolveScriptoriumOutputPath(paths, artifactCfg.OutputPath) + canonicalOutputPath, err := resolveScriptoriumOutputPath(paths, artifactCfg.OutputPath) if err != nil { return nil, fmt.Errorf("analyze: resolve output path: %w", err) } + outputPath, err := runLocalPathForCanonical(runLayout, paths, canonicalOutputPath) + if err != nil { + return nil, fmt.Errorf("analyze: resolve run-local output path: %w", err) + } stdoutLogPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".stdout.log") stderrLogPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".stderr.log") generatedConfigPath := filepath.Join(paths.ConfigDir, "scriptorium."+artifactName+".generated.yml") + if runLayout.Enabled { + stdoutLogPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".stdout.log") + stderrLogPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".stderr.log") + generatedConfigPath = filepath.Join(runLayout.ConfigDir, "scriptorium."+artifactName+".generated.yml") + } timeout, err := resolveScriptoriumTimeout(env.Config.Pipeline.Scriptorium.Timeout, artifactCfg.Timeout) if err != nil { @@ -168,9 +181,17 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if meta["render_debug_enabled"] == true { renderOutputPath := filepath.Join(paths.ArtifactsDir, artifactName+".render.json") + if runLayout.Enabled { + renderOutputPath = filepath.Join(runLayout.ReportsDir, artifactName+".render.json") + } renderStdoutPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".render.stdout.log") renderStderrPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".render.stderr.log") renderGeneratedConfigPath := filepath.Join(paths.ConfigDir, "scriptorium."+artifactName+".render.generated.yml") + if runLayout.Enabled { + renderStdoutPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".render.stdout.log") + renderStderrPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".render.stderr.log") + renderGeneratedConfigPath = filepath.Join(runLayout.ConfigDir, "scriptorium."+artifactName+".render.generated.yml") + } renderReq := scriptorium.RenderArtifactRequest{ Binary: env.Config.Pipeline.Scriptorium.Binary, @@ -257,17 +278,19 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err := requireNonEmptyFile(finalOutputPath, artifactName+" output"); err != nil { return nil, fmt.Errorf("analyze: %w", err) } - - artifactRef := artifacts.Ref{ - Kind: artifactName, - Category: "artifacts", - SessionID: sessionID, - AbsolutePath: finalOutputPath, + promotedArtifact, err := promoteRunLocalOutput(env.ArtifactStore, finalOutputPath, canonicalOutputPath, artifacts.Ref{ + Kind: artifactName, + Category: "artifacts", + SessionID: sessionID, + }) + if err != nil { + return nil, fmt.Errorf("analyze: promote artifact output: %w", err) } logPaths = append(logPaths, stdoutLogPath, stderrLogPath) generatedConfigs = append(generatedConfigs, generatedConfigPath) - meta["output_path"] = finalOutputPath + meta["run_output_path"] = finalOutputPath + meta["output_path"] = canonicalOutputPath meta["generated_config_path"] = generatedConfigPath meta["stdout_log_path"] = stdoutLogPath meta["stderr_log_path"] = stderrLogPath @@ -286,7 +309,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S } return &StageResult{ - Outputs: []artifacts.Ref{artifactRef}, + Outputs: []artifacts.Ref{promotedArtifact}, Logs: logPaths, GeneratedConfigs: generatedConfigs, Metadata: meta, diff --git a/internal/stage/merge.go b/internal/stage/merge.go index 7e0a046..af599ea 100644 --- a/internal/stage/merge.go +++ b/internal/stage/merge.go @@ -58,6 +58,10 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta } paths := sessionPathsForEnv(env, sessionID) + runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "merge") + if err != nil { + return nil, fmt.Errorf("merge: resolve run-stage layout: %w", err) + } inputs, err := discoverRawTranscripts(m, paths) if err != nil { @@ -81,13 +85,29 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta return nil, fmt.Errorf("merge: %w", err) } - mergedPath := filepath.Join(paths.TranscriptsDir, "merged.json") - reportPath := filepath.Join(paths.ArtifactsDir, "seriatim.report.json") + canonicalMergedPath := filepath.Join(paths.TranscriptsDir, "merged.json") + mergedPath, err := runLocalPathForCanonical(runLayout, paths, canonicalMergedPath) + if err != nil { + return nil, fmt.Errorf("merge: resolve run-local merged transcript path: %w", err) + } + canonicalReportPath := filepath.Join(paths.ArtifactsDir, "seriatim.report.json") + reportPath := canonicalReportPath + if runLayout.Enabled { + reportPath, err = runLocalPathForCanonical(runLayout, paths, canonicalReportPath) + if err != nil { + return nil, fmt.Errorf("merge: resolve run-local report path: %w", err) + } + } stdoutPath := filepath.Join(paths.LogsDir, "seriatim.stdout.log") stderrPath := filepath.Join(paths.LogsDir, "seriatim.stderr.log") genCfgPath := filepath.Join(paths.ConfigDir, "seriatim.generated.yml") + if runLayout.Enabled { + stdoutPath = filepath.Join(runLayout.LogsDir, "seriatim.stdout.log") + stderrPath = filepath.Join(runLayout.LogsDir, "seriatim.stderr.log") + genCfgPath = filepath.Join(runLayout.ConfigDir, "seriatim.generated.yml") + } - normalizedInputs, normalizeLogs, normalizeConfigs, normalizeMeta, err := normalizeMergeInputs(ctx, env, inputs, paths) + normalizedInputs, normalizeLogs, normalizeConfigs, normalizeMeta, err := normalizeMergeInputs(ctx, env, inputs, paths, runLayout) if err != nil { return nil, err } @@ -130,25 +150,35 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta } } - outputs := []artifacts.Ref{{ - Kind: "transcript_merged", - Category: "transcripts", - SessionID: sessionID, - AbsolutePath: finalMergedPath, - }} + promotedMerged, err := promoteRunLocalOutput(env.ArtifactStore, finalMergedPath, canonicalMergedPath, artifacts.Ref{ + Kind: "transcript_merged", + Category: "transcripts", + SessionID: sessionID, + }) + if err != nil { + return nil, fmt.Errorf("merge: promote merged transcript: %w", err) + } + outputs := []artifacts.Ref{promotedMerged} if reportEnabled { - outputs = append(outputs, artifacts.Ref{ - Kind: "seriatim_report", - Category: "artifacts", - SessionID: sessionID, - AbsolutePath: finalReportPath, + promotedReport, err := promoteRunLocalOutput(env.ArtifactStore, finalReportPath, canonicalReportPath, artifacts.Ref{ + Kind: "seriatim_report", + Category: "artifacts", + SessionID: sessionID, }) + if err != nil { + return nil, fmt.Errorf("merge: promote report: %w", err) + } + outputs = append(outputs, promotedReport) } coalesceGap := any(nil) if env.Config.Pipeline.Seriatim.CoalesceGap != nil { coalesceGap = *env.Config.Pipeline.Seriatim.CoalesceGap } + reportCanonicalPath := "" + if reportEnabled { + reportCanonicalPath = canonicalReportPath + } meta := map[string]any{ "stage": "merge", @@ -160,8 +190,10 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta "output_schema": env.Config.Pipeline.Seriatim.OutputSchema, "coalesce_gap": coalesceGap, "report_enabled": reportEnabled, - "output_path": finalMergedPath, - "report_path": finalReportPath, + "run_output_path": finalMergedPath, + "output_path": canonicalMergedPath, + "run_report_path": finalReportPath, + "report_path": reportCanonicalPath, "timeout": env.Config.Pipeline.Seriatim.Timeout, "binary": env.Config.Pipeline.Seriatim.Binary, "adapter_duration_ms": res.Duration.Milliseconds(), @@ -201,12 +233,21 @@ type normalizeMergeInputMeta struct { AdapterOutputPath string `json:"adapter_output_path,omitempty"` } -func normalizeMergeInputs(ctx context.Context, env *Env, rawInputs []string, paths artifacts.SessionPaths) ([]string, []string, []string, []normalizeMergeInputMeta, error) { +func normalizeMergeInputs( + ctx context.Context, + env *Env, + rawInputs []string, + paths artifacts.SessionPaths, + runLayout runStageLayout, +) ([]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)) normalizedDir := filepath.Join(paths.TranscriptsRawDir, "normalized") + if runLayout.Enabled { + normalizedDir = filepath.Join(runLayout.ScratchDir, "normalized") + } if err := os.MkdirAll(normalizedDir, 0o755); err != nil { return nil, nil, nil, nil, fmt.Errorf("merge: ensure normalized transcripts directory %q: %w", normalizedDir, err) } @@ -226,6 +267,11 @@ func normalizeMergeInputs(ctx context.Context, env *Env, rawInputs []string, pat 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") + if runLayout.Enabled { + stdoutPath = filepath.Join(runLayout.LogsDir, "seriatim.normalize."+base+".stdout.log") + stderrPath = filepath.Join(runLayout.LogsDir, "seriatim.normalize."+base+".stderr.log") + cfgPath = filepath.Join(runLayout.ConfigDir, "seriatim.normalize."+base+".generated.yml") + } req := seriatim.NormalizeRequest{ Binary: env.Config.Pipeline.Seriatim.Binary, diff --git a/internal/stage/normalize.go b/internal/stage/normalize.go index c28eef7..d116ec9 100644 --- a/internal/stage/normalize.go +++ b/internal/stage/normalize.go @@ -55,6 +55,10 @@ func (normalizeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) ( } paths := sessionPathsForEnv(env, sessionID) + runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "normalize") + if err != nil { + return nil, fmt.Errorf("normalize: resolve run-stage layout: %w", err) + } processedPath, processedSource, err := discoverProcessedTranscript(m, paths) if err != nil { return nil, fmt.Errorf("normalize: resolve processed transcript: %w", err) @@ -67,18 +71,34 @@ func (normalizeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) ( } normalizeCfg := normalizeConfigOrDefault(env.Config.Pipeline.Normalize) - normalizedPath, err := resolveScriptoriumOutputPath(paths, normalizeCfg.OutputPath) + canonicalNormalizedPath, err := resolveScriptoriumOutputPath(paths, normalizeCfg.OutputPath) if err != nil { return nil, fmt.Errorf("normalize: resolve normalized output path: %w", err) } + normalizedPath, err := runLocalPathForCanonical(runLayout, paths, canonicalNormalizedPath) + if err != nil { + return nil, fmt.Errorf("normalize: resolve run-local normalized output path: %w", err) + } reportEnabled := normalizeCfg.Report != nil && *normalizeCfg.Report reportPath := "" + canonicalReportPath := filepath.Join(paths.ArtifactsDir, "seriatim.normalize.report.json") if reportEnabled { - reportPath = filepath.Join(paths.ArtifactsDir, "seriatim.normalize.report.json") + reportPath = canonicalReportPath + if runLayout.Enabled { + reportPath, err = runLocalPathForCanonical(runLayout, paths, canonicalReportPath) + if err != nil { + return nil, fmt.Errorf("normalize: resolve run-local report path: %w", err) + } + } } stdoutPath := filepath.Join(paths.LogsDir, "seriatim.normalize.stdout.log") stderrPath := filepath.Join(paths.LogsDir, "seriatim.normalize.stderr.log") generatedConfigPath := filepath.Join(paths.ConfigDir, "seriatim.normalize.generated.yml") + if runLayout.Enabled { + stdoutPath = filepath.Join(runLayout.LogsDir, "seriatim.normalize.stdout.log") + stderrPath = filepath.Join(runLayout.LogsDir, "seriatim.normalize.stderr.log") + generatedConfigPath = filepath.Join(runLayout.ConfigDir, "seriatim.normalize.generated.yml") + } timeout, err := resolveTrimSeriatimTimeout(env.Config.Pipeline.Seriatim.Timeout) if err != nil { return nil, fmt.Errorf("normalize: resolve seriatim timeout: %w", err) @@ -113,44 +133,56 @@ func (normalizeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) ( } } - outputs := []artifacts.Ref{{ - Kind: "transcript_normalized", - Category: "transcripts", - SessionID: sessionID, - AbsolutePath: finalNormalizedPath, - }} + promotedNormalized, err := promoteRunLocalOutput(env.ArtifactStore, finalNormalizedPath, canonicalNormalizedPath, artifacts.Ref{ + Kind: "transcript_normalized", + Category: "transcripts", + SessionID: sessionID, + }) + if err != nil { + return nil, fmt.Errorf("normalize: promote normalized transcript: %w", err) + } + outputs := []artifacts.Ref{promotedNormalized} if reportEnabled { - outputs = append(outputs, artifacts.Ref{ - Kind: "seriatim_normalize_report", - Category: "artifacts", - SessionID: sessionID, - AbsolutePath: finalReportPath, + promotedReport, err := promoteRunLocalOutput(env.ArtifactStore, finalReportPath, canonicalReportPath, artifacts.Ref{ + Kind: "seriatim_normalize_report", + Category: "artifacts", + SessionID: sessionID, }) + if err != nil { + return nil, fmt.Errorf("normalize: promote report: %w", err) + } + outputs = append(outputs, promotedReport) + } + reportCanonicalPath := "" + if reportEnabled { + reportCanonicalPath = canonicalReportPath } meta := map[string]any{ - "stage": "normalize", - "processed_transcript_path": processedPath, - "processed_transcript_source": processedSource, - "normalized_transcript_path": finalNormalizedPath, - "normalized_transcript_source": "stage.normalize.output", - "output_schema": normalizeCfg.OutputSchema, - "report_enabled": reportEnabled, - "report_path": finalReportPath, - "timeout": env.Config.Pipeline.Seriatim.Timeout, - "binary": env.Config.Pipeline.Seriatim.Binary, - "stdout_log_path": stdoutPath, - "stderr_log_path": stderrPath, - "generated_config_path": generatedConfigPath, - "adapter_duration_ms": res.Duration.Milliseconds(), - "adapter_exit_code": res.ExitCode, - "adapter_invoked_binary": res.InvokedBinary, - "adapter_output_schema": res.OutputSchema, - "adapter_output_path": res.OutputNormalizedPath, - "adapter_report_path": res.ReportPath, - "adapter_generated_config": res.GeneratedConfigPath, - "adapter_stdout_log_path": res.StdoutLogPath, - "adapter_stderr_log_path": res.StderrLogPath, + "stage": "normalize", + "processed_transcript_path": processedPath, + "processed_transcript_source": processedSource, + "run_normalized_transcript_path": finalNormalizedPath, + "normalized_transcript_path": canonicalNormalizedPath, + "normalized_transcript_source": "stage.normalize.output", + "output_schema": normalizeCfg.OutputSchema, + "report_enabled": reportEnabled, + "run_report_path": finalReportPath, + "report_path": reportCanonicalPath, + "timeout": env.Config.Pipeline.Seriatim.Timeout, + "binary": env.Config.Pipeline.Seriatim.Binary, + "stdout_log_path": stdoutPath, + "stderr_log_path": stderrPath, + "generated_config_path": generatedConfigPath, + "adapter_duration_ms": res.Duration.Milliseconds(), + "adapter_exit_code": res.ExitCode, + "adapter_invoked_binary": res.InvokedBinary, + "adapter_output_schema": res.OutputSchema, + "adapter_output_path": res.OutputNormalizedPath, + "adapter_report_path": res.ReportPath, + "adapter_generated_config": res.GeneratedConfigPath, + "adapter_stdout_log_path": res.StdoutLogPath, + "adapter_stderr_log_path": res.StderrLogPath, } if res.Metadata != nil { meta["adapter_metadata"] = res.Metadata diff --git a/internal/stage/polish.go b/internal/stage/polish.go index 7e1b4f3..eeedf05 100644 --- a/internal/stage/polish.go +++ b/internal/stage/polish.go @@ -56,6 +56,10 @@ func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*St } paths := sessionPathsForEnv(env, sessionID) + runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "polish") + if err != nil { + return nil, fmt.Errorf("polish: resolve run-stage layout: %w", err) + } mergedPath, source, err := discoverMergedTranscript(m, paths) if err != nil { return nil, fmt.Errorf("polish: resolve merged transcript: %w", err) @@ -72,12 +76,29 @@ func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*St return nil, fmt.Errorf("polish: %w", err) } - processedPath := filepath.Join(paths.TranscriptsDir, "processed.json") - reportPath := filepath.Join(paths.ArtifactsDir, "audita.report.json") + canonicalProcessedPath := filepath.Join(paths.TranscriptsDir, "processed.json") + processedPath, err := runLocalPathForCanonical(runLayout, paths, canonicalProcessedPath) + if err != nil { + return nil, fmt.Errorf("polish: resolve run-local processed transcript path: %w", err) + } + canonicalReportPath := filepath.Join(paths.ArtifactsDir, "audita.report.json") + reportPath := canonicalReportPath + if runLayout.Enabled { + reportPath, err = runLocalPathForCanonical(runLayout, paths, canonicalReportPath) + if err != nil { + return nil, fmt.Errorf("polish: resolve run-local report path: %w", err) + } + } workDir := filepath.Join(paths.ArtifactsDir, "audita-work") stdoutPath := filepath.Join(paths.LogsDir, "audita.stdout.log") stderrPath := filepath.Join(paths.LogsDir, "audita.stderr.log") generatedConfigPath := filepath.Join(paths.ConfigDir, "audita.generated.yml") + if runLayout.Enabled { + workDir = filepath.Join(runLayout.ScratchDir, "audita-work") + stdoutPath = filepath.Join(runLayout.LogsDir, "audita.stdout.log") + stderrPath = filepath.Join(runLayout.LogsDir, "audita.stderr.log") + generatedConfigPath = filepath.Join(runLayout.ConfigDir, "audita.generated.yml") + } reportEnabled := env.Config.Pipeline.Audita.Report != nil && *env.Config.Pipeline.Audita.Report req := audita.PolishRequest{ @@ -128,19 +149,25 @@ func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*St } } - outputs := []artifacts.Ref{{ - Kind: "transcript_processed", - Category: "transcripts", - SessionID: sessionID, - AbsolutePath: finalProcessedPath, - }} + promotedProcessed, err := promoteRunLocalOutput(env.ArtifactStore, finalProcessedPath, canonicalProcessedPath, artifacts.Ref{ + Kind: "transcript_processed", + Category: "transcripts", + SessionID: sessionID, + }) + if err != nil { + return nil, fmt.Errorf("polish: promote processed transcript: %w", err) + } + outputs := []artifacts.Ref{promotedProcessed} if reportEnabled { - outputs = append(outputs, artifacts.Ref{ - Kind: "audita_report", - Category: "artifacts", - SessionID: sessionID, - AbsolutePath: finalReportPath, + promotedReport, err := promoteRunLocalOutput(env.ArtifactStore, finalReportPath, canonicalReportPath, artifacts.Ref{ + Kind: "audita_report", + Category: "artifacts", + SessionID: sessionID, }) + if err != nil { + return nil, fmt.Errorf("polish: promote report: %w", err) + } + outputs = append(outputs, promotedReport) } var validationConcurrency any @@ -155,14 +182,20 @@ func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*St if env.Config.Pipeline.Audita.ProposalLLMConcurrency != nil { proposalLLMConcurrency = *env.Config.Pipeline.Audita.ProposalLLMConcurrency } + reportCanonicalPath := "" + if reportEnabled { + reportCanonicalPath = canonicalReportPath + } meta := map[string]any{ "stage": "polish", "merged_transcript_path": mergedPath, "merged_transcript_source": source, "glossary_path": glossaryPath, - "output_path": finalProcessedPath, - "report_path": finalReportPath, + "run_output_path": finalProcessedPath, + "output_path": canonicalProcessedPath, + "run_report_path": finalReportPath, + "report_path": reportCanonicalPath, "audita_work_dir": workDir, "report_enabled": reportEnabled, "modules": append([]string(nil), req.Modules...), diff --git a/internal/stage/run_local.go b/internal/stage/run_local.go new file mode 100644 index 0000000..682bd22 --- /dev/null +++ b/internal/stage/run_local.go @@ -0,0 +1,132 @@ +package stage + +import ( + "fmt" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +type runStageLayout struct { + Enabled bool + Root string + OutputsDir string + LogsDir string + ReportsDir string + ConfigDir string + ScratchDir string +} + +func resolveRunStageLayout( + env *Env, + m *manifest.Manifest, + sessionPaths artifacts.SessionPaths, + sessionID, stageName string, +) (runStageLayout, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil { + return runStageLayout{}, fmt.Errorf("stage environment pipeline config is required") + } + if strings.TrimSpace(sessionID) == "" { + return runStageLayout{}, fmt.Errorf("session id is required") + } + stageName = strings.TrimSpace(stageName) + if stageName == "" { + return runStageLayout{}, fmt.Errorf("stage name is required") + } + + runID := "" + if m != nil { + runID = strings.TrimSpace(m.RunID) + } + campaign := strings.TrimSpace(env.Config.Session.Campaign) + if campaign == "" && m != nil { + campaign = strings.TrimSpace(m.Campaign) + } + + // Compatibility fallback for direct stage tests and older call paths + // that execute a stage without a run id. + if runID == "" || campaign == "" { + return runStageLayout{}, nil + } + + root := artifacts.SessionRunStageDirForCampaign( + env.Config.Pipeline.Workspace.Root, + campaign, + sessionID, + runID, + stageName, + ) + layout := runStageLayout{ + Enabled: true, + Root: root, + OutputsDir: filepath.Join(root, "outputs"), + LogsDir: filepath.Join(root, "logs"), + ReportsDir: filepath.Join(root, "reports"), + ConfigDir: filepath.Join(root, "config"), + ScratchDir: filepath.Join(root, "scratch"), + } + for _, dir := range []string{ + layout.Root, + layout.OutputsDir, + layout.LogsDir, + layout.ReportsDir, + layout.ConfigDir, + layout.ScratchDir, + } { + if err := os.MkdirAll(dir, 0o755); err != nil { + return runStageLayout{}, fmt.Errorf("create run-stage directory %q: %w", dir, err) + } + } + return layout, nil +} + +func runLocalPathForCanonical(layout runStageLayout, sessionPaths artifacts.SessionPaths, canonicalPath string) (string, error) { + if !layout.Enabled { + return filepath.Clean(canonicalPath), nil + } + cleanCanonical := filepath.Clean(strings.TrimSpace(canonicalPath)) + if cleanCanonical == "" { + return "", fmt.Errorf("canonical path is required") + } + rel, err := filepath.Rel(filepath.Clean(sessionPaths.Root), cleanCanonical) + if err != nil { + return "", fmt.Errorf("derive session-relative path for %q: %w", cleanCanonical, err) + } + rel = filepath.Clean(rel) + if rel == "." || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("canonical path %q is outside session root %q", cleanCanonical, sessionPaths.Root) + } + return filepath.Join(layout.OutputsDir, rel), nil +} + +func promoteRunLocalOutput( + store artifacts.Store, + srcPath, canonicalPath string, + ref artifacts.Ref, +) (artifacts.Ref, error) { + srcPath = filepath.Clean(strings.TrimSpace(srcPath)) + canonicalPath = filepath.Clean(strings.TrimSpace(canonicalPath)) + if srcPath == "" { + return artifacts.Ref{}, fmt.Errorf("source path is required") + } + if canonicalPath == "" { + return artifacts.Ref{}, fmt.Errorf("canonical destination path is required") + } + data, err := os.ReadFile(srcPath) + if err != nil { + return artifacts.Ref{}, fmt.Errorf("read run-local output %q: %w", srcPath, err) + } + if err := store.WriteFileAtomic(canonicalPath, data, 0o644); err != nil { + return artifacts.Ref{}, fmt.Errorf("promote output to %q: %w", canonicalPath, err) + } + checksum, err := store.Checksum(canonicalPath) + if err != nil { + return artifacts.Ref{}, fmt.Errorf("checksum promoted output %q: %w", canonicalPath, err) + } + ref.AbsolutePath = canonicalPath + ref.Checksum = checksum + return ref, nil +} diff --git a/internal/stage/transcribe.go b/internal/stage/transcribe.go index 636f463..46307ca 100644 --- a/internal/stage/transcribe.go +++ b/internal/stage/transcribe.go @@ -56,6 +56,10 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) } paths := sessionPathsForEnv(env, sessionID) + runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "transcribe") + if err != nil { + return nil, fmt.Errorf("transcribe: resolve run-stage layout: %w", err) + } audioFiles, err := discoverPreparedAudio(m, paths.AudioDir) if err != nil { return nil, fmt.Errorf("transcribe: resolve audio inputs: %w", err) @@ -88,10 +92,15 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) return nil, fmt.Errorf("transcribe: duplicate speaker/audio basename %q from %q and %q", base, prev, audioPath) } seenSpeaker[base] = audioPath + canonicalOut := filepath.Join(paths.TranscriptsRawDir, base+".json") + runOut, err := runLocalPathForCanonical(runLayout, paths, canonicalOut) + if err != nil { + return nil, fmt.Errorf("transcribe: resolve run-local output path for %q: %w", base, err) + } jobs = append(jobs, job{ speakerID: base, audioPath: audioPath, - outPath: filepath.Join(paths.TranscriptsRawDir, base+".json"), + outPath: runOut, }) } @@ -197,12 +206,19 @@ dispatch: sort.Strings(speakers) outputs := make([]artifacts.Ref, 0, len(speakers)) + runOutputPaths := make([]string, 0, len(speakers)) outputPaths := make([]string, 0, len(speakers)) orderedPerFile := make(map[string]any, len(speakers)) for _, speaker := range speakers { ref := outputRef[speaker] - outputs = append(outputs, ref) - outputPaths = append(outputPaths, ref.AbsolutePath) + runOutputPaths = append(runOutputPaths, ref.AbsolutePath) + canonicalOut := filepath.Join(paths.TranscriptsRawDir, speaker+".json") + promoted, err := promoteRunLocalOutput(env.ArtifactStore, ref.AbsolutePath, canonicalOut, ref) + if err != nil { + return nil, fmt.Errorf("transcribe: promote %q output: %w", speaker, err) + } + outputs = append(outputs, promoted) + outputPaths = append(outputPaths, canonicalOut) orderedPerFile[speaker] = perFile[speaker] } @@ -221,6 +237,7 @@ dispatch: "retries": retries, "retry_delay": env.Config.Pipeline.WhisperX.RetryDelay, "timeout": env.Config.Pipeline.WhisperX.Timeout, + "run_output_paths": runOutputPaths, "output_paths": outputPaths, "per_file": orderedPerFile, }, diff --git a/internal/stage/trim.go b/internal/stage/trim.go index 1f82706..bb1ea72 100644 --- a/internal/stage/trim.go +++ b/internal/stage/trim.go @@ -55,6 +55,10 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag } paths := sessionPathsForEnv(env, sessionID) + runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "trim") + if err != nil { + return nil, fmt.Errorf("trim: resolve run-stage layout: %w", err) + } normalizedPath, normalizedSource, err := discoverNormalizedTranscript(m, paths) if err != nil { return nil, fmt.Errorf("trim: resolve normalized transcript: %w", err) @@ -69,10 +73,14 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag trimCfg := env.Config.Pipeline.Trim enabled := trimCfg != nil && trimCfg.Enabled - trimmedPath, err := resolveTrimmedOutputPath(paths, trimCfg) + canonicalTrimmedPath, err := resolveTrimmedOutputPath(paths, trimCfg) if err != nil { return nil, fmt.Errorf("trim: resolve trimmed output path: %w", err) } + trimmedPath, err := runLocalPathForCanonical(runLayout, paths, canonicalTrimmedPath) + if err != nil { + return nil, fmt.Errorf("trim: resolve run-local trimmed output path: %w", err) + } logPaths := []string{} generatedConfigs := []string{} @@ -81,7 +89,8 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag "trim_enabled": enabled, "normalized_transcript_path": normalizedPath, "normalized_transcript_source": normalizedSource, - "trimmed_output_path": trimmedPath, + "run_trimmed_output_path": trimmedPath, + "trimmed_output_path": canonicalTrimmedPath, } if !enabled { @@ -91,14 +100,17 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag if err := validateProcessedTranscriptOutput(trimmedPath); err != nil { return nil, fmt.Errorf("trim: copied trimmed transcript %q invalid: %w", trimmedPath, err) } + promotedTrimmed, err := promoteRunLocalOutput(env.ArtifactStore, trimmedPath, canonicalTrimmedPath, artifacts.Ref{ + Kind: "transcript_trimmed", + Category: "transcripts", + SessionID: sessionID, + }) + if err != nil { + return nil, fmt.Errorf("trim: promote trimmed transcript: %w", err) + } metadata["trim_action"] = "copy_disabled" return &StageResult{ - Outputs: []artifacts.Ref{{ - Kind: "transcript_trimmed", - Category: "transcripts", - SessionID: sessionID, - AbsolutePath: trimmedPath, - }}, + Outputs: []artifacts.Ref{promotedTrimmed}, Metadata: metadata, }, nil } @@ -114,13 +126,22 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag } boundsCfg := trimCfg.Bounds - boundsOutputPath, err := resolveScriptoriumOutputPath(paths, boundsCfg.OutputPath) + canonicalBoundsOutputPath, err := resolveScriptoriumOutputPath(paths, boundsCfg.OutputPath) if err != nil { return nil, fmt.Errorf("trim: resolve bounds output path: %w", err) } + boundsOutputPath, err := runLocalPathForCanonical(runLayout, paths, canonicalBoundsOutputPath) + if err != nil { + return nil, fmt.Errorf("trim: resolve run-local 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") + if runLayout.Enabled { + boundsStdoutLogPath = filepath.Join(runLayout.LogsDir, "scriptorium.bounds.stdout.log") + boundsStderrLogPath = filepath.Join(runLayout.LogsDir, "scriptorium.bounds.stderr.log") + boundsGeneratedConfigPath = filepath.Join(runLayout.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) @@ -133,7 +154,8 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag metadata["bounds_prompt_id"] = boundsCfg.PromptID metadata["bounds_profile_id"] = boundsCfg.ProfileID - metadata["bounds_output_path"] = boundsOutputPath + metadata["run_bounds_output_path"] = boundsOutputPath + metadata["bounds_output_path"] = canonicalBoundsOutputPath metadata["bounds_timeout"] = boundsTimeout.String() metadata["bounds_input_name"] = boundsCfg.TranscriptInputName metadata["bounds_input_path"] = normalizedPath @@ -141,13 +163,22 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag renderOutputPath := "" if boundsCfg.RenderDebug { - renderOutputPath, err = resolveScriptoriumOutputPath(paths, boundsCfg.RenderOutputPath) + canonicalRenderOutputPath, err := resolveScriptoriumOutputPath(paths, boundsCfg.RenderOutputPath) if err != nil { return nil, fmt.Errorf("trim: resolve bounds render output path: %w", err) } + renderOutputPath, err = runLocalPathForCanonical(runLayout, paths, canonicalRenderOutputPath) + if err != nil { + return nil, fmt.Errorf("trim: resolve run-local 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") + if runLayout.Enabled { + renderStdoutLogPath = filepath.Join(runLayout.LogsDir, "scriptorium.bounds.render.stdout.log") + renderStderrLogPath = filepath.Join(runLayout.LogsDir, "scriptorium.bounds.render.stderr.log") + renderGeneratedConfigPath = filepath.Join(runLayout.ConfigDir, "scriptorium.bounds.render.generated.yml") + } renderReq := scriptorium.RenderArtifactRequest{ Binary: env.Config.Pipeline.Scriptorium.Binary, @@ -254,7 +285,7 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag metadata["end_segment_id"] = boundsPayload.EndSegmentID metadata["warnings"] = boundsPayload.Warnings metadata["keep_selector"] = keepSelector - metadata["bounds_output_path"] = finalBoundsOutputPath + metadata["run_bounds_output_path"] = finalBoundsOutputPath metadata["bounds_stdout_log_path"] = boundsStdoutLogPath metadata["bounds_stderr_log_path"] = boundsStderrLogPath metadata["bounds_generated_config_path"] = boundsGeneratedConfigPath @@ -279,6 +310,11 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag 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") + if runLayout.Enabled { + trimStdoutLogPath = filepath.Join(runLayout.LogsDir, "seriatim.trim.stdout.log") + trimStderrLogPath = filepath.Join(runLayout.LogsDir, "seriatim.trim.stderr.log") + trimGeneratedConfigPath = filepath.Join(runLayout.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) @@ -315,23 +351,25 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag 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, - }, + promotedTrimmed, err := promoteRunLocalOutput(env.ArtifactStore, trimmedPath, canonicalTrimmedPath, artifacts.Ref{ + Kind: "transcript_trimmed", + Category: "transcripts", + SessionID: sessionID, + }) + if err != nil { + return nil, fmt.Errorf("trim: promote trimmed transcript: %w", err) + } + promotedBounds, err := promoteRunLocalOutput(env.ArtifactStore, finalBoundsOutputPath, canonicalBoundsOutputPath, artifacts.Ref{ + Kind: "session_bounds", + Category: "artifacts", + SessionID: sessionID, + }) + if err != nil { + return nil, fmt.Errorf("trim: promote session bounds: %w", err) } return &StageResult{ - Outputs: outputs, + Outputs: []artifacts.Ref{promotedTrimmed, promotedBounds}, Logs: logPaths, GeneratedConfigs: generatedConfigs, Metadata: metadata,