diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index eddd024..44f9f83 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -86,7 +86,10 @@ failed in both manifests, persisting each transition. On success it records outputs, logs, generated configuration references, and metadata. 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 downstream session-stage records stale. +rerun marks only succeeded transitive dependent session-stage records stale. +The application owns a fixed dependency relation distinct from execution order; +dependents are returned in canonical order. Render and extract therefore never +stale one another, while either can stale analyze, publish, and notify. Starting an execution clears the current session-stage record's prior outputs, logs, generated configuration references, and metadata. Failed and skipped @@ -139,7 +142,7 @@ where a durable running record can require operator interpretation. - running, failed, and self-skipped stages do not retain result payloads from an earlier success. - stale stages retain prior details until replacement execution starts. -- force reruns stale downstream succeeded stages. +- force reruns stale succeeded stages in the fixed dependency relation. - run manifest does not replace session manifest as progress authority. - remote commitment is established by a verified current pointer and remote commit relationship, never by a mutable session-manifest boolean. diff --git a/docs/internal/overview.md b/docs/internal/overview.md index d1f89c2..52adc46 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -53,8 +53,8 @@ The implemented canonical order is: 4. [`polish`](stage-polish.md) 5. [`normalize`](stage-normalize.md) 6. [`trim`](stage-trim.md) -7. [`extract`](stage-extract.md) -8. [`render`](stage-render.md) +7. [`render`](stage-render.md) +8. [`extract`](stage-extract.md) 9. [`analyze`](stage-analyze.md) 10. [`publish`](stage-publish.md) 11. `notify` (no-op) @@ -65,6 +65,13 @@ mechanics. The [CLI](../cli.md) and [Operations](../operations.md) own user-visible invocation and execution semantics. +Execution order and invalidation are separate application contracts. The stage +registry owns the flat execution sequence. The application orchestration owner +uses a fixed, validated dependency relation to find transitive dependents in +canonical order. In particular, `render` and `extract` are sibling consumers of +trimmed transcript state: neither invalidates the other, while either can stale +`analyze`, `publish`, and `notify`. + ## Focused Documentation - [Adapter Internals](adapters.md): external adapter boundaries, composition, @@ -84,8 +91,8 @@ and execution semantics. - [`polish`](stage-polish.md) - [`normalize`](stage-normalize.md) - [`trim`](stage-trim.md) -- [`extract`](stage-extract.md) - [`render`](stage-render.md) +- [`extract`](stage-extract.md) - [`analyze`](stage-analyze.md) - [`publish`](stage-publish.md) diff --git a/docs/internal/stage-extract.md b/docs/internal/stage-extract.md index f058641..3f1ed59 100644 --- a/docs/internal/stage-extract.md +++ b/docs/internal/stage-extract.md @@ -2,7 +2,7 @@ ## Responsibility -`extract` runs after `trim` and before `render`. It converts the canonical +`extract` runs after `render` and before `analyze`. It converts the canonical `narratio.transcript.final_trimmed` JSON into configured Notarius lane artifacts. An omitted or disabled Notarius section makes the stage explicitly self-skip with reason `notarius_disabled`, no outputs, and no Notarius runner. @@ -50,8 +50,9 @@ Validation completes before promotion, so a rejected result cannot expose a partial durable bundle. Any executed extraction outcome that replaces a different effective outcome -marks succeeded downstream stages stale. Repeating the same disabled self-skip -with no outputs is stable and does not repeatedly invalidate downstream stages. +marks succeeded analysis and delivery dependents stale. Render is an independent +sibling and remains current. Repeating the same disabled self-skip with no +outputs is stable and does not repeatedly invalidate dependent stages. ## Resume Validation diff --git a/docs/internal/stage-render.md b/docs/internal/stage-render.md index 314a780..6a6f878 100644 --- a/docs/internal/stage-render.md +++ b/docs/internal/stage-render.md @@ -3,6 +3,10 @@ ## Purpose Render Markdown transcript artifacts from normalized JSON transcripts via Seriatim. +It runs after `trim` and before `extract` in the canonical sequence. Render and +extract are independent sibling consumers: replacing render output does not +invalidate extraction, but it does invalidate succeeded analysis and delivery +records that may consume rendered transcripts. ## Inputs diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index d7176cc..28103a3 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -87,8 +87,10 @@ merely on incidental files existing on disk. A failed or interrupted stage must not be presented as successful. Failure should preserve enough local state and diagnostics for inspection, recovery, -and resume. Forcing an upstream stage invalidates succeeded downstream work -according to the canonical stage order. +and resume. Forcing a stage invalidates succeeded transitive dependents +according to a fixed application-owned relation that is separate from canonical +execution order. The relation is validated against the stage inventory and is +not configurable. A stage may explicitly self-skip with a stable reason and no outputs. That outcome is persisted, clears older outputs owned by the stage, and is diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 6a0ad64..88cc398 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -90,6 +90,8 @@ failing tests or incomplete acceptance criteria. ## Stage 1 — Canonical Order And Dependency-Aware Invalidation +**Status: Completed** + ### Goal Move `render` before `extract` and replace suffix-based invalidation with the diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index 4eb61d0..a4c7b33 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -32,7 +32,7 @@ func TestExecuteValidCommands(t *testing.T) { wantOut string }{ {name: "run", args: []string{"run", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run: session 2026-05-03; executed=11 skipped=1; manifest="}, - {name: "session plan", args: []string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "prepare: skip\ntranscribe: skip\nmerge: skip\npolish: skip\nnormalize: skip\ntrim: skip\nextract: run\nrender: skip\nanalyze: skip\npublish: skip\nnotify: skip"}, + {name: "session plan", args: []string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "prepare: skip\ntranscribe: skip\nmerge: skip\npolish: skip\nnormalize: skip\ntrim: skip\nrender: skip\nextract: run\nanalyze: skip\npublish: skip\nnotify: skip"}, {name: "session status", args: []string{"session", "status", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "Session: 2026-05-03"}, {name: "run-stage", args: []string{"run-stage", "polish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run-stage: stage=polish executed=0 skipped=1 force=false; manifest="}, } diff --git a/internal/app/extract_lifecycle_test.go b/internal/app/extract_lifecycle_test.go index 857a536..3a56a76 100644 --- a/internal/app/extract_lifecycle_test.go +++ b/internal/app/extract_lifecycle_test.go @@ -256,7 +256,10 @@ func TestExtractLifecyclePreparedReferenceChangeRerunsExtractionAndInvalidatesDo if after.Stages["extract"].Status != manifest.StatusSucceeded { t.Fatalf("extract status = %#v", after.Stages["extract"]) } - for _, name := range []string{"render", "analyze", "publish"} { + if after.Stages["render"] == nil || after.Stages["render"].Status != manifest.StatusSucceeded { + t.Fatalf("render status = %#v, want succeeded sibling", after.Stages["render"]) + } + for _, name := range []string{"analyze", "publish"} { if after.Stages[name] == nil || after.Stages[name].Status != manifest.StatusStale { t.Fatalf("%s status = %#v, want stale", name, after.Stages[name]) } diff --git a/internal/app/plan_test.go b/internal/app/plan_test.go index 833cf4c..af57904 100644 --- a/internal/app/plan_test.go +++ b/internal/app/plan_test.go @@ -27,7 +27,7 @@ func TestPlanCreatesAndReusesWorkdir(t *testing.T) { if !strings.Contains(got, "narratio session plan: workdir prepared at") { t.Fatalf("first output = %q, want workdir prepared", got) } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"} { if !strings.Contains(got, name+": run") { t.Fatalf("first output = %q, missing stage %q", got, name) } diff --git a/internal/app/planner_test.go b/internal/app/planner_test.go index a76f6f4..88da2dd 100644 --- a/internal/app/planner_test.go +++ b/internal/app/planner_test.go @@ -4,7 +4,7 @@ import "testing" func TestBuildFullPlanOrder(t *testing.T) { got := BuildFullPlan() - want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} + want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"} if len(got) != len(want) { t.Fatalf("len(plan) = %d, want %d", len(got), len(want)) } diff --git a/internal/app/run_control.go b/internal/app/run_control.go index d7d504c..7d6c2c4 100644 --- a/internal/app/run_control.go +++ b/internal/app/run_control.go @@ -122,30 +122,153 @@ func canonicalStageNames() []string { return out } -func downstreamStageNames(stageName string) []string { - names := canonicalStageNames() - for i, name := range names { - if name != stageName { - continue - } - return append([]string(nil), names[i+1:]...) - } - return nil +type invalidationRelation struct { + canonical []string + direct map[string][]string } -func invalidateDownstreamSucceededStagesWithReason(m *manifest.Manifest, upstreamStage string, at time.Time, reason string) []string { - if m == nil || m.Stages == nil { +var canonicalInvalidationEdges = map[string][]string{ + "prepare": {"transcribe"}, + "transcribe": {"merge"}, + "merge": {"polish"}, + "polish": {"normalize"}, + "normalize": {"trim"}, + "trim": {"render", "extract"}, + "render": {"analyze"}, + "extract": {"analyze"}, + "analyze": {"publish"}, + "publish": {"notify"}, + "notify": {}, +} + +func newInvalidationRelation(registry []stage.Stage, direct map[string][]string) (*invalidationRelation, error) { + canonical := make([]string, 0, len(registry)) + known := make(map[string]struct{}, len(registry)) + for index, candidate := range registry { + if candidate == nil { + return nil, fmt.Errorf("canonical stage registry entry %d is nil", index) + } + name := strings.TrimSpace(candidate.Name()) + if name == "" { + return nil, fmt.Errorf("canonical stage registry entry %d has an empty name", index) + } + if _, duplicate := known[name]; duplicate { + return nil, fmt.Errorf("canonical stage registry contains duplicate stage %q", name) + } + known[name] = struct{}{} + canonical = append(canonical, name) + } + + cloned := make(map[string][]string, len(direct)) + for source, targets := range direct { + if _, ok := known[source]; !ok { + return nil, fmt.Errorf("invalidation relation classifies unknown stage %q", source) + } + cloned[source] = []string{} + seenTargets := make(map[string]struct{}, len(targets)) + for _, target := range targets { + if _, ok := known[target]; !ok { + return nil, fmt.Errorf("invalidation relation edge %q -> %q references an unknown stage", source, target) + } + if _, duplicate := seenTargets[target]; duplicate { + return nil, fmt.Errorf("invalidation relation contains duplicate edge %q -> %q", source, target) + } + seenTargets[target] = struct{}{} + cloned[source] = append(cloned[source], target) + } + } + for _, name := range canonical { + if _, classified := direct[name]; !classified { + return nil, fmt.Errorf("invalidation relation is missing classification for stage %q", name) + } + } + + relation := &invalidationRelation{canonical: canonical, direct: cloned} + visiting := make(map[string]bool, len(canonical)) + visited := make(map[string]bool, len(canonical)) + var visit func(string) error + visit = func(name string) error { + if visiting[name] { + return fmt.Errorf("invalidation relation contains a cycle involving stage %q", name) + } + if visited[name] { + return nil + } + visiting[name] = true + for _, target := range relation.direct[name] { + if err := visit(target); err != nil { + return err + } + } + visiting[name] = false + visited[name] = true return nil } + for _, name := range canonical { + if err := visit(name); err != nil { + return nil, err + } + } + return relation, nil +} + +func canonicalInvalidationRelation() (*invalidationRelation, error) { + return newInvalidationRelation(stage.All(), canonicalInvalidationEdges) +} + +func (r *invalidationRelation) Dependents(stageName string) ([]string, error) { + if r == nil { + return nil, fmt.Errorf("invalidation relation is nil") + } + if _, ok := r.direct[stageName]; !ok { + return nil, fmt.Errorf("unknown stage %q in invalidation relation", stageName) + } + reachable := make(map[string]bool, len(r.canonical)) + var collect func(string) + collect = func(name string) { + for _, target := range r.direct[name] { + if reachable[target] { + continue + } + reachable[target] = true + collect(target) + } + } + collect(stageName) + out := make([]string, 0, len(reachable)) + for _, name := range r.canonical { + if reachable[name] { + out = append(out, name) + } + } + return out, nil +} + +func dependentStageNames(stageName string) ([]string, error) { + relation, err := canonicalInvalidationRelation() + if err != nil { + return nil, err + } + return relation.Dependents(stageName) +} + +func invalidateDependentSucceededStagesWithReason(m *manifest.Manifest, upstreamStage string, at time.Time, reason string) ([]string, error) { + dependents, err := dependentStageNames(upstreamStage) + if err != nil { + return nil, err + } + if m == nil || m.Stages == nil { + return nil, nil + } invalidated := make([]string, 0) - for _, downstream := range downstreamStageNames(upstreamStage) { - sr := m.Stages[downstream] + for _, dependent := range dependents { + sr := m.Stages[dependent] if sr == nil || sr.Status != manifest.StatusSucceeded { continue } - m.MarkStageStale(downstream, at, reason) - invalidated = append(invalidated, downstream) + m.MarkStageStale(dependent, at, reason) + invalidated = append(invalidated, dependent) } - return invalidated + return invalidated, nil } diff --git a/internal/app/run_control_test.go b/internal/app/run_control_test.go index 7ccf7cf..4d2978d 100644 --- a/internal/app/run_control_test.go +++ b/internal/app/run_control_test.go @@ -1,11 +1,14 @@ package app import ( + "context" "reflect" + "strings" "testing" "time" "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" ) func TestDecideStageActions(t *testing.T) { @@ -30,40 +33,82 @@ func TestDecideStageActions(t *testing.T) { } } -func TestDownstreamStageNames(t *testing.T) { - got := downstreamStageNames("polish") - want := []string{"normalize", "trim", "extract", "render", "analyze", "publish", "notify"} - if !reflect.DeepEqual(got, want) { - t.Fatalf("downstreamStageNames(polish) = %#v, want %#v", got, want) +func TestInvalidationDependents(t *testing.T) { + tests := []struct { + stage string + want []string + }{ + {"prepare", []string{"transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, + {"transcribe", []string{"merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, + {"merge", []string{"polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, + {"polish", []string{"normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, + {"normalize", []string{"trim", "render", "extract", "analyze", "publish", "notify"}}, + {"trim", []string{"render", "extract", "analyze", "publish", "notify"}}, + {"render", []string{"analyze", "publish", "notify"}}, + {"extract", []string{"analyze", "publish", "notify"}}, + {"analyze", []string{"publish", "notify"}}, + {"publish", []string{"notify"}}, + {"notify", []string{}}, } - - missing := downstreamStageNames("unknown") - if len(missing) != 0 { - t.Fatalf("downstreamStageNames(unknown) = %#v, want empty", missing) + for _, test := range tests { + t.Run(test.stage, func(t *testing.T) { + got, err := dependentStageNames(test.stage) + if err != nil { + t.Fatalf("dependentStageNames(%q) error = %v", test.stage, err) + } + if !reflect.DeepEqual(got, test.want) { + t.Fatalf("dependentStageNames(%q) = %#v, want %#v", test.stage, got, test.want) + } + }) + } + if _, err := dependentStageNames("unknown"); err == nil || !strings.Contains(err.Error(), "unknown stage") { + t.Fatalf("dependentStageNames(unknown) error = %v, want unknown-stage error", err) } } -func TestInvalidateDownstreamSucceededStagesWithReason(t *testing.T) { +func TestInvalidationRelationRejectsInvalidInventory(t *testing.T) { + canonical := []stage.Stage{ + invalidationTestStage("one"), + invalidationTestStage("two"), + } + tests := []struct { + name string + registry []stage.Stage + edges map[string][]string + want string + }{ + {name: "duplicate registry name", registry: append(canonical, invalidationTestStage("one")), edges: map[string][]string{"one": {"two"}, "two": {}}, want: "duplicate stage"}, + {name: "unknown source", registry: canonical, edges: map[string][]string{"one": {"two"}, "two": {}, "three": {}}, want: "unknown stage"}, + {name: "unknown target", registry: canonical, edges: map[string][]string{"one": {"three"}, "two": {}}, want: "unknown stage"}, + {name: "missing classification", registry: canonical, edges: map[string][]string{"one": {"two"}}, want: "missing classification"}, + {name: "cycle", registry: canonical, edges: map[string][]string{"one": {"two"}, "two": {"one"}}, want: "cycle"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := newInvalidationRelation(test.registry, test.edges) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("newInvalidationRelation() error = %v, want %q", err, test.want) + } + }) + } +} + +func TestInvalidateDependentSucceededStagesWithReason(t *testing.T) { now := time.Now().UTC() m := manifest.New("2026-05-03", now) - m.MarkStageSucceeded("prepare", now, nil) - m.MarkStageSucceeded("transcribe", now, nil) - m.MarkStageSucceeded("merge", now, nil) - m.MarkStageSucceeded("polish", now, nil) - m.MarkStageSucceeded("normalize", now, nil) - m.MarkStageSucceeded("trim", now, nil) - m.MarkStageSucceeded("extract", now, nil) - m.MarkStageSucceeded("render", now, nil) - m.MarkStageFailed("analyze", now, "analysis failed") - m.MarkStageSucceeded("publish", now, nil) - m.MarkStageSucceeded("notify", now, nil) - - got := invalidateDownstreamSucceededStagesWithReason(m, "polish", now.Add(1*time.Second), staleReasonChangedResult) - want := []string{"normalize", "trim", "extract", "render", "publish", "notify"} - if !reflect.DeepEqual(got, want) { - t.Fatalf("invalidateDownstreamSucceededStagesWithReason() = %#v, want %#v", got, want) + for _, name := range canonicalStageNames() { + m.MarkStageSucceeded(name, now, nil) } + m.MarkStageFailed("analyze", now, "analysis failed") + got, err := invalidateDependentSucceededStagesWithReason(m, "polish", now.Add(time.Second), staleReasonChangedResult) + if err != nil { + t.Fatalf("invalidateDependentSucceededStagesWithReason() error = %v", err) + } + want := []string{"normalize", "trim", "render", "extract", "publish", "notify"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("invalidateDependentSucceededStagesWithReason() = %#v, want %#v", got, want) + } for _, stageName := range want { if m.Stages[stageName].Status != manifest.StatusStale { t.Fatalf("%s status = %q, want stale", stageName, m.Stages[stageName].Status) @@ -72,34 +117,38 @@ func TestInvalidateDownstreamSucceededStagesWithReason(t *testing.T) { if m.Stages["analyze"].Status != manifest.StatusFailed { t.Fatalf("analyze status = %q, want failed", m.Stages["analyze"].Status) } - if m.Stages["prepare"].Status != manifest.StatusSucceeded { - t.Fatalf("prepare status = %q, want succeeded", m.Stages["prepare"].Status) - } } -func TestExtractionPositionControlsForceInvalidation(t *testing.T) { +func TestRenderAndExtractInvalidationAreIndependent(t *testing.T) { now := time.Now().UTC() - tests := []struct { - upstream string - want []string - }{ - {upstream: "trim", want: []string{"extract", "render", "analyze", "publish", "notify"}}, - {upstream: "extract", want: []string{"render", "analyze", "publish", "notify"}}, - {upstream: "render", want: []string{"analyze", "publish", "notify"}}, - } - for _, test := range tests { - t.Run(test.upstream, func(t *testing.T) { + for _, upstream := range []string{"render", "extract"} { + t.Run(upstream, func(t *testing.T) { m := manifest.New("2026-05-03", now) for _, name := range canonicalStageNames() { m.MarkStageSucceeded(name, now, nil) } - got := invalidateDownstreamSucceededStagesWithReason(m, test.upstream, now.Add(time.Second), staleReasonForcedReplacement) - if !reflect.DeepEqual(got, test.want) { - t.Fatalf("invalidated = %#v, want %#v", got, test.want) + got, err := invalidateDependentSucceededStagesWithReason(m, upstream, now.Add(time.Second), staleReasonForcedReplacement) + if err != nil { + t.Fatalf("invalidate dependents: %v", err) } - if test.upstream == "render" && m.Stages["extract"].Status != manifest.StatusSucceeded { - t.Fatalf("forcing render changed extract: %#v", m.Stages["extract"]) + want := []string{"analyze", "publish", "notify"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("invalidated = %#v, want %#v", got, want) + } + sibling := "render" + if upstream == "render" { + sibling = "extract" + } + if m.Stages[sibling].Status != manifest.StatusSucceeded { + t.Fatalf("%s invalidated sibling %s: %#v", upstream, sibling, m.Stages[sibling]) } }) } } + +type invalidationTestStage string + +func (s invalidationTestStage) Name() string { return string(s) } +func (s invalidationTestStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + return &stage.StageResult{}, nil +} diff --git a/internal/app/run_stage_test.go b/internal/app/run_stage_test.go index eb50aa3..da28a16 100644 --- a/internal/app/run_stage_test.go +++ b/internal/app/run_stage_test.go @@ -225,7 +225,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) if err != nil { t.Fatalf("load manifest after force: %v", err) } - for _, name := range []string{"normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"normalize", "trim", "render", "extract", "analyze", "publish", "notify"} { if afterForce.Stages[name] == nil || afterForce.Stages[name].Status != manifest.StatusStale { t.Fatalf("stage %q = %#v, want stale", name, afterForce.Stages[name]) } diff --git a/internal/app/runner.go b/internal/app/runner.go index d41450c..3ad79f1 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -261,9 +261,14 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage if !validation.Resumable { staleAt := nowUTC() m.MarkStageStale(s.Name(), staleAt, validation.Reason) - invalidateDownstreamSucceededStagesWithReason( + if _, err := invalidateDependentSucceededStagesWithReason( m, s.Name(), staleAt, staleReasonNotResumable, - ) + ); err != nil { + return nil, persistTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, + fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", s.Name(), err), + ) + } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, persistTerminalFailure( ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, @@ -305,13 +310,20 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage } m.MarkStageRunning(s.Name(), now) if opts.Force { - invalidateDownstreamSucceededStagesWithReason(m, s.Name(), now, staleReasonForcedReplacement) + if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), now, staleReasonForcedReplacement); err != nil { + return nil, persistTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, + 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 { operationErr := fmt.Errorf("save manifest before stage %q: %w", s.Name(), err) m.MarkStageFailed(s.Name(), nowUTC(), operationErr.Error()) - invalidateDownstreamSucceededStagesWithReason(m, s.Name(), nowUTC(), staleReasonFailure) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), nowUTC(), staleReasonFailure); invalidationErr != nil { + operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after stage %q persistence failure: %w", s.Name(), invalidationErr)) + } return nil, persistTerminalFailure( ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, operationErr, ) @@ -325,7 +337,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage if err != nil { failedAt := nowUTC() m.MarkStageFailed(s.Name(), failedAt, err.Error()) - invalidateDownstreamSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil { + err = errors.Join(err, fmt.Errorf("invalidate dependents after stage %q failure: %w", s.Name(), invalidationErr)) + } runManifest.MarkStageFailed(s.Name(), failedAt, err.Error()) identity.applyToRunManifest(runManifest, manifestPath) env.Logger.Info("stage failed", "stage", s.Name(), "error", err) @@ -340,7 +354,12 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage m.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason) applyStageResultToManifest(m, s.Name(), result) if !priorOutcome.isSameSelfSkip(result.SkipReason) { - invalidateDownstreamSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip) + if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip); err != nil { + return nil, persistTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, + 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, persistTerminalFailure( @@ -367,7 +386,12 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage m.MarkStageSucceeded(s.Name(), succeededAt, outputs) applyStageResultToManifest(m, s.Name(), result) if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded { - invalidateDownstreamSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult) + if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult); err != nil { + return nil, persistTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, + fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err), + ) + } } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 9202e1a..95a4c0c 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -564,12 +564,12 @@ func TestExecuteStagesUsesOptionalResumeValidation(t *testing.T) { cfg := testConfig(t) store := &manifest.LocalStore{} seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) - seed.MarkStageSucceeded("checked", time.Now().UTC(), nil) + seed.MarkStageSucceeded("extract", time.Now().UTC(), nil) if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { t.Fatalf("Save() error = %v", err) } runs := 0 - candidate := resumeCheckingStage{name: "checked", validation: test.validation, runs: &runs} + candidate := resumeCheckingStage{name: "extract", validation: test.validation, runs: &runs} summary, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) @@ -581,7 +581,7 @@ func TestExecuteStagesUsesOptionalResumeValidation(t *testing.T) { } } -func TestExecuteStagesNonResumableResultRerunsSucceededDownstream(t *testing.T) { +func TestExecuteStagesNonResumableResultPreservesSucceededSibling(t *testing.T) { cfg := testConfig(t) store := &manifest.LocalStore{} seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) @@ -599,7 +599,7 @@ func TestExecuteStagesNonResumableResultRerunsSucceededDownstream(t *testing.T) if err != nil { t.Fatalf("executeStages() error = %v", err) } - if extractRuns != 1 || renderRuns != 1 || len(summary.Executed) != 2 || len(summary.Skipped) != 0 { + if extractRuns != 1 || renderRuns != 0 || len(summary.Executed) != 1 || len(summary.Skipped) != 1 { t.Fatalf("extract runs=%d render runs=%d summary=%#v", extractRuns, renderRuns, summary) } } @@ -1019,14 +1019,14 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) { store := &manifest.LocalStore{} manifestPath := manifestPathFor(cfg) seed := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) - seed.MarkStageSucceeded("optional", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{{ + seed.MarkStageSucceeded("extract", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{{ Kind: "old_output", SourceID: "narratio.example.old", LocalPath: "artifacts/old.json", }}) - seed.Stages["optional"].Logs = []string{"old.log"} - seed.Stages["optional"].GeneratedConfigs = []string{"old.yml"} - seed.Stages["optional"].Metadata = map[string]any{"old": true} + seed.Stages["extract"].Logs = []string{"old.log"} + seed.Stages["extract"].GeneratedConfigs = []string{"old.yml"} + seed.Stages["extract"].Metadata = map[string]any{"old": true} if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } @@ -1038,7 +1038,7 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) { optionalRuns := 0 stages := []stage.Stage{ resultStage{ - name: "optional", + name: "extract", runs: &optionalRuns, order: &order, result: &stage.StageResult{ @@ -1049,16 +1049,16 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) { Metadata: map[string]any{"enabled": false}, }, }, - resultStage{name: "later", order: &order, result: &stage.StageResult{}}, + resultStage{name: "analyze", order: &order, result: &stage.StageResult{}}, } summary, err := executeStages(context.Background(), cfg, stages, RunOptions{Force: true}) if err != nil { t.Fatalf("executeStages() error = %v", err) } - if strings.Join(order, ",") != "optional,later" { - t.Fatalf("execution order = %v, want optional then later", order) + if strings.Join(order, ",") != "extract,analyze" { + t.Fatalf("execution order = %v, want extract then analyze", order) } - if len(summary.Executed) != 2 || len(summary.Skipped) != 1 || summary.Skipped[0] != "optional" { + if len(summary.Executed) != 2 || len(summary.Skipped) != 1 || summary.Skipped[0] != "extract" { t.Fatalf("summary = %#v, want optional executed and self-skipped before later", summary) } @@ -1066,7 +1066,7 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) { if err != nil { t.Fatalf("Load() session manifest error = %v", err) } - selfSkipped := sessionManifest.Stages["optional"] + selfSkipped := sessionManifest.Stages["extract"] if selfSkipped == nil || selfSkipped.Status != manifest.StatusSkipped { t.Fatalf("optional stage = %#v, want skipped", selfSkipped) } @@ -1081,7 +1081,7 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) { selfSkipped.Metadata["enabled"] != false || selfSkipped.Metadata["old"] != nil { t.Fatalf("optional result details = %#v, want current bounded diagnostics and metadata", selfSkipped) } - if later := sessionManifest.Stages["later"]; later == nil || later.Status != manifest.StatusSucceeded { + if later := sessionManifest.Stages["analyze"]; later == nil || later.Status != manifest.StatusSucceeded { t.Fatalf("later stage = %#v, want succeeded", later) } @@ -1089,7 +1089,7 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) { if err != nil { t.Fatalf("LoadRun() error = %v", err) } - runStage := runManifest.Stages["optional"] + runStage := runManifest.Stages["extract"] if runStage == nil || runStage.Action != manifest.RunStageActionRun || runStage.Status != manifest.StatusSkipped { t.Fatalf("run optional stage = %#v, want run action with skipped status", runStage) } @@ -1108,7 +1108,7 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) { func TestExecuteStagesRejectsSkippedResultWithOutputs(t *testing.T) { cfg := testConfig(t) - invalid := resultStage{name: "optional", result: &stage.StageResult{ + invalid := resultStage{name: "extract", result: &stage.StageResult{ Disposition: stage.StageDispositionSkipped, SkipReason: "integration_disabled", Outputs: []artifacts.Ref{{Kind: "unexpected"}}, @@ -1129,7 +1129,7 @@ func TestExecuteStagesRejectsSkippedResultWithOutputs(t *testing.T) { if loadErr != nil { t.Fatalf("Load() session manifest error = %v", loadErr) } - if got := loaded.Stages["optional"]; got == nil || got.Status != manifest.StatusFailed { + if got := loaded.Stages["extract"]; got == nil || got.Status != manifest.StatusFailed { t.Fatalf("optional stage = %#v, want failed", got) } } diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index ad0031d..d033c62 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -66,8 +66,8 @@ func All() []Stage { polishStage{}, normalizeStage{}, trimStage{}, - extractStage{}, renderStage{}, + extractStage{}, analyzeStage{}, publishStage{}, placeholderStage{name: "notify"},