From 563acce0126d51174310b6db0bd4300185a684fa Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 5 Sep 2026 15:09:57 +0000 Subject: [PATCH] Define cleanup authorization and targets --- docs/config.md | 4 +- docs/roadmap/implementation.md | 2 +- internal/artifacts/remote_object_key.go | 26 ++++ internal/config/storage_publish_test.go | 40 +++++ internal/config/validate.go | 3 + internal/config/validate_publish.go | 13 ++ internal/stage/cleanup_plan.go | 170 ++++++++++++++++++++++ internal/stage/cleanup_plan_test.go | 185 ++++++++++++++++++++++++ 8 files changed, 440 insertions(+), 3 deletions(-) create mode 100644 internal/stage/cleanup_plan.go create mode 100644 internal/stage/cleanup_plan_test.go diff --git a/docs/config.md b/docs/config.md index 1d52119..1975fb8 100644 --- a/docs/config.md +++ b/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..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` | diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index c7d7406..e4a8278 100644 --- a/docs/roadmap/implementation.md +++ b/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 diff --git a/internal/artifacts/remote_object_key.go b/internal/artifacts/remote_object_key.go index b00d987..077a3ee 100644 --- a/internal/artifacts/remote_object_key.go +++ b/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 { diff --git a/internal/config/storage_publish_test.go b/internal/config/storage_publish_test.go index ad11e87..68628b7 100644 --- a/internal/config/storage_publish_test.go +++ b/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 diff --git a/internal/config/validate.go b/internal/config/validate.go index 9209213..125d48d 100644 --- a/internal/config/validate.go +++ b/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 } diff --git a/internal/config/validate_publish.go b/internal/config/validate_publish.go index 2558051..6a2b614 100644 --- a/internal/config/validate_publish.go +++ b/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 { diff --git a/internal/stage/cleanup_plan.go b/internal/stage/cleanup_plan.go new file mode 100644 index 0000000..64e6a04 --- /dev/null +++ b/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 +} diff --git a/internal/stage/cleanup_plan_test.go b/internal/stage/cleanup_plan_test.go new file mode 100644 index 0000000..4b9a4db --- /dev/null +++ b/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 +}