diff --git a/README.md b/README.md index aa07765..ff322b5 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,6 @@ Implemented now: Not implemented yet: -- `archive` stage behavior - `notify` stage behavior - additional analyze artifacts beyond `session_recap` - generic DAG orchestration @@ -79,7 +78,9 @@ Current boundaries: - `audio_dir`/`audio_files` and `audio_s3` are mutually exclusive - real S3-compatible backend now exists in the storage adapter package - storage backend tests use fake storage and do not require live S3 -- no archive-stage S3 upload/promotion behavior yet +- archive uploads successful run records under `runs/{run_id}/` +- archive does not upload local audio by default +- archive does not yet perform promotion writes - no `current/manifest.json` or `current/run_id.txt` uploads yet S3 input details and current boundaries are documented in [docs/s3-audio-input.md](docs/s3-audio-input.md). @@ -106,7 +107,10 @@ Key invariant: Current boundary: - `prepare` uses `List` + `Download` through the backend when `session.inputs.audio_s3` is configured -- archive-stage uploads/promotion are still not implemented +- `archive` uses `Upload` through the backend for successful run-record uploads +- promotion uploads and current-pointer writes are still not implemented + +Archive run-upload details and boundaries are documented in [docs/archive-storage.md](docs/archive-storage.md). ## Canonical Stage Order diff --git a/architecture.md b/architecture.md index ea72aeb..7734a53 100644 --- a/architecture.md +++ b/architecture.md @@ -40,8 +40,7 @@ Implemented: Still placeholder/future: -- `archive` stage behavior -- archive-stage S3 upload/promotion behavior +- archive-stage promotion behavior - `notify` stage behavior - additional Scriptorium artifact types beyond `session_recap` - artifact-to-artifact workflows beyond the initial single-artifact implementation @@ -168,7 +167,8 @@ Remote object-store backend scope: - callers pass full bucket-relative object keys - backend methods do not prepend `root_prefix` or infer campaign/session/run paths - `prepare` now uses object-store `List` and `Download` for S3 audio input -- archive upload/promotion behavior using `Upload`/`Exists` remains future work +- `archive` now uses object-store `Upload` for successful run-record upload under the run prefix +- archive promotion behavior using top-level writes and current pointers remains future work Prepare S3 audio behavior (implemented): @@ -183,6 +183,25 @@ Prepare S3 audio behavior (implemented): - record S3 provenance in manifest input records (bucket/key/metadata/local paths/checksum) - no AWS SDK types are used in stage code; storage implementation details stay in storage adapter packages +Archive run upload behavior (implemented): + +- `archive` verifies prerequisite stage success before upload: + - `prepare`, `transcribe`, `merge`, `polish`, `normalize`, `trim`, `analyze` +- only successful/completed runs are uploaded +- uploaded run record destination is: + - `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/runs/{run_id}/` +- uploaded existing local paths include: + - `inputs/`, `transcripts/`, `artifacts/`, optional `reports/`, `config/`, `logs/`, and `manifest.json` +- local `audio/` is intentionally excluded from upload by default +- file upload order is deterministic (sorted relative paths) +- `archive.enabled: false` and `archive.upload_run: false` skip upload cleanly +- stage metadata records non-secret upload context: + - `s3_bucket`, `s3_run_prefix`, `files_uploaded`, `uploaded_paths`, `audio_upload_skipped` +- archive does not yet: + - write top-level promoted outputs + - write `current/manifest.json` + - write `current/run_id.txt` + `pipeline.scriptorium` is optional. Existing pipelines without Scriptorium continue to work. `pipeline.trim` is optional. Existing pipelines without trim config continue to work. diff --git a/docs/archive-storage.md b/docs/archive-storage.md new file mode 100644 index 0000000..7c26ed1 --- /dev/null +++ b/docs/archive-storage.md @@ -0,0 +1,73 @@ +# Archive Storage + +This document describes implemented archive-stage storage behavior. + +## Scope + +Implemented: + +- archive uploads successful run records to remote object storage through the storage backend abstraction. +- upload destination is the run prefix: + - `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/runs/{run_id}/` +- tests use fake storage and do not require live S3. + +Not implemented: + +- promotion uploads to top-level `transcripts/` or `artifacts/` +- `current/manifest.json` write +- `current/run_id.txt` write +- audio re-upload from local workdir +- failed/incomplete run uploads + +## Prerequisites + +Archive verifies the following stages succeeded before upload: + +- `prepare` +- `transcribe` +- `merge` +- `polish` +- `normalize` +- `trim` +- `analyze` + +If any prerequisite is missing or not succeeded, archive fails and does not upload. + +## Uploaded Run Record + +Archive uploads existing files from the run workdir when present: + +- `inputs/` +- `transcripts/` +- `artifacts/` +- `reports/` (optional) +- `config/` +- `logs/` +- `manifest.json` + +Archive preserves relative file paths under `runs/{run_id}/`. + +Example: + +- local `transcripts/raw/speaker.json` +- remote `runs/{run_id}/transcripts/raw/speaker.json` + +## Audio Upload Policy + +Archive does not upload local `audio/` by default. +Original audio is expected at the session-level audio prefix and is not duplicated under `runs/{run_id}/`. + +## Config Controls + +- `archive.enabled: false` skips archive cleanly. +- `archive.upload_run: false` skips run upload cleanly. + +## Metadata + +Archive stage metadata includes non-secret upload context: + +- `s3_bucket` +- `s3_run_prefix` +- `files_uploaded` +- `uploaded_paths` +- `audio_upload_skipped` diff --git a/docs/roadmap/narratio-s3-archive-implementation-plan.md b/docs/roadmap/narratio-s3-archive-implementation-plan.md index 54a0941..dc7d307 100644 --- a/docs/roadmap/narratio-s3-archive-implementation-plan.md +++ b/docs/roadmap/narratio-s3-archive-implementation-plan.md @@ -100,10 +100,16 @@ Implemented in repository: - fake storage backend for deterministic, no-network testing - S3-compatible backend built from `storage.s3` config - backend construction helper from resolved config +- archive run upload behavior: + - archive validates required prior stage success before uploading + - archive uploads successful run records under `runs/{run_id}/` + - upload set includes run-record files (`inputs`, `transcripts`, `artifacts`, optional `reports`, `config`, `logs`, `manifest.json`) + - local audio is not uploaded by default + - upload uses storage backend abstraction and deterministic ordering + - archive skips cleanly when `archive.enabled` or `archive.upload_run` is false Not implemented yet: -- archive-stage S3 upload behavior - promotion uploads - writing `current/manifest.json` and `current/run_id.txt` to S3 diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index 58269e7..f034221 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -381,6 +381,9 @@ storage: backend: s3 s3: bucket: test-bucket +archive: + enabled: true + upload_run: false whisperx: transcribe_url: ` + url + ` timeout: 2s diff --git a/internal/app/runner.go b/internal/app/runner.go index 5b0c11a..4416ad7 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -81,7 +81,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage if env.Storage == nil { env.Storage = &storage.NoopBackend{} } - if env.ObjectStore == nil && env.Config != nil && env.Config.Session != nil && env.Config.Session.Inputs.AudioS3 != nil { + if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages) { objectStore, err := storage.NewObjectStoreFromConfig(ctx, env.Config) if err != nil { return nil, fmt.Errorf("initialize object store backend: %w", err) @@ -409,3 +409,34 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err func manifestPathFor(cfg *config.Config) string { return filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.SessionID, "manifest.json") } + +func needsObjectStoreForRun(cfg *config.Config, stages []stage.Stage) bool { + if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { + return false + } + stageRequested := func(name string) bool { + for _, s := range stages { + if s != nil && s.Name() == name { + return true + } + } + return false + } + + if cfg.Session.Inputs.AudioS3 != nil && stageRequested("prepare") { + return true + } + if !stageRequested("archive") { + return false + } + if cfg.Pipeline.Archive == nil { + return false + } + if cfg.Pipeline.Archive.Enabled != nil && !*cfg.Pipeline.Archive.Enabled { + return false + } + if cfg.Pipeline.Archive.UploadRun != nil && !*cfg.Pipeline.Archive.UploadRun { + return false + } + return true +} diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index b704656..1b615d7 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -144,6 +144,15 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { } continue } + if name == "archive" { + if sr.Metadata == nil || sr.Metadata["stage"] != "archive" { + t.Fatalf("archive metadata missing stage=archive: %#v", sr.Metadata) + } + if sr.Metadata["skipped"] != true { + t.Fatalf("archive metadata missing skipped=true for test config without archive section: %#v", sr.Metadata) + } + continue + } if sr.Metadata == nil || sr.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", name) } @@ -315,7 +324,7 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { {name: "merge", env: &Env{Seriatim: &seriatim.FakeRunner{Err: errors.New("merge fail")}}}, {name: "polish", env: &Env{Audita: &audita.FakeRunner{Err: errors.New("polish fail")}}}, {name: "analyze", env: &Env{Scriptorium: &scriptorium.FakeRunner{RunErr: errors.New("analyze fail")}}}, - {name: "archive", env: &Env{Storage: &storage.FakeBackend{Err: errors.New("archive fail")}}}, + {name: "archive", env: &Env{ObjectStore: &storage.FakeBackend{UploadErr: errors.New("archive fail")}}}, {name: "notify", env: &Env{Notifier: ¬ify.FakeSender{Err: errors.New("notify fail")}}}, } @@ -400,6 +409,41 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { }, } } + if tc.name == "archive" { + cfg.Pipeline.Archive = &config.ArchiveConfig{ + Enabled: boolPtr(true), + UploadRun: boolPtr(true), + } + cfg.Pipeline.Storage.S3 = &config.StorageS3Config{ + Bucket: "my-dnd-archive", + RootPrefix: "dnd", + } + runID := "20260516T010203Z-0a1b2c3d" + runWorkDir := filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.Campaign, cfg.Session.SessionID, runID) + if err := os.MkdirAll(filepath.Join(runWorkDir, "inputs"), 0o755); err != nil { + t.Fatalf("mkdir archive inputs dir: %v", err) + } + if err := os.WriteFile(filepath.Join(runWorkDir, "inputs", "session.yml"), []byte("session_id: 2026-05-03\n"), 0o644); err != nil { + t.Fatalf("write archive fixture session.yml: %v", err) + } + if err := os.WriteFile(filepath.Join(runWorkDir, "manifest.json"), []byte("{}\n"), 0o644); err != nil { + t.Fatalf("write archive fixture manifest.json: %v", err) + } + + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.Campaign = cfg.Session.Campaign + seed.RunID = runID + seed.LocalWorkDir = runWorkDir + seed.S3Bucket = "my-dnd-archive" + seed.S3SessionPrefix = "dnd/campaigns/" + cfg.Session.Campaign + "/sessions/" + cfg.Session.SessionID + "/" + seed.S3RunPrefix = seed.S3SessionPrefix + "runs/" + runID + "/" + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze"} { + seed.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + if err := tc.env.ManifestStore.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatalf("seed archive manifest: %v", err) + } + } _, runErr := executeStages(context.Background(), cfg, []stage.Stage{selected}, RunOptions{Env: tc.env}) if runErr == nil { @@ -462,3 +506,8 @@ func mustWriteFile(t *testing.T, path, contents string) { t.Fatalf("WriteFile(%q): %v", path, err) } } + +func boolPtr(v bool) *bool { + p := v + return &p +} diff --git a/internal/stage/archive.go b/internal/stage/archive.go new file mode 100644 index 0000000..86e4136 --- /dev/null +++ b/internal/stage/archive.go @@ -0,0 +1,282 @@ +package stage + +import ( + "context" + "fmt" + "io/fs" + "os" + "path/filepath" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +type archiveStage struct{} + +var archivePrerequisiteStages = []string{ + "prepare", + "transcribe", + "merge", + "polish", + "normalize", + "trim", + "analyze", +} + +var archiveRunUploadDirs = []string{ + "inputs", + "transcripts", + "artifacts", + "reports", + "config", + "logs", +} + +func (archiveStage) Name() string { return "archive" } + +func (archiveStage) Declares() IODecl { + return IODecl{ + Inputs: []artifacts.Ref{ + {Kind: "manifest", Category: "input", RelativePath: "manifest.json"}, + }, + } +} + +func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil { + return nil, fmt.Errorf("archive: resolved config must include pipeline and session") + } + + if archiveDisabled(env) { + return &StageResult{ + Metadata: map[string]any{ + "stage": "archive", + "skipped": true, + "archive_enabled": false, + "audio_upload_skipped": true, + }, + }, nil + } + if archiveRunUploadDisabled(env) { + return &StageResult{ + Metadata: map[string]any{ + "stage": "archive", + "skipped": true, + "upload_run_enabled": false, + "audio_upload_skipped": true, + }, + }, nil + } + + if err := validateArchivePrerequisites(m); err != nil { + return nil, fmt.Errorf("archive: %w", err) + } + if env.ObjectStore == nil { + return nil, fmt.Errorf("archive: remote object store backend is required when archive run upload is enabled") + } + + workDir, err := archiveWorkDir(env, m) + if err != nil { + return nil, fmt.Errorf("archive: resolve local workdir: %w", err) + } + info, err := os.Stat(workDir) + if err != nil { + return nil, fmt.Errorf("archive: local workdir %q: %w", workDir, err) + } + if !info.IsDir() { + return nil, fmt.Errorf("archive: local workdir %q is not a directory", workDir) + } + + runPrefix, err := archiveRunPrefix(env, m) + if err != nil { + return nil, fmt.Errorf("archive: resolve s3 run prefix: %w", err) + } + bucket := archiveBucket(env, m) + if bucket == "" { + return nil, fmt.Errorf("archive: resolve s3 bucket: bucket is required") + } + + relFiles, err := collectArchiveRunFiles(workDir) + if err != nil { + return nil, fmt.Errorf("archive: collect run files: %w", err) + } + + uploaded := make([]string, 0, len(relFiles)) + for _, rel := range relFiles { + localPath := filepath.Join(workDir, filepath.FromSlash(rel)) + key := artifacts.S3RunRelativeDestinationKey(runPrefix, rel) + if _, err := env.ObjectStore.Upload(ctx, localPath, key, storage.UploadOptions{}); err != nil { + return nil, fmt.Errorf("archive: upload %q to %q: %w", rel, key, err) + } + uploaded = append(uploaded, rel) + } + + return &StageResult{ + Metadata: map[string]any{ + "stage": "archive", + "uploaded": true, + "s3_bucket": bucket, + "s3_run_prefix": runPrefix, + "files_uploaded": len(uploaded), + "uploaded_paths": uploaded, + "audio_upload_skipped": true, + }, + }, nil +} + +func archiveDisabled(env *Env) bool { + cfg := env.Config.Pipeline.Archive + if cfg == nil { + return true + } + return cfg.Enabled != nil && !*cfg.Enabled +} + +func archiveRunUploadDisabled(env *Env) bool { + cfg := env.Config.Pipeline.Archive + if cfg == nil { + return true + } + return cfg.UploadRun != nil && !*cfg.UploadRun +} + +func validateArchivePrerequisites(m *manifest.Manifest) error { + if m == nil { + return fmt.Errorf("manifest is required") + } + for _, stageName := range archivePrerequisiteStages { + sr := m.Stages[stageName] + if sr == nil { + return fmt.Errorf("prerequisite stage %q has not succeeded", stageName) + } + if sr.Status != manifest.StatusSucceeded { + return fmt.Errorf("prerequisite stage %q status is %q (want %q)", stageName, sr.Status, manifest.StatusSucceeded) + } + } + return nil +} + +func archiveWorkDir(env *Env, m *manifest.Manifest) (string, error) { + workDir := strings.TrimSpace(m.LocalWorkDir) + if workDir != "" { + cleaned := filepath.Clean(workDir) + if info, err := os.Stat(cleaned); err == nil && info.IsDir() { + return cleaned, nil + } + } + + sessionID := strings.TrimSpace(env.Config.Session.SessionID) + if sessionID == "" { + sessionID = strings.TrimSpace(m.SessionID) + } + campaign := strings.TrimSpace(env.Config.Session.Campaign) + if campaign == "" { + campaign = strings.TrimSpace(m.Campaign) + } + runID := strings.TrimSpace(m.RunID) + if runID == "" { + return "", fmt.Errorf("run id is required") + } + if campaign == "" || sessionID == "" { + return "", fmt.Errorf("campaign and session id are required") + } + runScoped := artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID) + if info, err := os.Stat(runScoped); err == nil && info.IsDir() { + return runScoped, nil + } + legacy := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID) + return legacy, nil +} + +func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) { + runPrefix := strings.TrimSpace(m.S3RunPrefix) + if runPrefix != "" { + return runPrefix, nil + } + + sessionID := strings.TrimSpace(env.Config.Session.SessionID) + if sessionID == "" { + sessionID = strings.TrimSpace(m.SessionID) + } + campaign := strings.TrimSpace(env.Config.Session.Campaign) + if campaign == "" { + campaign = strings.TrimSpace(m.Campaign) + } + runID := strings.TrimSpace(m.RunID) + if runID == "" { + return "", fmt.Errorf("run id is required") + } + if env.Config.Pipeline.Storage.S3 == nil { + return "", fmt.Errorf("pipeline.storage.s3 configuration is required") + } + sessionPrefix := artifacts.S3SessionPrefix(env.Config.Pipeline.Storage.S3.RootPrefix, campaign, sessionID) + if strings.TrimSpace(sessionPrefix) == "" { + return "", fmt.Errorf("session prefix is required") + } + return artifacts.S3RunPrefix(sessionPrefix, runID), nil +} + +func archiveBucket(env *Env, m *manifest.Manifest) string { + if m != nil && strings.TrimSpace(m.S3Bucket) != "" { + return strings.TrimSpace(m.S3Bucket) + } + if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Pipeline.Storage.S3 == nil { + return "" + } + return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket) +} + +func collectArchiveRunFiles(workDir string) ([]string, error) { + files := make([]string, 0, 64) + + for _, dirName := range archiveRunUploadDirs { + fullDir := filepath.Join(workDir, dirName) + info, err := os.Stat(fullDir) + if err != nil { + if os.IsNotExist(err) { + continue + } + return nil, fmt.Errorf("stat %q: %w", fullDir, err) + } + if !info.IsDir() { + continue + } + + if err := filepath.WalkDir(fullDir, func(path string, d fs.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr + } + if d.IsDir() { + return nil + } + rel, err := filepath.Rel(workDir, path) + if err != nil { + return fmt.Errorf("relative path from %q to %q: %w", workDir, path, err) + } + rel = filepath.ToSlash(rel) + files = append(files, rel) + return nil + }); err != nil { + return nil, fmt.Errorf("walk %q: %w", fullDir, err) + } + } + + manifestPath := filepath.Join(workDir, "manifest.json") + manifestInfo, err := os.Stat(manifestPath) + if err != nil { + if os.IsNotExist(err) { + return nil, fmt.Errorf("manifest.json not found in workdir %q", workDir) + } + return nil, fmt.Errorf("stat %q: %w", manifestPath, err) + } + if manifestInfo.IsDir() { + return nil, fmt.Errorf("manifest path %q is a directory", manifestPath) + } + files = append(files, "manifest.json") + + sort.Strings(files) + return files, nil +} diff --git a/internal/stage/archive_test.go b/internal/stage/archive_test.go new file mode 100644 index 0000000..17bb62c --- /dev/null +++ b/internal/stage/archive_test.go @@ -0,0 +1,198 @@ +package stage + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestArchiveSkipsWhenDisabled(t *testing.T) { + env, m, _ := archiveFixture(t) + env.Config.Pipeline.Archive.Enabled = boolPtr(false) + + result, err := archiveStage{}.Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if result.Metadata["skipped"] != true { + t.Fatalf("metadata = %#v, want skipped=true", result.Metadata) + } + if len(env.ObjectStore.(*storage.FakeBackend).Objects) != 0 { + t.Fatalf("unexpected uploads when archive disabled") + } +} + +func TestArchiveSkipsRunUploadWhenDisabled(t *testing.T) { + env, m, _ := archiveFixture(t) + env.Config.Pipeline.Archive.UploadRun = boolPtr(false) + + result, err := archiveStage{}.Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if result.Metadata["skipped"] != true { + t.Fatalf("metadata = %#v, want skipped=true", result.Metadata) + } + if len(env.ObjectStore.(*storage.FakeBackend).Objects) != 0 { + t.Fatalf("unexpected uploads when upload_run disabled") + } +} + +func TestArchiveFailsWhenPrerequisiteNotSucceeded(t *testing.T) { + env, m, _ := archiveFixture(t) + m.Stages["trim"].Status = manifest.StatusFailed + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), `prerequisite stage "trim"`) { + t.Fatalf("Run() error = %v, want prerequisite failure", err) + } +} + +func TestArchiveUploadsRunRecordWithoutAudio(t *testing.T) { + env, m, workDir := archiveFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + + result, err := archiveStage{}.Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + + wantUploaded := []string{ + "artifacts/session_recap.md", + "config/audita.generated.yml", + "inputs/session.yml", + "logs/audita.stderr.log", + "manifest.json", + "reports/audita.report.json", + "transcripts/raw/speaker.json", + "transcripts/trimmed.json", + } + if got := result.Metadata["uploaded_paths"]; !reflect.DeepEqual(got, wantUploaded) { + t.Fatalf("uploaded_paths = %#v, want %#v", got, wantUploaded) + } + if got := result.Metadata["files_uploaded"]; got != len(wantUploaded) { + t.Fatalf("files_uploaded = %#v, want %d", got, len(wantUploaded)) + } + + for _, rel := range wantUploaded { + key := m.S3RunPrefix + rel + if _, ok := fake.Objects[key]; !ok { + t.Fatalf("missing uploaded key %q", key) + } + } + audioKey := m.S3RunPrefix + "audio/speaker.flac" + if _, ok := fake.Objects[audioKey]; ok { + t.Fatalf("audio key %q should not be uploaded", audioKey) + } + + // Ensure relative paths are preserved under runs/{run_id}/. + if _, err := os.Stat(filepath.Join(workDir, "transcripts", "raw", "speaker.json")); err != nil { + t.Fatalf("expected local fixture transcript file: %v", err) + } +} + +func TestArchiveFailsWhenUploadFails(t *testing.T) { + env, m, _ := archiveFixture(t) + env.ObjectStore.(*storage.FakeBackend).UploadErr = errors.New("upload failed") + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "upload") { + t.Fatalf("Run() error = %v, want upload failure", err) + } +} + +func TestArchiveFailsWithoutObjectStore(t *testing.T) { + env, m, _ := archiveFixture(t) + env.ObjectStore = nil + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "object store") { + t.Fatalf("Run() error = %v, want object store backend failure", err) + } +} + +func TestArchiveDoesNotWriteCurrentPointers(t *testing.T) { + env, m, _ := archiveFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + + if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; ok { + t.Fatalf("unexpected current/run_id.txt upload") + } + if _, ok := fake.Objects[m.S3SessionPrefix+"current/manifest.json"]; ok { + t.Fatalf("unexpected current/manifest.json upload") + } +} + +func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { + t.Helper() + + root := t.TempDir() + runID := "20260516T010203Z-1a2b3c4d" + campaign := "forsaken" + sessionID := "2026-04-19" + workDir := filepath.Join(root, "work", campaign, sessionID, runID) + + writeStageTestFile(t, filepath.Join(workDir, "inputs", "session.yml"), "session_id: 2026-04-19\n") + writeStageTestFile(t, filepath.Join(workDir, "transcripts", "raw", "speaker.json"), "{}\n") + writeStageTestFile(t, filepath.Join(workDir, "transcripts", "trimmed.json"), "{}\n") + writeStageTestFile(t, filepath.Join(workDir, "artifacts", "session_recap.md"), "# recap\n") + writeStageTestFile(t, filepath.Join(workDir, "reports", "audita.report.json"), "{}\n") + writeStageTestFile(t, filepath.Join(workDir, "config", "audita.generated.yml"), "key: value\n") + writeStageTestFile(t, filepath.Join(workDir, "logs", "audita.stderr.log"), "stderr\n") + writeStageTestFile(t, filepath.Join(workDir, "audio", "speaker.flac"), "flac") + writeStageTestFile(t, filepath.Join(workDir, "manifest.json"), "{}\n") + + m := manifest.New(sessionID, time.Now().UTC()) + m.Campaign = campaign + m.RunID = runID + m.LocalWorkDir = workDir + m.S3Bucket = "my-dnd-archive" + m.S3SessionPrefix = "dnd/campaigns/forsaken/sessions/2026-04-19/" + m.S3RunPrefix = m.S3SessionPrefix + "runs/" + runID + "/" + for _, name := range archivePrerequisiteStages { + m.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + + env := &Env{ + Config: &config.Config{ + Pipeline: &config.PipelineConfig{ + Workspace: config.WorkspaceConfig{Root: root}, + Storage: config.StorageConfig{ + S3: &config.StorageS3Config{ + Bucket: "my-dnd-archive", + RootPrefix: "dnd", + }, + }, + Archive: &config.ArchiveConfig{ + Enabled: boolPtr(true), + UploadRun: boolPtr(true), + }, + }, + Session: &config.SessionConfig{ + SessionID: sessionID, + Campaign: campaign, + }, + }, + ObjectStore: &storage.FakeBackend{}, + } + return env, m, workDir +} + +func boolPtr(v bool) *bool { + p := v + return &p +} diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index c7f6b85..50ba83a 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -3,10 +3,8 @@ package stage import ( "context" "fmt" - "path/filepath" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" - "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) @@ -47,25 +45,7 @@ func (s placeholderStage) Run(ctx context.Context, env *Env, m *manifest.Manifes return result, nil } - paths := env.ArtifactStore.SessionPaths(sessionID) - switch s.name { - case "archive": - if env.Storage != nil { - req := storage.ArchiveRequest{ - SessionID: sessionID, - ManifestPath: paths.ManifestPath, - Items: []storage.ArchiveItem{{ - Kind: "artifact", - LocalPath: filepath.Join(paths.ArtifactsDir, "session-log.md"), - RemoteKey: "sessions/" + sessionID + "/artifacts/session-log.md", - }}, - } - _, err := env.Storage.Archive(ctx, req) - if err != nil { - return nil, fmt.Errorf("placeholder archive adapter call failed: %w", err) - } - } case "notify": if env.Notifier != nil { req := notify.SendRequest{ @@ -95,7 +75,7 @@ func All() []Stage { normalizeStage{}, trimStage{}, analyzeStage{}, - placeholderStage{name: "archive"}, + archiveStage{}, placeholderStage{name: "notify"}, } } diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 229568d..9521205 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -49,7 +49,19 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { Config: &config.Config{ SessionPath: sessionPath, PipelinePath: pipelinePath, - Pipeline: &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: root}}, + Pipeline: &config.PipelineConfig{ + Workspace: config.WorkspaceConfig{Root: root}, + Storage: config.StorageConfig{ + S3: &config.StorageS3Config{ + Bucket: "my-dnd-archive", + RootPrefix: "dnd", + }, + }, + Archive: &config.ArchiveConfig{ + Enabled: boolPtr(true), + UploadRun: boolPtr(true), + }, + }, Session: &config.SessionConfig{ SessionID: "2026-05-03", Inputs: config.SessionInputsConfig{ @@ -66,10 +78,24 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { Audita: af, Scriptorium: sc, Storage: st, + ObjectStore: st, Notifier: nf, } m := manifest.New("2026-05-03", time.Now().UTC()) + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze"} { + m.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + m.RunID = "20260516T000000Z-abcdef12" + m.Campaign = "sample-campaign" + m.LocalWorkDir = filepath.Join(root, "work", "sample-campaign", "2026-05-03", m.RunID) + m.S3RunPrefix = "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/" + m.RunID + "/" + m.S3Bucket = "my-dnd-archive" + if err := os.MkdirAll(filepath.Join(m.LocalWorkDir, "inputs"), 0o755); err != nil { + t.Fatalf("mkdir workdir inputs: %v", err) + } + writeStageTestFile(t, filepath.Join(m.LocalWorkDir, "inputs", "session.yml"), "session_id: 2026-05-03\n") + writeStageTestFile(t, filepath.Join(m.LocalWorkDir, "manifest.json"), "{}\n") for _, s := range stages { result, err := s.Run(context.Background(), env, m) if err != nil { @@ -150,6 +176,15 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } continue } + if s.Name() == "archive" { + if result.Metadata["stage"] != "archive" { + t.Fatalf("archive metadata = %#v, want stage=archive", result.Metadata) + } + if result.Metadata["uploaded"] != true { + t.Fatalf("archive metadata = %#v, want uploaded=true", result.Metadata) + } + continue + } if result.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", s.Name()) } @@ -167,8 +202,11 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { if len(sc.RunRequests) != 0 { t.Fatalf("scriptorium run calls = %d, want 0 when scriptorium config is absent", len(sc.RunRequests)) } - if len(st.Requests) != 1 { - t.Fatalf("storage calls = %d, want 1", len(st.Requests)) + if len(st.Requests) != 0 { + t.Fatalf("storage archive calls = %d, want 0", len(st.Requests)) + } + if _, ok := st.Objects["dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/"+m.RunID+"/manifest.json"]; !ok { + t.Fatalf("archive upload missing manifest key in fake object store") } if len(nf.Requests) != 1 { t.Fatalf("notify calls = %d, want 1", len(nf.Requests)) @@ -181,7 +219,6 @@ func TestPlaceholderAdapterErrorPropagation(t *testing.T) { env *Env wantErr string }{ - {stageName: "archive", env: &Env{Storage: &storage.FakeBackend{Err: errors.New("sterr")}}, wantErr: "archive"}, {stageName: "notify", env: &Env{Notifier: ¬ify.FakeSender{Err: errors.New("nerr")}}, wantErr: "notify"}, }