package app import ( "context" "errors" "fmt" "os" "path/filepath" "strings" "testing" "time" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" "gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) type failingStage struct { name string err error } func (s failingStage) Name() string { return s.name } func (s failingStage) Declares() stage.IODecl { return stage.IODecl{} } func (s failingStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { return nil, s.err } type countingStage struct { name string runs *int } func (s countingStage) Name() string { return s.name } func (s countingStage) Declares() stage.IODecl { return stage.IODecl{} } func (s countingStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { *s.runs = *s.runs + 1 return &stage.StageResult{Metadata: map[string]any{"counting": true}}, nil } type captureSelectedArtifactsStage struct { name string captured *[]string } func (s captureSelectedArtifactsStage) Name() string { return s.name } func (s captureSelectedArtifactsStage) Declares() stage.IODecl { return stage.IODecl{} } func (s captureSelectedArtifactsStage) Run(_ context.Context, env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { if s.captured != nil { *s.captured = append((*s.captured)[:0], env.SelectedArtifactKeys...) } return &stage.StageResult{Metadata: map[string]any{"captured": true}}, nil } type analyzeOutputStage struct { output artifacts.Ref } func (s analyzeOutputStage) Name() string { return "analyze" } func (s analyzeOutputStage) Declares() stage.IODecl { return stage.IODecl{} } func (s analyzeOutputStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { return &stage.StageResult{ Outputs: []artifacts.Ref{s.output}, }, nil } type selectedAnalyzeArtifactStage struct { expected []string } func (s selectedAnalyzeArtifactStage) Name() string { return "analyze" } func (s selectedAnalyzeArtifactStage) Declares() stage.IODecl { return stage.IODecl{} } func (s selectedAnalyzeArtifactStage) Run(_ context.Context, env *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { if len(env.SelectedArtifactKeys) != len(s.expected) { return nil, fmt.Errorf("selected artifacts len = %d, want %d", len(env.SelectedArtifactKeys), len(s.expected)) } for i := range s.expected { if env.SelectedArtifactKeys[i] != s.expected[i] { return nil, fmt.Errorf("selected artifacts[%d] = %q, want %q", i, env.SelectedArtifactKeys[i], s.expected[i]) } } outputPath := filepath.Join( artifacts.SessionWorkDirForCampaign(env.Config.Pipeline.Workspace.Root, env.Config.Session.Campaign, m.SessionID), "artifacts", "player_handout.md", ) if err := os.MkdirAll(filepath.Dir(outputPath), 0o755); err != nil { return nil, fmt.Errorf("mkdir artifact dir: %w", err) } if err := os.WriteFile(outputPath, []byte("player handout\n"), 0o644); err != nil { return nil, fmt.Errorf("write player handout: %w", err) } return &stage.StageResult{ Outputs: []artifacts.Ref{ { Kind: "player_handout", Category: "artifacts", RelativePath: "artifacts/player_handout.md", AbsolutePath: outputPath, }, }, Metadata: map[string]any{ "stage": "analyze", }, }, nil } func TestExecuteStagesPropagatesSelectedArtifactsToEnv(t *testing.T) { cfg := testConfig(t) captured := []string{} stageToRun := captureSelectedArtifactsStage{name: "analyze", captured: &captured} _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{ SelectedArtifacts: []string{"player_handout", "session_recap"}, }) if err != nil { t.Fatalf("executeStages() error = %v", err) } if len(captured) != 2 { t.Fatalf("captured len = %d, want 2 (%v)", len(captured), captured) } if captured[0] != "player_handout" || captured[1] != "session_recap" { t.Fatalf("captured = %v, want [player_handout session_recap]", captured) } } func TestExecuteStagesAnalyzeOutputsPersistAsScriptoriumArtifacts(t *testing.T) { cfg := testConfig(t) storeForPaths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) sessionPaths := storeForPaths.SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) outputPath := filepath.Join(sessionPaths.ArtifactsDir, "session_recap.md") stageToRun := analyzeOutputStage{ output: artifacts.Ref{ Kind: "session_recap", Category: "artifacts", RelativePath: "artifacts/session_recap.md", AbsolutePath: outputPath, }, } summary, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } store := &manifest.LocalStore{} sessionManifest, err := store.Load(context.Background(), summary.ManifestPath) if err != nil { t.Fatalf("load session manifest: %v", err) } sessionStage := sessionManifest.Stages["analyze"] if sessionStage == nil { t.Fatal("session manifest analyze stage missing") } if len(sessionStage.Outputs) != 1 { t.Fatalf("session analyze outputs len = %d, want 1", len(sessionStage.Outputs)) } sessionOutput := sessionStage.Outputs[0] if sessionOutput.Kind != "scriptorium_artifact" { t.Fatalf("session output kind = %q, want scriptorium_artifact", sessionOutput.Kind) } if sessionOutput.SourceID != "narratio.artifact.session_recap" { t.Fatalf("session output source_id = %q, want narratio.artifact.session_recap", sessionOutput.SourceID) } if sessionOutput.LocalPath != outputPath { t.Fatalf("session output local_path = %q, want %q", sessionOutput.LocalPath, outputPath) } runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath) if err != nil { t.Fatalf("load run manifest: %v", err) } runStage := runManifest.Stages["analyze"] if runStage == nil { t.Fatal("run manifest analyze stage missing") } if len(runStage.Outputs) != 1 { t.Fatalf("run analyze outputs len = %d, want 1", len(runStage.Outputs)) } runOutput := runStage.Outputs[0] if runOutput.Kind != "scriptorium_artifact" { t.Fatalf("run output kind = %q, want scriptorium_artifact", runOutput.Kind) } if runOutput.SourceID != "narratio.artifact.session_recap" { t.Fatalf("run output source_id = %q, want narratio.artifact.session_recap", runOutput.SourceID) } if runOutput.LocalPath != outputPath { t.Fatalf("run output local_path = %q, want %q", runOutput.LocalPath, outputPath) } } func TestNeedsObjectStoreForRunPrepareWithPreviousRequirements(t *testing.T) { tests := []struct { name string previousSessionID string want bool }{ { name: "previous session configured", previousSessionID: "2026-05-10", want: true, }, { name: "previous session missing", previousSessionID: "", want: false, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { cfg := &config.Config{ Pipeline: &config.PipelineConfig{ Scriptorium: &config.ScriptoriumConfig{ Artifacts: map[string]config.ScriptoriumArtifactConfig{ "session_recap": { Enabled: true, Inputs: map[string]config.ScriptoriumInputConfig{ "previous_recap": { Source: "narratio.previous_session.artifact.session_recap", Required: true, }, }, }, }, }, }, Session: &config.SessionConfig{ PreviousSessionID: tt.previousSessionID, }, } got := needsObjectStoreForRun(cfg, []stage.Stage{countingStage{name: "prepare", runs: new(int)}}) if got != tt.want { t.Fatalf("needsObjectStoreForRun() = %v, want %v", got, tt.want) } }) } } func TestExecuteStagesArchiveSkipsRequiredUnselectedConfiguredPromotion(t *testing.T) { cfg := testConfig(t) cfg.Pipeline.Storage.S3 = &config.StorageS3Config{ Bucket: "my-dnd-archive", RootPrefix: "dnd", } cfg.Pipeline.Archive = &config.ArchiveConfig{ Enabled: boolPtr(true), UploadRun: boolPtr(true), PromoteArtifacts: []config.ArchivePromotionRule{ {Source: "narratio.artifact.session_recap", Dest: "artifacts/session_recap.md", Required: boolPtr(true)}, }, } cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ Artifacts: map[string]config.ScriptoriumArtifactConfig{ "session_recap": { OutputPath: "artifacts/session_recap.md", }, }, } store := &manifest.LocalStore{} manifestPath := manifestPathFor(cfg) seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) seed.Campaign = cfg.Session.Campaign for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim"} { seed.MarkStageSucceeded(stageName, time.Now().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, seed); err != nil { t.Fatalf("Save manifest error = %v", err) } archiveStageImpl, err := stage.Select("archive") if err != nil { t.Fatalf("Select(archive) error = %v", err) } summary, err := executeStages( context.Background(), cfg, []stage.Stage{ selectedAnalyzeArtifactStage{expected: []string{"player_handout"}}, archiveStageImpl, }, RunOptions{ SelectedArtifacts: []string{"player_handout"}, Env: &Env{ObjectStore: &storage.FakeBackend{}}, }, ) if err != nil { t.Fatalf("executeStages() error = %v", err) } if len(summary.Executed) != 2 || summary.Executed[0] != "analyze" || summary.Executed[1] != "archive" { t.Fatalf("executed = %#v, want analyze and archive", summary.Executed) } loadedManifest, err := store.Load(context.Background(), summary.ManifestPath) if err != nil { t.Fatalf("Load manifest error = %v", err) } meta := loadedManifest.Stages["archive"].Metadata skipped, ok := meta["skipped_unselected_promotions"].([]any) if !ok || len(skipped) != 1 { t.Fatalf("skipped_unselected_promotions = %#v, want one item", meta["skipped_unselected_promotions"]) } item, ok := skipped[0].(map[string]any) if !ok { t.Fatalf("skipped item = %#v, want object", skipped[0]) } if item["source"] != "narratio.artifact.session_recap" || item["dest"] != "artifacts/session_recap.md" || item["required"] != true { t.Fatalf("skipped item = %#v, want required session_recap promotion", item) } } func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { cfg := testConfig(t) summary, err := executeStages(context.Background(), cfg, BuildFullPlan(), RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } if len(summary.StageNames) != 9 || len(summary.Executed) != 9 || len(summary.Skipped) != 0 { t.Fatalf("summary = %#v, want all 9 executed", summary) } store := &manifest.LocalStore{} m, err := store.Load(context.Background(), summary.ManifestPath) if err != nil { t.Fatalf("Load manifest error = %v", err) } for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { sr := m.Stages[name] if sr == nil { t.Fatalf("missing stage record %q", name) } if sr.Status != manifest.StatusSucceeded { t.Fatalf("stage %q status = %q, want %q", name, sr.Status, manifest.StatusSucceeded) } if name == "prepare" { if sr.Metadata == nil || sr.Metadata["prepared"] != true { t.Fatalf("prepare metadata missing prepared=true: %#v", sr.Metadata) } continue } if name == "transcribe" { if sr.Metadata == nil || sr.Metadata["stage"] != "transcribe" { t.Fatalf("transcribe metadata missing stage=transcribe: %#v", sr.Metadata) } if len(sr.Outputs) == 0 { t.Fatalf("transcribe outputs missing") } continue } if name == "merge" { if sr.Metadata == nil || sr.Metadata["stage"] != "merge" { t.Fatalf("merge metadata missing stage=merge: %#v", sr.Metadata) } if len(sr.Outputs) == 0 { t.Fatalf("merge outputs missing") } if len(sr.Logs) == 0 { t.Fatalf("merge logs missing") } if len(sr.GeneratedConfigs) == 0 { t.Fatalf("merge generated configs missing") } continue } if name == "polish" { if sr.Metadata == nil || sr.Metadata["stage"] != "polish" { t.Fatalf("polish metadata missing stage=polish: %#v", sr.Metadata) } if len(sr.Outputs) == 0 { t.Fatalf("polish outputs missing") } if len(sr.Logs) == 0 { t.Fatalf("polish logs missing") } if len(sr.GeneratedConfigs) == 0 { t.Fatalf("polish generated configs missing") } continue } if name == "normalize" { if sr.Metadata == nil || sr.Metadata["stage"] != "normalize" { t.Fatalf("normalize metadata missing stage=normalize: %#v", sr.Metadata) } if len(sr.Outputs) == 0 { t.Fatalf("normalize outputs missing") } continue } if name == "analyze" { if sr.Metadata == nil || sr.Metadata["stage"] != "analyze" { t.Fatalf("analyze metadata missing stage=analyze: %#v", sr.Metadata) } if sr.Metadata["skipped"] != true { t.Fatalf("analyze metadata missing skipped=true when scriptorium is unconfigured: %#v", sr.Metadata) } continue } if name == "trim" { if sr.Metadata == nil || sr.Metadata["stage"] != "trim" { t.Fatalf("trim metadata missing stage=trim: %#v", sr.Metadata) } if sr.Metadata["trim_action"] != "copy_disabled" { t.Fatalf("trim metadata missing trim_action=copy_disabled: %#v", sr.Metadata) } if len(sr.Outputs) == 0 { t.Fatalf("trim outputs missing") } continue } if name == "archive" { if sr.Metadata == nil || sr.Metadata["stage"] != "archive" { t.Fatalf("archive metadata missing stage=archive: %#v", sr.Metadata) } if sr.Metadata["skipped"] != true { t.Fatalf("archive metadata missing skipped=true for test config without archive section: %#v", sr.Metadata) } continue } if sr.Metadata == nil || sr.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", name) } } if len(m.Inputs) == 0 { t.Fatalf("manifest inputs should be recorded by prepare") } if _, err := os.Stat(summary.ManifestPath); err != nil { t.Fatalf("manifest file missing at %q: %v", summary.ManifestPath, err) } } func TestExecuteStagesSkipSucceededWhenNotForced(t *testing.T) { cfg := testConfig(t) manifestPath := manifestPathFor(cfg) store := &manifest.LocalStore{} existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{{Kind: "transcript_raw", LocalPath: "existing.json"}}) existing.Stages["transcribe"].Metadata = map[string]any{"kept": true} 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) } runs := 0 stageToRun := countingStage{name: "transcribe", runs: &runs} summary, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } if runs != 0 { t.Fatalf("runs = %d, want 0 due to skip", runs) } if len(summary.Executed) != 0 || len(summary.Skipped) != 1 || summary.Skipped[0] != "transcribe" { t.Fatalf("summary = %#v, want skipped transcribe", summary) } loaded, err := store.Load(context.Background(), manifestPath) if err != nil { t.Fatalf("Load manifest error = %v", err) } sr := loaded.Stages["transcribe"] if sr == nil || sr.Status != manifest.StatusSucceeded { t.Fatalf("transcribe status = %#v, want succeeded", sr) } if sr.Metadata == nil || sr.Metadata["kept"] != true { t.Fatalf("transcribe metadata = %#v, want preserved", sr.Metadata) } } func TestExecuteStagesForceRerunsSucceeded(t *testing.T) { cfg := testConfig(t) manifestPath := manifestPathFor(cfg) store := &manifest.LocalStore{} 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) } runs := 0 stageToRun := countingStage{name: "transcribe", runs: &runs} summary, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: true}) if err != nil { t.Fatalf("executeStages() error = %v", err) } if runs != 1 { t.Fatalf("runs = %d, want 1 with force", runs) } if len(summary.Executed) != 1 || summary.Executed[0] != "transcribe" || len(summary.Skipped) != 0 { t.Fatalf("summary = %#v, want executed transcribe", summary) } } func TestExecuteStagesForceSuccessInvalidatesDownstreamSucceededStages(t *testing.T) { cfg := testConfig(t) manifestPath := manifestPathFor(cfg) store := &manifest.LocalStore{} existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "archive", "notify"} { existing.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) } existing.MarkStageFailed("analyze", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), "previous analyze failure") 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) } runs := 0 stageToRun := countingStage{name: "polish", runs: &runs} summary, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: true}) if err != nil { t.Fatalf("executeStages() error = %v", err) } if runs != 1 { t.Fatalf("runs = %d, want 1 with force", runs) } if len(summary.Executed) != 1 || summary.Executed[0] != "polish" || len(summary.Skipped) != 0 { t.Fatalf("summary = %#v, want executed polish", summary) } loaded, err := store.Load(context.Background(), manifestPath) if err != nil { t.Fatalf("Load manifest error = %v", err) } if loaded.Stages["polish"] == nil || loaded.Stages["polish"].Status != manifest.StatusSucceeded { t.Fatalf("polish status = %#v, want succeeded", loaded.Stages["polish"]) } for _, stageName := range []string{"normalize", "trim", "archive", "notify"} { if loaded.Stages[stageName] == nil || loaded.Stages[stageName].Status != manifest.StatusStale { t.Fatalf("%s status = %#v, want stale", stageName, loaded.Stages[stageName]) } } if loaded.Stages["analyze"] == nil || loaded.Stages["analyze"].Status != manifest.StatusFailed { t.Fatalf("analyze status = %#v, want preserved failed", loaded.Stages["analyze"]) } if loaded.Stages["transcribe"] == nil || loaded.Stages["transcribe"].Status != manifest.StatusSucceeded { t.Fatalf("transcribe status = %#v, want preserved succeeded", loaded.Stages["transcribe"]) } } func TestExecuteStagesFailureUpdatesManifest(t *testing.T) { cfg := testConfig(t) stages := []stage.Stage{ BuildFullPlan()[0], failingStage{name: "transcribe", err: errors.New("boom")}, BuildFullPlan()[2], } summary, err := executeStages(context.Background(), cfg, stages, RunOptions{}) if err == nil { t.Fatal("expected error, got nil") } if summary != nil { t.Fatalf("summary = %#v, want nil on failure", summary) } if !strings.Contains(err.Error(), "stage \"transcribe\" failed") { t.Fatalf("error = %q, want stage failure", err.Error()) } manifestPath := manifestPathFor(cfg) store := &manifest.LocalStore{} m, loadErr := store.Load(context.Background(), manifestPath) if loadErr != nil { t.Fatalf("Load manifest error = %v", loadErr) } if got := m.Stages["prepare"]; got == nil || got.Status != manifest.StatusSucceeded { t.Fatalf("prepare status = %#v, want succeeded", got) } if got := m.Stages["transcribe"]; got == nil || got.Status != manifest.StatusFailed { t.Fatalf("transcribe status = %#v, want failed", got) } if got := m.Stages["merge"]; got != nil { t.Fatalf("merge should not run, got %#v", got) } } func TestExecuteStagesLoadsExistingManifest(t *testing.T) { cfg := testConfig(t) manifestPath := manifestPathFor(cfg) store := &manifest.LocalStore{} 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.Campaign, cfg.Session.SessionID, "audio", "alice.flac", ) if err := os.MkdirAll(filepath.Dir(audioPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } if err := os.WriteFile(audioPath, []byte("audio"), 0o644); err != nil { t.Fatalf("WriteFile() error = %v", err) } existing.Inputs = append(existing.Inputs, manifest.InputRecord{ Kind: "audio", Path: audioPath, }) 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) } _, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } loaded, err := store.Load(context.Background(), manifestPath) if err != nil { t.Fatalf("Load manifest error = %v", err) } if loaded.Stages["prepare"] == nil || loaded.Stages["prepare"].Status != manifest.StatusSucceeded { t.Fatalf("existing stage prepare should remain succeeded") } if loaded.Stages["transcribe"] == nil || loaded.Stages["transcribe"].Status != manifest.StatusSucceeded { t.Fatalf("transcribe should be succeeded after run") } } 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 TestExecuteStagesRunLocalArtifactsAndCanonicalPromotion(t *testing.T) { cfg := testConfig(t) stages := []stage.Stage{ BuildFullPlan()[0], // prepare BuildFullPlan()[1], // transcribe BuildFullPlan()[2], // merge BuildFullPlan()[3], // polish BuildFullPlan()[4], // normalize BuildFullPlan()[5], // trim } summary, err := executeStages(context.Background(), cfg, stages, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } if summary.RunID == "" { t.Fatal("run id must be set") } runRoot := artifacts.SessionRunRootForCampaign( cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, summary.RunID, ) paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) runLocalChecks := []string{ filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "alice.json"), filepath.Join(runRoot, "merge", "logs", "seriatim.stdout.log"), filepath.Join(runRoot, "polish", "config", "audita.generated.yml"), filepath.Join(runRoot, "normalize", "logs", "seriatim.normalize.stdout.log"), filepath.Join(runRoot, "trim", "outputs", "transcripts", "final.trimmed.json"), } for _, p := range runLocalChecks { if _, statErr := os.Stat(p); statErr != nil { t.Fatalf("run-local artifact missing at %q: %v", p, statErr) } } canonicalChecks := []string{ filepath.Join(paths.TranscriptsRawDir, "alice.json"), filepath.Join(paths.TranscriptsDir, "base.json"), filepath.Join(paths.TranscriptsDir, "polished.json"), filepath.Join(paths.TranscriptsDir, "final.json"), filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), } for _, p := range canonicalChecks { if _, statErr := os.Stat(p); statErr != nil { t.Fatalf("canonical promoted artifact missing at %q: %v", p, statErr) } } store := &manifest.LocalStore{} sessionManifest, err := store.Load(context.Background(), summary.ManifestPath) if err != nil { t.Fatalf("Load manifest error = %v", err) } if got := sessionManifest.Stages["trim"]; got == nil || len(got.Outputs) == 0 { t.Fatalf("trim stage outputs missing in session manifest: %#v", got) } for _, out := range sessionManifest.Stages["trim"].Outputs { if strings.Contains(out.LocalPath, string(filepath.Separator)+"runs"+string(filepath.Separator)) { t.Fatalf("session manifest output should be canonical, got run-local path %q", out.LocalPath) } if out.ProducerRunID != summary.RunID { t.Fatalf("producer_run_id = %q, want %q", out.ProducerRunID, summary.RunID) } } runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath) if err != nil { t.Fatalf("LoadRun() error = %v", err) } mergeStage := runManifest.Stages["merge"] if mergeStage == nil || len(mergeStage.Logs) == 0 { t.Fatalf("merge logs missing in run manifest: %#v", mergeStage) } for _, logPath := range mergeStage.Logs { if !strings.Contains(logPath, filepath.Join("runs", summary.RunID, "merge", "logs")) { t.Fatalf("run manifest merge log path = %q, want run-local merge logs path", logPath) } } } func TestExecuteStagesSkippedStagePreservesExistingOutputsProvenance(t *testing.T) { cfg := testConfig(t) store := &manifest.LocalStore{} manifestPath := manifestPathFor(cfg) existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{ { Kind: "transcript_raw", LocalPath: "transcripts/raw/alice.json", ProducerRunID: "20260501T000000Z-deadbeef", }, }) if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } if err := store.Save(context.Background(), manifestPath, existing); err != nil { t.Fatalf("Save manifest error = %v", err) } summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } if len(summary.Skipped) != 1 || summary.Skipped[0] != "transcribe" { t.Fatalf("summary = %#v, want skipped transcribe", summary) } loaded, err := store.Load(context.Background(), manifestPath) if err != nil { t.Fatalf("Load manifest error = %v", err) } got := loaded.Stages["transcribe"] if got == nil || len(got.Outputs) != 1 { t.Fatalf("transcribe outputs = %#v, want one preserved output", got) } if got.Outputs[0].ProducerRunID != "20260501T000000Z-deadbeef" { t.Fatalf("producer_run_id = %q, want preserved value", got.Outputs[0].ProducerRunID) } } func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { cases := []struct { name string env *Env }{ {name: "transcribe", env: &Env{WhisperX: &whisperx.FakeClient{Err: errors.New("transcribe fail")}}}, {name: "merge", env: &Env{Seriatim: &seriatim.FakeRunner{Err: errors.New("merge fail")}}}, {name: "polish", env: &Env{Audita: &audita.FakeRunner{Err: errors.New("polish fail")}}}, {name: "analyze", env: &Env{Scriptorium: &scriptorium.FakeRunner{RunErr: errors.New("analyze fail")}}}, {name: "archive", env: &Env{ObjectStore: &storage.FakeBackend{UploadErr: errors.New("archive fail")}}}, {name: "notify", env: &Env{Notifier: ¬ify.FakeSender{Err: errors.New("notify fail")}}}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { cfg := testConfig(t) selected, err := stage.Select(tc.name) if err != nil { t.Fatalf("Select() error = %v", err) } artifactStore := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) tc.env.Config = cfg tc.env.ArtifactStore = artifactStore tc.env.ManifestStore = &manifest.LocalStore{} if tc.name == "transcribe" { paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } audioPath := filepath.Join(paths.AudioDir, "alice.flac") if err := os.WriteFile(audioPath, []byte("audio"), 0o644); err != nil { t.Fatalf("write transcribe fixture audio: %v", err) } m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) m.Inputs = append(m.Inputs, manifest.InputRecord{Kind: "audio", Path: audioPath}) if err := tc.env.ManifestStore.Save(context.Background(), manifestPathFor(cfg), m); err != nil { t.Fatalf("seed manifest: %v", err) } } if tc.name == "merge" { paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } rawPath := filepath.Join(paths.TranscriptsRawDir, "alice.json") if err := os.MkdirAll(filepath.Dir(rawPath), 0o755); err != nil { t.Fatalf("mkdir raw dir: %v", err) } if err := os.WriteFile(rawPath, []byte(`{"segments":[]}`), 0o644); err != nil { t.Fatalf("write raw transcript: %v", err) } if err := os.WriteFile(filepath.Join(paths.InputsDir, "speakers.yml"), []byte("match: []\n"), 0o644); err != nil { t.Fatalf("write speakers: %v", err) } if err := os.WriteFile(filepath.Join(paths.InputsDir, "autocorrect.yml"), []byte("rules: []\n"), 0o644); err != nil { t.Fatalf("write autocorrect: %v", err) } } if tc.name == "polish" { paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } if err := os.WriteFile(filepath.Join(paths.TranscriptsDir, "base.json"), []byte(`{"segments":[]}`), 0o644); err != nil { t.Fatalf("write merged transcript: %v", err) } if err := os.WriteFile(filepath.Join(paths.InputsDir, "glossary.yml"), []byte("terms: []\n"), 0o644); err != nil { t.Fatalf("write glossary: %v", err) } } if tc.name == "analyze" { paths, ensureErr := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if ensureErr != nil { t.Fatalf("EnsureLayout() error = %v", ensureErr) } if err := os.WriteFile(filepath.Join(paths.TranscriptsDir, "polished.json"), []byte(`{"segments":[]}`), 0o644); err != nil { t.Fatalf("write processed transcript: %v", err) } cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ Binary: "scriptorium", Timeout: "10m", Artifacts: map[string]config.ScriptoriumArtifactConfig{ "session_recap": { Enabled: true, PromptID: "dnd.session_recap", OutputPath: "artifacts/session_recap.md", Inputs: map[string]config.ScriptoriumInputConfig{ "transcript": {Source: "narratio.transcript.polished", Required: true}, }, }, }, } } if tc.name == "archive" { cfg.Pipeline.Archive = &config.ArchiveConfig{ Enabled: boolPtr(true), UploadRun: boolPtr(true), } cfg.Pipeline.Storage.S3 = &config.StorageS3Config{ Bucket: "my-dnd-archive", RootPrefix: "dnd", } runID := "20260516T010203Z-0a1b2c3d" runWorkDir := filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.Campaign, cfg.Session.SessionID, runID) if err := os.MkdirAll(filepath.Join(runWorkDir, "inputs"), 0o755); err != nil { t.Fatalf("mkdir archive inputs dir: %v", err) } if err := os.WriteFile(filepath.Join(runWorkDir, "inputs", "session.yml"), []byte("session_id: 2026-05-03\n"), 0o644); err != nil { t.Fatalf("write archive fixture session.yml: %v", err) } if err := os.WriteFile(filepath.Join(runWorkDir, "manifest.json"), []byte("{}\n"), 0o644); err != nil { t.Fatalf("write archive fixture manifest.json: %v", err) } seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) seed.Campaign = cfg.Session.Campaign seed.RunID = runID seed.LocalWorkDir = runWorkDir seed.S3Bucket = "my-dnd-archive" seed.S3SessionPrefix = "dnd/campaigns/" + cfg.Session.Campaign + "/sessions/" + cfg.Session.SessionID + "/" seed.S3RunPrefix = seed.S3SessionPrefix + "runs/" + runID + "/" for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze"} { seed.MarkStageSucceeded(name, time.Now().UTC(), nil) } if err := tc.env.ManifestStore.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { t.Fatalf("seed archive manifest: %v", err) } } _, runErr := executeStages(context.Background(), cfg, []stage.Stage{selected}, RunOptions{Env: tc.env}) if runErr == nil { t.Fatal("expected error, got nil") } m, loadErr := tc.env.ManifestStore.Load(context.Background(), manifestPathFor(cfg)) if loadErr != nil { t.Fatalf("load manifest error = %v", loadErr) } sr := m.Stages[tc.name] if sr == nil { t.Fatalf("missing stage record %q", tc.name) } if sr.Status != manifest.StatusFailed { t.Fatalf("status = %q, want %q", sr.Status, manifest.StatusFailed) } }) } } func testConfig(t *testing.T) *config.Config { t.Helper() workspace := t.TempDir() cfgDir := t.TempDir() sessionPath := filepath.Join(cfgDir, "session.yml") campaignPath := filepath.Join(cfgDir, "campaign.yml") pipelinePath := filepath.Join(cfgDir, "pipeline.yml") mustWriteFile(t, pipelinePath, "workspace:\n root: "+workspace+"\n") mustWriteFile(t, campaignPath, "campaign: sample-campaign\ninputs:\n speakers_file: ./speakers.yml\n autocorrect_file: ./autocorrect.yml\n glossary_file: ./glossary.yml\n") mustWriteFile(t, sessionPath, "session_id: 2026-05-03\ncampaign: sample-campaign\ninputs:\n audio_dir: ./audio\n") mustWriteFile(t, filepath.Join(cfgDir, "speakers.yml"), "alice: alice.flac\n") mustWriteFile(t, filepath.Join(cfgDir, "autocorrect.yml"), "[]\n") mustWriteFile(t, filepath.Join(cfgDir, "glossary.yml"), "[]\n") mustWriteFile(t, filepath.Join(cfgDir, "audio", "alice.flac"), "audio") return &config.Config{ Pipeline: &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: workspace}}, Campaign: &config.CampaignConfig{Campaign: "sample-campaign"}, PipelinePath: pipelinePath, CampaignPath: campaignPath, SessionPath: sessionPath, StableInputs: config.ResolvedStableInputs{ SpeakersFile: config.ResolvedInputFile{ Path: "./speakers.yml", ConfigPath: campaignPath, Source: "campaign_config", }, AutocorrectFile: config.ResolvedInputFile{ Path: "./autocorrect.yml", ConfigPath: campaignPath, Source: "campaign_config", }, GlossaryFile: config.ResolvedInputFile{ Path: "./glossary.yml", ConfigPath: campaignPath, Source: "campaign_config", }, }, Session: &config.SessionConfig{ SessionID: "2026-05-03", Campaign: "sample-campaign", Inputs: config.SessionInputsConfig{ AudioDir: "./audio", SpeakersFile: "./speakers.yml", AutocorrectFile: "./autocorrect.yml", GlossaryFile: "./glossary.yml", }, }, } } func TestBuildDefaultRunnersWithOmittedToolSections(t *testing.T) { dir := t.TempDir() pipelinePath := filepath.Join(dir, "pipeline.yml") campaignPath := filepath.Join(dir, "campaign.yml") sessionPath := filepath.Join(dir, "session.yml") pipelineYAML := `workspace: root: ` + t.TempDir() + ` whisperx: transcribe_url: https://example.com/transcribe notification: timeout: 10s ` campaignYAML := `campaign: sample-campaign inputs: speakers_file: ./speakers.yml autocorrect_file: ./autocorrect.yml glossary_file: ./glossary.yml ` sessionYAML := `session_id: 2026-05-03 campaign: sample-campaign inputs: audio_dir: ./audio speakers_file: ./speakers.yml autocorrect_file: ./autocorrect.yml glossary_file: ./glossary.yml ` mustWriteFile(t, pipelinePath, pipelineYAML) mustWriteFile(t, campaignPath, campaignYAML) mustWriteFile(t, sessionPath, sessionYAML) cfg, err := config.Load(pipelinePath, sessionPath) if err != nil { t.Fatalf("Load() error = %v", err) } if err := config.Validate(cfg); err != nil { t.Fatalf("Validate() error = %v", err) } serRunner, err := buildDefaultSeriatimRunner(cfg) if err != nil { t.Fatalf("buildDefaultSeriatimRunner() error = %v", err) } if _, ok := serRunner.(*seriatim.SubprocessRunner); !ok { t.Fatalf("seriatim runner type = %T, want *seriatim.SubprocessRunner", serRunner) } audRunner, err := buildDefaultAuditaRunner(cfg) if err != nil { t.Fatalf("buildDefaultAuditaRunner() error = %v", err) } if _, ok := audRunner.(*audita.SubprocessRunner); !ok { t.Fatalf("audita runner type = %T, want *audita.SubprocessRunner", audRunner) } } func mustWriteFile(t *testing.T, path, contents string) { t.Helper() if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { t.Fatalf("MkdirAll(%q): %v", path, err) } if err := os.WriteFile(path, []byte(contents), 0o644); err != nil { t.Fatalf("WriteFile(%q): %v", path, err) } } func boolPtr(v bool) *bool { p := v return &p }