Complete runner lifecycle refactor

2026-09-04 01:21:37 +00:00
parent dbda2a0161
commit 2e69c14898
18 changed files with 176 additions and 165 deletions
--- a/docs/internal/adapters.md
+++ b/docs/internal/adapters.md
@@ -35,8 +35,9 @@ Adapters do not own:

 ## Default Wiring

-`internal/app/runner.go` initializes default adapters when not injected and
-only when the selected execution plan needs them:
+The application composes default adapters per invocation when they were not
+injected through the stage environment, and only when the selected execution
+plan needs them:

 - WhisperX HTTP client for `transcribe`.
 - Seriatim subprocess runner for `merge`, `normalize`, `trim`, or `render`.
@@ -49,8 +50,10 @@ only when the selected execution plan needs them:
 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.
+application-owned composition. Constructor, clock, randomness, remote-lock,
+and post-publish deletion boundaries are scoped to the invocation; stages still
+receive their collaborators through the stage environment. Each selected stage
+retains its own fail-fast configuration and input validation.

 `session plan` is outside production adapter composition. It performs
 resume validation and models selected transitions against cloned manifest
@@ -88,7 +91,8 @@ configured filesystem secrets before adapter initialization.

 ## Implementation And Tests

- Composition: `internal/app/runner.go`, `internal/app/object_store.go`
+- Composition: `internal/app/runner_composition.go`,
+  `internal/app/runner_dependencies.go`, `internal/app/object_store.go`
 - Shared subprocess mechanics: `internal/adapters/subprocess`
 - Focused adapters: `internal/adapters/{whisperx,seriatim,audita,scriptorium,notarius,storage,notify}`
 - `internal/adapters/whisperx/http_test.go`
--- a/docs/internal/manifest.md
+++ b/docs/internal/manifest.md
@@ -155,11 +155,21 @@ durability guarantee. Callers reload after a returned save error before retrying

 ## Execution Semantics

-The application runner marks an executing stage running and then succeeded or
-failed in the session manifest, persisting each transition. On success it records
-outputs, logs, generated configuration references, metadata, and—when the
-stage implements the optional contract—a versioned semantic-configuration
-fingerprint. Artifact
+Planning and execution share one pure pre-execution decision over the recorded
+stage status, force selection, and normalized resume-validation result. That
+decision identifies run versus reuse and any required stale and dependent
+invalidation transition. Planning applies it only to a cloned manifest.
+Execution remains the sole owner of observed stage results and persistence.
+
+For each selected stage, the application runner computes semantic evidence,
+performs any eligible resume validation, applies the pre-execution decision,
+and persists required invalidation before either reusing the existing success
+or beginning an attempt. The same one-stage execution owner marks the attempt
+running before invoking the stage and then records and persists its failed,
+skipped, or successful terminal state before a later stage may start. On
+success it records outputs, logs, generated configuration references, metadata,
+and—when the stage implements the optional contract—a versioned
+semantic-configuration fingerprint. Artifact
 records may include optional contract and external provenance objects; old
 manifests remain compatible when those fields are absent. A successful forced
 rerun marks only succeeded transitive dependent session-stage records stale.
@@ -237,9 +247,9 @@ invalidates only stages whose semantic configuration changed. When a private
 external-tool model, module, prompt, or profile changes behind an unchanged
 configured identifier, use `--force` for the affected work.

-`session plan` computes the same current fingerprint and applies the same
-comparison and invalidation rules to a cloned manifest. It predicts the runner
-decision without persisting session state. The shared helper
+`session plan` computes the same current fingerprint and applies the shared
+decision and invalidation rules to a cloned manifest. It predicts the runner
+decision without persisting session state. The semantic fingerprint owner
 hashes deterministic JSON from stage-owned typed structs; stage providers must
 exclude secrets, complete effective-configuration dumps, and operational
 values that cannot affect canonical results. Concrete coverage is owned by the
--- a/docs/internal/overview.md
+++ b/docs/internal/overview.md
@@ -22,6 +22,14 @@ session safety controls, and runs commands. Pipeline commands execute the
 canonical stage sequence through adapter interfaces, while manifests record
 progress and artifact services resolve durable inputs and outputs.

+The pipeline runner follows one explicit invocation lifecycle: inspect bounded
+prerequisites without mutation, prepare the workspace layout, acquire one
+session lock, load or create the authoritative session manifest, compose only
+the collaborators needed by the selected range, execute each selected stage in
+order, complete any required post-publish cleanup, and then return the summary.
+The lock remains held from manifest initialization through finalization, and a
+failed stage or required manifest save prevents later stages from starting.
+
 ## Components

 | Area | Implemented owners | Responsibility |
