From 966b95b1766e7a49745455f631fe75898dcbf9d4 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 29 Aug 2026 18:17:05 +0000 Subject: [PATCH] Enforce bounded run prerequisites --- docs/internal/adapters.md | 21 +- docs/internal/manifest.md | 14 +- docs/operations.md | 24 +- docs/roadmap/implementation.md | 2 + docs/troubleshooting.md | 34 +++ internal/app/bounded_prerequisites.go | 62 +++++ internal/app/bounded_prerequisites_test.go | 288 +++++++++++++++++++++ internal/app/bounded_run_test.go | 11 + internal/app/plan.go | 13 +- internal/app/post_publish_cleanup_test.go | 27 +- internal/app/runner.go | 45 +++- 11 files changed, 509 insertions(+), 32 deletions(-) create mode 100644 internal/app/bounded_prerequisites.go create mode 100644 internal/app/bounded_prerequisites_test.go diff --git a/docs/internal/adapters.md b/docs/internal/adapters.md index c1f91db..a1e9676 100644 --- a/docs/internal/adapters.md +++ b/docs/internal/adapters.md @@ -35,16 +35,23 @@ Adapters do not own: ## Default Wiring -`internal/app/runner.go` initializes default adapters when not injected: +`internal/app/runner.go` initializes default adapters when not injected and +only when the selected execution plan needs them: -- WhisperX HTTP client from pipeline config. -- Seriatim subprocess runner. -- Audita subprocess runner. -- Scriptorium subprocess runner. -- Notarius subprocess runner when extraction is enabled. -- Noop notifier (`notify.NoopSender`). +- WhisperX HTTP client for `transcribe`. +- Seriatim subprocess runner for `merge`, `normalize`, `trim`, or `render`. +- Audita subprocess runner for `polish`. +- Scriptorium subprocess runner for `trim` or `analyze`. +- Notarius subprocess runner for `extract` when extraction is enabled. +- Noop notifier (`notify.NoopSender`) for `notify`. - Object store only when required by selected stages/config. +Remote publish locks are loaded only for a selected, enabled publish that +uploads a run. Shared session lifecycle setup still applies to every selected +range, but an unselected integration is neither initialized nor validated by +runner composition. Each selected stage retains its own fail-fast configuration +and input validation. + Notarius is composed only when extraction is enabled; the extract stage owns prepared reference resolution, receipt, bundle, and configured-lane policy. The adapter validates the ordered selector/absolute-path pairs and is the sole diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index 44f9f83..09a89eb 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -117,12 +117,20 @@ runner marks it stale and executes it. Session manifest is the authoritative stage-progress ledger across invocations. Run manifest is invocation-scoped audit state. +Before an explicitly bounded execution starts after `prepare`, the application +reads the session manifest and accepts only `succeeded` or `skipped` for every +excluded canonical prefix stage. The first other status or absent record fails +the request before layout mutation, adapter initialization, session-manifest +writes, or run-manifest creation. Excluded prefix records are not passed to +resume validators. Records after the selected end are not prerequisites and +may be made stale by selected work without being scheduled. + After a publish commits remotely, any configured local cleanup is first recorded as a session-manifest obligation before deletion begins. Each target becomes complete only after its confined deletion (or safe absence check) and a -successful manifest save. An incomplete obligation is retried on later -invocations independently of their selected stages and retains the committed -run and remote identity that authorized it. +successful manifest save. An incomplete obligation is retried when publish +executes again and retains the committed run and remote identity that authorized +it; an invocation that does not execute publish does not perform cleanup. Each invocation derives campaign, session, run, local-path, and remote-prefix metadata from the validated resolved configuration as one projection. A persisted diff --git a/docs/operations.md b/docs/operations.md index 7c76c8d..aa36c32 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -124,6 +124,24 @@ Omitting `--from` selects from `prepare`; omitting `--through` selects through The plan command uses the same selection contract and prints only the selected range. +Before a bounded run or plan whose range starts after `prepare`, every excluded +prefix stage must already have a session-manifest status of `succeeded` or +`skipped`. Narratio reports the first absent, pending, running, failed, stale, +or interrupted prerequisite without creating a run record or changing session +state. Widen `--from` to include that stage, or recover it explicitly before +retrying. Excluded prefix stages are not resume-validated or repaired as part +of the bounded invocation; selected stages still reject missing, unsafe, or +manifest-inconsistent inputs at their owning boundary. + +Stages after `--through` are not prerequisites and are never scheduled by the +bounded invocation. A selected forced stage can mark one of those succeeded +dependents stale through the fixed invalidation relation, but the dependent +does not execute until a later invocation selects it. Production composition +likewise initializes only collaborators needed by the selected range and +shared session lifecycle. In particular, render does not require Notarius or +Scriptorium, extract does not require Scriptorium, and analyze does not require +the transcription, Seriatim, Audita, or Notarius adapters. + ## Artifact Selection `--artifacts` can be used on `run`, `session plan`, `run-stage`, `analyze`, and @@ -451,7 +469,11 @@ Rules: - `pipeline.workspace.cleanup_after_publish=true` - Narratio first records the exact run-scoped cleanup obligation. If cleanup reports incomplete, the remote committed snapshot remains current; rerun - Narratio to retry only the outstanding confined local cleanup. + publish to retry only the outstanding confined local cleanup. + +Post-publish cleanup is evaluated only when `publish` actually executes in the +current invocation. A bounded range that excludes publish does not replay a +cleanup obligation as an unrelated side effect. ## Operational Caveats diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 3f0cf73..cd2b9b8 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -217,6 +217,8 @@ validation contract. ## Stage 4 — Bounded Runner Prerequisites And Composition +**Status: Completed** + ### Goal Make bounded execution honor its mutation boundary while still failing safely diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 2b544ee..bcbfa3d 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -117,6 +117,40 @@ Safe fix: Relevant reference: [CLI artifact selection](./cli.md). +## Bounded run prerequisite is unusable + +Symptom: + +- `run` or `session plan` reports that a prerequisite stage is absent or has a + pending, running, failed, stale, or interrupted status before the selected + start. + +Likely cause: + +- `--from` excludes upstream work that has not reached the terminal + `succeeded` or `skipped` state in the session manifest. + +Diagnostics: + +```bash +narratio session status 2026-04-04 +narratio session plan 2026-04-04 --from render --through analyze +``` + +Safe fix: + +- widen the bounded range to include the first reported stage, or recover that + stage explicitly with `run-stage` before retrying. The failed check does not + create a run record or modify the manifest. Narratio does not resume-validate + excluded prefix stages, and stages after `--through` are not prerequisites. + +If prerequisite statuses are terminal but a selected stage reports a missing, +unsafe, or checksum-inconsistent artifact, repair the artifact at the stage +that owns it; do not edit the manifest to bypass the selected stage's concrete +input validation. + +Relevant reference: [Operations: Stage Execution and Continuation Behavior](./operations.md#stage-execution-and-continuation-behavior). + ## Notarius executable missing Symptom: diff --git a/internal/app/bounded_prerequisites.go b/internal/app/bounded_prerequisites.go new file mode 100644 index 0000000..708c80c --- /dev/null +++ b/internal/app/bounded_prerequisites.go @@ -0,0 +1,62 @@ +package app + +import ( + "context" + "fmt" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func validateBoundedPrerequisites(plan BoundedPlan, m *manifest.Manifest) error { + if !plan.HasExplicitBounds() { + return nil + } + for _, name := range plan.PrefixNames() { + status := "absent" + if m != nil && m.Stages != nil && m.Stages[name] != nil { + stageStatus := m.Stages[name].Status + if stageStatus == manifest.StatusSucceeded || stageStatus == manifest.StatusSkipped { + continue + } + if stageStatus != "" { + status = string(stageStatus) + } + } + return fmt.Errorf( + "prerequisite stage %q has unusable status %q before selected start %q; widen the range with --from %s or recover %s explicitly", + name, + status, + plan.From(), + name, + name, + ) + } + return nil +} + +func inspectBoundedPrerequisites(ctx context.Context, cfg *config.Config, plan BoundedPlan, store manifest.Store) error { + if !plan.HasExplicitBounds() || len(plan.PrefixNames()) == 0 { + return nil + } + if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { + return fmt.Errorf("bounded prerequisite inspection requires resolved pipeline and session configuration") + } + if store == nil { + store = &manifest.LocalStore{} + } + path := artifacts.SessionManifestPathForCampaign( + cfg.Pipeline.Workspace.Root, + cfg.Session.Campaign, + cfg.Session.SessionID, + ) + m, present, err := loadManifestAtPathIfPresent(ctx, store, path) + if err != nil { + return err + } + if !present { + m = nil + } + return validateBoundedPrerequisites(plan, m) +} diff --git a/internal/app/bounded_prerequisites_test.go b/internal/app/bounded_prerequisites_test.go new file mode 100644 index 0000000..4a9b9e0 --- /dev/null +++ b/internal/app/bounded_prerequisites_test.go @@ -0,0 +1,288 @@ +package app + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "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" +) + +func TestValidateBoundedPrerequisitesRejectsFirstUnusablePrefixStatus(t *testing.T) { + plan := mustBoundedPlan(t, "render", "extract") + now := time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC) + for _, test := range []struct { + name string + status manifest.StageStatus + want string + }{ + {name: "absent", want: "absent"}, + {name: "pending", status: manifest.StatusPending, want: "pending"}, + {name: "running", status: manifest.StatusRunning, want: "running"}, + {name: "failed", status: manifest.StatusFailed, want: "failed"}, + {name: "stale", status: manifest.StatusStale, want: "stale"}, + {name: "interrupted", status: manifest.StatusInterrupted, want: "interrupted"}, + } { + t.Run(test.name, func(t *testing.T) { + m := manifest.New("session", now) + if test.status != "" { + m.Stages["prepare"] = &manifest.StageRecord{Name: "prepare", Status: test.status} + } + // A later terminal prefix must not hide the first unusable one. + m.MarkStageSucceeded("transcribe", now, nil) + err := validateBoundedPrerequisites(plan, m) + if err == nil { + t.Fatal("validateBoundedPrerequisites() error = nil") + } + for _, detail := range []string{`stage "prepare"`, `status "` + test.want + `"`, `selected start "render"`, "--from prepare", "recover prepare"} { + if !strings.Contains(err.Error(), detail) { + t.Fatalf("error = %q, want detail %q", err, detail) + } + } + }) + } +} + +func TestValidateBoundedPrerequisitesAcceptsSucceededAndSkippedPrefix(t *testing.T) { + plan := mustBoundedPlan(t, "render", "render") + m := manifest.New("session", time.Now().UTC()) + for index, name := range plan.PrefixNames() { + if index%2 == 0 { + m.MarkStageSucceeded(name, time.Now().UTC(), nil) + } else { + m.MarkStageSkipped(name, time.Now().UTC(), "not applicable") + } + } + if err := validateBoundedPrerequisites(plan, m); err != nil { + t.Fatalf("validateBoundedPrerequisites() error = %v", err) + } +} + +func TestValidateBoundedPrerequisitesHasNoPrefixAtPrepareAndIgnoresSuffix(t *testing.T) { + preparePlan := mustBoundedPlan(t, "prepare", "prepare") + if err := validateBoundedPrerequisites(preparePlan, nil); err != nil { + t.Fatalf("prepare prerequisite validation error = %v", err) + } + + renderPlan := mustBoundedPlan(t, "render", "render") + m := manifest.New("session", time.Now().UTC()) + markPrefixSucceeded(m, renderPlan) + m.MarkStageFailed("analyze", time.Now().UTC(), "later failure") + if err := validateBoundedPrerequisites(renderPlan, m); err != nil { + t.Fatalf("suffix status affected prerequisite validation: %v", err) + } +} + +func TestExecuteStagesRejectsBoundedPrerequisitesBeforePersistentMutation(t *testing.T) { + cfg := testConfig(t) + plan := mustBoundedPlan(t, "render", "render") + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Campaign = cfg.Session.Campaign + m.MarkStageRunning("prepare", time.Now().UTC()) + manifestPath := saveBoundedManifest(t, cfg, m) + before, err := os.ReadFile(manifestPath) + if err != nil { + t.Fatalf("read seeded manifest: %v", err) + } + + store := &prerequisiteMutationSpy{local: &manifest.LocalStore{}} + runs := 0 + _, err = executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "render", runs: &runs}}, RunOptions{ + Plan: plan, + Env: &Env{ManifestStore: store}, + }) + if err == nil || !strings.Contains(err.Error(), `stage "prepare" has unusable status "running"`) { + t.Fatalf("executeStages() error = %v, want running prerequisite", err) + } + if runs != 0 || store.creates != 0 || store.saves != 0 { + t.Fatalf("runs=%d manifest creates=%d saves=%d, want no mutation", runs, store.creates, store.saves) + } + after, err := os.ReadFile(manifestPath) + if err != nil { + t.Fatalf("read manifest after rejection: %v", err) + } + if string(after) != string(before) { + t.Fatal("manifest changed after prerequisite rejection") + } + if _, err := os.Stat(artifacts.SessionRunsDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)); !os.IsNotExist(err) { + t.Fatalf("runs directory stat error = %v, want not exist", err) + } +} + +func TestExecuteStagesBoundedCompositionUsesOnlySelectedCollaborators(t *testing.T) { + for _, test := range []struct { + name string + stageName string + configure func(*config.Config) + assertProbe func(*testing.T, *stage.Env) + }{ + { + name: "render", + stageName: "render", + assertProbe: func(t *testing.T, env *stage.Env) { + if env.Seriatim == nil || env.Notarius != nil || env.Scriptorium != nil { + t.Fatalf("render collaborators: seriatim=%v notarius=%v scriptorium=%v", env.Seriatim, env.Notarius, env.Scriptorium) + } + }, + }, + { + name: "extract", + stageName: "extract", + configure: func(cfg *config.Config) { + cfg.Pipeline.Notarius = &config.NotariusConfig{Enabled: true} + }, + assertProbe: func(t *testing.T, env *stage.Env) { + if env.Notarius == nil || env.Scriptorium != nil || env.Seriatim != nil { + t.Fatalf("extract collaborators: notarius=%v scriptorium=%v seriatim=%v", env.Notarius, env.Scriptorium, env.Seriatim) + } + }, + }, + { + name: "analyze", + stageName: "analyze", + assertProbe: func(t *testing.T, env *stage.Env) { + if env.Scriptorium == nil || env.Notarius != nil || env.Seriatim != nil || env.WhisperX != nil || env.Audita != nil { + t.Fatalf("analyze collaborators: scriptorium=%v notarius=%v seriatim=%v whisperx=%v audita=%v", env.Scriptorium, env.Notarius, env.Seriatim, env.WhisperX, env.Audita) + } + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + cfg := testConfig(t) + if test.configure != nil { + test.configure(cfg) + } + plan := mustBoundedPlan(t, test.stageName, test.stageName) + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Campaign = cfg.Session.Campaign + markPrefixSucceeded(m, plan) + saveBoundedManifest(t, cfg, m) + + var captured *stage.Env + _, err := executeStages(context.Background(), cfg, []stage.Stage{collaboratorProbeStage{name: test.stageName, captured: &captured}}, RunOptions{Plan: plan}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if captured == nil { + t.Fatal("selected stage did not run") + } + test.assertProbe(t, captured) + }) + } +} + +func TestExecuteStagesBoundedForceStalesButDoesNotRunDependentsOutsideRange(t *testing.T) { + cfg := testConfig(t) + plan := mustBoundedPlan(t, "render", "render") + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Campaign = cfg.Session.Campaign + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"} { + m.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + saveBoundedManifest(t, cfg, m) + runs := 0 + _, err := executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "render", runs: &runs}}, RunOptions{Plan: plan, Force: true}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if runs != 1 { + t.Fatalf("selected render runs = %d, want 1", runs) + } + loaded, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatalf("load manifest: %v", err) + } + for _, name := range []string{"analyze", "publish", "notify"} { + if loaded.Stages[name].Status != manifest.StatusStale { + t.Fatalf("stage %q status = %q, want stale", name, loaded.Stages[name].Status) + } + } + if loaded.Stages["extract"].Status != manifest.StatusSucceeded { + t.Fatalf("extract status = %q, want succeeded", loaded.Stages["extract"].Status) + } +} + +func TestExecuteStagesBoundedFailureStopsWithinSelectedRange(t *testing.T) { + cfg := testConfig(t) + plan := mustBoundedPlan(t, "render", "extract") + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Campaign = cfg.Session.Campaign + markPrefixSucceeded(m, plan) + saveBoundedManifest(t, cfg, m) + extractRuns := 0 + _, err := executeStages(context.Background(), cfg, []stage.Stage{ + failingStage{name: "render", err: errors.New("render failed")}, + countingStage{name: "extract", runs: &extractRuns}, + }, RunOptions{Plan: plan}) + if err == nil || !strings.Contains(err.Error(), "render failed") { + t.Fatalf("executeStages() error = %v", err) + } + if extractRuns != 0 { + t.Fatalf("extract runs = %d, want 0", extractRuns) + } +} + +type collaboratorProbeStage struct { + name string + captured **stage.Env +} + +func (s collaboratorProbeStage) Name() string { return s.name } +func (s collaboratorProbeStage) Run(_ context.Context, env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + *s.captured = env + return &stage.StageResult{}, nil +} + +type prerequisiteMutationSpy struct { + local *manifest.LocalStore + creates int + saves int +} + +func (s *prerequisiteMutationSpy) Create(ctx context.Context, sessionID string) (*manifest.Manifest, error) { + s.creates++ + return s.local.Create(ctx, sessionID) +} + +func (s *prerequisiteMutationSpy) Load(ctx context.Context, path string) (*manifest.Manifest, error) { + return s.local.Load(ctx, path) +} + +func (s *prerequisiteMutationSpy) Save(ctx context.Context, path string, m *manifest.Manifest) error { + s.saves++ + return s.local.Save(ctx, path, m) +} + +func mustBoundedPlan(t *testing.T, from, through string) BoundedPlan { + t.Helper() + plan, err := BuildBoundedPlan(from, through) + if err != nil { + t.Fatalf("BuildBoundedPlan(%q, %q) error = %v", from, through, err) + } + return plan +} + +func markPrefixSucceeded(m *manifest.Manifest, plan BoundedPlan) { + for _, name := range plan.PrefixNames() { + m.MarkStageSucceeded(name, time.Now().UTC(), nil) + } +} + +func saveBoundedManifest(t *testing.T, cfg *config.Config, m *manifest.Manifest) string { + t.Helper() + path := manifestPathFor(cfg) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatalf("create manifest directory: %v", err) + } + if err := (&manifest.LocalStore{}).Save(context.Background(), path, m); err != nil { + t.Fatalf("save manifest: %v", err) + } + return path +} diff --git a/internal/app/bounded_run_test.go b/internal/app/bounded_run_test.go index a51ea41..448cee9 100644 --- a/internal/app/bounded_run_test.go +++ b/internal/app/bounded_run_test.go @@ -8,8 +8,10 @@ import ( "reflect" "strings" "testing" + "time" "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) @@ -141,6 +143,15 @@ func TestRunPassesBoundedPlanToRunner(t *testing.T) { func TestPlanPrintsOnlyBoundedRange(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) + m.Campaign = "sample-campaign" + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim"} { + m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) + } + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") + if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPath, m); err != nil { + t.Fatalf("save prerequisite manifest: %v", err) + } var out bytes.Buffer err := Plan(context.Background(), []string{ "2026-05-03", diff --git a/internal/app/plan.go b/internal/app/plan.go index 9430de3..87d9426 100644 --- a/internal/app/plan.go +++ b/internal/app/plan.go @@ -12,7 +12,6 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/logging" - "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) // Plan validates configuration, prepares the local workdir, and prints stage order. @@ -37,6 +36,13 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { if _, err := resolveEffectiveArtifacts(cfg, request.SelectedArtifacts); err != nil { return fmt.Errorf("plan: %w", err) } + m, err := loadManifestIfPresent(ctx, cfg) + if err != nil { + return fmt.Errorf("plan: %w", err) + } + if err := validateBoundedPrerequisites(request.Plan, m); err != nil { + return fmt.Errorf("plan: %w", err) + } if _, err := loadSecretsFromConfig(cfg, logging.NewLogger(os.Stderr, slog.LevelInfo)); err != nil { return fmt.Errorf("plan: %w", err) } @@ -48,11 +54,6 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { } stages := request.Plan.Stages() - var m *manifest.Manifest - m, err = loadManifestIfPresent(ctx, cfg) - if err != nil { - return fmt.Errorf("plan: %w", err) - } decisions := decideStageActions(stages, m, request.Force) runCount := 0 diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go index c875274..4d1d04e 100644 --- a/internal/app/post_publish_cleanup_test.go +++ b/internal/app/post_publish_cleanup_test.go @@ -146,7 +146,12 @@ func TestPostPublishCleanupRetriesWhenInitialObligationSaveFails(t *testing.T) { store.fail = nil if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("retry executeStages() error = %v", err) + t.Fatalf("non-publish executeStages() error = %v", err) + } + assertExists(t, seed.spoolAudioDir) + assertCleanupPending(t, cfg) + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("publish retry executeStages() error = %v", err) } assertMissing(t, seed.spoolAudioDir) assertCleanupComplete(t, cfg) @@ -178,7 +183,12 @@ func TestPostPublishCleanupRetriesFailedDeletionWithoutTouchingOtherRuns(t *test removeRunScopedDirFn = originalRemove if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("retry executeStages() error = %v", err) + t.Fatalf("non-publish executeStages() error = %v", err) + } + assertExists(t, seed.runWorkDir) + assertCleanupPending(t, cfg) + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("publish retry executeStages() error = %v", err) } assertMissing(t, seed.runWorkDir) assertExists(t, seed.otherRunDir) @@ -218,12 +228,16 @@ func TestPostPublishCleanupRetriesWhenCompletionEvidenceSaveFails(t *testing.T) store.fail = nil if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("retry executeStages() error = %v", err) + t.Fatalf("non-publish executeStages() error = %v", err) + } + assertCleanupPending(t, cfg) + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("publish retry executeStages() error = %v", err) } assertCleanupComplete(t, cfg) if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("idempotent retry executeStages() error = %v", err) + t.Fatalf("idempotent non-publish executeStages() error = %v", err) } assertMissing(t, seed.spoolAudioDir) } @@ -328,7 +342,10 @@ func TestPostPublishCleanupFailsOnUnsafePath(t *testing.T) { t.Fatalf("Save() error = %v", err) } - _, err = executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) + if _, err = executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("non-publish executeStages() error = %v", err) + } + _, err = executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ObjectStore: &storage.FakeBackend{}}}) if err == nil || !strings.Contains(err.Error(), "refusing to delete path outside root") { t.Fatalf("executeStages() error = %v, want safe-path failure", err) } diff --git a/internal/app/runner.go b/internal/app/runner.go index 058f0de..3c261f8 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -44,6 +44,13 @@ type RunSummary struct { var executeStagesFn = executeStages func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (summary *RunSummary, resultErr error) { + var prerequisiteStore manifest.Store + if opts.Env != nil { + prerequisiteStore = opts.Env.ManifestStore + } + if err := inspectBoundedPrerequisites(ctx, cfg, opts.Plan, prerequisiteStore); err != nil { + return nil, fmt.Errorf("validate bounded run prerequisites: %w", err) + } effectiveArtifacts := opts.EffectiveArtifacts if !effectiveArtifacts.Resolved() && cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil { var err error @@ -170,7 +177,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage fmt.Errorf("load secrets from files: %w", err), ) } - if env.WhisperX == nil { + if env.WhisperX == nil && stagesContainAny(stages, "transcribe") { client, err := buildDefaultWhisperXClient(env.Config) if err != nil { return nil, persistTerminalFailure( @@ -180,7 +187,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage } env.WhisperX = client } - if env.Seriatim == nil { + if env.Seriatim == nil && stagesContainAny(stages, "merge", "normalize", "trim", "render") { runner, err := buildDefaultSeriatimRunner(env.Config) if err != nil { return nil, persistTerminalFailure( @@ -190,7 +197,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage } env.Seriatim = runner } - if env.Audita == nil { + if env.Audita == nil && stagesContainAny(stages, "polish") { runner, err := buildDefaultAuditaRunner(env.Config) if err != nil { return nil, persistTerminalFailure( @@ -203,7 +210,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage if env.Notarius == nil && needsNotariusForRun(env.Config, stages) { env.Notarius = notarius.NewSubprocessRunner() } - if env.Scriptorium == nil { + if env.Scriptorium == nil && stagesContainAny(stages, "trim", "analyze") { env.Scriptorium = scriptorium.NewSubprocessRunner() } if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages, effectiveArtifacts) { @@ -233,7 +240,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage return config.MergePublishLockRules(staticLocks, remote.Locks), nil } } - if env.Notifier == nil { + if env.Notifier == nil && stagesContainAny(stages, "notify") { env.Notifier = ¬ify.NoopSender{} } @@ -426,11 +433,13 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage // The run record lives inside the run work directory, which cleanup may // remove. Persist its completed publishing result before cleanup starts so a // successful deletion cannot be undone by a later diagnostic write. - if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil { - return nil, persistPostPublishCleanupFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("post-publish cleanup incomplete: %w", err), - ) + if containsStage(executed, "publish") { + if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil { + return nil, persistPostPublishCleanupFailure( + ctx, env.ManifestStore, manifestPath, m, + fmt.Errorf("post-publish cleanup incomplete: %w", err), + ) + } } return &RunSummary{ @@ -444,6 +453,22 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage }, nil } +func stagesContainAny(stages []stage.Stage, names ...string) bool { + wanted := make(map[string]struct{}, len(names)) + for _, name := range names { + wanted[name] = struct{}{} + } + for _, candidate := range stages { + if candidate == nil { + continue + } + if _, ok := wanted[candidate.Name()]; ok { + return true + } + } + return false +} + func persistPostPublishCleanupFailure( ctx context.Context, sessionStore manifest.Store,