eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 17 Wiki Activity

Use one execution plan throughout the runner

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-29 20:49:43 +00:00
parent 0dc8ff9b52
commit fd5ccc668b
12 changed files with 83 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
internal/app/analyze_artifacts_commands_test.go
View File

@@ -11,7 +11,6 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
func TestExecuteRunStageArtifactsUnsupportedStageFails(t *testing.T) {
@@ -43,8 +42,8 @@ func TestExecuteRunStagePublishPropagatesSelectedArtifacts(t *testing.T) {
t.Cleanup(func() {
executeStagesFn = origExecuteStagesFn
})
executeStagesFn = func(_ context.Context, _ *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
for _, s := range stages {
executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
for _, s := range plan.Stages() {
capturedStages = append(capturedStages, s.Name())
}
capturedArtifacts = append([]string(nil), opts.SelectedArtifacts...)
@@ -159,8 +158,8 @@ func TestExecuteAnalyzeForceRunsAnalyze(t *testing.T) {
t.Cleanup(func() {
executeStagesFn = origExecuteStagesFn
})
executeStagesFn = func(_ context.Context, _ *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
for _, s := range stages {
executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
for _, s := range plan.Stages() {
capturedStages = append(capturedStages, s.Name())
}
capturedForce = opts.Force
@@ -200,7 +199,7 @@ func TestExecuteAnalyzePropagatesSelectedArtifacts(t *testing.T) {
t.Cleanup(func() {
executeStagesFn = origExecuteStagesFn
})
executeStagesFn = func(_ context.Context, _ *config.Config, _ []stage.Stage, opts RunOptions) (*RunSummary, error) {
executeStagesFn = 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
}
@@ -293,8 +292,8 @@ func TestExecutePublishForceRunsPublish(t *testing.T) {
t.Cleanup(func() {
executeStagesFn = origExecuteStagesFn
})
executeStagesFn = func(_ context.Context, _ *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
for _, s := range stages {
executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
for _, s := range plan.Stages() {
capturedStages = append(capturedStages, s.Name())
}
capturedForce = opts.Force

14
internal/app/assembled_workflow_test.go
View File

@@ -54,9 +54,9 @@ func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.
var captured []capturedRequest
original := executeStagesFn
t.Cleanup(func() { executeStagesFn = original })
executeStagesFn = func(_ context.Context, _ *config.Config, stages []stage.Stage, options RunOptions) (*RunSummary, error) {
executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
captured = append(captured, capturedRequest{
stages: stageNames(stages), artifacts: append([]string(nil), options.SelectedArtifacts...), force: options.Force,
stages: plan.Names(), artifacts: append([]string(nil), options.SelectedArtifacts...), force: options.Force,
})
return &RunSummary{SessionID: "2026-05-03", ManifestPath: manifestPathForConfig(workspaceRoot)}, nil
}
@@ -99,9 +99,8 @@ func TestAssembledForcedSiblingIndependenceAndFailureBoundary(t *testing.T) {
seedAllStagesSucceeded(t, cfg)
plan := mustBoundedPlan(t, selected, selected)
runs := 0
summary, err := executeStages(context.Background(), cfg, []stage.Stage{
countingStage{name: selected, runs: &runs},
}, RunOptions{Plan: plan, Force: true})
plan.stages = []stage.Stage{countingStage{name: selected, runs: &runs}}
summary, err := executePlan(context.Background(), cfg, plan, RunOptions{Force: true})
if err != nil {
t.Fatal(err)
}
@@ -132,9 +131,8 @@ func TestAssembledForcedSiblingIndependenceAndFailureBoundary(t *testing.T) {
cfg := testConfig(t)
seedAllStagesSucceeded(t, cfg)
plan := mustBoundedPlan(t, "render", "render")
_, err := executeStages(context.Background(), cfg, []stage.Stage{
failingStage{name: "render", err: context.Canceled},
}, RunOptions{Plan: plan, Force: true})
plan.stages = []stage.Stage{failingStage{name: "render", err: context.Canceled}}
_, err := executePlan(context.Background(), cfg, plan, RunOptions{Force: true})
if err == nil {
t.Fatal("forced render failure returned nil")
}

23
internal/app/bounded_prerequisites_test.go
View File

@@ -94,9 +94,9 @@ func TestExecuteStagesRejectsBoundedPrerequisitesBeforePersistentMutation(t *tes
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},
plan.stages = []stage.Stage{countingStage{name: "render", runs: &runs}}
_, err = executePlan(context.Background(), cfg, plan, RunOptions{
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)
@@ -126,9 +126,9 @@ func TestExecuteStagesRechecksBoundedPrerequisitesUnderSessionLock(t *testing.T)
store := &prerequisiteChangingStore{local: &manifest.LocalStore{}}
runs := 0
_, err := executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "render", runs: &runs}}, RunOptions{
Plan: plan,
Env: &Env{ManifestStore: store},
plan.stages = []stage.Stage{countingStage{name: "render", runs: &runs}}
_, err := executePlan(context.Background(), cfg, plan, RunOptions{
Env: &Env{ManifestStore: store},
})
if err == nil || !strings.Contains(err.Error(), `stage "prepare" has unusable status "running"`) {
t.Fatalf("executeStages() error = %v, want changed prerequisite rejection", err)
@@ -191,7 +191,8 @@ func TestExecuteStagesBoundedCompositionUsesOnlySelectedCollaborators(t *testing
saveBoundedManifest(t, cfg, m)
var captured *stage.Env
_, err := executeStages(context.Background(), cfg, []stage.Stage{collaboratorProbeStage{name: test.stageName, captured: &captured}}, RunOptions{Plan: plan})
plan.stages = []stage.Stage{collaboratorProbeStage{name: test.stageName, captured: &captured}}
_, err := executePlan(context.Background(), cfg, plan, RunOptions{})
if err != nil {
t.Fatalf("executeStages() error = %v", err)
}
@@ -213,7 +214,8 @@ func TestExecuteStagesBoundedForceStalesButDoesNotRunDependentsOutsideRange(t *t
}
saveBoundedManifest(t, cfg, m)
runs := 0
_, err := executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "render", runs: &runs}}, RunOptions{Plan: plan, Force: true})
plan.stages = []stage.Stage{countingStage{name: "render", runs: &runs}}
_, err := executePlan(context.Background(), cfg, plan, RunOptions{Force: true})
if err != nil {
t.Fatalf("executeStages() error = %v", err)
}
@@ -242,10 +244,11 @@ func TestExecuteStagesBoundedFailureStopsWithinSelectedRange(t *testing.T) {
markPrefixSucceeded(m, plan)
saveBoundedManifest(t, cfg, m)
extractRuns := 0
_, err := executeStages(context.Background(), cfg, []stage.Stage{
plan.stages = []stage.Stage{
failingStage{name: "render", err: errors.New("render failed")},
countingStage{name: "extract", runs: &extractRuns},
}, RunOptions{Plan: plan})
}
_, err := executePlan(context.Background(), cfg, plan, RunOptions{})
if err == nil || !strings.Contains(err.Error(), "render failed") {
t.Fatalf("executeStages() error = %v", err)
}

9
internal/app/bounded_run_test.go
View File

@@ -12,7 +12,6 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
func TestBoundedRunParsingIsSharedByRunAndPlan(t *testing.T) {
@@ -112,11 +111,13 @@ func TestRunPassesBoundedPlanToRunner(t *testing.T) {
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
var capturedStages []string
var capturedPlan BoundedPlan
var capturedOptions RunOptions
original := executeStagesFn
t.Cleanup(func() { executeStagesFn = original })
executeStagesFn = func(_ context.Context, _ *config.Config, stages []stage.Stage, options RunOptions) (*RunSummary, error) {
capturedStages = stageNames(stages)
executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
capturedStages = plan.Names()
capturedPlan = plan
capturedOptions = options
return &RunSummary{SessionID: "2026-05-03", ManifestPath: filepath.Join(workspaceRoot, "manifest.json")}, nil
}
@@ -135,7 +136,7 @@ func TestRunPassesBoundedPlanToRunner(t *testing.T) {
t.Fatalf("Run() error = %v", err)
}
want := []string{"render", "extract"}
if !reflect.DeepEqual(capturedStages, want) || !reflect.DeepEqual(capturedOptions.Plan.Names(), want) || !capturedOptions.Force {
if !reflect.DeepEqual(capturedStages, want) || !reflect.DeepEqual(capturedPlan.Names(), want) || !capturedOptions.Force {
t.Fatalf("stages = %#v options = %#v", capturedStages, capturedOptions)
}
}

19
internal/app/planner.go
View File

@@ -140,3 +140,22 @@ func BuildSingleStagePlan(name string) ([]stage.Stage, error) {
}
return []stage.Stage{s}, nil
}
// buildSingleStageExecutionPlan selects one canonical stage without applying
// bounded-run prefix prerequisites. The run-stage family validates the stage's
// concrete inputs and intentionally retains its established direct-execution
// semantics.
func buildSingleStageExecutionPlan(name string) (BoundedPlan, error) {
stages, err := BuildSingleStagePlan(name)
if err != nil {
return BoundedPlan{}, err
}
plan, err := BuildBoundedPlan(name, name)
if err != nil {
return BoundedPlan{}, fmt.Errorf("build stage plan: %w", err)
}
plan.stages = stages
plan.explicitFrom = false
plan.explicitThrough = false
return plan, nil
}

5
internal/app/remote_session_test.go
View File

@@ -13,7 +13,6 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
func TestExecuteRemoteSessionFallbackLoadsFromObjectStore(t *testing.T) {
@@ -85,7 +84,7 @@ inputs:
`,
command: []string{"run", "2026-05-03"},
configureRun: func() {
executeStagesFn = func(context.Context, *config.Config, []stage.Stage, RunOptions) (*RunSummary, error) {
executeStagesFn = func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error) {
return nil, errors.New("adapter failed")
}
},
@@ -99,7 +98,7 @@ inputs:
`,
command: []string{"run", "2026-05-03"},
configureRun: func() {
executeStagesFn = func(context.Context, *config.Config, []stage.Stage, RunOptions) (*RunSummary, error) {
executeStagesFn = func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error) {
return nil, context.Canceled
}
},

9
internal/app/restore_workflow_test.go
View File

@@ -14,7 +14,6 @@ import (
"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 TestRestoreThenRunStageForceAnalyzeUsesRestoredDurableState(t *testing.T) {
@@ -34,12 +33,12 @@ func TestRestoreThenRunStageForceAnalyzeUsesRestoredDurableState(t *testing.T) {
t.Cleanup(func() {
executeStagesFn = origExecuteStagesFn
})
executeStagesFn = func(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
executeStagesFn = func(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
if opts.Env == nil {
opts.Env = &Env{}
}
opts.Env.Scriptorium = &scriptorium.NoopRunner{}
return executeStages(ctx, cfg, stages, opts)
return executePlan(ctx, cfg, plan, opts)
}
var stdout bytes.Buffer
@@ -224,12 +223,12 @@ previous_session_id: 2026-04-26
executeStagesFn = origExecuteStagesFn
newObjectStoreFromConfigFn = origObjectStoreFn
})
executeStagesFn = func(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
executeStagesFn = func(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
if opts.Env == nil {
opts.Env = &Env{}
}
opts.Env.Scriptorium = scriptoriumFake
return executeStages(ctx, cfg, stages, opts)
return executePlan(ctx, cfg, plan, opts)
}
newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
objectStoreConstructed = true

4
internal/app/run.go
View File

@@ -33,12 +33,10 @@ func Run(ctx context.Context, args []string, out io.Writer) error {
if err != nil {
return fmt.Errorf("run: %w", err)
}
stages := request.Plan.Stages()
summary, err := executeStagesFn(ctx, cfg, stages, RunOptions{
summary, err := executeStagesFn(ctx, cfg, request.Plan, RunOptions{
Force: request.Force,
SelectedArtifacts: request.SelectedArtifacts,
EffectiveArtifacts: effectiveArtifacts,
Plan: request.Plan,
})
if err != nil {
return fmt.Errorf("run: %w", err)

4
internal/app/run_stage.go
View File

@@ -199,7 +199,7 @@ type singleStageCommand struct {
}
func runSingleStageCommand(ctx context.Context, req singleStageCommand) (*RunSummary, error) {
stages, err := BuildSingleStagePlan(req.StageName)
plan, err := buildSingleStageExecutionPlan(req.StageName)
if err != nil {
return nil, fmt.Errorf("%s: %w", req.CommandName, err)
}
@@ -220,7 +220,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, stages, RunOptions{
summary, err := executeStagesFn(ctx, cfg, plan, RunOptions{
Force: req.Force,
SelectedArtifacts: req.SelectedArtifacts,
EffectiveArtifacts: effectiveArtifacts,

10
internal/app/runner.go
View File

@@ -26,7 +26,6 @@ type RunOptions struct {
Force bool
SelectedArtifacts []string
EffectiveArtifacts artifacts.EffectiveArtifactSet
Plan BoundedPlan
Env *Env
RunManifestStore manifest.RunStore
}
@@ -41,14 +40,15 @@ type RunSummary struct {
Skipped []string
}
var executeStagesFn = executeStages
var executeStagesFn = executePlan
func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (summary *RunSummary, resultErr error) {
func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (summary *RunSummary, resultErr error) {
stages := plan.Stages()
var prerequisiteStore manifest.Store
if opts.Env != nil {
prerequisiteStore = opts.Env.ManifestStore
}
if err := inspectBoundedPrerequisites(ctx, cfg, opts.Plan, prerequisiteStore); err != nil {
if err := inspectBoundedPrerequisites(ctx, cfg, plan, prerequisiteStore); err != nil {
return nil, fmt.Errorf("validate bounded run prerequisites: %w", err)
}
effectiveArtifacts := opts.EffectiveArtifacts
@@ -143,7 +143,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
// already-invalid request. Recheck the manifest protected by the session
// lock because another invocation may have changed prerequisite state while
// this invocation waited to acquire the lock.
if err := validateBoundedPrerequisites(opts.Plan, m); err != nil {
if err := validateBoundedPrerequisites(plan, m); err != nil {
return nil, fmt.Errorf("validate bounded run prerequisites under session lock: %w", err)
}
identity.applyToSessionManifest(m)

15
internal/app/runner_test_helpers_test.go Normal file
View File

@@ -0,0 +1,15 @@
package app
import (
"context"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
// executeStages keeps runner tests focused on controlled stage doubles. The
// production command path always supplies one validated BoundedPlan directly.
func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
plan := BoundedPlan{stages: append([]stage.Stage(nil), stages...)}
return executePlan(ctx, cfg, plan, opts)
}

9
internal/app/session_oriented_cli_test.go
View File

@@ -11,7 +11,6 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
func TestExecuteRunAcceptsPositionalSessionID(t *testing.T) {
@@ -21,7 +20,7 @@ func TestExecuteRunAcceptsPositionalSessionID(t *testing.T) {
var capturedSessionID string
origExecuteStagesFn := executeStagesFn
t.Cleanup(func() { executeStagesFn = origExecuteStagesFn })
executeStagesFn = func(_ context.Context, cfg *config.Config, _ []stage.Stage, _ RunOptions) (*RunSummary, error) {
executeStagesFn = func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
capturedSessionID = cfg.Session.SessionID
return &RunSummary{
SessionID: cfg.Session.SessionID,
@@ -117,7 +116,7 @@ inputs:
`)
origExecuteStagesFn := executeStagesFn
t.Cleanup(func() { executeStagesFn = origExecuteStagesFn })
executeStagesFn = func(_ context.Context, cfg *config.Config, _ []stage.Stage, _ RunOptions) (*RunSummary, error) {
executeStagesFn = func(_ context.Context, cfg *config.Config, _ BoundedPlan, _ RunOptions) (*RunSummary, error) {
return &RunSummary{
SessionID: cfg.Session.SessionID,
ManifestPath: filepath.Join(workspaceRoot, "manifest.json"),
@@ -194,8 +193,8 @@ func TestExecuteWorkflowCommandsAcceptPositionalSessionID(t *testing.T) {
var capturedArtifacts []string
origExecuteStagesFn := executeStagesFn
t.Cleanup(func() { executeStagesFn = origExecuteStagesFn })
executeStagesFn = func(_ context.Context, _ *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
for _, s := range stages {
executeStagesFn = func(_ context.Context, _ *config.Config, plan BoundedPlan, opts RunOptions) (*RunSummary, error) {
for _, s := range plan.Stages() {
capturedStages = append(capturedStages, s.Name())
}
capturedForce = opts.Force