From 8d9eb5cc408ac6cfbb82a337d0337b0bae57ae6a Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 5 Sep 2026 15:15:28 +0000 Subject: [PATCH] Implement idempotent cleanup behavior --- docs/roadmap/implementation.md | 2 +- internal/fileops/cleanup.go | 3 + internal/fileops/cleanup_test.go | 8 + internal/stage/cleanup.go | 101 +++++++++ internal/stage/cleanup_plan_test.go | 1 + internal/stage/cleanup_test.go | 306 ++++++++++++++++++++++++++++ 6 files changed, 420 insertions(+), 1 deletion(-) create mode 100644 internal/stage/cleanup.go create mode 100644 internal/stage/cleanup_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index e4a8278..ef03b0c 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -149,7 +149,7 @@ cannot be derived from the current invocation or uncommitted publication. ## Stage 2 — Implement Idempotent Cleanup Stage Behavior -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/fileops/cleanup.go b/internal/fileops/cleanup.go index f91de90..908d4ad 100644 --- a/internal/fileops/cleanup.go +++ b/internal/fileops/cleanup.go @@ -13,6 +13,9 @@ import ( func RemoveAllUnderRoot(rootPath, target string) error { root, targetName, err := openCleanupRoot(rootPath, target) if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } return err } defer func() { _ = root.Close() }() diff --git a/internal/fileops/cleanup_test.go b/internal/fileops/cleanup_test.go index 84d6ded..758483f 100644 --- a/internal/fileops/cleanup_test.go +++ b/internal/fileops/cleanup_test.go @@ -34,6 +34,14 @@ func TestRemoveAllUnderRootRemovesNestedTreeIdempotently(t *testing.T) { } } +func TestCleanupRemoveAllUnderRootAcceptsMissingRoot(t *testing.T) { + root := filepath.Join(t.TempDir(), "absent") + target := filepath.Join(root, "runs", "run-1") + if err := RemoveAllUnderRoot(root, target); err != nil { + t.Fatalf("RemoveAllUnderRoot() with missing root error = %v", err) + } +} + func TestRemoveAllUnderRootRejectsUnsafeTargets(t *testing.T) { root := t.TempDir() outside := t.TempDir() diff --git a/internal/stage/cleanup.go b/internal/stage/cleanup.go new file mode 100644 index 0000000..b8d7c2a --- /dev/null +++ b/internal/stage/cleanup.go @@ -0,0 +1,101 @@ +package stage + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/fileops" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +const ( + cleanupSkipReason = "cleanup policies are disabled" + cleanupSemanticConfigVersion = 1 +) + +type cleanupStage struct{} + +type cleanupSemanticConfig struct { + DeleteSpoolAudio bool `json:"delete_spool_audio"` + SpoolRoot string `json:"spool_root"` + DeleteWorkspace bool `json:"delete_workspace"` + WorkspaceRoot string `json:"workspace_root"` +} + +func (cleanupStage) Name() string { return "cleanup" } + +func (cleanupStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { + plan, err := buildCleanupPlan(env, m) + if err != nil { + return nil, err + } + if !plan.enabled { + return &StageResult{ + Disposition: StageDispositionSkipped, + SkipReason: cleanupSkipReason, + }, nil + } + + completedPolicies := make([]string, 0, len(plan.targets)) + for _, target := range plan.targets { + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("cleanup canceled before %s: %w", target.policy, err) + } + if err := fileops.RemoveAllUnderRoot(target.root, target.target); err != nil { + return nil, fmt.Errorf("apply %s: %w", target.policy, err) + } + completedPolicies = append(completedPolicies, target.policy) + } + + return &StageResult{Metadata: map[string]any{ + "stage": "cleanup", + "published_run_id": plan.publishedRunID, + "completed_policies": completedPolicies, + }}, nil +} + +func (cleanupStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil { + return manifest.SemanticConfigFingerprint{}, fmt.Errorf("resolved pipeline configuration is required") + } + pipeline := env.Config.Pipeline + return FingerprintSemanticConfig(cleanupSemanticConfigVersion, cleanupSemanticConfig{ + DeleteSpoolAudio: pipeline.Spool.DeleteAudioAfterPublish, + SpoolRoot: normalizedCleanupRoot(pipeline.Spool.Root), + DeleteWorkspace: pipeline.Workspace.CleanupAfterPublish, + WorkspaceRoot: normalizedCleanupRoot(pipeline.Workspace.Root), + }) +} + +func (cleanupStage) ValidateResume(ctx context.Context, env *Env, m *manifest.Manifest) (ResumeValidation, error) { + plan, err := buildCleanupPlan(env, m) + if err != nil { + return ResumeValidation{}, err + } + if !plan.enabled { + return NonResumable(cleanupSkipReason), nil + } + for _, target := range plan.targets { + if err := ctx.Err(); err != nil { + return ResumeValidation{}, fmt.Errorf("validate cleanup resume for %s: %w", target.policy, err) + } + if _, err := os.Lstat(target.target); err == nil { + return NonResumable(fmt.Sprintf("cleanup target for %s exists", target.policy)), nil + } else if !errors.Is(err, os.ErrNotExist) { + return ResumeValidation{}, fmt.Errorf("inspect cleanup target for %s: %w", target.policy, err) + } + } + return Resumable(), nil +} + +func normalizedCleanupRoot(root string) string { + root = strings.TrimSpace(root) + if root == "" { + return "" + } + return filepath.Clean(root) +} diff --git a/internal/stage/cleanup_plan_test.go b/internal/stage/cleanup_plan_test.go index 4b9a4db..d7c37f0 100644 --- a/internal/stage/cleanup_plan_test.go +++ b/internal/stage/cleanup_plan_test.go @@ -158,6 +158,7 @@ func cleanupPlanningFixture(t *testing.T) (*Env, *manifest.Manifest, string) { Pipeline: &config.PipelineConfig{ Workspace: config.WorkspaceConfig{Root: filepath.Join(root, "workspace")}, Spool: config.SpoolConfig{Root: filepath.Join(root, "spool")}, + Cache: config.CacheConfig{Root: filepath.Join(root, "cache")}, }, Session: &config.SessionConfig{Campaign: campaign, SessionID: sessionID}, }} diff --git a/internal/stage/cleanup_test.go b/internal/stage/cleanup_test.go new file mode 100644 index 0000000..419f91d --- /dev/null +++ b/internal/stage/cleanup_test.go @@ -0,0 +1,306 @@ +package stage + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" +) + +func TestCleanupStageDisabledSelfSkips(t *testing.T) { + env, _, _ := cleanupPlanningFixture(t) + result, err := (cleanupStage{}).Run(context.Background(), env, nil) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if result.Disposition != StageDispositionSkipped || result.SkipReason != cleanupSkipReason { + t.Fatalf("cleanup result = %#v", result) + } + if len(result.Outputs) != 0 || len(result.Logs) != 0 || len(result.GeneratedConfigs) != 0 || len(result.Metadata) != 0 { + t.Fatalf("disabled cleanup returned payloads: %#v", result) + } +} + +func TestCleanupStageRemovesOnlySelectedPublishedRunTargets(t *testing.T) { + tests := []struct { + name string + spool bool + workspace bool + }{ + {name: "spool only", spool: true}, + {name: "workspace only", workspace: true}, + {name: "both", spool: true, workspace: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + env, m, publishedRunID := cleanupPlanningFixture(t) + env.Config.Pipeline.Spool.DeleteAudioAfterPublish = test.spool + env.Config.Pipeline.Workspace.CleanupAfterPublish = test.workspace + plan, err := buildCleanupPlan(env, m) + if err != nil { + t.Fatal(err) + } + publishedSpool := artifacts.SessionSpoolAudioDir( + env.Config.Pipeline.Spool.Root, m.Campaign, m.SessionID, publishedRunID, + ) + publishedWorkspace := artifacts.SessionRunRootForCampaign( + env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, publishedRunID, + ) + writeStageTestFile(t, filepath.Join(publishedSpool, "nested", "audio.wav"), "audio") + writeStageTestFile(t, filepath.Join(publishedWorkspace, "nested", "result.txt"), "result") + + 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") + writeStageTestFile(t, filepath.Join(otherSpool, "audio.wav"), "other spool") + writeStageTestFile(t, filepath.Join(otherWorkspace, "result.txt"), "other workspace") + writeStageTestFile(t, cacheFile, "cache") + + result, err := (cleanupStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + for _, target := range plan.targets { + if _, err := os.Lstat(target.target); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("cleanup target %q still exists: %v", target.target, err) + } + } + if !test.spool { + if _, err := os.Stat(publishedSpool); err != nil { + t.Fatalf("unselected spool target changed: %v", err) + } + } + if !test.workspace { + if _, err := os.Stat(publishedWorkspace); err != nil { + t.Fatalf("unselected workspace target changed: %v", err) + } + } + for _, preserved := range []string{otherSpool, otherWorkspace, cacheFile} { + if _, err := os.Stat(preserved); err != nil { + t.Fatalf("preserved path %q: %v", preserved, err) + } + } + if len(result.Outputs) != 0 || len(result.Logs) != 0 || len(result.GeneratedConfigs) != 0 { + t.Fatalf("cleanup returned canonical or diagnostic outputs: %#v", result) + } + if len(result.Metadata) != 3 || result.Metadata["stage"] != "cleanup" || result.Metadata["published_run_id"] != publishedRunID { + t.Fatalf("cleanup metadata = %#v", result.Metadata) + } + wantPolicies := make([]string, 0, 2) + if test.spool { + wantPolicies = append(wantPolicies, cleanupSpoolPolicy) + } + if test.workspace { + wantPolicies = append(wantPolicies, cleanupWorkspacePolicy) + } + if got := result.Metadata["completed_policies"]; !reflect.DeepEqual(got, wantPolicies) { + t.Fatalf("completed policies = %#v, want %#v", got, wantPolicies) + } + }) + } +} + +func TestCleanupStageAcceptsAlreadyAbsentRootAndTarget(t *testing.T) { + tests := []struct { + name string + createRoot bool + }{ + {name: "missing root"}, + {name: "missing target", createRoot: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + env, m, _ := cleanupPlanningFixture(t) + env.Config.Pipeline.Spool.DeleteAudioAfterPublish = true + if test.createRoot { + if err := os.MkdirAll(env.Config.Pipeline.Spool.Root, 0o755); err != nil { + t.Fatal(err) + } + } + if _, err := (cleanupStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } + }) + } +} + +func TestCleanupStageHonorsCancellationBeforeDeletion(t *testing.T) { + env, m, _ := cleanupPlanningFixture(t) + env.Config.Pipeline.Spool.DeleteAudioAfterPublish = true + plan, err := buildCleanupPlan(env, m) + if err != nil { + t.Fatal(err) + } + writeStageTestFile(t, filepath.Join(plan.targets[0].target, "audio.wav"), "audio") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := (cleanupStage{}).Run(ctx, env, m); !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want context cancellation", err) + } + if _, err := os.Stat(plan.targets[0].target); err != nil { + t.Fatalf("cleanup target changed after cancellation: %v", err) + } +} + +func TestCleanupStageRejectsUnsafeIdentityBeforeDeletion(t *testing.T) { + env, m, _ := cleanupPlanningFixture(t) + env.Config.Pipeline.Spool.DeleteAudioAfterPublish = true + sentinel := filepath.Join(env.Config.Pipeline.Spool.Root, "sentinel.txt") + writeStageTestFile(t, sentinel, "keep") + m.Campaign = ".." + env.Config.Session.Campaign = ".." + + if _, err := (cleanupStage{}).Run(context.Background(), env, m); err == nil { + t.Fatal("Run() error = nil for unsafe campaign identity") + } + if contents, err := os.ReadFile(sentinel); err != nil || string(contents) != "keep" { + t.Fatalf("sentinel changed: contents %q error %v", contents, err) + } +} + +func TestCleanupStageRetriesAfterPartialDeletion(t *testing.T) { + env, m, _ := cleanupPlanningFixture(t) + env.Config.Pipeline.Spool.DeleteAudioAfterPublish = true + env.Config.Pipeline.Workspace.CleanupAfterPublish = true + plan, err := buildCleanupPlan(env, m) + if err != nil { + t.Fatal(err) + } + spoolTarget := plan.targets[0].target + workspaceTarget := plan.targets[1].target + writeStageTestFile(t, filepath.Join(spoolTarget, "audio.wav"), "audio") + if err := os.MkdirAll(filepath.Dir(env.Config.Pipeline.Workspace.Root), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(env.Config.Pipeline.Workspace.Root, []byte("not a directory"), 0o644); err != nil { + t.Fatal(err) + } + + if _, err := (cleanupStage{}).Run(context.Background(), env, m); err == nil { + t.Fatal("Run() error = nil with invalid workspace root") + } + if _, err := os.Lstat(spoolTarget); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("first target remains after partial cleanup: %v", err) + } + if err := os.Remove(env.Config.Pipeline.Workspace.Root); err != nil { + t.Fatal(err) + } + writeStageTestFile(t, filepath.Join(workspaceTarget, "result.txt"), "result") + + if _, err := (cleanupStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("retry Run() error = %v", err) + } + if _, err := os.Lstat(workspaceTarget); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("second target remains after retry: %v", err) + } +} + +func TestCleanupSemanticConfigFingerprintTracksPoliciesAndNormalizedRoots(t *testing.T) { + env, _, _ := cleanupPlanningFixture(t) + base, err := (cleanupStage{}).SemanticConfigFingerprint(env) + if err != nil { + t.Fatal(err) + } + + mutations := []struct { + name string + mutate func() + restore func() + }{ + { + name: "spool policy", + mutate: func() { env.Config.Pipeline.Spool.DeleteAudioAfterPublish = true }, + restore: func() { env.Config.Pipeline.Spool.DeleteAudioAfterPublish = false }, + }, + { + name: "workspace policy", + mutate: func() { env.Config.Pipeline.Workspace.CleanupAfterPublish = true }, + restore: func() { env.Config.Pipeline.Workspace.CleanupAfterPublish = false }, + }, + { + name: "spool root", + mutate: func() { env.Config.Pipeline.Spool.Root += "-changed" }, + restore: func() { + env.Config.Pipeline.Spool.Root = strings.TrimSuffix(env.Config.Pipeline.Spool.Root, "-changed") + }, + }, + { + name: "workspace root", + mutate: func() { env.Config.Pipeline.Workspace.Root += "-changed" }, + restore: func() { + env.Config.Pipeline.Workspace.Root = strings.TrimSuffix(env.Config.Pipeline.Workspace.Root, "-changed") + }, + }, + } + for _, mutation := range mutations { + t.Run(mutation.name, func(t *testing.T) { + mutation.mutate() + got, err := (cleanupStage{}).SemanticConfigFingerprint(env) + mutation.restore() + if err != nil { + t.Fatal(err) + } + if got == base { + t.Fatalf("fingerprint did not change: %#v", got) + } + }) + } + + originalWorkspaceRoot := env.Config.Pipeline.Workspace.Root + env.Config.Pipeline.Workspace.Root = originalWorkspaceRoot + string(os.PathSeparator) + "." + normalized, err := (cleanupStage{}).SemanticConfigFingerprint(env) + if err != nil { + t.Fatal(err) + } + if normalized != base { + t.Fatalf("equivalent normalized root changed fingerprint: base %#v normalized %#v", base, normalized) + } +} + +func TestCleanupResumeRequiresAbsentCurrentTargetsAndValidPublishEvidence(t *testing.T) { + env, m, _ := cleanupPlanningFixture(t) + env.Config.Pipeline.Workspace.CleanupAfterPublish = true + plan, err := buildCleanupPlan(env, m) + if err != nil { + t.Fatal(err) + } + writeStageTestFile(t, filepath.Join(plan.targets[0].target, "result.txt"), "result") + + validation, err := (cleanupStage{}).ValidateResume(context.Background(), env, m) + if err != nil { + t.Fatal(err) + } + if validation.Resumable { + t.Fatal("resume validation accepted existing cleanup target") + } + if _, err := (cleanupStage{}).Run(context.Background(), env, m); err != nil { + t.Fatal(err) + } + validation, err = (cleanupStage{}).ValidateResume(context.Background(), env, m) + if err != nil || !validation.Resumable { + t.Fatalf("resume validation after cleanup = %#v, error %v", validation, err) + } + writeStageTestFile(t, filepath.Join(plan.targets[0].target, "recreated.txt"), "recreated") + validation, err = (cleanupStage{}).ValidateResume(context.Background(), env, m) + if err != nil || validation.Resumable { + t.Fatalf("resume validation after recreation = %#v, error %v", validation, err) + } + + delete(m.Stages["publish"].Metadata, "remote_commit_key") + if _, err := (cleanupStage{}).ValidateResume(context.Background(), env, m); err == nil { + t.Fatal("resume validation accepted invalid publish evidence") + } + + env.Config.Pipeline.Workspace.CleanupAfterPublish = false + validation, err = (cleanupStage{}).ValidateResume(context.Background(), env, nil) + if err != nil || validation.Resumable { + t.Fatalf("disabled cleanup resume = %#v, error %v", validation, err) + } +}