Define cleanup authorization and targets

docs/config.md

@@ -278,7 +278,7 @@ Rules:
 | `composition.default_profile` | string | Conditional | selected when profiles exist and no caller explicitly selects one; must name a declared profile |
 | `composition.profiles.<name>.overlay` | string | Conditional | required for every declared profile; one confined `.yml` or `.yaml` overlay relative to the root pipeline directory |
 | `pipeline.workspace.root` | string | No | `/var/lib/narratio` |
-| `pipeline.workspace.cleanup_after_publish` | bool | No | `false` |
+| `pipeline.workspace.cleanup_after_publish` | bool | No | `false`; selects the published run's workspace target for final cleanup and requires effective `pipeline.publish.enabled=true` and `pipeline.publish.upload_run=true` |
 | `pipeline.campaigns.root` | string | No | `/usr/local/share/narratio/campaigns` |
 | `pipeline.campaigns.default_campaign_id` | string | No | empty |
 | `pipeline.secrets.env_dir` | string | No | empty |
@@ -291,7 +291,7 @@ Rules:
 | `pipeline.storage.s3.access_key_id_env` | string | No | `OBJECT_STORAGE_KEY_ID` |
 | `pipeline.storage.s3.secret_access_key_env` | string | No | `OBJECT_STORAGE_KEY` |
 | `pipeline.spool.root` | string | No | `/var/spool/narratio` |
-| `pipeline.spool.delete_audio_after_publish` | bool | No | `false` |
+| `pipeline.spool.delete_audio_after_publish` | bool | No | `false`; selects the published run's spool-audio target for final cleanup and requires effective `pipeline.publish.enabled=true` and `pipeline.publish.upload_run=true` |
 | `pipeline.cache.root` | string | No | `/var/cache/narratio` |
 | `pipeline.cache.s3_audio` | bool | No | `true` |
 | `pipeline.publish.enabled` | bool | No | `true` |

docs/roadmap/implementation.md

@@ -104,7 +104,7 @@ the same stage as the behavior it owns.
 
 ## Stage 1 — Define Cleanup Evidence, Targets, And Configuration Rules
 
-**Status:** Pending
+**Status:** Completed
 
 ### Goal
 

internal/artifacts/remote_object_key.go

@@ -66,6 +66,32 @@ func ParsePublishedOutputObjectKey(sessionPrefix, key string) (PublishedOutputOb
 	return PublishedOutputObjectKey{RunID: identity.RunID, Destination: destination}, nil
 }
 
