eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 27 Wiki Activity

Integrate cleanup into pipeline execution

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-05 15:27:08 +00:00
parent 8d9eb5cc40
commit adc60aa588
25 changed files with 405 additions and 1013 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
docs/cli.md
View File

@@ -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 <session_id> --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`

10
docs/internal/overview.md
View File

@@ -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.

54
docs/internal/stage-cleanup.md Normal file
View File

@@ -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 <session_id>` 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`

11
docs/internal/stage-publish.md
View File

@@ -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`

22
docs/internal/workspace.md
View File

@@ -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`

28
docs/operations.md
View File

@@ -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

18
docs/policy/architecture.md
View File

@@ -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

2
docs/roadmap/implementation.md
View File

@@ -193,7 +193,7 @@ filesystem and manifest contracts.
## Stage 3 — Integrate Cleanup Into Canonical Orchestration
**Status:** Pending
**Status:** Completed
### Goal

2
internal/app/analyze_artifacts_commands_test.go
View File

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

6
internal/app/assembled_workflow_test.go
View File

@@ -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])
}

201
internal/app/cleanup_orchestration_test.go Normal file
View File

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

10
internal/app/cleanup_targets.go
View File

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

4
internal/app/commands_test.go
View File

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

2
internal/app/config_commands_test.go
View File

@@ -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:

8
internal/app/plan_test.go
View File

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

27
internal/app/planner_test.go
View File

@@ -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) {

204
internal/app/post_publish_cleanup.go
View File

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

666
internal/app/post_publish_cleanup_test.go
View File

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

3
internal/app/run_control.go
View File

@@ -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) {

27
internal/app/run_control_test.go
View File

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

16
internal/app/run_stage_test.go
View File

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

20
internal/app/runner_dependencies.go
View File

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

50
internal/app/runner_finalization.go
View File

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

15
internal/app/runner_test.go
View File

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

1
internal/stage/placeholders.go
View File

@@ -71,6 +71,7 @@ func All() []Stage {
analyzeStage{},
publishStage{},
placeholderStage{name: "notify"},
cleanupStage{},
}
}