Verify Notarius cache lifecycle boundaries

docs/roadmap/implementation.md

@@ -173,7 +173,14 @@ policy, or permission regime was added.
 
 ## Stage 4 — Verify Session Lifecycle and Command Boundaries
 
-Status: Not started
+Status: Complete (2026-09-16)
+
+Completion: Added lifecycle coverage for stable retry snapshots, changed
+references, durable extract reuse after cache-control changes or snapshot loss,
+analysis-only preservation, and forced extraction/analysis regeneration. Added
+narrow cleanup, publish, and restore regressions proving the snapshot cache is
+retained and excluded from artifact transfer. Verified normal and race tests
+for `./internal/app ./internal/stage`.
 
 Read: `internal/app/extract_lifecycle_test.go`, `regenerate_artifacts.go`,
 `regenerate_artifacts_test.go`, `semantic_resume_test.go`,

internal/app/extract_lifecycle_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 	"sort"
 	"strings"
 	"testing"
@@ -398,6 +399,140 @@ func TestExtractLifecycleFailureInvalidatesAndOrdinaryRetryRerunsDownstream(t *t
 	}
 }
 
+func TestExtractLifecycleRetryReusesStableSnapshotsAndChangedReferenceSelectsNewPath(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	configureLifecycleReferences(t, cfg)
+	runner.failuresRemaining = 1
+	plan := prepareExtractLifecyclePlan(t)
+
+	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil {
+		t.Fatal("first executeStages() error = nil, want extraction failure")
+	}
+	if len(runner.requests) != 1 {
+		t.Fatalf("failed requests = %d, want 1", len(runner.requests))
+	}
+	firstPaths := lifecycleReferencePaths(runner.requests[0])
+	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
+		t.Fatalf("retry executeStages() error = %v", err)
+	}
+	if len(runner.requests) != 2 {
+		t.Fatalf("retry requests = %d, want 2", len(runner.requests))
+	}
+	if secondPaths := lifecycleReferencePaths(runner.requests[1]); !reflect.DeepEqual(secondPaths, firstPaths) {
+		t.Fatalf("snapshot paths changed on retry:\nfirst %#v\nsecond %#v", firstPaths, secondPaths)
+	}
+
+	partyDocument, err := config.ParseParty([]byte(`schema_version: narratio.party.v1
+characters:
+  changed:
+    player: {name: Eric}
+    character:
+      name: Changed
+      classes: [{name: wizard}]
+`))
+	if err != nil {
+		t.Fatalf("ParseParty(changed party) error = %v", err)
+	}
+	cfg.Party.Canonical = partyDocument
+	persisted := loadLifecycleManifest(t, cfg)
+	prepare, err := stage.Select("prepare")
+	if err != nil {
+		t.Fatalf("stage.Select(prepare) error = %v", err)
+	}
+	if _, err := prepare.Run(context.Background(), env, persisted); err != nil {
+		t.Fatalf("prepare.Run() error = %v", err)
+	}
+	if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), persisted); err != nil {
+		t.Fatalf("Save(reprepared manifest) error = %v", err)
+	}
+	extractPlan, _ := BuildSingleStagePlan("extract")
+	if _, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env}); err != nil {
+		t.Fatalf("changed-reference executeStages() error = %v", err)
+	}
+	if len(runner.requests) != 3 {
+		t.Fatalf("changed-reference requests = %d, want 3", len(runner.requests))
+	}
+	thirdPaths := lifecycleReferencePaths(runner.requests[2])
+	if thirdPaths["party"] == firstPaths["party"] {
+		t.Fatalf("changed party retained snapshot path %q", thirdPaths["party"])
+	}
+	if thirdPaths["players"] == firstPaths["players"] {
+		t.Fatalf("changed party retained derived players snapshot path %q", thirdPaths["players"])
+	}
+	for selector, firstPath := range firstPaths {
+		if selector != "party" && selector != "players" && thirdPaths[selector] != firstPath {
+			t.Fatalf("unchanged %s snapshot path = %q, want %q", selector, thirdPaths[selector], firstPath)
+		}
+		if _, err := os.Stat(firstPath); err != nil {
+			t.Fatalf("old %s snapshot %q was not retained: %v", selector, firstPath, err)
+		}
+	}
+}
+
+func TestExtractLifecycleCacheControlsAndSnapshotLossDoNotInvalidateCompleteResult(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	configureLifecycleReferences(t, cfg)
+	if _, err := executeStages(context.Background(), cfg, prepareExtractLifecyclePlan(t), RunOptions{Env: env}); err != nil {
+		t.Fatalf("initial executeStages() error = %v", err)
+	}
+	paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
+	cacheRoot := artifacts.SessionNotariusReferenceCacheDir(paths)
+	if err := os.RemoveAll(cacheRoot); err != nil {
+		t.Fatalf("RemoveAll(snapshot cache) error = %v", err)
+	}
+	cfg.Pipeline.Notarius.Resume = true
+	cfg.Pipeline.Notarius.ChunkCache = "bypass"
+	extractPlan, _ := BuildSingleStagePlan("extract")
+	summary, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env})
+	if err != nil {
+		t.Fatalf("resume executeStages() error = %v", err)
+	}
+	if len(summary.Executed) != 0 || !reflect.DeepEqual(summary.Skipped, []string{"extract"}) || len(runner.requests) != 1 {
+		t.Fatalf("resume summary = %#v requests=%d", summary, len(runner.requests))
+	}
+	if _, err := os.Stat(cacheRoot); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("skipped extraction recreated snapshot cache: %v", err)
+	}
+}
+
+func TestAnalyzeOnlyLifecyclePreservesExtractionWithoutInvokingNotarius(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	extractPlan, _ := BuildSingleStagePlan("extract")
+	if _, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env}); err != nil {
+		t.Fatalf("extract executeStages() error = %v", err)
+	}
+	before := loadLifecycleManifest(t, cfg).Stages["extract"]
+	beforeJSON, _ := json.Marshal(before)
+	analyzeRuns := 0
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "analyze", runs: &analyzeRuns}}, RunOptions{Env: env, Force: true}); err != nil {
+		t.Fatalf("analyze-only executeStages() error = %v", err)
+	}
+	after := loadLifecycleManifest(t, cfg).Stages["extract"]
+	afterJSON, _ := json.Marshal(after)
+	if analyzeRuns != 1 || len(runner.requests) != 1 || string(afterJSON) != string(beforeJSON) {
+		t.Fatalf("analyze-only lifecycle changed extraction: analyze=%d requests=%d\nbefore=%s\nafter=%s", analyzeRuns, len(runner.requests), beforeJSON, afterJSON)
+	}
+	for _, output := range after.Outputs {
+		if _, err := os.Stat(output.LocalPath); err != nil {
+			t.Fatalf("analyze-only lifecycle removed extract output %q: %v", output.LocalPath, err)
+		}
+	}
+}
+
+func TestForcedArtifactRegenerationUsesOrdinaryExtractAndAnalyzePath(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	cfg.Pipeline.Notarius.Resume = true
+	cfg.Pipeline.Notarius.ChunkCache = "bypass"
+	analyzeRuns := 0
+	plan := extractionLifecyclePlan(t, &analyzeRuns)
+	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env, Force: true, SelectedArtifacts: []string{"session_recap"}}); err != nil {
+		t.Fatalf("forced artifact regeneration error = %v", err)
+	}
+	if len(runner.requests) != 1 || runner.requests[0].Resume || runner.requests[0].ChunkCache != "refresh" || analyzeRuns != 1 {
+		t.Fatalf("forced path request=%#v analyze=%d", runner.requests, analyzeRuns)
+	}
+}
+
 func TestExtractLifecycleForcedSelfSkipInvalidatesDownstream(t *testing.T) {
 	cfg, env, _ := extractionLifecycleFixture(t, false)
 	markLifecycleStageSucceeded(t, cfg, "analyze")
@@ -787,6 +922,14 @@ func lifecycleInputChecksum(t *testing.T, m *manifest.Manifest, kind string) str
 	return ""
 }
 
