diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index 2cbb054..8c5cbf8 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -156,7 +156,9 @@ durable; callers must reload it before retrying. The application runner marks an executing stage running and then succeeded or failed in both manifests, persisting each transition. On success it records -outputs, logs, generated configuration references, and metadata. Artifact +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. @@ -165,7 +167,8 @@ 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 +logs, generated configuration references, metadata, and semantic fingerprint. +Failed and skipped transitions enforce the same clearing rule directly, while success repopulates only fields returned by the new result. Marking a record stale does not clear those details because resume validation and diagnosis may still require them @@ -184,12 +187,21 @@ session-stage record along with older logs, generated configuration references, and metadata, then applies any bounded details from the current skip and continues. This self-skip is distinct from deciding not to execute an already-succeeded stage and is reconsidered on later runs. Skipped results -cannot contain outputs. +cannot contain outputs. An intentional self-skip records the current semantic +fingerprint because it is a completed, reusable stage result; failed or +interrupted work never promotes one. When an already-succeeded stage is skipped, the invocation run manifest records the `skip` action and reason. The session manifest deliberately retains its existing succeeded record because it remains the cross-invocation progress -authority. Extraction and analyze have resume validators and may reject an +authority. If a stage supplies semantic configuration evidence, reuse first +requires the persisted positive schema version and lowercase SHA-256 digest to +match the current resolved stage semantics. Missing legacy evidence, malformed +evidence, or a mismatch makes the stage and its fixed transitive dependents +stale. The invocation skip copies the matched fingerprint for provenance but +does not rewrite session authority. The existing stage-specific resume +validator runs only after this semantic check succeeds; both checks are +required. Extraction and analyze have resume validators and may reject an otherwise eligible skip when their selected durable evidence is obsolete; the runner marks the aggregate record stale and executes it. Analyze's validator can still accept a partial selection when only unrelated artifact records are @@ -198,6 +210,14 @@ stale. Session manifest is the authoritative stage-progress ledger across invocations. Run manifest is invocation-scoped audit state. +`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 or invocation state. The shared helper +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 stage coverage is owned +by the focused stage documents as providers are added. + Before an explicitly bounded execution starts after `prepare`, the application reads the session manifest and accepts only `succeeded` or `skipped` for every excluded canonical prefix stage. The first other status or absent record fails @@ -228,6 +248,9 @@ where a durable running record can require operator interpretation. ## Invariants - stage resume/skip decisions are session-manifest driven. +- semantic fingerprint comparison precedes stage-specific resume validation. +- only successful and intentional-skipped results promote current semantic + evidence; invocation reuse copies evidence without replacing session state. - running, failed, and self-skipped stages do not retain result payloads from an earlier success. - stale stages retain prior details until replacement execution starts. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 89b0671..61e5d62 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -252,7 +252,7 @@ configuration behavior and one root-relative path base. ## Stage 3 — Versioned Stage Semantic Resume Framework -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/app/plan.go b/internal/app/plan.go index 1691471..12bb4f8 100644 --- a/internal/app/plan.go +++ b/internal/app/plan.go @@ -65,17 +65,19 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { return err } for _, selectedStage := range stages { + semanticConfig, err := currentStageSemanticConfig(selectedStage, stageEnv) + if err != nil { + return fmt.Errorf("plan: fingerprint semantic configuration for stage %q: %w", selectedStage.Name(), err) + } action := decideStageAction(selectedStage, model, request.Force) var validation *stage.ResumeValidation - if validator, ok := selectedStage.(stage.ResumeValidator); ok && - (action == stageActionSkip || selectedStage.Name() == "analyze") { - checked, validationErr := validator.ValidateResume(ctx, stageEnv, model) + if 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) } - checked = checked.Normalized() - validation = &checked - if action == stageActionSkip && !checked.Resumable { + validation = checked + if checked != nil && !checked.Resumable { at := time.Now().UTC() model.MarkStageStale(selectedStage.Name(), at, checked.Reason) if _, invalidationErr := invalidateDependentSucceededStagesWithReason( @@ -86,6 +88,16 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { action = stageActionRun } } + if action == stageActionRun && selectedStage.Name() == "analyze" { + if validator, ok := selectedStage.(stage.ResumeValidator); ok { + checked, validationErr := validator.ValidateResume(ctx, stageEnv, model) + if validationErr != nil { + return fmt.Errorf("plan: validate resume for stage %q: %w", selectedStage.Name(), validationErr) + } + checked = checked.Normalized() + validation = &checked + } + } if action == stageActionRun { runCount++ } else { @@ -100,7 +112,7 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { } } if action == stageActionRun { - if err := modelPlannedStageRun(model, selectedStage, cfg, request.Force); err != nil { + if err := modelPlannedStageRun(model, selectedStage, cfg, request.Force, semanticConfig); err != nil { return fmt.Errorf("plan: model stage %q: %w", selectedStage.Name(), err) } } @@ -129,7 +141,13 @@ func cloneManifestForPlan(source *manifest.Manifest, cfg *config.Config) (*manif return &cloned, nil } -func modelPlannedStageRun(model *manifest.Manifest, selectedStage stage.Stage, cfg *config.Config, force bool) error { +func modelPlannedStageRun( + model *manifest.Manifest, + selectedStage stage.Stage, + cfg *config.Config, + force bool, + semanticConfig *manifest.SemanticConfigFingerprint, +) error { prior := capturePriorStageOutcome(model, selectedStage.Name()) at := time.Now().UTC() model.MarkStageRunning(selectedStage.Name(), at) @@ -142,6 +160,7 @@ func modelPlannedStageRun(model *manifest.Manifest, selectedStage stage.Stage, c } 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, @@ -151,6 +170,7 @@ func modelPlannedStageRun(model *manifest.Manifest, selectedStage stage.Stage, c return nil } 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, diff --git a/internal/app/runner.go b/internal/app/runner.go index 81eb6d6..3e9e0b3 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -259,38 +259,42 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts for _, s := range stages { stageEnv.Force = opts.Force runNames = append(runNames, s.Name()) + semanticConfig, err := currentStageSemanticConfig(s, stageEnv) + if err != nil { + return nil, persistTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, + fmt.Errorf("fingerprint semantic configuration for stage %q: %w", s.Name(), err), + ) + } action := decideStageAction(s, m, opts.Force) if action == stageActionSkip { - if validator, ok := s.(stage.ResumeValidator); ok { - validation, err := validator.ValidateResume(ctx, stageEnv, m) - if err != nil { + validation, err := evaluateStageResume(ctx, s, stageEnv, m, semanticConfig) + if err != nil { + return nil, persistTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, + fmt.Errorf("validate resume for stage %q: %w", s.Name(), err), + ) + } + if validation != nil && !validation.Resumable { + staleAt := nowUTC() + m.MarkStageStale(s.Name(), staleAt, validation.Reason) + if _, err := invalidateDependentSucceededStagesWithReason( + m, s.Name(), staleAt, staleReasonNotResumable, + ); err != nil { return nil, persistTerminalFailure( ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, - fmt.Errorf("validate resume for stage %q: %w", s.Name(), err), + fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", s.Name(), err), ) } - validation = validation.Normalized() - if !validation.Resumable { - staleAt := nowUTC() - m.MarkStageStale(s.Name(), staleAt, validation.Reason) - 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, - 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 + if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { + return nil, persistTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, + 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 } } @@ -299,6 +303,7 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts skipAt := nowUTC() runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt) runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded") + setRunStageSemanticConfig(runManifest, s.Name(), semanticConfig) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, persistTerminalFailure( ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, @@ -380,6 +385,7 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts 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, persistTerminalFailure( @@ -396,6 +402,7 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts } runManifest.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason) applyStageResultToRunManifest(runManifest, s.Name(), result) + setRunStageSemanticConfig(runManifest, s.Name(), semanticConfig) identity.applyToRunManifest(runManifest, manifestPath) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, persistTerminalFailure( @@ -418,6 +425,7 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts m.MarkStageSucceeded(s.Name(), succeededAt, sessionOutputs) applyAnalyzeProjection(m, runManifest, 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, persistTerminalFailure( @@ -445,6 +453,7 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts } runManifest.MarkStageSucceeded(s.Name(), succeededAt, runOutputs) applyStageResultToRunManifest(runManifest, s.Name(), result) + setRunStageSemanticConfig(runManifest, s.Name(), semanticConfig) identity.applyToRunManifest(runManifest, manifestPath) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, persistTerminalFailure( diff --git a/internal/app/semantic_resume.go b/internal/app/semantic_resume.go new file mode 100644 index 0000000..b9e9f7d --- /dev/null +++ b/internal/app/semantic_resume.go @@ -0,0 +1,95 @@ +package app + +import ( + "context" + "fmt" + + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +func currentStageSemanticConfig(selected stage.Stage, env *stage.Env) (*manifest.SemanticConfigFingerprint, error) { + provider, ok := selected.(stage.SemanticConfigFingerprinter) + if !ok { + return nil, nil + } + record, err := provider.SemanticConfigFingerprint(env) + if err != nil { + return nil, err + } + if err := record.Validate(); err != nil { + return nil, fmt.Errorf("invalid current semantic configuration fingerprint: %w", err) + } + return cloneSemanticConfig(&record), nil +} + +func validateStageSemanticResume(selected stage.Stage, m *manifest.Manifest, current *manifest.SemanticConfigFingerprint) stage.ResumeValidation { + if current == nil { + return stage.Resumable() + } + var persisted *manifest.SemanticConfigFingerprint + if m != nil && m.Stages != nil && m.Stages[selected.Name()] != nil { + persisted = m.Stages[selected.Name()].SemanticConfig + } + if persisted == nil { + return stage.NonResumable("semantic configuration evidence is missing; rerun the stage to establish current evidence") + } + if err := persisted.Validate(); err != nil { + return stage.NonResumable("persisted semantic configuration evidence is malformed; rerun the stage") + } + if persisted.Version != current.Version { + return stage.NonResumable("semantic configuration contract version changed; rerun the stage") + } + if persisted.Digest != current.Digest { + return stage.NonResumable("result-affecting configuration changed; rerun the stage") + } + return stage.Resumable() +} + +func evaluateStageResume( + ctx context.Context, + selected stage.Stage, + env *stage.Env, + m *manifest.Manifest, + current *manifest.SemanticConfigFingerprint, +) (*stage.ResumeValidation, error) { + semantic := validateStageSemanticResume(selected, m, current).Normalized() + if !semantic.Resumable { + return &semantic, nil + } + validator, ok := selected.(stage.ResumeValidator) + if !ok { + if current == nil { + return nil, nil + } + return &semantic, nil + } + validation, err := validator.ValidateResume(ctx, env, m) + if err != nil { + return nil, err + } + validation = validation.Normalized() + return &validation, nil +} + +func setSessionStageSemanticConfig(m *manifest.Manifest, name string, value *manifest.SemanticConfigFingerprint) { + if m == nil || m.Stages == nil || m.Stages[name] == nil { + return + } + m.Stages[name].SemanticConfig = cloneSemanticConfig(value) +} + +func setRunStageSemanticConfig(m *manifest.RunManifest, name string, value *manifest.SemanticConfigFingerprint) { + if m == nil || m.Stages == nil || m.Stages[name] == nil { + return + } + m.Stages[name].SemanticConfig = cloneSemanticConfig(value) +} + +func cloneSemanticConfig(value *manifest.SemanticConfigFingerprint) *manifest.SemanticConfigFingerprint { + if value == nil { + return nil + } + copy := *value + return © +} diff --git a/internal/app/semantic_resume_test.go b/internal/app/semantic_resume_test.go new file mode 100644 index 0000000..41bfffb --- /dev/null +++ b/internal/app/semantic_resume_test.go @@ -0,0 +1,305 @@ +package app + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +type semanticStage struct { + name string + fingerprint manifest.SemanticConfigFingerprint + fingerprintErr error + result *stage.StageResult + runErr error + runs *int +} + +func (s semanticStage) Name() string { return s.name } +func (s semanticStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + if s.runs != nil { + *s.runs++ + } + return s.result, s.runErr +} +func (s semanticStage) SemanticConfigFingerprint(_ *stage.Env) (manifest.SemanticConfigFingerprint, error) { + return s.fingerprint, s.fingerprintErr +} + +type semanticResumeCheckingStage struct { + semanticStage + validation stage.ResumeValidation + validationCalls *int +} + +func (s semanticResumeCheckingStage) ValidateResume(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (stage.ResumeValidation, error) { + if s.validationCalls != nil { + *s.validationCalls++ + } + return s.validation, nil +} + +func TestSemanticResumeComparison(t *testing.T) { + current := semanticFingerprint(1, "a") + selected := semanticStage{name: "render", fingerprint: current} + for _, test := range []struct { + name string + persisted *manifest.SemanticConfigFingerprint + resumable bool + want string + }{ + {name: "matching", persisted: ¤t, resumable: true}, + {name: "missing", want: "missing"}, + {name: "version", persisted: fingerprintPointer(semanticFingerprint(2, "a")), want: "version"}, + {name: "digest", persisted: fingerprintPointer(semanticFingerprint(1, "b")), want: "configuration changed"}, + {name: "malformed", persisted: &manifest.SemanticConfigFingerprint{Version: 1, Digest: "bad"}, want: "malformed"}, + } { + t.Run(test.name, func(t *testing.T) { + model := manifest.New("session", time.Now().UTC()) + model.MarkStageSucceeded("render", time.Now().UTC(), nil) + model.Stages["render"].SemanticConfig = test.persisted + got := validateStageSemanticResume(selected, model, ¤t) + if got.Resumable != test.resumable || (test.want != "" && !strings.Contains(got.Reason, test.want)) { + t.Fatalf("validation = %#v, want resumable=%v reason %q", got, test.resumable, test.want) + } + }) + } +} + +func TestExecuteStagesSemanticEvidenceLifecycle(t *testing.T) { + for _, test := range []struct { + name string + persisted *manifest.SemanticConfigFingerprint + force bool + wantRuns int + wantSkipped int + }{ + {name: "matching skips", persisted: fingerprintPointer(semanticFingerprint(1, "current")), wantSkipped: 1}, + {name: "legacy missing reruns", wantRuns: 1}, + {name: "version reruns", persisted: fingerprintPointer(semanticFingerprint(2, "current")), wantRuns: 1}, + {name: "digest reruns", persisted: fingerprintPointer(semanticFingerprint(1, "old")), wantRuns: 1}, + {name: "force reruns matching", persisted: fingerprintPointer(semanticFingerprint(1, "current")), force: true, wantRuns: 1}, + } { + t.Run(test.name, func(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.MarkStageSucceeded("render", time.Now().UTC(), nil) + seed.Stages["render"].SemanticConfig = test.persisted + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatal(err) + } + runs := 0 + current := semanticFingerprint(1, "current") + candidate := semanticStage{name: "render", fingerprint: current, result: &stage.StageResult{}, runs: &runs} + summary, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{Force: test.force}) + if err != nil { + t.Fatal(err) + } + if runs != test.wantRuns || len(summary.Skipped) != test.wantSkipped { + t.Fatalf("runs=%d summary=%#v", runs, summary) + } + loaded, err := store.Load(context.Background(), summary.ManifestPath) + if err != nil { + t.Fatal(err) + } + if loaded.Stages["render"].SemanticConfig == nil || !loaded.Stages["render"].SemanticConfig.Equal(current) { + t.Fatalf("session evidence = %#v", loaded.Stages["render"].SemanticConfig) + } + run, err := store.LoadRun(context.Background(), summary.RunManifestPath) + if err != nil { + t.Fatal(err) + } + if run.Stages["render"].SemanticConfig == nil || !run.Stages["render"].SemanticConfig.Equal(current) { + t.Fatalf("run evidence = %#v", run.Stages["render"].SemanticConfig) + } + }) + } +} + +func TestSemanticEvidenceRequiresBothChecksAndNeverPromotesFailure(t *testing.T) { + t.Run("semantic mismatch precedes resume validator", func(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.MarkStageSucceeded("render", time.Now().UTC(), nil) + seed.Stages["render"].SemanticConfig = fingerprintPointer(semanticFingerprint(1, "old")) + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatal(err) + } + runs, validations := 0, 0 + candidate := semanticResumeCheckingStage{ + semanticStage: semanticStage{name: "render", fingerprint: semanticFingerprint(1, "new"), result: &stage.StageResult{}, runs: &runs}, + validation: stage.Resumable(), validationCalls: &validations, + } + if _, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}); err != nil { + t.Fatal(err) + } + if runs != 1 || validations != 0 { + t.Fatalf("runs=%d validations=%d", runs, validations) + } + }) + + t.Run("matching semantic evidence still requires validator", func(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + current := semanticFingerprint(1, "same") + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.MarkStageSucceeded("render", time.Now().UTC(), nil) + seed.Stages["render"].SemanticConfig = ¤t + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatal(err) + } + runs, validations := 0, 0 + candidate := semanticResumeCheckingStage{ + semanticStage: semanticStage{name: "render", fingerprint: current, result: &stage.StageResult{}, runs: &runs}, + validation: stage.NonResumable("output changed"), validationCalls: &validations, + } + if _, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}); err != nil { + t.Fatal(err) + } + if runs != 1 || validations != 1 { + t.Fatalf("runs=%d validations=%d", runs, validations) + } + }) + + t.Run("failed execution has no promoted evidence", func(t *testing.T) { + cfg := testConfig(t) + runs := 0 + candidate := semanticStage{ + name: "render", fingerprint: semanticFingerprint(1, "new"), runErr: errors.New("render failed"), runs: &runs, + } + _, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}) + if err == nil { + t.Fatal("executeStages() error = nil") + } + loaded, loadErr := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) + if loadErr != nil { + t.Fatal(loadErr) + } + if loaded.Stages["render"].SemanticConfig != nil || runs != 1 { + t.Fatalf("failed evidence=%#v runs=%d", loaded.Stages["render"].SemanticConfig, runs) + } + }) +} + +func TestIntentionalStageSkipPersistsSemanticEvidence(t *testing.T) { + cfg := testConfig(t) + current := semanticFingerprint(1, "disabled") + runs := 0 + candidate := semanticStage{ + name: "render", fingerprint: current, runs: &runs, + result: &stage.StageResult{Disposition: stage.StageDispositionSkipped, SkipReason: "render disabled"}, + } + summary, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + store := &manifest.LocalStore{} + session, err := store.Load(context.Background(), summary.ManifestPath) + if err != nil { + t.Fatal(err) + } + run, err := store.LoadRun(context.Background(), summary.RunManifestPath) + if err != nil { + t.Fatal(err) + } + if runs != 1 || session.Stages["render"].SemanticConfig == nil || run.Stages["render"].SemanticConfig == nil || + !session.Stages["render"].SemanticConfig.Equal(current) || !run.Stages["render"].SemanticConfig.Equal(current) { + t.Fatalf("runs=%d session=%#v run=%#v", runs, session.Stages["render"], run.Stages["render"]) + } +} + +func TestInvalidCurrentSemanticEvidenceStopsBeforeExecution(t *testing.T) { + cfg := testConfig(t) + runs := 0 + candidate := semanticStage{ + name: "render", fingerprint: manifest.SemanticConfigFingerprint{Version: 1, Digest: "invalid"}, runs: &runs, + } + _, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{Force: true}) + if err == nil || !strings.Contains(err.Error(), "invalid current semantic configuration fingerprint") || runs != 0 { + t.Fatalf("error=%v runs=%d", err, runs) + } +} + +func TestReadOnlySemanticResumeDecisionMutatesOnlyPlanModel(t *testing.T) { + cfg := testConfig(t) + original := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + original.MarkStageSucceeded("render", time.Now().UTC(), nil) + original.MarkStageSucceeded("extract", time.Now().UTC(), nil) + original.MarkStageSucceeded("analyze", time.Now().UTC(), nil) + original.Stages["render"].SemanticConfig = fingerprintPointer(semanticFingerprint(1, "old")) + model, err := cloneManifestForPlan(original, cfg) + if err != nil { + t.Fatal(err) + } + selected := semanticStage{name: "render", fingerprint: semanticFingerprint(1, "new")} + current, err := currentStageSemanticConfig(selected, &stage.Env{Config: cfg}) + if err != nil { + t.Fatal(err) + } + validation, err := evaluateStageResume(context.Background(), selected, &stage.Env{Config: cfg}, model, current) + if err != nil { + t.Fatal(err) + } + if validation == nil || validation.Resumable { + t.Fatalf("validation = %#v, want planned rerun", validation) + } + at := time.Now().UTC() + model.MarkStageStale("render", at, validation.Reason) + if _, err := invalidateDependentSucceededStagesWithReason(model, "render", at, staleReasonNotResumable); err != nil { + t.Fatal(err) + } + if model.Stages["render"].Status != manifest.StatusStale || model.Stages["analyze"].Status != manifest.StatusStale { + t.Fatalf("model render=%q analyze=%q", model.Stages["render"].Status, model.Stages["analyze"].Status) + } + if original.Stages["render"].Status != manifest.StatusSucceeded || original.Stages["analyze"].Status != manifest.StatusSucceeded { + t.Fatalf("authoritative manifest mutated: render=%q analyze=%q", original.Stages["render"].Status, original.Stages["analyze"].Status) + } +} + +func TestSemanticMismatchInvalidatesOnlyFixedDependents(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + for _, name := range []string{"render", "extract", "analyze", "publish", "notify"} { + seed.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + seed.Stages["render"].SemanticConfig = fingerprintPointer(semanticFingerprint(1, "old")) + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatal(err) + } + runs := 0 + candidate := semanticStage{name: "render", fingerprint: semanticFingerprint(1, "new"), result: &stage.StageResult{}, runs: &runs} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}); err != nil { + t.Fatal(err) + } + loaded, err := store.Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatal(err) + } + if loaded.Stages["render"].Status != manifest.StatusSucceeded || loaded.Stages["extract"].Status != manifest.StatusSucceeded { + t.Fatalf("render=%q extract=%q", loaded.Stages["render"].Status, loaded.Stages["extract"].Status) + } + for _, name := range []string{"analyze", "publish", "notify"} { + if loaded.Stages[name].Status != manifest.StatusStale { + t.Fatalf("%s status = %q, want stale", name, loaded.Stages[name].Status) + } + } +} + +func semanticFingerprint(version int, seed string) manifest.SemanticConfigFingerprint { + digest := sha256.Sum256([]byte(seed)) + return manifest.SemanticConfigFingerprint{Version: version, Digest: hex.EncodeToString(digest[:])} +} + +func fingerprintPointer(value manifest.SemanticConfigFingerprint) *manifest.SemanticConfigFingerprint { + return &value +} diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index a927a94..2de8425 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -55,6 +55,7 @@ type StageRecord struct { GeneratedConfigs []string `json:"generated_configs,omitempty"` Error *ErrorRecord `json:"error,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` + SemanticConfig *SemanticConfigFingerprint `json:"semantic_config,omitempty"` AnalyzeStateVersion int `json:"analyze_state_version,omitempty"` AnalyzeArtifacts map[string]AnalyzeArtifactRecord `json:"analyze_artifacts,omitempty"` } @@ -177,6 +178,7 @@ func (s *StageRecord) clearResultDetails() { s.Logs = nil s.GeneratedConfigs = nil s.Metadata = nil + s.SemanticConfig = nil } func (m *Manifest) ensureStage(name string, at time.Time) *StageRecord { diff --git a/internal/manifest/run_manifest.go b/internal/manifest/run_manifest.go index c0496e6..aff8ea2 100644 --- a/internal/manifest/run_manifest.go +++ b/internal/manifest/run_manifest.go @@ -34,6 +34,7 @@ type RunStageRecord struct { GeneratedConfigs []string `json:"generated_configs,omitempty"` Error *ErrorRecord `json:"error,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` + SemanticConfig *SemanticConfigFingerprint `json:"semantic_config,omitempty"` AnalyzeStateVersion int `json:"analyze_state_version,omitempty"` AnalyzeArtifacts map[string]AnalyzeArtifactRecord `json:"analyze_artifacts,omitempty"` } diff --git a/internal/manifest/semantic_fingerprint.go b/internal/manifest/semantic_fingerprint.go new file mode 100644 index 0000000..6959ec5 --- /dev/null +++ b/internal/manifest/semantic_fingerprint.go @@ -0,0 +1,32 @@ +package manifest + +import ( + "fmt" + "regexp" +) + +var lowercaseSHA256Pattern = regexp.MustCompile(`^[0-9a-f]{64}$`) + +// SemanticConfigFingerprint identifies the versioned, result-affecting +// configuration observed by one pipeline stage. +type SemanticConfigFingerprint struct { + Version int `json:"version"` + Digest string `json:"digest"` +} + +// Validate checks the durable fingerprint contract. +func (f SemanticConfigFingerprint) Validate() error { + if f.Version <= 0 { + return fmt.Errorf("version must be positive") + } + if !lowercaseSHA256Pattern.MatchString(f.Digest) { + return fmt.Errorf("digest must be a lowercase SHA-256 value") + } + return nil +} + +// Equal reports whether two valid fingerprint records identify the same +// semantic configuration contract and payload. +func (f SemanticConfigFingerprint) Equal(other SemanticConfigFingerprint) bool { + return f.Version == other.Version && f.Digest == other.Digest +} diff --git a/internal/manifest/semantic_fingerprint_test.go b/internal/manifest/semantic_fingerprint_test.go new file mode 100644 index 0000000..3fcfd38 --- /dev/null +++ b/internal/manifest/semantic_fingerprint_test.go @@ -0,0 +1,80 @@ +package manifest + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +const testSemanticDigest = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + +func TestSemanticConfigFingerprintRoundTripsAndLegacyRecordsRemainReadable(t *testing.T) { + ctx := context.Background() + store := &LocalStore{} + dir := t.TempDir() + fingerprint := &SemanticConfigFingerprint{Version: 3, Digest: testSemanticDigest} + + session := New("session", time.Now().UTC()) + session.MarkStageSucceeded("prepare", time.Now().UTC(), nil) + session.Stages["prepare"].SemanticConfig = fingerprint + sessionPath := filepath.Join(dir, "session.json") + if err := store.Save(ctx, sessionPath, session); err != nil { + t.Fatal(err) + } + loadedSession, err := store.Load(ctx, sessionPath) + if err != nil { + t.Fatal(err) + } + if loadedSession.Stages["prepare"].SemanticConfig == nil || !loadedSession.Stages["prepare"].SemanticConfig.Equal(*fingerprint) { + t.Fatalf("session fingerprint = %#v", loadedSession.Stages["prepare"].SemanticConfig) + } + + run := NewRun("session", "campaign", "run", false, []string{"prepare"}, time.Now().UTC()) + run.MarkStageSucceeded("prepare", time.Now().UTC(), nil) + run.Stages["prepare"].SemanticConfig = fingerprint + runPath := filepath.Join(dir, "run.json") + if err := store.SaveRun(ctx, runPath, run); err != nil { + t.Fatal(err) + } + loadedRun, err := store.LoadRun(ctx, runPath) + if err != nil { + t.Fatal(err) + } + if loadedRun.Stages["prepare"].SemanticConfig == nil || !loadedRun.Stages["prepare"].SemanticConfig.Equal(*fingerprint) { + t.Fatalf("run fingerprint = %#v", loadedRun.Stages["prepare"].SemanticConfig) + } + + legacyPath := filepath.Join(dir, "legacy.json") + legacy := `{"session_id":"session","created_at":"2026-01-01T00:00:00Z","updated_at":"2026-01-01T00:00:00Z","stages":{"prepare":{"name":"prepare","status":"succeeded","created_at":"2026-01-01T00:00:00Z","updated_at":"2026-01-01T00:00:00Z"}}}` + if err := os.WriteFile(legacyPath, []byte(legacy), 0o644); err != nil { + t.Fatal(err) + } + legacyManifest, err := store.Load(ctx, legacyPath) + if err != nil { + t.Fatal(err) + } + if legacyManifest.Stages["prepare"].SemanticConfig != nil { + t.Fatalf("legacy fingerprint = %#v, want nil", legacyManifest.Stages["prepare"].SemanticConfig) + } +} + +func TestSemanticConfigFingerprintValidation(t *testing.T) { + for _, test := range []struct { + name string + value SemanticConfigFingerprint + want string + }{ + {name: "zero version", value: SemanticConfigFingerprint{Digest: testSemanticDigest}, want: "positive"}, + {name: "uppercase digest", value: SemanticConfigFingerprint{Version: 1, Digest: strings.ToUpper(testSemanticDigest)}, want: "lowercase"}, + {name: "short digest", value: SemanticConfigFingerprint{Version: 1, Digest: "abcd"}, want: "lowercase"}, + } { + t.Run(test.name, func(t *testing.T) { + if err := test.value.Validate(); err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Validate() error = %v, want %q", err, test.want) + } + }) + } +} diff --git a/internal/manifest/store.go b/internal/manifest/store.go index 48a3bca..b6aa58f 100644 --- a/internal/manifest/store.go +++ b/internal/manifest/store.go @@ -275,6 +275,11 @@ func validateLoadedManifest(m *Manifest) error { if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil { return fmt.Errorf("stages.%s: %w", name, err) } + if stage.SemanticConfig != nil { + if err := stage.SemanticConfig.Validate(); err != nil { + return fmt.Errorf("stages.%s.semantic_config: %w", name, err) + } + } } return nil @@ -343,6 +348,11 @@ func validateLoadedRunManifest(m *RunManifest) error { if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil { return fmt.Errorf("stages.%s: %w", name, err) } + if stage.SemanticConfig != nil { + if err := stage.SemanticConfig.Validate(); err != nil { + return fmt.Errorf("stages.%s.semantic_config: %w", name, err) + } + } } return nil diff --git a/internal/stage/semantic_fingerprint.go b/internal/stage/semantic_fingerprint.go new file mode 100644 index 0000000..0d21bbc --- /dev/null +++ b/internal/stage/semantic_fingerprint.go @@ -0,0 +1,86 @@ +package stage + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "reflect" + + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +// SemanticConfigFingerprinter is implemented by stages whose reusable result +// depends on Narratio-observable semantic configuration. +type SemanticConfigFingerprinter interface { + SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) +} + +// FingerprintSemanticConfig hashes deterministic JSON for a stage-owned typed +// semantic configuration value. Callers must exclude secrets and operational +// settings that cannot affect the canonical result. +func FingerprintSemanticConfig(version int, value any) (manifest.SemanticConfigFingerprint, error) { + record := manifest.SemanticConfigFingerprint{Version: version} + if version <= 0 { + return record, fmt.Errorf("semantic configuration fingerprint version must be positive") + } + typeOf := reflect.TypeOf(value) + if typeOf == nil { + return record, fmt.Errorf("semantic configuration value is nil") + } + valueOf := reflect.ValueOf(value) + if valueOf.Kind() == reflect.Pointer && valueOf.IsNil() { + return record, fmt.Errorf("semantic configuration value is nil") + } + for typeOf.Kind() == reflect.Pointer { + typeOf = typeOf.Elem() + } + if typeOf.Kind() != reflect.Struct { + return record, fmt.Errorf("semantic configuration value must be a typed struct, got %s", typeOf.Kind()) + } + if err := validateSemanticConfigType(typeOf, map[reflect.Type]bool{}); err != nil { + return record, err + } + payload, err := json.Marshal(value) + if err != nil { + return record, fmt.Errorf("encode semantic configuration: %w", err) + } + digest := sha256.Sum256(payload) + record.Digest = hex.EncodeToString(digest[:]) + return record, nil +} + +func validateSemanticConfigType(value reflect.Type, visiting map[reflect.Type]bool) error { + for value.Kind() == reflect.Pointer { + value = value.Elem() + } + if visiting[value] { + return nil + } + visiting[value] = true + defer delete(visiting, value) + switch value.Kind() { + case reflect.Interface: + return fmt.Errorf("semantic configuration structs cannot contain arbitrary interface values") + case reflect.Struct: + for index := 0; index < value.NumField(); index++ { + field := value.Field(index) + if field.PkgPath != "" || field.Tag.Get("json") == "-" { + continue + } + if err := validateSemanticConfigType(field.Type, visiting); err != nil { + return fmt.Errorf("semantic configuration field %s: %w", field.Name, err) + } + } + case reflect.Slice, reflect.Array: + return validateSemanticConfigType(value.Elem(), visiting) + case reflect.Map: + if value.Key().Kind() != reflect.String { + return fmt.Errorf("semantic configuration maps must use string keys") + } + return validateSemanticConfigType(value.Elem(), visiting) + case reflect.Chan, reflect.Func, reflect.UnsafePointer: + return fmt.Errorf("semantic configuration contains unsupported %s value", value.Kind()) + } + return nil +} diff --git a/internal/stage/semantic_fingerprint_test.go b/internal/stage/semantic_fingerprint_test.go new file mode 100644 index 0000000..c5c2446 --- /dev/null +++ b/internal/stage/semantic_fingerprint_test.go @@ -0,0 +1,61 @@ +package stage + +import ( + "strings" + "testing" +) + +type semanticFingerprintFixture struct { + Model string `json:"model"` + Modules []string `json:"modules"` + Enabled bool `json:"enabled"` +} + +func TestFingerprintSemanticConfigIsDeterministicAndTyped(t *testing.T) { + value := semanticFingerprintFixture{Model: "quality", Modules: []string{"speaker", "terms"}, Enabled: true} + first, err := FingerprintSemanticConfig(2, value) + if err != nil { + t.Fatal(err) + } + second, err := FingerprintSemanticConfig(2, &value) + if err != nil { + t.Fatal(err) + } + if !first.Equal(second) || first.Version != 2 || len(first.Digest) != 64 || first.Validate() != nil { + t.Fatalf("fingerprints = %#v / %#v", first, second) + } + changed := value + changed.Model = "testing" + different, err := FingerprintSemanticConfig(2, changed) + if err != nil { + t.Fatal(err) + } + if first.Equal(different) { + t.Fatalf("changed semantic value retained digest %q", first.Digest) + } +} + +func TestFingerprintSemanticConfigRejectsInvalidInputs(t *testing.T) { + for _, test := range []struct { + name string + version int + value any + want string + }{ + {name: "version", value: semanticFingerprintFixture{}, want: "positive"}, + {name: "nil", version: 1, value: nil, want: "nil"}, + {name: "typed nil", version: 1, value: (*semanticFingerprintFixture)(nil), want: "nil"}, + {name: "untyped map", version: 1, value: map[string]any{"model": "quality"}, want: "typed struct"}, + {name: "arbitrary field", version: 1, value: struct { + Values map[string]any `json:"values"` + }{Values: map[string]any{"model": "quality"}}, want: "arbitrary interface"}, + {name: "slice", version: 1, value: []string{"quality"}, want: "typed struct"}, + } { + t.Run(test.name, func(t *testing.T) { + _, err := FingerprintSemanticConfig(test.version, test.value) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("error = %v, want %q", err, test.want) + } + }) + } +}