From c2806a161536c128cbf7a1a2194ace46af76de9b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 5 Sep 2026 15:42:00 +0000 Subject: [PATCH] Complete cleanup stage workstream --- docs/internal/manifest.md | 13 ++-- docs/internal/stage-analyze.md | 7 ++- docs/roadmap/implementation.md | 2 +- docs/roadmap/refactoring.md | 2 +- internal/app/cleanup_orchestration_test.go | 72 ++++++++++++++++++++++ internal/config/validate_publish.go | 4 +- 6 files changed, 87 insertions(+), 13 deletions(-) diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index f0cfd03..d463cbd 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -183,7 +183,7 @@ manifests remain compatible when those fields are absent. A successful forced rerun marks only succeeded transitive dependent session-stage records stale. The application owns a fixed dependency relation distinct from execution order; dependents are returned in canonical order. Render and extract therefore never -stale one another, while either can stale analyze, publish, and notify. +stale one another, while either can stale analyze, publish, notify, and cleanup. Starting an execution clears the current session-stage record's prior outputs, logs, generated configuration references, metadata, and semantic fingerprint. @@ -216,11 +216,11 @@ match the current resolved stage semantics. Missing semantic evidence, malformed evidence, or a mismatch makes the stage and its fixed transitive dependents stale. The existing stage-specific resume validator runs only after this semantic check succeeds; both checks are -required. Extraction and analyze have resume validators and may reject an -otherwise eligible skip when their selected durable evidence is obsolete; the -runner marks the aggregate record stale and executes it. Analyze's validator -can still accept a partial selection when only unrelated artifact records are -stale. +required. Extraction, analyze, and cleanup have resume validators and may +reject an otherwise eligible skip when their selected durable evidence is +obsolete; the runner marks the aggregate record stale and executes it. +Analyze's validator can still accept a partial selection when only unrelated +artifact records are stale. Implemented reuse coverage is deliberately split between aggregate semantic evidence and focused durable validators: @@ -238,6 +238,7 @@ evidence and focused durable validators: | analyze artifacts | per-artifact fingerprint, reconciliation, and output validator | [analyze](stage-analyze.md), [Scriptorium](../integrations/scriptorium.md) | | publish | aggregate semantic fingerprint plus immediate lock/commit checks | [publish](stage-publish.md), [storage adapter](adapters.md) | | notify | aggregate delivery-mode fingerprint | [pipeline overview](overview.md), [configuration](../config.md#notifications) | +| cleanup | aggregate semantic fingerprint plus publish-evidence and target-absence validator | [cleanup](stage-cleanup.md), [workspace](workspace.md) | These contracts record resolved choices Narratio can observe, not operational runner tuning. External model, module, prompt, profile, and configuration-file diff --git a/docs/internal/stage-analyze.md b/docs/internal/stage-analyze.md index 640e8b8..4043cd4 100644 --- a/docs/internal/stage-analyze.md +++ b/docs/internal/stage-analyze.md @@ -158,9 +158,10 @@ Supported source families: artifact failed without output authority, and stales its current transitive dependents. - the runner's ordinary error path marks aggregate analyze failed, clears its - aggregate outputs, invalidates publish and notify through the application - dependency relation, and persists the live retry collection. Terminal-save - errors are joined with the original artifact or checkpoint failure. + aggregate outputs, invalidates publish, notify, and cleanup through the + application dependency relation, and persists the live retry collection. + Terminal-save errors are joined with the original artifact or checkpoint + failure. ## Invariants diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index b130778..892488e 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -285,7 +285,7 @@ documents fail strict decoding rather than entering a compatibility path. ## Stage 5 — Consolidate, Validate, And Close Workstream 9 -**Status:** Pending +**Status:** Completed ### Goal diff --git a/docs/roadmap/refactoring.md b/docs/roadmap/refactoring.md index c31f533..d993c65 100644 --- a/docs/roadmap/refactoring.md +++ b/docs/roadmap/refactoring.md @@ -420,7 +420,7 @@ version of a locked artifact while unlocked artifacts advance. ### Workstream 9 — Cleanup as the final pipeline stage -**Status:** Pending +**Status:** Complete **Audit sources:** H20, H21, C02, and C04. diff --git a/internal/app/cleanup_orchestration_test.go b/internal/app/cleanup_orchestration_test.go index 383d696..52a953a 100644 --- a/internal/app/cleanup_orchestration_test.go +++ b/internal/app/cleanup_orchestration_test.go @@ -1,11 +1,13 @@ package app import ( + "bytes" "context" "errors" "os" "path/filepath" "reflect" + "strings" "testing" "time" @@ -162,6 +164,76 @@ func TestCleanupFailurePreservesPublishAndRetriesWithoutRepublishing(t *testing. } } +func TestRunStageCleanupUsesPublishEvidenceAndFailsClosedWithoutIt(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + enableCleanupForCommandTest(t, pipelinePath) + cfg, err := config.Load(pipelinePath, campaignPath, sessionPath) + if err != nil { + t.Fatalf("load cleanup config: %v", err) + } + + publishedRunID := "20260904T010203Z-a1b2c3d4" + m := manifest.New(cfg.Session.SessionID, time.Date(2026, 9, 4, 1, 2, 3, 0, time.UTC)) + m.Campaign = cfg.Session.Campaign + m.RunID = publishedRunID + m.S3SessionPrefix = artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, m.Campaign, m.SessionID) + m.MarkStageSucceeded("publish", time.Date(2026, 9, 4, 1, 3, 0, 0, time.UTC), nil) + m.Stages["publish"].Metadata = cleanupPublishMetadata(m) + manifestPath := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID) + if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPath, m); err != nil { + t.Fatalf("save publication evidence: %v", err) + } + workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID, publishedRunID) + spoolTarget := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, m.Campaign, m.SessionID, publishedRunID) + for _, target := range []string{workspaceTarget, spoolTarget} { + if err := os.MkdirAll(target, 0o755); err != nil { + t.Fatalf("create cleanup target: %v", err) + } + } + + var output bytes.Buffer + if err := RunStage(context.Background(), []string{ + "cleanup", cfg.Session.SessionID, "--config", pipelinePath, + "--campaign-file", campaignPath, "--session", sessionPath, + }, &output); err != nil { + t.Fatalf("RunStage(cleanup): %v", err) + } + if !strings.Contains(output.String(), "stage=cleanup executed=1 skipped=0") { + t.Fatalf("output = %q, want direct cleanup execution", output.String()) + } + for _, target := range []string{workspaceTarget, spoolTarget} { + if _, statErr := os.Lstat(target); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("cleanup target %q stat error = %v, want absent", target, statErr) + } + } + + missingRoot := t.TempDir() + missingPipeline, missingCampaign, missingSession := writeValidConfigFiles(t, missingRoot) + enableCleanupForCommandTest(t, missingPipeline) + err = RunStage(context.Background(), []string{ + "cleanup", "2026-05-03", "--config", missingPipeline, + "--campaign-file", missingCampaign, "--session", missingSession, + }, &bytes.Buffer{}) + if err == nil || !strings.Contains(err.Error(), "cleanup requires a recorded publish stage") { + t.Fatalf("RunStage(cleanup) without evidence error = %v", err) + } +} + +func enableCleanupForCommandTest(t *testing.T, pipelinePath string) { + t.Helper() + data, err := os.ReadFile(pipelinePath) + if err != nil { + t.Fatalf("read pipeline config: %v", err) + } + content := strings.Replace(string(data), "workspace:\n root: ", "workspace:\n cleanup_after_publish: true\n root: ", 1) + content = strings.Replace(content, "spool:\n root: ", "spool:\n delete_audio_after_publish: true\n root: ", 1) + content = strings.Replace(content, " upload_run: false", " upload_run: true", 1) + if err := os.WriteFile(pipelinePath, []byte(content), 0o644); err != nil { + t.Fatalf("write cleanup config: %v", err) + } +} + func cleanupOrchestrationConfig(t *testing.T) (*config.Config, *storage.FakeBackend) { t.Helper() cfg := testConfig(t) diff --git a/internal/config/validate_publish.go b/internal/config/validate_publish.go index 6a2b614..2fad95d 100644 --- a/internal/config/validate_publish.go +++ b/internal/config/validate_publish.go @@ -91,10 +91,10 @@ func validateCleanupPublishPolicy(cfg *PipelineConfig) error { 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") + return fmt.Errorf("cleanup stage 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 fmt.Errorf("cleanup stage policies require pipeline.publish.upload_run=true") } return nil }