From d5a9ad38f8e323da92931e03323e76d371c63a13 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 16 May 2026 23:09:39 +0000 Subject: [PATCH] Add post-archive cleanup policies --- README.md | 7 + architecture.md | 10 +- docs/archive-storage.md | 6 +- ...narratio-s3-archive-implementation-plan.md | 5 +- docs/s3-audio-input.md | 1 - examples/pipeline.minimal.yml | 1 + internal/app/post_archive_cleanup.go | 208 +++++++++ internal/app/post_archive_cleanup_test.go | 396 ++++++++++++++++++ internal/app/runner.go | 4 + internal/config/config.go | 3 +- internal/config/storage_archive_test.go | 3 + 11 files changed, 639 insertions(+), 5 deletions(-) create mode 100644 internal/app/post_archive_cleanup.go create mode 100644 internal/app/post_archive_cleanup_test.go diff --git a/README.md b/README.md index 5ad9979..1224e71 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,7 @@ Implemented foundations: Current defaults: - `pipeline.storage.s3.root_prefix`: `dnd` +- `pipeline.workspace.cleanup_after_archive`: `false` - `pipeline.spool.root`: `/var/spool/narratio` - `pipeline.spool.delete_audio_after_archive`: `false` - `pipeline.archive.enabled`: `true` @@ -104,6 +105,12 @@ Current boundaries: - archive uploads `current/run_id.txt` last as the effective commit marker - required missing promotions fail archive - optional missing promotions are skipped and recorded +- cleanup remains conservative and opt-in: + - `pipeline.spool.delete_audio_after_archive: true` removes only the run-scoped spool audio directory after successful archive commit + - `pipeline.workspace.cleanup_after_archive: true` removes only the run-scoped local workdir after successful archive commit + - cleanup executes only after all selected stages for the command invocation succeed + - cleanup does not run for failed, incomplete, skipped, or unarchived runs + - local development `audio_dir`/`audio_files` source inputs are never deleted by spool cleanup S3 input details and current boundaries are documented in [docs/s3-audio-input.md](docs/s3-audio-input.md). diff --git a/architecture.md b/architecture.md index 31aa6ec..b091d1e 100644 --- a/architecture.md +++ b/architecture.md @@ -152,7 +152,8 @@ Storage and archive foundations: - `endpoint` - `force_path_style` (default `false`) - `pipeline.spool.root` defaults to `/var/spool/narratio` -- `pipeline.spool.delete_audio_after_archive` defaults to `false` (cleanup behavior not implemented yet) +- `pipeline.workspace.cleanup_after_archive` defaults to `false` +- `pipeline.spool.delete_audio_after_archive` defaults to `false` - `pipeline.archive` is optional and defaults to: - `enabled: true` - `upload_run: true` @@ -230,6 +231,13 @@ Archive publishing behavior (implemented): - `current/run_id.txt` is the effective commit marker - if promotion or current-manifest upload fails, archive returns failure and does not write `current/run_id.txt` - failed/incomplete runs remain local and are not uploaded +- post-archive local cleanup (implemented, opt-in): + - cleanup runs only after archive succeeded and wrote `current/run_id.txt` + - cleanup is executed after all selected stages in the command invocation succeed (for example, a later `notify` failure leaves local files intact) + - `pipeline.spool.delete_audio_after_archive: true` removes only `{spool.root}/{campaign}/{session_id}/{run_id}/audio/` + - `pipeline.workspace.cleanup_after_archive: true` removes only `{workspace.root}/work/{campaign}/{session_id}/{run_id}/` + - cleanup does not run when archive is skipped/disabled/fails or when run upload is disabled + - local development `audio_dir`/`audio_files` inputs are never removed by spool cleanup `pipeline.scriptorium` is optional. Existing pipelines without Scriptorium continue to work. diff --git a/docs/archive-storage.md b/docs/archive-storage.md index 626803f..14829cb 100644 --- a/docs/archive-storage.md +++ b/docs/archive-storage.md @@ -20,11 +20,13 @@ Implemented: - archive uploads configured promoted outputs to session-level keys. - archive uploads `current/manifest.json`. - archive uploads `current/run_id.txt` last as the effective commit marker. +- optional post-archive local cleanup: + - `pipeline.spool.delete_audio_after_archive: true` removes only the run-scoped spool audio directory + - `pipeline.workspace.cleanup_after_archive: true` removes only the run-scoped local workdir - tests use fake storage and do not require live S3. Future work: -- spool audio cleanup/deletion behavior - `notify` stage behavior - stale detection - optional future source-audio upload mode @@ -89,6 +91,7 @@ Archive writes: Writing `current/run_id.txt` last makes it the effective commit marker for published session state. If any required run upload, promotion upload, or current-manifest upload fails, archive returns failure and does not write `current/run_id.txt`. +Cleanup runs only after this commit-marker write has succeeded. ## Audio Upload Policy @@ -99,6 +102,7 @@ Original audio is expected at the session-level audio prefix and is not duplicat - `archive.enabled: false` skips archive cleanly. - `archive.upload_run: false` skips run upload cleanly. +- both skip cases also skip post-archive local cleanup. ## Metadata diff --git a/docs/roadmap/narratio-s3-archive-implementation-plan.md b/docs/roadmap/narratio-s3-archive-implementation-plan.md index 1092a3a..5696a56 100644 --- a/docs/roadmap/narratio-s3-archive-implementation-plan.md +++ b/docs/roadmap/narratio-s3-archive-implementation-plan.md @@ -120,10 +120,13 @@ Implemented in repository: - `current/run_id.txt` is uploaded last as the effective commit marker - current pointer content is `{run_id}` plus trailing newline - if promotion/current manifest upload fails, current pointer is not written +- post-archive local cleanup behavior: + - `pipeline.spool.delete_audio_after_archive` removes run-scoped spool audio only after successful archive commit + - `pipeline.workspace.cleanup_after_archive` removes run-scoped workdir only after successful archive commit + - cleanup is skipped for failed/incomplete/skipped/unarchived runs Not implemented yet: -- spool audio cleanup / deletion behavior - `notify` stage behavior - generic stale detection based on input/config checksums - optional future mode for uploading source audio from local workspace/spool diff --git a/docs/s3-audio-input.md b/docs/s3-audio-input.md index 38770ad..4278de4 100644 --- a/docs/s3-audio-input.md +++ b/docs/s3-audio-input.md @@ -12,7 +12,6 @@ Implemented: Not implemented: -- spool cleanup/deletion behavior - uploads of failed runs ## Required Configuration diff --git a/examples/pipeline.minimal.yml b/examples/pipeline.minimal.yml index 1ffe9d6..0de6b36 100644 --- a/examples/pipeline.minimal.yml +++ b/examples/pipeline.minimal.yml @@ -1,5 +1,6 @@ workspace: root: ./tmp/narratio-workspace + cleanup_after_archive: false storage: backend: local diff --git a/internal/app/post_archive_cleanup.go b/internal/app/post_archive_cleanup.go new file mode 100644 index 0000000..28dc216 --- /dev/null +++ b/internal/app/post_archive_cleanup.go @@ -0,0 +1,208 @@ +package app + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m *manifest.Manifest, executed []string) error { + if env == nil || env.Config == nil || env.Config.Pipeline == nil || m == nil { + return nil + } + + spoolRequested := env.Config.Pipeline.Spool.DeleteAudioAfterArchive + workRequested := env.Config.Pipeline.Workspace.CleanupAfterArchive + if !spoolRequested && !workRequested { + return nil + } + + sr := archiveStageRecordForCleanup(m, executed) + if sr == nil { + return nil + } + if sr.Metadata == nil { + sr.Metadata = map[string]any{} + } + sr.Metadata["spool_cleanup_requested"] = spoolRequested + sr.Metadata["workdir_cleanup_requested"] = workRequested + + eligible, reason := archiveCleanupEligible(env.Config, sr) + if !eligible { + sr.Metadata["cleanup_skipped"] = true + sr.Metadata["cleanup_skipped_reason"] = reason + if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { + return fmt.Errorf("save manifest cleanup skip metadata %q: %w", manifestPath, err) + } + return nil + } + + spoolDir := strings.TrimSpace(m.LocalSpoolDir) + if spoolDir == "" { + spoolDir = artifacts.SessionSpoolAudioDir( + env.Config.Pipeline.Spool.Root, + strings.TrimSpace(env.Config.Session.Campaign), + strings.TrimSpace(env.Config.Session.SessionID), + strings.TrimSpace(m.RunID), + ) + } + workDir := strings.TrimSpace(m.LocalWorkDir) + if workDir == "" { + workDir = artifacts.SessionRunWorkDir( + env.Config.Pipeline.Workspace.Root, + strings.TrimSpace(env.Config.Session.Campaign), + strings.TrimSpace(env.Config.Session.SessionID), + strings.TrimSpace(m.RunID), + ) + } + + if spoolRequested { + if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Spool.Root), spoolDir, "pipeline.spool.delete_audio_after_archive"); err != nil { + sr.Metadata["cleanup_failed"] = true + sr.Metadata["cleanup_failed_policy"] = "pipeline.spool.delete_audio_after_archive" + sr.Metadata["cleanup_failed_path"] = spoolDir + _ = env.ManifestStore.Save(ctx, manifestPath, m) + return err + } + sr.Metadata["spool_cleanup_deleted"] = filepath.Clean(spoolDir) + } + + if !workRequested { + sr.Metadata["cleanup_completed"] = true + sr.Metadata["cleanup_skipped"] = false + if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { + return fmt.Errorf("save manifest cleanup metadata %q: %w", manifestPath, err) + } + return nil + } + + if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Workspace.Root), workDir, "pipeline.workspace.cleanup_after_archive"); err != nil { + sr.Metadata["cleanup_failed"] = true + sr.Metadata["cleanup_failed_policy"] = "pipeline.workspace.cleanup_after_archive" + sr.Metadata["cleanup_failed_path"] = workDir + _ = env.ManifestStore.Save(ctx, manifestPath, m) + return err + } + + sr.Metadata["workdir_cleanup_deleted"] = filepath.Clean(workDir) + sr.Metadata["cleanup_completed"] = true + sr.Metadata["cleanup_skipped"] = false + return nil +} + +func archiveStageRecordForCleanup(m *manifest.Manifest, executed []string) *manifest.StageRecord { + if m == nil { + return nil + } + archiveRan := false + for _, name := range executed { + if name == "archive" { + archiveRan = true + break + } + } + if !archiveRan { + return nil + } + sr := m.Stages["archive"] + if sr == nil || sr.Status != manifest.StatusSucceeded { + return nil + } + return sr +} + +func archiveCleanupEligible(cfg *config.Config, sr *manifest.StageRecord) (bool, string) { + if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Archive == nil { + return false, "archive configuration is missing" + } + enabled := true + if cfg.Pipeline.Archive.Enabled != nil { + enabled = *cfg.Pipeline.Archive.Enabled + } + if !enabled { + return false, "archive.enabled is false" + } + uploadRun := true + if cfg.Pipeline.Archive.UploadRun != nil { + uploadRun = *cfg.Pipeline.Archive.UploadRun + } + if !uploadRun { + return false, "archive.upload_run is false" + } + if sr == nil || sr.Metadata == nil { + return false, "archive metadata is missing" + } + if skipped, _ := sr.Metadata["skipped"].(bool); skipped { + return false, "archive stage was skipped" + } + if uploaded, _ := sr.Metadata["uploaded"].(bool); !uploaded { + return false, "archive did not upload run record" + } + if pointer, _ := sr.Metadata["current_pointer_written"].(bool); !pointer { + return false, "archive did not write current pointer" + } + if strings.TrimSpace(asString(sr.Metadata["current_run_id_key"])) == "" { + return false, "archive current run pointer key is missing" + } + return true, "" +} + +func removeRunScopedDir(root, target, policy string) error { + cleanRoot := strings.TrimSpace(root) + cleanTarget := strings.TrimSpace(target) + if cleanRoot == "" { + return fmt.Errorf("cleanup policy %s: root path is required", policy) + } + if cleanTarget == "" { + return 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) + } + targetAbs, err := filepath.Abs(cleanTarget) + if err != nil { + return 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) + } + if rel == "." { + return 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) + } + + info, err := os.Lstat(targetAbs) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return 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) + } + if !info.IsDir() { + return 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 +} + +func asString(v any) string { + s, _ := v.(string) + return s +} diff --git a/internal/app/post_archive_cleanup_test.go b/internal/app/post_archive_cleanup_test.go new file mode 100644 index 0000000..5aba559 --- /dev/null +++ b/internal/app/post_archive_cleanup_test.go @@ -0,0 +1,396 @@ +package app + +import ( + "context" + "errors" + "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 archiveSuccessStage struct { + metadata map[string]any +} + +func (archiveSuccessStage) Name() string { return "archive" } +func (archiveSuccessStage) Declares() stage.IODecl { return stage.IODecl{} } +func (s archiveSuccessStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + md := map[string]any{ + "stage": "archive", + "uploaded": true, + "current_pointer_written": true, + "current_run_id_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/run_id.txt", + } + 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) Declares() stage.IODecl { return stage.IODecl{} } +func (notifyFailStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + return nil, errors.New("notify failed") +} + +func TestPostArchiveCleanupDisabledKeepsLocalDirs(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = false + cfg.Pipeline.Workspace.CleanupAfterArchive = false + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, 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 TestPostArchiveCleanupSpoolOnly(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = false + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, 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 TestPostArchiveCleanupWorkdirOnly(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = false + cfg.Pipeline.Workspace.CleanupAfterArchive = true + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("executeStages() error = %v", err) + } + + assertExists(t, cfg.Pipeline.Workspace.Root) + assertExists(t, seed.otherRunDir) + assertMissing(t, seed.runWorkDir) + assertExists(t, seed.spoolAudioDir) +} + +func TestPostArchiveCleanupBothPolicies(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, 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) +} + +func TestPostArchiveCleanupNotRunWhenArchiveFails(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + + _, err := executeStages(context.Background(), cfg, []stage.Stage{failingStage{name: "archive", err: errors.New("archive failed")}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) + if err == nil || !strings.Contains(err.Error(), "stage \"archive\" failed") { + t.Fatalf("executeStages() error = %v, want archive failure", err) + } + + assertExists(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) +} + +func TestPostArchiveCleanupNotRunWhenArchiveSkipped(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"skipped": true}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("executeStages() error = %v", err) + } + + assertExists(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) +} + +func TestPostArchiveCleanupNotRunWhenCurrentPointerMissing(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"current_pointer_written": false}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("executeStages() error = %v", err) + } + + assertExists(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) +} + +func TestPostArchiveCleanupNotRunWhenArchiveUploadDisabled(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + cfg.Pipeline.Archive.UploadRun = boolPtr(false) + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { + t.Fatalf("executeStages() error = %v", err) + } + + assertExists(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) +} + +func TestPostArchiveCleanupWaitsUntilAllStagesSucceed(t *testing.T) { + cfg, seed := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + + _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}, 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 TestPostArchiveCleanupFailsOnUnsafePath(t *testing.T) { + cfg, _ := cleanupFixtureConfig(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = false + + manifestPath := manifestPathFor(cfg) + store := &manifest.LocalStore{} + m, err := store.Load(context.Background(), manifestPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + m.LocalSpoolDir = 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) + } + + _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{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 TestPostArchiveCleanupNotRunWhenPromotionIsMissing(t *testing.T) { + cfg, seed, runID := archiveStageCleanupFixture(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + cfg.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ + {From: "artifacts/missing.md", To: "artifacts/missing.md", Required: boolPtr(true)}, + } + + archiveStageImpl, err := stage.Select("archive") + if err != nil { + t.Fatalf("Select(archive) error = %v", err) + } + _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) + if err == nil || !strings.Contains(err.Error(), "required promotion source missing") { + t.Fatalf("executeStages() error = %v, want promotion-missing failure", err) + } + + assertExists(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) + assertExists(t, filepath.Join(seed.runWorkDir, "manifest.json")) + assertExists(t, artifacts.SessionRunWorkDir(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID)) +} + +func TestPostArchiveCleanupNotRunWhenCurrentManifestUploadFails(t *testing.T) { + cfg, seed, _ := archiveStageCleanupFixture(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + failKey := seed.sessionPrefix + "current/manifest.json" + + archiveStageImpl, err := stage.Select("archive") + if err != nil { + t.Fatalf("Select(archive) error = %v", err) + } + _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{ + Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}}, + }) + if err == nil || !strings.Contains(err.Error(), "current manifest") { + t.Fatalf("executeStages() error = %v, want current-manifest failure", err) + } + + assertExists(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) +} + +func TestPostArchiveCleanupNotRunWhenCurrentPointerUploadFails(t *testing.T) { + cfg, seed, _ := archiveStageCleanupFixture(t) + cfg.Pipeline.Spool.DeleteAudioAfterArchive = true + cfg.Pipeline.Workspace.CleanupAfterArchive = true + failKey := seed.sessionPrefix + "current/run_id.txt" + + archiveStageImpl, err := stage.Select("archive") + if err != nil { + t.Fatalf("Select(archive) error = %v", err) + } + _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{ + Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}}, + }) + if err == nil || !strings.Contains(err.Error(), "current run pointer") { + t.Fatalf("executeStages() error = %v, want current-run-pointer failure", err) + } + + assertExists(t, seed.spoolAudioDir) + assertExists(t, seed.runWorkDir) +} + +type cleanupSeed struct { + runWorkDir string + otherRunDir string + spoolAudioDir string + localSourceAudio string + sessionPrefix string +} + +func cleanupFixtureConfig(t *testing.T) (*config.Config, cleanupSeed) { + t.Helper() + + cfg := testConfig(t) + cfg.Pipeline.Archive = &config.ArchiveConfig{Enabled: boolPtr(true), UploadRun: boolPtr(true)} + cfg.Pipeline.Spool.Root = filepath.Join(t.TempDir(), "spool") + + runID := "20260516T010203Z-1a2b3c4d" + runWorkDir := artifacts.SessionRunWorkDir(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID) + otherRunDir := artifacts.SessionRunWorkDir(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) + + mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n") + 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") + + 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{ + runWorkDir: runWorkDir, + otherRunDir: otherRunDir, + spoolAudioDir: spoolAudioDir, + localSourceAudio: localSourceAudio, + sessionPrefix: seed.S3SessionPrefix, + } +} + +func archiveStageCleanupFixture(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.Archive = &config.ArchiveConfig{ + Enabled: boolPtr(true), + UploadRun: boolPtr(true), + PromoteArtifacts: []config.ArchivePromotionRule{ + {From: "transcripts/trimmed.json", To: "transcripts/trimmed.json", Required: boolPtr(true)}, + {From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)}, + }, + } + writeArchiveFixtureRunFiles(t, seed.runWorkDir) + + 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", "analyze"} { + seedManifest.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + 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 writeArchiveFixtureRunFiles(t *testing.T, runWorkDir string) { + t.Helper() + mustWriteFile(t, filepath.Join(runWorkDir, "inputs", "session.yml"), "session_id: 2026-05-03\n") + mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "raw", "speaker.json"), "{}\n") + mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "trimmed.json"), "{}\n") + mustWriteFile(t, filepath.Join(runWorkDir, "artifacts", "session_recap.md"), "# recap\n") + mustWriteFile(t, filepath.Join(runWorkDir, "reports", "audita.report.json"), "{}\n") + mustWriteFile(t, filepath.Join(runWorkDir, "config", "audita.generated.yml"), "key: value\n") + mustWriteFile(t, filepath.Join(runWorkDir, "logs", "audita.stderr.log"), "stderr\n") + mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n") +} + +type failKeyStore struct { + delegate *storage.FakeBackend + failKey string +} + +func (s *failKeyStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) { + return s.delegate.List(ctx, prefix) +} + +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 strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + return storage.ObjectInfo{}, errors.New("forced upload failure") + } + return s.delegate.Upload(ctx, localPath, key, opts) +} + +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) + } +} diff --git a/internal/app/runner.go b/internal/app/runner.go index 4416ad7..47cd59f 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -168,6 +168,10 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage env.Logger.Info("stage succeeded", "stage", s.Name()) } + if err := runPostArchiveCleanup(ctx, env, manifestPath, m, executed); err != nil { + return nil, fmt.Errorf("post-archive cleanup: %w", err) + } + return &RunSummary{ SessionID: cfg.Session.SessionID, ManifestPath: manifestPath, diff --git a/internal/config/config.go b/internal/config/config.go index e763210..432d934 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -36,7 +36,8 @@ type SessionConfig struct { // WorkspaceConfig configures local workspace behavior. type WorkspaceConfig struct { - Root string `yaml:"root"` + Root string `yaml:"root"` + CleanupAfterArchive bool `yaml:"cleanup_after_archive"` } // SecretsConfig configures optional local filesystem secret loading. diff --git a/internal/config/storage_archive_test.go b/internal/config/storage_archive_test.go index 32e5a44..d507f0f 100644 --- a/internal/config/storage_archive_test.go +++ b/internal/config/storage_archive_test.go @@ -47,6 +47,9 @@ func TestSpoolAndArchiveDefaults(t *testing.T) { if cfg.Pipeline.Spool.DeleteAudioAfterArchive { t.Fatalf("spool.delete_audio_after_archive = true, want false") } + if cfg.Pipeline.Workspace.CleanupAfterArchive { + t.Fatalf("workspace.cleanup_after_archive = true, want false") + } if cfg.Pipeline.Archive == nil { t.Fatal("archive should be initialized by defaults") }