package app import ( "bytes" "context" "errors" "os" "path/filepath" "reflect" "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" ) 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 TestRunStageCleanupUsesPublishEvidenceAndFailsClosedWithoutIt(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) enableCleanupForCommandTest(t, pipelinePath) cfg, err := config.Load(pipelinePath, campaignPath, sessionPath) if err != nil { t.Fatalf("load cleanup config: %v", err) } publishedRunID := "20260904T010203Z-a1b2c3d4" m := manifest.New(cfg.Session.SessionID, time.Date(2026, 9, 4, 1, 2, 3, 0, time.UTC)) m.Campaign = cfg.Session.Campaign m.RunID = publishedRunID m.S3SessionPrefix = artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, m.Campaign, m.SessionID) m.MarkStageSucceeded("publish", time.Date(2026, 9, 4, 1, 3, 0, 0, time.UTC), nil) m.Stages["publish"].Metadata = cleanupPublishMetadata(m) manifestPath := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID) if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPath, m); err != nil { t.Fatalf("save publication evidence: %v", err) } workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID, publishedRunID) spoolTarget := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, m.Campaign, m.SessionID, publishedRunID) for _, target := range []string{workspaceTarget, spoolTarget} { if err := os.MkdirAll(target, 0o755); err != nil { t.Fatalf("create cleanup target: %v", err) } } var output bytes.Buffer if err := RunStage(context.Background(), []string{ "cleanup", cfg.Session.SessionID, "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, }, &output); err != nil { t.Fatalf("RunStage(cleanup): %v", err) } if !strings.Contains(output.String(), "stage=cleanup executed=1 skipped=0") { t.Fatalf("output = %q, want direct cleanup execution", output.String()) } for _, target := range []string{workspaceTarget, spoolTarget} { if _, statErr := os.Lstat(target); !errors.Is(statErr, os.ErrNotExist) { t.Fatalf("cleanup target %q stat error = %v, want absent", target, statErr) } } cleaned := loadCleanupOrchestrationManifest(t, cfg) if cleaned.RunID == publishedRunID { t.Fatalf("cleanup invocation run id = %q, want a new invocation identity", cleaned.RunID) } if _, statErr := os.Lstat(cleaned.LocalWorkDir); !errors.Is(statErr, os.ErrNotExist) { t.Fatalf("cleanup invocation work directory stat error = %v, want absent", statErr) } missingRoot := t.TempDir() missingPipeline, missingCampaign, missingSession := writeValidConfigFiles(t, missingRoot) enableCleanupForCommandTest(t, missingPipeline) err = RunStage(context.Background(), []string{ "cleanup", "2026-05-03", "--config", missingPipeline, "--campaign-file", missingCampaign, "--session", missingSession, }, &bytes.Buffer{}) if err == nil || !strings.Contains(err.Error(), "cleanup requires a recorded publish stage") { t.Fatalf("RunStage(cleanup) without evidence error = %v", err) } } func enableCleanupForCommandTest(t *testing.T, pipelinePath string) { t.Helper() data, err := os.ReadFile(pipelinePath) if err != nil { t.Fatalf("read pipeline config: %v", err) } content := strings.Replace(string(data), "workspace:\n root: ", "workspace:\n cleanup_after_publish: true\n root: ", 1) content = strings.Replace(content, "spool:\n root: ", "spool:\n delete_audio_after_publish: true\n root: ", 1) content = strings.Replace(content, " upload_run: false", " upload_run: true", 1) if err := os.WriteFile(pipelinePath, []byte(content), 0o644); err != nil { t.Fatalf("write cleanup config: %v", err) } } func cleanupOrchestrationConfig(t *testing.T) (*config.Config, *storage.FakeBackend) { t.Helper() cfg := testConfig(t) 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 }