Implement idempotent cleanup behavior

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
 

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

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

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

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

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