From eac7e155a543e772d00efe6b379df15ad358b84c Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 20:29:00 +0000 Subject: [PATCH] Persist retryable post-publish cleanup obligations --- docs/internal/manifest.md | 10 + docs/internal/stage-publish.md | 3 + docs/internal/workspace.md | 8 +- docs/operations.md | 3 + docs/policy/architecture.md | 4 + docs/roadmap/implementation.md | 2 +- internal/app/post_publish_cleanup.go | 178 ++++++++-------- internal/app/post_publish_cleanup_test.go | 234 ++++++++++++++++++++-- internal/app/runner.go | 36 +++- internal/manifest/manifest.go | 50 +++-- internal/manifest/store.go | 27 +++ internal/stage/publish.go | 4 +- 12 files changed, 438 insertions(+), 121 deletions(-) diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index 0c48593..eddd024 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -16,6 +16,9 @@ Explain the session-progress and invocation-audit models implemented by - `inputs` records - durable `artifacts` records - per-stage `stages` map +- an optional `post_publish_cleanup` obligation, which binds a committed run, + remote commit identity, and each exact root-confined local target to its + completion evidence Session, campaign, and run identities in local and downloaded manifests must be portable opaque segments. Unsafe legacy identities are rejected with migration @@ -111,6 +114,13 @@ runner marks it stale and executes it. Session manifest is the authoritative stage-progress ledger across invocations. Run manifest is invocation-scoped audit state. +After a publish commits remotely, any configured local cleanup is first recorded +as a session-manifest obligation before deletion begins. Each target becomes +complete only after its confined deletion (or safe absence check) and a +successful manifest save. An incomplete obligation is retried on later +invocations independently of their selected stages and retains the committed +run and remote identity that authorized it. + Each invocation derives campaign, session, run, local-path, and remote-prefix metadata from the validated resolved configuration as one projection. A persisted session manifest must agree on campaign and session identity before execution; diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 6c71846..9065193 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -59,6 +59,7 @@ Includes counts/lists for: - skipped unselected outputs - locked outputs - remote commit and current-pointer key paths +- the run identifier selected by the commit ## Invariants @@ -72,6 +73,8 @@ Includes counts/lists for: archived only when recorded by the run manifest. - publish locks are not overridden by `--force`; remote locks are revalidated immediately before current-state selection. +- post-commit local cleanup is authorized by the committed publish metadata and + is durably recorded by the application lifecycle before any local deletion. The commit boundary and cleanup gate are normative architecture invariants; see [Architecture](../policy/architecture.md#publish-commit-boundary). diff --git a/docs/internal/workspace.md b/docs/internal/workspace.md index 1b507e3..b309eb3 100644 --- a/docs/internal/workspace.md +++ b/docs/internal/workspace.md @@ -54,13 +54,17 @@ lock only while their context remains active, and report a release failure. Automatic post-publish cleanup: -- only runs when publish actually executed and succeeded; +- is created only after a successful publish commit with complete publish + metadata, then is persisted before any deletion; - requires `uploaded=true`, a remote commit key, and a current commit-pointer key in publish metadata; - consumes the resolved cleanup policy described in [Configuration](../config.md); - refuses unsafe deletes (root delete, out-of-root delete, and symlinked - ancestors or entries). + ancestors or entries); +- retries any recorded incomplete target on later invocations even when no + publish work is selected. Missing targets are a successful, idempotent + cleanup result only after the completion evidence is saved. Manual cleanup uses the same root-confined deletion mechanism. Invocation syntax and exact deletion scope belong in [CLI](../cli.md#clean) and diff --git a/docs/operations.md b/docs/operations.md index acc37f0..2f816f5 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -381,6 +381,9 @@ Rules: - automatic post-publish cleanup is gated by successful publish commit plus: - `pipeline.spool.delete_audio_after_publish=true` - `pipeline.workspace.cleanup_after_publish=true` + - Narratio first records the exact run-scoped cleanup obligation. If cleanup + reports incomplete, the remote committed snapshot remains current; rerun + Narratio to retry only the outstanding confined local cleanup. ## Operational Caveats diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 79aa3b7..9e694fb 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -179,6 +179,10 @@ selection; loss of that check leaves the prior committed snapshot current. Automatic local cleanup is permitted only after a successful publish commit, only when explicitly configured, and only through the path-safety guardrails. +It is a durable local obligation bound to that committed run and its exact +targets, not an inferred side effect of the current stage list. A cleanup +failure makes the invocation incomplete while leaving the committed remote +snapshot authoritative; later invocations resume the recorded obligation. ## Security, Privacy, And Diagnostics diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index cf4e247..2ae3426 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -31,7 +31,7 @@ All stages are pending when this plan is created. | 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Completed | | 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Completed | | 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Completed | -| 16 | Persist retryable post-commit cleanup state | COR-006, COR-007 | Pending | +| 16 | Persist retryable post-commit cleanup state | COR-006, COR-007 | Completed | | 17 | Bind restore/status to a committed snapshot and reject conflicts | COR-008, COR-009, TST-005 | Pending | | 18 | Serialize restore transitions and make restored paths portable | RSK-006, RSK-008 | Pending | | 19 | Bind audio cache reuse to remote object identity | RSK-007 | Pending | diff --git a/internal/app/post_publish_cleanup.go b/internal/app/post_publish_cleanup.go index 791f9dd..47f080f 100644 --- a/internal/app/post_publish_cleanup.go +++ b/internal/app/post_publish_cleanup.go @@ -12,104 +12,104 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) +var removeRunScopedDirFn = removeRunScopedDir + func runPostPublishCleanup(ctx context.Context, env *Env, manifestPath string, m *manifest.Manifest, executed []string) error { if env == nil || env.Config == nil || env.Config.Pipeline == nil || m == nil { return nil } - spoolRequested := env.Config.Pipeline.Spool.DeleteAudioAfterPublish - workRequested := env.Config.Pipeline.Workspace.CleanupAfterPublish - if !spoolRequested && !workRequested { - return nil - } - - sr := publishStageRecordForCleanup(m, executed) - if sr == nil { - return nil - } - if sr.Metadata == nil { - sr.Metadata = map[string]any{} - } - sr.Metadata["spool_cleanup_requested"] = spoolRequested - sr.Metadata["workdir_cleanup_requested"] = workRequested - - eligible, reason := publishCleanupEligible(env.Config, sr) - if !eligible { - sr.Metadata["cleanup_skipped"] = true - sr.Metadata["cleanup_skipped_reason"] = reason - if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - return fmt.Errorf("save manifest cleanup skip metadata %q: %w", manifestPath, err) + cleanup := m.PostPublishCleanup + if cleanup == nil { + var err error + cleanup, err = createPostPublishCleanup(env.Config, m, executed) + if err != nil { + return err } - return nil + if cleanup == nil { + return nil + } + m.PostPublishCleanup = cleanup + if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { + return fmt.Errorf("persist post-publish cleanup obligation %q: %w", manifestPath, err) + } + } + + for index := range cleanup.Targets { + target := &cleanup.Targets[index] + if target.Completed { + continue + } + if err := removeRunScopedDirFn(target.Root, target.Path, target.Policy); err != nil { + return fmt.Errorf("complete post-publish cleanup for %q: %w", target.Path, err) + } + target.Completed = true + if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { + target.Completed = false + return fmt.Errorf("persist post-publish cleanup completion %q: %w", manifestPath, err) + } + } + return nil +} + +func createPostPublishCleanup(cfg *config.Config, m *manifest.Manifest, executed []string) (*manifest.PostPublishCleanup, error) { + spoolRequested := cfg.Pipeline.Spool.DeleteAudioAfterPublish + workRequested := cfg.Pipeline.Workspace.CleanupAfterPublish + if !spoolRequested && !workRequested { + return nil, nil + } + + sr := publishStageRecordForCleanup(m) + publishedRunID, eligible, _ := publishCleanupEligible(cfg, m, sr, executed) + if !eligible { + return nil, nil } spoolDir := strings.TrimSpace(m.LocalSpoolDir) if spoolDir == "" { spoolDir = artifacts.SessionSpoolAudioDir( - env.Config.Pipeline.Spool.Root, - strings.TrimSpace(env.Config.Session.Campaign), - strings.TrimSpace(env.Config.Session.SessionID), - strings.TrimSpace(m.RunID), + cfg.Pipeline.Spool.Root, + strings.TrimSpace(m.Campaign), + strings.TrimSpace(m.SessionID), + publishedRunID, ) } workDir := strings.TrimSpace(m.LocalWorkDir) if workDir == "" { workDir = artifacts.SessionRunRootForCampaign( - env.Config.Pipeline.Workspace.Root, - strings.TrimSpace(env.Config.Session.Campaign), - strings.TrimSpace(env.Config.Session.SessionID), - strings.TrimSpace(m.RunID), + cfg.Pipeline.Workspace.Root, + strings.TrimSpace(m.Campaign), + strings.TrimSpace(m.SessionID), + publishedRunID, ) } + cleanup := &manifest.PostPublishCleanup{ + CommittedRunID: publishedRunID, + RemoteCommitKey: strings.TrimSpace(asString(sr.Metadata["remote_commit_key"])), + CurrentCommitPointerKey: strings.TrimSpace(asString(sr.Metadata["current_commit_pointer_key"])), + } if spoolRequested { - if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Spool.Root), spoolDir, "pipeline.spool.delete_audio_after_publish"); err != nil { - sr.Metadata["cleanup_failed"] = true - sr.Metadata["cleanup_failed_policy"] = "pipeline.spool.delete_audio_after_publish" - sr.Metadata["cleanup_failed_path"] = spoolDir - _ = env.ManifestStore.Save(ctx, manifestPath, m) - return err - } - sr.Metadata["spool_cleanup_deleted"] = filepath.Clean(spoolDir) + cleanup.Targets = append(cleanup.Targets, manifest.CleanupTarget{ + Policy: "pipeline.spool.delete_audio_after_publish", + Root: strings.TrimSpace(cfg.Pipeline.Spool.Root), + Path: filepath.Clean(spoolDir), + }) } - - if !workRequested { - sr.Metadata["cleanup_completed"] = true - sr.Metadata["cleanup_skipped"] = false - if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - return fmt.Errorf("save manifest cleanup metadata %q: %w", manifestPath, err) - } - return nil + if workRequested { + cleanup.Targets = append(cleanup.Targets, manifest.CleanupTarget{ + Policy: "pipeline.workspace.cleanup_after_publish", + Root: strings.TrimSpace(cfg.Pipeline.Workspace.Root), + Path: filepath.Clean(workDir), + }) } - - if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Workspace.Root), workDir, "pipeline.workspace.cleanup_after_publish"); err != nil { - sr.Metadata["cleanup_failed"] = true - sr.Metadata["cleanup_failed_policy"] = "pipeline.workspace.cleanup_after_publish" - sr.Metadata["cleanup_failed_path"] = workDir - _ = env.ManifestStore.Save(ctx, manifestPath, m) - return err - } - - sr.Metadata["workdir_cleanup_deleted"] = filepath.Clean(workDir) - sr.Metadata["cleanup_completed"] = true - sr.Metadata["cleanup_skipped"] = false - return nil + return cleanup, nil } -func publishStageRecordForCleanup(m *manifest.Manifest, executed []string) *manifest.StageRecord { +func publishStageRecordForCleanup(m *manifest.Manifest) *manifest.StageRecord { if m == nil { return nil } - publishRan := false - for _, name := range executed { - if name == "publish" { - publishRan = true - break - } - } - if !publishRan { - return nil - } sr := m.Stages["publish"] if sr == nil || sr.Status != manifest.StatusSucceeded { return nil @@ -117,40 +117,56 @@ func publishStageRecordForCleanup(m *manifest.Manifest, executed []string) *mani return sr } -func publishCleanupEligible(cfg *config.Config, sr *manifest.StageRecord) (bool, string) { +func publishCleanupEligible(cfg *config.Config, m *manifest.Manifest, sr *manifest.StageRecord, executed []string) (string, bool, string) { if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Publish == nil { - return false, "publish configuration is missing" + return "", false, "publish configuration is missing" } enabled := true if cfg.Pipeline.Publish.Enabled != nil { enabled = *cfg.Pipeline.Publish.Enabled } if !enabled { - return false, "publish.enabled is false" + return "", false, "publish.enabled is false" } uploadRun := true if cfg.Pipeline.Publish.UploadRun != nil { uploadRun = *cfg.Pipeline.Publish.UploadRun } if !uploadRun { - return false, "publish.upload_run is false" + return "", false, "publish.upload_run is false" } if sr == nil || sr.Metadata == nil { - return false, "publish metadata is missing" + return "", false, "publish metadata is missing" } if skipped, _ := sr.Metadata["skipped"].(bool); skipped { - return false, "publish stage was skipped" + return "", false, "publish stage was skipped" } if uploaded, _ := sr.Metadata["uploaded"].(bool); !uploaded { - return false, "publish did not upload run record" + return "", false, "publish did not upload run record" } if strings.TrimSpace(asString(sr.Metadata["remote_commit_key"])) == "" { - return false, "publish remote commit key is missing" + return "", false, "publish remote commit key is missing" } if strings.TrimSpace(asString(sr.Metadata["current_commit_pointer_key"])) == "" { - return false, "publish current commit pointer key is missing" + return "", false, "publish current commit pointer key is missing" } - return true, "" + publishedRunID := strings.TrimSpace(asString(sr.Metadata["published_run_id"])) + if publishedRunID == "" && containsStage(executed, "publish") && m != nil { + publishedRunID = strings.TrimSpace(m.RunID) + } + if publishedRunID == "" { + return "", false, "publish run id is missing" + } + return publishedRunID, true, "" +} + +func containsStage(names []string, target string) bool { + for _, name := range names { + if name == target { + return true + } + } + return false } type scopedDir struct { diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go index 967fd9d..f540089 100644 --- a/internal/app/post_publish_cleanup_test.go +++ b/internal/app/post_publish_cleanup_test.go @@ -19,14 +19,32 @@ import ( type publishSuccessStage struct { metadata map[string]any + targets *cleanupSeed } func (publishSuccessStage) Name() string { return "publish" } func (publishSuccessStage) Declares() stage.IODecl { return stage.IODecl{} } -func (s publishSuccessStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { +func (s publishSuccessStage) Run(_ context.Context, _ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { + if s.targets != nil { + s.targets.runWorkDir = m.LocalWorkDir + s.targets.spoolAudioDir = m.LocalSpoolDir + if err := os.MkdirAll(filepath.Join(m.LocalWorkDir, "logs"), 0o755); err != nil { + return nil, err + } + if err := os.WriteFile(filepath.Join(m.LocalWorkDir, "logs", "stage.log"), []byte("log\n"), 0o644); err != nil { + return nil, err + } + if err := os.MkdirAll(m.LocalSpoolDir, 0o755); err != nil { + return nil, err + } + if err := os.WriteFile(filepath.Join(m.LocalSpoolDir, "speaker.flac"), []byte("flac\n"), 0o644); err != nil { + return nil, err + } + } md := map[string]any{ "stage": "publish", "uploaded": true, + "published_run_id": m.RunID, "remote_commit_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/20260519T010203Z-a1b2c3d4/commit.json", "current_commit_pointer_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/commit-pointer.json", } @@ -49,7 +67,7 @@ func TestPostPublishCleanupDisabledKeepsLocalDirs(t *testing.T) { cfg.Pipeline.Spool.DeleteAudioAfterPublish = false cfg.Pipeline.Workspace.CleanupAfterPublish = false - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -63,7 +81,7 @@ func TestPostPublishCleanupSpoolOnly(t *testing.T) { cfg.Pipeline.Spool.DeleteAudioAfterPublish = true cfg.Pipeline.Workspace.CleanupAfterPublish = false - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -77,7 +95,7 @@ func TestPostPublishCleanupWorkdirOnly(t *testing.T) { cfg.Pipeline.Spool.DeleteAudioAfterPublish = false cfg.Pipeline.Workspace.CleanupAfterPublish = true - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -93,7 +111,7 @@ func TestPostPublishCleanupBothPolicies(t *testing.T) { cfg.Pipeline.Spool.DeleteAudioAfterPublish = true cfg.Pipeline.Workspace.CleanupAfterPublish = true - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -103,6 +121,115 @@ func TestPostPublishCleanupBothPolicies(t *testing.T) { assertExists(t, seed.previousCachePath) } +func TestPostPublishCleanupRetriesWhenInitialObligationSaveFails(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterPublish = true + cfg.Pipeline.Workspace.CleanupAfterPublish = false + + failed := false + store := &cleanupFailingManifestStore{ + delegate: &manifest.LocalStore{}, + fail: func(m *manifest.Manifest) error { + if !failed && m.PostPublishCleanup != nil { + failed = true + return errors.New("injected obligation save failure") + } + return nil + }, + } + _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{ + Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}, + }) + if err == nil || !strings.Contains(err.Error(), "post-publish cleanup incomplete") { + t.Fatalf("executeStages() error = %v, want incomplete cleanup", err) + } + assertExists(t, seed.spoolAudioDir) + assertCleanupPending(t, cfg) + + store.fail = nil + if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + assertMissing(t, seed.spoolAudioDir) + assertCleanupComplete(t, cfg) +} + +func TestPostPublishCleanupRetriesFailedDeletionWithoutTouchingOtherRuns(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterPublish = true + cfg.Pipeline.Workspace.CleanupAfterPublish = true + + originalRemove := removeRunScopedDirFn + removeRunScopedDirFn = func(root, target, policy string) error { + if policy == "pipeline.workspace.cleanup_after_publish" { + return errors.New("injected deletion failure") + } + return originalRemove(root, target, policy) + } + t.Cleanup(func() { removeRunScopedDirFn = originalRemove }) + + _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) + if err == nil || !strings.Contains(err.Error(), "post-publish cleanup incomplete") { + t.Fatalf("executeStages() error = %v, want incomplete cleanup", err) + } + assertMissing(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) + assertCleanupPending(t, cfg) + assertExists(t, seed.otherRunDir) + assertExists(t, seed.previousCachePath) + + removeRunScopedDirFn = originalRemove + if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + assertMissing(t, seed.runWorkDir) + assertExists(t, seed.otherRunDir) + assertExists(t, seed.previousCachePath) + assertCleanupComplete(t, cfg) +} + +func TestPostPublishCleanupRetriesWhenCompletionEvidenceSaveFails(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterPublish = true + cfg.Pipeline.Workspace.CleanupAfterPublish = false + + failed := false + store := &cleanupFailingManifestStore{ + delegate: &manifest.LocalStore{}, + fail: func(m *manifest.Manifest) error { + if m.PostPublishCleanup == nil { + return nil + } + for _, target := range m.PostPublishCleanup.Targets { + if !failed && target.Completed { + failed = true + return errors.New("injected completion evidence failure") + } + } + return nil + }, + } + _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{ + Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}, + }) + if err == nil || !strings.Contains(err.Error(), "post-publish cleanup incomplete") { + t.Fatalf("executeStages() error = %v, want incomplete cleanup", err) + } + assertMissing(t, seed.spoolAudioDir) + assertCleanupPending(t, cfg) + + store.fail = nil + if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + assertCleanupComplete(t, cfg) + + if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("idempotent retry executeStages() error = %v", err) + } + assertMissing(t, seed.spoolAudioDir) +} + func TestPostPublishCleanupNotRunWhenPublishFails(t *testing.T) { cfg, seed := cleanupFixtureConfig(t) cfg.Pipeline.Spool.DeleteAudioAfterPublish = true @@ -122,7 +249,7 @@ func TestPostPublishCleanupNotRunWhenPublishSkipped(t *testing.T) { cfg.Pipeline.Spool.DeleteAudioAfterPublish = true cfg.Pipeline.Workspace.CleanupAfterPublish = true - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"skipped": true}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"skipped": true}, targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -135,7 +262,7 @@ func TestPostPublishCleanupNotRunWhenCommitPointerMissing(t *testing.T) { cfg.Pipeline.Spool.DeleteAudioAfterPublish = true cfg.Pipeline.Workspace.CleanupAfterPublish = true - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"current_commit_pointer_key": ""}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"current_commit_pointer_key": ""}, targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -149,7 +276,7 @@ func TestPostPublishCleanupNotRunWhenPublishUploadDisabled(t *testing.T) { cfg.Pipeline.Workspace.CleanupAfterPublish = true cfg.Pipeline.Publish.UploadRun = boolPtr(false) - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -182,12 +309,28 @@ func TestPostPublishCleanupFailsOnUnsafePath(t *testing.T) { if err != nil { t.Fatalf("Load() error = %v", err) } - m.LocalSpoolDir = filepath.Join(filepath.Dir(cfg.Pipeline.Spool.Root), "outside-spool") + m.MarkStageSucceeded("publish", time.Now().UTC(), nil) + m.Stages["publish"].Metadata = map[string]any{ + "uploaded": true, + "published_run_id": m.RunID, + "remote_commit_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/20260516T010203Z-1a2b3c4d/commit.json", + "current_commit_pointer_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/commit-pointer.json", + } + m.PostPublishCleanup = &manifest.PostPublishCleanup{ + CommittedRunID: m.RunID, + RemoteCommitKey: m.Stages["publish"].Metadata["remote_commit_key"].(string), + CurrentCommitPointerKey: m.Stages["publish"].Metadata["current_commit_pointer_key"].(string), + Targets: []manifest.CleanupTarget{{ + Policy: "pipeline.spool.delete_audio_after_publish", + Root: cfg.Pipeline.Spool.Root, + Path: filepath.Join(filepath.Dir(cfg.Pipeline.Spool.Root), "outside-spool"), + }}, + } if err := store.Save(context.Background(), manifestPath, m); err != nil { t.Fatalf("Save() error = %v", err) } - _, err = executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) + _, err = executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) if err == nil || !strings.Contains(err.Error(), "refusing to delete path outside root") { t.Fatalf("executeStages() error = %v, want safe-path failure", err) } @@ -220,14 +363,15 @@ func TestPostPublishCleanupNotRunWhenCommittedManifestUploadFails(t *testing.T) cfg, seed, _ := publishStageCleanupFixture(t) cfg.Pipeline.Spool.DeleteAudioAfterPublish = true cfg.Pipeline.Workspace.CleanupAfterPublish = true - failKey := artifacts.S3RunSessionManifestKey(seed.sessionPrefix, seed.runID) publishStageImpl, err := stage.Select("publish") if err != nil { t.Fatalf("Select(publish) error = %v", err) } _, err = executeStages(context.Background(), cfg, []stage.Stage{publishStageImpl}, RunOptions{ - Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}}, + Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, fail: func(key string) bool { + return strings.HasSuffix(key, "/session-manifest.json") + }}}, }) if err == nil || !strings.Contains(err.Error(), "immutable object") { t.Fatalf("executeStages() error = %v, want committed-manifest failure", err) @@ -268,6 +412,28 @@ type cleanupSeed struct { sessionPrefix string } +type cleanupFailingManifestStore struct { + delegate manifest.Store + fail func(*manifest.Manifest) error +} + +func (s *cleanupFailingManifestStore) Create(ctx context.Context, sessionID string) (*manifest.Manifest, error) { + return s.delegate.Create(ctx, sessionID) +} + +func (s *cleanupFailingManifestStore) Load(ctx context.Context, path string) (*manifest.Manifest, error) { + return s.delegate.Load(ctx, path) +} + +func (s *cleanupFailingManifestStore) Save(ctx context.Context, path string, m *manifest.Manifest) error { + if s.fail != nil { + if err := s.fail(m); err != nil { + return err + } + } + return s.delegate.Save(ctx, path, m) +} + func cleanupFixtureConfig(t *testing.T) (*config.Config, cleanupSeed) { t.Helper() @@ -388,6 +554,11 @@ func writePublishFixtureRunFiles(t *testing.T, runWorkDir, sessionRoot string) { type failKeyStore struct { delegate *storage.FakeBackend failKey string + fail func(string) bool +} + +func (s *failKeyStore) fails(key string) bool { + return strings.TrimSpace(key) == strings.TrimSpace(s.failKey) || (s.fail != nil && s.fail(key)) } func (s *failKeyStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) { @@ -403,21 +574,21 @@ func (s *failKeyStore) Download(ctx context.Context, key, localPath string) erro } func (s *failKeyStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) { - if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + if s.fails(key) { return storage.ObjectInfo{}, errors.New("forced upload failure") } return s.delegate.Upload(ctx, localPath, key, opts) } func (s *failKeyStore) UploadReader(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) { - if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + if s.fails(key) { return storage.ObjectInfo{}, errors.New("forced upload failure") } return s.delegate.UploadReader(ctx, source, key, opts) } func (s *failKeyStore) UploadConditional(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions, condition storage.WriteCondition) (storage.ObjectInfo, error) { - if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + if s.fails(key) { return storage.ObjectInfo{}, errors.New("forced upload failure") } return s.delegate.UploadConditional(ctx, source, key, opts, condition) @@ -440,3 +611,36 @@ func assertMissing(t *testing.T, path string) { t.Fatalf("expected path to be removed %q, stat err=%v", path, err) } } + +func assertCleanupPending(t *testing.T, cfg *config.Config) { + t.Helper() + m, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if m.PostPublishCleanup == nil { + t.Fatal("expected a persisted cleanup obligation") + } + for _, target := range m.PostPublishCleanup.Targets { + if !target.Completed { + return + } + } + t.Fatal("expected at least one cleanup target to remain incomplete") +} + +func assertCleanupComplete(t *testing.T, cfg *config.Config) { + t.Helper() + m, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if m.PostPublishCleanup == nil { + t.Fatal("expected a persisted cleanup obligation") + } + for _, target := range m.PostPublishCleanup.Targets { + if !target.Completed { + t.Fatalf("cleanup target remains incomplete: %#v", target) + } + } +} diff --git a/internal/app/runner.go b/internal/app/runner.go index 1ee8315..8e5f4ea 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -376,12 +376,6 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage env.Logger.Info("stage succeeded", "stage", s.Name()) } - if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil { - return nil, persistTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, - fmt.Errorf("post-publish cleanup: %w", err), - ) - } completedAt := nowUTC() runManifest.MarkSucceeded(completedAt) identity.applyToRunManifest(runManifest, manifestPath) @@ -391,6 +385,15 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage fmt.Errorf("save final run manifest %q: %w", runManifestPath, err), ) } + // The run record lives inside the run work directory, which cleanup may + // remove. Persist its completed publishing result before cleanup starts so a + // successful deletion cannot be undone by a later diagnostic write. + if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil { + return nil, persistPostPublishCleanupFailure( + ctx, env.ManifestStore, manifestPath, m, + fmt.Errorf("post-publish cleanup incomplete: %w", err), + ) + } return &RunSummary{ SessionID: cfg.Session.SessionID, @@ -403,6 +406,27 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage }, nil } +func persistPostPublishCleanupFailure( + ctx context.Context, + sessionStore manifest.Store, + sessionManifestPath string, + sessionManifest *manifest.Manifest, + operationErr error, +) error { + if operationErr == nil { + return nil + } + if sessionManifest != nil { + sessionManifest.RecordFailure(nowUTC(), operationErr.Error()) + } + // Do not rewrite the run manifest here: cleanup may have deliberately + // removed the directory that contains it. The completed run record remains + // the authoritative evidence of the remote publish; the session manifest + // records the outstanding local cleanup for the next invocation. + sessionErr := sessionStore.Save(ctx, sessionManifestPath, sessionManifest) + return errors.Join(operationErr, wrapTerminalPersistenceError("save terminal session manifest", sessionErr)) +} + func persistTerminalFailure( ctx context.Context, sessionStore manifest.Store, diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index f7f290f..1623a4b 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -57,23 +57,43 @@ type StageRecord struct { Metadata map[string]any `json:"metadata,omitempty"` } +// CleanupTarget records one root-confined local deletion requested by a +// committed publication. +type CleanupTarget struct { + Policy string `json:"policy"` + Root string `json:"root"` + Path string `json:"path"` + Completed bool `json:"completed"` +} + +// PostPublishCleanup records the durable deletion work left by a committed +// publish. It remains after completion as evidence of the exact committed run +// and local paths involved. +type PostPublishCleanup struct { + CommittedRunID string `json:"committed_run_id"` + RemoteCommitKey string `json:"remote_commit_key"` + CurrentCommitPointerKey string `json:"current_commit_pointer_key"` + Targets []CleanupTarget `json:"targets"` +} + // Manifest is the durable run-state record for a session execution. type Manifest struct { - SessionID string `json:"session_id"` - Campaign string `json:"campaign,omitempty"` - RunID string `json:"run_id,omitempty"` - LocalWorkDir string `json:"local_workdir,omitempty"` - LocalSpoolDir string `json:"local_spool_dir,omitempty"` - S3Bucket string `json:"s3_bucket,omitempty"` - S3SessionPrefix string `json:"s3_session_prefix,omitempty"` - S3RunPrefix string `json:"s3_run_prefix,omitempty"` - PipelineVersion string `json:"pipeline_version,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` - LastError *ErrorRecord `json:"last_error,omitempty"` - Inputs []InputRecord `json:"inputs,omitempty"` - Artifacts []ArtifactRecord `json:"artifacts,omitempty"` - Stages map[string]*StageRecord `json:"stages"` + SessionID string `json:"session_id"` + Campaign string `json:"campaign,omitempty"` + RunID string `json:"run_id,omitempty"` + LocalWorkDir string `json:"local_workdir,omitempty"` + LocalSpoolDir string `json:"local_spool_dir,omitempty"` + S3Bucket string `json:"s3_bucket,omitempty"` + S3SessionPrefix string `json:"s3_session_prefix,omitempty"` + S3RunPrefix string `json:"s3_run_prefix,omitempty"` + PipelineVersion string `json:"pipeline_version,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + LastError *ErrorRecord `json:"last_error,omitempty"` + Inputs []InputRecord `json:"inputs,omitempty"` + Artifacts []ArtifactRecord `json:"artifacts,omitempty"` + Stages map[string]*StageRecord `json:"stages"` + PostPublishCleanup *PostPublishCleanup `json:"post_publish_cleanup,omitempty"` } // New constructs a new manifest with deterministic timestamps. diff --git a/internal/manifest/store.go b/internal/manifest/store.go index 664df7f..bb287f4 100644 --- a/internal/manifest/store.go +++ b/internal/manifest/store.go @@ -107,6 +107,9 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error { if err := validateManifestIdentities(m.SessionID, m.Campaign, m.RunID); err != nil { return fmt.Errorf("save manifest: %w", err) } + if err := validatePostPublishCleanup(m.PostPublishCleanup); err != nil { + return fmt.Errorf("save manifest: %w", err) + } m.UpdatedAt = time.Now().UTC() if m.Stages == nil { @@ -244,6 +247,9 @@ func validateLoadedManifest(m *Manifest) error { if err := validateManifestIdentities(m.SessionID, m.Campaign, m.RunID); err != nil { return err } + if err := validatePostPublishCleanup(m.PostPublishCleanup); err != nil { + return err + } return nil } @@ -263,6 +269,27 @@ func normalizeManifest(m *Manifest) { } } +func validatePostPublishCleanup(cleanup *PostPublishCleanup) error { + if cleanup == nil { + return nil + } + if err := pathsafe.ValidateOpaqueSegment(cleanup.CommittedRunID); err != nil { + return fmt.Errorf("post_publish_cleanup.committed_run_id is not a portable opaque identifier: %w", err) + } + if strings.TrimSpace(cleanup.RemoteCommitKey) == "" || strings.TrimSpace(cleanup.CurrentCommitPointerKey) == "" { + return fmt.Errorf("post_publish_cleanup commit identity is required") + } + if len(cleanup.Targets) == 0 { + return fmt.Errorf("post_publish_cleanup.targets is required") + } + for index, target := range cleanup.Targets { + if strings.TrimSpace(target.Policy) == "" || strings.TrimSpace(target.Root) == "" || strings.TrimSpace(target.Path) == "" { + return fmt.Errorf("post_publish_cleanup.targets[%d] policy, root, and path are required", index) + } + } + return nil +} + func validateLoadedRunManifest(m *RunManifest) error { if m == nil { return fmt.Errorf("manifest is nil") diff --git a/internal/stage/publish.go b/internal/stage/publish.go index e8665ea..853b207 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -190,6 +190,7 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S manifestTempPath, err := writeCommittedSessionManifest(m, publishMetadata( bucket, runPrefix, + runID, runUploaded, publishedUploaded, previousUploaded, @@ -1188,7 +1189,7 @@ func writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[str } func publishMetadata( - bucket, runPrefix string, + bucket, runPrefix, runID string, runUploaded []string, publishedUploaded []string, previousUploaded []string, @@ -1203,6 +1204,7 @@ func publishMetadata( "uploaded": true, "s3_bucket": bucket, "s3_run_prefix": runPrefix, + "published_run_id": runID, "run_files_uploaded": len(runUploaded), "run_uploaded_paths": append([]string(nil), runUploaded...), "published_files_uploaded": len(publishedUploaded),