From adc60aa588be43a9c1e323b8973f6293cab90dac Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 5 Sep 2026 15:27:08 +0000 Subject: [PATCH] Integrate cleanup into pipeline execution --- docs/cli.md | 11 +- docs/internal/overview.md | 10 +- docs/internal/stage-cleanup.md | 54 ++ docs/internal/stage-publish.md | 11 +- docs/internal/workspace.md | 22 +- docs/operations.md | 28 +- docs/policy/architecture.md | 18 +- docs/roadmap/implementation.md | 2 +- .../app/analyze_artifacts_commands_test.go | 2 +- internal/app/assembled_workflow_test.go | 6 +- internal/app/cleanup_orchestration_test.go | 201 ++++++ internal/app/cleanup_targets.go | 10 + internal/app/commands_test.go | 4 +- internal/app/config_commands_test.go | 2 + internal/app/plan_test.go | 8 +- internal/app/planner_test.go | 27 +- internal/app/post_publish_cleanup.go | 204 ------ internal/app/post_publish_cleanup_test.go | 666 ------------------ internal/app/run_control.go | 3 +- internal/app/run_control_test.go | 27 +- internal/app/run_stage_test.go | 16 +- internal/app/runner_dependencies.go | 20 +- internal/app/runner_finalization.go | 50 -- internal/app/runner_test.go | 15 +- internal/stage/placeholders.go | 1 + 25 files changed, 405 insertions(+), 1013 deletions(-) create mode 100644 docs/internal/stage-cleanup.md create mode 100644 internal/app/cleanup_orchestration_test.go delete mode 100644 internal/app/post_publish_cleanup.go delete mode 100644 internal/app/post_publish_cleanup_test.go diff --git a/docs/cli.md b/docs/cli.md index 9ffbb73..6d6850a 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -153,7 +153,7 @@ Behavior: - evaluates one inclusive contiguous range of the canonical stage order; - defaults an omitted `--from` to `prepare` and an omitted `--through` to - `notify`, so omitting both retains full-pipeline behavior; + `cleanup`, so omitting both retains full-pipeline behavior; - rejects unknown endpoints and a `--from` endpoint after `--through`; - runs `render` before `extract`; an omitted or disabled Notarius configuration records an explicit `notarius_disabled` self-skip; @@ -213,6 +213,7 @@ Valid stage names: - `analyze` - `publish` - `notify` +- `cleanup` Rules: @@ -242,6 +243,10 @@ Equivalent to: narratio run-stage publish --force [...common config flags] ``` +This command selects only `publish`; it does not invoke cleanup. Use a normal +full run to include configured cleanup, or `run-stage cleanup` to retry cleanup +from valid publication evidence. + ### `clean` ```bash @@ -255,8 +260,8 @@ Behavior: - `--all` removes all local session work and spool state; - cache remains unless `--clear-cache` is provided. -See [Operations: Cleanup](./operations.md#cleanup) for deletion scope and -post-publish cleanup behavior. +See [Operations: Cleanup](./operations.md#cleanup) for manual and pipeline-stage +deletion scope. ### `session plan` diff --git a/docs/internal/overview.md b/docs/internal/overview.md index cc72668..ca5b6bd 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -26,7 +26,8 @@ The pipeline runner follows one explicit invocation lifecycle: inspect bounded prerequisites without mutation, prepare the workspace layout, acquire one session lock, load or create the authoritative session manifest, compose only the collaborators needed by the selected range, execute each selected stage in -order, complete any required post-publish cleanup, and then return the summary. +order, and then return the summary. Cleanup, when selected, is the ordinary +final stage rather than runner finalization work. The lock remains held from manifest initialization through finalization, and a failed stage or required manifest save prevents later stages from starting. @@ -35,7 +36,7 @@ failed stage or required manifest save prevents later stages from starting. | Area | Implemented owners | Responsibility | | --- | --- | --- | | Executable | `cmd/narratio` | Process entry, standard stream wiring, argument handoff, and exit status. | -| Application orchestration | `internal/app` | Command dispatch, configuration selection, secret-file environment loading, production composition, session locking, planning, execution, restore, cleanup gates, and user-facing reporting. | +| Application orchestration | `internal/app` | Command dispatch, configuration selection, secret-file environment loading, production composition, session locking, planning, execution, restore, and user-facing reporting. | | Configuration | [`internal/config`](configuration.md) | Presence-aware root/import/profile composition, canonical party and family expansion, strict YAML loading, defaults, normalization, session templating, and validation. | | Pipeline stages | `internal/stage` | Canonical stage registry, shared stage contract, execution dependencies, and implemented stage behavior. | | External boundaries | `internal/adapters`, `internal/audio` | WhisperX HTTP, downstream subprocesses, notification, object storage, and S3 audio materialization behind Narratio contracts. | @@ -78,6 +79,7 @@ The implemented canonical order is: 9. [`analyze`](stage-analyze.md) 10. [`publish`](stage-publish.md) 11. `notify` (no-op) +12. [`cleanup`](stage-cleanup.md) `notify` currently has no persisted pipeline outputs and uses the explicit `noop` notification mode. Its versioned semantic evidence records that delivery @@ -91,7 +93,8 @@ registry owns the flat execution sequence. The application orchestration owner uses a fixed, validated dependency relation to find transitive dependents in canonical order. In particular, `render` and `extract` are sibling consumers of trimmed transcript state: neither invalidates the other, while either can stale -`analyze`, `publish`, and `notify`. +`analyze`, `publish`, `notify`, and `cleanup`. Cleanup is a dependent of notify, +so any earlier replacement can stale a prior cleanup result. ## Focused Documentation @@ -118,6 +121,7 @@ trimmed transcript state: neither invalidates the other, while either can stale - [`extract`](stage-extract.md) - [`analyze`](stage-analyze.md) - [`publish`](stage-publish.md) +- [`cleanup`](stage-cleanup.md) Use this map to find an owner, then read its focused documentation and tests before changing behavior. diff --git a/docs/internal/stage-cleanup.md b/docs/internal/stage-cleanup.md new file mode 100644 index 0000000..2452e91 --- /dev/null +++ b/docs/internal/stage-cleanup.md @@ -0,0 +1,54 @@ +# Stage: cleanup + +## Purpose + +Remove explicitly configured local data for the run selected by a successful +publish commit. Cleanup is the twelfth and final canonical stage, after +`notify`. + +## Authorization And Targets + +Cleanup is disabled unless at least one of these policies is true: + +- `pipeline.spool.delete_audio_after_publish` +- `pipeline.workspace.cleanup_after_publish` + +An enabled cleanup requires the manifest's publish stage to be succeeded with +`uploaded=true`, a canonical `published_run_id`, and the exact remote commit and +current commit-pointer keys for the manifest campaign/session prefix. The +configured campaign and session must match the manifest. Cleanup derives its +only possible targets through the canonical artifact path helpers: + +- the published run's session spool-audio directory; and +- the published run's workspace run directory. + +It does not scan parent directories, infer the newest run, or read object +storage. Every deletion uses the root-confined file operation, which rejects a +root itself, traversal, symlink escape, and out-of-root targets. + +## Execution And Resume + +Disabled cleanup records an explicit skipped result. Enabled cleanup removes +targets in deterministic policy order and treats an already-missing target as +success. A successful record contains the published run ID and completed policy +names. + +A previous cleanup success is reusable only while its semantic configuration +matches and every selected target remains absent. A recreated target makes the +stage non-resumable. A failure is recorded against cleanup; publish remains +succeeded and remote current state is unchanged. On a later normal full run, +valid upstream stages are skipped and cleanup is retried without republishing. + +`narratio publish` selects publish only and never invokes cleanup. A bounded run +through `notify` or any earlier endpoint also excludes it. Operators may run +`narratio run-stage cleanup ` directly; missing, failed, malformed, +or identity-mismatched publication evidence fails before deletion. + +## Implementation And Tests + +- Planning and evidence validation: `internal/stage/cleanup_plan.go` +- Execution and resume: `internal/stage/cleanup.go` +- Root-confined deletion: `internal/fileops/cleanup.go` +- Focused tests: `internal/stage/cleanup_plan_test.go`, + `internal/stage/cleanup_test.go`, and + `internal/app/cleanup_orchestration_test.go` diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 77c6e05..1d81b9d 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -109,8 +109,8 @@ Includes counts/lists for: - independent publish locks are not overridden by `--force`; every initially unlocked relevant source is reread immediately before current-state selection. -- post-commit local cleanup is authorized by the committed publish metadata and - is durably recorded by the application lifecycle before any local deletion. +- successful uploaded publication records the exact run ID, remote commit key, + and current-pointer key consumed by the later cleanup stage. ## Resume Semantics @@ -123,9 +123,12 @@ and run identifiers are excluded. Independent remote locks remain mutable state and are read directly by publish at planning and commit boundaries; semantic evidence does not replace those safety checks. -The commit boundary and cleanup gate are normative architecture invariants; see +The commit boundary and cleanup authorization are normative architecture invariants; see [Architecture](../policy/architecture.md#publish-commit-boundary). +Publish never deletes local state. The standalone `publish` command selects +only this stage; a full run selects notify and cleanup afterward. + ## Related Contracts And Tests - [Configuration](../config.md#publish-configuration-summary) owns output @@ -141,4 +144,4 @@ The commit boundary and cleanup gate are normative architecture invariants; see `internal/stage/publish_test.go`, `internal/stage/semantic_contracts_delivery.go`, `internal/app/operator_helpers_test.go`, and - `internal/app/post_publish_cleanup_test.go` + `internal/app/cleanup_orchestration_test.go` diff --git a/docs/internal/workspace.md b/docs/internal/workspace.md index 8b8fcca..1938876 100644 --- a/docs/internal/workspace.md +++ b/docs/internal/workspace.md @@ -47,18 +47,18 @@ lock only while their context remains active, and report a release failure. ## Cleanup Semantics -Automatic post-publish cleanup: +The canonical cleanup stage: -- is created only after a successful publish commit with complete publish - metadata, then is persisted before any deletion; -- requires `uploaded=true`, a remote commit key, and a current commit-pointer - key in publish metadata; +- runs after notify only when selected by the invocation range; +- requires a succeeded publish record with `uploaded=true`, the published run + ID, remote commit key, and current commit-pointer key; - consumes the resolved cleanup policy described in [Configuration](../config.md); - refuses unsafe deletes (root delete, traversal, and out-of-root targets); -- retries any recorded incomplete target on later invocations even when no - publish work is selected. Missing targets are a successful, idempotent - cleanup result only after the completion evidence is saved. +- treats missing targets as successful, idempotent deletion; and +- uses its ordinary failed/succeeded manifest stage record for retry. A later + full run can skip reusable upstream stages and retry cleanup without + republishing. Manual cleanup uses the same root-confined deletion mechanism. Invocation syntax and exact deletion scope belong in [CLI](../cli.md#clean) and @@ -82,9 +82,11 @@ syntax and exact deletion scope belong in [CLI](../cli.md#clean) and - Immutable bundle installation: `internal/stage/extract.go` - Workspace modes: `internal/fileops/modes.go` - Cleanup confinement: `internal/fileops/cleanup.go`, - `internal/app/cleanup_targets.go`, `internal/app/post_publish_cleanup.go` + `internal/stage/cleanup.go`, `internal/stage/cleanup_plan.go`, and + `internal/app/cleanup_targets.go` - Tests: `internal/artifacts/paths_model_test.go`, `internal/artifacts/local_test.go`, `internal/stage/run_local_test.go`, `internal/fileops/modes_posix_test.go`, `internal/fileops/cleanup_test.go`, `internal/app/cleanup_targets_test.go`, - `internal/app/post_publish_cleanup_test.go` + `internal/stage/cleanup_test.go`, and + `internal/app/cleanup_orchestration_test.go` diff --git a/docs/operations.md b/docs/operations.md index c5e739b..4077e67 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -98,6 +98,7 @@ Canonical stage order: 9. `analyze` 10. `publish` 11. `notify` +12. `cleanup` Execution rules: @@ -154,7 +155,7 @@ narratio run 2026-04-04 --from extract --through analyze --force ``` Omitting `--from` selects from `prepare`; omitting `--through` selects through -`notify`. Force applies only within the selected range. Repeating `--from`, +`cleanup`. Force applies only within the selected range. Repeating `--from`, `--through`, or `--force` is rejected instead of resolving by argument order. The plan command uses the same selection contract and prints only the selected range. Planning is read-only: it clones the loaded manifest, models selected @@ -194,8 +195,8 @@ This command is a transparent expansion to a forced bounded `run` from `extract` through `analyze`. Extraction always rebuilds its complete configured bundle. Analysis rebuilds the selected targets and their required analysis prerequisites, or uses the normal default selection when no artifact names are -given. The command does not run publish or notify; delivery remains a separate -operator action. +given. The command does not run publish, notify, or cleanup; delivery and +retention remain separate operator actions. Inspect current artifact evidence, then publish explicitly when the regenerated set is ready: @@ -365,6 +366,9 @@ Equivalent: narratio run-stage publish 2026-04-04 --force ``` +This standalone command stops after publish and does not invoke cleanup. A +default full `run` continues through notify and cleanup. + Publish commit model: - uploads eligible run files under `{session_prefix}/runs/{run_id}/`, excluding @@ -591,16 +595,20 @@ Rules: - cache is preserved unless `--clear-cache` is set; - each deletion is confined beneath its configured workspace, spool, or cache root; root, traversal, and out-of-root targets are refused; -- automatic post-publish cleanup is gated by successful publish commit plus: +- the final cleanup stage is gated by successful publish evidence plus: - `pipeline.spool.delete_audio_after_publish=true` - `pipeline.workspace.cleanup_after_publish=true` - - Narratio first records the exact run-scoped cleanup obligation. If cleanup - reports incomplete, the remote committed snapshot remains current; rerun - publish to retry only the outstanding confined local cleanup. + - the manifest publish record has `uploaded=true`, the published run ID, the + exact remote commit key, and the current commit-pointer key. -Post-publish cleanup is evaluated only when `publish` actually executes in the -current invocation. A bounded range that excludes publish does not replay a -cleanup obligation as an unrelated side effect. +Cleanup derives only the configured run-scoped spool and workspace targets, +uses root-confined deletion, and does not read or mutate object storage. A +failure leaves publish succeeded and the remote committed snapshot current. +Retry with a normal full run: valid upstream work, including publish and notify, +is skipped while the failed cleanup stage runs again. `run-stage cleanup` is +also available when the manifest has valid publish evidence. A range ending at +`notify` or earlier and the standalone `publish` command never delete local +state. ## Operational Caveats diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 79a8769..f2b509f 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -185,8 +185,8 @@ availability. Writes, moves, replacements, and deletions must use narrow, explicit, root-confined destinations. Symlinks, traversal, broad roots, and ambiguous relative destinations must not expand the scope of an operation. Cleanup is -permitted only through explicit operator action or configured post-publish -gates, and it must preserve durable cache unless cache removal is explicitly +permitted only through explicit operator action or the configured final cleanup +stage, and it must preserve durable cache unless cache removal is explicitly requested. Physical layout, retention, and operational lifecycle belong in @@ -212,10 +212,16 @@ transaction boundary. Automatic local cleanup is permitted only after a successful publish commit, only when explicitly configured, and only through the path-safety guardrails. -It is a durable local obligation bound to that committed run and its exact -targets, not an inferred side effect of the current stage list. A cleanup -failure makes the invocation incomplete while leaving the committed remote -snapshot authoritative; later invocations resume the recorded obligation. +It is the final ordinary pipeline stage after `notify`. The stage derives its +exact run-scoped targets from validated publish metadata and the configured +campaign/session identity; it never infers a target from directory contents or +rereads remote storage. A cleanup failure makes the invocation incomplete while +leaving publish succeeded and the committed remote snapshot authoritative. A +normal later full run reuses valid upstream stages and retries cleanup. + +The standalone `publish` command selects only publish and therefore never +invokes cleanup. Operators may select cleanup directly only when the manifest +contains valid successful publication evidence. ## Security, Privacy, And Diagnostics diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index ef03b0c..8813ae2 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -193,7 +193,7 @@ filesystem and manifest contracts. ## Stage 3 — Integrate Cleanup Into Canonical Orchestration -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/app/analyze_artifacts_commands_test.go b/internal/app/analyze_artifacts_commands_test.go index 53c2810..b326438 100644 --- a/internal/app/analyze_artifacts_commands_test.go +++ b/internal/app/analyze_artifacts_commands_test.go @@ -151,7 +151,7 @@ func TestRunArtifactsWithSucceededAnalyzeSkipsUnlessForced(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=4 skipped=8") { + if !strings.Contains(out.String(), "executed=5 skipped=9") { t.Fatalf("output = %q, want extract reconsidered and legacy analyze plus delivery rebuilt", out.String()) } } diff --git a/internal/app/assembled_workflow_test.go b/internal/app/assembled_workflow_test.go index 89d49aa..57cac88 100644 --- a/internal/app/assembled_workflow_test.go +++ b/internal/app/assembled_workflow_test.go @@ -23,7 +23,7 @@ func TestAssembledFullRunUsesCanonicalOrder(t *testing.T) { cfg := testConfig(t) canonical := []string{ "prepare", "transcribe", "merge", "polish", "normalize", "trim", - "render", "extract", "analyze", "publish", "notify", + "render", "extract", "analyze", "publish", "notify", "cleanup", } var order []string stages := make([]stage.Stage, 0, len(canonical)) @@ -106,7 +106,7 @@ func TestAssembledForcedSiblingIndependenceAndFailureBoundary(t *testing.T) { if loaded.Stages[sibling].Status != manifest.StatusSucceeded { t.Fatalf("%s sibling = %#v, want succeeded", sibling, loaded.Stages[sibling]) } - for _, dependent := range []string{"analyze", "publish", "notify"} { + for _, dependent := range []string{"analyze", "publish", "notify", "cleanup"} { if loaded.Stages[dependent].Status != manifest.StatusStale { t.Fatalf("%s status = %q, want stale", dependent, loaded.Stages[dependent].Status) } @@ -127,7 +127,7 @@ func TestAssembledForcedSiblingIndependenceAndFailureBoundary(t *testing.T) { if loaded.Stages["extract"].Status != manifest.StatusSucceeded { t.Fatalf("extract sibling = %#v", loaded.Stages["extract"]) } - for _, outside := range []string{"analyze", "publish", "notify"} { + for _, outside := range []string{"analyze", "publish", "notify", "cleanup"} { if loaded.Stages[outside].Status != manifest.StatusStale { t.Fatalf("outside stage %s = %#v, want stale and unexecuted", outside, loaded.Stages[outside]) } diff --git a/internal/app/cleanup_orchestration_test.go b/internal/app/cleanup_orchestration_test.go new file mode 100644 index 0000000..383d696 --- /dev/null +++ b/internal/app/cleanup_orchestration_test.go @@ -0,0 +1,201 @@ +package app + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +const cleanupOrchestrationRunID = "20260905T010203Z-a1b2c3d4" + +type cleanupPublishEvidenceStage struct { + runs *int + skipSpoolTarget bool +} + +func (cleanupPublishEvidenceStage) Name() string { return "publish" } + +func (s cleanupPublishEvidenceStage) Run(_ context.Context, env *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { + if s.runs != nil { + *s.runs++ + } + if !s.skipSpoolTarget { + spoolTarget := artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, m.Campaign, m.SessionID, m.RunID) + if err := os.MkdirAll(spoolTarget, 0o755); err != nil { + return nil, err + } + if err := os.WriteFile(filepath.Join(spoolTarget, "audio.flac"), []byte("audio"), 0o644); err != nil { + return nil, err + } + } + workspaceTarget := artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID) + if err := os.MkdirAll(workspaceTarget, 0o755); err != nil { + return nil, err + } + return &stage.StageResult{Metadata: cleanupPublishMetadata(m)}, nil +} + +func cleanupPublishMetadata(m *manifest.Manifest) map[string]any { + return map[string]any{ + "uploaded": true, + "published_run_id": m.RunID, + "remote_commit_key": artifacts.S3RunCommitKey(m.S3SessionPrefix, m.RunID), + "current_commit_pointer_key": artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix), + } +} + +func TestCleanupStopsAtPublishAndNotifyFailureBoundaries(t *testing.T) { + tests := []struct { + name string + stages func(*int) []stage.Stage + }{ + { + name: "publish failure", + stages: func(_ *int) []stage.Stage { + return []stage.Stage{failingStage{name: "publish", err: errors.New("publish failed")}, selectedCleanupStage(t)} + }, + }, + { + name: "notify failure", + stages: func(publishRuns *int) []stage.Stage { + return []stage.Stage{ + cleanupPublishEvidenceStage{runs: publishRuns}, + failingStage{name: "notify", err: errors.New("notify failed")}, + selectedCleanupStage(t), + } + }, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + cfg, remote := cleanupOrchestrationConfig(t) + publishRuns := 0 + _, err := executePlanWithDependencies( + context.Background(), cfg, BoundedPlan{stages: test.stages(&publishRuns)}, + RunOptions{Env: &Env{ObjectStore: remote}}, cleanupOrchestrationDependencies(), + ) + if err == nil { + t.Fatal("execution error = nil") + } + loaded := loadCleanupOrchestrationManifest(t, cfg) + if loaded.Stages["cleanup"] != nil { + t.Fatalf("cleanup record = %#v, want no execution", loaded.Stages["cleanup"]) + } + if test.name == "notify failure" { + workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, cleanupOrchestrationRunID) + if _, statErr := os.Stat(workspaceTarget); statErr != nil { + t.Fatalf("workspace target removed after notify failure: %v", statErr) + } + } + }) + } +} + +func TestCleanupFailurePreservesPublishAndRetriesWithoutRepublishing(t *testing.T) { + cfg, remote := cleanupOrchestrationConfig(t) + badSpoolRoot := filepath.Join(t.TempDir(), "spool-file") + if err := os.WriteFile(badSpoolRoot, []byte("not a directory"), 0o644); err != nil { + t.Fatalf("write invalid spool root: %v", err) + } + cfg.Pipeline.Spool.Root = badSpoolRoot + remote.SeedObject(storage.FakeObject{Key: "sentinel/remote-state", Data: []byte("unchanged")}) + publishRuns := 0 + plan := BoundedPlan{stages: []stage.Stage{ + cleanupPublishEvidenceStage{runs: &publishRuns, skipSpoolTarget: true}, + resultStage{name: "notify"}, + selectedCleanupStage(t), + }} + + _, err := executePlanWithDependencies( + context.Background(), cfg, plan, RunOptions{Env: &Env{ObjectStore: remote}}, cleanupOrchestrationDependencies(), + ) + if err == nil { + t.Fatal("cleanup failure = nil") + } + loaded := loadCleanupOrchestrationManifest(t, cfg) + if loaded.Stages["publish"] == nil || loaded.Stages["publish"].Status != manifest.StatusSucceeded { + t.Fatalf("publish record = %#v, want succeeded", loaded.Stages["publish"]) + } + if loaded.Stages["cleanup"] == nil || loaded.Stages["cleanup"].Status != manifest.StatusFailed { + t.Fatalf("cleanup record = %#v, want failed", loaded.Stages["cleanup"]) + } + if got := string(remote.Objects["sentinel/remote-state"].Data); got != "unchanged" { + t.Fatalf("remote sentinel = %q, want unchanged", got) + } + if len(remote.Uploads)+len(remote.Downloads)+len(remote.Reads)+len(remote.Deletes)+len(remote.ExistsKeys) != 0 { + t.Fatalf("cleanup touched remote storage: %#v", remote) + } + + workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, cleanupOrchestrationRunID) + cfg.Pipeline.Spool.Root = t.TempDir() + spoolTarget := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID, cleanupOrchestrationRunID) + if err := os.MkdirAll(spoolTarget, 0o755); err != nil { + t.Fatalf("create retry spool target: %v", err) + } + summary, err := executePlanWithDependencies( + context.Background(), cfg, plan, RunOptions{Env: &Env{ObjectStore: remote}}, cleanupOrchestrationDependencies(), + ) + if err != nil { + t.Fatalf("retry cleanup: %v", err) + } + if publishRuns != 1 { + t.Fatalf("publish runs = %d, want one", publishRuns) + } + if !reflect.DeepEqual(summary.Executed, []string{"cleanup"}) || !reflect.DeepEqual(summary.Skipped, []string{"publish", "notify"}) { + t.Fatalf("retry summary = %#v", summary) + } + if _, statErr := os.Lstat(workspaceTarget); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("workspace target stat error = %v, want absent", statErr) + } + if _, statErr := os.Lstat(spoolTarget); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("spool target stat error = %v, want absent", statErr) + } +} + +func cleanupOrchestrationConfig(t *testing.T) (*config.Config, *storage.FakeBackend) { + t.Helper() + cfg := testConfig(t) + cfg.Pipeline.Spool.Root = t.TempDir() + cfg.Pipeline.Spool.DeleteAudioAfterPublish = true + cfg.Pipeline.Workspace.CleanupAfterPublish = true + cfg.Pipeline.Publish = &config.PublishConfig{Enabled: boolPtr(true), UploadRun: boolPtr(true)} + cfg.Pipeline.Storage.Backend = config.StorageBackendS3 + cfg.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "archive", RootPrefix: "dnd"} + return cfg, &storage.FakeBackend{} +} + +func selectedCleanupStage(t *testing.T) stage.Stage { + t.Helper() + selected, err := BuildSingleStagePlan("cleanup") + if err != nil { + t.Fatalf("select cleanup: %v", err) + } + return selected[0] +} + +func cleanupOrchestrationDependencies() invocationDependencies { + return invocationDependencies{ + now: func() time.Time { return time.Date(2026, 9, 5, 1, 2, 3, 0, time.UTC) }, + newRunID: func() (string, error) { return cleanupOrchestrationRunID, nil }, + } +} + +func loadCleanupOrchestrationManifest(t *testing.T, cfg *config.Config) *manifest.Manifest { + t.Helper() + path := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID) + loaded, err := (&manifest.LocalStore{}).Load(context.Background(), path) + if err != nil { + t.Fatalf("load manifest: %v", err) + } + return loaded +} diff --git a/internal/app/cleanup_targets.go b/internal/app/cleanup_targets.go index 4ca8e2c..fa7f3a5 100644 --- a/internal/app/cleanup_targets.go +++ b/internal/app/cleanup_targets.go @@ -7,6 +7,16 @@ import ( "strings" ) +type scopedDir struct { + RootAbs string + TargetAbs string + Exists bool +} + +func validateScopedDir(root, target, policy string) (scopedDir, error) { + return validateScopedTarget(root, target, policy, true) +} + func validateScopedTarget(root, target, policy string, requireDir bool) (scopedDir, error) { cleanRoot := strings.TrimSpace(root) cleanTarget := strings.TrimSpace(target) diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index 8ee322f..666ed17 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -32,7 +32,7 @@ func TestExecuteValidCommands(t *testing.T) { args []string wantOut string }{ - {name: "run", args: []string{"run", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run: session 2026-05-03; executed=11 skipped=1; manifest="}, + {name: "run", args: []string{"run", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run: session 2026-05-03; executed=12 skipped=2; manifest="}, {name: "session plan", args: []string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "analyze: skip\n targets: none\n prerequisites: none\n execute: none\n reuse: none\npublish: skip\nnotify: skip"}, {name: "session status", args: []string{"session", "status", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "Session: 2026-05-03"}, {name: "run-stage", args: []string{"run-stage", "polish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run-stage: stage=polish executed=0 skipped=1 force=false; manifest="}, @@ -336,7 +336,7 @@ func TestExecuteUsesDefaultPipelineConfigPathWhenConfigFlagOmitted(t *testing.T) if code != 0 { t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) } - if !strings.Contains(stdout.String(), "narratio run: session 2026-05-03; executed=11 skipped=1; manifest=") { + if !strings.Contains(stdout.String(), "narratio run: session 2026-05-03; executed=12 skipped=2; manifest=") { t.Fatalf("stdout = %q, want successful run output", stdout.String()) } } diff --git a/internal/app/config_commands_test.go b/internal/app/config_commands_test.go index 82418b3..f803ce6 100644 --- a/internal/app/config_commands_test.go +++ b/internal/app/config_commands_test.go @@ -116,6 +116,8 @@ func TestConfigDiffReportsSemanticProfileChanges(t *testing.T) { pipelinePath, _, _ := writeValidConfigFiles(t, workspaceRoot) writeInspectionProfiles(t, pipelinePath, `workspace: cleanup_after_publish: true +publish: + upload_run: true whisperx: language: en audita: diff --git a/internal/app/plan_test.go b/internal/app/plan_test.go index f82b7bf..74246bf 100644 --- a/internal/app/plan_test.go +++ b/internal/app/plan_test.go @@ -32,12 +32,12 @@ func TestPlanDoesNotCreateWorkdir(t *testing.T) { if !strings.Contains(got, "narratio session plan: read-only workdir at") { t.Fatalf("first output = %q, want read-only workdir", got) } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"} { if !strings.Contains(got, name+": run") { t.Fatalf("first output = %q, missing stage %q", got, name) } } - if !strings.Contains(got, "totals: run=11 skip=0") { + if !strings.Contains(got, "totals: run=12 skip=0") { t.Fatalf("first output = %q, want totals", got) } @@ -83,8 +83,8 @@ func TestPlanShowsRunAndSkipFromManifest(t *testing.T) { if !strings.Contains(got, "trim: run") { t.Fatalf("output = %q, want trim run", got) } - if !strings.Contains(got, "totals: run=9 skip=2") { - t.Fatalf("output = %q, want totals run=9 skip=2", got) + if !strings.Contains(got, "totals: run=10 skip=2") { + t.Fatalf("output = %q, want totals run=10 skip=2", got) } } diff --git a/internal/app/planner_test.go b/internal/app/planner_test.go index 80314bb..90dc58f 100644 --- a/internal/app/planner_test.go +++ b/internal/app/planner_test.go @@ -10,7 +10,7 @@ import ( func TestBuildFullPlanOrder(t *testing.T) { got := BuildFullPlan() - want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"} + want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"} if len(got) != len(want) { t.Fatalf("len(plan) = %d, want %d", len(got), len(want)) } @@ -22,15 +22,15 @@ func TestBuildFullPlanOrder(t *testing.T) { } func TestBuildSingleStagePlan(t *testing.T) { - stages, err := BuildSingleStagePlan("polish") + stages, err := BuildSingleStagePlan("cleanup") if err != nil { t.Fatalf("BuildSingleStagePlan() error = %v", err) } if len(stages) != 1 { t.Fatalf("len(plan) = %d, want 1", len(stages)) } - if stages[0].Name() != "polish" { - t.Fatalf("stage name = %q, want %q", stages[0].Name(), "polish") + if stages[0].Name() != "cleanup" { + t.Fatalf("stage name = %q, want %q", stages[0].Name(), "cleanup") } } @@ -92,6 +92,19 @@ func TestBuildBoundedPlanDefaultsToFullCanonicalPlan(t *testing.T) { } } +func TestBuildBoundedPlanThroughNotifyExcludesCleanup(t *testing.T) { + plan, err := BuildBoundedPlan("", "notify") + if err != nil { + t.Fatalf("BuildBoundedPlan() error = %v", err) + } + if plan.Contains("cleanup") { + t.Fatalf("plan = %#v, cleanup must be outside a notify-bounded run", plan.Names()) + } + if got := plan.Through(); got != "notify" { + t.Fatalf("through = %q, want notify", got) + } +} + func TestBuildBoundedPlanRejectsInvalidBounds(t *testing.T) { tests := []struct { name string @@ -99,9 +112,9 @@ func TestBuildBoundedPlanRejectsInvalidBounds(t *testing.T) { through string want []string }{ - {name: "unknown from", from: "missing", through: "analyze", want: []string{"unknown from stage", "missing", "prepare", "notify"}}, - {name: "unknown through", from: "extract", through: "missing", want: []string{"unknown through stage", "missing", "prepare", "notify"}}, - {name: "reversed", from: "publish", through: "render", want: []string{"publish", "occurs after", "render", "prepare", "notify"}}, + {name: "unknown from", from: "missing", through: "analyze", want: []string{"unknown from stage", "missing", "prepare", "cleanup"}}, + {name: "unknown through", from: "extract", through: "missing", want: []string{"unknown through stage", "missing", "prepare", "cleanup"}}, + {name: "reversed", from: "publish", through: "render", want: []string{"publish", "occurs after", "render", "prepare", "cleanup"}}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { diff --git a/internal/app/post_publish_cleanup.go b/internal/app/post_publish_cleanup.go deleted file mode 100644 index 17fe3c8..0000000 --- a/internal/app/post_publish_cleanup.go +++ /dev/null @@ -1,204 +0,0 @@ -package app - -import ( - "context" - "fmt" - "path/filepath" - "strings" - - "gitea.maximumdirect.net/eric/narratio/internal/artifacts" - "gitea.maximumdirect.net/eric/narratio/internal/config" - "gitea.maximumdirect.net/eric/narratio/internal/fileops" - "gitea.maximumdirect.net/eric/narratio/internal/manifest" -) - -func runPostPublishCleanup( - ctx context.Context, - env *Env, - manifestPath string, - m *manifest.Manifest, - executed []string, - removeScopedDir func(root, target, policy string) error, -) error { - if env == nil || env.Config == nil || env.Config.Pipeline == nil || m == nil { - return nil - } - - cleanup := m.PostPublishCleanup - if cleanup == nil { - var err error - cleanup, err = createPostPublishCleanup(env.Config, m, executed) - if err != nil { - return err - } - if cleanup == nil { - return nil - } - m.PostPublishCleanup = cleanup - if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - return fmt.Errorf("persist post-publish cleanup obligation %q: %w", manifestPath, err) - } - } - - for index := range cleanup.Targets { - target := &cleanup.Targets[index] - if target.Completed { - continue - } - if err := removeScopedDir(target.Root, target.Path, target.Policy); err != nil { - return fmt.Errorf("complete post-publish cleanup for %q: %w", target.Path, err) - } - target.Completed = true - if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - target.Completed = false - return fmt.Errorf("persist post-publish cleanup completion %q: %w", manifestPath, err) - } - } - return nil -} - -func createPostPublishCleanup(cfg *config.Config, m *manifest.Manifest, executed []string) (*manifest.PostPublishCleanup, error) { - spoolRequested := cfg.Pipeline.Spool.DeleteAudioAfterPublish - workRequested := cfg.Pipeline.Workspace.CleanupAfterPublish - if !spoolRequested && !workRequested { - return nil, nil - } - - sr := publishStageRecordForCleanup(m) - publishedRunID, eligible, _ := publishCleanupEligible(cfg, m, sr, executed) - if !eligible { - return nil, nil - } - - spoolDir := strings.TrimSpace(m.LocalSpoolDir) - if spoolDir == "" { - spoolDir = artifacts.SessionSpoolAudioDir( - cfg.Pipeline.Spool.Root, - strings.TrimSpace(m.Campaign), - strings.TrimSpace(m.SessionID), - publishedRunID, - ) - } - workDir := strings.TrimSpace(m.LocalWorkDir) - if workDir == "" { - workDir = artifacts.SessionRunRootForCampaign( - cfg.Pipeline.Workspace.Root, - strings.TrimSpace(m.Campaign), - strings.TrimSpace(m.SessionID), - publishedRunID, - ) - } - - cleanup := &manifest.PostPublishCleanup{ - CommittedRunID: publishedRunID, - RemoteCommitKey: strings.TrimSpace(asString(sr.Metadata["remote_commit_key"])), - CurrentCommitPointerKey: strings.TrimSpace(asString(sr.Metadata["current_commit_pointer_key"])), - } - if spoolRequested { - cleanup.Targets = append(cleanup.Targets, manifest.CleanupTarget{ - Policy: "pipeline.spool.delete_audio_after_publish", - Root: strings.TrimSpace(cfg.Pipeline.Spool.Root), - Path: filepath.Clean(spoolDir), - }) - } - if workRequested { - cleanup.Targets = append(cleanup.Targets, manifest.CleanupTarget{ - Policy: "pipeline.workspace.cleanup_after_publish", - Root: strings.TrimSpace(cfg.Pipeline.Workspace.Root), - Path: filepath.Clean(workDir), - }) - } - return cleanup, nil -} - -func publishStageRecordForCleanup(m *manifest.Manifest) *manifest.StageRecord { - if m == nil { - return nil - } - sr := m.Stages["publish"] - if sr == nil || sr.Status != manifest.StatusSucceeded { - return nil - } - return sr -} - -func publishCleanupEligible(cfg *config.Config, m *manifest.Manifest, sr *manifest.StageRecord, executed []string) (string, bool, string) { - if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Publish == nil { - return "", false, "publish configuration is missing" - } - enabled := true - if cfg.Pipeline.Publish.Enabled != nil { - enabled = *cfg.Pipeline.Publish.Enabled - } - if !enabled { - return "", false, "publish.enabled is false" - } - uploadRun := true - if cfg.Pipeline.Publish.UploadRun != nil { - uploadRun = *cfg.Pipeline.Publish.UploadRun - } - if !uploadRun { - return "", false, "publish.upload_run is false" - } - if sr == nil || sr.Metadata == nil { - return "", false, "publish metadata is missing" - } - if skipped, _ := sr.Metadata["skipped"].(bool); skipped { - return "", false, "publish stage was skipped" - } - if uploaded, _ := sr.Metadata["uploaded"].(bool); !uploaded { - return "", false, "publish did not complete an upload" - } - if strings.TrimSpace(asString(sr.Metadata["remote_commit_key"])) == "" { - return "", false, "publish remote commit key is missing" - } - if strings.TrimSpace(asString(sr.Metadata["current_commit_pointer_key"])) == "" { - return "", false, "publish current commit pointer key is missing" - } - publishedRunID := strings.TrimSpace(asString(sr.Metadata["published_run_id"])) - if publishedRunID == "" && containsStage(executed, "publish") && m != nil { - publishedRunID = strings.TrimSpace(m.RunID) - } - if publishedRunID == "" { - return "", false, "publish run id is missing" - } - return publishedRunID, true, "" -} - -func containsStage(names []string, target string) bool { - for _, name := range names { - if name == target { - return true - } - } - return false -} - -type scopedDir struct { - RootAbs string - TargetAbs string - Exists bool -} - -func removeRunScopedDir(root, target, policy string) error { - dir, err := validateScopedDir(root, target, policy) - if err != nil { - return err - } - if !dir.Exists { - return nil - } - if err := fileops.RemoveAllUnderRoot(dir.RootAbs, dir.TargetAbs); err != nil { - return fmt.Errorf("cleanup policy %s: remove %q: %w", policy, dir.TargetAbs, err) - } - return nil -} - -func validateScopedDir(root, target, policy string) (scopedDir, error) { - return validateScopedTarget(root, target, policy, true) -} - -func asString(v any) string { - s, _ := v.(string) - return s -} diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go deleted file mode 100644 index 92cb5ba..0000000 --- a/internal/app/post_publish_cleanup_test.go +++ /dev/null @@ -1,666 +0,0 @@ -package app - -import ( - "context" - "errors" - "io" - "os" - "path/filepath" - "strings" - "testing" - "time" - - "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" - "gitea.maximumdirect.net/eric/narratio/internal/artifacts" - "gitea.maximumdirect.net/eric/narratio/internal/config" - "gitea.maximumdirect.net/eric/narratio/internal/manifest" - "gitea.maximumdirect.net/eric/narratio/internal/stage" -) - -type publishSuccessStage struct { - metadata map[string]any - targets *cleanupSeed -} - -func (publishSuccessStage) Name() string { return "publish" } -func (s publishSuccessStage) Run(_ context.Context, _ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) { - if s.targets != nil { - s.targets.runWorkDir = m.LocalWorkDir - s.targets.spoolAudioDir = m.LocalSpoolDir - if err := os.MkdirAll(filepath.Join(m.LocalWorkDir, "logs"), 0o755); err != nil { - return nil, err - } - if err := os.WriteFile(filepath.Join(m.LocalWorkDir, "logs", "stage.log"), []byte("log\n"), 0o644); err != nil { - return nil, err - } - if err := os.MkdirAll(m.LocalSpoolDir, 0o755); err != nil { - return nil, err - } - if err := os.WriteFile(filepath.Join(m.LocalSpoolDir, "speaker.flac"), []byte("flac\n"), 0o644); err != nil { - return nil, err - } - } - md := map[string]any{ - "stage": "publish", - "uploaded": true, - "published_run_id": m.RunID, - "remote_commit_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/20260519T010203Z-a1b2c3d4/commit.json", - "current_commit_pointer_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/commit-pointer.json", - } - for k, v := range s.metadata { - md[k] = v - } - return &stage.StageResult{Metadata: md}, nil -} - -type notifyFailStage struct{} - -func (notifyFailStage) Name() string { return "notify" } -func (notifyFailStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { - return nil, errors.New("notify failed") -} - -func TestPostPublishCleanupDisabledKeepsLocalDirs(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = false - cfg.Pipeline.Workspace.CleanupAfterPublish = false - - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) - assertExists(t, seed.localSourceAudio) -} - -func TestPostPublishCleanupSpoolOnly(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = false - - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - - assertMissing(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) - assertExists(t, seed.localSourceAudio) -} - -func TestPostPublishCleanupWorkdirOnly(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = false - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - - assertExists(t, cfg.Pipeline.Workspace.Root) - assertExists(t, seed.otherRunDir) - assertExists(t, seed.previousCachePath) - assertMissing(t, seed.runWorkDir) - assertExists(t, seed.spoolAudioDir) -} - -func TestPostPublishCleanupBothPolicies(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - - assertMissing(t, seed.spoolAudioDir) - assertMissing(t, seed.runWorkDir) - assertExists(t, seed.otherRunDir) - assertExists(t, seed.previousCachePath) -} - -func TestPostPublishCleanupRetriesWhenInitialObligationSaveFails(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = false - - failed := false - store := &cleanupFailingManifestStore{ - delegate: &manifest.LocalStore{}, - fail: func(m *manifest.Manifest) error { - if !failed && m.PostPublishCleanup != nil { - failed = true - return errors.New("injected obligation save failure") - } - return nil - }, - } - _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{ - Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}, - }) - if err == nil || !strings.Contains(err.Error(), "post-publish cleanup incomplete") { - t.Fatalf("executeStages() error = %v, want incomplete cleanup", err) - } - assertExists(t, seed.spoolAudioDir) - assertCleanupPending(t, cfg) - - store.fail = nil - if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("non-publish executeStages() error = %v", err) - } - assertExists(t, seed.spoolAudioDir) - assertCleanupPending(t, cfg) - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("publish retry executeStages() error = %v", err) - } - assertMissing(t, seed.spoolAudioDir) - assertCleanupComplete(t, cfg) -} - -func TestPostPublishCleanupRetriesFailedDeletionWithoutTouchingOtherRuns(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - remove := func(root, target, policy string) error { - if policy == "pipeline.workspace.cleanup_after_publish" { - return errors.New("injected deletion failure") - } - return removeRunScopedDir(root, target, policy) - } - - _, err := executePlanWithDependencies( - context.Background(), - cfg, - BoundedPlan{stages: []stage.Stage{publishSuccessStage{targets: &seed}}}, - RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}, - invocationDependencies{removeScopedDir: remove}, - ) - if err == nil || !strings.Contains(err.Error(), "post-publish cleanup incomplete") { - t.Fatalf("executeStages() error = %v, want incomplete cleanup", err) - } - assertMissing(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) - assertCleanupPending(t, cfg) - assertExists(t, seed.otherRunDir) - assertExists(t, seed.previousCachePath) - - if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("non-publish executeStages() error = %v", err) - } - assertExists(t, seed.runWorkDir) - assertCleanupPending(t, cfg) - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("publish retry executeStages() error = %v", err) - } - assertMissing(t, seed.runWorkDir) - assertExists(t, seed.otherRunDir) - assertExists(t, seed.previousCachePath) - assertCleanupComplete(t, cfg) -} - -func TestPostPublishCleanupRetriesWhenCompletionEvidenceSaveFails(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = false - - failed := false - store := &cleanupFailingManifestStore{ - delegate: &manifest.LocalStore{}, - fail: func(m *manifest.Manifest) error { - if m.PostPublishCleanup == nil { - return nil - } - for _, target := range m.PostPublishCleanup.Targets { - if !failed && target.Completed { - failed = true - return errors.New("injected completion evidence failure") - } - } - return nil - }, - } - _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{ - Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}, - }) - if err == nil || !strings.Contains(err.Error(), "post-publish cleanup incomplete") { - t.Fatalf("executeStages() error = %v, want incomplete cleanup", err) - } - assertMissing(t, seed.spoolAudioDir) - assertCleanupPending(t, cfg) - - store.fail = nil - if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("non-publish executeStages() error = %v", err) - } - assertCleanupPending(t, cfg) - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("publish retry executeStages() error = %v", err) - } - assertCleanupComplete(t, cfg) - - if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ManifestStore: store, ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("idempotent non-publish executeStages() error = %v", err) - } - assertMissing(t, seed.spoolAudioDir) -} - -func TestPostPublishCleanupNotRunWhenPublishFails(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - _, err := executeStages(context.Background(), cfg, []stage.Stage{failingStage{name: "publish", err: errors.New("publish failed")}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) - if err == nil || !strings.Contains(err.Error(), "stage \"publish\" failed") { - t.Fatalf("executeStages() error = %v, want publish failure", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) -} - -func TestPostPublishCleanupNotRunWhenPublishSkipped(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"skipped": true}, targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) -} - -func TestPostPublishCleanupNotRunWhenCommitPointerMissing(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{metadata: map[string]any{"current_commit_pointer_key": ""}, targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) -} - -func TestPostPublishCleanupNotRunWhenPublishUploadDisabled(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - cfg.Pipeline.Publish.UploadRun = boolPtr(false) - - if _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("executeStages() error = %v", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) -} - -func TestPostPublishCleanupWaitsUntilAllStagesSucceed(t *testing.T) { - cfg, seed := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}, notifyFailStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) - if err == nil || !strings.Contains(err.Error(), "stage \"notify\" failed") { - t.Fatalf("executeStages() error = %v, want notify failure", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) -} - -func TestPostPublishCleanupFailsOnUnsafePath(t *testing.T) { - cfg, _ := cleanupFixtureConfig(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = false - - manifestPath := manifestPathFor(cfg) - store := &manifest.LocalStore{} - m, err := store.Load(context.Background(), manifestPath) - if err != nil { - t.Fatalf("Load() error = %v", err) - } - m.MarkStageSucceeded("publish", time.Now().UTC(), nil) - m.Stages["publish"].Metadata = map[string]any{ - "uploaded": true, - "published_run_id": m.RunID, - "remote_commit_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/20260516T010203Z-1a2b3c4d/commit.json", - "current_commit_pointer_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/commit-pointer.json", - } - m.PostPublishCleanup = &manifest.PostPublishCleanup{ - CommittedRunID: m.RunID, - RemoteCommitKey: m.Stages["publish"].Metadata["remote_commit_key"].(string), - CurrentCommitPointerKey: m.Stages["publish"].Metadata["current_commit_pointer_key"].(string), - Targets: []manifest.CleanupTarget{{ - Policy: "pipeline.spool.delete_audio_after_publish", - Root: cfg.Pipeline.Spool.Root, - Path: filepath.Join(filepath.Dir(cfg.Pipeline.Spool.Root), "outside-spool"), - }}, - } - if err := store.Save(context.Background(), manifestPath, m); err != nil { - t.Fatalf("Save() error = %v", err) - } - - if _, err = executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { - t.Fatalf("non-publish executeStages() error = %v", err) - } - _, err = executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{}}, RunOptions{Force: true, Env: &Env{ObjectStore: &storage.FakeBackend{}}}) - if err == nil || !strings.Contains(err.Error(), "refusing to delete path outside root") { - t.Fatalf("executeStages() error = %v, want safe-path failure", err) - } -} - -func TestPostPublishCleanupNotRunWhenOutputIsMissing(t *testing.T) { - cfg, seed, runID := publishStageCleanupFixture(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - cfg.Pipeline.Publish.Outputs = []config.PublishOutputRule{ - {Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)}, - } - - publishStageImpl, err := stage.Select("publish") - if err != nil { - t.Fatalf("Select(publish) error = %v", err) - } - _, err = executeStages(context.Background(), cfg, []stage.Stage{publishStageImpl}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) - if err == nil || !strings.Contains(err.Error(), "required output source unavailable") { - t.Fatalf("executeStages() error = %v, want required output source unavailable failure", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) - assertExists(t, artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID)) -} - -func TestPostPublishCleanupNotRunWhenCommittedManifestUploadFails(t *testing.T) { - cfg, seed, _ := publishStageCleanupFixture(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - - publishStageImpl, err := stage.Select("publish") - if err != nil { - t.Fatalf("Select(publish) error = %v", err) - } - _, err = executeStages(context.Background(), cfg, []stage.Stage{publishStageImpl}, RunOptions{ - Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, fail: func(key string) bool { - return strings.HasSuffix(key, "/session-manifest.json") - }}}, - }) - if err == nil || !strings.Contains(err.Error(), "immutable object") { - t.Fatalf("executeStages() error = %v, want committed-manifest failure", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) -} - -func TestPostPublishCleanupNotRunWhenCommitPointerUploadFails(t *testing.T) { - cfg, seed, _ := publishStageCleanupFixture(t) - cfg.Pipeline.Spool.DeleteAudioAfterPublish = true - cfg.Pipeline.Workspace.CleanupAfterPublish = true - failKey := artifacts.S3CurrentCommitPointerKey(seed.sessionPrefix) - - publishStageImpl, err := stage.Select("publish") - if err != nil { - t.Fatalf("Select(publish) error = %v", err) - } - _, err = executeStages(context.Background(), cfg, []stage.Stage{publishStageImpl}, RunOptions{ - Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}}, - }) - if err == nil || !strings.Contains(err.Error(), "current commit pointer") { - t.Fatalf("executeStages() error = %v, want current-commit-pointer failure", err) - } - - assertExists(t, seed.spoolAudioDir) - assertExists(t, seed.runWorkDir) -} - -type cleanupSeed struct { - runID string - runWorkDir string - otherRunDir string - spoolAudioDir string - localSourceAudio string - previousCachePath string - sessionPrefix string -} - -type cleanupFailingManifestStore struct { - delegate manifest.Store - fail func(*manifest.Manifest) error -} - -func (s *cleanupFailingManifestStore) Create(ctx context.Context, sessionID string) (*manifest.Manifest, error) { - return s.delegate.Create(ctx, sessionID) -} - -func (s *cleanupFailingManifestStore) Load(ctx context.Context, path string) (*manifest.Manifest, error) { - return s.delegate.Load(ctx, path) -} - -func (s *cleanupFailingManifestStore) Save(ctx context.Context, path string, m *manifest.Manifest) error { - if s.fail != nil { - if err := s.fail(m); err != nil { - return err - } - } - return s.delegate.Save(ctx, path, m) -} - -func cleanupFixtureConfig(t *testing.T) (*config.Config, cleanupSeed) { - t.Helper() - - cfg := testConfig(t) - cfg.Pipeline.Publish = &config.PublishConfig{Enabled: boolPtr(true), UploadRun: boolPtr(true)} - cfg.Pipeline.Spool.Root = filepath.Join(t.TempDir(), "spool") - - runID := "20260516T010203Z-1a2b3c4d" - runWorkDir := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID) - otherRunDir := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, "20260516T010204Z-5e6f7a8b") - spoolAudioDir := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID) - previousCachePath := mustPreviousArtifactPathForCampaign(t, - cfg.Pipeline.Workspace.Root, - cfg.Session.Campaign, - cfg.Session.SessionID, - "session_recap.md", - ) - - mustWriteFile(t, filepath.Join(runWorkDir, "logs", "stage.log"), "log\n") - mustWriteFile(t, filepath.Join(otherRunDir, "logs", "stage.log"), "other\n") - mustWriteFile(t, filepath.Join(spoolAudioDir, "speaker.flac"), "flac\n") - mustWriteFile(t, previousCachePath, "# previous recap\n") - - localSourceAudio := filepath.Join(filepath.Dir(cfg.SessionPath), "audio", "alice.flac") - mustWriteFile(t, localSourceAudio, "source\n") - - seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) - seed.Campaign = cfg.Session.Campaign - seed.RunID = runID - seed.LocalWorkDir = runWorkDir - seed.LocalSpoolDir = spoolAudioDir - seed.S3Bucket = "my-dnd-archive" - seed.S3SessionPrefix = "dnd/campaigns/sample-campaign/sessions/2026-05-03/" - seed.S3RunPrefix = seed.S3SessionPrefix + "runs/" + runID + "/" - - store := &manifest.LocalStore{} - if err := os.MkdirAll(filepath.Dir(manifestPathFor(cfg)), 0o755); err != nil { - t.Fatalf("MkdirAll() error = %v", err) - } - if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { - t.Fatalf("seed manifest save error = %v", err) - } - - return cfg, cleanupSeed{ - runID: runID, - runWorkDir: runWorkDir, - otherRunDir: otherRunDir, - spoolAudioDir: spoolAudioDir, - localSourceAudio: localSourceAudio, - previousCachePath: previousCachePath, - sessionPrefix: seed.S3SessionPrefix, - } -} - -func publishStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, string) { - t.Helper() - - cfg, seed := cleanupFixtureConfig(t) - runID := "20260516T010203Z-1a2b3c4d" - cfg.Pipeline.Storage.S3 = &config.StorageS3Config{ - Bucket: "my-dnd-archive", - RootPrefix: "dnd", - } - cfg.Pipeline.Publish = &config.PublishConfig{ - Enabled: boolPtr(true), - UploadRun: boolPtr(true), - Outputs: []config.PublishOutputRule{ - {Source: "narratio.transcript.final_trimmed", Dest: "transcripts/final.trimmed.json", Required: boolPtr(true)}, - {Source: "narratio.artifact.session_recap", Dest: "artifacts/session_recap.md", Required: boolPtr(true)}, - }, - } - cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ - Artifacts: map[string]config.ScriptoriumArtifactConfig{ - "session_recap": { - OutputPath: "artifacts/session_recap.md", - }, - }, - } - writePublishFixtureRunFiles( - t, - seed.runWorkDir, - artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID), - ) - - store := &manifest.LocalStore{} - seedManifest, err := store.Load(context.Background(), manifestPathFor(cfg)) - if err != nil { - t.Fatalf("Load() error = %v", err) - } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze"} { - seedManifest.MarkStageSucceeded(name, time.Now().UTC(), nil) - } - setCurrentBuiltInEvidence(t, seedManifest, "trim", artifacts.TranscriptOutputKindFinalTrimmed, artifacts.ArtifactTranscriptFinalTrimmed, filepath.Join(artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID), "transcripts", "final.trimmed.json")) - setAppAnalyzeEvidence(seedManifest, "session_recap", "artifacts/session_recap.md", []byte("# recap\n")) - seedManifest.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", cfg.Session.Campaign, cfg.Session.SessionID) - seedManifest.S3RunPrefix = artifacts.S3RunPrefix(seedManifest.S3SessionPrefix, runID) - if err := store.Save(context.Background(), manifestPathFor(cfg), seedManifest); err != nil { - t.Fatalf("Save() error = %v", err) - } - seed.sessionPrefix = seedManifest.S3SessionPrefix - return cfg, seed, runID -} - -func writePublishFixtureRunFiles(t *testing.T, runWorkDir, sessionRoot string) { - t.Helper() - mustWriteFile(t, filepath.Join(runWorkDir, "prepare", "inputs", "session.yml"), "session_id: 2026-05-03\n") - mustWriteFile(t, filepath.Join(runWorkDir, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n") - mustWriteFile(t, filepath.Join(runWorkDir, "trim", "outputs", "transcripts", "final.trimmed.json"), "{\"segments\":[]}\n") - mustWriteFile(t, filepath.Join(runWorkDir, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n") - mustWriteFile(t, filepath.Join(runWorkDir, "polish", "reports", "audita.report.json"), "{}\n") - mustWriteFile(t, filepath.Join(runWorkDir, "merge", "config", "seriatim.generated.yml"), "key: value\n") - mustWriteFile(t, filepath.Join(runWorkDir, "logs", "audita.stderr.log"), "stderr\n") - mustWriteFile(t, filepath.Join(sessionRoot, "transcripts", "final.trimmed.json"), "{\"segments\":[]}\n") - mustWriteFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n") -} - -type failKeyStore struct { - delegate *storage.FakeBackend - failKey string - fail func(string) bool -} - -func (s *failKeyStore) fails(key string) bool { - return strings.TrimSpace(key) == strings.TrimSpace(s.failKey) || (s.fail != nil && s.fail(key)) -} - -func (s *failKeyStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) { - return s.delegate.List(ctx, prefix) -} - -func (s *failKeyStore) Read(ctx context.Context, key string) (storage.ObjectInfo, io.ReadCloser, error) { - return s.delegate.Read(ctx, key) -} - -func (s *failKeyStore) Download(ctx context.Context, key, localPath string) error { - return s.delegate.Download(ctx, key, localPath) -} - -func (s *failKeyStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) { - if s.fails(key) { - return storage.ObjectInfo{}, errors.New("forced upload failure") - } - return s.delegate.Upload(ctx, localPath, key, opts) -} - -func (s *failKeyStore) UploadReader(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) { - if s.fails(key) { - return storage.ObjectInfo{}, errors.New("forced upload failure") - } - return s.delegate.UploadReader(ctx, source, key, opts) -} - -func (s *failKeyStore) UploadConditional(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions, condition storage.WriteCondition) (storage.ObjectInfo, error) { - if s.fails(key) { - return storage.ObjectInfo{}, errors.New("forced upload failure") - } - return s.delegate.UploadConditional(ctx, source, key, opts, condition) -} - -func (s *failKeyStore) Delete(ctx context.Context, key string) error { - return s.delegate.Delete(ctx, key) -} - -func (s *failKeyStore) Exists(ctx context.Context, key string) (bool, error) { - return s.delegate.Exists(ctx, key) -} - -func assertExists(t *testing.T, path string) { - t.Helper() - if _, err := os.Stat(path); err != nil { - t.Fatalf("expected path to exist %q: %v", path, err) - } -} - -func assertMissing(t *testing.T, path string) { - t.Helper() - if _, err := os.Stat(path); !os.IsNotExist(err) { - t.Fatalf("expected path to be removed %q, stat err=%v", path, err) - } -} - -func assertCleanupPending(t *testing.T, cfg *config.Config) { - t.Helper() - m, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) - if err != nil { - t.Fatalf("Load() error = %v", err) - } - if m.PostPublishCleanup == nil { - t.Fatal("expected a persisted cleanup obligation") - } - for _, target := range m.PostPublishCleanup.Targets { - if !target.Completed { - return - } - } - t.Fatal("expected at least one cleanup target to remain incomplete") -} - -func assertCleanupComplete(t *testing.T, cfg *config.Config) { - t.Helper() - m, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) - if err != nil { - t.Fatalf("Load() error = %v", err) - } - if m.PostPublishCleanup == nil { - t.Fatal("expected a persisted cleanup obligation") - } - for _, target := range m.PostPublishCleanup.Targets { - if !target.Completed { - t.Fatalf("cleanup target remains incomplete: %#v", target) - } - } -} diff --git a/internal/app/run_control.go b/internal/app/run_control.go index 000f004..76faa4b 100644 --- a/internal/app/run_control.go +++ b/internal/app/run_control.go @@ -191,7 +191,8 @@ var canonicalInvalidationEdges = map[string][]string{ "extract": {"analyze"}, "analyze": {"publish"}, "publish": {"notify"}, - "notify": {}, + "notify": {"cleanup"}, + "cleanup": {}, } func newInvalidationRelation(registry []stage.Stage, direct map[string][]string) (*invalidationRelation, error) { diff --git a/internal/app/run_control_test.go b/internal/app/run_control_test.go index b4b313a..7b74c0f 100644 --- a/internal/app/run_control_test.go +++ b/internal/app/run_control_test.go @@ -223,17 +223,18 @@ func TestInvalidationDependents(t *testing.T) { stage string want []string }{ - {"prepare", []string{"transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, - {"transcribe", []string{"merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, - {"merge", []string{"polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, - {"polish", []string{"normalize", "trim", "render", "extract", "analyze", "publish", "notify"}}, - {"normalize", []string{"trim", "render", "extract", "analyze", "publish", "notify"}}, - {"trim", []string{"render", "extract", "analyze", "publish", "notify"}}, - {"render", []string{"analyze", "publish", "notify"}}, - {"extract", []string{"analyze", "publish", "notify"}}, - {"analyze", []string{"publish", "notify"}}, - {"publish", []string{"notify"}}, - {"notify", []string{}}, + {"prepare", []string{"transcribe", "merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}}, + {"transcribe", []string{"merge", "polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}}, + {"merge", []string{"polish", "normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}}, + {"polish", []string{"normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}}, + {"normalize", []string{"trim", "render", "extract", "analyze", "publish", "notify", "cleanup"}}, + {"trim", []string{"render", "extract", "analyze", "publish", "notify", "cleanup"}}, + {"render", []string{"analyze", "publish", "notify", "cleanup"}}, + {"extract", []string{"analyze", "publish", "notify", "cleanup"}}, + {"analyze", []string{"publish", "notify", "cleanup"}}, + {"publish", []string{"notify", "cleanup"}}, + {"notify", []string{"cleanup"}}, + {"cleanup", []string{}}, } for _, test := range tests { t.Run(test.stage, func(t *testing.T) { @@ -290,7 +291,7 @@ func TestInvalidateDependentSucceededStagesWithReason(t *testing.T) { if err != nil { t.Fatalf("invalidateDependentSucceededStagesWithReason() error = %v", err) } - want := []string{"normalize", "trim", "render", "extract", "publish", "notify"} + want := []string{"normalize", "trim", "render", "extract", "publish", "notify", "cleanup"} if !reflect.DeepEqual(got, want) { t.Fatalf("invalidateDependentSucceededStagesWithReason() = %#v, want %#v", got, want) } @@ -316,7 +317,7 @@ func TestRenderAndExtractInvalidationAreIndependent(t *testing.T) { if err != nil { t.Fatalf("invalidate dependents: %v", err) } - want := []string{"analyze", "publish", "notify"} + want := []string{"analyze", "publish", "notify", "cleanup"} if !reflect.DeepEqual(got, want) { t.Fatalf("invalidated = %#v, want %#v", got, want) } diff --git a/internal/app/run_stage_test.go b/internal/app/run_stage_test.go index 721a20b..e0107df 100644 --- a/internal/app/run_stage_test.go +++ b/internal/app/run_stage_test.go @@ -38,8 +38,8 @@ func TestRunContinuesAfterCompletedStages(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=9 skipped=3") { - t.Fatalf("output = %q, want executed=9 skipped=3", out.String()) + if !strings.Contains(out.String(), "executed=10 skipped=4") { + t.Fatalf("output = %q, want executed=10 skipped=4", out.String()) } loaded, err := store.Load(context.Background(), manifestPath) @@ -76,7 +76,7 @@ func TestRunNoRemainingStagesRecordsSkippedStages(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=1 skipped=11") { + if !strings.Contains(out.String(), "executed=2 skipped=12") { t.Fatalf("output = %q, want disabled extraction to self-skip", out.String()) } loaded, err := store.Load(context.Background(), manifestPath) @@ -100,7 +100,7 @@ func TestRunForceRerunsSucceeded(t *testing.T) { store := &manifest.LocalStore{} m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify", "cleanup"} { m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) } if err := store.Save(context.Background(), manifestPath, m); err != nil { @@ -112,7 +112,7 @@ func TestRunForceRerunsSucceeded(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=11 skipped=1") { + if !strings.Contains(out.String(), "executed=12 skipped=2") { t.Fatalf("output = %q, want forced full rerun", out.String()) } } @@ -222,7 +222,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) if err != nil { t.Fatalf("load prepared manifest: %v", err) } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify", "cleanup"} { seed.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) } setCurrentBuiltInEvidence(t, seed, "merge", artifacts.TranscriptOutputKindBase, artifacts.ArtifactTranscriptBase, basePath) @@ -244,7 +244,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) if err != nil { t.Fatalf("load manifest after force: %v", err) } - for _, name := range []string{"normalize", "trim", "render", "extract", "analyze", "publish", "notify"} { + for _, name := range []string{"normalize", "trim", "render", "extract", "analyze", "publish", "notify", "cleanup"} { if afterForce.Stages[name] == nil || afterForce.Stages[name].Status != manifest.StatusStale { t.Fatalf("stage %q = %#v, want stale", name, afterForce.Stages[name]) } @@ -255,7 +255,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=7 skipped=5") { + if !strings.Contains(out.String(), "executed=8 skipped=6") { t.Fatalf("output = %q, want run to execute stale downstream stages", out.String()) } } diff --git a/internal/app/runner_dependencies.go b/internal/app/runner_dependencies.go index 85545be..ac80ad6 100644 --- a/internal/app/runner_dependencies.go +++ b/internal/app/runner_dependencies.go @@ -15,15 +15,14 @@ import ( ) type invocationDependencies struct { - now func() time.Time - newRunID func() (string, error) - newWhisperX func(*config.Config) (whisperx.Client, error) - newSeriatim func(*config.Config) (seriatim.Runner, error) - newAudita func(*config.Config) (audita.Runner, error) - newNotarius func() notarius.Runner - newScriptorium func() scriptorium.Runner - newObjectStore func(context.Context, *config.Config) (storage.ObjectStore, error) - removeScopedDir func(root, target, policy string) error + now func() time.Time + newRunID func() (string, error) + newWhisperX func(*config.Config) (whisperx.Client, error) + newSeriatim func(*config.Config) (seriatim.Runner, error) + newAudita func(*config.Config) (audita.Runner, error) + newNotarius func() notarius.Runner + newScriptorium func() scriptorium.Runner + newObjectStore func(context.Context, *config.Config) (storage.ObjectStore, error) } func (dependencies invocationDependencies) normalized() invocationDependencies { @@ -53,8 +52,5 @@ func (dependencies invocationDependencies) normalized() invocationDependencies { return initializeObjectStoreBackend(ctx, cfg, storage.NewObjectStoreFromConfig) } } - if dependencies.removeScopedDir == nil { - dependencies.removeScopedDir = removeRunScopedDir - } return dependencies } diff --git a/internal/app/runner_finalization.go b/internal/app/runner_finalization.go index 4689eb4..41e8216 100644 --- a/internal/app/runner_finalization.go +++ b/internal/app/runner_finalization.go @@ -1,35 +1,6 @@ package app -import ( - "context" - "errors" - "fmt" - "time" - - "gitea.maximumdirect.net/eric/narratio/internal/manifest" -) - func finalizeInvocation(runtime *invocationRuntime) (*RunSummary, error) { - if containsStage(runtime.executed, "publish") { - if err := runPostPublishCleanup( - runtime.ctx, - runtime.env, - runtime.manifestPath, - runtime.manifest, - runtime.executed, - runtime.dependencies.removeScopedDir, - ); err != nil { - return nil, persistPostPublishCleanupFailure( - runtime.ctx, - runtime.manifestStore, - runtime.manifestPath, - runtime.manifest, - fmt.Errorf("post-publish cleanup incomplete: %w", err), - runtime.dependencies.now(), - ) - } - } - return &RunSummary{ SessionID: runtime.identity.SessionID, RunID: runtime.identity.RunID, @@ -39,24 +10,3 @@ func finalizeInvocation(runtime *invocationRuntime) (*RunSummary, error) { Skipped: runtime.skipped, }, nil } - -func persistPostPublishCleanupFailure( - ctx context.Context, - sessionStore manifest.Store, - sessionManifestPath string, - sessionManifest *manifest.Manifest, - operationErr error, - at time.Time, -) error { - if operationErr == nil { - return nil - } - if sessionManifest != nil { - sessionManifest.RecordFailure(at, operationErr.Error()) - } - // Cleanup may have deliberately removed the selected run directory. The - // committed remote snapshot remains authoritative, while the session - // manifest records the outstanding local cleanup for the next invocation. - sessionErr := sessionStore.Save(ctx, sessionManifestPath, sessionManifest) - return errors.Join(operationErr, wrapTerminalPersistenceError("save terminal session manifest", sessionErr)) -} diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 3c92964..f75c922 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -7,6 +7,7 @@ import ( "fmt" "os" "path/filepath" + "reflect" "strings" "testing" "time" @@ -561,7 +562,7 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { if err != nil { t.Fatalf("executeStages() error = %v", err) } - if len(summary.StageNames) != 11 || len(summary.Executed) != 11 || len(summary.Skipped) != 1 || summary.Skipped[0] != "extract" { + if len(summary.StageNames) != 12 || len(summary.Executed) != 12 || !reflect.DeepEqual(summary.Skipped, []string{"extract", "cleanup"}) { t.Fatalf("summary = %#v, want full plan with disabled extraction self-skip", summary) } @@ -571,14 +572,18 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { t.Fatalf("Load manifest error = %v", err) } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify", "cleanup"} { sr := m.Stages[name] if sr == nil { t.Fatalf("missing stage record %q", name) } - if name == "extract" { - if sr.Status != manifest.StatusSkipped || sr.Error == nil || sr.Error.Message != "notarius_disabled" { - t.Fatalf("extract stage = %#v, want disabled skip", sr) + if name == "extract" || name == "cleanup" { + wantReason := "notarius_disabled" + if name == "cleanup" { + wantReason = "cleanup policies are disabled" + } + if sr.Status != manifest.StatusSkipped || sr.Error == nil || sr.Error.Message != wantReason { + t.Fatalf("%s stage = %#v, want disabled skip", name, sr) } continue } diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index d033c62..6f2e5fe 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -71,6 +71,7 @@ func All() []Stage { analyzeStage{}, publishStage{}, placeholderStage{name: "notify"}, + cleanupStage{}, } }