diff --git a/README.md b/README.md index 56d9919..aa07765 100644 --- a/README.md +++ b/README.md @@ -49,7 +49,7 @@ YAML decoding is strict (`KnownFields(true)`), so unknown fields fail fast. ## Storage And Archive Foundations -Narratio now includes configuration and path-model foundations for future S3 audio input and archive support. +Narratio now includes configuration and path-model foundations for archive support, plus implemented prepare-stage S3 audio input. Implemented foundations: @@ -58,7 +58,7 @@ Implemented foundations: - `pipeline.archive` config shape (`enabled`, `upload_run`, `promote_artifacts`) - promotion-rule validation (`from`/`to` required, relative-only paths, traversal rejected) - `session.campaign` requirement for campaign-aware path construction -- optional `session.inputs.audio_s3.prefix` modeling (not executed yet) +- optional `session.inputs.audio_s3.prefix` modeling and prepare-stage S3 audio download - run ID generation and S3/local path helper foundations - manifest run/path identity fields @@ -79,10 +79,11 @@ 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 prepare-stage S3 list/download behavior yet - no archive-stage S3 upload/promotion behavior yet - 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). + ## Remote Storage Backend Narratio includes an object-store backend layer for future prepare/archive work: @@ -104,7 +105,8 @@ Key invariant: Current boundary: -- this backend layer is implemented but not yet wired into prepare-stage audio retrieval or archive-stage uploads +- `prepare` uses `List` + `Download` through the backend when `session.inputs.audio_s3` is configured +- archive-stage uploads/promotion are still not implemented ## Canonical Stage Order diff --git a/architecture.md b/architecture.md index 592a162..ea72aeb 100644 --- a/architecture.md +++ b/architecture.md @@ -41,7 +41,6 @@ Implemented: Still placeholder/future: - `archive` stage behavior -- prepare-stage S3 audio download behavior - archive-stage S3 upload/promotion behavior - `notify` stage behavior - additional Scriptorium artifact types beyond `session_recap` @@ -152,9 +151,9 @@ Session input foundations: - `session.campaign` is required - local audio remains supported through `session.inputs.audio_dir` or `session.inputs.audio_files` -- optional S3 audio input shape is modeled as `session.inputs.audio_s3.prefix` +- optional S3 audio input shape is `session.inputs.audio_s3.prefix` - `audio_dir`/`audio_files` and `audio_s3` are mutually exclusive -- S3 input execution (object listing/downloading) is not implemented yet +- when `audio_s3` is configured, `prepare` lists and downloads `.flac` objects through the object-store backend Cross-config validation scope: @@ -168,7 +167,21 @@ Remote object-store backend scope: - S3 key/session path semantics remain outside the backend, with this invariant: - callers pass full bucket-relative object keys - backend methods do not prepend `root_prefix` or infer campaign/session/run paths -- the backend layer is available for future prepare/archive usage, but no stage currently invokes `List`/`Download`/`Upload`/`Exists` +- `prepare` now uses object-store `List` and `Download` for S3 audio input +- archive upload/promotion behavior using `Upload`/`Exists` remains future work + +Prepare S3 audio behavior (implemented): + +- compute session prefix as `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/` +- resolve `session.inputs.audio_s3.prefix` under that session prefix +- list objects under the computed audio prefix and filter `.flac` keys +- fail clearly when no `.flac` objects are found +- download selected objects to spool audio path: + - `{spool.root}/{campaign}/{session_id}/{run_id}/audio/` +- materialize audio files into workdir audio path: + - `{workspace.root}/work/{campaign}/{session_id}/{run_id}/audio/` +- 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 `pipeline.scriptorium` is optional. Existing pipelines without Scriptorium continue to work. diff --git a/docs/roadmap/narratio-s3-archive-implementation-plan.md b/docs/roadmap/narratio-s3-archive-implementation-plan.md index 0b765d9..54a0941 100644 --- a/docs/roadmap/narratio-s3-archive-implementation-plan.md +++ b/docs/roadmap/narratio-s3-archive-implementation-plan.md @@ -103,7 +103,6 @@ Implemented in repository: Not implemented yet: -- prepare-stage S3 list/download behavior - archive-stage S3 upload behavior - promotion uploads - writing `current/manifest.json` and `current/run_id.txt` to S3 @@ -314,7 +313,7 @@ work/audio transcribe ``` -The `prepare` stage should: +Implemented `prepare` behavior: 1. list `.flac` objects under the configured S3 audio prefix, 2. fail clearly if none are found, @@ -331,7 +330,11 @@ spool: delete_audio_after_archive: true ``` -For v1, prefer retaining local workdir audio until the run has successfully archived. The source of truth remains S3, but local diagnostics are valuable during development. +Current boundary: + +- downloaded audio is retained in spool/workdir +- spool cleanup policy remains future work +- archive does not upload source audio by default ## 6. Configuration Design diff --git a/docs/s3-audio-input.md b/docs/s3-audio-input.md new file mode 100644 index 0000000..24867ac --- /dev/null +++ b/docs/s3-audio-input.md @@ -0,0 +1,82 @@ +# S3 Audio Input + +This document describes implemented S3 audio input behavior in `prepare`. + +## Scope + +Implemented: + +- `prepare` can acquire source audio from S3 when `session.inputs.audio_s3.prefix` is configured. +- object listing and download go through the storage backend abstraction. +- tests use fake storage; no live S3 service is required for test runs. + +Not implemented: + +- archive uploads (run uploads, promotion uploads, current pointer writes) +- spool cleanup/deletion behavior +- uploads of failed runs + +## Required Configuration + +`pipeline.yml`: + +- `storage.s3.bucket` must be set when S3 audio input is used. +- `storage.s3.root_prefix` defaults to `dnd`. +- `spool.root` defaults to `/var/spool/narratio`. + +`session.yml`: + +- configure `session.campaign` and `session.session_id`. +- configure `session.inputs.audio_s3.prefix` for S3 audio input. +- do not configure `inputs.audio_dir` or `inputs.audio_files` at the same time as `inputs.audio_s3`. + +## Prefix Shape + +Session S3 root: + +`{root_prefix}/campaigns/{campaign}/sessions/{session_id}/` + +Audio prefix: + +`{session_root}/{audio_s3.prefix}` + +Example: + +`dnd/campaigns/forsaken/sessions/2026-04-19/audio/` + +Audio files must already exist in S3 before running Narratio. + +## Prepare Behavior + +When `inputs.audio_s3.prefix` is configured, `prepare`: + +1. lists objects under the computed S3 audio prefix +2. filters to `.flac` objects +3. fails when no `.flac` objects are found +4. downloads selected objects to spool audio: + - `{spool.root}/{campaign}/{session_id}/{run_id}/audio/` +5. materializes audio into workdir audio: + - `{workspace.root}/work/{campaign}/{session_id}/{run_id}/audio/` +6. records input provenance in the manifest (bucket, key, metadata, local paths, checksum) + +Notes: + +- `.flac` filtering is case-insensitive. +- ETag is recorded as provider metadata only and is not treated as a checksum. + +## Local Audio Development + +Local audio workflows remain supported: + +- `inputs.audio_dir` +- `inputs.audio_files` + +These options are mutually exclusive with `inputs.audio_s3`. + +## Archive Boundary + +Current archive behavior remains unimplemented in this area: + +- no upload of source audio by default +- no upload of failed runs +- no `current/manifest.json` or `current/run_id.txt` writes diff --git a/examples/session.minimal.yml b/examples/session.minimal.yml index 6207805..0ff8244 100644 --- a/examples/session.minimal.yml +++ b/examples/session.minimal.yml @@ -5,6 +5,7 @@ title: Sample Session inputs: audio_dir: ./audio # Optional S3 input alternative. Do not configure with audio_dir/audio_files. + # Narratio prepare lists this prefix and downloads .flac files. # audio_s3: # prefix: "audio/" speakers_file: ./speakers.yml diff --git a/internal/app/runner.go b/internal/app/runner.go index d62da62..5b0c11a 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -81,6 +81,13 @@ 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 { + objectStore, err := storage.NewObjectStoreFromConfig(ctx, env.Config) + if err != nil { + return nil, fmt.Errorf("initialize object store backend: %w", err) + } + env.ObjectStore = objectStore + } if env.Notifier == nil { env.Notifier = ¬ify.NoopSender{} } diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index befbb86..2ae3c5b 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -14,9 +14,15 @@ type ErrorRecord struct { // InputRecord captures one resolved input and optional checksum. type InputRecord struct { - Kind string `json:"kind"` - Path string `json:"path"` - Checksum string `json:"checksum,omitempty"` + Kind string `json:"kind"` + Path string `json:"path"` + Checksum string `json:"checksum,omitempty"` + Source string `json:"source,omitempty"` + S3Bucket string `json:"s3_bucket,omitempty"` + S3Key string `json:"s3_key,omitempty"` + S3Size int64 `json:"s3_size,omitempty"` + S3ETag string `json:"s3_etag,omitempty"` + SpoolPath string `json:"spool_path,omitempty"` } // ArtifactRecord captures one produced artifact and optional remote metadata. diff --git a/internal/stage/prepare.go b/internal/stage/prepare.go index 0c08d01..c2d5668 100644 --- a/internal/stage/prepare.go +++ b/internal/stage/prepare.go @@ -6,10 +6,12 @@ import ( "encoding/hex" "fmt" "os" + "path" "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/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" @@ -33,7 +35,7 @@ func (prepareStage) Declares() IODecl { } } -func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { +func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { if env == nil || env.Config == nil { return nil, fmt.Errorf("prepare: stage environment config is required") } @@ -89,13 +91,12 @@ func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*Sta } } - resolvedAudio, err := resolveAudioFiles(sessionDir, env.Config.Session.Inputs) + resolvedLocalAudio, useS3Audio, err := resolveAudioInputs(sessionDir, env.Config.Session.Inputs) if err != nil { return nil, fmt.Errorf("prepare: resolve audio inputs: %w", err) } - copiedByDest := map[string]string{} - inputs := make([]manifest.InputRecord, 0, 5+len(resolvedAudio)) + inputs := make([]manifest.InputRecord, 0, 5+len(resolvedLocalAudio)) registerInput := func(kind, path, checksum string) { inputs = append(inputs, manifest.InputRecord{Kind: kind, Path: path, Checksum: checksum}) } @@ -134,19 +135,14 @@ func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*Sta registerInput(cfgFile.kind, cfgFile.dst, checksum) } - for _, src := range resolvedAudio { - base := filepath.Base(src) - if prev, exists := copiedByDest[base]; exists && prev != src { - return nil, fmt.Errorf("prepare: duplicate audio basename %q from %q and %q", base, prev, src) + if useS3Audio { + if err := materializeS3AudioInputs(ctx, env, m, sessionID, &inputs); err != nil { + return nil, fmt.Errorf("prepare: materialize s3 audio: %w", err) } - copiedByDest[base] = src - - dst := filepath.Join(paths.AudioDir, base) - checksum, err := copyFileIfChanged(env.ArtifactStore, src, dst) - if err != nil { - return nil, fmt.Errorf("prepare: materialize audio %q: %w", base, err) + } else { + if err := materializeLocalAudioInputs(env, paths, resolvedLocalAudio, registerInput); err != nil { + return nil, fmt.Errorf("prepare: %w", err) } - registerInput("audio", dst, checksum) } sort.Slice(inputs, func(i, j int) bool { @@ -162,7 +158,7 @@ func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*Sta "prepared": true, "stage": "prepare", "inputs_count": len(inputs), - "audio_files_resolved": len(resolvedAudio), + "audio_files_resolved": countAudioInputs(inputs), }, }, nil } @@ -171,7 +167,23 @@ func renderResolvedPipeline(cfg *config.PipelineConfig) ([]byte, error) { return yaml.Marshal(cfg) } -func resolveAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) { +func resolveAudioInputs(sessionDir string, inputs config.SessionInputsConfig) ([]string, bool, error) { + hasLocal := strings.TrimSpace(inputs.AudioDir) != "" || len(inputs.AudioFiles) > 0 + if inputs.AudioS3 != nil { + if hasLocal { + return nil, false, fmt.Errorf("audio_dir/audio_files and audio_s3 are mutually exclusive") + } + return nil, true, nil + } + + local, err := resolveLocalAudioFiles(sessionDir, inputs) + if err != nil { + return nil, false, err + } + return local, false, nil +} + +func resolveLocalAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) { if len(inputs.AudioFiles) > 0 { out := make([]string, 0, len(inputs.AudioFiles)) for _, p := range inputs.AudioFiles { @@ -223,6 +235,142 @@ func resolveAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([] return out, nil } +func materializeLocalAudioInputs(env *Env, paths artifacts.SessionPaths, resolvedAudio []string, registerInput func(kind, path, checksum string)) error { + copiedByDest := map[string]string{} + for _, src := range resolvedAudio { + base := filepath.Base(src) + if prev, exists := copiedByDest[base]; exists && prev != src { + return fmt.Errorf("duplicate audio basename %q from %q and %q", base, prev, src) + } + copiedByDest[base] = src + + dst := filepath.Join(paths.AudioDir, base) + checksum, err := copyFileIfChanged(env.ArtifactStore, src, dst) + if err != nil { + return fmt.Errorf("materialize audio %q: %w", base, err) + } + registerInput("audio", dst, checksum) + } + return nil +} + +func materializeS3AudioInputs(ctx context.Context, env *Env, m *manifest.Manifest, sessionID string, inputs *[]manifest.InputRecord) error { + if env.ObjectStore == nil { + return fmt.Errorf("s3 audio input requires object store backend") + } + if env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil || env.Config.Pipeline.Storage.S3 == nil || env.Config.Session.Inputs.AudioS3 == nil { + return fmt.Errorf("s3 audio input requires pipeline.storage.s3 and session.inputs.audio_s3 configuration") + } + + campaign := strings.TrimSpace(env.Config.Session.Campaign) + if campaign == "" { + return fmt.Errorf("session campaign is required for s3 audio input") + } + runID := strings.TrimSpace(m.RunID) + if runID == "" { + return fmt.Errorf("run id is required for s3 audio input") + } + + sessionPrefix := artifacts.S3SessionPrefix(env.Config.Pipeline.Storage.S3.RootPrefix, campaign, sessionID) + audioPrefix := artifacts.S3AudioPrefix(sessionPrefix, env.Config.Session.Inputs.AudioS3.Prefix) + objects, err := env.ObjectStore.List(ctx, audioPrefix) + if err != nil { + return fmt.Errorf("list s3 audio objects under %q: %w", audioPrefix, err) + } + + audioObjects := make([]storage.ObjectInfo, 0, len(objects)) + for _, obj := range objects { + key := strings.TrimSpace(obj.Key) + if key == "" || strings.HasSuffix(key, "/") { + continue + } + if !isFlac(key) { + continue + } + audioObjects = append(audioObjects, obj) + } + sort.Slice(audioObjects, func(i, j int) bool { + return audioObjects[i].Key < audioObjects[j].Key + }) + if len(audioObjects) == 0 { + return fmt.Errorf("no .flac files found under s3 audio prefix %q", audioPrefix) + } + + spoolAudioDir := strings.TrimSpace(m.LocalSpoolDir) + if spoolAudioDir == "" { + spoolAudioDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, campaign, sessionID, runID) + } + workAudioDir := filepath.Join(pathsWorkDirForManifest(env, m, sessionID), "audio") + + if err := os.MkdirAll(spoolAudioDir, 0o755); err != nil { + return fmt.Errorf("create spool audio directory %q: %w", spoolAudioDir, err) + } + if err := os.MkdirAll(workAudioDir, 0o755); err != nil { + return fmt.Errorf("create work audio directory %q: %w", workAudioDir, err) + } + + seenBase := map[string]string{} + for _, obj := range audioObjects { + base := path.Base(obj.Key) + if prev, exists := seenBase[base]; exists && prev != obj.Key { + return fmt.Errorf("duplicate s3 audio basename %q from %q and %q", base, prev, obj.Key) + } + seenBase[base] = obj.Key + + spoolPath := filepath.Join(spoolAudioDir, base) + if err := env.ObjectStore.Download(ctx, obj.Key, spoolPath); err != nil { + return fmt.Errorf("download s3 audio object %q: %w", obj.Key, err) + } + + workPath := filepath.Join(workAudioDir, base) + checksum, err := copyFileIfChanged(env.ArtifactStore, spoolPath, workPath) + if err != nil { + return fmt.Errorf("materialize downloaded audio %q: %w", base, err) + } + + *inputs = append(*inputs, manifest.InputRecord{ + Kind: "audio", + Path: workPath, + Checksum: checksum, + Source: "s3", + S3Bucket: strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket), + S3Key: obj.Key, + S3Size: obj.Size, + S3ETag: obj.ETag, + SpoolPath: spoolPath, + }) + } + + return nil +} + +func countAudioInputs(inputs []manifest.InputRecord) int { + count := 0 + for _, in := range inputs { + if in.Kind == "audio" { + count++ + } + } + return count +} + +func pathsWorkDirForManifest(env *Env, m *manifest.Manifest, sessionID string) string { + if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil { + return "" + } + if m != nil && strings.TrimSpace(m.LocalWorkDir) != "" { + return strings.TrimSpace(m.LocalWorkDir) + } + runID := "" + if m != nil { + runID = strings.TrimSpace(m.RunID) + } + if runID != "" { + return artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, env.Config.Session.Campaign, sessionID, runID) + } + return artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID) +} + func resolvePath(baseDir, p string) (string, error) { trimmed := strings.TrimSpace(p) if trimmed == "" { diff --git a/internal/stage/prepare_test.go b/internal/stage/prepare_test.go index a920c44..8a495cf 100644 --- a/internal/stage/prepare_test.go +++ b/internal/stage/prepare_test.go @@ -8,6 +8,7 @@ import ( "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" @@ -150,6 +151,139 @@ func TestPrepareStageIdempotent(t *testing.T) { } } +func TestPrepareStageS3AudioDownloadAndMaterialization(t *testing.T) { + env, m := setupPrepareEnv(t) + env.Config.Session.Campaign = "forsaken" + env.Config.Session.Inputs.AudioDir = "" + env.Config.Session.Inputs.AudioFiles = nil + env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"} + env.Config.Pipeline.Spool = config.SpoolConfig{Root: filepath.Join(t.TempDir(), "spool")} + env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "my-dnd-archive", RootPrefix: "dnd"} + m.RunID = "20260515T031522Z-a1b2c3d4" + m.LocalWorkDir = artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, "forsaken", m.SessionID, m.RunID) + m.LocalSpoolDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, "forsaken", m.SessionID, m.RunID) + + fake := &storage.FakeBackend{} + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/alice.flac", Data: []byte("alice")}) + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/bob.FLAC", Data: []byte("bob")}) + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/ignore.txt", Data: []byte("x")}) + env.ObjectStore = fake + + result, err := (prepareStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + if result == nil || result.Metadata["audio_files_resolved"] != 2 { + t.Fatalf("result metadata = %#v, want audio_files_resolved=2", result) + } + + aliceWork := filepath.Join(m.LocalWorkDir, "audio", "alice.flac") + bobWork := filepath.Join(m.LocalWorkDir, "audio", "bob.FLAC") + aliceSpool := filepath.Join(m.LocalSpoolDir, "alice.flac") + bobSpool := filepath.Join(m.LocalSpoolDir, "bob.FLAC") + for _, p := range []string{aliceWork, bobWork, aliceSpool, bobSpool} { + if _, err := os.Stat(p); err != nil { + t.Fatalf("expected file %q: %v", p, err) + } + } + + audioInputs := 0 + for _, in := range m.Inputs { + if in.Kind != "audio" { + continue + } + audioInputs++ + if in.Source != "s3" { + t.Fatalf("audio input source = %q, want s3", in.Source) + } + if in.S3Bucket != "my-dnd-archive" { + t.Fatalf("audio input bucket = %q", in.S3Bucket) + } + if in.S3Key == "" || in.SpoolPath == "" || in.Checksum == "" { + t.Fatalf("audio input missing provenance: %#v", in) + } + } + if audioInputs != 2 { + t.Fatalf("audio input count = %d, want 2", audioInputs) + } +} + +func TestPrepareStageS3AudioFailures(t *testing.T) { + tests := []struct { + name string + setup func(env *Env, m *manifest.Manifest, fake *storage.FakeBackend) + wantError string + }{ + { + name: "no flac files", + setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) { + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/readme.txt", Data: []byte("x")}) + }, + wantError: "no .flac files found", + }, + { + name: "list error", + setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) { + fake.ListErr = os.ErrPermission + }, + wantError: "list s3 audio objects", + }, + { + name: "download error", + setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) { + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/alice.flac", Data: []byte("alice")}) + fake.DownloadErr = os.ErrPermission + }, + wantError: "download s3 audio object", + }, + { + name: "basename collision", + setup: func(_ *Env, _ *manifest.Manifest, fake *storage.FakeBackend) { + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/a/alice.flac", Data: []byte("a")}) + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/b/alice.flac", Data: []byte("b")}) + }, + wantError: "duplicate s3 audio basename", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + env, m := setupPrepareEnv(t) + env.Config.Session.Campaign = "forsaken" + env.Config.Session.Inputs.AudioDir = "" + env.Config.Session.Inputs.AudioFiles = nil + env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"} + env.Config.Pipeline.Spool = config.SpoolConfig{Root: filepath.Join(t.TempDir(), "spool")} + env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "my-dnd-archive", RootPrefix: "dnd"} + m.RunID = "20260515T031522Z-a1b2c3d4" + m.LocalWorkDir = artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, "forsaken", m.SessionID, m.RunID) + m.LocalSpoolDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, "forsaken", m.SessionID, m.RunID) + + fake := &storage.FakeBackend{} + tt.setup(env, m, fake) + env.ObjectStore = fake + + _, err := (prepareStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), tt.wantError) { + t.Fatalf("error = %v, want %q", err, tt.wantError) + } + }) + } +} + +func TestPrepareStageAudioSourceConflictFails(t *testing.T) { + env, m := setupPrepareEnv(t) + root := filepath.Dir(env.Config.SessionPath) + writeFile(t, filepath.Join(root, "audio", "a.flac"), "a") + env.Config.Session.Inputs.AudioDir = "./audio" + env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"} + + _, err := (prepareStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "mutually exclusive") { + t.Fatalf("error = %v, want mutually exclusive error", err) + } +} + func setupPrepareEnv(t *testing.T) (*Env, *manifest.Manifest) { t.Helper() workspace := t.TempDir() @@ -170,6 +304,7 @@ func setupPrepareEnv(t *testing.T) (*Env, *manifest.Manifest) { PipelinePath: pipelinePath, Session: &config.SessionConfig{ SessionID: "2026-05-03", + Campaign: "sample-campaign", Inputs: config.SessionInputsConfig{ AudioDir: "./audio", SpeakersFile: "./speakers.yml", diff --git a/internal/stage/stage.go b/internal/stage/stage.go index 6ba4861..55ad32b 100644 --- a/internal/stage/stage.go +++ b/internal/stage/stage.go @@ -29,6 +29,7 @@ type Env struct { Scriptorium scriptorium.Runner Analyzer analyzer.Runner Storage storage.Backend + ObjectStore storage.ObjectStore Notifier notify.Sender }