eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Implemented run-local stage execution + immediate promotion for core output-producing stages

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-18 00:55:35 +00:00
parent 622677d038
commit cb525c0f72
8 changed files with 508 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
internal/app/runner_test.go
View File

@@ -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

41
internal/stage/analyze.go
View File

@@ -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,

80
internal/stage/merge.go
View File

@@ -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,

102
internal/stage/normalize.go
View File

@@ -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

63
internal/stage/polish.go
View File

@@ -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...),

132
internal/stage/run_local.go Normal file
View File

@@ -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
}

23
internal/stage/transcribe.go
View File

@@ -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,
},

90
internal/stage/trim.go
View File

@@ -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,