@@ -44,9 +52,11 @@ adapters. The normative rules for these relationships remain in
 [Architecture](../policy/architecture.md).

 Pipeline execution and `session plan` share the same inclusive contiguous-range
-model. Planning clones session state and applies selected-stage transitions and
-resume validation in memory; it does not create invocation state or initialize
-stage-execution adapters. Command configuration loading can still retrieve a
+model and the same pure pre-execution decision over recorded status, force, and
+normalized resume validation. Planning applies that decision and modeled
+selected-stage transitions to cloned session state; it does not persist, create
+invocation state, initialize stage-execution adapters, or claim to observe an
+execution result. Command configuration loading can still retrieve a
 missing session file through configured remote storage. It retains the initially
 composed pipeline and selected campaign while resolving either a local or
 downloaded remote session, so one invocation cannot mix pipeline revisions.
--- a/docs/roadmap/implementation.md
+++ b/docs/roadmap/implementation.md
@@ -2,11 +2,13 @@

 ## Status And Purpose

+**Status: Completed**
+
 This is the decision-complete implementation plan for Workstream 5, **Runner
 lifecycle and pre-execution decisions**, in
 [`refactoring.md`](refactoring.md). Workstreams 1 through 4 are complete.
-Workstream 5 is the next pending workstream and is the only feature scope of
-this plan.
+Workstream 5 is complete; Workstream 6 is the next pending workstream. This
+plan's implementation scope remained limited to Workstream 5.

 The implementation must reduce the change blast and cognitive load of
 `internal/app.executePlan` without changing the pipeline's observable behavior.
@@ -408,7 +410,7 @@ are easier to read than the former monolith.

 ## Stage 6 — Remove Global Runner Seams And Close Workstream 5

-**Status: Pending**
+**Status: Completed**

 ### Goal

--- a/docs/roadmap/refactoring.md
+++ b/docs/roadmap/refactoring.md
@@ -222,7 +222,7 @@ manifest must remain synchronized with the progress ledger.

 ### Workstream 5 — Runner lifecycle and pre-execution decisions

-**Status:** Pending
+**Status:** Complete

 **Audit sources:** H04, H07, H14, H15, H17, H34, and H35.