+// ValidateRemoteSessionPrefixIdentity proves that a canonical remote session
+// prefix names the expected campaign and session.
+func ValidateRemoteSessionPrefixIdentity(sessionPrefix, campaign, sessionID string) error {
+	if err := validateRemoteSessionPrefix(sessionPrefix); err != nil {
+		return err
+	}
+	if err := ValidateSessionIdentity(campaign, sessionID); err != nil {
+		return err
+	}
+
+	parts := strings.Split(strings.TrimSuffix(sessionPrefix, "/"), "/")
+	prefixCampaign := parts[len(parts)-3]
+	prefixSessionID := parts[len(parts)-1]
+	if prefixCampaign != campaign || prefixSessionID != sessionID {
+		return fmt.Errorf(
+			"session prefix %q identifies campaign/session %q/%q, want %q/%q",
+			sessionPrefix,
+			prefixCampaign,
+			prefixSessionID,
+			campaign,
+			sessionID,
+		)
+	}
+	return nil
+}
+
 func validateCanonicalPublishSource(source string) error {
 	classified, err := artifactpolicy.ClassifySource(source)
 	if err != nil || classified.Kind == artifactpolicy.SourceKindPreviousArtifact {

internal/config/storage_publish_test.go

@@ -307,6 +307,46 @@ func TestSpoolAndPublishDefaults(t *testing.T) {
 	}
 }
 
+func TestCleanupPoliciesRequireEnabledRunPublication(t *testing.T) {
+	tests := []struct {
+		name             string
+		spoolCleanup     bool
+		workspaceCleanup bool
+		publishEnabled   bool
+		uploadRun        bool
+		wantError        bool
+	}{
+		{name: "spool cleanup rejects disabled publish", spoolCleanup: true, uploadRun: true, wantError: true},
+		{name: "workspace cleanup rejects disabled publish", workspaceCleanup: true, uploadRun: true, wantError: true},
+		{name: "spool cleanup rejects disabled run upload", spoolCleanup: true, publishEnabled: true, wantError: true},
+		{name: "workspace cleanup rejects disabled run upload", workspaceCleanup: true, publishEnabled: true, wantError: true},
+		{name: "spool cleanup accepts enabled run publication", spoolCleanup: true, publishEnabled: true, uploadRun: true},
+		{name: "workspace cleanup accepts enabled run publication", workspaceCleanup: true, publishEnabled: true, uploadRun: true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			pipelinePath, sessionPath := writeConfigFiles(t, testPipelineBaseYAML, testSessionBaseYAML)
+			cfg, err := Load(pipelinePath, sessionPath)
+			if err != nil {
+				t.Fatalf("Load() error = %v", err)
+			}
+			cfg.Pipeline.Spool.DeleteAudioAfterPublish = test.spoolCleanup
+			cfg.Pipeline.Workspace.CleanupAfterPublish = test.workspaceCleanup
+			cfg.Pipeline.Publish.Enabled = &test.publishEnabled
+			cfg.Pipeline.Publish.UploadRun = &test.uploadRun
+
+			err = Validate(cfg)
+			if test.wantError && err == nil {
+				t.Fatal("Validate() error = nil")
+			}
+			if !test.wantError && err != nil {
+				t.Fatalf("Validate() error = %v", err)
+			}
+		})
+	}
+}
+
 func TestPublishOutputValidation(t *testing.T) {
 	tests := []struct {
 		name    string

internal/config/validate.go

@@ -85,6 +85,9 @@ func validatePipeline(cfg *PipelineConfig) error {
 	if err := validatePublish(cfg.Publish, cfg.Scriptorium, cfg.Notarius); err != nil {
 		return err
 	}
+	if err := validateCleanupPublishPolicy(cfg); err != nil {
+		return err
+	}
 	if err := validateWhisperX(cfg.WhisperX); err != nil {
 		return err
 	}

internal/config/validate_publish.go

@@ -86,6 +86,19 @@ func validatePublish(cfg *PublishConfig, scriptorium *ScriptoriumConfig, notariu
 	return nil
 }
 
+func validateCleanupPublishPolicy(cfg *PipelineConfig) error {
+	if cfg == nil || (!cfg.Spool.DeleteAudioAfterPublish && !cfg.Workspace.CleanupAfterPublish) {
+		return nil
+	}
+	if cfg.Publish == nil || (cfg.Publish.Enabled != nil && !*cfg.Publish.Enabled) {
+		return fmt.Errorf("post-publish cleanup policies require pipeline.publish.enabled=true")
+	}
+	if cfg.Publish.UploadRun != nil && !*cfg.Publish.UploadRun {
+		return fmt.Errorf("post-publish cleanup policies require pipeline.publish.upload_run=true")
+	}
+	return nil
+}
+
 func scriptoriumOutputPathMap(scriptorium *ScriptoriumConfig) map[string]string {
 	out := map[string]string{}
 	if scriptorium == nil || len(scriptorium.Artifacts) == 0 {

internal/stage/cleanup_plan.go

@@ -0,0 +1,170 @@
+package stage
+
+import (
+	"fmt"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+const (
+	cleanupSpoolPolicy     = "pipeline.spool.delete_audio_after_publish"
+	cleanupWorkspacePolicy = "pipeline.workspace.cleanup_after_publish"
+)
+
+type cleanupPlan struct {
+	enabled        bool
+	publishedRunID string
+	targets        []cleanupPlanTarget
+}
+
+type cleanupPlanTarget struct {
+	root   string
+	target string
+	policy string
+}
+
+func buildCleanupPlan(env *Env, m *manifest.Manifest) (cleanupPlan, error) {
+	if env == nil || env.Config == nil || env.Config.Pipeline == nil {
+		return cleanupPlan{}, fmt.Errorf("resolved pipeline configuration is required")
+	}
+
+	pipeline := env.Config.Pipeline
+	spoolRequested := pipeline.Spool.DeleteAudioAfterPublish
+	workspaceRequested := pipeline.Workspace.CleanupAfterPublish
+	if !spoolRequested && !workspaceRequested {
+		return cleanupPlan{}, nil
+	}
+	if m == nil {
+		return cleanupPlan{}, fmt.Errorf("session manifest is required for cleanup")
+	}
+	if env.Config.Session == nil {
+		return cleanupPlan{}, fmt.Errorf("resolved session configuration is required for cleanup")
+	}
+
+	campaign := strings.TrimSpace(m.Campaign)
+	sessionID := strings.TrimSpace(m.SessionID)
+	if campaign != m.Campaign || sessionID != m.SessionID {
+		return cleanupPlan{}, fmt.Errorf("manifest campaign and session identities must be canonical")
+	}
+	if err := artifacts.ValidateSessionIdentity(campaign, sessionID); err != nil {
+		return cleanupPlan{}, fmt.Errorf("validate cleanup manifest identity: %w", err)
+	}
+	configuredCampaign := strings.TrimSpace(env.Config.Session.Campaign)
+	configuredSessionID := strings.TrimSpace(env.Config.Session.SessionID)
+	if configuredCampaign != env.Config.Session.Campaign || configuredSessionID != env.Config.Session.SessionID {
+		return cleanupPlan{}, fmt.Errorf("configured campaign and session identities must be canonical")
+	}
+	if err := artifacts.ValidateSessionIdentity(configuredCampaign, configuredSessionID); err != nil {
+		return cleanupPlan{}, fmt.Errorf("validate cleanup configuration identity: %w", err)
+	}
+	if configuredCampaign != campaign || configuredSessionID != sessionID {
+		return cleanupPlan{}, fmt.Errorf("configured campaign/session does not match the session manifest")
+	}
+
+	publishedRunID, err := validateCleanupPublishEvidence(m, campaign, sessionID)
+	if err != nil {
+		return cleanupPlan{}, err
+	}
+
+	plan := cleanupPlan{enabled: true, publishedRunID: publishedRunID}
+	if spoolRequested {
+		root := strings.TrimSpace(pipeline.Spool.Root)
+		if root == "" {
+			return cleanupPlan{}, fmt.Errorf("pipeline.spool.root is required for cleanup")
+		}
+		plan.targets = append(plan.targets, cleanupPlanTarget{
+			root: root,
+			target: artifacts.SessionSpoolAudioDir(
+				root,
+				campaign,
+				sessionID,
+				publishedRunID,
+			),
+			policy: cleanupSpoolPolicy,
+		})
+	}
+	if workspaceRequested {
+		root := strings.TrimSpace(pipeline.Workspace.Root)
+		if root == "" {
+			return cleanupPlan{}, fmt.Errorf("pipeline.workspace.root is required for cleanup")
+		}
+		plan.targets = append(plan.targets, cleanupPlanTarget{
+			root: root,
+			target: artifacts.SessionRunRootForCampaign(
+				root,
+				campaign,
+				sessionID,
+				publishedRunID,
+			),
+			policy: cleanupWorkspacePolicy,
+		})
+	}
+	return plan, nil
+}
+
+func validateCleanupPublishEvidence(m *manifest.Manifest, campaign, sessionID string) (string, error) {
+	publishRecord := m.Stages["publish"]
+	if publishRecord == nil || publishRecord.Name != "publish" {
+		return "", fmt.Errorf("cleanup requires a recorded publish stage")
+	}
+	if publishRecord.Status != manifest.StatusSucceeded {
+		return "", fmt.Errorf("cleanup requires a succeeded publish stage")
+	}
+	if publishRecord.Metadata == nil {
+		return "", fmt.Errorf("cleanup requires publish completion metadata")
+	}
+
+	uploaded, ok := publishRecord.Metadata["uploaded"].(bool)
+	if !ok || !uploaded {
+		return "", fmt.Errorf("cleanup requires publish metadata uploaded=true")
+	}
+	if skippedValue, exists := publishRecord.Metadata["skipped"]; exists {
+		skipped, ok := skippedValue.(bool)
+		if !ok {
+			return "", fmt.Errorf("cleanup publish metadata skipped must be boolean")
+		}
+		if skipped {
+			return "", fmt.Errorf("cleanup cannot use skipped publication evidence")
+		}
+	}
+
+	publishedRunID, err := canonicalCleanupMetadataString(publishRecord.Metadata, "published_run_id")
+	if err != nil {
+		return "", err
+	}
+	if err := artifacts.ValidateRunIdentity(publishedRunID); err != nil {
+		return "", fmt.Errorf("cleanup publish metadata published_run_id: %w", err)
+	}
+	remoteCommitKey, err := canonicalCleanupMetadataString(publishRecord.Metadata, "remote_commit_key")
+	if err != nil {
+		return "", err
+	}
+	pointerKey, err := canonicalCleanupMetadataString(publishRecord.Metadata, "current_commit_pointer_key")
+	if err != nil {
+		return "", err
+	}
+
+	sessionPrefix := m.S3SessionPrefix
+	if err := artifacts.ValidateRemoteSessionPrefixIdentity(sessionPrefix, campaign, sessionID); err != nil {
+		return "", fmt.Errorf("validate cleanup session prefix: %w", err)
+	}
+	wantCommitKey := artifacts.S3RunCommitKey(sessionPrefix, publishedRunID)
+	if remoteCommitKey != wantCommitKey {
+		return "", fmt.Errorf("cleanup publish metadata remote_commit_key %q does not match %q", remoteCommitKey, wantCommitKey)
+	}
+	wantPointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix)
+	if pointerKey != wantPointerKey {
+		return "", fmt.Errorf("cleanup publish metadata current_commit_pointer_key %q does not match %q", pointerKey, wantPointerKey)
+	}
+	return publishedRunID, nil
+}
+
+func canonicalCleanupMetadataString(metadata map[string]any, key string) (string, error) {
+	value, ok := metadata[key].(string)
+	if !ok || value == "" || strings.TrimSpace(value) != value {
+		return "", fmt.Errorf("cleanup publish metadata %s must be a canonical nonempty string", key)
+	}
+	return value, nil
+}

internal/stage/cleanup_plan_test.go

@@ -0,0 +1,185 @@
+package stage
+
+import (
+	"path/filepath"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+func TestBuildCleanupPlanDisabledNeedsNoPublishEvidence(t *testing.T) {
+	env := &Env{Config: &config.Config{Pipeline: &config.PipelineConfig{}}}
+	plan, err := buildCleanupPlan(env, nil)
+	if err != nil {
+		t.Fatalf("buildCleanupPlan() error = %v", err)
+	}
+	if plan.enabled || plan.publishedRunID != "" || len(plan.targets) != 0 {
+		t.Fatalf("disabled cleanup plan = %#v", plan)
+	}
+}
+
+func TestBuildCleanupPlanSelectsCanonicalPublishedRunTargets(t *testing.T) {
+	tests := []struct {
+		name              string
+		spool             bool
+		workspace         bool
+		wantPolicies      []string
+		wantTargetBuilder func(workspaceRoot, spoolRoot, campaign, sessionID, runID string) []string
+	}{
+		{
+			name:         "spool only",
+			spool:        true,
+			wantPolicies: []string{cleanupSpoolPolicy},
+			wantTargetBuilder: func(_, spoolRoot, campaign, sessionID, runID string) []string {
+				return []string{artifacts.SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID)}
+			},
+		},
+		{
+			name:         "workspace only",
+			workspace:    true,
+			wantPolicies: []string{cleanupWorkspacePolicy},
+			wantTargetBuilder: func(workspaceRoot, _, campaign, sessionID, runID string) []string {
+				return []string{artifacts.SessionRunRootForCampaign(workspaceRoot, campaign, sessionID, runID)}
+			},
+		},
+		{
+			name:         "spool precedes workspace",
+			spool:        true,
+			workspace:    true,
+			wantPolicies: []string{cleanupSpoolPolicy, cleanupWorkspacePolicy},
+			wantTargetBuilder: func(workspaceRoot, spoolRoot, campaign, sessionID, runID string) []string {
+				return []string{
+					artifacts.SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID),
+					artifacts.SessionRunRootForCampaign(workspaceRoot, campaign, sessionID, runID),
+				}
+			},
+		},
+	}
+
+	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.Fatalf("buildCleanupPlan() error = %v", err)
+			}
+			if !plan.enabled || plan.publishedRunID != publishedRunID {
+				t.Fatalf("cleanup plan identity = %#v", plan)
+			}
+			wantTargets := test.wantTargetBuilder(
+				env.Config.Pipeline.Workspace.Root,
+				env.Config.Pipeline.Spool.Root,
+				m.Campaign,
+				m.SessionID,
+				publishedRunID,
+			)
+			if len(plan.targets) != len(wantTargets) {
+				t.Fatalf("cleanup targets = %#v, want %d", plan.targets, len(wantTargets))
+			}
+			for index, target := range plan.targets {
+				if target.policy != test.wantPolicies[index] || target.target != wantTargets[index] {
+					t.Fatalf("cleanup target %d = %#v, want policy %q path %q", index, target, test.wantPolicies[index], wantTargets[index])
+				}
+				wantRoot := env.Config.Pipeline.Workspace.Root
+				if target.policy == cleanupSpoolPolicy {
+					wantRoot = env.Config.Pipeline.Spool.Root
+				}
+				if target.root != wantRoot {
+					t.Fatalf("cleanup target %d root = %q, want %q", index, target.root, wantRoot)
+				}
+				if target.target == m.LocalWorkDir || target.target == m.LocalSpoolDir || filepath.Base(target.target) == m.RunID {
+					t.Fatalf("cleanup target used current-invocation state: %#v", target)
+				}
+			}
+		})
+	}
+}
+
+func TestBuildCleanupPlanRejectsInvalidPublishEvidenceWithoutExecutablePlan(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*Env, *manifest.Manifest)
+	}{
+		{name: "missing publish record", mutate: func(_ *Env, m *manifest.Manifest) { delete(m.Stages, "publish") }},
+		{name: "failed publish record", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Status = manifest.StatusFailed }},
+		{name: "skipped publish record", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Status = manifest.StatusSkipped }},
+		{name: "missing metadata", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Metadata = nil }},
+		{name: "uploaded false", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Metadata["uploaded"] = false }},
+		{name: "uploaded mistyped", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Metadata["uploaded"] = "true" }},
+		{name: "skipped true", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Metadata["skipped"] = true }},
+		{name: "skipped mistyped", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Metadata["skipped"] = "false" }},
+		{name: "missing run identity", mutate: func(_ *Env, m *manifest.Manifest) { delete(m.Stages["publish"].Metadata, "published_run_id") }},
+		{name: "mistyped run identity", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Metadata["published_run_id"] = 42 }},
+		{name: "noncanonical run identity", mutate: func(_ *Env, m *manifest.Manifest) { m.Stages["publish"].Metadata["published_run_id"] = " ../run " }},
+		{name: "wrong commit key", mutate: func(_ *Env, m *manifest.Manifest) {
+			m.Stages["publish"].Metadata["remote_commit_key"] = "wrong/commit.json"
+		}},
+		{name: "wrong pointer key", mutate: func(_ *Env, m *manifest.Manifest) {
+			m.Stages["publish"].Metadata["current_commit_pointer_key"] = "wrong/pointer.json"
+		}},
+		{name: "wrong prefix identity", mutate: func(_ *Env, m *manifest.Manifest) {
+			m.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", "other", m.SessionID)
+		}},
+		{name: "configuration identity mismatch", mutate: func(env *Env, _ *manifest.Manifest) { env.Config.Session.SessionID = "other-session" }},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			env, m, _ := cleanupPlanningFixture(t)
+			env.Config.Pipeline.Spool.DeleteAudioAfterPublish = true
+			env.Config.Pipeline.Workspace.CleanupAfterPublish = true
+			test.mutate(env, m)
+
+			plan, err := buildCleanupPlan(env, m)
+			if err == nil {
+				t.Fatalf("buildCleanupPlan() error = nil, plan = %#v", plan)
+			}
+			if plan.enabled || plan.publishedRunID != "" || len(plan.targets) != 0 {
+				t.Fatalf("invalid evidence returned executable cleanup plan %#v", plan)
+			}
+		})
+	}
+}
+
+func cleanupPlanningFixture(t *testing.T) (*Env, *manifest.Manifest, string) {
+	t.Helper()
+	root := t.TempDir()
+	campaign := "forsaken"
+	sessionID := "2026-09-05"
+	publishedRunID := "published-run"
+	sessionPrefix := artifacts.S3SessionPrefix("dnd", campaign, sessionID)
+
+	env := &Env{Config: &config.Config{
+		Pipeline: &config.PipelineConfig{
+			Workspace: config.WorkspaceConfig{Root: filepath.Join(root, "workspace")},
+			Spool:     config.SpoolConfig{Root: filepath.Join(root, "spool")},
+		},
+		Session: &config.SessionConfig{Campaign: campaign, SessionID: sessionID},
+	}}
+	m := &manifest.Manifest{
+		Campaign:        campaign,
+		SessionID:       sessionID,
+		RunID:           "current-invocation-run",
+		LocalWorkDir:    filepath.Join(root, "untrusted-workdir"),
+		LocalSpoolDir:   filepath.Join(root, "untrusted-spool"),
+		S3SessionPrefix: sessionPrefix,
+		Stages: map[string]*manifest.StageRecord{
+			"publish": {
+				Name:   "publish",
+				Status: manifest.StatusSucceeded,
+				Metadata: map[string]any{
+					"uploaded":                   true,
+					"published_run_id":           publishedRunID,
+					"remote_commit_key":          artifacts.S3RunCommitKey(sessionPrefix, publishedRunID),
+					"current_commit_pointer_key": artifacts.S3CurrentCommitPointerKey(sessionPrefix),
+				},
+			},
+		},
+	}
+	return env, m, publishedRunID
+}