2 Commits

38 changed files with 1113 additions and 165 deletions

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

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

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

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)

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

View File

@@ -5,7 +5,6 @@ import (
"fmt"
"log/slog"
"os"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/audita"
@@ -28,7 +27,9 @@ type RunOptions struct {
type RunSummary struct {
SessionID string
RunID string
ManifestPath string
RunManifestPath string
StageNames []string
Executed []string
Skipped []string
@@ -93,12 +94,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)
}
@@ -111,7 +112,11 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if err != nil {
return nil, err
}
identityChanged, err := ensureManifestIdentity(cfg, m)
runID, err := artifacts.NewRunID()
if err != nil {
return nil, fmt.Errorf("generate run id: %w", err)
}
identityChanged, err := ensureManifestIdentity(cfg, m, runID)
if err != nil {
return nil, fmt.Errorf("initialize manifest identity: %w", err)
}
@@ -120,6 +125,29 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
return nil, fmt.Errorf("save manifest identity %q: %w", manifestPath, err)
}
}
runManifestPath := artifacts.SessionRunManifestPathForCampaign(
cfg.Pipeline.Workspace.Root,
cfg.Session.Campaign,
cfg.Session.SessionID,
runID,
)
runManifestStore := &manifest.LocalStore{}
runManifest, err := runManifestStore.CreateRun(
ctx,
cfg.Session.SessionID,
cfg.Session.Campaign,
runID,
opts.Force,
requestedStageNames(stages),
)
if err != nil {
return nil, fmt.Errorf("create run manifest: %w", err)
}
runManifest.SessionManifestPath = manifestPath
syncRunManifestIdentityFromSession(m, runManifest)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err)
}
stageEnv := env
@@ -134,12 +162,23 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if d.Action == stageActionSkip {
skipped = append(skipped, s.Name())
skipAt := nowUTC()
runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt)
runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded")
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, fmt.Errorf("save run manifest after skip %q: %w", s.Name(), err)
}
env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force)
continue
}
executed = append(executed, s.Name())
now := nowUTC()
runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now)
runManifest.MarkStageRunning(s.Name(), now)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err)
}
m.MarkStageRunning(s.Name(), now)
env.Logger.Info("starting stage", "stage", s.Name())
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
@@ -149,32 +188,59 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
result, err := s.Run(ctx, stageEnv, m)
if err != nil {
m.MarkStageFailed(s.Name(), nowUTC(), err.Error())
failedAt := nowUTC()
m.MarkStageFailed(s.Name(), failedAt, err.Error())
if saveErr := env.ManifestStore.Save(ctx, manifestPath, m); saveErr != nil {
return nil, fmt.Errorf("stage %q failed (%v) and manifest save failed (%v)", s.Name(), err, saveErr)
}
runManifest.MarkStageFailed(s.Name(), failedAt, err.Error())
syncRunManifestIdentityFromSession(m, runManifest)
if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil {
return nil, fmt.Errorf("stage %q failed (%v) and run-manifest save failed (%v)", s.Name(), err, saveErr)
}
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err)
}
outputs := mapResultOutputs(result)
m.MarkStageSucceeded(s.Name(), nowUTC(), outputs)
succeededAt := nowUTC()
m.MarkStageSucceeded(s.Name(), succeededAt, outputs)
applyStageResultToManifest(m, s.Name(), result)
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)
}
runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs)
applyStageResultToRunManifest(runManifest, s.Name(), result)
syncRunManifestIdentityFromSession(m, runManifest)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, fmt.Errorf("save run manifest after stage %q: %w", s.Name(), err)
}
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath)
env.Logger.Info("stage succeeded", "stage", s.Name())
}
if err := runPostArchiveCleanup(ctx, env, manifestPath, m, executed); err != nil {
failedAt := nowUTC()
runManifest.MarkFailed(failedAt, err.Error())
syncRunManifestIdentityFromSession(m, runManifest)
if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil {
return nil, fmt.Errorf("post-archive cleanup failed (%v) and run-manifest save failed (%v)", err, saveErr)
}
return nil, fmt.Errorf("post-archive cleanup: %w", err)
}
completedAt := nowUTC()
runManifest.MarkSucceeded(completedAt)
syncRunManifestIdentityFromSession(m, runManifest)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, fmt.Errorf("save final run manifest %q: %w", runManifestPath, err)
}
return &RunSummary{
SessionID: cfg.Session.SessionID,
RunID: runID,
ManifestPath: manifestPath,
RunManifestPath: runManifestPath,
StageNames: runNames,
Executed: executed,
Skipped: skipped,
@@ -358,7 +424,7 @@ func applyStageResultToManifest(m *manifest.Manifest, stageName string, result *
}
}
func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, error) {
func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest, runID string) (bool, error) {
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil || m == nil {
return false, nil
}
@@ -374,11 +440,8 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err
m.Campaign = campaign
changed = true
}
if m.RunID == "" {
runID, err := artifacts.NewRunID()
if err != nil {
return false, err
}
runID = strings.TrimSpace(runID)
if runID != "" && m.RunID != runID {
m.RunID = runID
changed = true
}
@@ -410,8 +473,54 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err
return changed, nil
}
func requestedStageNames(stages []stage.Stage) []string {
out := make([]string, 0, len(stages))
for _, s := range stages {
if s == nil {
continue
}
out = append(out, s.Name())
}
return out
}
func applyStageResultToRunManifest(m *manifest.RunManifest, stageName string, result *stage.StageResult) {
if m == nil || result == nil {
return
}
sr := m.Stages[stageName]
if sr == nil {
return
}
if len(result.Logs) > 0 {
sr.Logs = append([]string(nil), result.Logs...)
}
if len(result.GeneratedConfigs) > 0 {
sr.GeneratedConfigs = append([]string(nil), result.GeneratedConfigs...)
}
if len(result.Metadata) > 0 {
sr.Metadata = result.Metadata
}
}
func syncRunManifestIdentityFromSession(session *manifest.Manifest, run *manifest.RunManifest) {
if session == nil || run == nil {
return
}
run.Campaign = session.Campaign
run.LocalWorkDir = session.LocalWorkDir
run.LocalSpoolDir = session.LocalSpoolDir
run.S3Bucket = session.S3Bucket
run.S3SessionPrefix = session.S3SessionPrefix
run.S3RunPrefix = session.S3RunPrefix
}
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 {

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)
}
@@ -315,6 +322,82 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) {
}
}
func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) {
cfg := testConfig(t)
run1, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{})
if err != nil {
t.Fatalf("first executeStages() error = %v", err)
}
run2, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{Force: true})
if err != nil {
t.Fatalf("second executeStages() error = %v", err)
}
if run1.RunID == "" || run2.RunID == "" {
t.Fatalf("run ids must be set, got %q and %q", run1.RunID, run2.RunID)
}
if run1.RunID == run2.RunID {
t.Fatalf("expected distinct run ids, got %q", run1.RunID)
}
if run1.RunManifestPath == "" || run2.RunManifestPath == "" {
t.Fatalf("run manifest paths must be set, got %q and %q", run1.RunManifestPath, run2.RunManifestPath)
}
if run1.RunManifestPath == run2.RunManifestPath {
t.Fatalf("expected distinct run manifest paths, got %q", run1.RunManifestPath)
}
for _, path := range []string{run1.RunManifestPath, run2.RunManifestPath} {
if _, statErr := os.Stat(path); statErr != nil {
t.Fatalf("run manifest missing at %q: %v", path, statErr)
}
}
store := &manifest.LocalStore{}
sessionManifest, err := store.Load(context.Background(), run2.ManifestPath)
if err != nil {
t.Fatalf("Load session manifest error = %v", err)
}
if sessionManifest.RunID != run2.RunID {
t.Fatalf("session manifest run_id = %q, want latest run id %q", sessionManifest.RunID, run2.RunID)
}
}
func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) {
cfg := testConfig(t)
store := &manifest.LocalStore{}
manifestPath := manifestPathFor(cfg)
existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC))
existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil)
if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil {
t.Fatalf("MkdirAll() error = %v", err)
}
if err := store.Save(context.Background(), manifestPath, existing); err != nil {
t.Fatalf("Save manifest error = %v", err)
}
summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{})
if err != nil {
t.Fatalf("executeStages() error = %v", err)
}
if len(summary.Skipped) != 1 || summary.Skipped[0] != "transcribe" {
t.Fatalf("summary = %#v, want skipped transcribe", summary)
}
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
sr := runManifest.Stages["transcribe"]
if sr == nil {
t.Fatal("run manifest transcribe stage missing")
}
if sr.Action != manifest.RunStageActionSkip {
t.Fatalf("action = %q, want %q", sr.Action, manifest.RunStageActionSkip)
}
if sr.Status != manifest.StatusSkipped {
t.Fatalf("status = %q, want %q", sr.Status, manifest.StatusSkipped)
}
}
func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) {
cases := []struct {
name string
@@ -341,7 +424,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 +439,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 +458,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 +470,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)
}

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)

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 {

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

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,51 @@ 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)
}
// SessionRunManifestPathForCampaign returns the canonical run manifest path under runs/{run_id}/manifest.json.
func SessionRunManifestPathForCampaign(rootDir, campaign, sessionID, runID string) string {
return filepath.Join(SessionRunRootForCampaign(rootDir, campaign, sessionID, runID), config.PathManifestFile)
}
// 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 +72,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 +94,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),
}