--- a/internal/app/analyze_artifacts_commands_test.go
+++ b/internal/app/analyze_artifacts_commands_test.go
@@ -38,11 +38,7 @@ func TestExecuteRunStagePublishPropagatesSelectedArtifacts(t *testing.T) {

 	var capturedStages []string
 	var capturedArtifacts []string
-	origExecuteStagesFn := executeStagesFn
-	t.Cleanup(func() {
-		executeStagesFn = origExecuteStagesFn
-	})
-	executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
 		for _, s := range plan.Stages() {
 			capturedStages = append(capturedStages, s.Name())
 		}
@@ -52,7 +48,7 @@ func TestExecuteRunStagePublishPropagatesSelectedArtifacts(t *testing.T) {

 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	code := Execute(
+	code := executeWithPipelineExecutor(
 		[]string{
 			"run-stage", "publish", "2026-05-03",
 			"--config", pipelinePath,
@@ -62,6 +58,7 @@ func TestExecuteRunStagePublishPropagatesSelectedArtifacts(t *testing.T) {
 		},
 		&stdout,
 		&stderr,
+		executor,
 	)
 	if code != 0 {
 		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
@@ -165,11 +162,7 @@ func TestExecuteAnalyzeForceRunsAnalyze(t *testing.T) {

 	var capturedStages []string
 	var capturedForce bool
-	origExecuteStagesFn := executeStagesFn
-	t.Cleanup(func() {
-		executeStagesFn = origExecuteStagesFn
-	})
-	executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
 		for _, s := range plan.Stages() {
 			capturedStages = append(capturedStages, s.Name())
 		}
@@ -182,10 +175,11 @@ func TestExecuteAnalyzeForceRunsAnalyze(t *testing.T) {

 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	code := Execute(
+	code := executeWithPipelineExecutor(
 		[]string{"analyze", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath},
 		&stdout,
 		&stderr,
+		executor,
 	)
 	if code != 0 {
 		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
@@ -206,18 +200,14 @@ func TestExecuteAnalyzePropagatesSelectedArtifacts(t *testing.T) {
 	pipelinePath, campaignPath, sessionPath := writeValidConfigFilesWithScriptoriumArtifacts(t, workspaceRoot)

 	var capturedArtifacts []string
-	origExecuteStagesFn := executeStagesFn
-	t.Cleanup(func() {
-		executeStagesFn = origExecuteStagesFn
-	})
-	executeStagesFn = func(_ context.Context, _ *config.Config, _ BoundedPlan, opts RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, _ *config.Config, _ BoundedPlan, opts RunOptions) (*RunSummary, error) {
 		capturedArtifacts = append([]string(nil), opts.SelectedArtifacts...)
 		return &RunSummary{ManifestPath: filepath.Join(workspaceRoot, "manifest.json"), Executed: []string{"analyze"}}, nil
 	}

 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	code := Execute(
+	code := executeWithPipelineExecutor(
 		[]string{
 			"analyze",
 			"2026-05-03",
@@ -228,6 +218,7 @@ func TestExecuteAnalyzePropagatesSelectedArtifacts(t *testing.T) {
 		},
 		&stdout,
 		&stderr,
+		executor,
 	)
 	if code != 0 {
 		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
@@ -299,11 +290,7 @@ func TestExecutePublishForceRunsPublish(t *testing.T) {
 	var capturedStages []string
 	var capturedForce bool
 	var capturedArtifacts []string
-	origExecuteStagesFn := executeStagesFn
-	t.Cleanup(func() {
-		executeStagesFn = origExecuteStagesFn
-	})
-	executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
 		for _, s := range plan.Stages() {
 			capturedStages = append(capturedStages, s.Name())
 		}
@@ -317,10 +304,11 @@ func TestExecutePublishForceRunsPublish(t *testing.T) {

 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	code := Execute(
+	code := executeWithPipelineExecutor(
 		[]string{"publish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--artifacts", "session_recap"},
 		&stdout,
 		&stderr,
+		executor,
 	)
 	if code != 0 {
 		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
--- a/internal/app/assembled_workflow_test.go
+++ b/internal/app/assembled_workflow_test.go
@@ -48,9 +48,7 @@ func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.
 		force     bool
 	}
 	var captured []capturedRequest
-	original := executeStagesFn
-	t.Cleanup(func() { executeStagesFn = original })
-	executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, _ *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
 		captured = append(captured, capturedRequest{
 			stages: plan.Names(), artifacts: append([]string(nil), options.SelectedArtifacts...), force: options.Force,
 		})
@@ -60,11 +58,11 @@ func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.
 		"2026-05-03", "--force", "--from", "extract", "--through", "analyze",
 		"--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath,
 	}
-	if err := Run(context.Background(), base, &bytes.Buffer{}); err != nil {
+	if err := runWithPipelineExecutor(context.Background(), base, &bytes.Buffer{}, executor); err != nil {
 		t.Fatalf("canonical unselected Run() error = %v", err)
 	}
 	selected := append(append([]string(nil), base...), "--artifacts", "session_recap")
-	if err := Run(context.Background(), selected, &bytes.Buffer{}); err != nil {
+	if err := runWithPipelineExecutor(context.Background(), selected, &bytes.Buffer{}, executor); err != nil {
 		t.Fatalf("canonical selected Run() error = %v", err)
 	}
 	var stdout, stderr bytes.Buffer
@@ -72,7 +70,7 @@ func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.
 		"regenerate-artifacts", "2026-05-03", "--artifacts", "session_recap",
 		"--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath,
 	}
-	if code := Execute(alias, &stdout, &stderr); code != 0 {
+	if code := executeWithPipelineExecutor(alias, &stdout, &stderr, executor); code != 0 {
 		t.Fatalf("alias exit=%d stderr=%q", code, stderr.String())
 	}
 	if len(captured) != 3 {
@@ -289,12 +287,10 @@ func TestAssembledSplitBundleCommandsUseOneEffectiveFamilyConfiguration(t *testi
 		t.Fatalf("plan output = %q, want testing provenance", planned.String())
 	}

-	original := executeStagesFn
-	t.Cleanup(func() { executeStagesFn = original })
 	var capturedCfg *config.Config
 	var capturedPlan BoundedPlan
 	var capturedOptions RunOptions
-	executeStagesFn = func(_ context.Context, cfg *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, cfg *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
 		capturedCfg = cfg
 		capturedPlan = plan
 		capturedOptions = options
@@ -302,7 +298,7 @@ func TestAssembledSplitBundleCommandsUseOneEffectiveFamilyConfiguration(t *testi
 	}
 	runArgs := append(append([]string(nil), common...), "--from", "analyze", "--through", "analyze", "--artifacts", "character_items_arannis")
 	var runOut bytes.Buffer
-	if err := Run(context.Background(), runArgs, &runOut); err != nil {
+	if err := runWithPipelineExecutor(context.Background(), runArgs, &runOut, executor); err != nil {
 		t.Fatalf("bounded family run: %v", err)
 	}
 	if !reflect.DeepEqual(capturedPlan.Names(), []string{"analyze"}) {
--- a/internal/app/bounded_run_test.go
+++ b/internal/app/bounded_run_test.go
@@ -140,9 +140,7 @@ func TestRunPassesBoundedPlanToRunner(t *testing.T) {
 	var capturedStages []string
 	var capturedPlan BoundedPlan
 	var capturedOptions RunOptions
-	original := executeStagesFn
-	t.Cleanup(func() { executeStagesFn = original })
-	executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, _ *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
 		capturedStages = plan.Names()
 		capturedPlan = plan
 		capturedOptions = options
@@ -150,7 +148,7 @@ func TestRunPassesBoundedPlanToRunner(t *testing.T) {
 	}

 	var out bytes.Buffer
-	err := Run(context.Background(), []string{
+	err := runWithPipelineExecutor(context.Background(), []string{
 		"2026-05-03",
 		"--from", "render",
 		"--through", "extract",
@@ -158,7 +156,7 @@ func TestRunPassesBoundedPlanToRunner(t *testing.T) {
 		"--config", pipelinePath,
 		"--campaign-file", campaignPath,
 		"--session", sessionPath,
-	}, &out)
+	}, &out, executor)
 	if err != nil {
 		t.Fatalf("Run() error = %v", err)
 	}
--- a/internal/app/commands.go
+++ b/internal/app/commands.go
@@ -9,10 +9,12 @@ import (

 var supportedCommands = []string{"version", "run", "regenerate-artifacts", "run-stage", "analyze", "publish", "clean", "session", "config"}

-var runCommandFn = Run
-
 // Execute dispatches CLI commands and returns a process exit code.
 func Execute(args []string, stdout, stderr io.Writer) int {
+	return executeWithPipelineExecutor(args, stdout, stderr, executePlan)
+}
+
+func executeWithPipelineExecutor(args []string, stdout, stderr io.Writer, executor pipelineExecutor) int {
 	if len(args) == 0 {
 		printUsage(stderr)
 		return 1
@@ -27,15 +29,15 @@ func Execute(args []string, stdout, stderr io.Writer) int {
 	case "version":
 		err = Version(cmdArgs, stdout)
 	case "run":
-		err = runCommandFn(ctx, cmdArgs, stdout)
+		err = runWithPipelineExecutor(ctx, cmdArgs, stdout, executor)
 	case "regenerate-artifacts":
-		err = RegenerateArtifacts(ctx, cmdArgs, stdout)
+		err = regenerateArtifactsWithPipelineExecutor(ctx, cmdArgs, stdout, executor)
 	case "run-stage":
-		err = RunStage(ctx, cmdArgs, stdout)
+		err = runStageWithPipelineExecutor(ctx, cmdArgs, stdout, executor)
 	case "analyze":
-		err = Analyze(ctx, cmdArgs, stdout)
+		err = analyzeWithPipelineExecutor(ctx, cmdArgs, stdout, executor)
 	case "publish":
-		err = Publish(ctx, cmdArgs, stdout)
+		err = publishWithPipelineExecutor(ctx, cmdArgs, stdout, executor)
 	case "session":
 		err = Session(ctx, cmdArgs, stdout)
 	case "clean":
--- a/internal/app/profile_commands_test.go
+++ b/internal/app/profile_commands_test.go
@@ -16,26 +16,24 @@ func TestRunProfileSelectionFlowsThroughSharedLoader(t *testing.T) {
 	pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
 	enableCommandTestProfiles(t, pipelinePath)

-	original := executeStagesFn
-	t.Cleanup(func() { executeStagesFn = original })
 	var language string
-	executeStagesFn = func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
 		language = cfg.Pipeline.WhisperX.Language
 		return &RunSummary{SessionID: cfg.Session.SessionID, ManifestPath: "manifest.json"}, nil
 	}

-	if err := Run(context.Background(), []string{
+	if err := runWithPipelineExecutor(context.Background(), []string{
 		"2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--profile", "testing",
-	}, io.Discard); err != nil {
+	}, io.Discard, executor); err != nil {
 		t.Fatal(err)
 	}
 	if language != "fr" {
 		t.Fatalf("explicit profile language = %q, want fr", language)
 	}

-	if err := Run(context.Background(), []string{
+	if err := runWithPipelineExecutor(context.Background(), []string{
 		"2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath,
-	}, io.Discard); err != nil {
+	}, io.Discard, executor); err != nil {
 		t.Fatal(err)
 	}
 	if language != "en" {
@@ -43,9 +41,9 @@ func TestRunProfileSelectionFlowsThroughSharedLoader(t *testing.T) {
 	}

 	for _, profile := range []string{"unknown", ""} {
-		err := Run(context.Background(), []string{
+		err := runWithPipelineExecutor(context.Background(), []string{
 			"2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--profile", profile,
-		}, io.Discard)
+		}, io.Discard, executor)
 		if err == nil || !strings.Contains(err.Error(), "profile") {
 			t.Fatalf("profile %q error = %v, want selection rejection", profile, err)
 		}
--- a/internal/app/regenerate_artifacts.go
+++ b/internal/app/regenerate_artifacts.go
@@ -9,6 +9,10 @@ import (
 // RegenerateArtifacts expands the convenience command into its canonical run
 // invocation. The run command remains the sole owner of parsing and execution.
 func RegenerateArtifacts(ctx context.Context, args []string, out io.Writer) error {
+	return regenerateArtifactsWithPipelineExecutor(ctx, args, out, executePlan)
+}
+
+func regenerateArtifactsWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
 	if containsHelpOption(args) {
 		printRegenerateArtifactsHelp(out)
 		return nil
@@ -21,7 +25,7 @@ func RegenerateArtifacts(ctx context.Context, args []string, out io.Writer) erro
 	}
 	expanded = append(expanded, "--force", "--from", "extract", "--through", "analyze")
 	expanded = append(expanded, args...)
-	return runCommandFn(ctx, expanded, out)
+	return runWithPipelineExecutor(ctx, expanded, out, executor)
 }

 func containsHelpOption(args []string) bool {
--- a/internal/app/regenerate_artifacts_test.go
+++ b/internal/app/regenerate_artifacts_test.go
@@ -7,82 +7,80 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
 )

-func TestRegenerateArtifactsForwardsExactCanonicalRunArguments(t *testing.T) {
-	original := runCommandFn
-	t.Cleanup(func() { runCommandFn = original })
-	var captured []string
-	runCommandFn = func(_ context.Context, args []string, _ io.Writer) error {
-		captured = append([]string(nil), args...)
-		return nil
+func TestRegenerateArtifactsForwardsCanonicalPipelineRequest(t *testing.T) {
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, sessionPath := writeValidConfigFilesWithScriptoriumArtifacts(t, workspaceRoot)
+	enableCommandTestProfiles(t, pipelinePath)
+	var capturedPlan BoundedPlan
+	var capturedOptions RunOptions
+	executor := func(_ context.Context, cfg *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
+		capturedPlan = plan
+		capturedOptions = options
+		return &RunSummary{SessionID: cfg.Session.SessionID, ManifestPath: "manifest.json"}, nil
 	}

-	code := Execute([]string{
+	code := executeWithPipelineExecutor([]string{
 		"regenerate-artifacts", "2026-05-03",
 		"--artifacts", "session_recap,player_handout",
 		"--artifacts=player_handout",
-		"--config", "pipeline.yml",
-		"--campaign", "sample-campaign",
+		"--config", pipelinePath,
+		"--campaign-file", campaignPath,
+		"--session", sessionPath,
 		"--profile", "testing",
-	}, io.Discard, io.Discard)
+	}, io.Discard, io.Discard, executor)
 	if code != 0 {
-		t.Fatalf("Execute() code = %d, want 0", code)
+		t.Fatalf("executeWithPipelineExecutor() code = %d, want 0", code)
 	}
-	want := []string{
-		"2026-05-03", "--force", "--from", "extract", "--through", "analyze",
-		"--artifacts", "session_recap,player_handout",
-		"--artifacts=player_handout",
-		"--config", "pipeline.yml",
-		"--campaign", "sample-campaign",
-		"--profile", "testing",
+	if got, want := capturedPlan.Names(), []string{"extract", "analyze"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("plan = %#v, want %#v", got, want)
 	}
-	if !reflect.DeepEqual(captured, want) {
-		t.Fatalf("forwarded args = %#v, want %#v", captured, want)
+	if !capturedOptions.Force {
+		t.Fatal("force = false, want true")
+	}
+	if got, want := capturedOptions.SelectedArtifacts, []string{"player_handout", "session_recap"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("selected artifacts = %#v, want %#v", got, want)
 	}
 }

 func TestRegenerateArtifactsForwardsSessionIDCompatibilityFlag(t *testing.T) {
-	original := runCommandFn
-	t.Cleanup(func() { runCommandFn = original })
-	var captured []string
-	runCommandFn = func(_ context.Context, args []string, _ io.Writer) error {
-		captured = append([]string(nil), args...)
-		return nil
+	workspaceRoot := t.TempDir()
+	pipelinePath, campaignPath, sessionPath := writeValidConfigFilesWithScriptoriumArtifacts(t, workspaceRoot)
+	var capturedSessionID string
+	executor := func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
+		capturedSessionID = cfg.Session.SessionID
+		return &RunSummary{SessionID: cfg.Session.SessionID, ManifestPath: "manifest.json"}, nil
 	}

-	code := Execute([]string{
+	code := executeWithPipelineExecutor([]string{
 		"regenerate-artifacts",
-		"--config", "pipeline.yml",
+		"--config", pipelinePath,
+		"--campaign-file", campaignPath,
+		"--session", sessionPath,
 		"--session-id", "2026-05-03",
 		"--artifacts", "session_recap",
-	}, io.Discard, io.Discard)
+	}, io.Discard, io.Discard, executor)
 	if code != 0 {
-		t.Fatalf("Execute() code = %d, want 0", code)
+		t.Fatalf("executeWithPipelineExecutor() code = %d, want 0", code)
 	}
-	want := []string{
-		"--force", "--from", "extract", "--through", "analyze",
-		"--config", "pipeline.yml",
-		"--session-id", "2026-05-03",
-		"--artifacts", "session_recap",
-	}
-	if !reflect.DeepEqual(captured, want) {
-		t.Fatalf("forwarded args = %#v, want %#v", captured, want)
+	if capturedSessionID != "2026-05-03" {
+		t.Fatalf("session id = %q, want compatibility flag value", capturedSessionID)
 	}
 }

 func TestRegenerateArtifactsHelpDoesNotInvokeRun(t *testing.T) {
-	original := runCommandFn
-	t.Cleanup(func() { runCommandFn = original })
 	called := false
-	runCommandFn = func(_ context.Context, _ []string, _ io.Writer) error {
+	executor := func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error) {
 		called = true
-		return nil
+		return nil, nil
 	}

 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	if code := Execute([]string{"regenerate-artifacts", "--help"}, &stdout, &stderr); code != 0 {
+	if code := executeWithPipelineExecutor([]string{"regenerate-artifacts", "--help"}, &stdout, &stderr, executor); code != 0 {
 		t.Fatalf("Execute() code = %d, stderr = %q", code, stderr.String())
 	}
 	if called {
--- a/internal/app/remote_session_test.go
+++ b/internal/app/remote_session_test.go
@@ -92,7 +92,7 @@ func TestRemoteSessionConfigIsRemovedAfterEveryCommandExit(t *testing.T) {
 		name           string
 		sessionYAML    string
 		command        []string
-		configureRun   func()
+		executor       pipelineExecutor
 		wantSuccessful bool
 	}{
 		{
@@ -126,10 +126,8 @@ inputs:
    prefix: audio/
 `,
 			command: []string{"run", "2026-05-03"},
-			configureRun: func() {
-				executeStagesFn = func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error) {
-					return nil, errors.New("adapter failed")
-				}
+			executor: func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error) {
+				return nil, errors.New("adapter failed")
 			},
 		},
 		{
@@ -140,10 +138,8 @@ inputs:
    prefix: audio/
 `,
 			command: []string{"run", "2026-05-03"},
-			configureRun: func() {
-				executeStagesFn = func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error) {
-					return nil, context.Canceled
-				}
+			executor: func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error) {
+				return nil, context.Canceled
 			},
 		},
 	}
@@ -159,16 +155,15 @@ inputs:

 			var downloadedPath string
 			captureRemoteSessionTempPath(t, &downloadedPath)
-			if tt.configureRun != nil {
-				origExecuteStagesFn := executeStagesFn
-				t.Cleanup(func() { executeStagesFn = origExecuteStagesFn })
-				tt.configureRun()
+			executor := tt.executor
+			if executor == nil {
+				executor = executePlan
 			}

 			args := append(append([]string(nil), tt.command...), "--config", pipelinePath, "--campaign-file", campaignPath)
 			var stdout bytes.Buffer
 			var stderr bytes.Buffer
-			code := Execute(args, &stdout, &stderr)
+			code := executeWithPipelineExecutor(args, &stdout, &stderr, executor)
 			if tt.wantSuccessful && code != 0 {
 				t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
 			}
--- a/internal/app/restore_workflow_test.go
+++ b/internal/app/restore_workflow_test.go
@@ -29,11 +29,7 @@ func TestRestoreThenRunStageForceAnalyzeUsesRestoredDurableState(t *testing.T) {

 	restoreWithStoreAndRealPhases(t, fake)

-	origExecuteStagesFn := executeStagesFn
-	t.Cleanup(func() {
-		executeStagesFn = origExecuteStagesFn
-	})
-	executeStagesFn = func(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
+	executor := func(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
 		if opts.Env == nil {
 			opts.Env = &Env{}
 		}
@@ -81,7 +77,7 @@ func TestRestoreThenRunStageForceAnalyzeUsesRestoredDurableState(t *testing.T) {

 	stdout.Reset()
 	stderr.Reset()
-	runStageCode := Execute(
+	runStageCode := executeWithPipelineExecutor(
 		[]string{
 			"run-stage", "analyze", cfg.Session.SessionID,
 			"--config", pipelinePath,
@@ -92,6 +88,7 @@ func TestRestoreThenRunStageForceAnalyzeUsesRestoredDurableState(t *testing.T) {
 		},
 		&stdout,
 		&stderr,
+		executor,
 	)
 	if runStageCode != 0 {
 		t.Fatalf("run-stage exit code = %d, want 0; stderr=%q", runStageCode, stderr.String())
@@ -209,14 +206,12 @@ previous_session_id: 2026-04-26
 	mustReadEquals(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n")

 	scriptoriumFake := &scriptorium.FakeRunner{}
-	origExecuteStagesFn := executeStagesFn
 	origObjectStoreFn := newObjectStoreFromConfigFn
 	objectStoreConstructed := false
 	t.Cleanup(func() {
-		executeStagesFn = origExecuteStagesFn
 		newObjectStoreFromConfigFn = origObjectStoreFn
 	})
-	executeStagesFn = func(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
+	executor := func(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
 		if opts.Env == nil {
 			opts.Env = &Env{}
 		}
@@ -230,7 +225,7 @@ previous_session_id: 2026-04-26

 	stdout.Reset()
 	stderr.Reset()
-	runStageCode := Execute(
+	runStageCode := executeWithPipelineExecutor(
 		[]string{
 			"run-stage", "analyze", cfg.Session.SessionID,
 			"--config", pipelinePath,
@@ -241,6 +236,7 @@ previous_session_id: 2026-04-26
 		},
 		&stdout,
 		&stderr,
+		executor,
 	)
 	if runStageCode != 0 {
 		t.Fatalf("run-stage exit code = %d, want 0; stderr=%q", runStageCode, stderr.String())
--- a/internal/app/run.go
+++ b/internal/app/run.go
@@ -12,6 +12,10 @@ import (

 // Run executes the pipeline plan and persists manifest state.
 func Run(ctx context.Context, args []string, out io.Writer) error {
+	return runWithPipelineExecutor(ctx, args, out, executePlan)
+}
+
+func runWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
 	request, err := parseBoundedRunRequest("run", args, out)
 	if err != nil {
 		if errors.Is(err, flag.ErrHelp) {
@@ -37,7 +41,7 @@ func Run(ctx context.Context, args []string, out io.Writer) error {
 	if err != nil {
 		return fmt.Errorf("run: %w", err)
 	}
-	summary, err := executeStagesFn(ctx, cfg, request.Plan, RunOptions{
+	summary, err := executor(ctx, cfg, request.Plan, RunOptions{
 		Force:              request.Force,
 		SelectedArtifacts:  selectedArtifacts,
 		EffectiveArtifacts: effectiveArtifacts,
--- a/internal/app/run_stage.go
+++ b/internal/app/run_stage.go
@@ -12,6 +12,10 @@ import (

 // RunStage executes exactly one selected stage.
 func RunStage(ctx context.Context, args []string, out io.Writer) error {
+	return runStageWithPipelineExecutor(ctx, args, out, executePlan)
+}
+
+func runStageWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
 	var stageName string
 	var positionalSessionID string
 	if len(args) >= 2 && !isCLIFlagToken(args[0]) && !isCLIFlagToken(args[1]) {
@@ -74,7 +78,7 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error {
 		PreviousSessionID: flags.previousSessionID,
 		Force:             force,
 		SelectedArtifacts: normalizedArtifacts,
-	})
+	}, executor)
 	if err != nil {
 		return err
 	}
@@ -93,6 +97,10 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error {

 // Analyze force-runs the analyze stage.
 func Analyze(ctx context.Context, args []string, out io.Writer) error {
+	return analyzeWithPipelineExecutor(ctx, args, out, executePlan)
+}
+
+func analyzeWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
 	fs := flag.NewFlagSet("analyze", flag.ContinueOnError)
 	fs.SetOutput(io.Discard)

@@ -123,7 +131,7 @@ func Analyze(ctx context.Context, args []string, out io.Writer) error {
 		PreviousSessionID: flags.previousSessionID,
 		Force:             true,
 		SelectedArtifacts: normalizedArtifacts,
-	})
+	}, executor)
 	if err != nil {
 		return err
 	}
@@ -140,6 +148,10 @@ func Analyze(ctx context.Context, args []string, out io.Writer) error {

 // Publish force-runs the publish stage.
 func Publish(ctx context.Context, args []string, out io.Writer) error {
+	return publishWithPipelineExecutor(ctx, args, out, executePlan)
+}
+
+func publishWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
 	fs := flag.NewFlagSet("publish", flag.ContinueOnError)
 	fs.SetOutput(io.Discard)

@@ -170,7 +182,7 @@ func Publish(ctx context.Context, args []string, out io.Writer) error {
 		PreviousSessionID: flags.previousSessionID,
 		Force:             true,
 		SelectedArtifacts: normalizedArtifacts,
-	})
+	}, executor)
 	if err != nil {
 		return err
 	}
@@ -198,7 +210,7 @@ type singleStageCommand struct {
 	SelectedArtifacts []string
 }

-func runSingleStageCommand(ctx context.Context, req singleStageCommand) (*RunSummary, error) {
+func runSingleStageCommand(ctx context.Context, req singleStageCommand, executor pipelineExecutor) (*RunSummary, error) {
 	plan, err := buildSingleStageExecutionPlan(req.StageName)
 	if err != nil {
 		return nil, fmt.Errorf("%s: %w", req.CommandName, err)
@@ -224,7 +236,7 @@ func runSingleStageCommand(ctx context.Context, req singleStageCommand) (*RunSum
 	if err != nil {
 		return nil, fmt.Errorf("%s: %w", req.CommandName, err)
 	}
-	summary, err := executeStagesFn(ctx, cfg, plan, RunOptions{
+	summary, err := executor(ctx, cfg, plan, RunOptions{
 		Force:              req.Force,
 		SelectedArtifacts:  selectedArtifacts,
 		EffectiveArtifacts: effectiveArtifacts,
--- a/internal/app/runner.go
+++ b/internal/app/runner.go
@@ -34,7 +34,7 @@ type RunSummary struct {
 	Skipped      []string
 }

-var executeStagesFn = executePlan
+type pipelineExecutor func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error)

 func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (summary *RunSummary, resultErr error) {
 	return executePlanWithDependencies(ctx, cfg, plan, opts, invocationDependencies{})
--- a/internal/app/session_oriented_cli_test.go
+++ b/internal/app/session_oriented_cli_test.go
@@ -18,9 +18,7 @@ func TestExecuteRunAcceptsPositionalSessionID(t *testing.T) {
 	pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)

 	var capturedSessionID string
-	origExecuteStagesFn := executeStagesFn
-	t.Cleanup(func() { executeStagesFn = origExecuteStagesFn })
-	executeStagesFn = func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
 		capturedSessionID = cfg.Session.SessionID
 		return &RunSummary{
 			SessionID:    cfg.Session.SessionID,
@@ -31,13 +29,13 @@ func TestExecuteRunAcceptsPositionalSessionID(t *testing.T) {

 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	code := Execute([]string{
+	code := executeWithPipelineExecutor([]string{
 		"run",
 		"2026-05-03",
 		"--config", pipelinePath,
 		"--campaign-file", campaignPath,
 		"--session", sessionPath,
-	}, &stdout, &stderr)
+	}, &stdout, &stderr, executor)
 	if code != 0 {
 		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
 	}
@@ -114,9 +112,7 @@ inputs:
  audio_s3:
    prefix: audio/
 `)
-	origExecuteStagesFn := executeStagesFn
-	t.Cleanup(func() { executeStagesFn = origExecuteStagesFn })
-	executeStagesFn = func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
+	executor := func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
 		return &RunSummary{
 			SessionID:    cfg.Session.SessionID,
 			ManifestPath: filepath.Join(workspaceRoot, "manifest.json"),
@@ -126,12 +122,12 @@ inputs:

 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
-	code := Execute([]string{
+	code := executeWithPipelineExecutor([]string{
 		"run",
 		"2026-06-07",
 		"--config", pipelinePath,
 		"--campaign-file", campaignPath,
-	}, &stdout, &stderr)
+	}, &stdout, &stderr, executor)
 	if code != 0 {
 		t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
 	}
@@ -191,9 +187,7 @@ func TestExecuteWorkflowCommandsAcceptPositionalSessionID(t *testing.T) {
 			var capturedStages []string
 			var capturedForce bool
 			var capturedArtifacts []string
-			origExecuteStagesFn := executeStagesFn
-			t.Cleanup(func() { executeStagesFn = origExecuteStagesFn })
-			executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
+			executor := func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
 				for _, s := range plan.Stages() {
 					capturedStages = append(capturedStages, s.Name())
 				}
@@ -208,7 +202,7 @@ func TestExecuteWorkflowCommandsAcceptPositionalSessionID(t *testing.T) {

 			var stdout bytes.Buffer
 			var stderr bytes.Buffer
-			code := Execute(tt.args, &stdout, &stderr)
+			code := executeWithPipelineExecutor(tt.args, &stdout, &stderr, executor)
 			if code != 0 {
 				t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
 			}