diff --git a/docs/cli.md b/docs/cli.md index f48a369..0dd4c5e 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -6,7 +6,7 @@ narratio run --session-id 2026-04-04 ``` -This command uses default discovery for `pipeline.yml`, `campaign.yml`, and `session.yml`; all three files must be discoverable unless you pass explicit `--config`, `--campaign`, and `--session` paths. +This command uses default discovery for `pipeline.yml`, `campaign.yml`, and local `session.yml`. If local session discovery misses and S3 storage is configured, `--session-id` can load remote `session.yml` from the canonical session prefix. ## Command Overview @@ -110,6 +110,7 @@ Success output: Common failure cases: - missing default config/campaign/session paths when flags omitted. +- missing local session plus missing/unavailable remote `session.yml`. - invalid template/rendered session mismatch. - unknown/invalid `--artifacts` value. - `--artifacts` with unknown configured artifact key. @@ -132,6 +133,7 @@ Success output includes: Common failure cases: - same config/campaign/session discovery and validation failures as `run`. +- remote session fallback failures when local session discovery misses. - secrets directory read failures when `pipeline.secrets.env_dir` is configured. ### `resume` diff --git a/docs/config.md b/docs/config.md index 5af8fd5..c82b54d 100644 --- a/docs/config.md +++ b/docs/config.md @@ -6,7 +6,7 @@ Narratio loads three YAML files: - `pipeline.yml`: pipeline-level runtime configuration. - `campaign.yml`: stable campaign identity and campaign-level input defaults. -- `session.yml`: per-session metadata and input selection. +- `session.yml`: per-session metadata and input selection, loaded locally or from the configured S3 backend. These commands load and validate all three files before running: @@ -20,6 +20,7 @@ Behavior: - strict YAML decode is enabled (`KnownFields(true)`): unknown fields fail. - session templates render before session YAML decode. +- remote `session.yml` uses the same strict decode and template behavior as local `session.yml`. - defaults are applied for optional pipeline fields. - campaign-level stable input paths fill missing session input paths. - session-level stable input paths override campaign-level input paths. @@ -49,11 +50,14 @@ Campaign config lookup for `run`, `plan`, `resume`, `run-stage`, and `restore`: Session config lookup for `run`, `plan`, `resume`, `run-stage`, and `restore`: - if `--session ` is provided, that path is used. -- if omitted, Narratio searches in order: +- if `--session` is omitted, Narratio searches locally in order: 1. `./session.yml` 2. `/usr/local/etc/narratio/session.yml` 3. `/etc/narratio/session.yml` -- first existing file wins. +- first existing local file wins. +- if no local session file is found, `--session-id ` is present, storage is configured, and campaign identity is resolved, Narratio loads remote `session.yml` from: + - `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/session.yml` +- local discovery always runs before remote fallback. Template behavior: diff --git a/docs/internal/artifacts.md b/docs/internal/artifacts.md index 87c1b97..b24717c 100644 --- a/docs/internal/artifacts.md +++ b/docs/internal/artifacts.md @@ -16,6 +16,7 @@ Outputs: - resolved artifact path + provenance (`ResolvedSessionArtifact`); - runtime catalog entries for built-ins and configured artifacts; - requirement sets for canonical previous-session inputs. +- canonical S3 session, run, current, session config, audio, and promoted artifact keys. ## Boundaries Owns: @@ -43,6 +44,14 @@ Does not own: - configured artifact: `narratio.artifact.` - canonical previous-session artifact: `narratio.previous_session.artifact.` +## S3 key helpers +- session prefix: `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/` +- session config: `{session_prefix}/session.yml` +- run prefix: `{session_prefix}/runs/{run_id}/` +- audio prefix: `{session_prefix}/{session.inputs.audio_s3.prefix}` +- current manifest: `{session_prefix}/current/manifest.json` +- current run pointer: `{session_prefix}/current/run_id.txt` + ## Runtime catalog model Catalog entries track: - `planned`: source is registered for this run; diff --git a/docs/internal/stage-prepare.md b/docs/internal/stage-prepare.md index 3ee8823..6becd7c 100644 --- a/docs/internal/stage-prepare.md +++ b/docs/internal/stage-prepare.md @@ -11,6 +11,7 @@ Prepare owns: ## Inputs and outputs Inputs: - resolved config/campaign/session (`pipeline.yml`, `campaign.yml`, `session.yml`); +- remote session provenance when `session.yml` was loaded from S3; - campaign or session input files (`speakers`, `autocorrect`, `glossary`); - audio source: - local: `session.inputs.audio_dir` or `session.inputs.audio_files`; @@ -72,6 +73,7 @@ Does not own: ## State and manifest behavior - Ensures workspace layout exists. - Materializes canonical input files and audio files. +- Records `inputs/session.yml` provenance as local `session_config` or remote `session_config.s3`. - Resolves campaign-provided stable input paths relative to `campaign.yml`. - Resolves session-provided stable input overrides relative to `session.yml`. - Scans enabled configured artifact inputs for canonical sources: diff --git a/docs/internal/storage.md b/docs/internal/storage.md index fc777c3..ceafa87 100644 --- a/docs/internal/storage.md +++ b/docs/internal/storage.md @@ -6,7 +6,7 @@ Document Narratio's remote storage backend contracts and implementations under ` ## Inputs and outputs Inputs: - Resolved storage config (`pipeline.storage.*`). -- Bucket-relative object keys and local file paths from stage/app orchestration. +- Bucket-relative object keys and local file paths from app/stage orchestration. Outputs: - Listed/downloaded/uploaded object metadata (`ObjectInfo`). @@ -57,6 +57,7 @@ Implementations: - `S3Backend` constructor fails when required bucket is missing or AWS client setup fails. - CRUD operations return contextual errors (including not-found behavior via `Exists`). - Key normalization is applied before operations (`\\` to `/`, leading slash trimmed). +- Remote session loading uses `List` to find the exact `session.yml` key and `Download` to materialize it to a local temp file. ## Tests to inspect before changing - `internal/adapters/storage/factory_test.go` diff --git a/docs/operations.md b/docs/operations.md index 14aeb6c..8c1bd4a 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -6,19 +6,21 @@ For field-level configuration, see [docs/config.md](./config.md). For full comma ## Normal workflow (S3-first path) -1. Upload session `.flac` files to object storage under the configured session audio prefix. -2. Run Narratio: +1. Upload `session.yml` to the configured session prefix, or keep a local `session.yml` available. +2. Upload session `.flac` files to object storage under the configured session audio prefix. +3. Run Narratio: ```bash narratio run --session-id 2026-04-04 ``` -3. Read success output: +4. Read success output: - `narratio run: session ; executed= skipped=; manifest=` - use `manifest=` with `status` for inspection. Notes: - default config/campaign/session discovery applies unless `--config`, `--campaign`, and `--session` are passed. +- when local `session.yml` discovery misses, `--session-id` loads remote `session.yml` from `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/session.yml`. - S3 audio mode requires `session.inputs.audio_s3.prefix` and valid object-store access. ## Restore workflow diff --git a/docs/roadmap/remote.md b/docs/roadmap/remote.md index 9d287ab..44ea4ea 100644 --- a/docs/roadmap/remote.md +++ b/docs/roadmap/remote.md @@ -2,13 +2,13 @@ ## Purpose -This roadmap describes planned, unimplemented work for three related feature areas: +This roadmap tracks implementation work for three related feature areas: 1. `campaign.yml` configuration for stable campaign-level inputs. 2. Remote `session.yml` loading from the existing S3 object-store backend. 3. Logical archive locks that prevent selected top-level transcript/artifact promotions from overwriting curated archive state while still preserving run-local outputs. -Treat this document as an implementation plan, not as current behavior. Keep planned behavior under `docs/roadmap/` until each phase is implemented and canonical docs are updated. +Treat future sections of this document as an implementation plan, not as current behavior. Keep planned behavior under `docs/roadmap/` until each item is implemented and canonical docs are updated. ## Current Code Facts @@ -43,7 +43,7 @@ Implementation must preserve these constraints: - Keep raw secrets out of configs, manifests, logs, generated configs, archive metadata, and roadmap examples. - Update canonical user-facing docs only after behavior is implemented. -## Phase 1: Add `campaign.yml` +## Phase 1: Add `campaign.yml` (implemented) Add campaign-level configuration for stable campaign identity and stable input files. Do not add remote campaign loading in this phase. @@ -116,7 +116,7 @@ Update `prepare` to materialize the resolved campaign/session inputs into canoni Continue recording deterministic `manifest.Inputs` records with checksums. If a prepared input came from `campaign.yml`, record source/provenance using the existing manifest input fields where practical; add narrow metadata only if the existing fields cannot describe it. -## Phase 2: Load Remote `session.yml` +## Phase 2: Load Remote `session.yml` (implemented) Support running with no local session file when a remote session file exists under the canonical session prefix. diff --git a/internal/adapters/storage/fake_test.go b/internal/adapters/storage/fake_test.go index 22a1bd2..e60b3c8 100644 --- a/internal/adapters/storage/fake_test.go +++ b/internal/adapters/storage/fake_test.go @@ -56,7 +56,7 @@ func TestFakeBackendDownload(t *testing.T) { fake.SeedObject(FakeObject{Key: "audio/a.flac", Data: []byte("audio-a")}) dst := filepath.Join(t.TempDir(), "nested", "a.flac") - if err := fake.Download(context.Background(), "audio/a.flac", dst); err != nil { + if err := fake.Download(context.Background(), `audio\a.flac`, dst); err != nil { t.Fatalf("Download() error = %v", err) } data, err := os.ReadFile(dst) diff --git a/internal/app/config_loader.go b/internal/app/config_loader.go new file mode 100644 index 0000000..f2ae819 --- /dev/null +++ b/internal/app/config_loader.go @@ -0,0 +1,133 @@ +package app + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +func loadCommandConfig(ctx context.Context, pipelineFlag, campaignFlag, sessionFlag string, sessionOpts config.SessionLoadOptions) (*config.Config, error) { + resolvedPipelinePath, err := resolvePipelineConfigPath(pipelineFlag) + if err != nil { + return nil, err + } + resolvedCampaignPath, err := resolveCampaignConfigPath(campaignFlag) + if err != nil { + return nil, err + } + + if explicitSession := strings.TrimSpace(sessionFlag); explicitSession != "" { + return config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, explicitSession, sessionOpts) + } + + discoveredSession, err := discoverSessionConfigPathWithCandidates(config.DefaultSessionConfigSearchPaths) + if err != nil { + return nil, err + } + if discoveredSession.Path != "" { + return config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, discoveredSession.Path, sessionOpts) + } + + pipelineCfg, err := config.LoadPipeline(resolvedPipelinePath) + if err != nil { + return nil, err + } + campaignCfg, err := config.LoadCampaign(resolvedCampaignPath) + if err != nil { + return nil, err + } + + sessionID := strings.TrimSpace(sessionOpts.SessionID) + if sessionID == "" { + return nil, missingSessionConfigError(discoveredSession.Searched, "remote session loading requires --session-id") + } + + sessionPrefix := artifacts.S3SessionPrefix(pipelineCfg.Storage.S3.RootPrefix, campaignCfg.Campaign, sessionID) + remoteKey := artifacts.S3SessionConfigKey(sessionPrefix) + partialCfg := &config.Config{ + Pipeline: pipelineCfg, + Campaign: campaignCfg, + PipelinePath: resolvedPipelinePath, + CampaignPath: resolvedCampaignPath, + } + store, err := newObjectStoreFromConfigFn(ctx, partialCfg) + if err != nil { + return nil, missingSessionConfigError(discoveredSession.Searched, fmt.Sprintf("remote session %q unavailable: %v", remoteKey, err)) + } + + sessionInfo, err := findRemoteSessionConfig(ctx, store, sessionPrefix, remoteKey) + if err != nil { + return nil, missingSessionConfigError(discoveredSession.Searched, err.Error()) + } + sessionTempPath, err := downloadRemoteSessionConfig(ctx, store, remoteKey) + if err != nil { + return nil, missingSessionConfigError(discoveredSession.Searched, fmt.Sprintf("remote session %q download failed: %v", remoteKey, err)) + } + sessionBytes, err := os.ReadFile(sessionTempPath) + if err != nil { + return nil, fmt.Errorf("read downloaded remote session %q: %w", sessionTempPath, err) + } + sessionCfg, err := config.LoadSessionBytesWithOptions("s3://"+s3BucketName(pipelineCfg)+"/"+remoteKey, sessionBytes, sessionOpts) + if err != nil { + return nil, err + } + + return config.Resolve( + resolvedPipelinePath, + pipelineCfg, + resolvedCampaignPath, + campaignCfg, + sessionTempPath, + sessionCfg, + config.SessionSource{ + Source: "session_config.s3", + LocalPath: sessionTempPath, + S3Bucket: s3BucketName(pipelineCfg), + S3Key: remoteKey, + S3Size: sessionInfo.Size, + S3ETag: sessionInfo.ETag, + SpoolPath: sessionTempPath, + }, + ) +} + +func findRemoteSessionConfig(ctx context.Context, store storage.ObjectStore, sessionPrefix, remoteKey string) (storage.ObjectInfo, error) { + objects, err := store.List(ctx, sessionPrefix) + if err != nil { + return storage.ObjectInfo{}, fmt.Errorf("remote session %q list failed: %w", remoteKey, err) + } + for _, obj := range objects { + if obj.Key == remoteKey { + return obj, nil + } + } + return storage.ObjectInfo{}, fmt.Errorf("remote session %q not found", remoteKey) +} + +func downloadRemoteSessionConfig(ctx context.Context, store storage.ObjectStore, remoteKey string) (string, error) { + f, err := os.CreateTemp("", "narratio-session-*.yml") + if err != nil { + return "", fmt.Errorf("create temp file: %w", err) + } + path := f.Name() + if err := f.Close(); err != nil { + return "", fmt.Errorf("close temp file %q: %w", path, err) + } + if err := store.Download(ctx, remoteKey, path); err != nil { + return "", err + } + return filepath.Clean(path), nil +} + +func s3BucketName(cfg *config.PipelineConfig) string { + if cfg == nil || cfg.Storage.S3 == nil { + return "" + } + return strings.TrimSpace(cfg.Storage.S3.Bucket) +} diff --git a/internal/app/plan.go b/internal/app/plan.go index 0b7f1e2..04e8457 100644 --- a/internal/app/plan.go +++ b/internal/app/plan.go @@ -38,20 +38,7 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { if fs.NArg() != 0 { return fmt.Errorf("plan: unexpected positional arguments") } - resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath) - if err != nil { - return fmt.Errorf("plan: %w", err) - } - resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath) - if err != nil { - return fmt.Errorf("plan: %w", err) - } - resolvedSessionPath, err := resolveSessionConfigPath(sessionPath) - if err != nil { - return fmt.Errorf("plan: %w", err) - } - - cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{ + cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{ SessionID: sessionID, PreviousSessionID: previousSessionID, }) diff --git a/internal/app/remote_session_test.go b/internal/app/remote_session_test.go new file mode 100644 index 0000000..1a7087e --- /dev/null +++ b/internal/app/remote_session_test.go @@ -0,0 +1,204 @@ +package app + +import ( + "bytes" + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +func TestExecuteRemoteSessionFallbackLoadsFromObjectStore(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + remoteKey := seedRemoteSessionConfig(t, fake, "2026-05-03", `session_id: "{{ session_id }}" +inputs: + audio_s3: + prefix: audio/ +`) + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if storeInitCalls != 1 { + t.Fatalf("object store init calls = %d, want 1", storeInitCalls) + } + if !strings.Contains(stdout.String(), "narratio plan: workdir prepared") { + t.Fatalf("stdout = %q, want plan output", stdout.String()) + } + if _, ok := fake.Objects[remoteKey]; !ok { + t.Fatalf("remote session key %q was not seeded", remoteKey) + } +} + +func TestExecuteExplicitLocalSessionPrecedenceSkipsRemote(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--session-id", "2026-05-03"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if storeInitCalls != 0 { + t.Fatalf("object store init calls = %d, want 0", storeInitCalls) + } +} + +func TestExecuteLocalSessionDiscoveryPrecedenceSkipsRemote(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, config.DefaultSessionConfigSearchPaths) + + originalWD, err := os.Getwd() + if err != nil { + t.Fatalf("Getwd(): %v", err) + } + if err := os.Chdir(filepath.Dir(sessionPath)); err != nil { + t.Fatalf("Chdir(%q): %v", filepath.Dir(sessionPath), err) + } + t.Cleanup(func() { _ = os.Chdir(originalWD) }) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if storeInitCalls != 0 { + t.Fatalf("object store init calls = %d, want 0", storeInitCalls) + } +} + +func TestExecuteRemoteSessionMissingObjectFailsClearly(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + var storeInitCalls int + missingSessionPath := filepath.Join(t.TempDir(), "session.yml") + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{missingSessionPath}) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr) + if code == 0 { + t.Fatal("exit code = 0, want non-zero") + } + if !strings.Contains(stderr.String(), "remote session") || !strings.Contains(stderr.String(), "session.yml") || !strings.Contains(stderr.String(), "not found") { + t.Fatalf("stderr = %q, want remote session not found context", stderr.String()) + } + if !strings.Contains(stderr.String(), missingSessionPath) { + t.Fatalf("stderr = %q, want local searched path", stderr.String()) + } +} + +func TestExecuteRemoteSessionRequiresSessionID(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) + var storeInitCalls int + restoreAppConfigTestGlobals(t, &storage.FakeBackend{}, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath}, &stdout, &stderr) + if code == 0 { + t.Fatal("exit code = 0, want non-zero") + } + if !strings.Contains(stderr.String(), "remote session loading requires --session-id") { + t.Fatalf("stderr = %q, want session-id guidance", stderr.String()) + } + if storeInitCalls != 0 { + t.Fatalf("object store init calls = %d, want 0", storeInitCalls) + } +} + +func TestExecuteRemoteSessionStorageInitErrorFailsClearly(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) + origStoreFn := newObjectStoreFromConfigFn + origSessionDefaults := append([]string(nil), config.DefaultSessionConfigSearchPaths...) + config.DefaultSessionConfigSearchPaths = []string{filepath.Join(t.TempDir(), "session.yml")} + newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) { + return nil, errors.New("storage unavailable") + } + t.Cleanup(func() { + newObjectStoreFromConfigFn = origStoreFn + config.DefaultSessionConfigSearchPaths = origSessionDefaults + }) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr) + if code == 0 { + t.Fatal("exit code = 0, want non-zero") + } + if !strings.Contains(stderr.String(), "storage unavailable") || !strings.Contains(stderr.String(), "remote session") { + t.Fatalf("stderr = %q, want remote storage context", stderr.String()) + } +} + +func TestExecuteRemoteSessionMalformedYAMLFailsStrictDecode(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + seedRemoteSessionConfig(t, fake, "2026-05-03", "session_id: 2026-05-03\nunknown: true\n") + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"plan", "--config", pipelinePath, "--campaign", campaignPath, "--session-id", "2026-05-03"}, &stdout, &stderr) + if code == 0 { + t.Fatal("exit code = 0, want non-zero") + } + if !strings.Contains(stderr.String(), "strict decode failed") { + t.Fatalf("stderr = %q, want strict decode context", stderr.String()) + } +} + +func restoreAppConfigTestGlobals(t *testing.T, fake *storage.FakeBackend, storeInitCalls *int, sessionDefaults []string) { + t.Helper() + origStoreFn := newObjectStoreFromConfigFn + origSessionDefaults := append([]string(nil), config.DefaultSessionConfigSearchPaths...) + config.DefaultSessionConfigSearchPaths = append([]string(nil), sessionDefaults...) + newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) { + if storeInitCalls != nil { + (*storeInitCalls)++ + } + return fake, nil + } + t.Cleanup(func() { + newObjectStoreFromConfigFn = origStoreFn + config.DefaultSessionConfigSearchPaths = origSessionDefaults + }) +} + +func seedRemoteSessionConfig(t *testing.T, fake *storage.FakeBackend, sessionID, content string) string { + t.Helper() + sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", sessionID) + remoteKey := artifacts.S3SessionConfigKey(sessionPrefix) + fake.SeedObject(storage.FakeObject{ + Key: remoteKey, + Data: []byte(content), + ETag: "remote-session-etag", + }) + return remoteKey +} diff --git a/internal/app/restore.go b/internal/app/restore.go index 0d1bbb2..402d7ed 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -57,20 +57,7 @@ func Restore(ctx context.Context, args []string, out io.Writer) error { if fs.NArg() != 0 { return fmt.Errorf("restore: unexpected positional arguments") } - resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath) - if err != nil { - return fmt.Errorf("restore: %w", err) - } - resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath) - if err != nil { - return fmt.Errorf("restore: %w", err) - } - resolvedSessionPath, err := resolveSessionConfigPath(sessionPath) - if err != nil { - return fmt.Errorf("restore: %w", err) - } - - cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{ + cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{ SessionID: sessionID, PreviousSessionID: previousSessionID, }) diff --git a/internal/app/resume.go b/internal/app/resume.go index e137181..f09b45f 100644 --- a/internal/app/resume.go +++ b/internal/app/resume.go @@ -37,20 +37,7 @@ func Resume(ctx context.Context, args []string, out io.Writer) error { if fs.NArg() != 0 { return fmt.Errorf("resume: unexpected positional arguments") } - resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath) - if err != nil { - return fmt.Errorf("resume: %w", err) - } - resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath) - if err != nil { - return fmt.Errorf("resume: %w", err) - } - resolvedSessionPath, err := resolveSessionConfigPath(sessionPath) - if err != nil { - return fmt.Errorf("resume: %w", err) - } - - cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{ + cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{ SessionID: sessionID, PreviousSessionID: previousSessionID, }) diff --git a/internal/app/run.go b/internal/app/run.go index 2e0d2b5..410b6f9 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -35,20 +35,7 @@ func Run(ctx context.Context, args []string, out io.Writer) error { if fs.NArg() != 0 { return fmt.Errorf("run: unexpected positional arguments") } - resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath) - if err != nil { - return fmt.Errorf("run: %w", err) - } - resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath) - if err != nil { - return fmt.Errorf("run: %w", err) - } - resolvedSessionPath, err := resolveSessionConfigPath(sessionPath) - if err != nil { - return fmt.Errorf("run: %w", err) - } - - cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{ + cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{ SessionID: sessionID, PreviousSessionID: previousSessionID, }) diff --git a/internal/app/run_stage.go b/internal/app/run_stage.go index 0e1e400..808479e 100644 --- a/internal/app/run_stage.go +++ b/internal/app/run_stage.go @@ -48,20 +48,7 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error { return fmt.Errorf("run-stage: %w", err) } - resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath) - if err != nil { - return fmt.Errorf("run-stage: %w", err) - } - resolvedCampaignPath, err := resolveCampaignConfigPath(campaignPath) - if err != nil { - return fmt.Errorf("run-stage: %w", err) - } - resolvedSessionPath, err := resolveSessionConfigPath(sessionPath) - if err != nil { - return fmt.Errorf("run-stage: %w", err) - } - - cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedCampaignPath, resolvedSessionPath, config.SessionLoadOptions{ + cfg, err := loadCommandConfig(ctx, pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{ SessionID: sessionID, PreviousSessionID: previousSessionID, }) diff --git a/internal/app/session_config_path.go b/internal/app/session_config_path.go index 5305b20..c4ecaf3 100644 --- a/internal/app/session_config_path.go +++ b/internal/app/session_config_path.go @@ -19,6 +19,22 @@ func resolveSessionConfigPathWithCandidates(flagValue string, candidates []strin return explicit, nil } + resolved, err := discoverSessionConfigPathWithCandidates(candidates) + if err != nil { + return "", err + } + if resolved.Path != "" { + return resolved.Path, nil + } + return "", missingSessionConfigError(resolved.Searched, "") +} + +type sessionConfigDiscovery struct { + Path string + Searched []string +} + +func discoverSessionConfigPathWithCandidates(candidates []string) (sessionConfigDiscovery, error) { ordered := make([]string, 0, len(candidates)) for _, raw := range candidates { path := strings.TrimSpace(raw) @@ -31,19 +47,32 @@ func resolveSessionConfigPathWithCandidates(flagValue string, candidates []strin if info.IsDir() { continue } - return filepath.Clean(path), nil + return sessionConfigDiscovery{Path: filepath.Clean(path), Searched: ordered}, nil } if errors.Is(err, os.ErrNotExist) { continue } - return "", fmt.Errorf("check default session config %q: %w", path, err) + return sessionConfigDiscovery{}, fmt.Errorf("check default session config %q: %w", path, err) } + return sessionConfigDiscovery{Searched: ordered}, nil +} + +func missingSessionConfigError(searched []string, remoteDetail string) error { + ordered := append([]string(nil), searched...) if len(ordered) == 0 { - return "", fmt.Errorf("no session config path provided and no default locations configured") + if strings.TrimSpace(remoteDetail) != "" { + return fmt.Errorf("no session config path provided and no default locations configured; %s", remoteDetail) + } + return fmt.Errorf("no session config path provided and no default locations configured") } - return "", fmt.Errorf( - "no session config path provided and no default session config found; searched: %s; pass --session to use an explicit path", + msg := fmt.Sprintf( + "no session config path provided and no default session config found; searched: %s", strings.Join(ordered, ", "), ) + if strings.TrimSpace(remoteDetail) != "" { + msg += "; " + strings.TrimSpace(remoteDetail) + } + msg += "; pass --session to use an explicit path" + return fmt.Errorf("%s", msg) } diff --git a/internal/artifacts/s3_keys.go b/internal/artifacts/s3_keys.go index fceb63e..f86c0e3 100644 --- a/internal/artifacts/s3_keys.go +++ b/internal/artifacts/s3_keys.go @@ -34,6 +34,12 @@ func S3AudioPrefix(sessionPrefix, audioPrefix string) string { return ensureS3TrailingSlash(key) } +// S3SessionConfigKey returns the session config key. +// Format: {session_prefix}/session.yml +func S3SessionConfigKey(sessionPrefix string) string { + return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "session.yml") +} + // S3CurrentManifestKey returns the current manifest pointer key. // Format: {session_prefix}/current/manifest.json func S3CurrentManifestKey(sessionPrefix string) string { diff --git a/internal/artifacts/s3_keys_test.go b/internal/artifacts/s3_keys_test.go index 5a382e5..3946cb2 100644 --- a/internal/artifacts/s3_keys_test.go +++ b/internal/artifacts/s3_keys_test.go @@ -17,6 +17,11 @@ func TestS3KeyConstruction(t *testing.T) { t.Fatalf("audioPrefix = %q", audioPrefix) } + sessionConfigKey := S3SessionConfigKey(`dnd\campaigns\forsaken\sessions\2026-04-19\`) + if sessionConfigKey != "dnd/campaigns/forsaken/sessions/2026-04-19/session.yml" { + t.Fatalf("session config key = %q", sessionConfigKey) + } + runPrefix := S3RunPrefix(sessionPrefix, runID) wantRunPrefix := "dnd/campaigns/forsaken/sessions/2026-04-19/runs/" + runID + "/" if runPrefix != wantRunPrefix { diff --git a/internal/config/config.go b/internal/config/config.go index 29586cb..b6ef96d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -10,7 +10,8 @@ type Config struct { CampaignPath string SessionPath string - StableInputs ResolvedStableInputs + StableInputs ResolvedStableInputs + SessionSource SessionSource } // PipelineConfig contains durable pipeline-level settings. @@ -262,3 +263,14 @@ type ResolvedInputFile struct { ConfigPath string Source string } + +// SessionSource records where session.yml came from before materialization. +type SessionSource struct { + Source string + LocalPath string + S3Bucket string + S3Key string + S3Size int64 + S3ETag string + SpoolPath string +} diff --git a/internal/config/load.go b/internal/config/load.go index 20303b8..9523a1c 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -49,20 +49,25 @@ func LoadSessionWithOptions(path string, opts SessionLoadOptions) (*SessionConfi if err != nil { return nil, fmt.Errorf("load session config: session file %q: open: %w", path, err) } + return LoadSessionBytesWithOptions(path, sessionBytes, opts) +} - rendered, err := renderSessionTemplate(string(sessionBytes), opts) +// LoadSessionBytesWithOptions loads session configuration from YAML bytes with +// strict field checking after template rendering. +func LoadSessionBytesWithOptions(label string, data []byte, opts SessionLoadOptions) (*SessionConfig, error) { + rendered, err := renderSessionTemplate(string(data), opts) if err != nil { return nil, fmt.Errorf("load session config: %w", err) } var cfg SessionConfig - if err := decodeStrictYAMLFromReader("session", path, strings.NewReader(rendered), &cfg); err != nil { + if err := decodeStrictYAMLFromReader("session", label, strings.NewReader(rendered), &cfg); err != nil { return nil, fmt.Errorf("load session config: %w", err) } if strings.TrimSpace(opts.SessionID) != "" && strings.TrimSpace(cfg.SessionID) != "" && strings.TrimSpace(cfg.SessionID) != strings.TrimSpace(opts.SessionID) { return nil, fmt.Errorf( "load session config: session file %q: session_id mismatch: --session-id %q does not match rendered session_id %q", - path, + label, strings.TrimSpace(opts.SessionID), strings.TrimSpace(cfg.SessionID), ) @@ -72,7 +77,7 @@ func LoadSessionWithOptions(path string, opts SessionLoadOptions) (*SessionConfi strings.TrimSpace(cfg.PreviousSessionID) != strings.TrimSpace(opts.PreviousSessionID) { return nil, fmt.Errorf( "load session config: session file %q: previous_session_id mismatch: --previous-session-id %q does not match rendered previous_session_id %q", - path, + label, strings.TrimSpace(opts.PreviousSessionID), strings.TrimSpace(cfg.PreviousSessionID), ) @@ -109,19 +114,35 @@ func LoadWithSessionOptions(pipelinePath, campaignPath, sessionPath string, sess return nil, err } + return Resolve(pipelinePath, pipelineCfg, campaignPath, campaignCfg, sessionPath, sessionCfg, SessionSource{ + Source: "session_config", + LocalPath: sessionPath, + }) +} + +// Resolve builds final stage-facing configuration from already loaded +// pipeline, campaign, and session documents. +func Resolve(pipelinePath string, pipelineCfg *PipelineConfig, campaignPath string, campaignCfg *CampaignConfig, sessionPath string, sessionCfg *SessionConfig, sessionSource SessionSource) (*Config, error) { stableInputs, err := mergeCampaignSession(campaignCfg, sessionCfg, campaignPath, sessionPath) if err != nil { return nil, err } + if strings.TrimSpace(sessionSource.Source) == "" { + sessionSource.Source = "session_config" + } + if strings.TrimSpace(sessionSource.LocalPath) == "" { + sessionSource.LocalPath = sessionPath + } return &Config{ - Pipeline: pipelineCfg, - Campaign: campaignCfg, - Session: sessionCfg, - PipelinePath: pipelinePath, - CampaignPath: campaignPath, - SessionPath: sessionPath, - StableInputs: stableInputs, + Pipeline: pipelineCfg, + Campaign: campaignCfg, + Session: sessionCfg, + PipelinePath: pipelinePath, + CampaignPath: campaignPath, + SessionPath: sessionPath, + StableInputs: stableInputs, + SessionSource: sessionSource, }, nil } diff --git a/internal/config/session_template_test.go b/internal/config/session_template_test.go index cb490f5..afb94db 100644 --- a/internal/config/session_template_test.go +++ b/internal/config/session_template_test.go @@ -241,3 +241,38 @@ inputs: t.Fatalf("SessionID = %q, want 2026-05-03", cfg.SessionID) } } + +func TestLoadSessionBytesWithOptionsUsesSameTemplateAndStrictDecode(t *testing.T) { + sessionYAML := []byte(`session_id: "{{ session_id }}" +campaign: sample-campaign +inputs: + audio_s3: + prefix: audio/ +`) + + cfg, err := LoadSessionBytesWithOptions("s3://bucket/session.yml", sessionYAML, SessionLoadOptions{SessionID: "2026-05-03"}) + if err != nil { + t.Fatalf("LoadSessionBytesWithOptions() error = %v", err) + } + if cfg.SessionID != "2026-05-03" { + t.Fatalf("SessionID = %q, want 2026-05-03", cfg.SessionID) + } + + _, err = LoadSessionBytesWithOptions("s3://bucket/session.yml", []byte("session_id: 2026-05-03\nunknown: true\n"), SessionLoadOptions{}) + if err == nil { + t.Fatal("expected strict decode error, got nil") + } + if !strings.Contains(err.Error(), "strict decode failed") { + t.Fatalf("error = %q, want strict decode context", err.Error()) + } +} + +func TestLoadSessionBytesWithOptionsMismatchFails(t *testing.T) { + _, err := LoadSessionBytesWithOptions("s3://bucket/session.yml", []byte("session_id: 2026-05-03\n"), SessionLoadOptions{SessionID: "2026-04-04"}) + if err == nil { + t.Fatal("expected mismatch error, got nil") + } + if !strings.Contains(err.Error(), "session_id mismatch") { + t.Fatalf("error = %q, want mismatch context", err.Error()) + } +} diff --git a/internal/stage/prepare.go b/internal/stage/prepare.go index e08556d..7eb1269 100644 --- a/internal/stage/prepare.go +++ b/internal/stage/prepare.go @@ -113,6 +113,23 @@ func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S registerConfigInput := func(kind, path, checksum, source string) { inputs = append(inputs, manifest.InputRecord{Kind: kind, Path: path, Checksum: checksum, Source: source}) } + registerSessionConfigInput := func(path, checksum string) { + source := env.Config.SessionSource + if strings.TrimSpace(source.Source) == "" { + source.Source = "session_config" + } + inputs = append(inputs, manifest.InputRecord{ + Kind: "session_config", + Path: path, + Checksum: checksum, + Source: source.Source, + S3Bucket: source.S3Bucket, + S3Key: source.S3Key, + S3Size: source.S3Size, + S3ETag: source.S3ETag, + SpoolPath: source.SpoolPath, + }) + } campaignDst := filepath.Join(paths.InputsDir, "campaign.yml") campaignChecksum, err := copyFileIfChanged(env.ArtifactStore, campaignSrc, campaignDst) @@ -126,7 +143,7 @@ func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err != nil { return nil, fmt.Errorf("prepare: materialize session.yml: %w", err) } - registerConfigInput("session_config", sessionDst, sessionChecksum, "session_config") + registerSessionConfigInput(sessionDst, sessionChecksum) pipelineResolvedBytes, err := renderResolvedPipeline(env.Config.Pipeline) if err != nil { diff --git a/internal/stage/prepare_test.go b/internal/stage/prepare_test.go index f8a8114..568f9fa 100644 --- a/internal/stage/prepare_test.go +++ b/internal/stage/prepare_test.go @@ -152,6 +152,64 @@ func TestPrepareStageIdempotent(t *testing.T) { } } +func TestPrepareStageRecordsLocalSessionProvenance(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.AudioFiles = []string{"./audio/a.flac"} + + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + sessionInput := findManifestInput(t, m.Inputs, "session_config") + if sessionInput.Source != "session_config" { + t.Fatalf("session source = %q, want session_config", sessionInput.Source) + } + if sessionInput.S3Bucket != "" || sessionInput.S3Key != "" || sessionInput.SpoolPath != "" { + t.Fatalf("local session input has unexpected remote provenance: %#v", sessionInput) + } +} + +func TestPrepareStageRecordsRemoteSessionProvenance(t *testing.T) { + env, m := setupPrepareEnv(t) + remoteSessionPath := filepath.Join(t.TempDir(), "downloaded-session.yml") + writeFile(t, remoteSessionPath, "session_id: 2026-05-03\ninputs:\n audio_s3:\n prefix: audio/\n") + env.Config.SessionPath = remoteSessionPath + env.Config.Session.Inputs.AudioDir = "" + env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"} + env.Config.SessionSource = config.SessionSource{ + Source: "session_config.s3", + LocalPath: remoteSessionPath, + S3Bucket: "my-dnd-archive", + S3Key: "dnd/campaigns/sample-campaign/sessions/2026-05-03/session.yml", + S3Size: 58, + S3ETag: "session-etag", + SpoolPath: remoteSessionPath, + } + 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.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, "sample-campaign", m.SessionID, m.RunID) + m.LocalSpoolDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, "sample-campaign", m.SessionID, m.RunID) + fake := &storage.FakeBackend{} + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/sample-campaign/sessions/2026-05-03/audio/alice.flac", Data: []byte("alice")}) + env.ObjectStore = fake + + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + sessionInput := findManifestInput(t, m.Inputs, "session_config") + if sessionInput.Source != "session_config.s3" { + t.Fatalf("session source = %q, want session_config.s3", sessionInput.Source) + } + if sessionInput.S3Bucket != "my-dnd-archive" || sessionInput.S3Key != "dnd/campaigns/sample-campaign/sessions/2026-05-03/session.yml" { + t.Fatalf("remote session input missing bucket/key: %#v", sessionInput) + } + if sessionInput.S3Size != 58 || sessionInput.S3ETag != "session-etag" || sessionInput.SpoolPath != remoteSessionPath { + t.Fatalf("remote session input missing metadata: %#v", sessionInput) + } +} + func TestPrepareStageS3AudioDownloadAndMaterialization(t *testing.T) { env, m := setupPrepareEnv(t) env.Config.Session.Campaign = "forsaken" @@ -543,3 +601,14 @@ func snapshotInputs(inputs []manifest.InputRecord) map[string]string { } return out } + +func findManifestInput(t *testing.T, inputs []manifest.InputRecord, kind string) manifest.InputRecord { + t.Helper() + for _, input := range inputs { + if input.Kind == kind { + return input + } + } + t.Fatalf("manifest input kind %q not found in %#v", kind, inputs) + return manifest.InputRecord{} +}