From 7dc79e052f8dbc7550e191b600840ed4c4793405 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 18 May 2026 01:13:51 +0000 Subject: [PATCH] Aligned the archive stage with the new work directory layout --- internal/app/post_archive_cleanup.go | 10 +- internal/app/post_archive_cleanup_test.go | 23 +- internal/app/runner.go | 16 +- internal/app/runner_test.go | 44 ++++ internal/manifest/manifest.go | 6 +- internal/stage/analyze_test.go | 26 ++ internal/stage/archive.go | 280 ++++++++++------------ internal/stage/archive_test.go | 98 +++++--- internal/stage/merge_test.go | 33 +++ internal/stage/normalize_test.go | 27 +++ internal/stage/placeholders_test.go | 3 +- internal/stage/polish_test.go | 30 +++ internal/stage/prepare.go | 6 +- internal/stage/transcribe_test.go | 30 +++ internal/stage/trim_test.go | 33 +++ 15 files changed, 466 insertions(+), 199 deletions(-) diff --git a/internal/app/post_archive_cleanup.go b/internal/app/post_archive_cleanup.go index 28dc216..38df42a 100644 --- a/internal/app/post_archive_cleanup.go +++ b/internal/app/post_archive_cleanup.go @@ -54,12 +54,20 @@ func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m } workDir := strings.TrimSpace(m.LocalWorkDir) if workDir == "" { - workDir = artifacts.SessionRunWorkDir( + workDir = artifacts.SessionRunRootForCampaign( env.Config.Pipeline.Workspace.Root, strings.TrimSpace(env.Config.Session.Campaign), strings.TrimSpace(env.Config.Session.SessionID), strings.TrimSpace(m.RunID), ) + if info, err := os.Stat(workDir); err != nil || !info.IsDir() { + workDir = artifacts.SessionRunWorkDir( + env.Config.Pipeline.Workspace.Root, + strings.TrimSpace(env.Config.Session.Campaign), + strings.TrimSpace(env.Config.Session.SessionID), + strings.TrimSpace(m.RunID), + ) + } } if spoolRequested { diff --git a/internal/app/post_archive_cleanup_test.go b/internal/app/post_archive_cleanup_test.go index 5aba559..4fd193b 100644 --- a/internal/app/post_archive_cleanup_test.go +++ b/internal/app/post_archive_cleanup_test.go @@ -326,7 +326,11 @@ func archiveStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, stri {From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)}, }, } - writeArchiveFixtureRunFiles(t, seed.runWorkDir) + writeArchiveFixtureRunFiles( + t, + seed.runWorkDir, + artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID), + ) store := &manifest.LocalStore{} seedManifest, err := store.Load(context.Background(), manifestPathFor(cfg)) @@ -345,16 +349,19 @@ func archiveStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, stri return cfg, seed, runID } -func writeArchiveFixtureRunFiles(t *testing.T, runWorkDir string) { +func writeArchiveFixtureRunFiles(t *testing.T, runWorkDir, sessionRoot string) { t.Helper() - mustWriteFile(t, filepath.Join(runWorkDir, "inputs", "session.yml"), "session_id: 2026-05-03\n") - mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "raw", "speaker.json"), "{}\n") - mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "trimmed.json"), "{}\n") - mustWriteFile(t, filepath.Join(runWorkDir, "artifacts", "session_recap.md"), "# recap\n") - mustWriteFile(t, filepath.Join(runWorkDir, "reports", "audita.report.json"), "{}\n") - mustWriteFile(t, filepath.Join(runWorkDir, "config", "audita.generated.yml"), "key: value\n") + mustWriteFile(t, filepath.Join(runWorkDir, "prepare", "inputs", "session.yml"), "session_id: 2026-05-03\n") + mustWriteFile(t, filepath.Join(runWorkDir, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n") + mustWriteFile(t, filepath.Join(runWorkDir, "trim", "outputs", "transcripts", "trimmed.json"), "{}\n") + mustWriteFile(t, filepath.Join(runWorkDir, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n") + mustWriteFile(t, filepath.Join(runWorkDir, "polish", "reports", "audita.report.json"), "{}\n") + mustWriteFile(t, filepath.Join(runWorkDir, "merge", "config", "seriatim.generated.yml"), "key: value\n") mustWriteFile(t, filepath.Join(runWorkDir, "logs", "audita.stderr.log"), "stderr\n") mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n") + + mustWriteFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{}\n") + mustWriteFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n") } type failKeyStore struct { diff --git a/internal/app/runner.go b/internal/app/runner.go index a244476..abe3b8e 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -202,7 +202,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err) } - outputs := mapResultOutputs(result) + outputs := mapResultOutputs(result, runID) succeededAt := nowUTC() m.MarkStageSucceeded(s.Name(), succeededAt, outputs) applyStageResultToManifest(m, s.Name(), result) @@ -383,11 +383,12 @@ func fileExists(path string) (bool, error) { return false, err } -func mapResultOutputs(result *stage.StageResult) []manifest.ArtifactRecord { +func mapResultOutputs(result *stage.StageResult, runID string) []manifest.ArtifactRecord { if result == nil || len(result.Outputs) == 0 { return nil } + runID = strings.TrimSpace(runID) out := make([]manifest.ArtifactRecord, 0, len(result.Outputs)) for _, ref := range result.Outputs { localPath := ref.AbsolutePath @@ -395,10 +396,11 @@ func mapResultOutputs(result *stage.StageResult) []manifest.ArtifactRecord { localPath = ref.RelativePath } out = append(out, manifest.ArtifactRecord{ - Kind: ref.Kind, - LocalPath: localPath, - RemoteKey: ref.RemoteKey, - Checksum: ref.Checksum, + Kind: ref.Kind, + LocalPath: localPath, + ProducerRunID: runID, + RemoteKey: ref.RemoteKey, + Checksum: ref.Checksum, }) } @@ -446,7 +448,7 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest, runID stri changed = true } if m.LocalWorkDir == "" && campaign != "" && sessionID != "" && m.RunID != "" { - m.LocalWorkDir = artifacts.SessionRunWorkDir(cfg.Pipeline.Workspace.Root, campaign, sessionID, m.RunID) + m.LocalWorkDir = artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, campaign, sessionID, m.RunID) changed = true } if m.LocalSpoolDir == "" && campaign != "" && sessionID != "" && m.RunID != "" && strings.TrimSpace(cfg.Pipeline.Spool.Root) != "" { diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index f6619de..0641ce6 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -463,6 +463,9 @@ func TestExecuteStagesRunLocalArtifactsAndCanonicalPromotion(t *testing.T) { 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) } + if out.ProducerRunID != summary.RunID { + t.Fatalf("producer_run_id = %q, want %q", out.ProducerRunID, summary.RunID) + } } runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath) @@ -480,6 +483,47 @@ func TestExecuteStagesRunLocalArtifactsAndCanonicalPromotion(t *testing.T) { } } +func TestExecuteStagesSkippedStagePreservesExistingOutputsProvenance(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + manifestPath := manifestPathFor(cfg) + + existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) + existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{ + { + Kind: "transcript_raw", + LocalPath: "transcripts/raw/alice.json", + ProducerRunID: "20260501T000000Z-deadbeef", + }, + }) + if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { + t.Fatalf("MkdirAll() error = %v", err) + } + if err := store.Save(context.Background(), manifestPath, existing); err != nil { + t.Fatalf("Save manifest error = %v", err) + } + + summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if len(summary.Skipped) != 1 || summary.Skipped[0] != "transcribe" { + t.Fatalf("summary = %#v, want skipped transcribe", summary) + } + + loaded, err := store.Load(context.Background(), manifestPath) + if err != nil { + t.Fatalf("Load manifest error = %v", err) + } + got := loaded.Stages["transcribe"] + if got == nil || len(got.Outputs) != 1 { + t.Fatalf("transcribe outputs = %#v, want one preserved output", got) + } + if got.Outputs[0].ProducerRunID != "20260501T000000Z-deadbeef" { + t.Fatalf("producer_run_id = %q, want preserved value", got.Outputs[0].ProducerRunID) + } +} + func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { cases := []struct { name string diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index 2ae3c5b..28c3634 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -29,8 +29,10 @@ type InputRecord struct { type ArtifactRecord struct { Kind string `json:"kind"` LocalPath string `json:"local_path"` - RemoteKey string `json:"remote_key,omitempty"` - Checksum string `json:"checksum,omitempty"` + // ProducerRunID identifies the run that produced this durable artifact. + ProducerRunID string `json:"producer_run_id,omitempty"` + RemoteKey string `json:"remote_key,omitempty"` + Checksum string `json:"checksum,omitempty"` } // StageRecord tracks lifecycle and provenance for one pipeline stage. diff --git a/internal/stage/analyze_test.go b/internal/stage/analyze_test.go index 5093db8..9951e94 100644 --- a/internal/stage/analyze_test.go +++ b/internal/stage/analyze_test.go @@ -256,6 +256,32 @@ func TestAnalyzeOmitsOptionalPreviousRecapWhenUnavailable(t *testing.T) { } } +func TestAnalyzeUsesRunLocalPathsAndPromotesCanonical(t *testing.T) { + env, m, fake := setupAnalyzeEnv(t) + env.Config.Session.Campaign = "sample-campaign" + m.Campaign = "sample-campaign" + m.RunID = "20260518T010203Z-abcdef12" + paths := sessionPathsForEnv(env, m.SessionID) + writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) + + result, err := (analyzeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(fake.RunRequests) != 1 { + t.Fatalf("run requests = %d, want 1", len(fake.RunRequests)) + } + if !strings.Contains(fake.RunRequests[0].OutputPath, filepath.Join("runs", m.RunID, "analyze", "outputs")) { + t.Fatalf("run output path = %q, want run-local path", fake.RunRequests[0].OutputPath) + } + if len(result.Outputs) != 1 { + t.Fatalf("outputs len = %d, want 1", len(result.Outputs)) + } + if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { + t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath) + } +} + type orderedScriptoriumRunner struct { Calls []string RenderErr error diff --git a/internal/stage/archive.go b/internal/stage/archive.go index a1793ea..63280e9 100644 --- a/internal/stage/archive.go +++ b/internal/stage/archive.go @@ -33,15 +33,6 @@ var archivePrerequisiteStages = []string{ "analyze", } -var archiveRunUploadDirs = []string{ - "inputs", - "transcripts", - "artifacts", - "reports", - "config", - "logs", -} - func (archiveStage) Name() string { return "archive" } func (archiveStage) Declares() IODecl { @@ -87,16 +78,16 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("archive: remote object store backend is required when archive run upload is enabled") } - workDir, err := archiveWorkDir(env, m) + runRoot, err := resolveArchiveRunRoot(env, m) if err != nil { - return nil, fmt.Errorf("archive: resolve local workdir: %w", err) + return nil, fmt.Errorf("archive: resolve run root: %w", err) } - workDirInfo, err := os.Stat(workDir) + runRootInfo, err := os.Stat(runRoot) if err != nil { - return nil, fmt.Errorf("archive: local workdir %q: %w", workDir, err) + return nil, fmt.Errorf("archive: run root %q: %w", runRoot, err) } - if !workDirInfo.IsDir() { - return nil, fmt.Errorf("archive: local workdir %q is not a directory", workDir) + if !runRootInfo.IsDir() { + return nil, fmt.Errorf("archive: run root %q is not a directory", runRoot) } runPrefix, err := archiveRunPrefix(env, m) @@ -116,17 +107,20 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("archive: run id is required") } - manifestSource, err := resolveArchiveManifestSource(env, m, workDir) + manifestSource, err := resolveArchiveRunManifestSource(runRoot) if err != nil { - return nil, fmt.Errorf("archive: resolve manifest source: %w", err) + return nil, fmt.Errorf("archive: resolve run manifest source: %w", err) } - runFiles, err := collectArchiveRunFiles(workDir, manifestSource) + runFiles, err := collectArchiveRunFiles(runRoot, manifestSource) if err != nil { return nil, fmt.Errorf("archive: collect run files: %w", err) } - workDirCandidates := archiveSourceWorkDirs(env, m, workDir) - promotions, err := resolveArchivePromotions(workDirCandidates, env.Config.Pipeline.Archive.PromoteArtifacts) + sessionRoot, err := resolveArchiveSessionRoot(env, m) + if err != nil { + return nil, fmt.Errorf("archive: resolve session root for promotions: %w", err) + } + promotions, err := resolveArchivePromotions(sessionRoot, env.Config.Pipeline.Archive.PromoteArtifacts) if err != nil { return nil, fmt.Errorf("archive: resolve promotion rules: %w", err) } @@ -249,39 +243,67 @@ func validateArchivePrerequisites(m *manifest.Manifest) error { return nil } -func archiveWorkDir(env *Env, m *manifest.Manifest) (string, error) { - workDir := strings.TrimSpace(m.LocalWorkDir) - if workDir != "" { - cleaned := filepath.Clean(workDir) - if info, err := os.Stat(cleaned); err == nil && info.IsDir() { - return cleaned, nil - } - } - +func resolveArchiveRunRoot(env *Env, m *manifest.Manifest) (string, error) { sessionID := strings.TrimSpace(env.Config.Session.SessionID) - if sessionID == "" { + if sessionID == "" && m != nil { sessionID = strings.TrimSpace(m.SessionID) } campaign := strings.TrimSpace(env.Config.Session.Campaign) - if campaign == "" { + if campaign == "" && m != nil { campaign = strings.TrimSpace(m.Campaign) } - runID := strings.TrimSpace(m.RunID) + runID := "" + if m != nil { + runID = strings.TrimSpace(m.RunID) + } + if sessionID == "" || campaign == "" { + return "", fmt.Errorf("campaign and session id are required") + } if runID == "" { return "", fmt.Errorf("run id is required") } - if campaign == "" || sessionID == "" { - return "", fmt.Errorf("campaign and session id are required") + + canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)) + legacy := filepath.Clean(artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)) + + canonicalExists, err := directoryExists(canonical) + if err != nil { + return "", fmt.Errorf("check canonical run root %q: %w", canonical, err) } - runScoped := artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID) - if info, err := os.Stat(runScoped); err == nil && info.IsDir() { - return runScoped, nil + legacyExists, err := directoryExists(legacy) + if err != nil { + return "", fmt.Errorf("check legacy run root %q: %w", legacy, err) + } + + switch { + case canonicalExists && legacyExists && canonical != legacy: + return "", fmt.Errorf("ambiguous run roots for campaign %q session %q run %q: canonical=%q legacy=%q", campaign, sessionID, runID, canonical, legacy) + case canonicalExists: + return canonical, nil + case legacyExists: + return legacy, nil + default: + return "", fmt.Errorf("run root not found for campaign %q session %q run %q (checked canonical=%q legacy=%q)", campaign, sessionID, runID, canonical, legacy) + } +} + +func resolveArchiveSessionRoot(env *Env, m *manifest.Manifest) (string, error) { + sessionID := strings.TrimSpace(env.Config.Session.SessionID) + if sessionID == "" && m != nil { + sessionID = strings.TrimSpace(m.SessionID) + } + campaign := strings.TrimSpace(env.Config.Session.Campaign) + if campaign == "" && m != nil { + campaign = strings.TrimSpace(m.Campaign) + } + if sessionID == "" { + return "", fmt.Errorf("session id is required") } paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(campaign, sessionID) if err != nil { return "", err } - return paths.Root, nil + return filepath.Clean(paths.Root), nil } func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) { @@ -334,9 +356,10 @@ func archiveBucket(env *Env, m *manifest.Manifest) string { return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket) } -func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) { - if len(workDirs) == 0 { - return nil, fmt.Errorf("at least one workdir candidate is required") +func resolveArchivePromotions(sessionRoot string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) { + sessionRoot = filepath.Clean(strings.TrimSpace(sessionRoot)) + if sessionRoot == "" { + return nil, fmt.Errorf("session root is required") } out := make([]archivePromotion, 0, len(rules)) for _, rule := range rules { @@ -344,28 +367,16 @@ func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotion to := strings.TrimSpace(rule.To) required := rule.Required == nil || *rule.Required - var ( - localPath string - exists bool - ) - for _, candidateRoot := range workDirs { - resolvedPath, err := resolveWorkDirRelativePath(candidateRoot, from) - if err != nil { - return nil, fmt.Errorf("promotion from %q: %w", from, err) - } - info, err := os.Stat(resolvedPath) - if err == nil && !info.IsDir() { - localPath = resolvedPath - exists = true - break - } - if err != nil && !os.IsNotExist(err) { - return nil, fmt.Errorf("promotion source %q: %w", from, err) - } - if localPath == "" { - localPath = resolvedPath - } + resolvedPath, err := resolveWorkDirRelativePath(sessionRoot, from) + if err != nil { + return nil, fmt.Errorf("promotion from %q: %w", from, err) } + info, err := os.Stat(resolvedPath) + exists := err == nil && !info.IsDir() + if err != nil && !os.IsNotExist(err) { + return nil, fmt.Errorf("promotion source %q: %w", from, err) + } + localPath := resolvedPath out = append(out, archivePromotion{ From: from, @@ -378,35 +389,6 @@ func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotion return out, nil } -func archiveSourceWorkDirs(env *Env, m *manifest.Manifest, runWorkDir string) []string { - candidates := make([]string, 0, 2) - if strings.TrimSpace(runWorkDir) != "" { - candidates = append(candidates, filepath.Clean(runWorkDir)) - } - sessionID := strings.TrimSpace(env.Config.Session.SessionID) - if sessionID == "" && m != nil { - sessionID = strings.TrimSpace(m.SessionID) - } - if sessionID != "" { - paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(strings.TrimSpace(env.Config.Session.Campaign), sessionID) - if err == nil { - if cleanRoot := filepath.Clean(paths.Root); cleanRoot != "" { - candidates = append(candidates, cleanRoot) - } - } - } - seen := make(map[string]struct{}, len(candidates)) - out := make([]string, 0, len(candidates)) - for _, c := range candidates { - if _, ok := seen[c]; ok { - continue - } - seen[c] = struct{}{} - out = append(out, c) - } - return out -} - func resolveWorkDirRelativePath(workDir, rel string) (string, error) { rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel))) if rel == "." || rel == "" { @@ -425,42 +407,40 @@ func resolveWorkDirRelativePath(workDir, rel string) (string, error) { return cleanedFull, nil } -func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile, error) { +func collectArchiveRunFiles(runRoot, manifestPath string) ([]archiveUploadFile, error) { files := make([]archiveUploadFile, 0, 64) - - for _, dirName := range archiveRunUploadDirs { - fullDir := filepath.Join(workDir, dirName) - info, err := os.Stat(fullDir) - if err != nil { - if os.IsNotExist(err) { - continue - } - return nil, fmt.Errorf("stat %q: %w", fullDir, err) + err := filepath.WalkDir(runRoot, func(path string, d fs.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr } - if !info.IsDir() { - continue - } - - if err := filepath.WalkDir(fullDir, func(path string, d fs.DirEntry, walkErr error) error { - if walkErr != nil { - return walkErr - } - if d.IsDir() { + if d.IsDir() { + if path == runRoot { return nil } - rel, err := filepath.Rel(workDir, path) + relDir, err := filepath.Rel(runRoot, path) if err != nil { - return fmt.Errorf("relative path from %q to %q: %w", workDir, path, err) + return fmt.Errorf("relative dir from %q to %q: %w", runRoot, path, err) + } + relDir = filepath.ToSlash(relDir) + // Preserve existing behavior: audio is not uploaded in archive run record. + if relDir == "audio" || strings.HasPrefix(relDir, "audio/") { + return filepath.SkipDir } - rel = filepath.ToSlash(rel) - files = append(files, archiveUploadFile{ - RelativePath: rel, - LocalPath: path, - }) return nil - }); err != nil { - return nil, fmt.Errorf("walk %q: %w", fullDir, err) } + rel, err := filepath.Rel(runRoot, path) + if err != nil { + return fmt.Errorf("relative path from %q to %q: %w", runRoot, path, err) + } + rel = filepath.ToSlash(rel) + files = append(files, archiveUploadFile{ + RelativePath: rel, + LocalPath: path, + }) + return nil + }) + if err != nil { + return nil, fmt.Errorf("walk %q: %w", runRoot, err) } manifestInfo, err := os.Stat(manifestPath) @@ -477,6 +457,14 @@ func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile, RelativePath: "manifest.json", LocalPath: manifestPath, }) + seen := map[string]archiveUploadFile{} + for _, file := range files { + seen[file.RelativePath] = file + } + files = files[:0] + for _, file := range seen { + files = append(files, file) + } sort.Slice(files, func(i, j int) bool { return files[i].RelativePath < files[j].RelativePath @@ -484,42 +472,30 @@ func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile, return files, nil } -func resolveArchiveManifestSource(env *Env, m *manifest.Manifest, workDir string) (string, error) { - candidates := make([]string, 0, 3) - candidates = append(candidates, filepath.Join(workDir, "manifest.json")) +func resolveArchiveRunManifestSource(runRoot string) (string, error) { + path := filepath.Join(filepath.Clean(runRoot), "manifest.json") + info, err := os.Stat(path) + if err != nil { + if os.IsNotExist(err) { + return "", fmt.Errorf("manifest.json not found in run root %q", runRoot) + } + return "", fmt.Errorf("stat %q: %w", path, err) + } + if info.IsDir() { + return "", fmt.Errorf("manifest path %q is a directory", path) + } + return path, nil +} - sessionID := strings.TrimSpace(env.Config.Session.SessionID) - if sessionID == "" && m != nil { - sessionID = strings.TrimSpace(m.SessionID) +func directoryExists(path string) (bool, error) { + info, err := os.Stat(path) + if err == nil { + return info.IsDir(), nil } - if sessionID != "" { - paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(strings.TrimSpace(env.Config.Session.Campaign), sessionID) - if err != nil { - return "", err - } - candidates = append(candidates, paths.ManifestPath) + if os.IsNotExist(err) { + return false, nil } - - seen := make(map[string]struct{}, len(candidates)) - for _, candidate := range candidates { - clean := filepath.Clean(candidate) - if _, ok := seen[clean]; ok { - continue - } - seen[clean] = struct{}{} - info, err := os.Stat(clean) - if err != nil { - if os.IsNotExist(err) { - continue - } - return "", fmt.Errorf("stat %q: %w", clean, err) - } - if info.IsDir() { - continue - } - return clean, nil - } - return "", fmt.Errorf("manifest.json not found in run workdir or session workdir") + return false, err } func writeCurrentManifestSnapshot(m *manifest.Manifest, archiveMetadata map[string]any) (string, error) { diff --git a/internal/stage/archive_test.go b/internal/stage/archive_test.go index 1b2f6a1..dcade48 100644 --- a/internal/stage/archive_test.go +++ b/internal/stage/archive_test.go @@ -73,14 +73,15 @@ func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) { runPrefix := m.S3RunPrefix sessionPrefix := m.S3SessionPrefix wantRunUploads := []string{ - "artifacts/session_recap.md", - "config/audita.generated.yml", - "inputs/session.yml", + "analyze/outputs/artifacts/session_recap.md", + "merge/config/seriatim.generated.yml", + "prepare/inputs/session.yml", + "prepare/outputs/audio/speaker.flac", "logs/audita.stderr.log", "manifest.json", - "reports/audita.report.json", - "transcripts/raw/speaker.json", - "transcripts/trimmed.json", + "polish/reports/audita.report.json", + "transcribe/outputs/transcripts/raw/speaker.json", + "trim/outputs/transcripts/trimmed.json", } for _, rel := range wantRunUploads { key := runPrefix + rel @@ -132,12 +133,11 @@ func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) { } func TestArchiveUsesCustomPromotionRules(t *testing.T) { - env, m, workDir := archiveFixture(t) + env, m, _ := archiveFixture(t) env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ {From: "transcripts/trimmed.json", To: "published/trimmed.json", Required: boolPtr(true)}, {From: "artifacts/session_recap.md", To: "published/recap.md", Required: boolPtr(true)}, } - writeStageTestFile(t, filepath.Join(workDir, "published", "ignored.txt"), "ignore\n") _, err := archiveStage{}.Run(context.Background(), env, m) if err != nil { @@ -184,14 +184,9 @@ func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) { } func TestArchivePromotionFailsOnAmbiguousSessionRoots(t *testing.T) { - env, m, runWorkDir := archiveFixture(t) + env, m, _ := archiveFixture(t) sessionWorkDir := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, m.SessionID) - sessionTrimmed := filepath.Join(sessionWorkDir, "transcripts", "trimmed.json") - - if err := os.Remove(filepath.Join(runWorkDir, "transcripts", "trimmed.json")); err != nil { - t.Fatalf("remove run scoped trimmed transcript: %v", err) - } - writeStageTestFile(t, sessionTrimmed, "{}\n") + writeStageTestFile(t, filepath.Join(sessionWorkDir, "transcripts", "trimmed.json"), "{}\n") _, err := archiveStage{}.Run(context.Background(), env, m) if err == nil { @@ -202,6 +197,50 @@ func TestArchivePromotionFailsOnAmbiguousSessionRoots(t *testing.T) { } } +func TestArchiveFallsBackToLegacyRunRootWhenCanonicalMissing(t *testing.T) { + env, m, runRoot := archiveFixture(t) + legacyRoot := artifacts.SessionRunWorkDir( + env.Config.Pipeline.Workspace.Root, + env.Config.Session.Campaign, + env.Config.Session.SessionID, + m.RunID, + ) + if err := os.MkdirAll(filepath.Dir(legacyRoot), 0o755); err != nil { + t.Fatalf("create legacy run-root parent: %v", err) + } + if err := os.Rename(runRoot, legacyRoot); err != nil { + t.Fatalf("move canonical run root to legacy root: %v", err) + } + m.LocalWorkDir = legacyRoot + + if _, err := (archiveStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestArchiveFailsOnAmbiguousCanonicalAndLegacyRunRoots(t *testing.T) { + env, m, runRoot := archiveFixture(t) + legacyRoot := artifacts.SessionRunWorkDir( + env.Config.Pipeline.Workspace.Root, + env.Config.Session.Campaign, + env.Config.Session.SessionID, + m.RunID, + ) + writeStageTestFile(t, filepath.Join(legacyRoot, "manifest.json"), "{}\n") + writeStageTestFile(t, filepath.Join(legacyRoot, "prepare", "inputs", "session.yml"), "session_id: 2026-04-19\n") + if runRoot == legacyRoot { + t.Fatalf("test requires distinct canonical and legacy run roots, got %q", runRoot) + } + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected ambiguity error, got nil") + } + if !strings.Contains(err.Error(), "ambiguous run roots") { + t.Fatalf("error = %v, want run-root ambiguity", err) + } +} + func TestArchiveDoesNotWriteCurrentPointerWhenPromotionUploadFails(t *testing.T) { env, m, _ := archiveFixture(t) fake := env.ObjectStore.(*storage.FakeBackend) @@ -260,22 +299,27 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { runID := "20260516T010203Z-1a2b3c4d" campaign := "forsaken" sessionID := "2026-04-19" - workDir := filepath.Join(root, "work", campaign, sessionID, runID) + sessionRoot := artifacts.SessionWorkDirForCampaign(root, campaign, sessionID) + runRoot := artifacts.SessionRunRootForCampaign(root, campaign, sessionID, runID) - writeStageTestFile(t, filepath.Join(workDir, "inputs", "session.yml"), "session_id: 2026-04-19\n") - writeStageTestFile(t, filepath.Join(workDir, "transcripts", "raw", "speaker.json"), "{}\n") - writeStageTestFile(t, filepath.Join(workDir, "transcripts", "trimmed.json"), "{}\n") - writeStageTestFile(t, filepath.Join(workDir, "artifacts", "session_recap.md"), "# recap\n") - writeStageTestFile(t, filepath.Join(workDir, "reports", "audita.report.json"), "{}\n") - writeStageTestFile(t, filepath.Join(workDir, "config", "audita.generated.yml"), "key: value\n") - writeStageTestFile(t, filepath.Join(workDir, "logs", "audita.stderr.log"), "stderr\n") - writeStageTestFile(t, filepath.Join(workDir, "audio", "speaker.flac"), "flac") - writeStageTestFile(t, filepath.Join(workDir, "manifest.json"), "{}\n") + writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{}\n") + writeStageTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n") + + writeStageTestFile(t, filepath.Join(runRoot, "prepare", "inputs", "session.yml"), "session_id: 2026-04-19\n") + writeStageTestFile(t, filepath.Join(runRoot, "prepare", "outputs", "audio", "speaker.flac"), "flac\n") + writeStageTestFile(t, filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n") + writeStageTestFile(t, filepath.Join(runRoot, "trim", "outputs", "transcripts", "trimmed.json"), "{}\n") + writeStageTestFile(t, filepath.Join(runRoot, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n") + writeStageTestFile(t, filepath.Join(runRoot, "polish", "reports", "audita.report.json"), "{}\n") + writeStageTestFile(t, filepath.Join(runRoot, "merge", "config", "seriatim.generated.yml"), "key: value\n") + writeStageTestFile(t, filepath.Join(runRoot, "logs", "audita.stderr.log"), "stderr\n") + writeStageTestFile(t, filepath.Join(runRoot, "audio", "speaker.flac"), "flac") + writeStageTestFile(t, filepath.Join(runRoot, "manifest.json"), "{}\n") m := manifest.New(sessionID, time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC)) m.Campaign = campaign m.RunID = runID - m.LocalWorkDir = workDir + m.LocalWorkDir = runRoot m.S3Bucket = "my-dnd-archive" m.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", campaign, sessionID) m.S3RunPrefix = artifacts.S3RunPrefix(m.S3SessionPrefix, runID) @@ -309,7 +353,7 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { }, ObjectStore: &storage.FakeBackend{}, } - return env, m, workDir + return env, m, runRoot } type promotionFailingStore struct { diff --git a/internal/stage/merge_test.go b/internal/stage/merge_test.go index 2b51324..c1db7b3 100644 --- a/internal/stage/merge_test.go +++ b/internal/stage/merge_test.go @@ -289,6 +289,39 @@ func TestMergeStageFailsWhenNormalizedOutputInvalid(t *testing.T) { } } +func TestMergeStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) { + env, m := setupMergeEnv(t) + env.Config.Session.Campaign = "sample-campaign" + m.Campaign = "sample-campaign" + m.RunID = "20260518T010203Z-abcdef12" + paths := sessionPathsForEnv(env, m.SessionID) + + in := filepath.Join(paths.TranscriptsRawDir, "alice.json") + writeFile(t, in, `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + + fake := &seriatim.FakeRunner{} + env.Seriatim = fake + result, err := (mergeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("merge.Run() error = %v", err) + } + if len(fake.Requests) != 1 { + t.Fatalf("requests = %d, want 1", len(fake.Requests)) + } + req := fake.Requests[0] + if !strings.Contains(req.OutputMergedTranscriptPath, filepath.Join("runs", m.RunID, "merge", "outputs")) { + t.Fatalf("run output path = %q, want run-local path", req.OutputMergedTranscriptPath) + } + if len(result.Outputs) == 0 { + t.Fatalf("outputs = %#v, want promoted outputs", result.Outputs) + } + if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { + t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath) + } +} + func TestMergeStageResolvesSessionRelativeManifestOutputs(t *testing.T) { env, m := setupMergeEnv(t) paths := sessionPathsForEnv(env, m.SessionID) diff --git a/internal/stage/normalize_test.go b/internal/stage/normalize_test.go index e08a1a3..948ae34 100644 --- a/internal/stage/normalize_test.go +++ b/internal/stage/normalize_test.go @@ -211,6 +211,33 @@ func TestNormalizeStageReportEnabledFailsWhenReportMissing(t *testing.T) { } } +func TestNormalizeStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) { + env, m, ser := setupNormalizeEnv(t) + env.Config.Session.Campaign = "sample-campaign" + m.Campaign = "sample-campaign" + m.RunID = "20260518T010203Z-abcdef12" + paths := sessionPathsForEnv(env, m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`) + + result, err := (normalizeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("normalize.Run() error = %v", err) + } + if len(ser.NormalizeRequests) != 1 { + t.Fatalf("normalize requests = %d, want 1", len(ser.NormalizeRequests)) + } + req := ser.NormalizeRequests[0] + if !strings.Contains(req.OutputNormalizedPath, filepath.Join("runs", m.RunID, "normalize", "outputs")) { + t.Fatalf("run output path = %q, want run-local path", req.OutputNormalizedPath) + } + if len(result.Outputs) == 0 { + t.Fatalf("outputs = %#v, want promoted outputs", result.Outputs) + } + if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { + t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath) + } +} + func setupNormalizeEnv(t *testing.T) (*Env, *manifest.Manifest, *seriatim.FakeRunner) { t.Helper() workspace := t.TempDir() diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 91a9e59..3908233 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -64,6 +64,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { }, Session: &config.SessionConfig{ SessionID: "2026-05-03", + Campaign: "sample-campaign", Inputs: config.SessionInputsConfig{ AudioDir: "./audio", SpeakersFile: "./speakers.yml", @@ -88,7 +89,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } m.RunID = "20260516T000000Z-abcdef12" m.Campaign = "sample-campaign" - m.LocalWorkDir = filepath.Join(root, "work", "sample-campaign", "2026-05-03", m.RunID) + m.LocalWorkDir = artifacts.SessionRunRootForCampaign(root, "sample-campaign", "2026-05-03", m.RunID) m.S3RunPrefix = "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/" + m.RunID + "/" m.S3Bucket = "my-dnd-archive" if err := os.MkdirAll(filepath.Join(m.LocalWorkDir, "inputs"), 0o755); err != nil { diff --git a/internal/stage/polish_test.go b/internal/stage/polish_test.go index 83fe395..7ef63f4 100644 --- a/internal/stage/polish_test.go +++ b/internal/stage/polish_test.go @@ -238,6 +238,36 @@ func TestPolishStageFailsWhenReportInvalid(t *testing.T) { } } +func TestPolishStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) { + env, m := setupPolishEnv(t) + env.Config.Session.Campaign = "sample-campaign" + m.Campaign = "sample-campaign" + m.RunID = "20260518T010203Z-abcdef12" + paths := sessionPathsForEnv(env, m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") + + fake := &audita.FakeRunner{} + env.Audita = fake + result, err := (polishStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("polish.Run() error = %v", err) + } + if len(fake.Requests) != 1 { + t.Fatalf("requests = %d, want 1", len(fake.Requests)) + } + req := fake.Requests[0] + if !strings.Contains(req.OutputProcessedPath, filepath.Join("runs", m.RunID, "polish", "outputs")) { + t.Fatalf("run output path = %q, want run-local path", req.OutputProcessedPath) + } + if len(result.Outputs) == 0 { + t.Fatalf("outputs = %#v, want promoted outputs", result.Outputs) + } + if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { + t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath) + } +} + func setupPolishEnv(t *testing.T) (*Env, *manifest.Manifest) { t.Helper() workspace := t.TempDir() diff --git a/internal/stage/prepare.go b/internal/stage/prepare.go index eca9d9d..de0c695 100644 --- a/internal/stage/prepare.go +++ b/internal/stage/prepare.go @@ -366,7 +366,11 @@ func pathsWorkDirForManifest(env *Env, m *manifest.Manifest, sessionID string) s runID = strings.TrimSpace(m.RunID) } if runID != "" { - return artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, env.Config.Session.Campaign, sessionID, runID) + campaign := strings.TrimSpace(env.Config.Session.Campaign) + if campaign != "" { + return artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID) + } + return artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID) } campaign := strings.TrimSpace(env.Config.Session.Campaign) if campaign == "" { diff --git a/internal/stage/transcribe_test.go b/internal/stage/transcribe_test.go index 857997b..fec6415 100644 --- a/internal/stage/transcribe_test.go +++ b/internal/stage/transcribe_test.go @@ -191,6 +191,36 @@ func TestTranscribeStageInvalidJSONFails(t *testing.T) { } } +func TestTranscribeStageUsesRunLocalOutputAndPromotesCanonical(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac"}) + env.Config.Session.Campaign = "sample-campaign" + m.Campaign = "sample-campaign" + m.RunID = "20260518T010203Z-abcdef12" + fake := &whisperx.FakeClient{} + env.WhisperX = fake + + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + result, err := (transcribeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("transcribe.Run() error = %v", err) + } + if len(fake.Requests) != 1 { + t.Fatalf("requests = %d, want 1", len(fake.Requests)) + } + runOut := fake.Requests[0].OutputRawTranscriptPath + if !strings.Contains(runOut, filepath.Join("runs", m.RunID, "transcribe", "outputs")) { + t.Fatalf("run-local output path = %q, want runs/{run_id}/transcribe/outputs path", runOut) + } + if len(result.Outputs) != 1 { + t.Fatalf("outputs = %#v, want one output", result.Outputs) + } + if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { + t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath) + } +} + func setupTranscribeEnv(t *testing.T, audioFiles []string) (*Env, *manifest.Manifest) { t.Helper() diff --git a/internal/stage/trim_test.go b/internal/stage/trim_test.go index 62e5958..af7084d 100644 --- a/internal/stage/trim_test.go +++ b/internal/stage/trim_test.go @@ -298,6 +298,39 @@ func TestTrimStageDisabledCopiesNormalizedTranscript(t *testing.T) { } } +func TestTrimStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) { + env, m, scr, ser := setupTrimEnv(t) + env.Config.Session.Campaign = "sample-campaign" + m.Campaign = "sample-campaign" + m.RunID = "20260518T010203Z-abcdef12" + paths := sessionPathsForEnv(env, m.SessionID) + normalized := filepath.Join(paths.TranscriptsDir, "normalized.json") + writeFile(t, normalized, `{"segments":[{"id":10},{"id":11}]}`) + scr.BoundsBody = `{"trim_action":"trim","start_segment_id":10,"end_segment_id":11,"warnings":[]}` + + result, err := (trimStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("trim.Run() error = %v", err) + } + if len(scr.RunRequests) != 1 || len(ser.TrimRequests) != 1 { + t.Fatalf("scriptorium run=%d seriatim trim=%d, want 1/1", len(scr.RunRequests), len(ser.TrimRequests)) + } + if !strings.Contains(scr.RunRequests[0].OutputPath, filepath.Join("runs", m.RunID, "trim", "outputs")) { + t.Fatalf("bounds run output path = %q, want run-local path", scr.RunRequests[0].OutputPath) + } + if !strings.Contains(ser.TrimRequests[0].OutputTrimmedPath, filepath.Join("runs", m.RunID, "trim", "outputs")) { + t.Fatalf("trim output path = %q, want run-local path", ser.TrimRequests[0].OutputTrimmedPath) + } + if len(result.Outputs) < 2 { + t.Fatalf("outputs = %#v, want promoted trimmed+bounds outputs", result.Outputs) + } + for _, out := range result.Outputs { + if strings.Contains(out.AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { + t.Fatalf("promoted output path = %q, want canonical session path", out.AbsolutePath) + } + } +} + type boundsScriptoriumRunner struct { RunRequests []scriptorium.RunArtifactRequest RenderRequests []scriptorium.RenderArtifactRequest