eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 17 Wiki Activity

Preserve incremental analysis state on failure

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-29 19:49:48 +00:00
parent 99f4f9a0db
commit 4e4e2b7d96
5 changed files with 365 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
internal/stage/analyze.go
View File

@@ -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 {

217
internal/stage/analyze_failure_test.go Normal file
View File

@@ -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"},
}
}