From 4e4e2b7d96e62b3eac5021464554b55e1ba702e3 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 29 Aug 2026 19:49:48 +0000 Subject: [PATCH] Preserve incremental analysis state on failure --- docs/internal/manifest.md | 9 + docs/internal/stage-analyze.md | 16 +- docs/roadmap/implementation.md | 2 + internal/stage/analyze.go | 136 ++++++++++++++-- internal/stage/analyze_failure_test.go | 217 +++++++++++++++++++++++++ 5 files changed, 365 insertions(+), 15 deletions(-) create mode 100644 internal/stage/analyze_failure_test.go diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index d7b84d6..26b066e 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -92,6 +92,15 @@ 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. +The incremental executor constructs this restricted projection at each +scheduled artifact boundary. 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 +invocation subset. Session failure state is persisted before invocation failure +state. If either terminal save fails, its persistence error is joined with the +original adapter, validation, or filesystem cause; a failed projection save +does not turn incidental canonical bytes into manifest authority. + ## Run Manifest `manifest.RunManifest` is created for each invocation and records: diff --git a/docs/internal/stage-analyze.md b/docs/internal/stage-analyze.md index 80164d1..347b8b4 100644 --- a/docs/internal/stage-analyze.md +++ b/docs/internal/stage-analyze.md @@ -125,6 +125,19 @@ 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 + with the active artifact marked `failed`, a bounded error, and no output + authority. Current transitive dependents become stale without execution. +- earlier artifacts from the invocation remain current only after their + run-local output passed validation and canonical materialization. They remain + in invocation history; unattempted later artifacts do not appear there. +- 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 persists a valid partial projection 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. ## Invariants @@ -154,4 +167,5 @@ Supported source families: `internal/stage/analyze_reconciliation.go`, and `internal/stage/analyze_reconciliation_test.go`, `internal/stage/analyze_plan.go`, `internal/stage/analyze_plan_test.go`, and - `internal/stage/analyze_incremental_execution_test.go` + `internal/stage/analyze_incremental_execution_test.go`, and + `internal/stage/analyze_failure_test.go` diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 47cf634..d3c8cf9 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -625,6 +625,8 @@ results, and preserve or invalidate records according to actual output identity. ## Stage 13 — Incremental Analyze Failure Safety +**Status: Completed** + ### Goal Complete the incremental executor with conservative, durable behavior for diff --git a/internal/stage/analyze.go b/internal/stage/analyze.go index 8ccd8b5..5e8f309 100644 --- a/internal/stage/analyze.go +++ b/internal/stage/analyze.go @@ -177,13 +177,30 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S execution, ) if err != nil { - return nil, fmt.Errorf("analyze: compute execution fingerprint for artifact %q: %w", item.Key, err) + executionErr := fmt.Errorf("analyze: compute execution fingerprint for artifact %q: %w", item.Key, err) + return failedAnalyzeResult( + execution, + item, + artifactCfg, + "", + executionErr, + sessionRecords, + invocationRecords, + ), executionErr } priorRecord, hadPriorRecord := priorCurrentRecords[item.Key] plan := analyzeArtifactExecutionPlan{Name: item.Key, Cfg: artifactCfg} artifactResult, err := executeAnalyzeArtifact(ctx, execution, plan) if err != nil { - return nil, err + return failedAnalyzeResult( + execution, + item, + artifactCfg, + fingerprint, + err, + sessionRecords, + invocationRecords, + ), err } logs = append(logs, artifactResult.Logs...) @@ -207,14 +224,27 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S artifactResult, ) if err != nil { - return nil, fmt.Errorf("analyze: record artifact %q: %w", plan.Name, err) + recordErr := fmt.Errorf("analyze: record artifact %q: %w", plan.Name, err) + return failedAnalyzeResult( + execution, + item, + artifactCfg, + fingerprint, + recordErr, + sessionRecords, + invocationRecords, + ), recordErr } sessionRecords[plan.Name] = record invocationRecords[plan.Name] = record sourceID, ok := runtimeCatalog.SourceIDForConfiguredKey(plan.Name) if !ok { - return nil, fmt.Errorf("analyze: source id not found for artifact %q", plan.Name) + catalogErr := fmt.Errorf("analyze: source id not found for artifact %q", plan.Name) + return failedAnalyzeResult( + execution, item, artifactCfg, fingerprint, catalogErr, + sessionRecords, invocationRecords, + ), catalogErr } if err := runtimeCatalog.MarkAvailableGeneratedEvidence( sourceID, @@ -224,7 +254,11 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S record.OutputSize, record.Output.Contract, ); err != nil { - return nil, fmt.Errorf("analyze: mark generated artifact %q available: %w", sourceID, err) + catalogErr := fmt.Errorf("analyze: mark generated artifact %q available: %w", sourceID, err) + return failedAnalyzeResult( + execution, item, artifactCfg, fingerprint, catalogErr, + sessionRecords, invocationRecords, + ), catalogErr } if !hadPriorRecord || !sameAnalyzeOutputIdentity(priorRecord, record) { staleUnscheduledAnalyzeDependents( @@ -263,6 +297,77 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S }, nil } +func failedAnalyzeResult( + execution analyzeExecutionContext, + item analyzePlanItem, + artifactCfg config.ScriptoriumArtifactConfig, + fingerprint string, + cause error, + sessionRecords map[string]manifest.AnalyzeArtifactRecord, + invocationRecords map[string]manifest.AnalyzeArtifactRecord, +) *StageResult { + record := manifest.AnalyzeArtifactRecord{ + Key: item.Key, + Status: manifest.AnalyzeArtifactFailed, + Dependencies: normalizedAnalyzeDependencyKeys(artifactCfg.DependsOn), + ProducerRunID: analyzeProducerRunID(execution), + UpdatedAt: time.Now().UTC(), + Error: NonResumable(cause.Error()).Reason, + } + if fingerprint != "" { + record.FingerprintVersion = manifest.AnalyzeFingerprintContractVersion + record.Fingerprint = fingerprint + } + if artifactCfg.PromptID != "" || artifactCfg.ProfileID != "" { + record.Scriptorium = &manifest.AnalyzeArtifactProvenance{ + PromptID: artifactCfg.PromptID, ProfileID: artifactCfg.ProfileID, + } + } + sessionRecords[item.Key] = record + invocationRecords[item.Key] = record + staleAnalyzeDependents(artifactCfgMap(execution), item.Key, sessionRecords) + return &StageResult{AnalyzeState: &AnalyzeStateProjection{ + Session: manifest.CloneAnalyzeArtifactCollection(sessionRecords), + Invocation: manifest.CloneAnalyzeArtifactCollection(invocationRecords), + }} +} + +func artifactCfgMap(execution analyzeExecutionContext) map[string]config.ScriptoriumArtifactConfig { + if execution.Env == nil || execution.Env.Config == nil || execution.Env.Config.Pipeline == nil || + execution.Env.Config.Pipeline.Scriptorium == nil { + return nil + } + return execution.Env.Config.Pipeline.Scriptorium.Artifacts +} + +func staleAnalyzeDependents( + configured map[string]config.ScriptoriumArtifactConfig, + changed string, + records map[string]manifest.AnalyzeArtifactRecord, +) { + reverse := make(map[string][]string, len(configured)) + for key, artifactCfg := range configured { + for _, dependency := range normalizedAnalyzeDependencyKeys(artifactCfg.DependsOn) { + reverse[dependency] = append(reverse[dependency], key) + } + } + for key := range reverse { + sort.Strings(reverse[key]) + } + queue := append([]string(nil), reverse[changed]...) + seen := make(map[string]struct{}, len(queue)) + for len(queue) > 0 { + key := queue[0] + queue = queue[1:] + if _, visited := seen[key]; visited { + continue + } + seen[key] = struct{}{} + staleProjectedAnalyzeRecord(records, key) + queue = append(queue, reverse[key]...) + } +} + func currentAnalyzeArtifactRecord( execution analyzeExecutionContext, plan analyzeArtifactExecutionPlan, @@ -272,15 +377,7 @@ func currentAnalyzeArtifactRecord( if result == nil { return manifest.AnalyzeArtifactRecord{}, fmt.Errorf("execution result is required") } - producerRunID := "" - if execution.Manifest != nil { - producerRunID = strings.TrimSpace(execution.Manifest.RunID) - } - if producerRunID == "" { - // Direct stage callers predate invocation manifests. Application-owned - // execution always supplies the actual run identity. - producerRunID = "direct-analyze" - } + producerRunID := analyzeProducerRunID(execution) relativePath, err := normalizedAnalyzeOutputIdentity(plan.Cfg.OutputPath) if err != nil { return manifest.AnalyzeArtifactRecord{}, err @@ -322,6 +419,17 @@ func currentAnalyzeArtifactRecord( return record, nil } +func analyzeProducerRunID(execution analyzeExecutionContext) string { + if execution.Manifest != nil { + if runID := strings.TrimSpace(execution.Manifest.RunID); runID != "" { + return runID + } + } + // Direct stage callers predate invocation manifests. Application-owned + // execution always supplies the actual run identity. + return "direct-analyze" +} + func sameAnalyzeOutputIdentity(left, right manifest.AnalyzeArtifactRecord) bool { if left.Status != manifest.AnalyzeArtifactCurrent || right.Status != manifest.AnalyzeArtifactCurrent || left.Output == nil || right.Output == nil || left.Output.Contract == nil || right.Output.Contract == nil { diff --git a/internal/stage/analyze_failure_test.go b/internal/stage/analyze_failure_test.go new file mode 100644 index 0000000..d03e8ce --- /dev/null +++ b/internal/stage/analyze_failure_test.go @@ -0,0 +1,217 @@ +package stage + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestAnalyzeAdapterFailureProjectsCompletedAndFailedArtifacts(t *testing.T) { + for _, failAt := range []int{1, 2, 3} { + t.Run(map[int]string{1: "first", 2: "middle", 3: "last"}[failAt], func(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + env.Config.Pipeline.Scriptorium.Artifacts = independentAnalyzeArtifacts() + cause := errors.New("injected adapter failure") + runner := &indexedAnalyzeFailureRunner{FailAt: failAt, Err: cause} + env.Scriptorium = runner + + result, err := (analyzeStage{}).Run(context.Background(), env, m) + if !errors.Is(err, cause) { + t.Fatalf("Run() error = %v, want injected cause", err) + } + if result == nil || result.AnalyzeState == nil { + t.Fatal("failure result has no analyze projection") + } + keys := []string{"alpha", "beta", "gamma"} + for index, key := range keys { + record, exists := result.AnalyzeState.Invocation[key] + switch { + case index < failAt-1: + if !exists || record.Status != manifest.AnalyzeArtifactCurrent || record.Output == nil { + t.Fatalf("completed %s = %#v", key, record) + } + case index == failAt-1: + if !exists || record.Status != manifest.AnalyzeArtifactFailed || record.Output != nil || record.Error == "" { + t.Fatalf("failed %s = %#v", key, record) + } + default: + if exists { + t.Fatalf("unattempted %s appeared in invocation: %#v", key, record) + } + } + } + }) + } +} + +func TestAnalyzeMiddleFailureStalesDependentsAndPreservesUnrelatedCurrent(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + env.Config.Pipeline.Scriptorium.Artifacts = chainedAnalyzeArtifacts() + first, err := (analyzeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatal(err) + } + installAnalyzeProjection(m, first.AnalyzeState) + unrelated := m.Stages["analyze"].AnalyzeArtifacts["unrelated"] + stale := m.Stages["analyze"].AnalyzeArtifacts["alpha"] + stale.Status = manifest.AnalyzeArtifactStale + stale.Output = nil + stale.OutputSize = 0 + m.Stages["analyze"].AnalyzeArtifacts["alpha"] = stale + env.SelectedArtifactKeys = []string{"gamma"} + cause := errors.New("middle artifact failed") + env.Scriptorium = &indexedAnalyzeFailureRunner{FailAt: 2, Err: cause} + + result, err := (analyzeStage{}).Run(context.Background(), env, m) + if !errors.Is(err, cause) { + t.Fatalf("Run() error = %v", err) + } + if result.AnalyzeState.Session["alpha"].Status != manifest.AnalyzeArtifactCurrent { + t.Fatalf("completed prerequisite = %#v", result.AnalyzeState.Session["alpha"]) + } + if result.AnalyzeState.Session["beta"].Status != manifest.AnalyzeArtifactFailed { + t.Fatalf("failed middle = %#v", result.AnalyzeState.Session["beta"]) + } + if target := result.AnalyzeState.Session["gamma"]; target.Status != manifest.AnalyzeArtifactStale || target.Output != nil { + t.Fatalf("dependent target = %#v, want stale", target) + } + if got := result.AnalyzeState.Session["unrelated"]; got.Status != manifest.AnalyzeArtifactCurrent || + got.Fingerprint != unrelated.Fingerprint || got.Output == nil || got.Output.Checksum != unrelated.Output.Checksum { + t.Fatalf("unrelated record = %#v, want preserved %#v", got, unrelated) + } + if _, attempted := result.AnalyzeState.Invocation["gamma"]; attempted { + t.Fatal("dependent target was reported as attempted") + } + if old := m.Stages["analyze"].AnalyzeArtifacts["beta"]; old.Output == nil { + t.Fatal("fixture did not retain old canonical evidence before failure") + } + if failed := result.AnalyzeState.Session["beta"]; failed.Output != nil { + t.Fatal("old canonical bytes made failed artifact current") + } +} + +func TestAnalyzeInvalidOutputReturnsFailedProjection(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`) + m.Campaign = env.Config.Session.Campaign + m.RunID = "run-invalid-output" + env.Scriptorium = invalidAnalyzeOutputRunner{} + + result, err := (analyzeStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("Run() error = nil") + } + record := result.AnalyzeState.Session["session_recap"] + if record.Status != manifest.AnalyzeArtifactFailed || record.Output != nil { + t.Fatalf("failed record = %#v", record) + } +} + +func TestAnalyzeMaterializationFailureReturnsFailedProjection(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`) + m.Campaign = env.Config.Session.Campaign + m.RunID = "run-materialize-failure" + cause := errors.New("injected canonical write failure") + canonical := filepath.Join(paths.ArtifactsDir, "session_recap.md") + env.ArtifactStore = &failingAnalyzeArtifactStore{Store: env.ArtifactStore, FailPath: canonical, Err: cause} + + result, err := (analyzeStage{}).Run(context.Background(), env, m) + if !errors.Is(err, cause) { + t.Fatalf("Run() error = %v, want write cause", err) + } + record := result.AnalyzeState.Session["session_recap"] + if record.Status != manifest.AnalyzeArtifactFailed || record.Output != nil { + t.Fatalf("failed record = %#v", record) + } + if _, statErr := os.Stat(canonical); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("canonical output stat error = %v, want absent", statErr) + } +} + +type indexedAnalyzeFailureRunner struct { + Calls int + FailAt int + Err error +} + +func (r *indexedAnalyzeFailureRunner) RunArtifact(_ context.Context, req scriptorium.RunArtifactRequest) (scriptorium.ArtifactResult, error) { + r.Calls++ + if r.Calls == r.FailAt { + return scriptorium.ArtifactResult{}, r.Err + } + writeAnalyzeFileNoTest(req.OutputPath, "generated "+req.PromptID+"\n") + return scriptorium.ArtifactResult{ + OutputPath: req.OutputPath, CommandMode: scriptorium.CommandModeRun, + PromptID: req.PromptID, ProfileID: req.ProfileID, + }, nil +} + +func (r *indexedAnalyzeFailureRunner) RenderArtifact(context.Context, scriptorium.RenderArtifactRequest) (scriptorium.ArtifactResult, error) { + return scriptorium.ArtifactResult{}, errors.New("unexpected render") +} + +type invalidAnalyzeOutputRunner struct{} + +func (invalidAnalyzeOutputRunner) RunArtifact(_ context.Context, req scriptorium.RunArtifactRequest) (scriptorium.ArtifactResult, error) { + if err := os.MkdirAll(filepath.Dir(req.OutputPath), 0o755); err != nil { + return scriptorium.ArtifactResult{}, err + } + target := req.OutputPath + ".target" + if err := os.WriteFile(target, []byte("unsafe\n"), 0o644); err != nil { + return scriptorium.ArtifactResult{}, err + } + if err := os.Symlink(target, req.OutputPath); err != nil { + return scriptorium.ArtifactResult{}, err + } + return scriptorium.ArtifactResult{OutputPath: req.OutputPath, CommandMode: scriptorium.CommandModeRun}, nil +} + +func (invalidAnalyzeOutputRunner) RenderArtifact(context.Context, scriptorium.RenderArtifactRequest) (scriptorium.ArtifactResult, error) { + return scriptorium.ArtifactResult{}, errors.New("unexpected render") +} + +type failingAnalyzeArtifactStore struct { + artifacts.Store + FailPath string + Err error +} + +func (s *failingAnalyzeArtifactStore) WriteFileAtomic(path string, data []byte, perm os.FileMode) error { + if filepath.Clean(path) == filepath.Clean(s.FailPath) { + return s.Err + } + return s.Store.WriteFileAtomic(path, data, perm) +} + +func independentAnalyzeArtifacts() map[string]config.ScriptoriumArtifactConfig { + return map[string]config.ScriptoriumArtifactConfig{ + "alpha": {Enabled: true, PromptID: "alpha", OutputPath: "artifacts/alpha.md"}, + "beta": {Enabled: true, PromptID: "beta", OutputPath: "artifacts/beta.md"}, + "gamma": {Enabled: true, PromptID: "gamma", OutputPath: "artifacts/gamma.md"}, + } +} + +func chainedAnalyzeArtifacts() map[string]config.ScriptoriumArtifactConfig { + return map[string]config.ScriptoriumArtifactConfig{ + "alpha": {Enabled: true, PromptID: "alpha", OutputPath: "artifacts/alpha.md"}, + "beta": { + Enabled: true, DependsOn: []string{"alpha"}, PromptID: "beta", OutputPath: "artifacts/beta.md", + Inputs: map[string]config.ScriptoriumInputConfig{"alpha": {Source: "narratio.artifact.alpha", Required: true}}, + }, + "gamma": { + Enabled: true, DependsOn: []string{"beta"}, PromptID: "gamma", OutputPath: "artifacts/gamma.md", + Inputs: map[string]config.ScriptoriumInputConfig{"beta": {Source: "narratio.artifact.beta", Required: true}}, + }, + "unrelated": {Enabled: true, PromptID: "unrelated", OutputPath: "artifacts/unrelated.md"}, + } +}