Enforce bounded run prerequisites

This commit is contained in:
2026-08-29 18:17:05 +00:00
parent 3bcf2c08dd
commit 966b95b176
11 changed files with 509 additions and 32 deletions

View File

@@ -35,16 +35,23 @@ Adapters do not own:
## Default Wiring ## 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. - WhisperX HTTP client for `transcribe`.
- Seriatim subprocess runner. - Seriatim subprocess runner for `merge`, `normalize`, `trim`, or `render`.
- Audita subprocess runner. - Audita subprocess runner for `polish`.
- Scriptorium subprocess runner. - Scriptorium subprocess runner for `trim` or `analyze`.
- Notarius subprocess runner when extraction is enabled. - Notarius subprocess runner for `extract` when extraction is enabled.
- Noop notifier (`notify.NoopSender`). - Noop notifier (`notify.NoopSender`) for `notify`.
- Object store only when required by selected stages/config. - 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 Notarius is composed only when extraction is enabled; the extract stage owns
prepared reference resolution, receipt, bundle, and configured-lane policy. prepared reference resolution, receipt, bundle, and configured-lane policy.
The adapter validates the ordered selector/absolute-path pairs and is the sole The adapter validates the ordered selector/absolute-path pairs and is the sole

View File

@@ -117,12 +117,20 @@ runner marks it stale and executes it.
Session manifest is the authoritative stage-progress ledger across invocations. Session manifest is the authoritative stage-progress ledger across invocations.
Run manifest is invocation-scoped audit state. 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 After a publish commits remotely, any configured local cleanup is first recorded
as a session-manifest obligation before deletion begins. Each target becomes as a session-manifest obligation before deletion begins. Each target becomes
complete only after its confined deletion (or safe absence check) and a complete only after its confined deletion (or safe absence check) and a
successful manifest save. An incomplete obligation is retried on later successful manifest save. An incomplete obligation is retried when publish
invocations independently of their selected stages and retains the committed executes again and retains the committed run and remote identity that authorized
run and remote identity that authorized it. it; an invocation that does not execute publish does not perform cleanup.
Each invocation derives campaign, session, run, local-path, and remote-prefix Each invocation derives campaign, session, run, local-path, and remote-prefix
metadata from the validated resolved configuration as one projection. A persisted metadata from the validated resolved configuration as one projection. A persisted

View File

@@ -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 The plan command uses the same selection contract and prints only the selected
range. 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 ## Artifact Selection
`--artifacts` can be used on `run`, `session plan`, `run-stage`, `analyze`, and `--artifacts` can be used on `run`, `session plan`, `run-stage`, `analyze`, and
@@ -451,7 +469,11 @@ Rules:
- `pipeline.workspace.cleanup_after_publish=true` - `pipeline.workspace.cleanup_after_publish=true`
- Narratio first records the exact run-scoped cleanup obligation. If cleanup - Narratio first records the exact run-scoped cleanup obligation. If cleanup
reports incomplete, the remote committed snapshot remains current; rerun 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 ## Operational Caveats

View File

@@ -217,6 +217,8 @@ validation contract.
## Stage 4 — Bounded Runner Prerequisites And Composition ## Stage 4 — Bounded Runner Prerequisites And Composition
**Status: Completed**
### Goal ### Goal
Make bounded execution honor its mutation boundary while still failing safely Make bounded execution honor its mutation boundary while still failing safely

View File

@@ -117,6 +117,40 @@ Safe fix:
Relevant reference: [CLI artifact selection](./cli.md). 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 ## Notarius executable missing
Symptom: Symptom:

View File

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

View File

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

View File

