From 903dc706826de19905335958158c2ae3b51fcbf5 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 29 Aug 2026 18:36:21 +0000 Subject: [PATCH] Persist partial analysis artifact results --- docs/internal/manifest.md | 17 ++ docs/roadmap/implementation.md | 2 + internal/app/analyze_projection.go | 134 +++++++++ internal/app/analyze_projection_test.go | 346 ++++++++++++++++++++++++ internal/app/runner.go | 39 ++- internal/manifest/analyze_state.go | 19 ++ internal/stage/stage.go | 9 + 7 files changed, 562 insertions(+), 4 deletions(-) create mode 100644 internal/app/analyze_projection.go create mode 100644 internal/app/analyze_projection_test.go diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index 675f578..1c24605 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -66,6 +66,23 @@ audit of only the artifacts evaluated or attempted by that run. 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 and the invocation subset. Aggregate session outputs are rebuilt in +configured-key order from current session records only; invocation outputs are +limited to current records produced by that invocation's run ID. Ordinary +stage outputs cannot accompany this projection, so there is one source of +artifact authority. + +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 and run 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. + ## Run Manifest `manifest.RunManifest` is created for each invocation and records: diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 245fa9a..449ae3c 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -362,6 +362,8 @@ can represent independently current artifacts without treating them as stages. ## Stage 7 — Runner Projection And Partial-Error State Boundary +**Status: Completed** + ### Goal Give analyze one safe way to promote reconciled session state and invocation diff --git a/internal/app/analyze_projection.go b/internal/app/analyze_projection.go new file mode 100644 index 0000000..8584bac --- /dev/null +++ b/internal/app/analyze_projection.go @@ -0,0 +1,134 @@ +package app + +import ( + "fmt" + "reflect" + "sort" + + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +type validatedAnalyzeProjection struct { + session map[string]manifest.AnalyzeArtifactRecord + invocation 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) + invocation := manifest.CloneAnalyzeArtifactCollection(projection.Invocation) + if err := manifest.ValidateAnalyzeArtifactCollection(manifest.AnalyzeStateContractVersion, session); err != nil { + return nil, fmt.Errorf("validate reconciled session analyze state: %w", err) + } + if err := manifest.ValidateAnalyzeArtifactCollection(manifest.AnalyzeStateContractVersion, invocation); err != nil { + return nil, fmt.Errorf("validate invocation analyze state: %w", err) + } + for key, invocationRecord := range invocation { + sessionRecord, ok := session[key] + if !ok { + return nil, fmt.Errorf("invocation analyze artifact %q is absent from reconciled session state", key) + } + if !reflect.DeepEqual(invocationRecord, sessionRecord) { + return nil, fmt.Errorf("invocation analyze artifact %q contradicts reconciled session state", key) + } + } + return &validatedAnalyzeProjection{session: session, invocation: invocation}, nil +} + +func applyAnalyzeProjection( + sessionManifest *manifest.Manifest, + runManifest *manifest.RunManifest, + 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) + } + if runManifest != nil && runManifest.Stages["analyze"] != nil { + record := runManifest.Stages["analyze"] + record.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion + record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(projection.invocation) + } +} + +func analyzeProjectionOutputs(records map[string]manifest.AnalyzeArtifactRecord, producerRunID string) []manifest.ArtifactRecord { + keys := make([]string, 0, len(records)) + for key, record := range records { + if record.Status != manifest.AnalyzeArtifactCurrent || record.Output == nil { + continue + } + if producerRunID != "" && record.ProducerRunID != producerRunID { + 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 new file mode 100644 index 0000000..24f0cc0 --- /dev/null +++ b/internal/app/analyze_projection_test.go @@ -0,0 +1,346 @@ +package app + +import ( + "context" + "errors" + "os" + "path/filepath" + "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 TestExecuteStagesProjectsSeparateSessionAndInvocationAnalyzeState(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, + Invocation: map[string]manifest.AnalyzeArtifactRecord{ + "player_handout": oldRecord, + "session_recap": newRecord, + }, + }, + }, 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) + } + + runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath) + if err != nil { + t.Fatalf("LoadRun() error = %v", err) + } + runAnalyze := runManifest.Stages["analyze"] + if len(runAnalyze.AnalyzeArtifacts) != 2 || runAnalyze.AnalyzeArtifacts["session_recap"].Status != manifest.AnalyzeArtifactCurrent { + t.Fatalf("invocation analyze state = %#v", runAnalyze.AnalyzeArtifacts) + } + if got := analyzeArtifactOutputKeys(runAnalyze.Outputs); !reflect.DeepEqual(got, []string{"session_recap"}) { + t.Fatalf("invocation outputs = %#v, want produced artifact only", got) + } +} + +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, + Invocation: map[string]manifest.AnalyzeArtifactRecord{ + "quest_log": failed, + "session_recap": completed, + }, + }}, 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) + } + + runsDir := artifacts.SessionRunsDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID) + entries, err := os.ReadDir(runsDir) + if err != nil || len(entries) != 1 { + t.Fatalf("run directory entries = %#v, error = %v", entries, err) + } + runManifest, err := store.LoadRun(context.Background(), filepath.Join(runsDir, entries[0].Name(), "manifest.json")) + if err != nil { + t.Fatalf("LoadRun() error = %v", err) + } + runAnalyze := runManifest.Stages["analyze"] + if runAnalyze.Status != manifest.StatusFailed || len(runAnalyze.AnalyzeArtifacts) != 2 || len(runAnalyze.Outputs) != 0 { + t.Fatalf("partial invocation projection = %#v", runAnalyze) + } +} + +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}, + Invocation: 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, + }, + Invocation: map[string]manifest.AnalyzeArtifactRecord{"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}, + Invocation: 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.go b/internal/app/runner.go index 3c261f8..540f6c3 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -253,6 +253,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage skipped := make([]string, 0, len(decisions)) for _, d := range decisions { s := d.Stage + stageEnv.Force = opts.Force runNames = append(runNames, s.Name()) d.Action = decideStageAction(s, m, opts.Force) @@ -305,6 +306,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage } executed = append(executed, s.Name()) priorOutcome := capturePriorStageOutcome(m, s.Name()) + priorAnalyzeState := captureAnalyzeState(m, s.Name()) now := nowUTC() runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now) @@ -339,10 +341,23 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath) result, err := s.Run(ctx, stageEnv, m) + var analyzeProjection *validatedAnalyzeProjection if err == nil { err = validateStageResult(result) + if err == nil { + analyzeProjection, err = validateSuccessfulAnalyzeProjection(s.Name(), result) + } + } else if result != nil && result.AnalyzeState != nil { + var projectionErr error + analyzeProjection, projectionErr = validateFailedAnalyzeProjection(s.Name(), result) + if projectionErr != nil { + err = errors.Join(err, projectionErr) + } } if err != nil { + if analyzeProjection != nil { + applyAnalyzeProjection(m, runManifest, analyzeProjection) + } failedAt := nowUTC() m.MarkStageFailed(s.Name(), failedAt, err.Error()) if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil { @@ -389,9 +404,15 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage continue } - outputs := mapResultOutputs(s.Name(), result, runID) + sessionOutputs := mapResultOutputs(s.Name(), result, runID) + runOutputs := sessionOutputs + if analyzeProjection != nil { + sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session, "") + runOutputs = analyzeProjectionOutputs(analyzeProjection.invocation, runID) + } succeededAt := nowUTC() - m.MarkStageSucceeded(s.Name(), succeededAt, outputs) + m.MarkStageSucceeded(s.Name(), succeededAt, sessionOutputs) + applyAnalyzeProjection(m, runManifest, analyzeProjection) applyStageResultToManifest(m, s.Name(), result) if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded { if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult); err != nil { @@ -403,12 +424,22 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { + operationErr := fmt.Errorf("save manifest after stage %q: %w", s.Name(), err) + if analyzeProjection != nil { + restoreAnalyzeState(m, priorAnalyzeState) + failedAt := nowUTC() + m.MarkStageFailed(s.Name(), failedAt, operationErr.Error()) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil { + operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after analyze projection persistence failure: %w", invalidationErr)) + } + runManifest.MarkStageFailed(s.Name(), failedAt, operationErr.Error()) + } return nil, persistTerminalFailure( ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, - fmt.Errorf("save manifest after stage %q: %w", s.Name(), err), + operationErr, ) } - runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs) + runManifest.MarkStageSucceeded(s.Name(), succeededAt, runOutputs) applyStageResultToRunManifest(runManifest, s.Name(), result) identity.applyToRunManifest(runManifest, manifestPath) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { diff --git a/internal/manifest/analyze_state.go b/internal/manifest/analyze_state.go index 97e8458..db8aef6 100644 --- a/internal/manifest/analyze_state.go +++ b/internal/manifest/analyze_state.go @@ -84,6 +84,25 @@ func ValidateAnalyzeArtifactCollection(version int, records map[string]AnalyzeAr return nil } +// CloneAnalyzeArtifactCollection returns a deep copy in canonical dependency +// order so projections cannot be mutated after application. +func CloneAnalyzeArtifactCollection(records map[string]AnalyzeArtifactRecord) map[string]AnalyzeArtifactRecord { + if records == nil { + return nil + } + cloned := make(map[string]AnalyzeArtifactRecord, len(records)) + for key, record := range records { + record.Dependencies = cloneStrings(record.Dependencies) + record.Logs = cloneStrings(record.Logs) + record.GeneratedConfigs = cloneStrings(record.GeneratedConfigs) + record.Output = cloneArtifactRecord(record.Output) + record.Scriptorium = cloneAnalyzeProvenance(record.Scriptorium) + cloned[key] = record + } + normalizeAnalyzeArtifactCollection(cloned) + return cloned +} + func validateAnalyzeStageState(stageName string, version int, records map[string]AnalyzeArtifactRecord) error { if stageName != "analyze" { if version != 0 || records != nil { diff --git a/internal/stage/stage.go b/internal/stage/stage.go index 3c08d3b..0957580 100644 --- a/internal/stage/stage.go +++ b/internal/stage/stage.go @@ -26,6 +26,7 @@ type Env struct { ArtifactStore artifacts.Store ManifestStore manifest.Store Logger *slog.Logger + Force bool WhisperX whisperx.Client Seriatim seriatim.Runner @@ -105,4 +106,12 @@ type StageResult struct { Logs []string GeneratedConfigs []string Metadata map[string]any + AnalyzeState *AnalyzeStateProjection +} + +// AnalyzeStateProjection carries analyze-owned reconciled session authority and +// the invocation subset evaluated by the current run. +type AnalyzeStateProjection struct { + Session map[string]manifest.AnalyzeArtifactRecord + Invocation map[string]manifest.AnalyzeArtifactRecord }