View File

@@ -14,6 +14,50 @@ 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 TestSessionRunManifestPathForCampaign(t *testing.T) {
root := "/tmp/workspace"
runID := "20260515T031522Z-a1b2c3d4"
got := SessionRunManifestPathForCampaign(root, "forsaken", "2026-04-19", runID)
want := filepath.Join(root, "work", "forsaken", "2026-04-19", "runs", runID, "manifest.json")
if got != want {
t.Fatalf("SessionRunManifestPathForCampaign() = %q, want %q", got, want)
}
}
func TestSessionSpoolAudioDir(t *testing.T) {
root := "/var/spool/narratio"
got := SessionSpoolAudioDir(root, "forsaken", "2026-04-19", "20260515T031522Z-a1b2c3d4")

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

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

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
}

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"

View File

@@ -0,0 +1,164 @@
package manifest
import (
"strings"
"time"
)
type RunManifestStatus string
const (
RunManifestStatusRunning RunManifestStatus = "running"
RunManifestStatusSucceeded RunManifestStatus = "succeeded"
RunManifestStatusFailed RunManifestStatus = "failed"
)
type RunStageAction string
const (
RunStageActionRun RunStageAction = "run"
RunStageActionSkip RunStageAction = "skip"
)
// RunStageRecord tracks lifecycle and provenance for one stage within a single invocation.
type RunStageRecord struct {
Name string `json:"name"`
Action RunStageAction `json:"action"`
Status StageStatus `json:"status"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
Outputs []ArtifactRecord `json:"outputs,omitempty"`
Logs []string `json:"logs,omitempty"`
GeneratedConfigs []string `json:"generated_configs,omitempty"`
Error *ErrorRecord `json:"error,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
// RunManifest is the invocation-scoped execution record under runs/{run_id}/manifest.json.
type RunManifest struct {
SessionID string `json:"session_id"`
Campaign string `json:"campaign,omitempty"`
RunID string `json:"run_id"`
Force bool `json:"force"`
RequestedStages []string `json:"requested_stages,omitempty"`
SessionManifestPath string `json:"session_manifest_path,omitempty"`
LocalWorkDir string `json:"local_workdir,omitempty"`
LocalSpoolDir string `json:"local_spool_dir,omitempty"`
S3Bucket string `json:"s3_bucket,omitempty"`
S3SessionPrefix string `json:"s3_session_prefix,omitempty"`
S3RunPrefix string `json:"s3_run_prefix,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
Status RunManifestStatus `json:"status"`
LastError *ErrorRecord `json:"last_error,omitempty"`
Stages map[string]*RunStageRecord `json:"stages"`
Metadata map[string]any `json:"metadata,omitempty"`
}
// NewRun constructs a new run manifest with deterministic timestamps.
func NewRun(sessionID, campaign, runID string, force bool, requestedStages []string, now time.Time) *RunManifest {
return &RunManifest{
SessionID: strings.TrimSpace(sessionID),
Campaign: strings.TrimSpace(campaign),
RunID: strings.TrimSpace(runID),
Force: force,
RequestedStages: append([]string(nil), requestedStages...),
CreatedAt: now,
UpdatedAt: now,
StartedAt: timePtr(now),
Status: RunManifestStatusRunning,
Stages: map[string]*RunStageRecord{},
}
}
func (m *RunManifest) SetStageAction(name string, action RunStageAction, at time.Time) {
s := m.ensureStage(name, at)
s.Action = action
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkStageRunning(name string, at time.Time) {
s := m.ensureStage(name, at)
s.Status = StatusRunning
s.StartedAt = timePtr(at)
s.CompletedAt = nil
s.Error = nil
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkStageSucceeded(name string, at time.Time, outputs []ArtifactRecord) {
s := m.ensureStage(name, at)
s.Status = StatusSucceeded
s.CompletedAt = timePtr(at)
s.Error = nil
s.Outputs = append([]ArtifactRecord(nil), outputs...)
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkStageFailed(name string, at time.Time, message string) {
s := m.ensureStage(name, at)
s.Status = StatusFailed
s.CompletedAt = timePtr(at)
s.Error = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)}
s.UpdatedAt = at
m.LastError = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)}
m.UpdatedAt = at
m.Status = RunManifestStatusFailed
m.CompletedAt = timePtr(at)
}
func (m *RunManifest) MarkStageSkipped(name string, at time.Time, reason string) {
s := m.ensureStage(name, at)
s.Status = StatusSkipped
s.CompletedAt = timePtr(at)
s.Error = &ErrorRecord{Message: strings.TrimSpace(reason), Code: "skipped", At: timePtr(at)}
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkSucceeded(at time.Time) {
m.Status = RunManifestStatusSucceeded
m.CompletedAt = timePtr(at)
m.UpdatedAt = at
}
func (m *RunManifest) MarkFailed(at time.Time, message string) {
m.Status = RunManifestStatusFailed
m.CompletedAt = timePtr(at)
m.LastError = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)}
m.UpdatedAt = at
}
func (m *RunManifest) ensureStage(name string, at time.Time) *RunStageRecord {
if m.Stages == nil {
m.Stages = map[string]*RunStageRecord{}
}
stageName := strings.TrimSpace(name)
s, ok := m.Stages[stageName]
if !ok || s == nil {
s = &RunStageRecord{
Name: stageName,
Action: RunStageActionRun,
Status: StatusPending,
CreatedAt: at,
UpdatedAt: at,
}
m.Stages[stageName] = s
}
if s.Name == "" {
s.Name = stageName
}
if s.CreatedAt.IsZero() {
s.CreatedAt = at
}
return s
}

View File

@@ -0,0 +1,50 @@
package manifest
import (
"testing"
"time"
)
func TestRunManifestStageMarkHelpers(t *testing.T) {
rm := NewRun("2026-05-03", "forsaken", "20260517T000000Z-abcdef12", false, []string{"prepare"}, time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
if rm.Status != RunManifestStatusRunning {
t.Fatalf("status = %q, want %q", rm.Status, RunManifestStatusRunning)
}
runningAt := time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC)
rm.SetStageAction("prepare", RunStageActionRun, runningAt)
rm.MarkStageRunning("prepare", runningAt)
rm.MarkStageSucceeded("prepare", runningAt.Add(30*time.Second), []ArtifactRecord{
{Kind: "input", LocalPath: "inputs/session.yml"},
})
stage := rm.Stages["prepare"]
if stage == nil {
t.Fatal("prepare stage missing")
}
if stage.Action != RunStageActionRun {
t.Fatalf("action = %q, want %q", stage.Action, RunStageActionRun)
}
if stage.Status != StatusSucceeded {
t.Fatalf("status = %q, want %q", stage.Status, StatusSucceeded)
}
skippedAt := runningAt.Add(1 * time.Minute)
rm.SetStageAction("notify", RunStageActionSkip, skippedAt)
rm.MarkStageSkipped("notify", skippedAt, "already_succeeded")
skipped := rm.Stages["notify"]
if skipped == nil {
t.Fatal("notify stage missing")
}
if skipped.Action != RunStageActionSkip {
t.Fatalf("action = %q, want %q", skipped.Action, RunStageActionSkip)
}
if skipped.Status != StatusSkipped {
t.Fatalf("status = %q, want %q", skipped.Status, StatusSkipped)
}
rm.MarkSucceeded(skippedAt.Add(10 * time.Second))
if rm.Status != RunManifestStatusSucceeded {
t.Fatalf("status = %q, want %q", rm.Status, RunManifestStatusSucceeded)
}
}

View File

@@ -129,6 +129,89 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error {
return nil
}
// CreateRun returns a new in-memory run manifest for one invocation.
func (s *LocalStore) CreateRun(
ctx context.Context,
sessionID, campaign, runID string,
force bool,
requestedStages []string,
) (*RunManifest, error) {
if err := checkContext(ctx); err != nil {
return nil, err
}
if strings.TrimSpace(sessionID) == "" {
return nil, fmt.Errorf("create run manifest: session_id is required")
}
if strings.TrimSpace(runID) == "" {
return nil, fmt.Errorf("create run manifest: run_id is required")
}
now := time.Now().UTC()
return NewRun(sessionID, campaign, runID, force, requestedStages, now), nil
}
// LoadRun reads and validates a local JSON run manifest from path.
func (s *LocalStore) LoadRun(ctx context.Context, path string) (*RunManifest, error) {
if err := checkContext(ctx); err != nil {
return nil, err
}
if strings.TrimSpace(path) == "" {
return nil, fmt.Errorf("load run manifest: path is required")
}
data, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("load run manifest %q: %w", path, err)
}
var m RunManifest
if err := json.Unmarshal(data, &m); err != nil {
return nil, fmt.Errorf("decode run manifest %q: %w", path, err)
}
if err := validateLoadedRunManifest(&m); err != nil {
return nil, fmt.Errorf("run manifest %q invalid: %w", path, err)
}
normalizeRunManifest(&m)
return &m, nil
}
// SaveRun writes the run manifest to path atomically via temp file + rename.
func (s *LocalStore) SaveRun(ctx context.Context, path string, m *RunManifest) error {
if err := checkContext(ctx); err != nil {
return err
}
if strings.TrimSpace(path) == "" {
return fmt.Errorf("save run manifest: path is required")
}
if m == nil {
return fmt.Errorf("save run manifest: manifest is nil")
}
if strings.TrimSpace(m.SessionID) == "" {
return fmt.Errorf("save run manifest: session_id is required")
}
if strings.TrimSpace(m.RunID) == "" {
return fmt.Errorf("save run manifest: run_id is required")
}
if m.CreatedAt.IsZero() {
return fmt.Errorf("save run manifest: created_at is required")
}
m.UpdatedAt = time.Now().UTC()
if m.Stages == nil {
m.Stages = map[string]*RunStageRecord{}
}
data, err := json.MarshalIndent(m, "", " ")
if err != nil {
return fmt.Errorf("save run manifest: marshal: %w", err)
}
data = append(data, '\n')
return writeJSONAtomically(ctx, path, ".run-manifest.json.tmp-*", data)
}
func validateLoadedManifest(m *Manifest) error {
if m == nil {
return fmt.Errorf("manifest is nil")
@@ -161,6 +244,88 @@ func normalizeManifest(m *Manifest) {
}
}
func validateLoadedRunManifest(m *RunManifest) error {
if m == nil {
return fmt.Errorf("manifest is nil")
}
if strings.TrimSpace(m.SessionID) == "" {
return fmt.Errorf("session_id is required")
}
if strings.TrimSpace(m.RunID) == "" {
return fmt.Errorf("run_id is required")
}
if m.CreatedAt.IsZero() {
return fmt.Errorf("created_at is required")
}
if m.UpdatedAt.IsZero() {
return fmt.Errorf("updated_at is required")
}
return nil
}
func normalizeRunManifest(m *RunManifest) {
if m.Stages == nil {
m.Stages = map[string]*RunStageRecord{}
}
for name, stage := range m.Stages {
if stage == nil {
stage = &RunStageRecord{
Name: name,
Action: RunStageActionRun,
Status: StatusPending,
CreatedAt: m.CreatedAt,
UpdatedAt: m.UpdatedAt,
}
m.Stages[name] = stage
}
if stage.Name == "" {
stage.Name = name
}
}
}
func writeJSONAtomically(ctx context.Context, path, tempPattern string, data []byte) error {
dir := filepath.Dir(path)
if err := os.MkdirAll(dir, 0o755); err != nil {
return fmt.Errorf("create directory %q: %w", dir, err)
}
tmp, err := os.CreateTemp(dir, tempPattern)
if err != nil {
return fmt.Errorf("create temp file: %w", err)
}
tmpName := tmp.Name()
removeTmp := true
defer func() {
if removeTmp {
_ = os.Remove(tmpName)
}
}()
if _, err := tmp.Write(data); err != nil {
_ = tmp.Close()
return fmt.Errorf("write temp file: %w", err)
}
if err := tmp.Sync(); err != nil {
_ = tmp.Close()
return fmt.Errorf("sync temp file: %w", err)
}
if err := tmp.Close(); err != nil {
return fmt.Errorf("close temp file: %w", err)
}
if err := checkContext(ctx); err != nil {
return err
}
if err := os.Rename(tmpName, path); err != nil {
return fmt.Errorf("rename temp file: %w", err)
}
removeTmp = false
return nil
}
func checkContext(ctx context.Context) error {
if ctx == nil {
return nil

View File

@@ -150,3 +150,73 @@ func TestLoadRejectsInvalidManifest(t *testing.T) {
t.Fatalf("error = %q, want session_id validation", err.Error())
}
}
func TestLocalStoreCreateSaveLoadRunManifestRoundTrip(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
run, err := store.CreateRun(
ctx,
"2026-05-03",
"forsaken",
"20260517T000000Z-abcdef12",
true,
[]string{"prepare", "transcribe"},
)
if err != nil {
t.Fatalf("CreateRun() error = %v", err)
}
run.SessionManifestPath = "/var/lib/narratio/work/forsaken/2026-05-03/manifest.json"
run.MarkStageRunning("prepare", time.Date(2026, 5, 3, 12, 1, 0, 0, time.UTC))
run.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 12, 2, 0, 0, time.UTC), []ArtifactRecord{
{Kind: "input", LocalPath: "inputs/session.yml"},
})
run.MarkSucceeded(time.Date(2026, 5, 3, 12, 3, 0, 0, time.UTC))
path := filepath.Join(t.TempDir(), "run-manifest.json")
if err := store.SaveRun(ctx, path, run); err != nil {
t.Fatalf("SaveRun() error = %v", err)
}
loaded, err := store.LoadRun(ctx, path)
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
if loaded.SessionID != "2026-05-03" {
t.Fatalf("SessionID = %q, want %q", loaded.SessionID, "2026-05-03")
}
if loaded.Campaign != "forsaken" {
t.Fatalf("Campaign = %q, want %q", loaded.Campaign, "forsaken")
}
if loaded.RunID != "20260517T000000Z-abcdef12" {
t.Fatalf("RunID = %q, want %q", loaded.RunID, "20260517T000000Z-abcdef12")
}
if loaded.Status != RunManifestStatusSucceeded {
t.Fatalf("Status = %q, want %q", loaded.Status, RunManifestStatusSucceeded)
}
if loaded.Stages["prepare"] == nil || loaded.Stages["prepare"].Status != StatusSucceeded {
t.Fatalf("prepare stage = %#v, want succeeded", loaded.Stages["prepare"])
}
if loaded.Stages["prepare"].Action != RunStageActionRun {
t.Fatalf("prepare action = %q, want %q", loaded.Stages["prepare"].Action, RunStageActionRun)
}
}
func TestLoadRunRejectsInvalidManifest(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
path := filepath.Join(t.TempDir(), "run-manifest.json")
if err := os.WriteFile(path, []byte(`{"session_id":"2026-05-03"}`), 0o644); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
_, err := store.LoadRun(ctx, path)
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "run_id is required") {
t.Fatalf("error = %q, want run_id validation", err.Error())
}
}

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{

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

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

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

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 {

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{

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)

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

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

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)

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{

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

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

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

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)

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

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)

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