@@ -8,8 +8,10 @@ import (
"reflect" "reflect"
"strings" "strings"
"testing" "testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/stage" "gitea.maximumdirect.net/eric/narratio/internal/stage"
) )
@@ -141,6 +143,15 @@ func TestRunPassesBoundedPlanToRunner(t *testing.T) {
func TestPlanPrintsOnlyBoundedRange(t *testing.T) { func TestPlanPrintsOnlyBoundedRange(t *testing.T) {
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) 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 var out bytes.Buffer
err := Plan(context.Background(), []string{ err := Plan(context.Background(), []string{
"2026-05-03", "2026-05-03",

View File

@@ -12,7 +12,6 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/logging" "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. // 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 { if _, err := resolveEffectiveArtifacts(cfg, request.SelectedArtifacts); err != nil {
return fmt.Errorf("plan: %w", err) 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 { if _, err := loadSecretsFromConfig(cfg, logging.NewLogger(os.Stderr, slog.LevelInfo)); err != nil {
return fmt.Errorf("plan: %w", err) 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() 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) decisions := decideStageActions(stages, m, request.Force)
runCount := 0 runCount := 0

View File

@@ -146,7 +146,12 @@ func TestPostPublishCleanupRetriesWhenInitialObligationSaveFails(t *testing.T) {
store.fail = nil store.fail = nil
if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != 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) assertMissing(t, seed.spoolAudioDir)
assertCleanupComplete(t, cfg) assertCleanupComplete(t, cfg)
@@ -178,7 +183,12 @@ func TestPostPublishCleanupRetriesFailedDeletionWithoutTouchingOtherRuns(t *test
removeRunScopedDirFn = originalRemove removeRunScopedDirFn = originalRemove
if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { 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) assertMissing(t, seed.runWorkDir)
assertExists(t, seed.otherRunDir) assertExists(t, seed.otherRunDir)
@@ -218,12 +228,16 @@ func TestPostPublishCleanupRetriesWhenCompletionEvidenceSaveFails(t *testing.T)
store.fail = nil store.fail = nil
if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != 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) assertCleanupComplete(t, cfg)
if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { 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) assertMissing(t, seed.spoolAudioDir)
} }
@@ -328,7 +342,10 @@ func TestPostPublishCleanupFailsOnUnsafePath(t *testing.T) {
t.Fatalf("Save() error = %v", err) 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") { if err == nil || !strings.Contains(err.Error(), "refusing to delete path outside root") {
t.Fatalf("executeStages() error = %v, want safe-path failure", err) t.Fatalf("executeStages() error = %v, want safe-path failure", err)
} }

View File

@@ -44,6 +44,13 @@ type RunSummary struct {
var executeStagesFn = executeStages var executeStagesFn = executeStages
func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (summary *RunSummary, resultErr error) { 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 effectiveArtifacts := opts.EffectiveArtifacts
if !effectiveArtifacts.Resolved() && cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil { if !effectiveArtifacts.Resolved() && cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil {
var err error 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), fmt.Errorf("load secrets from files: %w", err),
) )
} }
if env.WhisperX == nil { if env.WhisperX == nil && stagesContainAny(stages, "transcribe") {
client, err := buildDefaultWhisperXClient(env.Config) client, err := buildDefaultWhisperXClient(env.Config)
if err != nil { if err != nil {
return nil, persistTerminalFailure( return nil, persistTerminalFailure(
@@ -180,7 +187,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
} }
env.WhisperX = client env.WhisperX = client
} }
if env.Seriatim == nil { if env.Seriatim == nil && stagesContainAny(stages, "merge", "normalize", "trim", "render") {
runner, err := buildDefaultSeriatimRunner(env.Config) runner, err := buildDefaultSeriatimRunner(env.Config)
if err != nil { if err != nil {
return nil, persistTerminalFailure( return nil, persistTerminalFailure(
@@ -190,7 +197,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
} }
env.Seriatim = runner env.Seriatim = runner
} }
if env.Audita == nil { if env.Audita == nil && stagesContainAny(stages, "polish") {
runner, err := buildDefaultAuditaRunner(env.Config) runner, err := buildDefaultAuditaRunner(env.Config)
if err != nil { if err != nil {
return nil, persistTerminalFailure( 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) { if env.Notarius == nil && needsNotariusForRun(env.Config, stages) {
env.Notarius = notarius.NewSubprocessRunner() env.Notarius = notarius.NewSubprocessRunner()
} }
if env.Scriptorium == nil { if env.Scriptorium == nil && stagesContainAny(stages, "trim", "analyze") {
env.Scriptorium = scriptorium.NewSubprocessRunner() env.Scriptorium = scriptorium.NewSubprocessRunner()
} }
if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages, effectiveArtifacts) { 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 return config.MergePublishLockRules(staticLocks, remote.Locks), nil
} }
} }
if env.Notifier == nil { if env.Notifier == nil && stagesContainAny(stages, "notify") {
env.Notifier = &notify.NoopSender{} env.Notifier = &notify.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 // The run record lives inside the run work directory, which cleanup may
// remove. Persist its completed publishing result before cleanup starts so a // remove. Persist its completed publishing result before cleanup starts so a
// successful deletion cannot be undone by a later diagnostic write. // successful deletion cannot be undone by a later diagnostic write.
if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil { if containsStage(executed, "publish") {
return nil, persistPostPublishCleanupFailure( if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil {
ctx, env.ManifestStore, manifestPath, m, return nil, persistPostPublishCleanupFailure(
fmt.Errorf("post-publish cleanup incomplete: %w", err), ctx, env.ManifestStore, manifestPath, m,
) fmt.Errorf("post-publish cleanup incomplete: %w", err),
)
}
} }
return &RunSummary{ return &RunSummary{
@@ -444,6 +453,22 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
}, nil }, 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( func persistPostPublishCleanupFailure(
ctx context.Context, ctx context.Context,
sessionStore manifest.Store, sessionStore manifest.Store,