+func lifecycleReferencePaths(request notarius.RunRequest) map[string]string {
+	paths := make(map[string]string, len(request.References))
+	for _, reference := range request.References {
+		paths[reference.Selector] = reference.Path
+	}
+	return paths
+}
+
 func lifecycleDndOutputs() map[string]config.NotariusOutputConfig {
 	return map[string]config.NotariusOutputConfig{
 		"item_registry":        {LaneID: "item-registry", MediaType: "application/json", SchemaID: "notarius.dnd.item_registry", SchemaVersion: "v1", ModuleKey: "dnd/item-registry"},

internal/app/restore_plan_test.go

@@ -24,6 +24,7 @@ func TestRestorePlanDefaultScope(t *testing.T) {
 	seedCurrentRestoreObject(t, store, current, current.SessionPrefix+"audio/alice.flac", []byte("audio"))
 	seedRestoreObject(store, current.SessionPrefix+"runs/20260519T010203Z-a1b2/manifest.json", []byte("{}"))
 	seedRestoreObject(store, current.SessionPrefix+"logs/publish.log", []byte("log"))
+	seedRestoreObject(store, current.SessionPrefix+"cache/notarius/references/ignored/party.yml", []byte("cache"))
 
 	plan, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{})
 	if err != nil {

internal/stage/cleanup_test.go

@@ -58,9 +58,12 @@ func TestCleanupStageRemovesOnlySelectedPublishedRunTargets(t *testing.T) {
 			otherSpool := filepath.Join(env.Config.Pipeline.Spool.Root, m.Campaign, m.SessionID, "other-run", "audio")
 			otherWorkspace := filepath.Join(env.Config.Pipeline.Workspace.Root, "work", m.Campaign, m.SessionID, "runs", "other-run")
 			cacheFile := filepath.Join(env.Config.Pipeline.Cache.Root, "s3", "audio.cache")
+			sessionPaths := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).SessionPathsFor(m.Campaign, m.SessionID)
+			referenceCacheFile := filepath.Join(artifacts.SessionNotariusReferenceCacheDir(sessionPaths), strings.Repeat("a", 64), "party.yml")
 			writeStageTestFile(t, filepath.Join(otherSpool, "audio.wav"), "other spool")
 			writeStageTestFile(t, filepath.Join(otherWorkspace, "result.txt"), "other workspace")
 			writeStageTestFile(t, cacheFile, "cache")
+			writeStageTestFile(t, referenceCacheFile, "reference cache")
 
 			result, err := (cleanupStage{}).Run(context.Background(), env, m)
 			if err != nil {
@@ -81,7 +84,7 @@ func TestCleanupStageRemovesOnlySelectedPublishedRunTargets(t *testing.T) {
 					t.Fatalf("unselected workspace target changed: %v", err)
 				}
 			}
-			for _, preserved := range []string{otherSpool, otherWorkspace, cacheFile} {
+			for _, preserved := range []string{otherSpool, otherWorkspace, cacheFile, referenceCacheFile} {
 				if _, err := os.Stat(preserved); err != nil {
 					t.Fatalf("preserved path %q: %v", preserved, err)
 				}

internal/stage/publish_test.go

@@ -324,6 +324,9 @@ func TestPublishRejectsDeclaredDirectoryBeforeUploading(t *testing.T) {
 func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) {
 	env, m, runRoot := publishFixture(t)
 	writeStageTestFile(t, filepath.Join(runRoot, "unlisted.txt"), "outside of manifest\n")
+	paths := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).SessionPathsFor(m.Campaign, m.SessionID)
+	cacheFile := filepath.Join(artifacts.SessionNotariusReferenceCacheDir(paths), strings.Repeat("a", 64), "party.yml")
+	writeStageTestFile(t, cacheFile, "local Notarius reference cache\n")
 
 	if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil {
 		t.Fatalf("Run() error = %v", err)
@@ -333,8 +336,8 @@ func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) {
 		t.Fatal("undeclared run file was uploaded")
 	}
 	for _, object := range fake.Objects {
-		if strings.Contains(string(object.Data), "outside of manifest") {
-			t.Fatalf("undeclared bytes were uploaded in %q", object.Key)
+		if strings.Contains(string(object.Data), "outside of manifest") || strings.Contains(string(object.Data), "local Notarius reference cache") {
+			t.Fatalf("undeclared or cache bytes were uploaded in %q", object.Key)
 		}
 	}
 }