From 2e69c1489859ee9d6d65f82d7618428f82debc17 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 4 Sep 2026 01:21:37 +0000 Subject: [PATCH] Complete runner lifecycle refactor --- docs/internal/adapters.md | 14 +-- docs/internal/manifest.md | 26 ++++-- docs/internal/overview.md | 16 +++- docs/roadmap/implementation.md | 8 +- docs/roadmap/refactoring.md | 2 +- .../app/analyze_artifacts_commands_test.go | 36 +++----- internal/app/assembled_workflow_test.go | 16 ++-- internal/app/bounded_run_test.go | 8 +- internal/app/commands.go | 16 ++-- internal/app/profile_commands_test.go | 16 ++-- internal/app/regenerate_artifacts.go | 6 +- internal/app/regenerate_artifacts_test.go | 86 +++++++++---------- internal/app/remote_session_test.go | 23 ++--- internal/app/restore_workflow_test.go | 16 ++-- internal/app/run.go | 6 +- internal/app/run_stage.go | 22 +++-- internal/app/runner.go | 2 +- internal/app/session_oriented_cli_test.go | 22 ++--- 18 files changed, 176 insertions(+), 165 deletions(-) diff --git a/docs/internal/adapters.md b/docs/internal/adapters.md index d2fd007..a41486f 100644 --- 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` diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index dc9c465..d0395ce 100644 --- 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 diff --git a/docs/internal/overview.md b/docs/internal/overview.md index 3ff8a10..cc72668 100644 --- 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. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 788d997..99ce983 100644 --- 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 diff --git a/docs/roadmap/refactoring.md b/docs/roadmap/refactoring.md index 70a638b..09e3456 100644 --- 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. diff --git a/internal/app/analyze_artifacts_commands_test.go b/internal/app/analyze_artifacts_commands_test.go index ecb71de..53c2810 100644 --- 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()) diff --git a/internal/app/assembled_workflow_test.go b/internal/app/assembled_workflow_test.go index e1aad5e..89d49aa 100644 --- 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"}) { diff --git a/internal/app/bounded_run_test.go b/internal/app/bounded_run_test.go index 1bd58e3..8d7f67e 100644 --- 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) } diff --git a/internal/app/commands.go b/internal/app/commands.go index dd524fd..0743985 100644 --- 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": diff --git a/internal/app/profile_commands_test.go b/internal/app/profile_commands_test.go index fad3dfe..fb1dd5f 100644 --- 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) } diff --git a/internal/app/regenerate_artifacts.go b/internal/app/regenerate_artifacts.go index 45fa36b..b8a53b8 100644 --- 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 { diff --git a/internal/app/regenerate_artifacts_test.go b/internal/app/regenerate_artifacts_test.go index 0b5f02d..ff04959 100644 --- 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 { diff --git a/internal/app/remote_session_test.go b/internal/app/remote_session_test.go index 7bda59a..19148d3 100644 --- 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()) } diff --git a/internal/app/restore_workflow_test.go b/internal/app/restore_workflow_test.go index 6d100ae..2edb8e8 100644 --- 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()) diff --git a/internal/app/run.go b/internal/app/run.go index ee05977..9c71f69 100644 --- 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, diff --git a/internal/app/run_stage.go b/internal/app/run_stage.go index b9897d3..6b3cd77 100644 --- 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, diff --git a/internal/app/runner.go b/internal/app/runner.go index 1c25c36..1872653 100644 --- 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{}) diff --git a/internal/app/session_oriented_cli_test.go b/internal/app/session_oriented_cli_test.go index 7e8f5c0..f07c685 100644 --- 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()) }