From 74977b40648e31aff4ae11d9070171f39e487002 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 4 Sep 2026 16:04:59 +0000 Subject: [PATCH] Checkpoint analyze artifacts incrementally --- docs/roadmap/implementation.md | 2 +- internal/app/runner_stage_execution.go | 10 +- internal/app/runner_terminal_failure_test.go | 52 +++++ internal/stage/analyze.go | 5 +- internal/stage/analyze_checkpoint_test.go | 182 ++++++++++++++++++ internal/stage/analyze_execution_state.go | 79 +++++++- .../analyze_incremental_execution_test.go | 15 +- internal/stage/analyze_test.go | 33 +++- 8 files changed, 343 insertions(+), 35 deletions(-) create mode 100644 internal/stage/analyze_checkpoint_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 974be6d..0ceae17 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -236,7 +236,7 @@ the existing end-of-stage persistence boundary temporarily. ## Stage 3 — Cut Analyze Over To Durable Per-Artifact Checkpoints -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/app/runner_stage_execution.go b/internal/app/runner_stage_execution.go index 4d04977..5b337fa 100644 --- a/internal/app/runner_stage_execution.go +++ b/internal/app/runner_stage_execution.go @@ -134,11 +134,11 @@ func executeSelectedStage(runtime *invocationRuntime, selected stage.Stage) (sta 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 { - operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after analyze projection persistence failure: %w", invalidationErr)) - } + } + failedAt := runtime.dependencies.now() + m.MarkStageFailed(selected.Name(), failedAt, operationErr.Error()) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), failedAt, staleReasonFailure); invalidationErr != nil { + operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after stage %q persistence failure: %w", selected.Name(), invalidationErr)) } return stageExecutionOutcome{}, runtime.persistTerminalFailure(operationErr) } diff --git a/internal/app/runner_terminal_failure_test.go b/internal/app/runner_terminal_failure_test.go index ed9a3a1..8635eb5 100644 --- a/internal/app/runner_terminal_failure_test.go +++ b/internal/app/runner_terminal_failure_test.go @@ -17,6 +17,31 @@ type terminalSessionStore struct { saved *manifest.Manifest } +type successfulStageSaveFailingStore struct { + delegate manifest.Store + stageName string + dependent string + failed bool +} + +func (s *successfulStageSaveFailingStore) Create(ctx context.Context, sessionID string) (*manifest.Manifest, error) { + return s.delegate.Create(ctx, sessionID) +} + +func (s *successfulStageSaveFailingStore) Load(ctx context.Context, path string) (*manifest.Manifest, error) { + return s.delegate.Load(ctx, path) +} + +func (s *successfulStageSaveFailingStore) Save(ctx context.Context, path string, m *manifest.Manifest) error { + record := m.Stages[s.stageName] + if !s.failed && record != nil && record.Status == manifest.StatusSucceeded { + s.failed = true + m.MarkStageSucceeded(s.dependent, time.Now().UTC(), nil) + return errors.New("injected successful-stage save failure") + } + return s.delegate.Save(ctx, path, m) +} + func (s *terminalSessionStore) Create(_ context.Context, sessionID string) (*manifest.Manifest, error) { return manifest.New(sessionID, time.Now().UTC()), nil } @@ -114,3 +139,30 @@ func TestExecuteStagesResumeValidationFailureRecordsSessionFailure(t *testing.T) t.Fatalf("session manifest = %#v, want preserved success with terminal error", persisted) } } + +func TestExecuteStagesSuccessfulSaveFailureMarksStageFailedAndInvalidatesDependents(t *testing.T) { + cfg := testConfig(t) + store := &successfulStageSaveFailingStore{ + delegate: &manifest.LocalStore{}, + stageName: "polish", + dependent: "normalize", + } + + _, err := executeStages(context.Background(), cfg, []stage.Stage{ + countingStage{name: "polish", runs: new(int)}, + }, RunOptions{Env: &Env{ManifestStore: store}}) + if err == nil || !strings.Contains(err.Error(), "save manifest after stage \"polish\"") { + t.Fatalf("executeStages() error = %v, want successful-stage save failure", err) + } + + persisted, loadErr := store.delegate.Load(context.Background(), manifestPathFor(cfg)) + if loadErr != nil { + t.Fatalf("load session manifest: %v", loadErr) + } + if got := persisted.Stages["polish"]; got == nil || got.Status != manifest.StatusFailed { + t.Fatalf("polish stage = %#v, want failed", got) + } + if got := persisted.Stages["normalize"]; got == nil || got.Status != manifest.StatusStale { + t.Fatalf("normalize stage = %#v, want stale", got) + } +} diff --git a/internal/stage/analyze.go b/internal/stage/analyze.go index 68b378a..9bb0073 100644 --- a/internal/stage/analyze.go +++ b/internal/stage/analyze.go @@ -143,6 +143,9 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S } state := newAnalyzeExecutionState(execution, workPlan, reconciliation) + if err := state.checkpointStartingCollection(ctx); err != nil { + return nil, fmt.Errorf("analyze: checkpoint reconciled artifact collection before execution: %w", err) + } for _, item := range workPlan.ExecutionOrder { artifactCfg := env.Config.Pipeline.Scriptorium.Artifacts[item.Key] @@ -160,7 +163,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err != nil { return state.failedResult(item, artifactCfg, fingerprint, err), err } - if err := state.acceptArtifact(plan, fingerprint, artifactResult); err != nil { + if err := state.acceptArtifact(ctx, plan, fingerprint, artifactResult); err != nil { return state.failedResult(item, artifactCfg, fingerprint, err), err } } diff --git a/internal/stage/analyze_checkpoint_test.go b/internal/stage/analyze_checkpoint_test.go new file mode 100644 index 0000000..c6e4150 --- /dev/null +++ b/internal/stage/analyze_checkpoint_test.go @@ -0,0 +1,182 @@ +package stage + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "sort" + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestAnalyzeCheckpointsCompletedArtifactsBeforeInvokingNext(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + env.Config.Pipeline.Scriptorium.Artifacts = independentAnalyzeArtifacts() + manifestPath := artifacts.SessionManifestPathForCampaign( + env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, + ) + runner := &checkpointObservingRunner{Store: env.ManifestStore, ManifestPath: manifestPath} + env.Scriptorium = runner + + if _, err := (analyzeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } + want := [][]string{nil, {"alpha"}, {"alpha", "beta"}} + if !reflect.DeepEqual(runner.CurrentBeforeCall, want) { + t.Fatalf("current records before calls = %#v, want %#v", runner.CurrentBeforeCall, want) + } +} + +func TestAnalyzeCheckpointFailureStopsLaterWorkWithoutPromotingOutput(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + env.Config.Pipeline.Scriptorium.Artifacts = independentAnalyzeArtifacts() + cause := errors.New("injected checkpoint failure") + baseStore := env.ManifestStore + env.ManifestStore = &failingAnalyzeManifestStore{Store: baseStore, FailAt: 2, Err: cause} + runner := &indexedAnalyzeFailureRunner{} + env.Scriptorium = runner + + _, err := (analyzeStage{}).Run(context.Background(), env, m) + if !errors.Is(err, cause) { + t.Fatalf("Run() error = %v, want checkpoint cause", err) + } + if runner.Calls != 2 { + t.Fatalf("adapter calls = %d, want alpha and beta only", runner.Calls) + } + paths := sessionPathsForEnv(env, m.SessionID) + if _, statErr := os.Stat(filepath.Join(paths.ArtifactsDir, "beta.md")); statErr != nil { + t.Fatalf("materialized beta output missing: %v", statErr) + } + manifestPath := artifacts.SessionManifestPathForCampaign( + env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, + ) + durable, loadErr := baseStore.Load(context.Background(), manifestPath) + if loadErr != nil { + t.Fatalf("load durable manifest: %v", loadErr) + } + records := durable.Stages["analyze"].AnalyzeArtifacts + if records["alpha"].Status != manifest.AnalyzeArtifactCurrent { + t.Fatalf("durable alpha = %#v, want current", records["alpha"]) + } + if beta, exists := records["beta"]; exists && beta.Status == manifest.AnalyzeArtifactCurrent { + t.Fatalf("durable beta = %#v, incidental output became authoritative", beta) + } + if _, exists := records["gamma"]; exists { + t.Fatalf("unattempted gamma was persisted: %#v", records["gamma"]) + } + if beta := m.Stages["analyze"].AnalyzeArtifacts["beta"]; beta.Status != manifest.AnalyzeArtifactFailed || beta.Output != nil { + t.Fatalf("live beta = %#v, want failed without output for terminal save", beta) + } +} + +func TestAnalyzeReconciledCheckpointFailurePreventsFirstAdapterCall(t *testing.T) { + env, m, fake := setupAnalyzeEnv(t) + if _, err := (analyzeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("seed Run() error = %v", err) + } + env.Force = true + fake.RunRequests = nil + cause := errors.New("injected starting checkpoint failure") + env.ManifestStore = &failingAnalyzeManifestStore{Store: env.ManifestStore, FailAt: 1, Err: cause} + + _, err := (analyzeStage{}).Run(context.Background(), env, m) + if !errors.Is(err, cause) { + t.Fatalf("Run() error = %v, want starting checkpoint cause", err) + } + if len(fake.RunRequests) != 0 { + t.Fatalf("adapter requests = %d, want none before starting checkpoint", len(fake.RunRequests)) + } +} + +func TestAnalyzeOutputsIncludeEveryCurrentRecordWithOriginalProducer(t *testing.T) { + env, m, _ := setupAnalyzeEnv(t) + env.Config.Pipeline.Scriptorium.Artifacts = independentAnalyzeArtifacts() + if _, err := (analyzeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("seed Run() error = %v", err) + } + originalRunID := m.RunID + m.RunID = "replacement-run" + env.SelectedArtifactKeys = []string{"beta"} + env.Force = true + env.Scriptorium = &indexedAnalyzeFailureRunner{} + + result, err := (analyzeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("replacement Run() error = %v", err) + } + wantSources := []string{ + "narratio.artifact.alpha", "narratio.artifact.beta", "narratio.artifact.gamma", + } + gotSources := make([]string, 0, len(result.Outputs)) + gotProducers := make([]string, 0, len(result.Outputs)) + for _, output := range result.Outputs { + gotSources = append(gotSources, output.SourceID) + gotProducers = append(gotProducers, output.ProducerRunID) + } + if !reflect.DeepEqual(gotSources, wantSources) { + t.Fatalf("output sources = %#v, want %#v", gotSources, wantSources) + } + wantProducers := []string{originalRunID, "replacement-run", originalRunID} + if !reflect.DeepEqual(gotProducers, wantProducers) { + t.Fatalf("output producer run ids = %#v, want %#v", gotProducers, wantProducers) + } +} + +type checkpointObservingRunner struct { + Store manifest.Store + ManifestPath string + CurrentBeforeCall [][]string +} + +func (r *checkpointObservingRunner) RunArtifact( + ctx context.Context, + req scriptorium.RunArtifactRequest, +) (scriptorium.ArtifactResult, error) { + m, err := r.Store.Load(ctx, r.ManifestPath) + if err != nil { + return scriptorium.ArtifactResult{}, err + } + current := make([]string, 0) + for key, record := range m.Stages["analyze"].AnalyzeArtifacts { + if record.Status == manifest.AnalyzeArtifactCurrent { + current = append(current, key) + } + } + sort.Strings(current) + if len(current) == 0 { + current = nil + } + r.CurrentBeforeCall = append(r.CurrentBeforeCall, current) + writeAnalyzeFileNoTest(req.OutputPath, "generated "+req.PromptID+"\n") + return scriptorium.ArtifactResult{ + OutputPath: req.OutputPath, CommandMode: scriptorium.CommandModeRun, + PromptID: req.PromptID, ProfileID: req.ProfileID, + }, nil +} + +func (*checkpointObservingRunner) RenderArtifact( + context.Context, + scriptorium.RenderArtifactRequest, +) (scriptorium.ArtifactResult, error) { + return scriptorium.ArtifactResult{}, errors.New("unexpected render") +} + +type failingAnalyzeManifestStore struct { + manifest.Store + Saves int + FailAt int + Err error +} + +func (s *failingAnalyzeManifestStore) Save(ctx context.Context, path string, m *manifest.Manifest) error { + s.Saves++ + if s.Saves == s.FailAt { + return s.Err + } + return s.Store.Save(ctx, path, m) +} diff --git a/internal/stage/analyze_execution_state.go b/internal/stage/analyze_execution_state.go index 9dc094a..a167451 100644 --- a/internal/stage/analyze_execution_state.go +++ b/internal/stage/analyze_execution_state.go @@ -1,7 +1,9 @@ package stage import ( + "context" "fmt" + "reflect" "sort" "strings" "time" @@ -16,6 +18,7 @@ type analyzeExecutionState struct { execution analyzeExecutionContext workPlan analyzeWorkPlan records map[string]manifest.AnalyzeArtifactRecord + committedRecords map[string]manifest.AnalyzeArtifactRecord priorCurrentRecords map[string]manifest.AnalyzeArtifactRecord evaluatedKeys map[string]struct{} catalog *artifacts.ArtifactCatalog @@ -41,6 +44,11 @@ func newAnalyzeExecutionState( artifactMetadata: make([]map[string]any, 0, len(workPlan.ExecutionOrder)), reusedSeen: make(map[string]struct{}), } + if execution.Manifest != nil && execution.Manifest.Stages["analyze"] != nil { + state.committedRecords = manifest.CloneAnalyzeArtifactCollection( + execution.Manifest.Stages["analyze"].AnalyzeArtifacts, + ) + } for _, item := range reconciliation.Ordered { if item.Stored != nil && item.Stored.Status == manifest.AnalyzeArtifactCurrent { state.priorCurrentRecords[item.Key] = *item.Stored @@ -56,6 +64,7 @@ func newAnalyzeExecutionState( } func (state *analyzeExecutionState) acceptArtifact( + ctx context.Context, plan analyzeArtifactExecutionPlan, fingerprint string, result *analyzeArtifactExecutionResult, @@ -90,6 +99,67 @@ func (state *analyzeExecutionState) acceptArtifact( state.records, ) } + if err := state.checkpointCollection(ctx, state.records); err != nil { + state.records = manifest.CloneAnalyzeArtifactCollection(state.committedRecords) + return fmt.Errorf("analyze: checkpoint artifact %q: %w", plan.Name, err) + } + return nil +} + +func (state *analyzeExecutionState) checkpointStartingCollection(ctx context.Context) error { + m := state.execution.Manifest + if m != nil { + if record := m.Stages["analyze"]; record != nil && + record.AnalyzeStateVersion == manifest.AnalyzeStateContractVersion && + reflect.DeepEqual(record.AnalyzeArtifacts, state.records) { + return nil + } + } + return state.checkpointCollection(ctx, state.records) +} + +func (state *analyzeExecutionState) checkpointCollection( + ctx context.Context, + records map[string]manifest.AnalyzeArtifactRecord, +) error { + m := state.execution.Manifest + if m == nil { + return fmt.Errorf("session manifest is required") + } + if m.Stages == nil || m.Stages["analyze"] == nil { + return fmt.Errorf("analyze stage record is required") + } + if state.execution.Env == nil || state.execution.Env.ManifestStore == nil { + return fmt.Errorf("manifest store is required") + } + if err := manifest.ValidateAnalyzeArtifactCollection(manifest.AnalyzeStateContractVersion, records); err != nil { + return fmt.Errorf("validate analyze artifact collection: %w", err) + } + + candidate := *m + candidate.Stages = make(map[string]*manifest.StageRecord, len(m.Stages)) + for name, record := range m.Stages { + candidate.Stages[name] = record + } + analyzeRecord := *m.Stages["analyze"] + analyzeRecord.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion + analyzeRecord.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(records) + candidate.Stages["analyze"] = &analyzeRecord + + manifestPath := artifacts.SessionManifestPathForCampaign( + state.execution.Paths.WorkspaceRoot, + m.Campaign, + m.SessionID, + ) + if err := state.execution.Env.ManifestStore.Save(ctx, manifestPath, &candidate); err != nil { + return fmt.Errorf("save session manifest %q: %w", manifestPath, err) + } + liveRecord := m.Stages["analyze"] + liveRecord.AnalyzeStateVersion = candidate.Stages["analyze"].AnalyzeStateVersion + liveRecord.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(candidate.Stages["analyze"].AnalyzeArtifacts) + m.UpdatedAt = candidate.UpdatedAt + state.records = manifest.CloneAnalyzeArtifactCollection(liveRecord.AnalyzeArtifacts) + state.committedRecords = manifest.CloneAnalyzeArtifactCollection(liveRecord.AnalyzeArtifacts) return nil } @@ -143,7 +213,7 @@ func (state *analyzeExecutionState) failedResult( state.records[item.Key] = record staleAnalyzeDependents(artifactCfgMap(state.execution), item.Key, state.records) state.installManifestCollection() - return &StageResult{AnalyzeState: state.projection()} + return &StageResult{} } func (state *analyzeExecutionState) successResult() *StageResult { @@ -162,12 +232,11 @@ func (state *analyzeExecutionState) successResult() *StageResult { metadata[key] = value } } - state.installManifestCollection() return &StageResult{ + Outputs: analyzeArtifactRefs(state.records), Logs: dedupeAndSortPaths(state.logs), GeneratedConfigs: dedupeAndSortPaths(state.generatedConfigs), Metadata: metadata, - AnalyzeState: state.projection(), } } @@ -188,10 +257,6 @@ func (state *analyzeExecutionState) installManifestCollection() { record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(state.records) } -func (state *analyzeExecutionState) projection() *AnalyzeStateProjection { - return &AnalyzeStateProjection{Session: manifest.CloneAnalyzeArtifactCollection(state.records)} -} - func analyzeArtifactRefs(records map[string]manifest.AnalyzeArtifactRecord) []artifacts.Ref { keys := make([]string, 0, len(records)) for key, record := range records { diff --git a/internal/stage/analyze_incremental_execution_test.go b/internal/stage/analyze_incremental_execution_test.go index 6a9f89a..5f0dbce 100644 --- a/internal/stage/analyze_incremental_execution_test.go +++ b/internal/stage/analyze_incremental_execution_test.go @@ -15,11 +15,10 @@ func TestAnalyzeReusesCurrentArtifactWithoutScriptorium(t *testing.T) { paths := sessionPathsForEnv(env, m.SessionID) writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`) - first, err := (analyzeStage{}).Run(context.Background(), env, m) + _, err := (analyzeStage{}).Run(context.Background(), env, m) if err != nil { t.Fatal(err) } - assertAnalyzeProjectionMatchesManifest(t, m, first.AnalyzeState) if len(fake.RunRequests) != 1 { t.Fatalf("initial requests = %d, want 1", len(fake.RunRequests)) } @@ -235,15 +234,3 @@ func addAnalyzeDependentArtifact(env *Env) { }, } } - -func assertAnalyzeProjectionMatchesManifest( - t *testing.T, - m *manifest.Manifest, - projection *AnalyzeStateProjection, -) { - t.Helper() - if projection == nil || m.Stages["analyze"] == nil || - !reflect.DeepEqual(projection.Session, m.Stages["analyze"].AnalyzeArtifacts) { - t.Fatalf("projection = %#v, manifest records = %#v", projection, m.Stages["analyze"]) - } -} diff --git a/internal/stage/analyze_test.go b/internal/stage/analyze_test.go index 76a3e4d..9185a52 100644 --- a/internal/stage/analyze_test.go +++ b/internal/stage/analyze_test.go @@ -34,6 +34,9 @@ func TestAnalyzeGeneratesSessionRecapFromTrimmedTranscript(t *testing.T) { } req := fake.RunRequests[0] + runRoot := artifacts.SessionRunStageDirForCampaign( + env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID, "analyze", + ) if req.PromptID != "dnd.session_recap" { t.Fatalf("prompt id = %q, want dnd.session_recap", req.PromptID) } @@ -43,16 +46,16 @@ func TestAnalyzeGeneratesSessionRecapFromTrimmedTranscript(t *testing.T) { if req.InputPaths["transcript"] != filepath.Join(paths.TranscriptsDir, "final.trimmed.json") { t.Fatalf("transcript input = %q, want trimmed transcript path", req.InputPaths["transcript"]) } - if req.OutputPath != filepath.Join(paths.ArtifactsDir, "session_recap.md") { - t.Fatalf("output path = %q, want %q", req.OutputPath, filepath.Join(paths.ArtifactsDir, "session_recap.md")) + if req.OutputPath != filepath.Join(runRoot, "outputs", "artifacts", "session_recap.md") { + t.Fatalf("output path = %q, want run-local output", req.OutputPath) } - if req.StdoutLogPath != filepath.Join(paths.LogsDir, "scriptorium.session_recap.stdout.log") { + if req.StdoutLogPath != filepath.Join(runRoot, "logs", "scriptorium.session_recap.stdout.log") { t.Fatalf("stdout log path = %q", req.StdoutLogPath) } - if req.StderrLogPath != filepath.Join(paths.LogsDir, "scriptorium.session_recap.stderr.log") { + if req.StderrLogPath != filepath.Join(runRoot, "logs", "scriptorium.session_recap.stderr.log") { t.Fatalf("stderr log path = %q", req.StderrLogPath) } - if req.GeneratedConfigPath != filepath.Join(paths.ConfigDir, "scriptorium.session_recap.generated.yml") { + if req.GeneratedConfigPath != filepath.Join(runRoot, "config", "scriptorium.session_recap.generated.yml") { t.Fatalf("generated config path = %q", req.GeneratedConfigPath) } if req.Timeout != 2*time.Minute { @@ -183,7 +186,10 @@ func TestAnalyzeRenderOutputPathIsRecorded(t *testing.T) { if fake.RunRequests[0].Vars["session_id"] != "narratio-session-2026-05-03" { t.Fatalf("run session_id var = %q, want sticky narratio session id", fake.RunRequests[0].Vars["session_id"]) } - if result.Metadata["render_output_path"] != filepath.Join(paths.ArtifactsDir, "session_recap.render.json") { + runRoot := artifacts.SessionRunStageDirForCampaign( + env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID, "analyze", + ) + if result.Metadata["render_output_path"] != filepath.Join(runRoot, "reports", "session_recap.render.json") { t.Fatalf("render_output_path = %#v, want session_recap.render.json path", result.Metadata["render_output_path"]) } if len(result.Logs) != 4 { @@ -1728,10 +1734,16 @@ func setupAnalyzeEnv(t *testing.T) (*Env, *manifest.Manifest, *scriptorium.FakeR env := &Env{ Config: cfg, ArtifactStore: store, + ManifestStore: &manifest.LocalStore{}, Scriptorium: fake, } - m := manifest.New("2026-05-03", time.Now().UTC()) paths := store.SessionPathsFor("icewind-dale", "2026-05-03") + now := time.Now().UTC() + m := manifest.New("2026-05-03", now) + m.Campaign = "icewind-dale" + m.RunID = "analyze-test-run" + m.LocalWorkDir = paths.Root + m.MarkStageRunning("analyze", now) trimmedPath := filepath.Join(paths.TranscriptsDir, "final.trimmed.json") writeAnalyzeFile(t, trimmedPath, `{"segments":[]}`) checksum, err := artifacts.SHA256File(trimmedPath) @@ -1742,6 +1754,13 @@ func setupAnalyzeEnv(t *testing.T) (*Env, *manifest.Manifest, *scriptorium.FakeR Kind: artifacts.TranscriptOutputKindFinalTrimmed, SourceID: artifacts.ArtifactTranscriptFinalTrimmed, LocalPath: trimmedPath, ProducerRunID: "trim-run", Checksum: checksum, }}) + if err := env.ManifestStore.Save( + context.Background(), + artifacts.SessionManifestPathForCampaign(workspace, m.Campaign, m.SessionID), + m, + ); err != nil { + t.Fatalf("save initial analyze manifest: %v", err) + } return env, m, fake }