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

Add campaign-aware workspace path foundation

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-17 20:57:27 +00:00
parent e58e545686
commit 550288e008
34 changed files with 448 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
internal/app/commands_test.go
View File

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

2
internal/app/plan.go
View File

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

4
internal/app/plan_test.go
View File

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

8
internal/app/resume.go
View File

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

24
internal/app/resume_run_stage_test.go
View File

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

11
internal/app/runner.go
View File

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

17
internal/app/runner_test.go
View File

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

2
internal/app/whisperx_wiring_test.go
View File

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

108
internal/artifacts/local.go
View File

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

58
internal/artifacts/local_test.go
View File

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

50
internal/artifacts/paths.go
View File

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

34
internal/artifacts/paths_model_test.go
View File

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

6
internal/artifacts/resolve_test.go
View File

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

25
internal/artifacts/s3.go
View File

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

5
internal/artifacts/store.go
View File

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

3
internal/config/defaults.go
View File

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

2
internal/stage/analyze.go
View File

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

44
internal/stage/analyze_test.go
View File

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

22
internal/stage/archive.go
View File

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

16
internal/stage/archive_test.go
View File

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

2
internal/stage/merge.go
View File

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

22
internal/stage/merge_test.go
View File

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

2
internal/stage/normalize.go
View File

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

22
internal/stage/normalize_test.go
View File

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

2
internal/stage/placeholders_test.go
View File

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

2
internal/stage/polish.go
View File

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

18
internal/stage/polish_test.go
View File

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

8
internal/stage/prepare.go
View File

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

4
internal/stage/prepare_test.go
View File

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

29
internal/stage/session_paths.go Normal file
View File

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

2
internal/stage/transcribe.go
View File

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

4
internal/stage/transcribe_test.go
View File

@@ -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, "|") {

2
internal/stage/trim.go
View File

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

28
internal/stage/trim_test.go
View File

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