Checkpoint analyze artifacts incrementally

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)
 	}

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)
+	}
+}

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
 		}
 	}

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)
+}

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 {

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"])
-	}
-}

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
 }