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

Unify pre-execution stage decisions

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-04 01:01:46 +00:00
parent 9331373245
commit ca1b694aa9
5 changed files with 297 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/roadmap/implementation.md
View File

@@ -121,7 +121,7 @@ During every stage:
## Stage 1 — Unify Pre-Execution Decisions
**Status: Pending**
**Status: Completed**
### Goal

65
internal/app/plan.go
View File

@@ -73,26 +73,20 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
if err != nil {
return fmt.Errorf("plan: fingerprint semantic configuration for stage %q: %w", selectedStage.Name(), err)
}
action := decideStageAction(selectedStage, model, request.Force)
decision := decideStageExecution(recordedStageStatus(model, selectedStage.Name()), request.Force, nil)
var validation *stage.ResumeValidation
if action == stageActionSkip {
if decision.action == stageActionSkip {
checked, validationErr := evaluateStageResume(ctx, selectedStage, stageEnv, model, semanticConfig)
if validationErr != nil {
return fmt.Errorf("plan: validate resume for stage %q: %w", selectedStage.Name(), validationErr)
}
validation = checked
if checked != nil && !checked.Resumable {
at := time.Now().UTC()
model.MarkStageStale(selectedStage.Name(), at, checked.Reason)
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(
model, selectedStage.Name(), at, staleReasonNotResumable,
); invalidationErr != nil {
return fmt.Errorf("plan: model resume invalidation for stage %q: %w", selectedStage.Name(), invalidationErr)
}
action = stageActionRun
}
decision = decideStageExecution(recordedStageStatus(model, selectedStage.Name()), request.Force, checked)
}
if action == stageActionRun && selectedStage.Name() == "analyze" {
if _, err := applyPreExecutionDecision(model, selectedStage.Name(), decision, time.Now().UTC()); err != nil {
return fmt.Errorf("plan: model resume invalidation for stage %q: %w", selectedStage.Name(), err)
}
if decision.action == stageActionRun && selectedStage.Name() == "analyze" {
if validator, ok := selectedStage.(stage.ResumeValidator); ok {
checked, validationErr := validator.ValidateResume(ctx, stageEnv, model)
if validationErr != nil {
@@ -102,12 +96,12 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
validation = &checked
}
}
if action == stageActionRun {
if decision.action == stageActionRun {
runCount++
} else {
skipCount++
}
if _, err := fmt.Fprintf(out, "%s: %s\n", selectedStage.Name(), action); err != nil {
if _, err := fmt.Fprintf(out, "%s: %s\n", selectedStage.Name(), decision.action); err != nil {
return err
}
if validation != nil && validation.Analyze != nil {
@@ -115,8 +109,8 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
return err
}
}
if action == stageActionRun {
if err := modelPlannedStageRun(model, selectedStage, cfg, request.Force, semanticConfig); err != nil {
if decision.action == stageActionRun {
if err := modelPlannedStageRun(model, selectedStage, cfg, decision, semanticConfig); err != nil {
return fmt.Errorf("plan: model stage %q: %w", selectedStage.Name(), err)
}
}
@@ -149,40 +143,23 @@ func modelPlannedStageRun(
model *manifest.Manifest,
selectedStage stage.Stage,
cfg *config.Config,
force bool,
decision preExecutionDecision,
semanticConfig *manifest.SemanticConfigFingerprint,
) error {
prior := capturePriorStageOutcome(model, selectedStage.Name())
at := time.Now().UTC()
model.MarkStageRunning(selectedStage.Name(), at)
if force {
if _, err := invalidateDependentSucceededStagesWithReason(
model, selectedStage.Name(), at, staleReasonForcedReplacement,
); err != nil {
return err
}
if _, err := beginStageAttempt(model, selectedStage.Name(), decision, at); err != nil {
return err
}
if reason := plannedSelfSkipReason(selectedStage.Name(), cfg); reason != "" {
model.MarkStageSkipped(selectedStage.Name(), at, reason)
setSessionStageSemanticConfig(model, selectedStage.Name(), semanticConfig)
if !prior.isSameSelfSkip(reason) {
_, err := invalidateDependentSucceededStagesWithReason(
model, selectedStage.Name(), at, staleReasonSelfSkip,
)
return err
}
return nil
_, err := applyCompletedStageOutcome(model, selectedStage.Name(), &stage.StageResult{
Disposition: stage.StageDispositionSkipped,
SkipReason: reason,
}, nil, semanticConfig, prior, at)
return err
}
model.MarkStageSucceeded(selectedStage.Name(), at, nil)
setSessionStageSemanticConfig(model, selectedStage.Name(), semanticConfig)
if !prior.exists || prior.status != manifest.StatusSucceeded {
if _, err := invalidateDependentSucceededStagesWithReason(
model, selectedStage.Name(), at, staleReasonChangedResult,
); err != nil {
return err
}
}
return nil
_, err := applyCompletedStageOutcome(model, selectedStage.Name(), &stage.StageResult{}, nil, semanticConfig, prior, at)
return err
}
func plannedSelfSkipReason(stageName string, cfg *config.Config) string {

88
internal/app/run_control.go
View File

@@ -34,21 +34,89 @@ type priorStageOutcome struct {
outputs int
}
func decideStageAction(s stage.Stage, m *manifest.Manifest, force bool) stageAction {
// A succeeded aggregate record is the initial skip candidate. The runner and
// planner then let stage-owned resume validation refine that decision.
if !force && stageSucceeded(m, s.Name()) {
return stageActionSkip
}
return stageActionRun
type preExecutionDecision struct {
action stageAction
markSelectedStale bool
selectedStaleReason string
invalidationReason string
invalidateWhenStarted bool
}
func stageSucceeded(m *manifest.Manifest, name string) bool {
func decideStageExecution(status manifest.StageStatus, force bool, resume *stage.ResumeValidation) preExecutionDecision {
if force {
return preExecutionDecision{
action: stageActionRun,
invalidationReason: staleReasonForcedReplacement,
invalidateWhenStarted: true,
}
}
if status != manifest.StatusSucceeded {
return preExecutionDecision{action: stageActionRun}
}
if resume != nil && !resume.Resumable {
return preExecutionDecision{
action: stageActionRun,
markSelectedStale: true,
selectedStaleReason: resume.Reason,
invalidationReason: staleReasonNotResumable,
}
}
return preExecutionDecision{action: stageActionSkip}
}
func recordedStageStatus(m *manifest.Manifest, name string) manifest.StageStatus {
if m == nil || m.Stages == nil {
return false
return ""
}
sr := m.Stages[name]
return sr != nil && sr.Status == manifest.StatusSucceeded
if sr == nil {
return ""
}
return sr.Status
}
func applyPreExecutionDecision(m *manifest.Manifest, name string, decision preExecutionDecision, at time.Time) ([]string, error) {
if !decision.markSelectedStale {
return nil, nil
}
m.MarkStageStale(name, at, decision.selectedStaleReason)
return invalidateDependentSucceededStagesWithReason(m, name, at, decision.invalidationReason)
}
func beginStageAttempt(m *manifest.Manifest, name string, decision preExecutionDecision, at time.Time) ([]string, error) {
m.MarkStageRunning(name, at)
if !decision.invalidateWhenStarted {
return nil, nil
}
return invalidateDependentSucceededStagesWithReason(m, name, at, decision.invalidationReason)
}
func applyCompletedStageOutcome(
m *manifest.Manifest,
name string,
result *stage.StageResult,
outputs []manifest.ArtifactRecord,
semanticConfig *manifest.SemanticConfigFingerprint,
prior priorStageOutcome,
at time.Time,
) ([]string, error) {
if result != nil && result.Disposition == stage.StageDispositionSkipped {
m.MarkStageSkipped(name, at, result.SkipReason)
applyStageResultToManifest(m, name, result)
setSessionStageSemanticConfig(m, name, semanticConfig)
if prior.isSameSelfSkip(result.SkipReason) {
return nil, nil
}
return invalidateDependentSucceededStagesWithReason(m, name, at, staleReasonSelfSkip)
}
m.MarkStageSucceeded(name, at, outputs)
applyStageResultToManifest(m, name, result)
setSessionStageSemanticConfig(m, name, semanticConfig)
if prior.exists && prior.status == manifest.StatusSucceeded {
return nil, nil
}
return invalidateDependentSucceededStagesWithReason(m, name, at, staleReasonChangedResult)
}
func capturePriorStageOutcome(m *manifest.Manifest, name string) priorStageOutcome {

185
internal/app/run_control_test.go
View File

@@ -2,6 +2,7 @@ package app
import (
"context"
"fmt"
"reflect"
"strings"
"testing"
@@ -11,23 +12,185 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
func TestDecideStageAction(t *testing.T) {
stages := BuildFullPlan()[:2]
m := manifest.New("2026-05-03", time.Now().UTC())
m.MarkStageSucceeded("prepare", time.Now().UTC(), nil)
if got := decideStageAction(stages[0], m, false); got != stageActionSkip {
t.Fatalf("prepare action = %q, want %q", got, stageActionSkip)
func TestDecideStageExecution(t *testing.T) {
statuses := []manifest.StageStatus{"", manifest.StatusSucceeded, manifest.StatusFailed, manifest.StatusStale, manifest.StatusSkipped}
validations := []struct {
name string
value *stage.ResumeValidation
}{
{name: "none"},
{name: "resumable", value: resumeValidation(stage.Resumable())},
{name: "not resumable", value: resumeValidation(stage.NonResumable("output changed"))},
}
if got := decideStageAction(stages[1], m, false); got != stageActionRun {
t.Fatalf("transcribe action = %q, want %q", got, stageActionRun)
for _, status := range statuses {
for _, force := range []bool{false, true} {
for _, validation := range validations {
name := fmt.Sprintf("status=%s/force=%t/resume=%s", status, force, validation.name)
t.Run(name, func(t *testing.T) {
got := decideStageExecution(status, force, validation.value)
want := preExecutionDecision{action: stageActionRun}
switch {
case force:
want.invalidationReason = staleReasonForcedReplacement
want.invalidateWhenStarted = true
case status == manifest.StatusSucceeded && validation.value != nil && !validation.value.Resumable:
want.markSelectedStale = true
want.selectedStaleReason = "output changed"
want.invalidationReason = staleReasonNotResumable
case status == manifest.StatusSucceeded:
want.action = stageActionSkip
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("decision = %#v, want %#v", got, want)
}
})
}
}
}
}
func resumeValidation(value stage.ResumeValidation) *stage.ResumeValidation { return &value }
func TestPlanningAndExecutionTransitionParity(t *testing.T) {
tests := []struct {
name string
seed func(*manifest.Manifest, time.Time)
force bool
resume *stage.ResumeValidation
result *stage.StageResult
wantAction stageAction
wantStatus manifest.StageStatus
}{
{
name: "ordinary skip",
seed: func(m *manifest.Manifest, at time.Time) {
m.MarkStageSucceeded("prepare", at, nil)
m.MarkStageSucceeded("transcribe", at, nil)
},
wantAction: stageActionSkip,
wantStatus: manifest.StatusSucceeded,
},
{
name: "non-resumable success",
seed: func(m *manifest.Manifest, at time.Time) {
m.MarkStageSucceeded("prepare", at, nil)
m.MarkStageSucceeded("transcribe", at, nil)
},
resume: resumeValidation(stage.NonResumable("output changed")),
result: &stage.StageResult{},
wantAction: stageActionRun,
wantStatus: manifest.StatusStale,
},
{
name: "forced replacement",
seed: func(m *manifest.Manifest, at time.Time) {
m.MarkStageSucceeded("prepare", at, nil)
m.MarkStageSucceeded("transcribe", at, nil)
},
force: true,
result: &stage.StageResult{},
wantAction: stageActionRun,
wantStatus: manifest.StatusStale,
},
{
name: "unchanged self-skip",
seed: func(m *manifest.Manifest, at time.Time) {
m.MarkStageSkipped("prepare", at, "disabled")
m.MarkStageSucceeded("transcribe", at, nil)
},
result: &stage.StageResult{
Disposition: stage.StageDispositionSkipped,
SkipReason: "disabled",
},
wantAction: stageActionRun,
wantStatus: manifest.StatusSucceeded,
},
{
name: "changed self-skip",
seed: func(m *manifest.Manifest, at time.Time) {
m.MarkStageSkipped("prepare", at, "old reason")
m.MarkStageSucceeded("transcribe", at, nil)
},
result: &stage.StageResult{
Disposition: stage.StageDispositionSkipped,
SkipReason: "new reason",
},
wantAction: stageActionRun,
wantStatus: manifest.StatusStale,
},
{
name: "success replacing non-success",
seed: func(m *manifest.Manifest, at time.Time) {
m.MarkStageFailed("prepare", at, "failed")
m.MarkStageSucceeded("transcribe", at, nil)
},
result: &stage.StageResult{},
wantAction: stageActionRun,
wantStatus: manifest.StatusStale,
},
}
if got := decideStageAction(stages[0], m, true); got != stageActionRun {
t.Fatalf("forced prepare action = %q, want %q", got, stageActionRun)
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
at := time.Date(2026, time.May, 3, 12, 0, 0, 0, time.UTC)
planned := manifest.New("2026-05-03", at)
executed := manifest.New("2026-05-03", at)
test.seed(planned, at)
test.seed(executed, at)
plannedAction := applyParityTransition(t, planned, test.force, test.resume, test.result, false, at.Add(time.Second))
executedAction := applyParityTransition(t, executed, test.force, test.resume, test.result, true, at.Add(time.Second))
if plannedAction != test.wantAction || executedAction != test.wantAction {
t.Fatalf("actions = planned %q, executed %q; want %q", plannedAction, executedAction, test.wantAction)
}
if got := planned.Stages["transcribe"].Status; got != test.wantStatus {
t.Fatalf("planned dependent status = %q, want %q", got, test.wantStatus)
}
if got := executed.Stages["transcribe"].Status; got != test.wantStatus {
t.Fatalf("executed dependent status = %q, want %q", got, test.wantStatus)
}
plannedError := planned.Stages["transcribe"].Error
executedError := executed.Stages["transcribe"].Error
if !reflect.DeepEqual(plannedError, executedError) {
t.Fatalf("dependent errors differ: planned %#v, executed %#v", plannedError, executedError)
}
})
}
}
func applyParityTransition(
t *testing.T,
m *manifest.Manifest,
force bool,
resume *stage.ResumeValidation,
result *stage.StageResult,
actual bool,
at time.Time,
) stageAction {
t.Helper()
decision := decideStageExecution(recordedStageStatus(m, "prepare"), force, resume)
if _, err := applyPreExecutionDecision(m, "prepare", decision, at); err != nil {
t.Fatalf("apply pre-execution decision: %v", err)
}
if decision.action == stageActionSkip {
return decision.action
}
prior := capturePriorStageOutcome(m, "prepare")
if _, err := beginStageAttempt(m, "prepare", decision, at); err != nil {
t.Fatalf("begin attempt: %v", err)
}
if result != nil {
modeled := *result
if actual {
modeled.Logs = []string{"run.log"}
}
if _, err := applyCompletedStageOutcome(m, "prepare", &modeled, nil, nil, prior, at); err != nil {
t.Fatalf("apply completed outcome: %v", err)
}
}
return decision.action
}
func TestInvalidationDependents(t *testing.T) {
tests := []struct {
stage string

62
internal/app/runner.go
View File

@@ -244,9 +244,9 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
fmt.Errorf("fingerprint semantic configuration for stage %q: %w", s.Name(), err),
)
}
action := decideStageAction(s, m, opts.Force)
decision := decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, nil)
if action == stageActionSkip {
if decision.action == stageActionSkip {
validation, err := evaluateStageResume(ctx, s, stageEnv, m, semanticConfig)
if err != nil {
return nil, persistSessionTerminalFailure(
@@ -254,12 +254,10 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
fmt.Errorf("validate resume for stage %q: %w", s.Name(), err),
)
}
if validation != nil && !validation.Resumable {
decision = decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, validation)
if decision.markSelectedStale {
staleAt := nowUTC()
m.MarkStageStale(s.Name(), staleAt, validation.Reason)
if _, err := invalidateDependentSucceededStagesWithReason(
m, s.Name(), staleAt, staleReasonNotResumable,
); err != nil {
if _, err := applyPreExecutionDecision(m, s.Name(), decision, staleAt); err != nil {
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", s.Name(), err),
@@ -271,12 +269,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err),
)
}
env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", validation.Reason)
action = stageActionRun
env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", decision.selectedStaleReason)
}
}
if action == stageActionSkip {
if decision.action == stageActionSkip {
skipped = append(skipped, s.Name())
env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force)
continue
@@ -286,14 +283,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
priorAnalyzeState := captureAnalyzeState(m, s.Name())
now := nowUTC()
m.MarkStageRunning(s.Name(), now)
if opts.Force {
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), now, staleReasonForcedReplacement); err != nil {
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents before forced stage %q: %w", s.Name(), err),
)
}
if _, err := beginStageAttempt(m, s.Name(), decision, now); err != nil {
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents before forced stage %q: %w", s.Name(), err),
)
}
env.Logger.Info("starting stage", "stage", s.Name())
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
@@ -340,16 +334,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if result != nil && result.Disposition == stage.StageDispositionSkipped {
skipped = append(skipped, s.Name())
skippedAt := nowUTC()
m.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
applyStageResultToManifest(m, s.Name(), result)
setSessionStageSemanticConfig(m, s.Name(), semanticConfig)
if !priorOutcome.isSameSelfSkip(result.SkipReason) {
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip); err != nil {
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents after stage %q self-skip: %w", s.Name(), err),
)
}
if _, err := applyCompletedStageOutcome(m, s.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil {
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents after stage %q self-skip: %w", s.Name(), err),
)
}
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, persistSessionTerminalFailure(
@@ -367,18 +356,13 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session)
}
succeededAt := nowUTC()
m.MarkStageSucceeded(s.Name(), succeededAt, sessionOutputs)
applyAnalyzeProjection(m, analyzeProjection)
applyStageResultToManifest(m, s.Name(), result)
setSessionStageSemanticConfig(m, s.Name(), semanticConfig)
if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded {
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult); err != nil {
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err),
)
}
if _, err := applyCompletedStageOutcome(m, s.Name(), result, sessionOutputs, semanticConfig, priorOutcome, succeededAt); err != nil {
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err),
)
}
applyAnalyzeProjection(m, analyzeProjection)
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
operationErr := fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)