diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index c432f7a..4d2fc94 100644 --- a/docs/roadmap/implementation.md +++ b/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`, diff --git a/internal/app/extract_lifecycle_test.go b/internal/app/extract_lifecycle_test.go index d780210..ef0b985 100644 --- a/internal/app/extract_lifecycle_test.go +++ b/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"}, diff --git a/internal/app/restore_plan_test.go b/internal/app/restore_plan_test.go index 2d933f3..32c98d9 100644 --- a/internal/app/restore_plan_test.go +++ b/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 { diff --git a/internal/stage/cleanup_test.go b/internal/stage/cleanup_test.go index 419f91d..9320e13 100644 --- a/internal/stage/cleanup_test.go +++ b/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) } diff --git a/internal/stage/publish_test.go b/internal/stage/publish_test.go index 0df7efd..1dc18db 100644 --- a/internal/stage/publish_test.go +++ b/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) } } }