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

Add clean command

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-21 22:48:18 -05:00
parent b817a5b772
commit 2937696024
10 changed files with 748 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
docs/cli.md
View File

@@ -24,6 +24,7 @@ Implemented commands:
- `session init`: create local or remote `session.yml`.
- `artifacts list`: list effective artifact source IDs.
- `locks`: list, add, and remove archive promotion locks.
- `clean`: remove local workspace/spool state for one session or all local sessions.
Unknown commands print usage and exit non-zero.
@@ -94,6 +95,17 @@ Valid stage names:
- `--force`: overwrite local conflicting files with remote archive files.
- `--include-audio`: include durable archived `audio/**` files in restore scope.
### `clean`
- `--session-id <value>`: required for session cleanup unless `--all` is set.
- `--config <path>`
- `--campaign <path>`
- `--session <path>`
- `--previous-session-id <value>`
- `--all`: clean all local session work/spool state using pipeline config only.
- `--dry-run`: print cleanup targets without deleting.
- `--clear-cache`: also remove matching S3 audio cache entries.
### `status`
- `--manifest <path>`: inspect one manifest file.
@@ -378,6 +390,38 @@ Common failure cases:
When `--include-audio` is set, S3 audio files are restored through the shared audio cache. Cache hits avoid re-downloading large audio objects.
### `clean`
Purpose:
- Remove local Narratio work/spool state for testing, reruns, or recovery from corrupted local files.
- Preserve durable S3 audio cache state unless `--clear-cache` is passed.
Syntax:
```bash
narratio clean --session-id <id> [--config <pipeline.yml>] [--campaign <campaign.yml>] [--session <session.yml>] [--previous-session-id <id>] [--dry-run] [--clear-cache]
narratio clean --all [--config <pipeline.yml>] [--dry-run] [--clear-cache]
```
Session cleanup deletes:
- `{workspace.root}/work/{campaign}/{session_id}`
- `{spool.root}/{campaign}/{session_id}`
All-session cleanup deletes:
- `{workspace.root}/work`
- the contents of `{spool.root}`, while preserving the spool root directory itself.
Cache behavior:
- cache is preserved by default.
- `--clear-cache` in session mode removes cached S3 audio files for the resolved session.
- `--all --clear-cache` removes the configured Narratio S3 audio cache namespace for the configured bucket/root prefix.
- `--clear-cache` does not delete arbitrary files under `pipeline.cache.root`.
Common failure cases:
- missing `--session-id` when `--all` is not set.
- combining `--all` with `--campaign`, `--session`, `--session-id`, or `--previous-session-id`.
- unsafe cleanup target, such as a symlink, a non-directory session target, a configured root directory, or a path outside the configured root.
## Common Workflows
Default-discovery run:
@@ -423,6 +467,20 @@ Rehydrate canonical previous-session inputs after artifact-input changes:
narratio run-stage --session-id 2026-04-04 --force prepare
```
Reset local state before testing restore:
```bash
narratio clean --session-id 2026-04-04 --dry-run
narratio clean --session-id 2026-04-04
narratio restore --session-id 2026-04-04 --include-audio
```
Clean all local sessions while keeping cached S3 audio:
```bash
narratio clean --all
```
## Diagnostic / Recovery Commands
Inspect stage status:

2
docs/config.md
View File

@@ -245,6 +245,7 @@ Operational notes:
- operator-created mutable locks are stored at `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/locks.yml` and are merged with static locks.
- Narratio does not auto-promote all generated analyze artifacts.
- `restore` reads the same config/campaign/session inputs and restore scope is bounded by committed archive current state.
- `clean` removes workspace/spool state by default and preserves `pipeline.cache.root` unless `--clear-cache` is passed.
## 6. Full pipeline reference
@@ -408,6 +409,7 @@ Restore-related implications:
- restore remote identity requires archive S3 identity to resolve (`pipeline.storage.s3.bucket` and session prefix derivation inputs).
- restore scope considers committed current state and durable paths (`manifest.json`, `transcripts/**`, `artifacts/**`, `previous/**`, optional `audio/**`).
- S3 audio downloads use `pipeline.spool.root` for active downloads and `pipeline.cache.root` for reusable cached audio when `pipeline.cache.s3_audio` is true.
- `pipeline.cache.root` is durable local cache state. It is not workspace state and is preserved by default by `narratio clean`.
## 7. Full campaign reference

8
docs/internal/workspace.md
View File

@@ -46,7 +46,10 @@ None directly in this subsystem. Stages may use object storage adapters and then
- During each run, stage outputs are often written run-local first (`runs/{run_id}/{stage}/outputs/...`) and promoted to canonical session paths after stage success.
- `manifest.Artifacts` entries record `ProducerRunID` for durable outputs.
- For S3 audio sessions, `prepare` records work/cache paths, S3 provenance, and spool path when the invocation downloaded the object.
- Durable cache state under `pipeline.cache.root` is not workspace state and is not part of session cleanup semantics.
- Durable cache state under `pipeline.cache.root` is not workspace state and is preserved by default by `narratio clean`.
- `narratio clean --session-id <id>` removes the session work root and session spool root.
- `narratio clean --all` removes all local session work under `workspace.root/work` and spool children under `spool.root`.
- `narratio clean --clear-cache` is the explicit opt-in for deleting matching S3 audio cache entries.
## Skip and Resume Behavior
- Skip/resume decisions are made in `internal/app` (`run_control.go`, `resume.go`) using stage status in the session manifest.
@@ -70,4 +73,5 @@ None directly in this subsystem. Stages may use object storage adapters and then
- Run roots are always nested: `runs/{run_id}` under the session root.
- Run-local output promotion must end in canonical session paths.
- `previous/**` is session-durable state and must not be treated as run-local output scratch state.
- Cleanup only targets run-scoped directories and must never delete configured root directories.
- Automatic post-archive cleanup only targets run-scoped directories and must never delete configured root directories.
- Manual `clean` may delete session-scoped directories or the `workspace.root/work` directory, but it must preserve configured root directories and reject unsafe targets.

22
docs/operations.md
View File

@@ -68,6 +68,16 @@ Restore default scope:
- includes `audio/**` only with `--include-audio`
- excludes `runs/**`, `logs/**`, `reports/**`, `config/**`, `inputs/**`, and `current/**` (except remote `current/manifest.json` as source)
Reset local state before restore testing:
```bash
narratio clean --session-id 2026-04-04 --dry-run
narratio clean --session-id 2026-04-04
narratio restore --session-id 2026-04-04 --include-audio
```
`clean` removes the local session work directory and session spool directory. It preserves the durable S3 audio cache by default, so repeated restore or forced prepare tests do not re-download large audio files.
## Local filesystem layout and state artifacts
Session root:
@@ -183,13 +193,19 @@ Safe rerun pattern:
## Cleanup behavior
Cleanup is considered only when archive stage executed and succeeded.
Automatic post-archive cleanup is considered only when archive stage executed and succeeded.
Cleanup toggles:
Automatic cleanup toggles:
- `pipeline.spool.delete_audio_after_archive=true` deletes run-scoped spool audio.
- `pipeline.workspace.cleanup_after_archive=true` deletes run-scoped local run directory.
The S3 audio cache under `pipeline.cache.root` is durable input cache state, not workspace or spool state. Cleanup does not delete it.
Manual cleanup:
- `narratio clean --session-id <id>` deletes `{workspace.root}/work/{campaign}/{session_id}` and `{spool.root}/{campaign}/{session_id}`.
- `narratio clean --all` deletes all local session work under `{workspace.root}/work` and all spool children under `{spool.root}`.
- `--dry-run` prints targets without deleting.
- `--clear-cache` also removes matching S3 audio cache files. Without it, cache is preserved.
The S3 audio cache under `pipeline.cache.root` is durable input cache state, not workspace or spool state. Automatic cleanup and default manual cleanup do not delete it.
Cleanup eligibility gates:
- archive enabled

336
internal/app/clean.go Normal file
View File

@@ -0,0 +1,336 @@
package app
import (
"context"
"flag"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
)
// Clean removes local workspace/spool state while preserving durable cache
// state unless cache cleanup is explicitly requested.
func Clean(ctx context.Context, args []string, out io.Writer) error {
fs := flag.NewFlagSet("clean", flag.ContinueOnError)
fs.SetOutput(io.Discard)
var flags commonConfigFlags
var all bool
var dryRun bool
var clearCache bool
addCommonConfigFlags(fs, &flags)
fs.BoolVar(&all, "all", false, "clean all local session work/spool state")
fs.BoolVar(&dryRun, "dry-run", false, "print cleanup targets without deleting")
fs.BoolVar(&clearCache, "clear-cache", false, "also clear durable S3 audio cache entries")
if err := fs.Parse(args); err != nil {
return fmt.Errorf("clean: invalid flags: %w", err)
}
if fs.NArg() != 0 {
return fmt.Errorf("clean: unexpected positional arguments")
}
if all {
return cleanAllLocal(flags, dryRun, clearCache, out)
}
return cleanSession(ctx, flags, dryRun, clearCache, out)
}
func cleanSession(ctx context.Context, flags commonConfigFlags, dryRun, clearCache bool, out io.Writer) error {
if strings.TrimSpace(flags.sessionID) == "" {
return fmt.Errorf("clean: --session-id is required unless --all is set")
}
cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.sessionPath, flags.sessionOptions())
if err != nil {
return fmt.Errorf("clean: %w", err)
}
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
return fmt.Errorf("clean: resolved pipeline and session config are required")
}
campaign := strings.TrimSpace(cfg.Session.Campaign)
sessionID := strings.TrimSpace(cfg.Session.SessionID)
if campaign == "" || sessionID == "" {
return fmt.Errorf("clean: campaign and session_id are required")
}
if dryRun {
fmt.Fprintf(out, "Clean plan for %s/%s\n", campaign, sessionID)
} else {
fmt.Fprintf(out, "Cleaned %s/%s\n", campaign, sessionID)
}
workDir := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, campaign, sessionID)
spoolDir := artifacts.SessionSpoolDir(cfg.Pipeline.Spool.Root, campaign, sessionID)
if err := reportCleanScopedDir(out, cfg.Pipeline.Workspace.Root, workDir, "clean.workspace.session", dryRun); err != nil {
return fmt.Errorf("clean: %w", err)
}
if err := reportCleanScopedDir(out, cfg.Pipeline.Spool.Root, spoolDir, "clean.spool.session", dryRun); err != nil {
return fmt.Errorf("clean: %w", err)
}
if clearCache {
if err := cleanSessionAudioCache(ctx, cfg, dryRun, out); err != nil {
return fmt.Errorf("clean: %w", err)
}
} else {
fmt.Fprintln(out, "Cache: preserved")
}
return nil
}
func cleanAllLocal(flags commonConfigFlags, dryRun, clearCache bool, out io.Writer) error {
if strings.TrimSpace(flags.campaignPath) != "" ||
strings.TrimSpace(flags.sessionPath) != "" ||
strings.TrimSpace(flags.sessionID) != "" ||
strings.TrimSpace(flags.previousSessionID) != "" {
return fmt.Errorf("clean: --all cannot be combined with --campaign, --session, --session-id, or --previous-session-id")
}
resolvedPipelinePath, err := resolvePipelineConfigPath(flags.pipelinePath)
if err != nil {
return fmt.Errorf("clean: %w", err)
}
pipelineCfg, err := config.LoadPipeline(resolvedPipelinePath)
if err != nil {
return fmt.Errorf("clean: %w", err)
}
if dryRun {
fmt.Fprintln(out, "Clean plan for all local sessions")
} else {
fmt.Fprintln(out, "Cleaned all local sessions")
}
workRoot := filepath.Join(pipelineCfg.Workspace.Root, config.PathWorkDirSegment)
if err := reportCleanScopedDir(out, pipelineCfg.Workspace.Root, workRoot, "clean.workspace.all", dryRun); err != nil {
return fmt.Errorf("clean: %w", err)
}
if err := reportCleanRootChildren(out, pipelineCfg.Spool.Root, "clean.spool.all", dryRun); err != nil {
return fmt.Errorf("clean: %w", err)
}
if clearCache {
if err := cleanAllAudioCache(pipelineCfg, dryRun, out); err != nil {
return fmt.Errorf("clean: %w", err)
}
} else {
fmt.Fprintln(out, "Cache: preserved")
}
return nil
}
func reportCleanScopedDir(out io.Writer, root, target, policy string, dryRun bool) error {
dir, err := validateScopedDir(root, target, policy)
if err != nil {
return err
}
if dryRun {
if dir.Exists {
fmt.Fprintf(out, "Would delete: %s\n", dir.TargetAbs)
} else {
fmt.Fprintf(out, "Would skip missing: %s\n", dir.TargetAbs)
}
return nil
}
if !dir.Exists {
fmt.Fprintf(out, "Missing: %s\n", dir.TargetAbs)
return nil
}
if err := os.RemoveAll(dir.TargetAbs); err != nil {
return fmt.Errorf("cleanup policy %s: remove %q: %w", policy, dir.TargetAbs, err)
}
fmt.Fprintf(out, "Deleted: %s\n", dir.TargetAbs)
return nil
}
func reportCleanRootChildren(out io.Writer, root, policy string, dryRun bool) error {
rootAbs, entries, err := cleanableRootChildren(root, policy)
if err != nil {
return err
}
if len(entries) == 0 {
if dryRun {
fmt.Fprintf(out, "Would skip empty: %s\n", rootAbs)
} else {
fmt.Fprintf(out, "Empty: %s\n", rootAbs)
}
return nil
}
for _, entry := range entries {
if dryRun {
fmt.Fprintf(out, "Would delete: %s\n", entry)
continue
}
if err := os.RemoveAll(entry); err != nil {
return fmt.Errorf("cleanup policy %s: remove %q: %w", policy, entry, err)
}
fmt.Fprintf(out, "Deleted: %s\n", entry)
}
return nil
}
func cleanableRootChildren(root, policy string) (string, []string, error) {
cleanRoot := strings.TrimSpace(root)
if cleanRoot == "" {
return "", nil, fmt.Errorf("cleanup policy %s: root path is required", policy)
}
rootAbs, err := filepath.Abs(cleanRoot)
if err != nil {
return "", nil, fmt.Errorf("cleanup policy %s: resolve root %q: %w", policy, cleanRoot, err)
}
info, err := os.Lstat(rootAbs)
if err != nil {
if os.IsNotExist(err) {
return rootAbs, nil, nil
}
return "", nil, fmt.Errorf("cleanup policy %s: stat root %q: %w", policy, rootAbs, err)
}
if info.Mode()&os.ModeSymlink != 0 {
return "", nil, fmt.Errorf("cleanup policy %s: refusing to clean symlink root %q", policy, rootAbs)
}
if !info.IsDir() {
return "", nil, fmt.Errorf("cleanup policy %s: root %q is not a directory", policy, rootAbs)
}
entries, err := os.ReadDir(rootAbs)
if err != nil {
return "", nil, fmt.Errorf("cleanup policy %s: read root %q: %w", policy, rootAbs, err)
}
out := make([]string, 0, len(entries))
for _, entry := range entries {
path := filepath.Join(rootAbs, entry.Name())
info, err := os.Lstat(path)
if err != nil {
return "", nil, fmt.Errorf("cleanup policy %s: stat child %q: %w", policy, path, err)
}
if info.Mode()&os.ModeSymlink != 0 {
return "", nil, fmt.Errorf("cleanup policy %s: refusing to delete symlink path %q", policy, path)
}
out = append(out, path)
}
return rootAbs, out, nil
}
func cleanSessionAudioCache(ctx context.Context, cfg *config.Config, dryRun bool, out io.Writer) error {
if cfg.Session.Inputs.AudioS3 == nil {
fmt.Fprintln(out, "Cache: skipped (session does not use audio_s3)")
return nil
}
if cfg.Pipeline.Storage.S3 == nil || strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) == "" {
return fmt.Errorf("clear cache requires pipeline.storage.s3.bucket")
}
store, err := newCommandObjectStore(ctx, cfg, nil)
if err != nil {
return fmt.Errorf("initialize object store for cache cleanup: %w", err)
}
sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID)
audioPrefix := artifacts.S3AudioPrefix(sessionPrefix, cfg.Session.Inputs.AudioS3.Prefix)
objects, err := store.List(ctx, audioPrefix)
if err != nil {
return fmt.Errorf("list s3 audio objects under %q: %w", audioPrefix, err)
}
count := 0
for _, obj := range objects {
key := strings.TrimSpace(obj.Key)
if key == "" || strings.HasSuffix(key, "/") || !cleanIsFlac(key) {
continue
}
cachePath, err := artifacts.S3AudioCachePath(cfg.Pipeline.Cache.Root, cfg.Pipeline.Storage.S3.Bucket, key)
if err != nil {
return err
}
deleted, err := reportCleanScopedFile(out, cfg.Pipeline.Cache.Root, cachePath, "clean.cache.session", dryRun)
if err != nil {
return err
}
if deleted {
count++
}
}
if count == 0 {
fmt.Fprintf(out, "Cache: no cached S3 audio files found for %s\n", audioPrefix)
}
return nil
}
func cleanAllAudioCache(cfg *config.PipelineConfig, dryRun bool, out io.Writer) error {
if cfg.Storage.S3 == nil || strings.TrimSpace(cfg.Storage.S3.Bucket) == "" {
return fmt.Errorf("clear cache requires pipeline.storage.s3.bucket")
}
namespaceDir, err := artifacts.S3AudioCacheNamespaceDir(cfg.Cache.Root, cfg.Storage.S3.Bucket, cfg.Storage.S3.RootPrefix)
if err != nil {
return err
}
return reportCleanScopedDir(out, cfg.Cache.Root, namespaceDir, "clean.cache.all", dryRun)
}
func reportCleanScopedFile(out io.Writer, root, target, policy string, dryRun bool) (bool, error) {
file, err := validateScopedFile(root, target, policy)
if err != nil {
return false, err
}
if dryRun {
if file.Exists {
fmt.Fprintf(out, "Would delete cache file: %s\n", file.TargetAbs)
return true, nil
}
fmt.Fprintf(out, "Would skip missing cache file: %s\n", file.TargetAbs)
return false, nil
}
if !file.Exists {
fmt.Fprintf(out, "Missing cache file: %s\n", file.TargetAbs)
return false, nil
}
if err := os.Remove(file.TargetAbs); err != nil {
return false, fmt.Errorf("cleanup policy %s: remove %q: %w", policy, file.TargetAbs, err)
}
fmt.Fprintf(out, "Deleted cache file: %s\n", file.TargetAbs)
return true, nil
}
func validateScopedFile(root, target, policy string) (scopedDir, error) {
cleanRoot := strings.TrimSpace(root)
cleanTarget := strings.TrimSpace(target)
if cleanRoot == "" {
return scopedDir{}, fmt.Errorf("cleanup policy %s: root path is required", policy)
}
if cleanTarget == "" {
return scopedDir{}, fmt.Errorf("cleanup policy %s: target path is required", policy)
}
rootAbs, err := filepath.Abs(cleanRoot)
if err != nil {
return scopedDir{}, fmt.Errorf("cleanup policy %s: resolve root %q: %w", policy, cleanRoot, err)
}
targetAbs, err := filepath.Abs(cleanTarget)
if err != nil {
return scopedDir{}, fmt.Errorf("cleanup policy %s: resolve target %q: %w", policy, cleanTarget, err)
}
rel, err := filepath.Rel(rootAbs, targetAbs)
if err != nil {
return scopedDir{}, fmt.Errorf("cleanup policy %s: relative path from %q to %q: %w", policy, rootAbs, targetAbs, err)
}
if rel == "." {
return scopedDir{}, fmt.Errorf("cleanup policy %s: refusing to delete root directory %q", policy, rootAbs)
}
if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return scopedDir{}, fmt.Errorf("cleanup policy %s: refusing to delete path outside root: root=%q target=%q", policy, rootAbs, targetAbs)
}
info, err := os.Lstat(targetAbs)
if err != nil {
if os.IsNotExist(err) {
return scopedDir{RootAbs: rootAbs, TargetAbs: targetAbs, Exists: false}, nil
}
return scopedDir{}, fmt.Errorf("cleanup policy %s: stat target %q: %w", policy, targetAbs, err)
}
if info.Mode()&os.ModeSymlink != 0 {
return scopedDir{}, fmt.Errorf("cleanup policy %s: refusing to delete symlink path %q", policy, targetAbs)
}
if info.IsDir() {
return scopedDir{}, fmt.Errorf("cleanup policy %s: target %q is a directory", policy, targetAbs)
}
return scopedDir{RootAbs: rootAbs, TargetAbs: targetAbs, Exists: true}, nil
}
func cleanIsFlac(path string) bool {
return strings.EqualFold(filepath.Ext(path), ".flac")
}

255
internal/app/clean_test.go Normal file
View File

@@ -0,0 +1,255 @@
package app
import (
"bytes"
"os"
"path/filepath"
"strings"
"testing"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
)
func TestExecuteCleanSessionDeletesWorkAndSpoolButPreservesCache(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
workDir := artifacts.SessionWorkDirForCampaign(workspaceRoot, "sample-campaign", "2026-05-03")
spoolDir := artifacts.SessionSpoolDir(filepath.Join(workspaceRoot, "spool"), "sample-campaign", "2026-05-03")
cachePath, err := artifacts.S3AudioCachePath(filepath.Join(workspaceRoot, "cache"), "test-bucket", "dnd/campaigns/sample-campaign/sessions/2026-05-03/audio/alice.flac")
if err != nil {
t.Fatalf("S3AudioCachePath() error = %v", err)
}
mustWriteTestFile(t, filepath.Join(workDir, "manifest.json"), "{}")
mustWriteTestFile(t, filepath.Join(spoolDir, "run-1", "audio", "alice.flac"), "audio")
mustWriteTestFile(t, cachePath, "cached-audio")
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
cleanAssertMissing(t, workDir)
cleanAssertMissing(t, spoolDir)
cleanAssertExists(t, cachePath)
if !strings.Contains(stdout.String(), "Cache: preserved") {
t.Fatalf("stdout = %q, want cache preserved", stdout.String())
}
}
func TestExecuteCleanSessionDryRunDeletesNothing(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
workDir := artifacts.SessionWorkDirForCampaign(workspaceRoot, "sample-campaign", "2026-05-03")
spoolDir := artifacts.SessionSpoolDir(filepath.Join(workspaceRoot, "spool"), "sample-campaign", "2026-05-03")
mustWriteTestFile(t, filepath.Join(workDir, "manifest.json"), "{}")
mustWriteTestFile(t, filepath.Join(spoolDir, "run-1", "audio", "alice.flac"), "audio")
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--session-id", "2026-05-03", "--dry-run"}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
cleanAssertExists(t, workDir)
cleanAssertExists(t, spoolDir)
if !strings.Contains(stdout.String(), "Would delete:") {
t.Fatalf("stdout = %q, want dry-run delete plan", stdout.String())
}
}
func TestExecuteCleanMissingSessionPathsSucceeds(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--session-id", "2026-05-03"}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
if !strings.Contains(stdout.String(), "Missing:") {
t.Fatalf("stdout = %q, want missing path output", stdout.String())
}
}
func TestExecuteCleanSessionClearCacheRemovesOnlyS3AudioCache(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
if err := os.WriteFile(sessionPath, []byte(`session_id: 2026-05-03
inputs:
audio_s3:
prefix: audio/
`), 0o644); err != nil {
t.Fatalf("write session: %v", err)
}
audioKey := "dnd/campaigns/sample-campaign/sessions/2026-05-03/audio/alice.flac"
fake := &storage.FakeBackend{}
fake.SeedObject(storage.FakeObject{Key: audioKey, Data: []byte("audio")})
var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
cacheRoot := filepath.Join(workspaceRoot, "cache")
cachePath, err := artifacts.S3AudioCachePath(cacheRoot, "test-bucket", audioKey)
if err != nil {
t.Fatalf("S3AudioCachePath() error = %v", err)
}
otherCachePath, err := artifacts.S3AudioCachePath(cacheRoot, "test-bucket", "dnd/campaigns/other/sessions/2026-05-03/audio/bob.flac")
if err != nil {
t.Fatalf("S3AudioCachePath() error = %v", err)
}
mustWriteTestFile(t, cachePath, "cached-audio")
mustWriteTestFile(t, otherCachePath, "other-audio")
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--session-id", "2026-05-03", "--clear-cache"}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
cleanAssertMissing(t, cachePath)
cleanAssertExists(t, otherCachePath)
if storeInitCalls != 1 {
t.Fatalf("object store init calls = %d, want 1", storeInitCalls)
}
}
func TestExecuteCleanLocalAudioClearCacheIsNoop(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--session-id", "2026-05-03", "--clear-cache"}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
if !strings.Contains(stdout.String(), "Cache: skipped (session does not use audio_s3)") {
t.Fatalf("stdout = %q, want local audio cache no-op", stdout.String())
}
}
func TestExecuteCleanAllDeletesWorkAndSpoolContentsButPreservesCache(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, _, _ := writeValidConfigFiles(t, workspaceRoot)
workRoot := filepath.Join(workspaceRoot, "work")
spoolRoot := filepath.Join(workspaceRoot, "spool")
cachePath := filepath.Join(workspaceRoot, "cache", "keep.txt")
mustWriteTestFile(t, filepath.Join(workRoot, "sample-campaign", "2026-05-03", "manifest.json"), "{}")
mustWriteTestFile(t, filepath.Join(spoolRoot, "sample-campaign", "2026-05-03", "run-1", "audio", "alice.flac"), "audio")
mustWriteTestFile(t, cachePath, "cache")
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--all"}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
cleanAssertMissing(t, workRoot)
cleanAssertExists(t, spoolRoot)
cleanAssertMissing(t, filepath.Join(spoolRoot, "sample-campaign"))
cleanAssertExists(t, cachePath)
}
func TestExecuteCleanAllClearCacheRemovesS3AudioNamespaceOnly(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, _, _ := writeValidConfigFiles(t, workspaceRoot)
cacheRoot := filepath.Join(workspaceRoot, "cache")
audioCachePath, err := artifacts.S3AudioCachePath(cacheRoot, "test-bucket", "dnd/campaigns/sample-campaign/sessions/2026-05-03/audio/alice.flac")
if err != nil {
t.Fatalf("S3AudioCachePath() error = %v", err)
}
otherCachePath, err := artifacts.S3AudioCachePath(cacheRoot, "test-bucket", "other-root/campaigns/sample-campaign/sessions/2026-05-03/audio/alice.flac")
if err != nil {
t.Fatalf("S3AudioCachePath() error = %v", err)
}
mustWriteTestFile(t, audioCachePath, "cached-audio")
mustWriteTestFile(t, otherCachePath, "other-cache")
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--all", "--clear-cache"}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
cleanAssertMissing(t, audioCachePath)
cleanAssertExists(t, otherCachePath)
}
func TestExecuteCleanAllRejectsSessionScopedFlags(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean", "--config", pipelinePath, "--campaign", campaignPath, "--all"}, &stdout, &stderr)
if code == 0 {
t.Fatal("exit code = 0, want non-zero")
}
if !strings.Contains(stderr.String(), "--all cannot be combined") {
t.Fatalf("stderr = %q, want --all conflict", stderr.String())
}
}
func TestCleanRequiresSessionID(t *testing.T) {
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clean"}, &stdout, &stderr)
if code == 0 {
t.Fatal("exit code = 0, want non-zero")
}
if !strings.Contains(stderr.String(), "--session-id is required unless --all is set") {
t.Fatalf("stderr = %q, want missing session-id", stderr.String())
}
}
func TestCleanRejectsUnsafeTargets(t *testing.T) {
root := t.TempDir()
outside := t.TempDir()
if err := reportCleanScopedDir(&bytes.Buffer{}, root, filepath.Join(outside, "target"), "test.outside", false); err == nil {
t.Fatal("outside target error = nil, want error")
}
if err := reportCleanScopedDir(&bytes.Buffer{}, root, root, "test.root", false); err == nil {
t.Fatal("root target error = nil, want error")
}
filePath := filepath.Join(root, "file.txt")
mustWriteTestFile(t, filePath, "file")
if err := reportCleanScopedDir(&bytes.Buffer{}, root, filePath, "test.file", false); err == nil {
t.Fatal("file target error = nil, want error")
}
symlinkPath := filepath.Join(root, "link")
if err := os.Symlink(filepath.Join(root, "missing"), symlinkPath); err != nil {
t.Fatalf("Symlink() error = %v", err)
}
if err := reportCleanScopedDir(&bytes.Buffer{}, root, symlinkPath, "test.symlink", false); err == nil {
t.Fatal("symlink target error = nil, want error")
}
}
func TestClearIsNotCommandAlias(t *testing.T) {
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"clear"}, &stdout, &stderr)
if code == 0 {
t.Fatal("exit code = 0, want non-zero")
}
if !strings.Contains(stderr.String(), `unknown command: "clear"`) {
t.Fatalf("stderr = %q, want unknown clear command", stderr.String())
}
}
func cleanAssertExists(t *testing.T, path string) {
t.Helper()
if _, err := os.Stat(path); err != nil {
t.Fatalf("expected %q to exist: %v", path, err)
}
}
func cleanAssertMissing(t *testing.T, path string) {
t.Helper()
if _, err := os.Stat(path); !os.IsNotExist(err) {
t.Fatalf("expected %q to be missing, stat err=%v", path, err)
}
}

4
internal/app/commands.go
View File

@@ -7,7 +7,7 @@ import (
"strings"
)
var supportedCommands = []string{"run", "plan", "status", "resume", "run-stage", "restore", "session", "artifacts", "locks"}
var supportedCommands = []string{"run", "plan", "status", "resume", "run-stage", "restore", "session", "artifacts", "locks", "clean"}
// Execute dispatches CLI commands and returns a process exit code.
func Execute(args []string, stdout, stderr io.Writer) int {
@@ -40,6 +40,8 @@ func Execute(args []string, stdout, stderr io.Writer) int {
err = Artifacts(ctx, cmdArgs, stdout)
case "locks":
err = Locks(ctx, cmdArgs, stdout)
case "clean":
err = Clean(ctx, cmdArgs, stdout)
default:
fmt.Fprintf(stderr, "unknown command: %q\n\n", cmd)
printUsage(stderr)

47
internal/app/post_archive_cleanup.go
View File

@@ -153,53 +153,70 @@ func archiveCleanupEligible(cfg *config.Config, sr *manifest.StageRecord) (bool,
return true, ""
}
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 := os.RemoveAll(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) {
cleanRoot := strings.TrimSpace(root)
cleanTarget := strings.TrimSpace(target)
if cleanRoot == "" {
return fmt.Errorf("cleanup policy %s: root path is required", policy)
return scopedDir{}, fmt.Errorf("cleanup policy %s: root path is required", policy)
}
if cleanTarget == "" {
return fmt.Errorf("cleanup policy %s: target path is required", policy)
return scopedDir{}, fmt.Errorf("cleanup policy %s: target path is required", policy)
}
rootAbs, err := filepath.Abs(cleanRoot)
if err != nil {
return fmt.Errorf("cleanup policy %s: resolve root %q: %w", policy, cleanRoot, err)
return scopedDir{}, fmt.Errorf("cleanup policy %s: resolve root %q: %w", policy, cleanRoot, err)
}
targetAbs, err := filepath.Abs(cleanTarget)
if err != nil {
return fmt.Errorf("cleanup policy %s: resolve target %q: %w", policy, cleanTarget, err)
return scopedDir{}, fmt.Errorf("cleanup policy %s: resolve target %q: %w", policy, cleanTarget, err)
}
rel, err := filepath.Rel(rootAbs, targetAbs)
if err != nil {
return fmt.Errorf("cleanup policy %s: relative path from %q to %q: %w", policy, rootAbs, targetAbs, err)
return scopedDir{}, fmt.Errorf("cleanup policy %s: relative path from %q to %q: %w", policy, rootAbs, targetAbs, err)
}
if rel == "." {
return fmt.Errorf("cleanup policy %s: refusing to delete root directory %q", policy, rootAbs)
return scopedDir{}, fmt.Errorf("cleanup policy %s: refusing to delete root directory %q", policy, rootAbs)
}
if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return fmt.Errorf("cleanup policy %s: refusing to delete path outside root: root=%q target=%q", policy, rootAbs, targetAbs)
return scopedDir{}, fmt.Errorf("cleanup policy %s: refusing to delete path outside root: root=%q target=%q", policy, rootAbs, targetAbs)
}
info, err := os.Lstat(targetAbs)
if err != nil {
if os.IsNotExist(err) {
return nil
return scopedDir{RootAbs: rootAbs, TargetAbs: targetAbs, Exists: false}, nil
}
return fmt.Errorf("cleanup policy %s: stat target %q: %w", policy, targetAbs, err)
return scopedDir{}, fmt.Errorf("cleanup policy %s: stat target %q: %w", policy, targetAbs, err)
}
if info.Mode()&os.ModeSymlink != 0 {
return fmt.Errorf("cleanup policy %s: refusing to delete symlink path %q", policy, targetAbs)
return scopedDir{}, fmt.Errorf("cleanup policy %s: refusing to delete symlink path %q", policy, targetAbs)
}
if !info.IsDir() {
return fmt.Errorf("cleanup policy %s: target %q is not a directory", policy, targetAbs)
return scopedDir{}, fmt.Errorf("cleanup policy %s: target %q is not a directory", policy, targetAbs)
}
if err := os.RemoveAll(targetAbs); err != nil {
return fmt.Errorf("cleanup policy %s: remove %q: %w", policy, targetAbs, err)
}
return nil
return scopedDir{RootAbs: rootAbs, TargetAbs: targetAbs, Exists: true}, nil
}
func asString(v any) string {

18
internal/artifacts/paths.go
View File

@@ -91,9 +91,14 @@ func SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID string) string {
return filepath.Join(spoolRoot, campaign, sessionID, runID, config.PathAudioDirSegment)
}
// SessionSpoolDir returns the campaign/session scoped local spool root.
func SessionSpoolDir(spoolRoot, campaign, sessionID string) string {
return filepath.Join(spoolRoot, campaign, sessionID)
}
// SessionSpoolRestoreAudioDir returns the local spool audio path for restore downloads.
func SessionSpoolRestoreAudioDir(spoolRoot, campaign, sessionID string) string {
return filepath.Join(spoolRoot, campaign, sessionID, "restore", config.PathAudioDirSegment)
return filepath.Join(SessionSpoolDir(spoolRoot, campaign, sessionID), "restore", config.PathAudioDirSegment)
}
// S3AudioCachePath returns the durable local cache path for one S3 audio object.
@@ -120,6 +125,17 @@ func S3AudioCachePath(cacheRoot, bucket, key string) (string, error) {
return filepath.Join(root, "s3", bucket, filepath.FromSlash(cleanKey)), nil
}
// S3AudioCacheNamespaceDir returns the durable local cache namespace for all
// Narratio S3 audio objects under one bucket/root prefix.
func S3AudioCacheNamespaceDir(cacheRoot, bucket, rootPrefix string) (string, error) {
sentinelKey := path.Join(cleanS3PathPart(rootPrefix), config.S3CampaignsSegment, ".narratio-cache-sentinel")
sentinelPath, err := S3AudioCachePath(cacheRoot, bucket, sentinelKey)
if err != nil {
return "", err
}
return filepath.Dir(sentinelPath), nil
}
func cleanCacheS3Key(key string) string {
normalized := strings.ReplaceAll(strings.TrimSpace(key), `\`, "/")
normalized = strings.Trim(normalized, "/")

20
internal/artifacts/paths_model_test.go
View File

@@ -115,6 +115,15 @@ func TestSessionSpoolAudioDir(t *testing.T) {
}
}
func TestSessionSpoolDir(t *testing.T) {
root := "/var/spool/narratio"
got := SessionSpoolDir(root, "forsaken", "2026-04-19")
want := filepath.Join(root, "forsaken", "2026-04-19")
if got != want {
t.Fatalf("SessionSpoolDir() = %q, want %q", got, want)
}
}
func TestSessionSpoolRestoreAudioDir(t *testing.T) {
root := "/var/spool/narratio"
got := SessionSpoolRestoreAudioDir(root, "forsaken", "2026-04-19")
@@ -135,6 +144,17 @@ func TestS3AudioCachePath(t *testing.T) {
}
}
func TestS3AudioCacheNamespaceDir(t *testing.T) {
got, err := S3AudioCacheNamespaceDir("/var/cache/narratio", "my-dnd-archive", "dnd")
if err != nil {
t.Fatalf("S3AudioCacheNamespaceDir() error = %v", err)
}
want := filepath.Join("/var/cache/narratio", "s3", "my-dnd-archive", "dnd", "campaigns")
if got != want {
t.Fatalf("S3AudioCacheNamespaceDir() = %q, want %q", got, want)
}
}
func TestS3AudioCachePathRejectsUnsafeInputs(t *testing.T) {
tests := []struct {
name string