diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index ee63021..5500dc8 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -111,7 +111,7 @@ func TestExecuteRunStageUnknownFails(t *testing.T) { func TestExecuteRunStageNormalizeIsAccepted(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot, "https://example.com/transcribe") - workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "processed.json"), `{"segments":[{"id":1}]}`) var stdout bytes.Buffer @@ -156,7 +156,7 @@ func TestExecuteRunStageTranscribeUsesConfiguredWhisperXServer(t *testing.T) { t.Fatal("expected whisperx server to be called at least once") } - outPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "transcripts", "raw", "alice.json") + outPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "transcripts", "raw", "alice.json") data, err := os.ReadFile(outPath) if err != nil { t.Fatalf("ReadFile(%q): %v", outPath, err) @@ -233,7 +233,7 @@ inputs: _ = os.Chdir(originalWD) }) - workRoot := filepath.Join(workspaceRoot, "work", sessionID) + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", sessionID) mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "merged.json"), `{"schema":"seriatim-intermediate","segments":[]}`) mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "glossary.yml"), "[]\n") diff --git a/internal/app/plan.go b/internal/app/plan.go index 23f936d..89f9b90 100644 --- a/internal/app/plan.go +++ b/internal/app/plan.go @@ -57,7 +57,7 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { } store := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) - paths, err := store.EnsureLayout(cfg.Session.SessionID) + paths, err := store.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if err != nil { return fmt.Errorf("plan: prepare workdir: %w", err) } diff --git a/internal/app/plan_test.go b/internal/app/plan_test.go index e2dfaef..a2afe54 100644 --- a/internal/app/plan_test.go +++ b/internal/app/plan_test.go @@ -36,7 +36,7 @@ func TestPlanCreatesAndReusesWorkdir(t *testing.T) { t.Fatalf("first output = %q, want totals", got) } - sessionWorkdir := artifacts.SessionWorkDir(workspaceRoot, "2026-05-03") + sessionWorkdir := artifacts.SessionWorkDirForCampaign(workspaceRoot, "sample-campaign", "2026-05-03") expectedDirs := []string{ sessionWorkdir, filepath.Join(sessionWorkdir, "inputs"), @@ -63,7 +63,7 @@ func TestPlanCreatesAndReusesWorkdir(t *testing.T) { func TestPlanShowsRunAndSkipFromManifest(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") store := &manifest.LocalStore{} m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) diff --git a/internal/app/resume.go b/internal/app/resume.go index c0629c5..2c312c3 100644 --- a/internal/app/resume.go +++ b/internal/app/resume.go @@ -6,6 +6,7 @@ import ( "fmt" "io" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) @@ -83,7 +84,12 @@ func Resume(ctx context.Context, args []string, out io.Writer) error { } func loadManifestIfPresent(ctx context.Context, cfg *config.Config) (*manifest.Manifest, error) { - path := manifestPathFor(cfg) + localStore := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) + paths, err := localStore.ResolveSessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + if err != nil { + return nil, fmt.Errorf("resolve session workspace paths: %w", err) + } + path := paths.ManifestPath exists, err := fileExists(path) if err != nil { return nil, fmt.Errorf("check manifest %q: %w", path, err) diff --git a/internal/app/resume_run_stage_test.go b/internal/app/resume_run_stage_test.go index 13afb58..9e24fed 100644 --- a/internal/app/resume_run_stage_test.go +++ b/internal/app/resume_run_stage_test.go @@ -16,7 +16,7 @@ import ( func TestResumeStartsAfterCompletedStages(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") store := &manifest.LocalStore{} m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) @@ -25,7 +25,7 @@ func TestResumeStartsAfterCompletedStages(t *testing.T) { if err := store.Save(context.Background(), manifestPath, m); err != nil { t.Fatalf("save manifest: %v", err) } - workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "raw", "alice.json"), `{"segments":[]}`) mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "speakers.yml"), "match:\n - speaker: Alice\n match: [\"alice\"]\n") mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "autocorrect.yml"), "[]\n") @@ -52,7 +52,7 @@ func TestResumeStartsAfterCompletedStages(t *testing.T) { func TestResumeNoRemainingStages(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") store := &manifest.LocalStore{} m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) @@ -81,7 +81,7 @@ func TestResumeForceRerunsSucceeded(t *testing.T) { })) defer srv.Close() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot, srv.URL) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") store := &manifest.LocalStore{} m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) @@ -105,8 +105,8 @@ func TestResumeForceRerunsSucceeded(t *testing.T) { func TestRunStageExecutesOnlySelectedStage(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") - workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "merged.json"), `{"segments":[]}`) mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "glossary.yml"), "terms: []\n") @@ -135,8 +135,8 @@ func TestRunStageExecutesOnlySelectedStage(t *testing.T) { func TestRunStageSkipAndForce(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") - workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "merged.json"), `{"segments":[]}`) mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "glossary.yml"), "terms: []\n") @@ -169,8 +169,8 @@ func TestRunStageSkipAndForce(t *testing.T) { func TestRunStageTrimExecutes(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") - workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "normalized.json"), `{"segments":[{"id":1},{"id":2}]}`) var out bytes.Buffer @@ -198,8 +198,8 @@ func TestRunStageTrimExecutes(t *testing.T) { func TestRunStageNormalizeExecutes(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) - manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") - workRoot := filepath.Join(workspaceRoot, "work", "2026-05-03") + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) var out bytes.Buffer diff --git a/internal/app/runner.go b/internal/app/runner.go index 6efdf0d..1963636 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -5,7 +5,6 @@ import ( "fmt" "log/slog" "os" - "path/filepath" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" @@ -93,12 +92,12 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage } artifactStore := env.ArtifactStore - paths, err := artifactStore.EnsureLayout(cfg.Session.SessionID) + paths, err := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if err != nil { return nil, fmt.Errorf("prepare workdir: %w", err) } - lock, err := artifactStore.AcquireSessionLock(cfg.Session.SessionID) + lock, err := artifactStore.AcquireSessionLockFor(cfg.Session.Campaign, cfg.Session.SessionID) if err != nil { return nil, fmt.Errorf("acquire session lock: %w", err) } @@ -411,7 +410,11 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err } func manifestPathFor(cfg *config.Config) string { - return filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.SessionID, "manifest.json") + return artifacts.SessionManifestPathForCampaign( + cfg.Pipeline.Workspace.Root, + cfg.Session.Campaign, + cfg.Session.SessionID, + ) } func needsObjectStoreForRun(cfg *config.Config, stages []stage.Stage) bool { diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index d729252..5e6863f 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -280,7 +280,14 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) { existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) existing.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) - audioPath := filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.SessionID, "audio", "alice.flac") + audioPath := filepath.Join( + cfg.Pipeline.Workspace.Root, + "work", + cfg.Session.Campaign, + cfg.Session.SessionID, + "audio", + "alice.flac", + ) if err := os.MkdirAll(filepath.Dir(audioPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } @@ -341,7 +348,7 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { tc.env.ArtifactStore = artifactStore tc.env.ManifestStore = &manifest.LocalStore{} if tc.name == "transcribe" { - paths, ensureErr := artifactStore.EnsureLayout(cfg.Session.SessionID) + paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } @@ -356,7 +363,7 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { } } if tc.name == "merge" { - paths, ensureErr := artifactStore.EnsureLayout(cfg.Session.SessionID) + paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } @@ -375,7 +382,7 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { } } if tc.name == "polish" { - paths, ensureErr := artifactStore.EnsureLayout(cfg.Session.SessionID) + paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } @@ -387,7 +394,7 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { } } if tc.name == "analyze" { - paths, ensureErr := artifactStore.EnsureLayout(cfg.Session.SessionID) + paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } diff --git a/internal/app/whisperx_wiring_test.go b/internal/app/whisperx_wiring_test.go index 4558d19..268cb57 100644 --- a/internal/app/whisperx_wiring_test.go +++ b/internal/app/whisperx_wiring_test.go @@ -63,7 +63,7 @@ func TestExecuteStagesDefaultWiringUsesWhisperXHTTPClient(t *testing.T) { t.Fatal("audio file payload was empty") } - outPath := filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.SessionID, "transcripts", "raw", "alice.json") + outPath := filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.Campaign, cfg.Session.SessionID, "transcripts", "raw", "alice.json") data, err := os.ReadFile(outPath) if err != nil { t.Fatalf("ReadFile(%q) error = %v", outPath, err) diff --git a/internal/artifacts/local.go b/internal/artifacts/local.go index 0f4b822..187cc1d 100644 --- a/internal/artifacts/local.go +++ b/internal/artifacts/local.go @@ -30,13 +30,18 @@ func NewLocalStore(workspaceRoot string) *LocalStore { return &LocalStore{WorkspaceRoot: workspaceRoot} } -// SessionPaths resolves canonical paths for a session workdir. +// SessionPaths resolves legacy paths for a session workdir. func (s *LocalStore) SessionPaths(sessionID string) SessionPaths { - return buildSessionPaths(s.WorkspaceRoot, sessionID) + return buildLegacySessionPaths(s.WorkspaceRoot, sessionID) } -// EnsureLayout creates and verifies the canonical session workdir directory layout. -func (s *LocalStore) EnsureLayout(sessionID string) (SessionPaths, error) { +// SessionPathsFor resolves canonical campaign-aware paths for a session workdir. +func (s *LocalStore) SessionPathsFor(campaign, sessionID string) SessionPaths { + return buildSessionPaths(s.WorkspaceRoot, campaign, sessionID) +} + +// ResolveSessionPathsFor resolves the active session path with legacy compatibility. +func (s *LocalStore) ResolveSessionPathsFor(campaign, sessionID string) (SessionPaths, error) { if strings.TrimSpace(s.WorkspaceRoot) == "" { return SessionPaths{}, fmt.Errorf("workspace root is required") } @@ -44,7 +49,62 @@ func (s *LocalStore) EnsureLayout(sessionID string) (SessionPaths, error) { return SessionPaths{}, fmt.Errorf("sessionID is required") } - paths := s.SessionPaths(sessionID) + campaign = strings.TrimSpace(campaign) + if campaign == "" { + return s.SessionPaths(sessionID), nil + } + + canonical := s.SessionPathsFor(campaign, sessionID) + legacy := s.SessionPaths(sessionID) + canonicalExists, err := dirExists(canonical.Root) + if err != nil { + return SessionPaths{}, fmt.Errorf("check canonical session root %q: %w", canonical.Root, err) + } + legacyExists, err := dirExists(legacy.Root) + if err != nil { + return SessionPaths{}, fmt.Errorf("check legacy session root %q: %w", legacy.Root, err) + } + + switch { + case canonicalExists && legacyExists: + return SessionPaths{}, fmt.Errorf( + "ambiguous session workspace roots for campaign %q session %q: canonical=%q legacy=%q", + campaign, + sessionID, + canonical.Root, + legacy.Root, + ) + case canonicalExists: + return canonical, nil + case legacyExists: + return legacy, nil + default: + return canonical, nil + } +} + +// EnsureLayout creates and verifies the canonical session workdir directory layout. +func (s *LocalStore) EnsureLayout(sessionID string) (SessionPaths, error) { + return s.ensureLayout(s.SessionPaths(sessionID)) +} + +// EnsureLayoutFor creates and verifies campaign-aware session layout, with controlled legacy compatibility. +func (s *LocalStore) EnsureLayoutFor(campaign, sessionID string) (SessionPaths, error) { + paths, err := s.ResolveSessionPathsFor(campaign, sessionID) + if err != nil { + return SessionPaths{}, err + } + return s.ensureLayout(paths) +} + +func (s *LocalStore) ensureLayout(paths SessionPaths) (SessionPaths, error) { + if strings.TrimSpace(s.WorkspaceRoot) == "" { + return SessionPaths{}, fmt.Errorf("workspace root is required") + } + if strings.TrimSpace(paths.SessionID) == "" { + return SessionPaths{}, fmt.Errorf("sessionID is required") + } + dirs := []string{ paths.Root, paths.InputsDir, @@ -53,8 +113,11 @@ func (s *LocalStore) EnsureLayout(sessionID string) (SessionPaths, error) { paths.TranscriptsRawDir, paths.TranscriptsTrimmedDir, paths.ArtifactsDir, + paths.ReportsDir, paths.ConfigDir, paths.LogsDir, + paths.CurrentDir, + paths.RunsDir, } for _, dir := range dirs { @@ -72,7 +135,19 @@ func (s *LocalStore) CopyInput(sessionID, srcPath, destRelativePath string) (Ref if err != nil { return Ref{}, err } + return s.copyInputWithPaths(paths, sessionID, srcPath, destRelativePath) +} +// CopyInputFor copies an input file into the campaign-aware session workdir under destRelativePath. +func (s *LocalStore) CopyInputFor(campaign, sessionID, srcPath, destRelativePath string) (Ref, error) { + paths, err := s.EnsureLayoutFor(campaign, sessionID) + if err != nil { + return Ref{}, err + } + return s.copyInputWithPaths(paths, sessionID, srcPath, destRelativePath) +} + +func (s *LocalStore) copyInputWithPaths(paths SessionPaths, sessionID, srcPath, destRelativePath string) (Ref, error) { destAbs, err := resolveInRoot(paths.Root, destRelativePath) if err != nil { return Ref{}, fmt.Errorf("copy input: %w", err) @@ -179,7 +254,19 @@ func (s *LocalStore) AcquireSessionLock(sessionID string) (*LockHandle, error) { if err != nil { return nil, err } + return s.acquireSessionLockForPaths(paths) +} +// AcquireSessionLockFor acquires an exclusive lock file for a campaign/session workdir. +func (s *LocalStore) AcquireSessionLockFor(campaign, sessionID string) (*LockHandle, error) { + paths, err := s.EnsureLayoutFor(campaign, sessionID) + if err != nil { + return nil, err + } + return s.acquireSessionLockForPaths(paths) +} + +func (s *LocalStore) acquireSessionLockForPaths(paths SessionPaths) (*LockHandle, error) { f, err := os.OpenFile(paths.LockPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644) if err != nil { if errors.Is(err, os.ErrExist) { @@ -203,6 +290,17 @@ func (s *LocalStore) AcquireSessionLock(sessionID string) (*LockHandle, error) { return &LockHandle{path: paths.LockPath, file: f}, nil } +func dirExists(path string) (bool, error) { + info, err := os.Stat(path) + if err == nil { + return info.IsDir(), nil + } + if errors.Is(err, os.ErrNotExist) { + return false, nil + } + return false, err +} + // ReleaseSessionLock releases a previously acquired session lock. func (s *LocalStore) ReleaseSessionLock(lock *LockHandle) error { if lock == nil { diff --git a/internal/artifacts/local_test.go b/internal/artifacts/local_test.go index e6b8d9f..07ae95d 100644 --- a/internal/artifacts/local_test.go +++ b/internal/artifacts/local_test.go @@ -10,9 +10,9 @@ import ( func TestEnsureLayoutCreatesExpectedDirectories(t *testing.T) { store := NewLocalStore(t.TempDir()) - paths, err := store.EnsureLayout("session-1") + paths, err := store.EnsureLayoutFor("sample-campaign", "session-1") if err != nil { - t.Fatalf("EnsureLayout() error = %v", err) + t.Fatalf("EnsureLayoutFor() error = %v", err) } checkDirExists(t, paths.Root) @@ -22,8 +22,11 @@ func TestEnsureLayoutCreatesExpectedDirectories(t *testing.T) { checkDirExists(t, paths.TranscriptsRawDir) checkDirExists(t, paths.TranscriptsTrimmedDir) checkDirExists(t, paths.ArtifactsDir) + checkDirExists(t, paths.ReportsDir) checkDirExists(t, paths.ConfigDir) checkDirExists(t, paths.LogsDir) + checkDirExists(t, paths.CurrentDir) + checkDirExists(t, paths.RunsDir) if filepath.Base(paths.ManifestPath) != "manifest.json" { t.Fatalf("ManifestPath = %q, want basename manifest.json", paths.ManifestPath) @@ -33,6 +36,43 @@ func TestEnsureLayoutCreatesExpectedDirectories(t *testing.T) { } } +func TestResolveSessionPathsForLegacyFallback(t *testing.T) { + root := t.TempDir() + store := NewLocalStore(root) + legacyRoot := SessionWorkDir(root, "session-1") + if err := os.MkdirAll(legacyRoot, 0o755); err != nil { + t.Fatalf("MkdirAll() error = %v", err) + } + + paths, err := store.ResolveSessionPathsFor("sample-campaign", "session-1") + if err != nil { + t.Fatalf("ResolveSessionPathsFor() error = %v", err) + } + if paths.Root != legacyRoot { + t.Fatalf("paths.Root = %q, want legacy root %q", paths.Root, legacyRoot) + } +} + +func TestResolveSessionPathsForAmbiguousRoots(t *testing.T) { + root := t.TempDir() + store := NewLocalStore(root) + legacyRoot := SessionWorkDir(root, "session-1") + canonicalRoot := SessionWorkDirForCampaign(root, "sample-campaign", "session-1") + for _, dir := range []string{legacyRoot, canonicalRoot} { + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("MkdirAll(%q) error = %v", dir, err) + } + } + + _, err := store.ResolveSessionPathsFor("sample-campaign", "session-1") + if err == nil { + t.Fatal("expected ambiguity error, got nil") + } + if !strings.Contains(err.Error(), "ambiguous session workspace roots") { + t.Fatalf("error = %v, want ambiguity message", err) + } +} + func TestChecksumCalculation(t *testing.T) { store := NewLocalStore(t.TempDir()) path := filepath.Join(t.TempDir(), "sample.txt") @@ -53,9 +93,9 @@ func TestChecksumCalculation(t *testing.T) { func TestLockAcquireRelease(t *testing.T) { store := NewLocalStore(t.TempDir()) - lock, err := store.AcquireSessionLock("session-1") + lock, err := store.AcquireSessionLockFor("sample-campaign", "session-1") if err != nil { - t.Fatalf("AcquireSessionLock() error = %v", err) + t.Fatalf("AcquireSessionLockFor() error = %v", err) } exists, err := store.Exists(lock.path) @@ -81,15 +121,15 @@ func TestLockAcquireRelease(t *testing.T) { func TestLockConflict(t *testing.T) { store := NewLocalStore(t.TempDir()) - lock1, err := store.AcquireSessionLock("session-1") + lock1, err := store.AcquireSessionLockFor("sample-campaign", "session-1") if err != nil { - t.Fatalf("first AcquireSessionLock() error = %v", err) + t.Fatalf("first AcquireSessionLockFor() error = %v", err) } defer func() { _ = store.ReleaseSessionLock(lock1) }() - _, err = store.AcquireSessionLock("session-1") + _, err = store.AcquireSessionLockFor("sample-campaign", "session-1") if err == nil { t.Fatal("expected lock conflict error, got nil") } @@ -138,9 +178,9 @@ func TestCopyInput(t *testing.T) { t.Fatalf("WriteFile() error = %v", err) } - ref, err := store.CopyInput("session-1", srcPath, "inputs/speakers.yml") + ref, err := store.CopyInputFor("sample-campaign", "session-1", srcPath, "inputs/speakers.yml") if err != nil { - t.Fatalf("CopyInput() error = %v", err) + t.Fatalf("CopyInputFor() error = %v", err) } if ref.Kind != "input" { diff --git a/internal/artifacts/paths.go b/internal/artifacts/paths.go index e5ba59e..ccd3f5c 100644 --- a/internal/artifacts/paths.go +++ b/internal/artifacts/paths.go @@ -9,6 +9,8 @@ import ( // SessionPaths contains canonical local paths for one session work directory. type SessionPaths struct { WorkspaceRoot string + CampaignID string + SessionID string Root string InputsDir string AudioDir string @@ -16,18 +18,46 @@ type SessionPaths struct { TranscriptsRawDir string TranscriptsTrimmedDir string ArtifactsDir string + ReportsDir string ConfigDir string LogsDir string + CurrentDir string + RunsDir string ManifestPath string LockPath string } -// SessionWorkDir returns the work directory for one session. +// SessionWorkDir returns the legacy work directory for one session. func SessionWorkDir(rootDir, sessionID string) string { return filepath.Join(rootDir, config.PathWorkDirSegment, sessionID) } -// SessionRunWorkDir returns the campaign/session/run scoped local work directory. +// SessionWorkDirForCampaign returns the canonical campaign-aware work directory for one session. +func SessionWorkDirForCampaign(rootDir, campaign, sessionID string) string { + return filepath.Join(rootDir, config.PathWorkDirSegment, campaign, sessionID) +} + +// SessionManifestPathForCampaign returns the canonical session manifest path. +func SessionManifestPathForCampaign(rootDir, campaign, sessionID string) string { + return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathManifestFile) +} + +// SessionRunsDirForCampaign returns the canonical runs directory for one session. +func SessionRunsDirForCampaign(rootDir, campaign, sessionID string) string { + return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathRunsDirSegment) +} + +// SessionRunRootForCampaign returns the canonical run root under runs/{run_id}. +func SessionRunRootForCampaign(rootDir, campaign, sessionID, runID string) string { + return filepath.Join(SessionRunsDirForCampaign(rootDir, campaign, sessionID), runID) +} + +// SessionRunStageDirForCampaign returns the canonical stage directory under runs/{run_id}/{stage}. +func SessionRunStageDirForCampaign(rootDir, campaign, sessionID, runID, stageName string) string { + return filepath.Join(SessionRunRootForCampaign(rootDir, campaign, sessionID, runID), stageName) +} + +// SessionRunWorkDir returns the legacy campaign/session/run scoped local work directory. func SessionRunWorkDir(rootDir, campaign, sessionID, runID string) string { return filepath.Join(rootDir, config.PathWorkDirSegment, campaign, sessionID, runID) } @@ -37,10 +67,21 @@ func SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID string) string { return filepath.Join(spoolRoot, campaign, sessionID, runID, config.PathAudioDirSegment) } -func buildSessionPaths(workspaceRoot, sessionID string) SessionPaths { +func buildLegacySessionPaths(workspaceRoot, sessionID string) SessionPaths { root := SessionWorkDir(workspaceRoot, sessionID) + return buildSessionPathsFromRoot(workspaceRoot, "", sessionID, root) +} + +func buildSessionPaths(workspaceRoot, campaign, sessionID string) SessionPaths { + root := SessionWorkDirForCampaign(workspaceRoot, campaign, sessionID) + return buildSessionPathsFromRoot(workspaceRoot, campaign, sessionID, root) +} + +func buildSessionPathsFromRoot(workspaceRoot, campaign, sessionID, root string) SessionPaths { return SessionPaths{ WorkspaceRoot: workspaceRoot, + CampaignID: campaign, + SessionID: sessionID, Root: root, InputsDir: filepath.Join(root, config.PathInputsDirSegment), AudioDir: filepath.Join(root, config.PathAudioDirSegment), @@ -48,8 +89,11 @@ func buildSessionPaths(workspaceRoot, sessionID string) SessionPaths { TranscriptsRawDir: filepath.Join(root, filepath.FromSlash(config.PathTranscriptsRaw)), TranscriptsTrimmedDir: filepath.Join(root, filepath.FromSlash(config.PathTranscriptsTrimmed)), ArtifactsDir: filepath.Join(root, config.PathArtifactsDirSegment), + ReportsDir: filepath.Join(root, config.PathReportsDirSegment), ConfigDir: filepath.Join(root, config.PathConfigDirSegment), LogsDir: filepath.Join(root, config.PathLogsDirSegment), + CurrentDir: filepath.Join(root, config.PathCurrentDirSegment), + RunsDir: filepath.Join(root, config.PathRunsDirSegment), ManifestPath: filepath.Join(root, config.PathManifestFile), LockPath: filepath.Join(root, config.PathLockFile), } diff --git a/internal/artifacts/paths_model_test.go b/internal/artifacts/paths_model_test.go index 8263a13..4a427b3 100644 --- a/internal/artifacts/paths_model_test.go +++ b/internal/artifacts/paths_model_test.go @@ -14,6 +14,40 @@ func TestSessionRunWorkDir(t *testing.T) { } } +func TestSessionWorkDirForCampaign(t *testing.T) { + root := "/tmp/workspace" + got := SessionWorkDirForCampaign(root, "forsaken", "2026-04-19") + want := filepath.Join(root, "work", "forsaken", "2026-04-19") + if got != want { + t.Fatalf("SessionWorkDirForCampaign() = %q, want %q", got, want) + } +} + +func TestSessionManifestPathForCampaign(t *testing.T) { + root := "/tmp/workspace" + got := SessionManifestPathForCampaign(root, "forsaken", "2026-04-19") + want := filepath.Join(root, "work", "forsaken", "2026-04-19", "manifest.json") + if got != want { + t.Fatalf("SessionManifestPathForCampaign() = %q, want %q", got, want) + } +} + +func TestSessionRunRootAndStageDirForCampaign(t *testing.T) { + root := "/tmp/workspace" + runID := "20260515T031522Z-a1b2c3d4" + runRoot := SessionRunRootForCampaign(root, "forsaken", "2026-04-19", runID) + wantRoot := filepath.Join(root, "work", "forsaken", "2026-04-19", "runs", runID) + if runRoot != wantRoot { + t.Fatalf("SessionRunRootForCampaign() = %q, want %q", runRoot, wantRoot) + } + + stageDir := SessionRunStageDirForCampaign(root, "forsaken", "2026-04-19", runID, "transcribe") + wantStage := filepath.Join(root, "work", "forsaken", "2026-04-19", "runs", runID, "transcribe") + if stageDir != wantStage { + t.Fatalf("SessionRunStageDirForCampaign() = %q, want %q", stageDir, wantStage) + } +} + func TestSessionSpoolAudioDir(t *testing.T) { root := "/var/spool/narratio" got := SessionSpoolAudioDir(root, "forsaken", "2026-04-19", "20260515T031522Z-a1b2c3d4") diff --git a/internal/artifacts/resolve_test.go b/internal/artifacts/resolve_test.go index 71af272..01e0136 100644 --- a/internal/artifacts/resolve_test.go +++ b/internal/artifacts/resolve_test.go @@ -8,7 +8,7 @@ import ( func TestResolveSessionLocalPathForRead(t *testing.T) { workspace := t.TempDir() - paths := buildSessionPaths(workspace, "s-1") + paths := buildSessionPaths(workspace, "sample-campaign", "s-1") if err := os.MkdirAll(paths.TranscriptsRawDir, 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } @@ -41,7 +41,7 @@ func TestResolveSessionLocalPathForReadRelativeWorkspaceRootQualifiedPath(t *tes t.Fatalf("Rel() error = %v", err) } - paths := buildSessionPaths(workspaceRel, "s-1") + paths := buildSessionPaths(workspaceRel, "sample-campaign", "s-1") target := filepath.Join(paths.TranscriptsRawDir, "alice.json") if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) @@ -50,7 +50,7 @@ func TestResolveSessionLocalPathForReadRelativeWorkspaceRootQualifiedPath(t *tes t.Fatalf("WriteFile() error = %v", err) } - manifestPath := filepath.Join(workspaceRel, "work", "s-1", "transcripts", "raw", "alice.json") + manifestPath := filepath.Join(workspaceRel, "work", "sample-campaign", "s-1", "transcripts", "raw", "alice.json") got := ResolveSessionLocalPathForRead(paths, manifestPath) if got != filepath.Clean(manifestPath) { t.Fatalf("resolution = %q, want %q", got, filepath.Clean(manifestPath)) diff --git a/internal/artifacts/s3.go b/internal/artifacts/s3.go index 56331a9..a0b3843 100644 --- a/internal/artifacts/s3.go +++ b/internal/artifacts/s3.go @@ -16,16 +16,36 @@ func (s *S3Store) SessionPaths(_ string) SessionPaths { return SessionPaths{} } +// SessionPathsFor is not implemented for S3-backed storage. +func (s *S3Store) SessionPathsFor(_, _ string) SessionPaths { + return SessionPaths{} +} + +// ResolveSessionPathsFor is not implemented for S3-backed storage. +func (s *S3Store) ResolveSessionPathsFor(_, _ string) (SessionPaths, error) { + return SessionPaths{}, fmt.Errorf("artifacts s3 resolve session paths: not yet implemented") +} + // EnsureLayout returns a not-yet-implemented error in the scaffold. func (s *S3Store) EnsureLayout(_ string) (SessionPaths, error) { return SessionPaths{}, fmt.Errorf("artifacts s3 ensure layout: not yet implemented") } +// EnsureLayoutFor returns a not-yet-implemented error in the scaffold. +func (s *S3Store) EnsureLayoutFor(_, _ string) (SessionPaths, error) { + return SessionPaths{}, fmt.Errorf("artifacts s3 ensure layout for campaign/session: not yet implemented") +} + // CopyInput returns a not-yet-implemented error in the scaffold. func (s *S3Store) CopyInput(_, _, _ string) (Ref, error) { return Ref{}, fmt.Errorf("artifacts s3 copy input: not yet implemented") } +// CopyInputFor returns a not-yet-implemented error in the scaffold. +func (s *S3Store) CopyInputFor(_, _, _, _ string) (Ref, error) { + return Ref{}, fmt.Errorf("artifacts s3 copy input for campaign/session: not yet implemented") +} + // Exists returns a not-yet-implemented error in the scaffold. func (s *S3Store) Exists(_ string) (bool, error) { return false, fmt.Errorf("artifacts s3 exists: not yet implemented") @@ -51,6 +71,11 @@ func (s *S3Store) AcquireSessionLock(_ string) (*LockHandle, error) { return nil, fmt.Errorf("artifacts s3 acquire lock: not yet implemented") } +// AcquireSessionLockFor returns a not-yet-implemented error in the scaffold. +func (s *S3Store) AcquireSessionLockFor(_, _ string) (*LockHandle, error) { + return nil, fmt.Errorf("artifacts s3 acquire lock for campaign/session: not yet implemented") +} + // ReleaseSessionLock returns a not-yet-implemented error in the scaffold. func (s *S3Store) ReleaseSessionLock(_ *LockHandle) error { return fmt.Errorf("artifacts s3 release lock: not yet implemented") diff --git a/internal/artifacts/store.go b/internal/artifacts/store.go index 7a8445d..b894780 100644 --- a/internal/artifacts/store.go +++ b/internal/artifacts/store.go @@ -16,12 +16,17 @@ type Ref struct { // Store is the local artifact/workdir abstraction used by orchestration code. type Store interface { SessionPaths(sessionID string) SessionPaths + SessionPathsFor(campaign, sessionID string) SessionPaths + ResolveSessionPathsFor(campaign, sessionID string) (SessionPaths, error) EnsureLayout(sessionID string) (SessionPaths, error) + EnsureLayoutFor(campaign, sessionID string) (SessionPaths, error) CopyInput(sessionID, srcPath, destRelativePath string) (Ref, error) + CopyInputFor(campaign, sessionID, srcPath, destRelativePath string) (Ref, error) Exists(path string) (bool, error) ExistsRef(ref Ref) (bool, error) WriteFileAtomic(path string, data []byte, perm os.FileMode) error Checksum(path string) (string, error) AcquireSessionLock(sessionID string) (*LockHandle, error) + AcquireSessionLockFor(campaign, sessionID string) (*LockHandle, error) ReleaseSessionLock(lock *LockHandle) error } diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 6af6f3d..42d0dc3 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -49,8 +49,11 @@ const ( PathTranscriptsRaw = "transcripts/raw" PathTranscriptsTrimmed = "transcripts/trimmed" PathArtifactsDirSegment = "artifacts" + PathReportsDirSegment = "reports" PathConfigDirSegment = "config" PathLogsDirSegment = "logs" + PathCurrentDirSegment = "current" + PathRunsDirSegment = "runs" PathManifestFile = "manifest.json" PathLockFile = ".lock" PathTranscriptMerged = "transcripts/merged.json" diff --git a/internal/stage/analyze.go b/internal/stage/analyze.go index 344dd91..4d427a0 100644 --- a/internal/stage/analyze.go +++ b/internal/stage/analyze.go @@ -58,7 +58,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("analyze: session id is required") } - paths := env.ArtifactStore.SessionPaths(sessionID) + paths := sessionPathsForEnv(env, sessionID) if env.Config.Pipeline.Scriptorium == nil { return &StageResult{ Metadata: map[string]any{ diff --git a/internal/stage/analyze_test.go b/internal/stage/analyze_test.go index 6bc16b6..5093db8 100644 --- a/internal/stage/analyze_test.go +++ b/internal/stage/analyze_test.go @@ -17,7 +17,7 @@ import ( func TestAnalyzeGeneratesSessionRecapFromTrimmedTranscript(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) result, err := (analyzeStage{}).Run(context.Background(), env, m) @@ -70,7 +70,7 @@ func TestAnalyzeGeneratesSessionRecapFromTrimmedTranscript(t *testing.T) { func TestAnalyzeRenderDebugFalseDoesNotCallRenderArtifact(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.RenderDebug = false @@ -86,7 +86,7 @@ func TestAnalyzeRenderDebugFalseDoesNotCallRenderArtifact(t *testing.T) { func TestAnalyzeRenderDebugArtifactOverrideFalseWinsOverGlobalTrue(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.RenderDebug = true @@ -109,7 +109,7 @@ func TestAnalyzeRenderDebugArtifactOverrideFalseWinsOverGlobalTrue(t *testing.T) func TestAnalyzeRenderDebugTrueCallsRenderBeforeRun(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.RenderDebug = true @@ -130,7 +130,7 @@ func TestAnalyzeRenderDebugTrueCallsRenderBeforeRun(t *testing.T) { func TestAnalyzeRenderOutputPathIsRecorded(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.RenderDebug = true @@ -154,7 +154,7 @@ func TestAnalyzeRenderOutputPathIsRecorded(t *testing.T) { func TestAnalyzeRenderFailurePreventsRun(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.RenderDebug = true fake.RenderErr = errors.New("render boom") @@ -176,7 +176,7 @@ func TestAnalyzeRenderFailurePreventsRun(t *testing.T) { func TestAnalyzeRenderInvalidJSONFailsClearly(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.RenderDebug = true runner := &orderedScriptoriumRunner{ @@ -199,7 +199,7 @@ func TestAnalyzeRenderInvalidJSONFailsClearly(t *testing.T) { func TestAnalyzeRunStillSucceedsWhenRenderSucceeds(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.RenderDebug = true @@ -220,7 +220,7 @@ func TestAnalyzeRunStillSucceedsWhenRenderSucceeds(t *testing.T) { func TestAnalyzeOmitsOptionalPreviousRecapWhenUnavailable(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = config.ScriptoriumArtifactConfig{ @@ -318,7 +318,7 @@ func (r *orderedScriptoriumRunner) RunArtifact(_ context.Context, req scriptoriu func TestAnalyzeIncludesPreviousRecapWhenConfiguredAndAvailable(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[]}`) previousRecapPath := filepath.Join(filepath.Dir(env.Config.SessionPath), "previous", "session_recap.md") @@ -359,7 +359,7 @@ func TestAnalyzeIncludesPreviousRecapWhenConfiguredAndAvailable(t *testing.T) { func TestAnalyzeFailsWhenRequiredPreviousRecapMissing(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = config.ScriptoriumArtifactConfig{ @@ -390,7 +390,7 @@ func TestAnalyzeFailsWhenRequiredPreviousRecapMissing(t *testing.T) { func TestAnalyzeFailsWhenOutputPathMissing(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) artifact := env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] @@ -425,7 +425,7 @@ func TestAnalyzeFailsWhenTrimmedTranscriptMissing(t *testing.T) { func TestAnalyzeSupportsProcessedTranscriptSourceWhenConfigured(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[]}`) artifact := env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] @@ -449,7 +449,7 @@ func TestAnalyzeSupportsProcessedTranscriptSourceWhenConfigured(t *testing.T) { func TestAnalyzeSupportsNormalizedTranscriptSourceWhenConfigured(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) normalizedPath := filepath.Join(paths.TranscriptsDir, "normalized.json") writeAnalyzeFile(t, normalizedPath, `{"segments":[{"id":1}]}`) @@ -474,7 +474,7 @@ func TestAnalyzeSupportsNormalizedTranscriptSourceWhenConfigured(t *testing.T) { func TestAnalyzeSupportsNormalizedTranscriptSourceFromManifestOutput(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) fallbackPath := filepath.Join(paths.TranscriptsDir, "normalized.json") manifestPath := filepath.Join(paths.ArtifactsDir, "normalized.from-manifest.json") writeAnalyzeFile(t, fallbackPath, `{"segments":[{"id":999}]}`) @@ -528,7 +528,7 @@ func TestAnalyzeFailsWhenNormalizedTranscriptMissing(t *testing.T) { func TestAnalyzeFailsWhenProcessedTranscriptInvalidJSON(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{not-json`) _, err := (analyzeStage{}).Run(context.Background(), env, m) @@ -542,7 +542,7 @@ func TestAnalyzeFailsWhenProcessedTranscriptInvalidJSON(t *testing.T) { func TestAnalyzeFailsWhenProcessedTranscriptMissingSegmentsArray(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"not_segments":[]}`) _, err := (analyzeStage{}).Run(context.Background(), env, m) @@ -556,7 +556,7 @@ func TestAnalyzeFailsWhenProcessedTranscriptMissingSegmentsArray(t *testing.T) { func TestAnalyzeRecordsRefsAndMetadata(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) result, err := (analyzeStage{}).Run(context.Background(), env, m) @@ -585,7 +585,7 @@ func TestAnalyzeRecordsRefsAndMetadata(t *testing.T) { func TestAnalyzeHandlesAdapterError(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) fake.RunErr = errors.New("adapter boom") @@ -600,7 +600,7 @@ func TestAnalyzeHandlesAdapterError(t *testing.T) { func TestAnalyzeHandlesValidationFailedResultAsError(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) fake.RunResult = scriptorium.ArtifactResult{ ValidationFailed: true, @@ -619,7 +619,7 @@ func TestAnalyzeHandlesValidationFailedResultAsError(t *testing.T) { func TestAnalyzeSkipsWhenNoEnabledScriptoriumArtifactsConfigured(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`) env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = config.ScriptoriumArtifactConfig{ @@ -692,7 +692,7 @@ func setupAnalyzeEnv(t *testing.T) (*Env, *manifest.Manifest, *scriptorium.FakeR } store := artifacts.NewLocalStore(workspace) - if _, err := store.EnsureLayout("2026-05-03"); err != nil { + if _, err := store.EnsureLayoutFor("sample-campaign", "2026-05-03"); err != nil { t.Fatalf("EnsureLayout() error = %v", err) } diff --git a/internal/stage/archive.go b/internal/stage/archive.go index 97af070..a1793ea 100644 --- a/internal/stage/archive.go +++ b/internal/stage/archive.go @@ -277,8 +277,11 @@ func archiveWorkDir(env *Env, m *manifest.Manifest) (string, error) { if info, err := os.Stat(runScoped); err == nil && info.IsDir() { return runScoped, nil } - legacy := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID) - return legacy, nil + paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(campaign, sessionID) + if err != nil { + return "", err + } + return paths.Root, nil } func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) { @@ -385,9 +388,11 @@ func archiveSourceWorkDirs(env *Env, m *manifest.Manifest, runWorkDir string) [] sessionID = strings.TrimSpace(m.SessionID) } if sessionID != "" { - sessionWorkDir := filepath.Clean(artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID)) - if sessionWorkDir != "" { - candidates = append(candidates, sessionWorkDir) + 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)) @@ -488,8 +493,11 @@ func resolveArchiveManifestSource(env *Env, m *manifest.Manifest, workDir string sessionID = strings.TrimSpace(m.SessionID) } if sessionID != "" { - sessionManifest := filepath.Join(artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID), "manifest.json") - candidates = append(candidates, sessionManifest) + 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) } seen := make(map[string]struct{}, len(candidates)) diff --git a/internal/stage/archive_test.go b/internal/stage/archive_test.go index 3a695f1..1b2f6a1 100644 --- a/internal/stage/archive_test.go +++ b/internal/stage/archive_test.go @@ -183,7 +183,7 @@ func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) { } } -func TestArchivePromotionFallsBackToSessionWorkDir(t *testing.T) { +func TestArchivePromotionFailsOnAmbiguousSessionRoots(t *testing.T) { env, m, runWorkDir := archiveFixture(t) sessionWorkDir := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, m.SessionID) sessionTrimmed := filepath.Join(sessionWorkDir, "transcripts", "trimmed.json") @@ -193,16 +193,12 @@ func TestArchivePromotionFallsBackToSessionWorkDir(t *testing.T) { } writeStageTestFile(t, sessionTrimmed, "{}\n") - result, err := archiveStage{}.Run(context.Background(), env, m) - if err != nil { - t.Fatalf("Run() error = %v", err) + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected ambiguity error, got nil") } - if result.Metadata["promoted_files_uploaded"] != 2 { - t.Fatalf("metadata promoted_files_uploaded = %#v, want 2", result.Metadata["promoted_files_uploaded"]) - } - fake := env.ObjectStore.(*storage.FakeBackend) - if _, ok := fake.Objects[m.S3SessionPrefix+"transcripts/trimmed.json"]; !ok { - t.Fatalf("missing promoted trimmed key from session fallback") + if !strings.Contains(err.Error(), "ambiguous session workspace roots") { + t.Fatalf("error = %v, want ambiguity error", err) } } diff --git a/internal/stage/merge.go b/internal/stage/merge.go index 4cfcfff..7e0a046 100644 --- a/internal/stage/merge.go +++ b/internal/stage/merge.go @@ -57,7 +57,7 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta return nil, fmt.Errorf("merge: session id is required") } - paths := env.ArtifactStore.SessionPaths(sessionID) + paths := sessionPathsForEnv(env, sessionID) inputs, err := discoverRawTranscripts(m, paths) if err != nil { diff --git a/internal/stage/merge_test.go b/internal/stage/merge_test.go index 0877121..2b51324 100644 --- a/internal/stage/merge_test.go +++ b/internal/stage/merge_test.go @@ -33,7 +33,7 @@ func (r *normalizeDirAssertingRunner) Normalize(ctx context.Context, req seriati func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) inA := filepath.Join(paths.TranscriptsRawDir, "alice.json") inB := filepath.Join(paths.TranscriptsRawDir, "bob.json") @@ -122,7 +122,7 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) { func TestMergeStageFailsWhenNoRawTranscripts(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") env.Seriatim = &seriatim.FakeRunner{} @@ -138,7 +138,7 @@ func TestMergeStageFailsWhenNoRawTranscripts(t *testing.T) { func TestMergeStageFailsOnInvalidInputJSON(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), "not-json") writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") @@ -155,7 +155,7 @@ func TestMergeStageFailsOnInvalidInputJSON(t *testing.T) { func TestMergeStageFailsWhenAdapterFails(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`) writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") @@ -172,7 +172,7 @@ func TestMergeStageFailsWhenAdapterFails(t *testing.T) { func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`) writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") @@ -193,7 +193,7 @@ func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testin func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t *testing.T) { env, m := setupMergeEnvWithRelativeWorkspaceRoot(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) rawPath := filepath.Join(paths.TranscriptsRawDir, "alice.json") writeFile(t, rawPath, `{"segments":[]}`) @@ -224,7 +224,7 @@ func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t func TestMergeStageCreatesNormalizedRawDirectoryBeforeNormalize(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) rawPath := filepath.Join(paths.TranscriptsRawDir, "alice.json") writeFile(t, rawPath, `{"segments":[]}`) @@ -248,7 +248,7 @@ func TestMergeStageCreatesNormalizedRawDirectoryBeforeNormalize(t *testing.T) { func TestMergeStageFailsWhenNormalizeAdapterFails(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + 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") @@ -266,7 +266,7 @@ func TestMergeStageFailsWhenNormalizeAdapterFails(t *testing.T) { func TestMergeStageFailsWhenNormalizedOutputInvalid(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + 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") @@ -291,7 +291,7 @@ func TestMergeStageFailsWhenNormalizedOutputInvalid(t *testing.T) { func TestMergeStageResolvesSessionRelativeManifestOutputs(t *testing.T) { env, m := setupMergeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) rawPath := filepath.Join(paths.TranscriptsRawDir, "alice.json") writeFile(t, rawPath, `{"segments":[]}`) @@ -373,7 +373,7 @@ func setupMergeEnvWithWorkspace(t *testing.T, workspace string) (*Env, *manifest } store := artifacts.NewLocalStore(workspace) - if _, err := store.EnsureLayout("2026-05-03"); err != nil { + if _, err := store.EnsureLayoutFor("sample-campaign", "2026-05-03"); err != nil { t.Fatalf("EnsureLayout() error = %v", err) } return &Env{ diff --git a/internal/stage/normalize.go b/internal/stage/normalize.go index 2ca2ccb..c28eef7 100644 --- a/internal/stage/normalize.go +++ b/internal/stage/normalize.go @@ -54,7 +54,7 @@ func (normalizeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) ( return nil, fmt.Errorf("normalize: session id is required") } - paths := env.ArtifactStore.SessionPaths(sessionID) + paths := sessionPathsForEnv(env, sessionID) processedPath, processedSource, err := discoverProcessedTranscript(m, paths) if err != nil { return nil, fmt.Errorf("normalize: resolve processed transcript: %w", err) diff --git a/internal/stage/normalize_test.go b/internal/stage/normalize_test.go index a3ca1d3..e08a1a3 100644 --- a/internal/stage/normalize_test.go +++ b/internal/stage/normalize_test.go @@ -16,7 +16,7 @@ import ( func TestNormalizeStageConsumesProcessedTranscriptFromManifest(t *testing.T) { env, m, ser := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) manifestProcessed := filepath.Join(paths.ArtifactsDir, "processed.from-manifest.json") writeFile(t, manifestProcessed, `{"segments":[{"id":10}]}`) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":99}]}`) @@ -43,7 +43,7 @@ func TestNormalizeStageConsumesProcessedTranscriptFromManifest(t *testing.T) { func TestNormalizeStageFallsBackToProcessedTranscriptPath(t *testing.T) { env, m, ser := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) fallback := filepath.Join(paths.TranscriptsDir, "processed.json") writeFile(t, fallback, `{"segments":[{"id":1}]}`) @@ -72,7 +72,7 @@ func TestNormalizeStageFailsWhenProcessedTranscriptMissing(t *testing.T) { func TestNormalizeStageFailsWhenProcessedTranscriptInvalidJSON(t *testing.T) { env, m, _ := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), "not-json") _, err := (normalizeStage{}).Run(context.Background(), env, m) @@ -86,7 +86,7 @@ func TestNormalizeStageFailsWhenProcessedTranscriptInvalidJSON(t *testing.T) { func TestNormalizeStageFailsWhenProcessedTranscriptMissingSegments(t *testing.T) { env, m, _ := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"schema":"audita.processed.v1"}`) _, err := (normalizeStage{}).Run(context.Background(), env, m) @@ -100,7 +100,7 @@ func TestNormalizeStageFailsWhenProcessedTranscriptMissingSegments(t *testing.T) func TestNormalizeStagePassesConfiguredOutputSchemaToAdapter(t *testing.T) { env, m, ser := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`) env.Config.Pipeline.Normalize.OutputSchema = "seriatim-full" @@ -121,7 +121,7 @@ func TestNormalizeStagePassesConfiguredOutputSchemaToAdapter(t *testing.T) { func TestNormalizeStageRecordsNormalizedTranscriptOutputKind(t *testing.T) { env, m, _ := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`) result, err := (normalizeStage{}).Run(context.Background(), env, m) @@ -138,7 +138,7 @@ func TestNormalizeStageRecordsNormalizedTranscriptOutputKind(t *testing.T) { func TestNormalizeStageRecordsReportLogAndGeneratedConfigRefs(t *testing.T) { env, m, _ := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`) report := true env.Config.Pipeline.Normalize.Report = &report @@ -164,7 +164,7 @@ func TestNormalizeStageRecordsReportLogAndGeneratedConfigRefs(t *testing.T) { func TestNormalizeStageFailsWhenAdapterReturnsError(t *testing.T) { env, m, ser := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`) ser.NormalizeErr = errors.New("normalize failed") @@ -179,7 +179,7 @@ func TestNormalizeStageFailsWhenAdapterReturnsError(t *testing.T) { func TestNormalizeStageFailsWhenNormalizedOutputInvalid(t *testing.T) { env, m, ser := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`) badOutput := filepath.Join(paths.TranscriptsDir, "normalized.bad.json") writeFile(t, badOutput, "not-json") @@ -196,7 +196,7 @@ func TestNormalizeStageFailsWhenNormalizedOutputInvalid(t *testing.T) { func TestNormalizeStageReportEnabledFailsWhenReportMissing(t *testing.T) { env, m, ser := setupNormalizeEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`) report := true env.Config.Pipeline.Normalize.Report = &report @@ -243,7 +243,7 @@ func setupNormalizeEnv(t *testing.T) (*Env, *manifest.Manifest, *seriatim.FakeRu } store := artifacts.NewLocalStore(workspace) - if _, err := store.EnsureLayout("2026-05-03"); err != nil { + if _, err := store.EnsureLayoutFor("sample-campaign", "2026-05-03"); err != nil { t.Fatalf("EnsureLayout() error = %v", err) } diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 9521205..91a9e59 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -224,7 +224,7 @@ func TestPlaceholderAdapterErrorPropagation(t *testing.T) { root := t.TempDir() store := artifacts.NewLocalStore(root) - _, err := store.EnsureLayout("2026-05-03") + _, err := store.EnsureLayoutFor("sample-campaign", "2026-05-03") if err != nil { t.Fatalf("EnsureLayout() error = %v", err) } diff --git a/internal/stage/polish.go b/internal/stage/polish.go index 42e52a8..7e1b4f3 100644 --- a/internal/stage/polish.go +++ b/internal/stage/polish.go @@ -55,7 +55,7 @@ func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*St return nil, fmt.Errorf("polish: session id is required") } - paths := env.ArtifactStore.SessionPaths(sessionID) + paths := sessionPathsForEnv(env, sessionID) mergedPath, source, err := discoverMergedTranscript(m, paths) if err != nil { return nil, fmt.Errorf("polish: resolve merged transcript: %w", err) diff --git a/internal/stage/polish_test.go b/internal/stage/polish_test.go index 8f52d18..83fe395 100644 --- a/internal/stage/polish_test.go +++ b/internal/stage/polish_test.go @@ -16,7 +16,7 @@ import ( func TestPolishStagePolishesMergedTranscriptAndRecordsMetadata(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) mergedPath := filepath.Join(paths.TranscriptsDir, "merged.json") writeFile(t, mergedPath, `{"segments":[]}`) @@ -120,7 +120,7 @@ func TestPolishStagePolishesMergedTranscriptAndRecordsMetadata(t *testing.T) { func TestPolishStageFallsBackToMergedTranscriptPath(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) mergedPath := filepath.Join(paths.TranscriptsDir, "merged.json") writeFile(t, mergedPath, `{"segments":[]}`) writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") @@ -142,7 +142,7 @@ func TestPolishStageFallsBackToMergedTranscriptPath(t *testing.T) { func TestPolishStageFailsWhenMergedTranscriptMissing(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") env.Audita = &audita.FakeRunner{} @@ -157,7 +157,7 @@ func TestPolishStageFailsWhenMergedTranscriptMissing(t *testing.T) { func TestPolishStageFailsWhenMergedTranscriptInvalidJSON(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), "not-json") writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") env.Audita = &audita.FakeRunner{} @@ -173,7 +173,7 @@ func TestPolishStageFailsWhenMergedTranscriptInvalidJSON(t *testing.T) { func TestPolishStageFailsWhenGlossaryMissing(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) env.Audita = &audita.FakeRunner{} @@ -188,7 +188,7 @@ func TestPolishStageFailsWhenGlossaryMissing(t *testing.T) { func TestPolishStageFailsWhenAdapterFails(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") env.Audita = &audita.FakeRunner{Err: errors.New("audita failed")} @@ -204,7 +204,7 @@ func TestPolishStageFailsWhenAdapterFails(t *testing.T) { func TestPolishStageFailsWhenProcessedOutputInvalid(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") badPath := filepath.Join(paths.TranscriptsDir, "processed.invalid.json") @@ -222,7 +222,7 @@ func TestPolishStageFailsWhenProcessedOutputInvalid(t *testing.T) { func TestPolishStageFailsWhenReportInvalid(t *testing.T) { env, m := setupPolishEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`) writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n") badReport := filepath.Join(paths.ArtifactsDir, "bad.report.json") @@ -284,7 +284,7 @@ func setupPolishEnv(t *testing.T) (*Env, *manifest.Manifest) { } store := artifacts.NewLocalStore(workspace) - if _, err := store.EnsureLayout("2026-05-03"); err != nil { + if _, err := store.EnsureLayoutFor("sample-campaign", "2026-05-03"); err != nil { t.Fatalf("EnsureLayout() error = %v", err) } return &Env{ diff --git a/internal/stage/prepare.go b/internal/stage/prepare.go index c2d5668..eca9d9d 100644 --- a/internal/stage/prepare.go +++ b/internal/stage/prepare.go @@ -54,7 +54,7 @@ func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("prepare: session id is required") } - paths, err := env.ArtifactStore.EnsureLayout(sessionID) + paths, err := ensureLayoutForEnv(env, sessionID) if err != nil { return nil, fmt.Errorf("prepare: ensure workdir layout: %w", err) } @@ -368,7 +368,11 @@ func pathsWorkDirForManifest(env *Env, m *manifest.Manifest, sessionID string) s if runID != "" { return artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, env.Config.Session.Campaign, sessionID, runID) } - return artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID) + campaign := strings.TrimSpace(env.Config.Session.Campaign) + if campaign == "" { + return artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID) + } + return artifacts.SessionWorkDirForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID) } func resolvePath(baseDir, p string) (string, error) { diff --git a/internal/stage/prepare_test.go b/internal/stage/prepare_test.go index 8a495cf..16ad7cf 100644 --- a/internal/stage/prepare_test.go +++ b/internal/stage/prepare_test.go @@ -33,7 +33,7 @@ func TestPrepareStageExplicitAudioFiles(t *testing.T) { t.Fatalf("result metadata = %#v, want prepared=true", result) } - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) for _, p := range []string{ filepath.Join(paths.InputsDir, "session.yml"), filepath.Join(paths.InputsDir, "pipeline.resolved.yml"), @@ -76,7 +76,7 @@ func TestPrepareStageAudioDirEnumeration(t *testing.T) { t.Fatalf("audio_files_resolved = %#v, want 1", got) } - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) if _, err := os.Stat(filepath.Join(paths.AudioDir, "a.flac")); err != nil { t.Fatalf("expected copied flac: %v", err) } diff --git a/internal/stage/session_paths.go b/internal/stage/session_paths.go new file mode 100644 index 0000000..f4004a9 --- /dev/null +++ b/internal/stage/session_paths.go @@ -0,0 +1,29 @@ +package stage + +import ( + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" +) + +func sessionPathsForEnv(env *Env, sessionID string) artifacts.SessionPaths { + campaign := "" + if env != nil && env.Config != nil && env.Config.Session != nil { + campaign = strings.TrimSpace(env.Config.Session.Campaign) + } + if campaign == "" { + return env.ArtifactStore.SessionPaths(sessionID) + } + return env.ArtifactStore.SessionPathsFor(campaign, sessionID) +} + +func ensureLayoutForEnv(env *Env, sessionID string) (artifacts.SessionPaths, error) { + campaign := "" + if env != nil && env.Config != nil && env.Config.Session != nil { + campaign = strings.TrimSpace(env.Config.Session.Campaign) + } + if campaign == "" { + return env.ArtifactStore.EnsureLayout(sessionID) + } + return env.ArtifactStore.EnsureLayoutFor(campaign, sessionID) +} diff --git a/internal/stage/transcribe.go b/internal/stage/transcribe.go index bb1589f..636f463 100644 --- a/internal/stage/transcribe.go +++ b/internal/stage/transcribe.go @@ -55,7 +55,7 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) return nil, fmt.Errorf("transcribe: session id is required") } - paths := env.ArtifactStore.SessionPaths(sessionID) + paths := sessionPathsForEnv(env, sessionID) audioFiles, err := discoverPreparedAudio(m, paths.AudioDir) if err != nil { return nil, fmt.Errorf("transcribe: resolve audio inputs: %w", err) diff --git a/internal/stage/transcribe_test.go b/internal/stage/transcribe_test.go index 0796dd1..857997b 100644 --- a/internal/stage/transcribe_test.go +++ b/internal/stage/transcribe_test.go @@ -53,8 +53,8 @@ func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) { } sort.Strings(gotPaths) wantPaths := []string{ - filepath.Join(env.ArtifactStore.SessionPaths(m.SessionID).TranscriptsRawDir, "alice.json"), - filepath.Join(env.ArtifactStore.SessionPaths(m.SessionID).TranscriptsRawDir, "bob.json"), + filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json"), + filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "bob.json"), } sort.Strings(wantPaths) if strings.Join(gotPaths, "|") != strings.Join(wantPaths, "|") { diff --git a/internal/stage/trim.go b/internal/stage/trim.go index 28fbd0a..1f82706 100644 --- a/internal/stage/trim.go +++ b/internal/stage/trim.go @@ -54,7 +54,7 @@ func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Stag return nil, fmt.Errorf("trim: session id is required") } - paths := env.ArtifactStore.SessionPaths(sessionID) + paths := sessionPathsForEnv(env, sessionID) normalizedPath, normalizedSource, err := discoverNormalizedTranscript(m, paths) if err != nil { return nil, fmt.Errorf("trim: resolve normalized transcript: %w", err) diff --git a/internal/stage/trim_test.go b/internal/stage/trim_test.go index 903a83c..62e5958 100644 --- a/internal/stage/trim_test.go +++ b/internal/stage/trim_test.go @@ -18,7 +18,7 @@ import ( func TestTrimStageConsumesNormalizedAndProducesTrimmedTranscript(t *testing.T) { env, m, scr, ser := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) normalized := filepath.Join(paths.TranscriptsDir, "normalized.json") writeFile(t, normalized, `{"segments":[{"id":10},{"id":868}]}`) m.MarkStageSucceeded("normalize", time.Now().UTC(), []manifest.ArtifactRecord{ @@ -70,7 +70,7 @@ func TestTrimStageConsumesNormalizedAndProducesTrimmedTranscript(t *testing.T) { func TestTrimStageUsesConfiguredScriptoriumInputName(t *testing.T) { env, m, scr, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + 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}` @@ -94,7 +94,7 @@ func TestTrimStageUsesConfiguredScriptoriumInputName(t *testing.T) { func TestTrimStageRecordsLogAndGeneratedConfigRefs(t *testing.T) { env, m, scr, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"segments":[{"id":1},{"id":2}]}`) scr.BoundsBody = `{"trim_action":"trim","start_segment_id":1,"end_segment_id":2}` @@ -114,7 +114,7 @@ func TestTrimStageRecordsLogAndGeneratedConfigRefs(t *testing.T) { func TestTrimStageRenderDebugDiagnosticsAreNotStageOutputs(t *testing.T) { env, m, scr, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"segments":[{"id":1},{"id":2}]}`) scr.BoundsBody = `{"trim_action":"trim","start_segment_id":1,"end_segment_id":2}` @@ -152,7 +152,7 @@ func TestTrimStageFailsWhenNormalizedTranscriptMissing(t *testing.T) { func TestTrimStageDoesNotFallBackToProcessedTranscript(t *testing.T) { env, m, scr, ser := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1},{"id":2}]}`) _, err := (trimStage{}).Run(context.Background(), env, m) @@ -172,7 +172,7 @@ func TestTrimStageDoesNotFallBackToProcessedTranscript(t *testing.T) { func TestTrimStageFailsWhenNormalizedTranscriptInvalidJSON(t *testing.T) { env, m, _, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), "not-json") _, err := (trimStage{}).Run(context.Background(), env, m) if err == nil { @@ -185,7 +185,7 @@ func TestTrimStageFailsWhenNormalizedTranscriptInvalidJSON(t *testing.T) { func TestTrimStageFailsWhenNormalizedTranscriptMissingSegmentsArray(t *testing.T) { env, m, _, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"schema":"audita.processed.v1"}`) _, err := (trimStage{}).Run(context.Background(), env, m) if err == nil { @@ -198,7 +198,7 @@ func TestTrimStageFailsWhenNormalizedTranscriptMissingSegmentsArray(t *testing.T func TestTrimStageFailsWhenBoundsOutputInvalidJSON(t *testing.T) { env, m, scr, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"segments":[{"id":1},{"id":2}]}`) scr.BoundsBody = "not-json" _, err := (trimStage{}).Run(context.Background(), env, m) @@ -212,7 +212,7 @@ func TestTrimStageFailsWhenBoundsOutputInvalidJSON(t *testing.T) { func TestTrimStageFailsWhenBoundsRangeIsDescending(t *testing.T) { env, m, scr, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"segments":[{"id":1},{"id":2}]}`) scr.BoundsBody = `{"trim_action":"trim","start_segment_id":2,"end_segment_id":1}` _, err := (trimStage{}).Run(context.Background(), env, m) @@ -226,7 +226,7 @@ func TestTrimStageFailsWhenBoundsRangeIsDescending(t *testing.T) { func TestTrimStageFailsWhenBoundsIDsMissingFromTranscript(t *testing.T) { env, m, scr, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"segments":[{"id":20},{"id":21}]}`) scr.BoundsBody = `{"trim_action":"trim","start_segment_id":10,"end_segment_id":21}` _, err := (trimStage{}).Run(context.Background(), env, m) @@ -240,7 +240,7 @@ func TestTrimStageFailsWhenBoundsIDsMissingFromTranscript(t *testing.T) { func TestTrimStageFailsWhenScriptoriumAdapterFails(t *testing.T) { env, m, scr, _ := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"segments":[{"id":1},{"id":2}]}`) scr.RunErr = errors.New("bounds failed") _, err := (trimStage{}).Run(context.Background(), env, m) @@ -254,7 +254,7 @@ func TestTrimStageFailsWhenScriptoriumAdapterFails(t *testing.T) { func TestTrimStageFailsWhenSeriatimTrimAdapterFails(t *testing.T) { env, m, scr, ser := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) writeFile(t, filepath.Join(paths.TranscriptsDir, "normalized.json"), `{"segments":[{"id":1},{"id":2}]}`) scr.BoundsBody = `{"trim_action":"trim","start_segment_id":1,"end_segment_id":2}` ser.TrimErr = errors.New("trim failed") @@ -269,7 +269,7 @@ func TestTrimStageFailsWhenSeriatimTrimAdapterFails(t *testing.T) { func TestTrimStageDisabledCopiesNormalizedTranscript(t *testing.T) { env, m, scr, ser := setupTrimEnv(t) - paths := env.ArtifactStore.SessionPaths(m.SessionID) + paths := sessionPathsForEnv(env, m.SessionID) normalized := filepath.Join(paths.TranscriptsDir, "normalized.json") normalizedBody := `{"segments":[{"id":1,"text":"alpha"},{"id":2,"text":"beta"}]}` writeFile(t, normalized, normalizedBody) @@ -430,7 +430,7 @@ func setupTrimEnv(t *testing.T) (*Env, *manifest.Manifest, *boundsScriptoriumRun } store := artifacts.NewLocalStore(workspace) - if _, err := store.EnsureLayout("2026-05-03"); err != nil { + if _, err := store.EnsureLayoutFor("sample-campaign", "2026-05-03"); err != nil { t.Fatalf("EnsureLayout() error = %v", err) }