diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index d0395ce..5ecadbc 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -82,11 +82,12 @@ The session-stage collection is the reconciled authority across invocations. These records remain analyze-owned data inside the fixed stage; they are not dynamic stages or generic subtasks. -The stage result contract has one analyze-specific projection boundary. On -success, the runner validates and deep-copies the complete reconciled session -collection. Aggregate session outputs are rebuilt in configured-key order from -current session records only. Ordinary stage outputs cannot accompany this -projection, so there is one source of artifact authority. +Analyze checkpoints one complete validated candidate collection after each +successful artifact and before exposing that artifact to later work. It saves a +detached manifest candidate and installs its version, collection, and updated +timestamp into live state only after the store reports success. Aggregate stage +outputs are ordinary result outputs rebuilt in configured-key order from current +session records only. Successful incremental execution replaces only evaluated artifact records and preserves valid unrelated current records. Rebuilt outputs are compared by @@ -95,31 +96,29 @@ the same recomputed fingerprint to remain current, while a changed identity removes output authority from every unselected transitive dependent by marking it stale. A partial analyze invocation can therefore succeed while unrelated configured records remain stale. Existing canonical files never create current -records without validated execution and projection. +records without validated execution and a successful checkpoint. Aggregate analyze status is deliberately coarser than this collection. Resume validation may skip a succeeded aggregate record when the selected artifact closure is current even if unrelated records are stale. Conversely, a stale aggregate record may cross the ordinary runner boundary and perform zero Scriptorium calls when reconciliation proves every selected artifact current; -the successful projection then restores the aggregate status. +ordinary successful stage completion then restores the aggregate status. -Analyze may return a projection together with an error. That restricted result -cannot carry ordinary outputs, skip state, aggregate logs, generated configs, -or metadata. The runner persists only the validated per-artifact collections, -then marks the aggregate analyze state failed and invalidates delivery -dependents conservatively. Unrelated current records survive because the -session projection is complete. A malformed projection is not applied, and a -failed session projection save restores the prior per-artifact authority before -terminal failure persistence. +On an artifact error, analyze installs a complete retry collection on the live +manifest before returning. The active record is failed without output, current +transitive dependents are stale, unrelated current records survive, and only +earlier checkpointed completions remain current. The runner's ordinary failure +path marks aggregate analyze failed, clears aggregate outputs, invalidates +delivery dependents, and saves that collection. A terminal session-save failure +is joined with the original adapter, validation, filesystem, cancellation, or +checkpoint cause. -The incremental executor constructs this restricted projection over its -scheduled artifact work. The active record is failed without output, -current transitive dependents are stale, unrelated current records survive, and -only earlier validated and materialized completions remain current in the -reconciled session collection. A terminal session-save failure is joined with -the original adapter, validation, or filesystem cause; a failed projection save -does not turn incidental canonical bytes into manifest authority. +If an artifact checkpoint fails, analyze rolls its working collection back to +the last committed collection before constructing the active failed record. +Newly materialized canonical bytes therefore remain incidental and cannot +become current authority. A later ordinary run reconciles the durable records, +reuses earlier current artifacts, and resumes with the failed or stale work. ## Remote Commit Manifest diff --git a/docs/internal/stage-analyze.md b/docs/internal/stage-analyze.md index 325e18a..8576149 100644 --- a/docs/internal/stage-analyze.md +++ b/docs/internal/stage-analyze.md @@ -89,9 +89,9 @@ Supported source families: rebuilt when required, while unrelated disabled artifacts are excluded. - the work plan carries explicit targets, prerequisite-only work, deterministic execution and reuse lists, invalidated and removed records, and a cloned - projected record collection. Valid unrelated configured records survive the - projection, removed records are omitted, and incidental files never become - current without manifest-backed regeneration. + reconciled starting collection. Valid unrelated configured records survive + reconciliation, removed records are omitted, and incidental files never + become current without manifest-backed regeneration. - implements aggregate resume validation by running the same read-only catalog, fingerprint reconciliation, and work planner used by execution. A succeeded aggregate record is reusable exactly when the selected closure schedules no @@ -112,15 +112,16 @@ Supported source families: - records each successful artifact's freshly computed fingerprint, canonical relative output path, contract, checksum, size, producer run ID, bounded Scriptorium provenance, logs, and generated configuration references in the - analyze-owned projection. + analyze-owned manifest collection. - preserves valid unrelated current records during partial execution. If a rebuilt output's bytes and contract are unchanged, unselected dependents may remain current. If that semantic identity changes, unselected transitive dependents become stale without being executed; dependents included in the invocation are evaluated in dependency order instead. -- returns one complete reconciled session collection. The runner reconstructs - aggregate session outputs from every current session record; unrelated stale - records do not make an otherwise successful partial invocation fail. +- returns aggregate outputs from every current session record in normalized-key + order after the complete reconciled collection has been checkpointed; + unrelated stale records do not make an otherwise successful partial + invocation fail. - resolves previous-session sources from local `previous/` cache only. - runs optional render-debug, then artifact execution. - validates non-empty output files and materializes canonical outputs. @@ -135,20 +136,20 @@ Supported source families: guidance. - dependency cycles or unavailable required dependencies fail. - adapter validation failures fail stage. -- a scheduled artifact failure returns the restricted analyze-state projection +- a scheduled artifact failure installs a retry collection on the live manifest with the active artifact marked `failed`, a bounded error, and no output authority. Current transitive dependents become stale without execution. - earlier artifacts remain current only after their run-local output passed - validation and canonical materialization; unattempted later artifacts retain - their reconciled session state. + validation, canonical materialization, and a successful complete-collection + manifest checkpoint; unattempted later artifacts retain their reconciled + session state. - unrelated current records survive a partial failure. Old canonical bytes for - the failed artifact and newly materialized bytes whose projection cannot be - persisted are incidental, not current evidence. -- the runner applies one valid partial projection at the existing end-of-stage - save boundary before it marks aggregate analyze failed and invalidates publish - and notify through the application dependency relation. Projection-persistence - errors retain the last durable per-artifact authority and are joined with the - original failure context. Per-artifact checkpoint persistence is future work. + the failed artifact and newly materialized bytes whose checkpoint fails are + incidental, not current evidence. +- the runner's ordinary error path marks aggregate analyze failed, clears its + aggregate outputs, invalidates publish and notify through the application + dependency relation, and persists the live retry collection. Terminal-save + errors are joined with the original artifact or checkpoint failure. ## Invariants diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 0ceae17..f18b9b4 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -318,7 +318,7 @@ stop using the end-of-stage projection transport for real analyze execution. ## Stage 4 — Complete Failure Recovery And Remove Projection Machinery -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/app/analyze_checkpoint_recovery_test.go b/internal/app/analyze_checkpoint_recovery_test.go new file mode 100644 index 0000000..67b5d50 --- /dev/null +++ b/internal/app/analyze_checkpoint_recovery_test.go @@ -0,0 +1,329 @@ +package app + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +func TestExecuteStagesRetriesFromLastAnalyzeCheckpoint(t *testing.T) { + cfg := testConfig(t) + cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{Artifacts: checkpointRecoveryArtifacts("original")} + store := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.Campaign = cfg.Session.Campaign + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim"} { + seed.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + saveBoundedManifest(t, cfg, seed) + analyze := mustSelectStage(t, "analyze") + + baselineRunner := &analyzeCheckpointRunner{} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{analyze}, RunOptions{ + Env: &Env{ManifestStore: store, Scriptorium: baselineRunner}, + }); err != nil { + t.Fatalf("baseline executeStages() error = %v", err) + } + baseline, err := store.Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatalf("load baseline manifest: %v", err) + } + baseline.MarkStageSucceeded("publish", time.Now().UTC(), nil) + baseline.MarkStageSucceeded("notify", time.Now().UTC(), nil) + if err := store.Save(context.Background(), manifestPathFor(cfg), baseline); err != nil { + t.Fatalf("save delivery baseline: %v", err) + } + upstream := cloneStageRecords(baseline, "prepare", "transcribe", "merge", "polish", "normalize", "trim") + unrelated := baseline.Stages["analyze"].AnalyzeArtifacts["unrelated"] + + cfg.Pipeline.Scriptorium.Artifacts = checkpointRecoveryArtifacts("revised") + failure := errors.New("injected quest-log failure") + failingRunner := &analyzeCheckpointRunner{FailPrompt: "quest-log-revised", Err: failure} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{analyze}, RunOptions{ + Env: &Env{ManifestStore: store, Scriptorium: failingRunner}, + }); !errors.Is(err, failure) { + t.Fatalf("executeStages() error = %v, want quest-log failure", err) + } + if want := []string{"chronicle-revised", "encounters-revised", "quest-log-revised"}; !reflect.DeepEqual(failingRunner.Calls, want) { + t.Fatalf("failed-run calls = %#v, want %#v", failingRunner.Calls, want) + } + + failed := loadAnalyzeRecoveryManifest(t, store, cfg) + failedAnalyze := failed.Stages["analyze"] + if failedAnalyze.Status != manifest.StatusFailed || len(failedAnalyze.Outputs) != 0 { + t.Fatalf("aggregate analyze = %#v, want failed without outputs", failedAnalyze) + } + assertAnalyzeArtifactStatus(t, failedAnalyze, "chronicle", manifest.AnalyzeArtifactCurrent) + assertAnalyzeArtifactStatus(t, failedAnalyze, "encounters", manifest.AnalyzeArtifactCurrent) + assertAnalyzeArtifactStatus(t, failedAnalyze, "quest_log", manifest.AnalyzeArtifactFailed) + assertAnalyzeArtifactStatus(t, failedAnalyze, "player_handout", manifest.AnalyzeArtifactStale) + if got := failedAnalyze.AnalyzeArtifacts["unrelated"]; !reflect.DeepEqual(got, unrelated) { + t.Fatalf("unrelated artifact = %#v, want preserved %#v", got, unrelated) + } + for _, name := range []string{"publish", "notify"} { + if got := failed.Stages[name]; got == nil || got.Status != manifest.StatusStale { + t.Fatalf("%s stage = %#v, want stale", name, got) + } + } + assertStageRecordsUnchanged(t, failed, upstream) + + retryRunner := &analyzeCheckpointRunner{} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{analyze}, RunOptions{ + Env: &Env{ManifestStore: store, Scriptorium: retryRunner}, + }); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + if want := []string{"quest-log-revised", "player-handout-revised"}; !reflect.DeepEqual(retryRunner.Calls, want) { + t.Fatalf("retry calls = %#v, want %#v", retryRunner.Calls, want) + } + + completed := loadAnalyzeRecoveryManifest(t, store, cfg) + if completed.Stages["analyze"].Status != manifest.StatusSucceeded { + t.Fatalf("aggregate analyze = %#v, want succeeded", completed.Stages["analyze"]) + } + for _, key := range []string{"chronicle", "encounters", "player_handout", "quest_log", "unrelated"} { + assertAnalyzeArtifactStatus(t, completed.Stages["analyze"], key, manifest.AnalyzeArtifactCurrent) + } + assertStageRecordsUnchanged(t, completed, upstream) +} + +func TestExecuteStagesJoinsAnalyzeFailureWithTerminalSaveFailure(t *testing.T) { + cfg := testConfig(t) + cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ + Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "chronicle": {Enabled: true, PromptID: "chronicle", OutputPath: "artifacts/chronicle.md"}, + }, + } + artifactErr := errors.New("injected analyze failure") + terminalErr := errors.New("injected terminal save failure") + store := &failedAnalyzeSaveStore{Store: &manifest.LocalStore{}, Err: terminalErr} + runner := &analyzeCheckpointRunner{FailPrompt: "chronicle", Err: artifactErr} + + _, err := executeStages(context.Background(), cfg, []stage.Stage{mustSelectStage(t, "analyze")}, RunOptions{ + Env: &Env{ManifestStore: store, Scriptorium: runner}, + }) + if !errors.Is(err, artifactErr) || !errors.Is(err, terminalErr) { + t.Fatalf("executeStages() error = %v, want analyze and terminal save failures", err) + } + if !store.Failed { + t.Fatal("terminal save failure was not injected") + } +} + +func TestExecuteStagesPersistsAnalyzeCheckpointFailureRecovery(t *testing.T) { + cfg := testConfig(t) + cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ + Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "chronicle": {Enabled: true, PromptID: "chronicle", OutputPath: "artifacts/chronicle.md"}, + "encounters": {Enabled: true, PromptID: "encounters", OutputPath: "artifacts/encounters.md"}, + "quest_log": {Enabled: true, PromptID: "quest-log", OutputPath: "artifacts/quest_log.md"}, + }, + } + baseStore := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.Campaign = cfg.Session.Campaign + seed.MarkStageSucceeded("publish", time.Now().UTC(), nil) + seed.MarkStageSucceeded("notify", time.Now().UTC(), nil) + saveBoundedManifest(t, cfg, seed) + checkpointErr := errors.New("injected artifact checkpoint failure") + store := ¤tAnalyzeCheckpointFailingStore{ + Store: baseStore, + CurrentLimit: 2, + Err: checkpointErr, + } + runner := &analyzeCheckpointRunner{} + + _, err := executeStages(context.Background(), cfg, []stage.Stage{mustSelectStage(t, "analyze")}, RunOptions{ + Env: &Env{ManifestStore: store, Scriptorium: runner}, + }) + if !errors.Is(err, checkpointErr) { + t.Fatalf("executeStages() error = %v, want checkpoint failure", err) + } + if want := []string{"chronicle", "encounters"}; !reflect.DeepEqual(runner.Calls, want) { + t.Fatalf("adapter calls = %#v, want %#v", runner.Calls, want) + } + + persisted := loadAnalyzeRecoveryManifest(t, baseStore, cfg) + if persisted.Stages["analyze"].Status != manifest.StatusFailed { + t.Fatalf("aggregate analyze = %#v, want failed", persisted.Stages["analyze"]) + } + assertAnalyzeArtifactStatus(t, persisted.Stages["analyze"], "chronicle", manifest.AnalyzeArtifactCurrent) + assertAnalyzeArtifactStatus(t, persisted.Stages["analyze"], "encounters", manifest.AnalyzeArtifactFailed) + if _, exists := persisted.Stages["analyze"].AnalyzeArtifacts["quest_log"]; exists { + t.Fatalf("unattempted quest log appeared: %#v", persisted.Stages["analyze"].AnalyzeArtifacts["quest_log"]) + } + for _, name := range []string{"publish", "notify"} { + if got := persisted.Stages[name]; got == nil || got.Status != manifest.StatusStale { + t.Fatalf("%s stage = %#v, want stale", name, got) + } + } +} + +type analyzeCheckpointRunner struct { + Calls []string + FailPrompt string + Err error +} + +type failedAnalyzeSaveStore struct { + manifest.Store + Err error + Failed bool +} + +type currentAnalyzeCheckpointFailingStore struct { + manifest.Store + CurrentLimit int + Err error + Failed bool +} + +func (s *currentAnalyzeCheckpointFailingStore) Save(ctx context.Context, path string, m *manifest.Manifest) error { + if !s.Failed { + current := 0 + if record := m.Stages["analyze"]; record != nil && record.Status == manifest.StatusRunning { + for _, artifact := range record.AnalyzeArtifacts { + if artifact.Status == manifest.AnalyzeArtifactCurrent { + current++ + } + } + } + if current >= s.CurrentLimit { + s.Failed = true + return s.Err + } + } + return s.Store.Save(ctx, path, m) +} + +func (s *failedAnalyzeSaveStore) Save(ctx context.Context, path string, m *manifest.Manifest) error { + if record := m.Stages["analyze"]; !s.Failed && record != nil && record.Status == manifest.StatusFailed { + s.Failed = true + return s.Err + } + return s.Store.Save(ctx, path, m) +} + +func (r *analyzeCheckpointRunner) RunArtifact( + ctx context.Context, + req scriptorium.RunArtifactRequest, +) (scriptorium.ArtifactResult, error) { + if err := ctx.Err(); err != nil { + return scriptorium.ArtifactResult{}, err + } + r.Calls = append(r.Calls, req.PromptID) + if req.PromptID == r.FailPrompt { + return scriptorium.ArtifactResult{}, r.Err + } + if err := os.MkdirAll(filepath.Dir(req.OutputPath), 0o755); err != nil { + return scriptorium.ArtifactResult{}, err + } + if err := os.WriteFile(req.OutputPath, []byte("generated "+req.PromptID+"\n"), 0o644); err != nil { + return scriptorium.ArtifactResult{}, err + } + return scriptorium.ArtifactResult{ + OutputPath: req.OutputPath, + CommandMode: scriptorium.CommandModeRun, + PromptID: req.PromptID, + ProfileID: req.ProfileID, + }, nil +} + +func (*analyzeCheckpointRunner) RenderArtifact( + context.Context, + scriptorium.RenderArtifactRequest, +) (scriptorium.ArtifactResult, error) { + return scriptorium.ArtifactResult{}, errors.New("unexpected render") +} + +func checkpointRecoveryArtifacts(revision string) map[string]config.ScriptoriumArtifactConfig { + return map[string]config.ScriptoriumArtifactConfig{ + "chronicle": { + Enabled: true, PromptID: "chronicle-" + revision, OutputPath: "artifacts/chronicle.md", + }, + "encounters": { + Enabled: true, DependsOn: []string{"chronicle"}, PromptID: "encounters-" + revision, + OutputPath: "artifacts/encounters.md", + Inputs: map[string]config.ScriptoriumInputConfig{ + "chronicle": {Source: "narratio.artifact.chronicle", Required: true}, + }, + }, + "quest_log": { + Enabled: true, DependsOn: []string{"encounters"}, PromptID: "quest-log-" + revision, + OutputPath: "artifacts/quest_log.md", + Inputs: map[string]config.ScriptoriumInputConfig{ + "encounters": {Source: "narratio.artifact.encounters", Required: true}, + }, + }, + "player_handout": { + Enabled: true, DependsOn: []string{"quest_log"}, PromptID: "player-handout-" + revision, + OutputPath: "artifacts/player_handout.md", + Inputs: map[string]config.ScriptoriumInputConfig{ + "quest_log": {Source: "narratio.artifact.quest_log", Required: true}, + }, + }, + "unrelated": { + Enabled: true, PromptID: "unrelated", OutputPath: "artifacts/unrelated.md", + }, + } +} + +func mustSelectStage(t *testing.T, name string) stage.Stage { + t.Helper() + selected, err := stage.Select(name) + if err != nil { + t.Fatalf("Select(%q) error = %v", name, err) + } + return selected +} + +func loadAnalyzeRecoveryManifest(t *testing.T, store manifest.Store, cfg *config.Config) *manifest.Manifest { + t.Helper() + m, err := store.Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatalf("load session manifest: %v", err) + } + return m +} + +func assertAnalyzeArtifactStatus( + t *testing.T, + record *manifest.StageRecord, + key string, + want manifest.AnalyzeArtifactStatus, +) { + t.Helper() + got, exists := record.AnalyzeArtifacts[key] + if !exists || got.Status != want { + t.Fatalf("artifact %q = %#v, want status %q", key, got, want) + } + if want != manifest.AnalyzeArtifactCurrent && got.Output != nil { + t.Fatalf("artifact %q retained output authority: %#v", key, got) + } +} + +func cloneStageRecords(m *manifest.Manifest, names ...string) map[string]manifest.StageRecord { + clones := make(map[string]manifest.StageRecord, len(names)) + for _, name := range names { + clones[name] = *m.Stages[name] + } + return clones +} + +func assertStageRecordsUnchanged(t *testing.T, m *manifest.Manifest, want map[string]manifest.StageRecord) { + t.Helper() + for name, expected := range want { + if got := m.Stages[name]; got == nil || !reflect.DeepEqual(*got, expected) { + t.Fatalf("upstream stage %q = %#v, want unchanged %#v", name, got, expected) + } + } +} diff --git a/internal/app/analyze_projection.go b/internal/app/analyze_projection.go deleted file mode 100644 index 46658ef..0000000 --- a/internal/app/analyze_projection.go +++ /dev/null @@ -1,110 +0,0 @@ -package app - -import ( - "fmt" - "sort" - - "gitea.maximumdirect.net/eric/narratio/internal/manifest" - "gitea.maximumdirect.net/eric/narratio/internal/stage" -) - -type validatedAnalyzeProjection struct { - session map[string]manifest.AnalyzeArtifactRecord -} - -type analyzeStateSnapshot struct { - version int - records map[string]manifest.AnalyzeArtifactRecord -} - -func captureAnalyzeState(manifestValue *manifest.Manifest, stageName string) analyzeStateSnapshot { - if manifestValue == nil || stageName != "analyze" || manifestValue.Stages["analyze"] == nil { - return analyzeStateSnapshot{} - } - record := manifestValue.Stages["analyze"] - return analyzeStateSnapshot{ - version: record.AnalyzeStateVersion, - records: manifest.CloneAnalyzeArtifactCollection(record.AnalyzeArtifacts), - } -} - -func restoreAnalyzeState(manifestValue *manifest.Manifest, snapshot analyzeStateSnapshot) { - if manifestValue == nil || manifestValue.Stages["analyze"] == nil { - return - } - record := manifestValue.Stages["analyze"] - record.AnalyzeStateVersion = snapshot.version - record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(snapshot.records) -} - -func validateSuccessfulAnalyzeProjection(stageName string, result *stage.StageResult) (*validatedAnalyzeProjection, error) { - if result == nil || result.AnalyzeState == nil { - return nil, nil - } - if stageName != "analyze" { - return nil, fmt.Errorf("stage %q returned analyze-owned state projection", stageName) - } - if result.Disposition == stage.StageDispositionSkipped { - return nil, fmt.Errorf("skipped analyze result cannot contain analyze-owned state projection") - } - if len(result.Outputs) != 0 { - return nil, fmt.Errorf("analyze result with state projection cannot contain ordinary outputs") - } - return validateAndCloneAnalyzeProjection(result.AnalyzeState) -} - -func validateFailedAnalyzeProjection(stageName string, result *stage.StageResult) (*validatedAnalyzeProjection, error) { - if result == nil || result.AnalyzeState == nil { - return nil, nil - } - if stageName != "analyze" { - return nil, fmt.Errorf("stage %q returned analyze-owned state projection with an error", stageName) - } - if result.Disposition != stage.StageDispositionSucceeded || result.SkipReason != "" || len(result.Outputs) != 0 || len(result.Logs) != 0 || len(result.GeneratedConfigs) != 0 || len(result.Metadata) != 0 { - return nil, fmt.Errorf("analyze result with an error may contain only analyze-owned state projection") - } - return validateAndCloneAnalyzeProjection(result.AnalyzeState) -} - -func validateAndCloneAnalyzeProjection(projection *stage.AnalyzeStateProjection) (*validatedAnalyzeProjection, error) { - session := manifest.CloneAnalyzeArtifactCollection(projection.Session) - if err := manifest.ValidateAnalyzeArtifactCollection(manifest.AnalyzeStateContractVersion, session); err != nil { - return nil, fmt.Errorf("validate reconciled session analyze state: %w", err) - } - return &validatedAnalyzeProjection{session: session}, nil -} - -func applyAnalyzeProjection( - sessionManifest *manifest.Manifest, - projection *validatedAnalyzeProjection, -) { - if projection == nil { - return - } - if sessionManifest != nil && sessionManifest.Stages["analyze"] != nil { - record := sessionManifest.Stages["analyze"] - record.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion - record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(projection.session) - } -} - -func analyzeProjectionOutputs(records map[string]manifest.AnalyzeArtifactRecord) []manifest.ArtifactRecord { - keys := make([]string, 0, len(records)) - for key, record := range records { - if record.Status != manifest.AnalyzeArtifactCurrent || record.Output == nil { - continue - } - keys = append(keys, key) - } - sort.Strings(keys) - outputs := make([]manifest.ArtifactRecord, 0, len(keys)) - for _, key := range keys { - record := manifest.CloneAnalyzeArtifactCollection(map[string]manifest.AnalyzeArtifactRecord{key: records[key]})[key] - output := *record.Output - if output.ProducerRunID == "" { - output.ProducerRunID = record.ProducerRunID - } - outputs = append(outputs, output) - } - return outputs -} diff --git a/internal/app/analyze_projection_test.go b/internal/app/analyze_projection_test.go deleted file mode 100644 index 7549de4..0000000 --- a/internal/app/analyze_projection_test.go +++ /dev/null @@ -1,309 +0,0 @@ -package app - -import ( - "context" - "errors" - "reflect" - "strings" - "testing" - "time" - - "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" - "gitea.maximumdirect.net/eric/narratio/internal/artifacts" - "gitea.maximumdirect.net/eric/narratio/internal/manifest" - "gitea.maximumdirect.net/eric/narratio/internal/stage" -) - -type projectionStage struct { - name string - run func(*stage.Env, *manifest.Manifest) (*stage.StageResult, error) -} - -func (s projectionStage) Name() string { return s.name } -func (s projectionStage) Run(_ context.Context, env *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - return s.run(env, m) -} - -func TestExecuteStagesProjectsReconciledSessionAnalyzeState(t *testing.T) { - cfg := testConfig(t) - oldAt := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC) - oldRecord := appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", oldAt) - - stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - newRecord := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC()) - staleRecord := appAnalyzeRecord("quest_log", manifest.AnalyzeArtifactStale, m.RunID, time.Now().UTC()) - session := map[string]manifest.AnalyzeArtifactRecord{ - "player_handout": oldRecord, - "quest_log": staleRecord, - "session_recap": newRecord, - } - return &stage.StageResult{ - Logs: []string{"aggregate-analyze.log"}, - AnalyzeState: &stage.AnalyzeStateProjection{ - Session: session, - }, - }, nil - }} - - summary, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{}) - if err != nil { - t.Fatalf("executeStages() error = %v", err) - } - store := &manifest.LocalStore{} - sessionManifest, err := store.Load(context.Background(), summary.ManifestPath) - if err != nil { - t.Fatalf("Load(session) error = %v", err) - } - analyze := sessionManifest.Stages["analyze"] - if analyze.AnalyzeStateVersion != manifest.AnalyzeStateContractVersion || len(analyze.AnalyzeArtifacts) != 3 { - t.Fatalf("session analyze state = %#v", analyze) - } - if got := analyzeArtifactOutputKeys(analyze.Outputs); !reflect.DeepEqual(got, []string{"player_handout", "session_recap"}) { - t.Fatalf("session aggregate outputs = %#v, want current records only", got) - } - if len(analyze.Logs) != 1 || analyze.Logs[0] != "aggregate-analyze.log" { - t.Fatalf("session aggregate logs = %#v", analyze.Logs) - } - -} - -func TestExecuteStagesPersistsRestrictedAnalyzeStateOnPartialError(t *testing.T) { - cfg := testConfig(t) - now := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC) - seed := manifest.New(cfg.Session.SessionID, now) - seed.Campaign = cfg.Session.Campaign - seed.MarkStageSucceeded("analyze", now, nil) - seed.Stages["analyze"].AnalyzeStateVersion = manifest.AnalyzeStateContractVersion - seed.Stages["analyze"].AnalyzeArtifacts = map[string]manifest.AnalyzeArtifactRecord{ - "player_handout": appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", now), - } - seed.MarkStageSucceeded("publish", now, nil) - saveBoundedManifest(t, cfg, seed) - - stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - unrelated := seed.Stages["analyze"].AnalyzeArtifacts["player_handout"] - completed := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC()) - failed := appAnalyzeRecord("quest_log", manifest.AnalyzeArtifactFailed, m.RunID, time.Now().UTC()) - session := map[string]manifest.AnalyzeArtifactRecord{ - "player_handout": unrelated, - "quest_log": failed, - "session_recap": completed, - } - return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{ - Session: session, - }}, errors.New("quest log failed") - }} - - _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: true}) - if err == nil || !strings.Contains(err.Error(), "quest log failed") { - t.Fatalf("executeStages() error = %v", err) - } - store := &manifest.LocalStore{} - loaded, err := store.Load(context.Background(), manifestPathFor(cfg)) - if err != nil { - t.Fatalf("Load(session) error = %v", err) - } - analyze := loaded.Stages["analyze"] - if analyze.Status != manifest.StatusFailed || len(analyze.Outputs) != 0 { - t.Fatalf("aggregate analyze state = %#v, want failed without outputs", analyze) - } - if analyze.AnalyzeArtifacts["player_handout"].Status != manifest.AnalyzeArtifactCurrent || - analyze.AnalyzeArtifacts["session_recap"].Status != manifest.AnalyzeArtifactCurrent || - analyze.AnalyzeArtifacts["quest_log"].Status != manifest.AnalyzeArtifactFailed { - t.Fatalf("partial session projection = %#v", analyze.AnalyzeArtifacts) - } - if loaded.Stages["publish"].Status != manifest.StatusStale { - t.Fatalf("publish status = %q, want stale", loaded.Stages["publish"].Status) - } - -} - -func TestExecuteStagesRejectsInvalidAnalyzeProjectionWithoutReplacingPriorState(t *testing.T) { - cfg := testConfig(t) - now := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC) - prior := appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", now) - seed := manifest.New(cfg.Session.SessionID, now) - seed.Campaign = cfg.Session.Campaign - seed.MarkStageSucceeded("analyze", now, nil) - seed.Stages["analyze"].AnalyzeStateVersion = manifest.AnalyzeStateContractVersion - seed.Stages["analyze"].AnalyzeArtifacts = map[string]manifest.AnalyzeArtifactRecord{"player_handout": prior} - saveBoundedManifest(t, cfg, seed) - - stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - invalid := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC()) - invalid.Output.Checksum = "invalid" - return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{ - Session: map[string]manifest.AnalyzeArtifactRecord{"session_recap": invalid}, - }}, errors.New("analysis failed") - }} - - _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: true}) - if err == nil || !strings.Contains(err.Error(), "checksum") { - t.Fatalf("executeStages() error = %v, want projection validation failure", err) - } - loaded, loadErr := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) - if loadErr != nil { - t.Fatalf("Load() error = %v", loadErr) - } - if len(loaded.Stages["analyze"].AnalyzeArtifacts) != 1 || !reflect.DeepEqual(loaded.Stages["analyze"].AnalyzeArtifacts["player_handout"], prior) { - t.Fatalf("prior state replaced by invalid projection: %#v", loaded.Stages["analyze"].AnalyzeArtifacts) - } -} - -func TestExecuteStagesRollsBackAnalyzeAuthorityWhenProjectionSaveFails(t *testing.T) { - cfg := testConfig(t) - now := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC) - prior := appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", now) - seed := manifest.New(cfg.Session.SessionID, now) - seed.Campaign = cfg.Session.Campaign - seed.MarkStageSucceeded("analyze", now, nil) - seed.Stages["analyze"].AnalyzeStateVersion = manifest.AnalyzeStateContractVersion - seed.Stages["analyze"].AnalyzeArtifacts = map[string]manifest.AnalyzeArtifactRecord{"player_handout": prior} - saveBoundedManifest(t, cfg, seed) - - stageReturned := false - store := &analyzeProjectionFailingStore{delegate: &manifest.LocalStore{}, shouldFail: func(m *manifest.Manifest) bool { - return stageReturned && m.Stages["analyze"] != nil && m.Stages["analyze"].Status == manifest.StatusSucceeded && m.Stages["analyze"].AnalyzeArtifacts["session_recap"].Status == manifest.AnalyzeArtifactCurrent - }} - stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - stageReturned = true - newRecord := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC()) - return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{ - Session: map[string]manifest.AnalyzeArtifactRecord{ - "player_handout": prior, - "session_recap": newRecord, - }, - }}, nil - }} - - _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{ - Force: true, - Env: &Env{ManifestStore: store}, - }) - if err == nil || !strings.Contains(err.Error(), "injected analyze projection save failure") { - t.Fatalf("executeStages() error = %v", err) - } - if !store.failed { - t.Fatal("projection persistence failure was not injected") - } - loaded, loadErr := store.delegate.Load(context.Background(), manifestPathFor(cfg)) - if loadErr != nil { - t.Fatalf("Load() error = %v", loadErr) - } - analyze := loaded.Stages["analyze"] - if analyze.Status != manifest.StatusFailed || len(analyze.AnalyzeArtifacts) != 1 || !reflect.DeepEqual(analyze.AnalyzeArtifacts["player_handout"], prior) { - t.Fatalf("durable analyze state after rollback = %#v", analyze) - } -} - -func TestExecuteStagesRejectsAnalyzeProjectionFromOtherStage(t *testing.T) { - cfg := testConfig(t) - stageToRun := projectionStage{name: "prepare", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - record := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC()) - return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{ - Session: map[string]manifest.AnalyzeArtifactRecord{"session_recap": record}, - }}, nil - }} - _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{}) - if err == nil || !strings.Contains(err.Error(), "returned analyze-owned state projection") { - t.Fatalf("executeStages() error = %v", err) - } -} - -func TestExecuteStagesRejectsContradictoryAnalyzeResultWithError(t *testing.T) { - cfg := testConfig(t) - stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - record := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC()) - return &stage.StageResult{ - Outputs: []artifacts.Ref{{Kind: "session_recap", RelativePath: "artifacts/session-recap.md"}}, - AnalyzeState: &stage.AnalyzeStateProjection{ - Session: map[string]manifest.AnalyzeArtifactRecord{"session_recap": record}, - }, - }, errors.New("analysis failed") - }} - _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{}) - if err == nil || !strings.Contains(err.Error(), "may contain only analyze-owned state projection") { - t.Fatalf("executeStages() error = %v", err) - } -} - -func TestExecuteStagesExposesSelectedForceDecisionToStage(t *testing.T) { - for _, force := range []bool{false, true} { - t.Run(strings.ToLower(strings.TrimSpace(map[bool]string{false: "ordinary", true: "forced"}[force])), func(t *testing.T) { - cfg := testConfig(t) - captured := !force - stageToRun := projectionStage{name: "prepare", run: func(env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { - captured = env.Force - return &stage.StageResult{}, nil - }} - if _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: force}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - if captured != force { - t.Fatalf("stage env force = %v, want %v", captured, force) - } - }) - } -} - -func appAnalyzeRecord(key string, status manifest.AnalyzeArtifactStatus, producerRunID string, at time.Time) manifest.AnalyzeArtifactRecord { - record := manifest.AnalyzeArtifactRecord{ - Key: key, - Status: status, - ProducerRunID: producerRunID, - UpdatedAt: at, - } - if status == manifest.AnalyzeArtifactFailed { - record.Error = "scriptorium failed" - return record - } - if status != manifest.AnalyzeArtifactCurrent { - record.FingerprintVersion = manifest.AnalyzeFingerprintContractVersion - record.Fingerprint = strings.Repeat("b", 64) - return record - } - record.FingerprintVersion = manifest.AnalyzeFingerprintContractVersion - record.Fingerprint = strings.Repeat("a", 64) - record.OutputSize = 42 - record.Output = &manifest.ArtifactRecord{ - Kind: key, - SourceID: artifacts.ConfiguredArtifactSourceID(key), - LocalPath: "artifacts/" + strings.ReplaceAll(key, "_", "-") + ".md", - ProducerRunID: producerRunID, - Checksum: strings.Repeat("c", 64), - Contract: &artifactmodel.ContractMetadata{ - MediaType: "text/markdown", SchemaID: "narratio." + key, SchemaVersion: "1", - }, - } - return record -} - -func analyzeArtifactOutputKeys(outputs []manifest.ArtifactRecord) []string { - keys := make([]string, 0, len(outputs)) - for _, output := range outputs { - keys = append(keys, strings.TrimPrefix(output.SourceID, "narratio.artifact.")) - } - return keys -} - -type analyzeProjectionFailingStore struct { - delegate *manifest.LocalStore - shouldFail func(*manifest.Manifest) bool - failed bool -} - -func (s *analyzeProjectionFailingStore) Create(ctx context.Context, sessionID string) (*manifest.Manifest, error) { - return s.delegate.Create(ctx, sessionID) -} - -func (s *analyzeProjectionFailingStore) Load(ctx context.Context, path string) (*manifest.Manifest, error) { - return s.delegate.Load(ctx, path) -} - -func (s *analyzeProjectionFailingStore) Save(ctx context.Context, path string, m *manifest.Manifest) error { - if !s.failed && s.shouldFail != nil && s.shouldFail(m) { - s.failed = true - return errors.New("injected analyze projection save failure") - } - return s.delegate.Save(ctx, path, m) -} diff --git a/internal/app/runner_stage_execution.go b/internal/app/runner_stage_execution.go index 5b337fa..e8bfa8f 100644 --- a/internal/app/runner_stage_execution.go +++ b/internal/app/runner_stage_execution.go @@ -55,7 +55,6 @@ func executeSelectedStage(runtime *invocationRuntime, selected stage.Stage) (sta } priorOutcome := capturePriorStageOutcome(m, selected.Name()) - priorAnalyzeState := captureAnalyzeState(m, selected.Name()) attempted := stageExecutionOutcome{executed: true} startedAt := runtime.dependencies.now() if _, err := beginStageAttempt(m, selected.Name(), decision, startedAt); err != nil { @@ -75,23 +74,10 @@ func executeSelectedStage(runtime *invocationRuntime, selected stage.Stage) (sta env.Logger.Debug("manifest saved", "stage", selected.Name(), "transition", "running", "path", runtime.manifestPath) result, err := selected.Run(runtime.ctx, env, m) - var analyzeProjection *validatedAnalyzeProjection if err == nil { err = validateStageResult(result) - if err == nil { - analyzeProjection, err = validateSuccessfulAnalyzeProjection(selected.Name(), result) - } - } else if result != nil && result.AnalyzeState != nil { - var projectionErr error - analyzeProjection, projectionErr = validateFailedAnalyzeProjection(selected.Name(), result) - if projectionErr != nil { - err = errors.Join(err, projectionErr) - } } if err != nil { - if analyzeProjection != nil { - applyAnalyzeProjection(m, analyzeProjection) - } failedAt := runtime.dependencies.now() m.MarkStageFailed(selected.Name(), failedAt, err.Error()) if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), failedAt, staleReasonFailure); invalidationErr != nil { @@ -119,22 +105,14 @@ func executeSelectedStage(runtime *invocationRuntime, selected stage.Stage) (sta } outputs := mapResultOutputs(selected.Name(), result, runtime.identity.RunID) - if analyzeProjection != nil { - outputs = analyzeProjectionOutputs(analyzeProjection.session) - } succeededAt := runtime.dependencies.now() if _, err := applyCompletedStageOutcome(m, selected.Name(), result, outputs, semanticConfig, priorOutcome, succeededAt); err != nil { return stageExecutionOutcome{}, runtime.persistTerminalFailure( fmt.Errorf("invalidate dependents after changed stage %q result: %w", selected.Name(), err), ) } - applyAnalyzeProjection(m, analyzeProjection) - if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil { operationErr := fmt.Errorf("save manifest after stage %q: %w", selected.Name(), err) - if analyzeProjection != nil { - restoreAnalyzeState(m, priorAnalyzeState) - } failedAt := runtime.dependencies.now() m.MarkStageFailed(selected.Name(), failedAt, operationErr.Error()) if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), failedAt, staleReasonFailure); invalidationErr != nil { diff --git a/internal/manifest/analyze_state.go b/internal/manifest/analyze_state.go index 1e234dc..1d7b541 100644 --- a/internal/manifest/analyze_state.go +++ b/internal/manifest/analyze_state.go @@ -87,7 +87,7 @@ func ValidateAnalyzeArtifactCollection(version int, records map[string]AnalyzeAr } // CloneAnalyzeArtifactCollection returns a deep copy in canonical dependency -// order so projections cannot be mutated after application. +// order so candidate or installed authority cannot share mutable record data. func CloneAnalyzeArtifactCollection(records map[string]AnalyzeArtifactRecord) map[string]AnalyzeArtifactRecord { if records == nil { return nil diff --git a/internal/stage/analyze_failure_test.go b/internal/stage/analyze_failure_test.go index 0b44294..65be168 100644 --- a/internal/stage/analyze_failure_test.go +++ b/internal/stage/analyze_failure_test.go @@ -48,10 +48,55 @@ func TestAnalyzeAdapterFailureRecordsCompletedAndFailedArtifacts(t *testing.T) { } } } + + manifestPath := artifacts.SessionManifestPathForCampaign( + env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, + ) + durable, loadErr := env.ManifestStore.Load(context.Background(), manifestPath) + if loadErr != nil { + t.Fatalf("load durable manifest: %v", loadErr) + } + for index, key := range keys { + record, exists := durable.Stages["analyze"].AnalyzeArtifacts[key] + if index < failAt-1 { + if !exists || record.Status != manifest.AnalyzeArtifactCurrent { + t.Fatalf("durable completed %s = %#v", key, record) + } + } else if exists && record.Status == manifest.AnalyzeArtifactCurrent { + t.Fatalf("durable uncommitted %s = %#v", key, record) + } + } }) } } +func TestAnalyzeCancellationPreservesEarlierCheckpoint(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + env.Config.Pipeline.Scriptorium.Artifacts = independentAnalyzeArtifacts() + env.Scriptorium = &indexedAnalyzeFailureRunner{FailAt: 2, Err: context.Canceled} + + _, err := (analyzeStage{}).Run(context.Background(), env, m) + if !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want cancellation", err) + } + manifestPath := artifacts.SessionManifestPathForCampaign( + env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, + ) + durable, loadErr := env.ManifestStore.Load(context.Background(), manifestPath) + if loadErr != nil { + t.Fatalf("load durable manifest: %v", loadErr) + } + if got := durable.Stages["analyze"].AnalyzeArtifacts["alpha"]; got.Status != manifest.AnalyzeArtifactCurrent { + t.Fatalf("durable alpha = %#v, want current", got) + } + if _, exists := durable.Stages["analyze"].AnalyzeArtifacts["beta"]; exists { + t.Fatalf("durable beta should be absent after cancellation: %#v", durable.Stages["analyze"].AnalyzeArtifacts["beta"]) + } + if got := m.Stages["analyze"].AnalyzeArtifacts["beta"]; got.Status != manifest.AnalyzeArtifactFailed || got.Output != nil { + t.Fatalf("live beta = %#v, want failed without output", got) + } +} + func TestAnalyzeMiddleFailureStalesDependentsAndPreservesUnrelatedCurrent(t *testing.T) { env, m, _ := setupAnalyzeEnv(t) env.Config.Pipeline.Scriptorium.Artifacts = chainedAnalyzeArtifacts() diff --git a/internal/stage/analyze_incremental_execution_test.go b/internal/stage/analyze_incremental_execution_test.go index 5f0dbce..44bd76c 100644 --- a/internal/stage/analyze_incremental_execution_test.go +++ b/internal/stage/analyze_incremental_execution_test.go @@ -39,7 +39,7 @@ func TestAnalyzeReusesCurrentArtifactWithoutScriptorium(t *testing.T) { } } -func TestAnalyzePartialForcePreservesUnrelatedCurrentRecord(t *testing.T) { +func TestAnalyzeIdenticalReplacementPreservesUnselectedDependent(t *testing.T) { env, m, fake := setupAnalyzeEnv(t) paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`) diff --git a/internal/stage/stage.go b/internal/stage/stage.go index 9b6eda6..4d41dcc 100644 --- a/internal/stage/stage.go +++ b/internal/stage/stage.go @@ -129,11 +129,4 @@ type StageResult struct { Logs []string GeneratedConfigs []string Metadata map[string]any - AnalyzeState *AnalyzeStateProjection -} - -// AnalyzeStateProjection carries the reconciled session authority proposed for -// the runner's existing end-of-stage session-manifest save. -type AnalyzeStateProjection struct { - Session map[string]manifest.AnalyzeArtifactRecord }