From 228c348e4278599f5873c94f798a0c8a736a39bb Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 21 May 2026 11:50:20 -0500 Subject: [PATCH] Implemented operations helper commands for validation, locking, and status --- docs/cli.md | 143 +++- docs/config.md | 11 +- docs/internal/artifacts.md | 3 +- docs/internal/stage-archive.md | 7 +- docs/operations.md | 41 +- docs/roadmap/operations.md | 53 ++ docs/roadmap/remote.md | 317 --------- internal/app/commands.go | 12 +- internal/app/operator_helpers.go | 870 ++++++++++++++++++++++++ internal/app/operator_helpers_test.go | 195 ++++++ internal/app/remote_locks.go | 157 +++++ internal/app/runner.go | 35 +- internal/app/status.go | 67 -- internal/artifacts/s3_keys.go | 6 + internal/artifacts/s3_keys_test.go | 5 + internal/config/config.go | 5 + internal/config/load.go | 27 + internal/config/storage_archive_test.go | 48 ++ internal/config/validate.go | 59 +- 19 files changed, 1653 insertions(+), 408 deletions(-) create mode 100644 docs/roadmap/operations.md delete mode 100644 docs/roadmap/remote.md create mode 100644 internal/app/operator_helpers.go create mode 100644 internal/app/operator_helpers_test.go create mode 100644 internal/app/remote_locks.go delete mode 100644 internal/app/status.go diff --git a/docs/cli.md b/docs/cli.md index 4dd1436..400a934 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -17,9 +17,15 @@ Implemented commands: - `run`: execute pipeline stages and persist manifest state. - `plan`: validate config, prepare workspace layout, and print stage run/skip decisions. - `resume`: continue from first non-succeeded stage unless forced. -- `status`: read and print stage statuses from an existing manifest. +- `status`: read an existing manifest or inspect local/remote state for a session. - `run-stage`: execute exactly one stage. - `restore`: restore durable local session state from the committed remote archive state. +- `session validate`: run read-only preflight checks for a session. +- `session init`: create local or remote `session.yml`. +- `artifacts list`: list effective artifact source IDs. +- `locks`: list effective archive promotion locks. +- `lock`: add or update a remote session lock. +- `unlock`: remove a remote session lock. Unknown commands print usage and exit non-zero. @@ -92,7 +98,55 @@ Valid stage names: ### `status` -- `--manifest `: required manifest path. +- `--manifest `: inspect one manifest file. +- `--config ` +- `--campaign ` +- `--session ` +- `--session-id ` +- `--previous-session-id ` + +### `session validate` + +- `--config ` +- `--campaign ` +- `--session ` +- `--session-id ` +- `--previous-session-id ` + +### `session init` + +- `--config `: required. +- `--campaign `: required. +- `--session-id `: required. +- `--output `: local `session.yml` target; mutually exclusive with `--remote`. +- `--remote`: write remote `session.yml` to the canonical session prefix; mutually exclusive with `--output`. +- `--previous-session-id ` +- `--date ` +- `--title ` +- `--audio-s3-prefix `: defaults to `audio/` when neither audio flag is provided. +- `--audio-dir `: local audio directory; mutually exclusive with `--audio-s3-prefix`. +- `--force`: overwrite existing local or remote target. + +### `artifacts list` + +- `--config ` +- `--campaign ` +- `--session ` +- `--session-id ` +- `--previous-session-id ` +- `--remote`: check promoted remote object availability. + +### `locks`, `lock`, `unlock` + +- `--config ` +- `--campaign ` +- `--session ` +- `--session-id ` +- `--previous-session-id ` +- `lock ` positional source ID. +- `lock --reason ` optional remote lock reason. +- `lock --force` updates an existing remote lock. +- `unlock ` positional source ID. ## Command Reference @@ -161,22 +215,101 @@ Common failure cases: ### `status` Purpose: -- Inspect one manifest file without executing stages. +- Inspect one manifest file, or inspect configured local/remote state for a session. Syntax: ```bash narratio status --manifest +narratio status [--config ] [--campaign ] [--session ] [--session-id ] [--previous-session-id ] ``` -Success output includes: +Manifest output includes: - `session_id: ` - `updated_at: ` - `stages:` entries (`- : `) +Session output includes: +- session ID, campaign, workspace, session config source. +- local manifest state when present. +- remote current archive state when storage is configured. +- effective archive locks and conservative next actions. + Common failure cases: -- missing `--manifest`. +- missing `--manifest` when no config/session flags are provided. - unreadable or invalid manifest path. +- invalid config or remote session fallback failure in session mode. + +### `session validate` + +Purpose: +- Run read-only preflight checks for a session. + +Syntax: + +```bash +narratio session validate [--config ] [--campaign ] [--session ] [--session-id ] [--previous-session-id ] +``` + +Checks include: +- effective config and session source. +- stable input files. +- local or remote audio availability. +- previous-session requirements. +- archive promotions and effective locks. + +Warnings do not fail the command. Any `ERROR` finding exits non-zero. + +### `session init` + +Purpose: +- Create a strict-decoded session skeleton locally or in object storage. + +Syntax: + +```bash +narratio session init --config --campaign --session-id --output ./session.yml +narratio session init --config --campaign --session-id --remote +``` + +Behavior: +- exactly one of `--output` or `--remote` is required. +- remote writes target `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/session.yml`. +- existing local or remote targets fail unless `--force` is passed. +- remote writes use existence checks, not compare-and-swap. + +### `artifacts list` + +Purpose: +- List built-in, configured, previous-session, promoted, and locked artifact sources. + +Syntax: + +```bash +narratio artifacts list [--config ] [--campaign ] [--session ] [--session-id ] [--previous-session-id ] [--remote] +``` + +`--remote` checks promoted top-level object availability through the storage adapter. + +### `locks`, `lock`, and `unlock` + +Purpose: +- Inspect and mutate source-based archive promotion locks. + +Syntax: + +```bash +narratio locks [--config ] [--campaign ] [--session ] [--session-id ] +narratio lock [flags] +narratio unlock [flags] +``` + +Behavior: +- static locks from `pipeline.archive.locks` and remote locks from `{session_prefix}/locks.yml` are merged. +- static locks win when sources duplicate remote locks. +- `lock` writes or updates only remote locks. +- `unlock` removes only remote locks and cannot remove static pipeline locks. +- `lock --force` is required to update an existing remote lock reason. ### `run-stage` diff --git a/docs/config.md b/docs/config.md index 3cf37a3..1951bef 100644 --- a/docs/config.md +++ b/docs/config.md @@ -237,7 +237,8 @@ Operational notes: - archive promotion is explicit and source-based via `archive.promote_artifacts`. - `source` is required; `dest` is optional and derived when omitted. -- `archive.locks` skips top-level promotion overwrites for locked sources while preserving run-local uploads. +- `archive.locks` skips top-level promotion overwrites for static locked sources while preserving run-local uploads. +- operator-created mutable locks are stored at `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/locks.yml` and are merged with static locks. - Narratio does not auto-promote all generated analyze artifacts. - `restore` reads the same config/campaign/session inputs and restore scope is bounded by committed archive current state. @@ -384,10 +385,18 @@ Archive lock rules: - locks are source-based and do not accept `dest`. - duplicate lock sources are rejected. +- static `pipeline.archive.locks` win over remote mutable locks for the same source. - locked promotions are recorded as intentional skips in archive metadata. - locked required promotions do not fail archive by default. - ordinary `--force` reruns do not override locks. +Remote mutable lock store: + +- path: `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/locks.yml`. +- strict YAML shape: top-level `locks`, each with `source` and optional `reason`. +- `narratio lock` and `narratio unlock` mutate only the remote lock store. +- writes use existence checks plus `--force` for updates; they are not compare-and-swap atomic. + Restore-related implications: - restore remote identity requires archive S3 identity to resolve (`pipeline.storage.s3.bucket` and session prefix derivation inputs). diff --git a/docs/internal/artifacts.md b/docs/internal/artifacts.md index 4c1fbe0..e241c14 100644 --- a/docs/internal/artifacts.md +++ b/docs/internal/artifacts.md @@ -16,7 +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. +- canonical S3 session, run, current, session config, session locks, audio, and promoted artifact keys. ## Boundaries Owns: @@ -47,6 +47,7 @@ Does not own: ## S3 key helpers - session prefix: `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/` - session config: `{session_prefix}/session.yml` +- session lock store: `{session_prefix}/locks.yml` - run prefix: `{session_prefix}/runs/{run_id}/` - audio prefix: `{session_prefix}/{session.inputs.audio_s3.prefix}` - current manifest: `{session_prefix}/current/manifest.json` diff --git a/docs/internal/stage-archive.md b/docs/internal/stage-archive.md index fa41cf7..b4425fd 100644 --- a/docs/internal/stage-archive.md +++ b/docs/internal/stage-archive.md @@ -8,7 +8,7 @@ Inputs: - session manifest and prerequisite stage records - run root contents under `runs/{run_id}/` - promotion rules with artifact `source` IDs and archive `dest` paths (`archive.promote_artifacts`) -- source-based promotion locks (`archive.locks`) +- effective source-based promotion locks from static config and remote session lock store - session-level `previous/**` cache files when present Outputs: @@ -37,6 +37,7 @@ Does not own: - `pipeline.archive.upload_run` - `pipeline.archive.promote_artifacts` - `pipeline.archive.locks` +- `{session_prefix}/locks.yml` loaded by app orchestration before archive execution - `pipeline.storage.s3.bucket` - `pipeline.storage.s3.root_prefix` - `pipeline.workspace.root` @@ -50,7 +51,8 @@ Does not own: - Requires `prepare`, `transcribe`, `merge`, `polish`, `normalize`, `trim`, and `analyze` status `succeeded`. - Resolves bucket/prefix from manifest identity first, then config fallback. - Uploads session `previous/**` files as durable session state when the local `previous/` directory exists. -- Skips top-level promotion uploads for sources listed in `pipeline.archive.locks`; run-local uploads still publish. +- Skips top-level promotion uploads for effective locked sources; run-local uploads still publish. +- Effective locks are the union of `pipeline.archive.locks` and remote `{session_prefix}/locks.yml`; static pipeline locks win on duplicate sources. - Writes metadata including: - upload counts/paths - `previous_files_uploaded` and `previous_uploaded_paths` @@ -77,5 +79,6 @@ Does not own: - Run upload excludes `audio/` subtree. - Session `previous/**` is archiveable durable input/provenance state, not run-local output. - Ordinary `--force` does not override archive locks. +- Malformed or unreadable remote lock store fails archive-capable execution before promotion. - `current/manifest.json` uploads before `current/run_id.txt`. - `current/run_id.txt` is the remote publish commit marker. diff --git a/docs/operations.md b/docs/operations.md index 238713b..67b6328 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -6,7 +6,7 @@ For field-level configuration, see [docs/config.md](./config.md). For full comma ## Normal workflow (S3-first path) -1. Upload `session.yml` to the configured session prefix, or pass a local `session.yml` explicitly. +1. Create or upload `session.yml`, or pass a local `session.yml` explicitly. 2. Upload session `.flac` files to object storage under the configured session audio prefix. 3. Run Narratio: @@ -23,6 +23,20 @@ Notes: - 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. +Initialize a remote session skeleton: + +```bash +narratio session init --config /etc/narratio/pipeline.yml --campaign /etc/narratio/campaign.yml --session-id 2026-04-04 --remote +``` + +Remote init writes `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/session.yml`. It fails if the object already exists unless `--force` is passed. + +Validate before running: + +```bash +narratio session validate --session-id 2026-04-04 +``` + ## Restore workflow Use restore when local durable session state is missing or stale and archive current state is authoritative. @@ -119,6 +133,7 @@ When archive is enabled and run upload is enabled, archive publishes under: Archive uploads: - run record files from run root (excluding `audio/`). - promoted files from explicit `archive.promote_artifacts` rules. +- mutable session locks from helper commands live at `{session_prefix}/locks.yml`. Publish order: 1. upload `current/manifest.json` @@ -133,8 +148,17 @@ Archive promotion is explicit and source-based: - missing optional promotion sources are skipped. - invalid resolved artifacts fail archive stage. - `archive.locks` skips top-level promotion overwrites for locked sources while run-local uploads still publish. +- remote locks from `{session_prefix}/locks.yml` are merged with static `archive.locks`; static locks win on duplicate sources. - locked required promotions are treated as intentional successful skips and are recorded in archive metadata. +Lock helper behavior: +- `narratio locks --session-id ` lists effective static and remote locks. +- `narratio lock --session-id --reason ` writes or updates a remote lock. +- `narratio unlock --session-id ` removes only a remote lock. +- `lock --force` is required to update an existing remote lock reason. +- `unlock` cannot remove static pipeline locks. +- remote lock writes check whether the lock store exists, but are not compare-and-swap atomic. + ## Resume, retry, restore, and safe rerun behavior Default skip: @@ -191,18 +215,24 @@ Recommended recovery: 1. inspect state: +```bash +narratio status --session-id 2026-04-04 +``` + +2. for one manifest file, run: + ```bash narratio status --manifest ``` -2. for restore-specific checks, run: +3. for restore-specific checks, run: ```bash narratio restore --session-id 2026-04-04 --dry-run ``` -3. fix root cause (config/input/credentials/storage/service availability). -4. continue with `resume`, or targeted `run-stage --force` followed by `resume`. +4. fix root cause (config/input/credentials/storage/service availability). +5. continue with `resume`, or targeted `run-stage --force` followed by `resume`. ## Restore report @@ -219,7 +249,8 @@ Dry-run does not write restore report files. ## Operational caveats -- `status` requires explicit `--manifest`; there is no session-id lookup command. +- `status` with no config/session flags still requires explicit `--manifest`. +- `status --session-id ` uses normal config/session loading, including remote session fallback. - local and S3 audio input modes are mutually exclusive. - archive publish requires upstream stages through `analyze` to be `succeeded`. - required promotion rules can fail when selected analyze artifacts did not generate a required file path. diff --git a/docs/roadmap/operations.md b/docs/roadmap/operations.md new file mode 100644 index 0000000..14a13b6 --- /dev/null +++ b/docs/roadmap/operations.md @@ -0,0 +1,53 @@ +# Roadmap: Operator Helper Commands + +## Status + +Implemented. + +The operator helper command set is no longer conceptual. Current behavior is documented in: + +- `docs/cli.md` +- `docs/operations.md` +- `docs/config.md` +- `docs/internal/artifacts.md` +- `docs/internal/stage-archive.md` + +## Implemented Commands + +- `narratio session validate` +- `narratio status --manifest ` +- `narratio status --session-id ` +- `narratio session init --output ` +- `narratio session init --remote` +- `narratio artifacts list` +- `narratio artifacts list --remote` +- `narratio locks` +- `narratio lock ` +- `narratio unlock ` + +## Implemented Decisions + +- Helper output is text-only. No JSON schema exists yet. +- `status` remains a top-level command. +- `session validate`, `session init`, and `artifacts list` are nested helper commands. +- `lock`, `unlock`, and `locks` are top-level commands. +- Remote session initialization requires explicit `--remote`. +- Local session initialization requires `--output`. +- Remote artifact availability is opt-in with `artifacts list --remote`. +- Mutable locks are source-based and stored at `{session_prefix}/locks.yml`. +- The remote lock store uses strict YAML with top-level `locks`. +- Static `pipeline.archive.locks` and remote locks are merged; static locks win on duplicate sources. +- `unlock` removes only remote locks. +- Ordinary execution `--force` does not override locks. +- Remote lock writes use existence checks and `--force` for updates; there is no compare-and-swap protection. + +## Remaining Future Enhancements + +These are intentionally not implemented: + +- `--json` output for helper commands. +- Optimistic concurrency or ETag compare-and-swap for remote lock mutations. +- Rich remote artifact availability across historical run-local objects. +- Session-lock acquisition for remote mutation helpers. +- Broader campaign helper commands such as `campaign validate` or `campaign publish`. + diff --git a/docs/roadmap/remote.md b/docs/roadmap/remote.md deleted file mode 100644 index 3a916c1..0000000 --- a/docs/roadmap/remote.md +++ /dev/null @@ -1,317 +0,0 @@ -# Roadmap: Campaign State, Remote Sessions, and Archive Locks - -## Purpose - -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 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 - -The current codebase already settles several design choices: - -- CLI commands use short noun flags: `--config`, `--session`, `--session-id`, `--previous-session-id`, `--force`, and `--artifacts`. -- `run-stage` uses flags before the positional stage name, for example: - - narratio run-stage --config ./pipeline.yml --campaign ./campaign.yml --session ./session.yml prepare - -- `session.yml` is represented by `config.SessionConfig` and currently owns `session_id`, `previous_session_id`, `campaign`, `date`, `title`, and `inputs`. -- Strict YAML decoding is already implemented with `yaml.Decoder.KnownFields(true)`. -- Default local config discovery uses system paths under `/usr/local/etc/narratio/` and `/etc/narratio/`; working-directory files are used only when passed explicitly. -- The canonical S3 session prefix is already: - - {root_prefix}/campaigns/{campaign}/sessions/{session_id}/ - -- Archive promotion is already source-based through `archive.promote_artifacts[].source`, with destination derivation and validation in `internal/config`. -- Storage adapters receive bucket-relative keys and do not infer campaign, session, run, or root-prefix semantics. - -## Guardrails - -Keep Narratio explicit and stage-driven. Do not introduce a generic workflow engine, broad config language, or stage behavior that reaches through adapter boundaries. - -Implementation must preserve these constraints: - -- Keep storage details behind `internal/adapters/storage`. -- Compute session, campaign, archive, and remote config keys in app/artifact/path helpers, not inside storage implementations. -- Use centralized path helpers in `internal/artifacts` or the established local path model. -- Preserve manifest-driven resume and stage status semantics. -- Keep strict YAML decoding for `pipeline.yml`, `campaign.yml`, and `session.yml`. -- 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` (implemented) - -Add campaign-level configuration for stable campaign identity and stable input files. Do not add remote campaign loading in this phase. - -### CLI and Discovery - -Add `--campaign ` to `run`, `plan`, `resume`, `run-stage`, and `restore`. - -Examples: - - narratio run --campaign ./campaign.yml --session ./session.yml - narratio plan --campaign ./campaign.yml --session ./session.yml - narratio resume --campaign ./campaign.yml --session ./session.yml - narratio run-stage --campaign ./campaign.yml --session ./session.yml prepare - narratio restore --campaign ./campaign.yml --session ./session.yml - -Campaign config discovery order: - -1. explicit `--campaign `; -2. `/usr/local/etc/narratio/campaign.yml`; -3. `/etc/narratio/campaign.yml`. - -Implement this in the same style as `resolvePipelineConfigPath` and `resolveSessionConfigPath`. Add default path constants and a search-path variable in `internal/config/defaults.go`. - -### Config Shape - -Initial `campaign.yml` fields: - - campaign: icewind-dale - inputs: - speakers_file: ./speakers.yml - autocorrect_file: ./autocorrect.yml - glossary_file: ./glossary.yml - -Do not add speculative campaign artifact defaults, prompt defaults, or title conventions in the first implementation. - -### Merge Behavior - -Add `CampaignConfig` and keep the final stage-facing config explicit. - -Required behavior: - -- `pipeline.yml` remains host/runtime configuration. -- `campaign.yml` supplies campaign identity and stable input file defaults. -- `session.yml` remains the source for `session_id`, `previous_session_id`, `date`, `title`, and audio input. -- Campaign-level `speakers_file`, `autocorrect_file`, and `glossary_file` fill missing session-level stable input fields. -- Session-level stable input fields override campaign-level stable input fields. -- If both `campaign.yml` and `session.yml` specify `campaign`, the values must match. -- The resolved session must satisfy the existing session validation rules before stages run. -- Unknown fields in `campaign.yml` fail strict decode. - -Path resolution must preserve source-file locality: - -- campaign-provided stable input paths resolve relative to `campaign.yml`; -- session-provided stable input overrides resolve relative to `session.yml`; -- absolute paths keep existing behavior. - -Track enough provenance in the resolved config or prepare inputs so `prepare` can copy the correct source files without guessing which file supplied each path. - -### Prepare Behavior - -Update `prepare` to materialize the resolved campaign/session inputs into canonical session input paths: - - inputs/campaign.yml - inputs/session.yml - inputs/pipeline.resolved.yml - inputs/speakers.yml - inputs/autocorrect.yml - inputs/glossary.yml - -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` (implemented) - -Support running with no local session file when a remote session file exists under the canonical session prefix. - -### Preconditions - -Build this phase after `campaign.yml`, because campaign identity is required to compute the remote session key. Do not infer campaign identity from object-store listing. - -### Loading Precedence - -Session loading order: - -1. If `--session ` is supplied, load that local file. -2. If `--session` is omitted, use existing local discovery: `/usr/local/etc/narratio/session.yml`, `/etc/narratio/session.yml`. -3. If no local session file is found, `--session-id` is present, storage is configured, and campaign identity is resolved, load remote `session.yml`. -4. If no local or remote session can be loaded, fail with a message that lists the local search paths and the remote key that was attempted when applicable. - -Do not make remote loading mask local discovery. Existing local discovery remains the local fallback before remote is attempted. Once remote loading is attempted, a missing remote object, storage init error, or malformed remote YAML fails clearly because no local session was available. - -### Remote Key Layout - -Use the existing canonical S3 layout: - - session prefix: {root_prefix}/campaigns/{campaign}/sessions/{session_id}/ - session file: {session_prefix}/session.yml - audio prefix: {session_prefix}/{session.inputs.audio_s3.prefix} - -Add a centralized helper near `internal/artifacts/s3_keys.go`: - - S3SessionConfigKey(sessionPrefix string) string - -The helper should return `{session_prefix}/session.yml` using the same key normalization style as `S3CurrentManifestKey`, `S3CurrentRunPointerKey`, and `S3PromotedArtifactKey`. - -### Decode, Template, and Provenance - -Remote `session.yml` uses the same template variables and mismatch checks as local sessions: - -- `{{session_id}}` -- `{{ session_id }}` -- `{{previous_session_id}}` -- `{{ previous_session_id }}` - -Decode remote session YAML with strict known-field validation. Reuse the current session template/render/decode path by adding a byte/string-based loader rather than duplicating YAML decode logic. - -When `prepare` materializes a remote session into `inputs/session.yml`, record that it came from S3. Preserve useful non-secret provenance when available: - -- bucket; -- key; -- ETag; -- size; -- local checksum; -- downloaded temp/materialized path. - -## Phase 3: Add Logical Archive Locks (implemented) - -Narratio supports source-based archive locks under `pipeline.archive.locks`. The promotion system is source-based; destination-based locks are not supported. - -### Config Shape - -Lock entries: - - archive: - locks: - - source: narratio.transcript.polished - reason: Human-reviewed transcript; do not overwrite automatically. - - source: narratio.artifact.session_recap - reason: Final recap was manually edited. - -Validation rules: - -- `source` is required. -- `source` must be a built-in source ID or configured `narratio.artifact.` accepted by the same source validation used for `promote_artifacts`. -- `reason` is optional and non-secret. -- duplicate lock sources fail validation. -- lock entries do not support `dest` in the first implementation; unknown fields already fail strict decode. - -### Archive Behavior - -Archive continues uploading complete run-local outputs under `runs/{run_id}/`. - -Promotion behavior: - -1. Resolve promotion source and destination using existing source-based promotion logic. -2. If the promotion source is unlocked, upload the top-level promoted object normally. -3. If the promotion source is locked, skip only the top-level promotion overwrite. -4. Treat locked required promotions as intentional successful skips by default. -5. Continue archive commit when all run-local uploads and all non-locked required promotions succeed. -6. Upload `current/manifest.json` and `current/run_id.txt` in the existing order, with `current/run_id.txt` last. - -Ordinary `--force` does not override locks. A lock-break override remains out of scope. - -### Metadata - -Archive records locked promotion skips in archive metadata/reporting so operators can distinguish missing optional promotions from lock-protected promotions. - -Include: - -- source ID; -- destination relative path and remote key; -- reason; -- local resolved path; -- resolved provenance; -- whether the original promotion rule was required. - -Keep existing metadata such as `promoted_paths`, `skipped_optional_promotions`, `current_manifest_key`, `current_run_id_key`, and `current_pointer_written`. - -## Phase 4: Future Operator Helpers - -These commands are future work only. Do not implement them with the first campaign, remote-session, or lock changes. - -Potential helper shapes: - - narratio session validate --session-id 2026-06-07 - narratio session init --session-id 2026-06-07 --title "The Black Cabin" - narratio status --session-id 2026-06-07 - narratio locks --session-id 2026-06-07 - narratio lock narratio.artifact.session_recap --session-id 2026-06-07 - narratio unlock narratio.artifact.session_recap --session-id 2026-06-07 - -Potential behavior: - -- validate remote session config; -- check audio object availability; -- inspect committed remote current state; -- list promoted transcripts/artifacts; -- list archive lock status; -- initialize a remote session skeleton; -- publish or sync campaign assets. - -## Implementation Sequence - -Use small, reviewable commits. - -1. Campaign config types and discovery: - add `CampaignConfig`, strict loading, defaults/search paths, `--campaign` flags, and config/app tests. -2. Campaign/session merge: - implement resolved stable input merge, path provenance, validation, and prepare materialization. -3. Campaign docs after implementation: - update canonical docs and examples only for implemented behavior. -4. Remote session key and loader: - add `S3SessionConfigKey`, byte/string session loading, remote download through `ObjectStore`, and app-level precedence tests. -5. Remote session prepare provenance: - materialize downloaded session config and record S3 provenance. -6. Remote session docs after implementation: - update canonical docs and examples only after behavior exists. -7. Archive lock config: - add lock config structs, strict decode coverage, source validation, and duplicate detection. -8. Archive lock enforcement: - skip locked top-level promotions, preserve run-local uploads, record lock metadata, and protect commit ordering. -9. Final sweep: - run focused tests, then `go test ./...`; verify planned behavior remains only in roadmap docs until implemented. - -## Test Plan - -Add focused coverage in these packages: - -- `internal/config`: campaign load, strict decode, discovery constants, merge validation, campaign/session mismatch, lock validation, duplicate lock rejection. -- `internal/app`: `--campaign` parsing on `run`, `plan`, `resume`, `run-stage`, and `restore`; campaign discovery; explicit `--session` precedence; local discovery before remote; remote session fallback when local discovery misses. -- `internal/artifacts`: `S3SessionConfigKey`; canonical session prefix compatibility; source ID validation for lock sources. -- `internal/stage/prepare`: campaign/session stable input materialization; campaign-relative and session-relative path resolution; remote session provenance in `manifest.Inputs`. -- `internal/stage/archive`: locked required promotion succeeds as skipped; unlocked promotion uploads; run-local outputs upload when top-level promotion is locked; `--force` does not break locks; `current/run_id.txt` remains the last upload. -- `internal/adapters/storage`: fake object key normalization and remote session download expectations. - -Run at least: - - go test ./internal/config -v - go test ./internal/app -run TestExecute -v - go test ./internal/artifacts -v - go test ./internal/stage -run 'Prepare|Archive' -v - go test ./internal/adapters/storage -v - go test ./... - -Use fake storage for remote-session and archive-lock behavior. Ordinary tests must not require live S3. - -## Documentation Updates After Implementation - -After each phase is implemented, update only docs for behavior that exists. - -Likely files: - -- `docs/config.md` -- `docs/cli.md` -- `docs/operations.md` -- `docs/internal/stage-prepare.md` -- `docs/internal/stage-archive.md` -- `docs/internal/storage.md` -- `docs/internal/artifacts.md` -- relevant examples under `examples/` - -Do not document remote campaign loading, helper commands, or lock override flags as current behavior until implemented. - -## Remaining Open Decisions - -The codebase resolves the campaign flag name, campaign discovery order, remote session layout, local-vs-remote session precedence, source-based archive lock model, and locked required promotion policy. - -Remaining decisions: - -1. Whether `campaign.yml` should eventually be loadable from S3. Do not implement remote campaign loading in the first phase. -2. Whether a future explicit lock override command or flag is needed. Do not make ordinary `--force` break locks. -3. Whether future helper commands should be top-level commands or subcommands. Keep them out of the first implementation. diff --git a/internal/app/commands.go b/internal/app/commands.go index a92043f..1b2a9a5 100644 --- a/internal/app/commands.go +++ b/internal/app/commands.go @@ -7,7 +7,7 @@ import ( "strings" ) -var supportedCommands = []string{"run", "plan", "status", "resume", "run-stage", "restore"} +var supportedCommands = []string{"run", "plan", "status", "resume", "run-stage", "restore", "session", "artifacts", "locks", "lock", "unlock"} // Execute dispatches CLI commands and returns a process exit code. func Execute(args []string, stdout, stderr io.Writer) int { @@ -34,6 +34,16 @@ func Execute(args []string, stdout, stderr io.Writer) int { err = RunStage(ctx, cmdArgs, stdout) case "restore": err = Restore(ctx, cmdArgs, stdout) + case "session": + err = Session(ctx, cmdArgs, stdout) + case "artifacts": + err = Artifacts(ctx, cmdArgs, stdout) + case "locks": + err = Locks(ctx, cmdArgs, stdout) + case "lock": + err = Lock(ctx, cmdArgs, stdout) + case "unlock": + err = Unlock(ctx, cmdArgs, stdout) default: fmt.Fprintf(stderr, "unknown command: %q\n\n", cmd) printUsage(stderr) diff --git a/internal/app/operator_helpers.go b/internal/app/operator_helpers.go new file mode 100644 index 0000000..369cd35 --- /dev/null +++ b/internal/app/operator_helpers.go @@ -0,0 +1,870 @@ +package app + +import ( + "context" + "flag" + "fmt" + "io" + "os" + "path/filepath" + "regexp" + "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" + "gopkg.in/yaml.v3" +) + +type commonConfigFlags struct { + pipelinePath string + campaignPath string + sessionPath string + sessionID string + previousSessionID string +} + +type finding struct { + Severity string + Category string + Message string +} + +type findingError struct { + count int +} + +func (e findingError) Error() string { + return fmt.Sprintf("%d validation error(s)", e.count) +} + +func addCommonConfigFlags(fs *flag.FlagSet, flags *commonConfigFlags) { + fs.StringVar(&flags.pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)") + fs.StringVar(&flags.campaignPath, "campaign", "", "path to campaign.yml (optional; defaults searched)") + fs.StringVar(&flags.sessionPath, "session", "", "path to session.yml") + fs.StringVar(&flags.sessionID, "session-id", "", "session identifier for session.yml templates") + fs.StringVar(&flags.previousSessionID, "previous-session-id", "", "previous session identifier for session.yml templates") +} + +func (f commonConfigFlags) sessionOptions() config.SessionLoadOptions { + return config.SessionLoadOptions{ + SessionID: f.sessionID, + PreviousSessionID: f.previousSessionID, + } +} + +// Session dispatches session helper subcommands. +func Session(ctx context.Context, args []string, out io.Writer) error { + if len(args) == 0 { + return fmt.Errorf("session: expected subcommand: validate|init") + } + switch args[0] { + case "validate": + return SessionValidate(ctx, args[1:], out) + case "init": + return SessionInit(ctx, args[1:], out) + default: + return fmt.Errorf("session: unknown subcommand %q", args[0]) + } +} + +// Artifacts dispatches artifact helper subcommands. +func Artifacts(ctx context.Context, args []string, out io.Writer) error { + if len(args) == 0 { + return fmt.Errorf("artifacts: expected subcommand: list") + } + switch args[0] { + case "list": + return ArtifactsList(ctx, args[1:], out) + default: + return fmt.Errorf("artifacts: unknown subcommand %q", args[0]) + } +} + +// SessionValidate performs a read-only session preflight. +func SessionValidate(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("session validate", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + addCommonConfigFlags(fs, &flags) + if err := fs.Parse(args); err != nil { + return fmt.Errorf("session validate: invalid flags: %w", err) + } + if fs.NArg() != 0 { + return fmt.Errorf("session validate: unexpected positional arguments") + } + + findings := []finding{} + cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.sessionPath, flags.sessionOptions()) + if err != nil { + findings = append(findings, errorFinding("config", err.Error())) + return renderFindings(out, "", "", findings) + } + if err := config.Validate(cfg); err != nil { + findings = append(findings, errorFinding("config", err.Error())) + } else { + findings = append(findings, okFinding("config", "resolved pipeline, campaign, and session config")) + } + findings = append(findings, okFinding("session", fmt.Sprintf("session source: %s", sessionSourceSummary(cfg)))) + + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + findings = append(findings, validateStableInputFindings(cfg)...) + findings = append(findings, validateLocalAudioFindings(cfg)...) + + store, storeErr := objectStoreIfConfigured(ctx, cfg) + if storeErr != nil { + findings = append(findings, errorFinding("storage", storeErr.Error())) + } + if cfg.Session.Inputs.AudioS3 != nil { + if storeErr != nil { + findings = append(findings, errorFinding("audio", "remote audio cannot be checked because storage is unavailable")) + } else { + findings = append(findings, validateRemoteAudioFinding(ctx, cfg, store)) + } + } + + requirements := artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) + if len(requirements) == 0 { + findings = append(findings, okFinding("previous", "no previous-session artifacts required")) + } else if strings.TrimSpace(cfg.Session.PreviousSessionID) == "" { + findings = append(findings, errorFinding("previous", "previous_session_id is required by configured previous-session artifacts")) + } else if storeErr != nil { + findings = append(findings, errorFinding("previous", "previous-session artifacts cannot be checked because storage is unavailable")) + } else { + findings = append(findings, validatePreviousArtifactFindings(ctx, cfg, store, requirements)...) + } + + locks, lockErr := loadEffectiveLocks(ctx, cfg, store) + if lockErr != nil { + findings = append(findings, errorFinding("locks", lockErr.Error())) + } else if len(locks.All) == 0 { + findings = append(findings, okFinding("locks", "no effective archive locks")) + } else { + for _, lock := range locks.All { + findings = append(findings, warnFinding("locks", fmt.Sprintf("%s locked: %s", lock.Source, strings.TrimSpace(lock.Reason)))) + } + } + if paths.ManifestPath != "" { + findings = append(findings, infoFinding("workspace", "manifest path: "+paths.ManifestPath)) + } + return renderFindings(out, cfg.Session.Campaign, cfg.Session.SessionID, findings) +} + +// Status reports either a requested manifest or effective local/remote session state. +func Status(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("status", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var manifestPath string + var flags commonConfigFlags + fs.StringVar(&manifestPath, "manifest", "", "path to manifest.json") + addCommonConfigFlags(fs, &flags) + if err := fs.Parse(args); err != nil { + return fmt.Errorf("status: invalid flags: %w", err) + } + if fs.NArg() != 0 { + return fmt.Errorf("status: unexpected positional arguments") + } + if strings.TrimSpace(manifestPath) != "" { + return statusManifest(ctx, manifestPath, out) + } + if flags.pipelinePath == "" && flags.campaignPath == "" && flags.sessionPath == "" && flags.sessionID == "" && flags.previousSessionID == "" { + return fmt.Errorf("status: --manifest is required") + } + cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.sessionPath, flags.sessionOptions()) + if err != nil { + return fmt.Errorf("status: %w", err) + } + if err := config.Validate(cfg); err != nil { + return fmt.Errorf("status: %w", err) + } + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + fmt.Fprintf(out, "Session: %s\n", cfg.Session.SessionID) + fmt.Fprintf(out, "Campaign: %s\n", cfg.Session.Campaign) + fmt.Fprintf(out, "Workspace: %s\n", paths.Root) + fmt.Fprintf(out, "Session config: %s\n", sessionSourceSummary(cfg)) + + if m, err := loadLocalManifest(ctx, paths.ManifestPath); err != nil { + fmt.Fprintf(out, "Local manifest: error: %v\n", err) + } else if m == nil { + fmt.Fprintln(out, "Local manifest: missing") + } else { + fmt.Fprintf(out, "Local manifest: %s\n", paths.ManifestPath) + writeStageStatuses(out, m) + } + + store, storeErr := objectStoreIfConfigured(ctx, cfg) + if storeErr != nil { + fmt.Fprintf(out, "Remote archive: unavailable: %v\n", storeErr) + } else if store != nil { + current, err := discoverRemoteCurrentStateFn(ctx, cfg, store) + if err != nil { + fmt.Fprintf(out, "Remote archive: missing or unavailable: %v\n", err) + } else { + fmt.Fprintf(out, "Remote archive: current run %s\n", current.RunID) + fmt.Fprintf(out, "Remote manifest: %s\n", current.CurrentManifestKey) + } + } + + locks, err := loadEffectiveLocks(ctx, cfg, store) + if err != nil { + fmt.Fprintf(out, "Archive locks: error: %v\n", err) + } else { + writeLocks(out, cfg, locks) + } + fmt.Fprintln(out, "Next actions:") + fmt.Fprintf(out, "- narratio session validate --session-id %s\n", cfg.Session.SessionID) + fmt.Fprintf(out, "- narratio restore --session-id %s --dry-run\n", cfg.Session.SessionID) + return nil +} + +func statusManifest(ctx context.Context, manifestPath string, out io.Writer) error { + store := &manifest.LocalStore{} + m, err := store.Load(ctx, manifestPath) + if err != nil { + return fmt.Errorf("status: %w", err) + } + if _, err := fmt.Fprintf(out, "session_id: %s\n", m.SessionID); err != nil { + return err + } + if _, err := fmt.Fprintf(out, "updated_at: %s\n", m.UpdatedAt.UTC().Format("2006-01-02T15:04:05Z07:00")); err != nil { + return err + } + writeStageStatuses(out, m) + return nil +} + +// SessionInit creates a local or remote session.yml skeleton. +func SessionInit(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("session init", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var pipelinePath, campaignPath, sessionID, previousSessionID, date, title, output, audioS3Prefix, audioDir string + var remote, force bool + fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml") + fs.StringVar(&campaignPath, "campaign", "", "path to campaign.yml") + fs.StringVar(&sessionID, "session-id", "", "session identifier") + fs.StringVar(&previousSessionID, "previous-session-id", "", "previous session identifier") + fs.StringVar(&date, "date", "", "session date") + fs.StringVar(&title, "title", "", "session title") + fs.StringVar(&output, "output", "", "local output session.yml path") + fs.StringVar(&audioS3Prefix, "audio-s3-prefix", "", "session audio S3 prefix") + fs.StringVar(&audioDir, "audio-dir", "", "local audio directory") + fs.BoolVar(&remote, "remote", false, "write session.yml to S3 session prefix") + fs.BoolVar(&force, "force", false, "overwrite existing target") + if err := fs.Parse(args); err != nil { + return fmt.Errorf("session init: invalid flags: %w", err) + } + if fs.NArg() != 0 { + return fmt.Errorf("session init: unexpected positional arguments") + } + if strings.TrimSpace(pipelinePath) == "" || strings.TrimSpace(campaignPath) == "" || strings.TrimSpace(sessionID) == "" { + return fmt.Errorf("session init: --config, --campaign, and --session-id are required") + } + if (strings.TrimSpace(output) == "") == !remote { + return fmt.Errorf("session init: specify exactly one target: --output or --remote") + } + if strings.TrimSpace(audioDir) != "" && strings.TrimSpace(audioS3Prefix) != "" { + return fmt.Errorf("session init: --audio-dir and --audio-s3-prefix are mutually exclusive") + } + + resolvedPipeline, err := resolvePipelineConfigPath(pipelinePath) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + resolvedCampaign, err := resolveCampaignConfigPath(campaignPath) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + pipelineCfg, err := config.LoadPipeline(resolvedPipeline) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + campaignCfg, err := config.LoadCampaign(resolvedCampaign) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + + data, err := buildSessionYAML(campaignCfg.Campaign, sessionID, previousSessionID, date, title, audioS3Prefix, audioDir) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + label := strings.TrimSpace(output) + if label == "" { + label = "remote session.yml" + } + sessionCfg, err := config.LoadSessionBytesWithOptions(label, data, config.SessionLoadOptions{ + SessionID: sessionID, + PreviousSessionID: previousSessionID, + }) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + cfg, err := config.Resolve(resolvedPipeline, pipelineCfg, resolvedCampaign, campaignCfg, label, sessionCfg, config.SessionSource{Source: "session_config", LocalPath: label}) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + if err := config.Validate(cfg); err != nil { + return fmt.Errorf("session init: %w", err) + } + + if !remote { + if err := writeLocalFile(output, data, force); err != nil { + return fmt.Errorf("session init: %w", err) + } + _, err := fmt.Fprintf(out, "narratio session init: wrote %s\n", filepath.Clean(output)) + return err + } + + store, err := newObjectStoreFromConfigFn(ctx, cfg) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + sessionPrefix := artifacts.S3SessionPrefix(pipelineCfg.Storage.S3.RootPrefix, campaignCfg.Campaign, sessionID) + key := artifacts.S3SessionConfigKey(sessionPrefix) + exists, err := store.Exists(ctx, key) + if err != nil { + return fmt.Errorf("session init: check remote session %q: %w", key, err) + } + if exists && !force { + return fmt.Errorf("session init: remote session %q already exists; pass --force to overwrite", key) + } + tmp, err := os.CreateTemp("", "narratio-session-init-*.yml") + if err != nil { + return fmt.Errorf("session init: create temp file: %w", err) + } + tmpPath := tmp.Name() + defer func() { _ = os.Remove(tmpPath) }() + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return fmt.Errorf("session init: write temp file: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("session init: close temp file: %w", err) + } + if _, err := store.Upload(ctx, tmpPath, key, storage.UploadOptions{ContentType: "application/x-yaml; charset=utf-8"}); err != nil { + return fmt.Errorf("session init: upload remote session %q: %w", key, err) + } + _, err = fmt.Fprintf(out, "narratio session init: wrote s3://%s/%s\n", s3BucketName(pipelineCfg), key) + return err +} + +// ArtifactsList lists effective artifact sources. +func ArtifactsList(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("artifacts list", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + var remote bool + addCommonConfigFlags(fs, &flags) + fs.BoolVar(&remote, "remote", false, "inspect remote archive availability") + if err := fs.Parse(args); err != nil { + return fmt.Errorf("artifacts list: invalid flags: %w", err) + } + if fs.NArg() != 0 { + return fmt.Errorf("artifacts list: unexpected positional arguments") + } + cfg, store, locks, _, err := loadHelperContext(ctx, flags, remote) + if err != nil { + return fmt.Errorf("artifacts list: %w", err) + } + catalog, err := buildHelperArtifactCatalog(cfg) + if err != nil { + return fmt.Errorf("artifacts list: %w", err) + } + remoteState := map[string]string{} + if remote && store != nil { + remoteState = remoteArtifactAvailability(ctx, cfg, store, catalog) + } + writeArtifactList(out, cfg, catalog, locks, remoteState) + return nil +} + +// Locks lists effective archive locks. +func Locks(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("locks", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + addCommonConfigFlags(fs, &flags) + if err := fs.Parse(args); err != nil { + return fmt.Errorf("locks: invalid flags: %w", err) + } + if fs.NArg() != 0 { + return fmt.Errorf("locks: unexpected positional arguments") + } + cfg, _, locks, _, err := loadHelperContext(ctx, flags, true) + if err != nil { + return fmt.Errorf("locks: %w", err) + } + writeLocks(out, cfg, locks) + return nil +} + +// Lock adds or updates one remote lock. +func Lock(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("lock", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + var reason string + var force bool + addCommonConfigFlags(fs, &flags) + fs.StringVar(&reason, "reason", "", "lock reason") + fs.BoolVar(&force, "force", false, "update existing remote lock") + if err := fs.Parse(args); err != nil { + return fmt.Errorf("lock: invalid flags: %w", err) + } + if fs.NArg() != 1 { + return fmt.Errorf("lock: expected exactly one source id") + } + source := strings.TrimSpace(fs.Arg(0)) + cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) + if err != nil { + return fmt.Errorf("lock: %w", err) + } + if _, err := config.ValidateArchiveLockRules([]config.ArchiveLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "lock"); err != nil { + return fmt.Errorf("lock: %w", err) + } + if _, ok := lockSourceSet(locks.Static)[source]; ok { + return fmt.Errorf("lock: source %q is locked by pipeline config and cannot be modified remotely", source) + } + remoteSet := lockSourceSet(locks.Remote) + if _, exists := remoteSet[source]; exists && !force { + return fmt.Errorf("lock: remote lock for %q already exists; pass --force to update", source) + } + remoteSet[source] = config.ArchiveLockRule{Source: source, Reason: strings.TrimSpace(reason)} + remoteLocks := lockMapValues(remoteSet) + if _, err := config.ValidateArchiveLockRules(remoteLocks, cfg.Pipeline.Scriptorium, "locks"); err != nil { + return fmt.Errorf("lock: %w", err) + } + if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.ArchiveLockStore{Locks: remoteLocks}); err != nil { + return fmt.Errorf("lock: %w", err) + } + _, err = fmt.Fprintf(out, "narratio lock: locked %s\n", source) + return err +} + +// Unlock removes one remote lock. +func Unlock(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("unlock", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + addCommonConfigFlags(fs, &flags) + if err := fs.Parse(args); err != nil { + return fmt.Errorf("unlock: invalid flags: %w", err) + } + if fs.NArg() != 1 { + return fmt.Errorf("unlock: expected exactly one source id") + } + source := strings.TrimSpace(fs.Arg(0)) + cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) + if err != nil { + return fmt.Errorf("unlock: %w", err) + } + if _, err := config.ValidateArchiveLockRules([]config.ArchiveLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "unlock"); err != nil { + return fmt.Errorf("unlock: %w", err) + } + remoteSet := lockSourceSet(locks.Remote) + if _, ok := remoteSet[source]; !ok { + if _, static := lockSourceSet(locks.Static)[source]; static { + return fmt.Errorf("unlock: source %q is locked by pipeline config and cannot be unlocked remotely", source) + } + return fmt.Errorf("unlock: remote lock for %q does not exist", source) + } + delete(remoteSet, source) + remoteLocks := lockMapValues(remoteSet) + if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.ArchiveLockStore{Locks: remoteLocks}); err != nil { + return fmt.Errorf("unlock: %w", err) + } + _, err = fmt.Fprintf(out, "narratio unlock: unlocked %s\n", source) + return err +} + +func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore bool) (*config.Config, storage.ObjectStore, *effectiveLocks, *manifest.Manifest, error) { + cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.sessionPath, flags.sessionOptions()) + if err != nil { + return nil, nil, nil, nil, err + } + if err := config.Validate(cfg); err != nil { + return nil, nil, nil, nil, err + } + var store storage.ObjectStore + if needStore { + store, err = newObjectStoreFromConfigFn(ctx, cfg) + if err != nil { + return nil, nil, nil, nil, err + } + } else { + store, _ = objectStoreIfConfigured(ctx, cfg) + } + locks, err := loadEffectiveLocks(ctx, cfg, store) + if err != nil { + return nil, nil, nil, nil, err + } + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + m, err := loadLocalManifest(ctx, paths.ManifestPath) + if err != nil { + return nil, nil, nil, nil, err + } + return cfg, store, locks, m, nil +} + +func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.ObjectStore, error) { + if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Storage.S3 == nil || strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) == "" { + return nil, nil + } + store, err := newObjectStoreFromConfigFn(ctx, cfg) + if err != nil { + return nil, err + } + return store, nil +} + +func buildSessionYAML(campaign, sessionID, previousSessionID, date, title, audioS3Prefix, audioDir string) ([]byte, error) { + if strings.TrimSpace(date) == "" && regexp.MustCompile(`^\d{4}-\d{2}-\d{2}$`).MatchString(strings.TrimSpace(sessionID)) { + date = strings.TrimSpace(sessionID) + } + type audioS3 struct { + Prefix string `yaml:"prefix"` + } + type inputs struct { + AudioDir string `yaml:"audio_dir,omitempty"` + AudioS3 *audioS3 `yaml:"audio_s3,omitempty"` + } + type sessionYAML struct { + Campaign string `yaml:"campaign"` + SessionID string `yaml:"session_id"` + PreviousSessionID string `yaml:"previous_session_id,omitempty"` + Date string `yaml:"date,omitempty"` + Title string `yaml:"title,omitempty"` + Inputs inputs `yaml:"inputs"` + } + in := inputs{AudioDir: strings.TrimSpace(audioDir)} + if in.AudioDir == "" { + prefix := strings.TrimSpace(audioS3Prefix) + if prefix == "" { + prefix = "audio/" + } + in.AudioS3 = &audioS3{Prefix: prefix} + } + data, err := yaml.Marshal(sessionYAML{ + Campaign: strings.TrimSpace(campaign), + SessionID: strings.TrimSpace(sessionID), + PreviousSessionID: strings.TrimSpace(previousSessionID), + Date: strings.TrimSpace(date), + Title: strings.TrimSpace(title), + Inputs: in, + }) + if err != nil { + return nil, err + } + return data, nil +} + +func renderFindings(out io.Writer, campaign, sessionID string, findings []finding) error { + if campaign != "" || sessionID != "" { + fmt.Fprintf(out, "Campaign: %s\n", campaign) + fmt.Fprintf(out, "Session: %s\n\n", sessionID) + } + errorsCount := 0 + for _, f := range findings { + if f.Severity == "ERROR" { + errorsCount++ + } + fmt.Fprintf(out, "%-5s %-10s %s\n", f.Severity, f.Category, f.Message) + } + if errorsCount > 0 { + return findingError{count: errorsCount} + } + return nil +} + +func okFinding(category, msg string) finding { return finding{"OK", category, msg} } +func infoFinding(category, msg string) finding { return finding{"INFO", category, msg} } +func warnFinding(category, msg string) finding { return finding{"WARN", category, msg} } +func errorFinding(category, msg string) finding { return finding{"ERROR", category, msg} } + +func sessionSourceSummary(cfg *config.Config) string { + source := cfg.SessionSource.Source + if source == "" { + source = "session_config" + } + if cfg.SessionSource.S3Key != "" { + return source + " " + cfg.SessionSource.S3Key + } + return source + " " + cfg.SessionPath +} + +func validateStableInputFindings(cfg *config.Config) []finding { + items := []struct { + name string + in config.ResolvedInputFile + }{ + {"speakers", cfg.StableInputs.SpeakersFile}, + {"autocorrect", cfg.StableInputs.AutocorrectFile}, + {"glossary", cfg.StableInputs.GlossaryFile}, + } + out := make([]finding, 0, len(items)) + for _, item := range items { + path, err := resolveHelperConfigRelativePath(item.in) + if err != nil { + out = append(out, errorFinding("inputs", item.name+": "+err.Error())) + continue + } + if _, err := os.Stat(path); err != nil { + out = append(out, errorFinding("inputs", fmt.Sprintf("%s missing: %v", item.name, err))) + } else { + out = append(out, okFinding("inputs", item.name+": "+path)) + } + } + return out +} + +func resolveHelperConfigRelativePath(input config.ResolvedInputFile) (string, error) { + if strings.TrimSpace(input.ConfigPath) == "" { + return "", fmt.Errorf("source config path is required") + } + path := strings.TrimSpace(input.Path) + if path == "" { + return "", fmt.Errorf("path is required") + } + if filepath.IsAbs(path) { + return filepath.Clean(path), nil + } + return filepath.Clean(filepath.Join(filepath.Dir(input.ConfigPath), path)), nil +} + +func validateLocalAudioFindings(cfg *config.Config) []finding { + if cfg.Session.Inputs.AudioS3 != nil { + return nil + } + audioDir := strings.TrimSpace(cfg.Session.Inputs.AudioDir) + if audioDir == "" && len(cfg.Session.Inputs.AudioFiles) == 0 { + return []finding{errorFinding("audio", "audio_dir, audio_files, or audio_s3 is required")} + } + base := filepath.Dir(cfg.SessionPath) + paths := []string{} + if audioDir != "" { + dir := audioDir + if !filepath.IsAbs(dir) { + dir = filepath.Join(base, dir) + } + matches, err := filepath.Glob(filepath.Join(dir, "*.flac")) + if err != nil || len(matches) == 0 { + return []finding{errorFinding("audio", "no .flac files found in "+dir)} + } + paths = append(paths, matches...) + } + for _, file := range cfg.Session.Inputs.AudioFiles { + p := file + if !filepath.IsAbs(p) { + p = filepath.Join(base, p) + } + paths = append(paths, p) + } + for _, p := range paths { + if _, err := os.Stat(p); err != nil { + return []finding{errorFinding("audio", fmt.Sprintf("audio file missing: %v", err))} + } + } + return []finding{okFinding("audio", fmt.Sprintf("%d local audio file(s)", len(paths)))} +} + +func validateRemoteAudioFinding(ctx context.Context, cfg *config.Config, store storage.ObjectStore) finding { + sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) + audioPrefix := artifacts.S3AudioPrefix(sessionPrefix, cfg.Session.Inputs.AudioS3.Prefix) + objects, err := store.List(ctx, audioPrefix) + if err != nil { + return errorFinding("audio", err.Error()) + } + count := 0 + for _, obj := range objects { + if strings.HasSuffix(strings.ToLower(obj.Key), ".flac") { + count++ + } + } + if count == 0 { + return errorFinding("audio", "no remote .flac objects found under "+audioPrefix) + } + return okFinding("audio", fmt.Sprintf("%d remote .flac object(s)", count)) +} + +func validatePreviousArtifactFindings(ctx context.Context, cfg *config.Config, store storage.ObjectStore, requirements []artifacts.PreviousArtifactRequirement) []finding { + out := []finding{} + prefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.PreviousSessionID) + manifestKey, runIDKey := artifacts.ResolveArchiveCurrentStateKeys(prefix) + for _, key := range []string{runIDKey, manifestKey} { + exists, err := store.Exists(ctx, key) + if err != nil { + out = append(out, errorFinding("previous", fmt.Sprintf("check %s: %v", key, err))) + return out + } + if !exists { + out = append(out, errorFinding("previous", "missing "+key)) + return out + } + } + for _, req := range requirements { + out = append(out, okFinding("previous", fmt.Sprintf("%s required=%t", req.Name, req.Required))) + } + return out +} + +func loadLocalManifest(ctx context.Context, path string) (*manifest.Manifest, error) { + if _, err := os.Stat(path); err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + store := &manifest.LocalStore{} + return store.Load(ctx, path) +} + +func writeStageStatuses(out io.Writer, m *manifest.Manifest) { + if m == nil || len(m.Stages) == 0 { + fmt.Fprintln(out, "stages: no stages recorded") + return + } + fmt.Fprintln(out, "stages:") + names := make([]string, 0, len(m.Stages)) + for name := range m.Stages { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + fmt.Fprintf(out, "- %s: %s\n", name, m.Stages[name].Status) + } +} + +func buildHelperArtifactCatalog(cfg *config.Config) (*artifacts.ArtifactCatalog, error) { + catalog := artifacts.NewArtifactCatalog() + if err := catalog.RegisterBuiltIns(); err != nil { + return nil, err + } + configured := map[string]artifacts.ConfiguredArtifactDefinition{} + if cfg.Pipeline.Scriptorium != nil { + for key, item := range cfg.Pipeline.Scriptorium.Artifacts { + configured[key] = artifacts.ConfiguredArtifactDefinition{Enabled: item.Enabled, OutputPath: item.OutputPath} + } + } + if err := catalog.RegisterConfiguredArtifacts(configured, nil); err != nil { + return nil, err + } + return catalog, nil +} + +func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.ArtifactCatalog, locks *effectiveLocks, remoteState map[string]string) { + lockSet := lockSourceSet(locks.All) + fmt.Fprintln(out, "Built-in:") + for _, id := range []string{ + artifacts.ArtifactTranscriptMerged, + artifacts.ArtifactTranscriptPolished, + artifacts.ArtifactTranscriptFull, + artifacts.ArtifactTranscriptTrimmed, + artifacts.ArtifactBoundsSession, + } { + writeArtifactLine(out, id, lockSet, remoteState) + } + fmt.Fprintln(out, "Configured:") + for _, entry := range catalog.ListConfigured() { + writeArtifactLine(out, entry.SourceID, lockSet, remoteState) + } + fmt.Fprintln(out, "Previous-session:") + for _, req := range artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) { + fmt.Fprintf(out, "- narratio.previous_session.artifact.%s required=%t\n", req.Name, req.Required) + } + fmt.Fprintln(out, "Promoted:") + for _, rule := range cfg.Pipeline.Archive.PromoteArtifacts { + writeArtifactLine(out, rule.Source, lockSet, remoteState) + } +} + +func writeArtifactLine(out io.Writer, source string, lockSet map[string]config.ArchiveLockRule, remoteState map[string]string) { + parts := []string{source} + if _, ok := lockSet[source]; ok { + parts = append(parts, "locked") + } + if state := remoteState[source]; state != "" { + parts = append(parts, state) + } + fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) +} + +func remoteArtifactAvailability(ctx context.Context, cfg *config.Config, store storage.ObjectStore, catalog *artifacts.ArtifactCatalog) map[string]string { + out := map[string]string{} + sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) + for _, source := range allCatalogSources(catalog) { + entry, ok := catalog.Lookup(source) + if !ok || strings.TrimSpace(entry.CanonicalRelPath) == "" { + continue + } + key := artifacts.S3PromotedArtifactKey(sessionPrefix, entry.CanonicalRelPath) + if exists, err := store.Exists(ctx, key); err == nil && exists { + out[source] = "remote=promoted" + } else if err != nil { + out[source] = "remote=error" + } else { + out[source] = "remote=missing" + } + } + return out +} + +func allCatalogSources(catalog *artifacts.ArtifactCatalog) []string { + out := []string{ + artifacts.ArtifactTranscriptMerged, + artifacts.ArtifactTranscriptPolished, + artifacts.ArtifactTranscriptFull, + artifacts.ArtifactTranscriptTrimmed, + artifacts.ArtifactBoundsSession, + } + for _, entry := range catalog.ListConfigured() { + out = append(out, entry.SourceID) + } + return out +} + +func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) { + if locks == nil || len(locks.All) == 0 { + fmt.Fprintln(out, "Archive locks: none") + return + } + fmt.Fprintln(out, "Archive locks:") + promoted := map[string]config.ArchivePromotionRule{} + if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Archive != nil { + for _, rule := range cfg.Pipeline.Archive.PromoteArtifacts { + promoted[strings.TrimSpace(rule.Source)] = rule + } + } + staticSet := lockSourceSet(locks.Static) + for _, lock := range locks.All { + origin := "remote" + if _, ok := staticSet[lock.Source]; ok { + origin = "pipeline" + } + promo := "not-promoted" + if _, ok := promoted[lock.Source]; ok { + promo = "promoted" + } + reason := strings.TrimSpace(lock.Reason) + if reason == "" { + reason = "(no reason)" + } + fmt.Fprintf(out, "- %s origin=%s %s reason=%s\n", lock.Source, origin, promo, reason) + } +} + +func lockMapValues(in map[string]config.ArchiveLockRule) []config.ArchiveLockRule { + keys := make([]string, 0, len(in)) + for key := range in { + keys = append(keys, key) + } + sort.Strings(keys) + out := make([]config.ArchiveLockRule, 0, len(keys)) + for _, key := range keys { + item := in[key] + item.Source = key + item.Reason = strings.TrimSpace(item.Reason) + out = append(out, item) + } + return out +} diff --git a/internal/app/operator_helpers_test.go b/internal/app/operator_helpers_test.go new file mode 100644 index 0000000..767d2a6 --- /dev/null +++ b/internal/app/operator_helpers_test.go @@ -0,0 +1,195 @@ +package app + +import ( + "bytes" + "context" + "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" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestExecuteSessionInitRemoteWritesCanonicalSessionConfig(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := 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{ + "session", "init", + "--config", pipelinePath, + "--campaign", campaignPath, + "--session-id", "2026-06-07", + "--title", "The Black Cabin", + "--remote", + }, &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + key := artifacts.S3SessionConfigKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-06-07")) + obj, ok := fake.Objects[key] + if !ok { + t.Fatalf("remote session key %q not uploaded; objects=%v", key, fake.Objects) + } + if !strings.Contains(string(obj.Data), `session_id: "2026-06-07"`) || !strings.Contains(string(obj.Data), "prefix: audio/") { + t.Fatalf("remote session data = %q", string(obj.Data)) + } + if storeInitCalls != 1 { + t.Fatalf("object store init calls = %d, want 1", storeInitCalls) + } +} + +func TestExecuteLockAndUnlockUseRemoteLockStore(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{ + "lock", + "--config", pipelinePath, + "--campaign", campaignPath, + "--session", sessionPath, + "--reason", "manual edit", + "narratio.transcript.trimmed", + }, &stdout, &stderr) + if code != 0 { + t.Fatalf("lock exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + key := artifacts.S3SessionLocksKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")) + obj, ok := fake.Objects[key] + if !ok { + t.Fatalf("remote locks key %q not uploaded", key) + } + if !strings.Contains(string(obj.Data), "source: narratio.transcript.trimmed") || !strings.Contains(string(obj.Data), "reason: manual edit") { + t.Fatalf("lock store data = %q", string(obj.Data)) + } + + stdout.Reset() + stderr.Reset() + code = Execute([]string{ + "unlock", + "--config", pipelinePath, + "--campaign", campaignPath, + "--session", sessionPath, + "narratio.transcript.trimmed", + }, &stdout, &stderr) + if code != 0 { + t.Fatalf("unlock exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + store, err := config.LoadArchiveLockStoreBytes("locks.yml", fake.Objects[key].Data, nil) + if err != nil { + t.Fatalf("LoadArchiveLockStoreBytes() error = %v", err) + } + if len(store.Locks) != 0 { + t.Fatalf("locks after unlock = %#v, want empty", store.Locks) + } + if storeInitCalls != 2 { + t.Fatalf("object store init calls = %d, want 2", storeInitCalls) + } +} + +func TestExecuteArtifactsListRemoteReportsPromotedAvailability(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + trimmedKey := artifacts.S3PromotedArtifactKey( + artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"), + "transcripts/trimmed.json", + ) + fake.SeedObject(storage.FakeObject{Key: trimmedKey, Data: []byte(`{"segments":[]}`)}) + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{ + "artifacts", "list", + "--config", pipelinePath, + "--campaign", campaignPath, + "--session", sessionPath, + "--remote", + }, &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if !strings.Contains(stdout.String(), "narratio.transcript.trimmed remote=promoted") { + t.Fatalf("stdout = %q, want promoted remote availability", stdout.String()) + } +} + +func TestExecuteArchiveLoadsRemoteLocks(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidArchiveConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + lockKey := artifacts.S3SessionLocksKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")) + fake.SeedObject(storage.FakeObject{Key: lockKey, Data: []byte("locks:\n - source: narratio.transcript.trimmed\n reason: remote review\n")}) + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) + + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") + for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze"} { + // The archive stage only checks the manifest statuses and source files. + _ = stageName + } + mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "trimmed.json"), `{"segments":[]}`) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"run-stage", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--force", "archive"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + promotedKey := artifacts.S3PromotedArtifactKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"), "transcripts/trimmed.json") + if _, ok := fake.Objects[promotedKey]; ok { + t.Fatalf("locked promoted key %q was uploaded", promotedKey) + } +} + +func writeValidArchiveConfigFiles(t *testing.T, workspaceRoot string) (string, string, string) { + t.Helper() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + data, err := os.ReadFile(pipelinePath) + if err != nil { + t.Fatalf("read pipeline: %v", err) + } + updated := strings.Replace(string(data), "upload_run: false", "upload_run: true", 1) + if err := os.WriteFile(pipelinePath, []byte(updated), 0o644); err != nil { + t.Fatalf("write pipeline: %v", err) + } + ctx := context.Background() + cfg, err := config.LoadWithSessionOptions(pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{}) + if err != nil { + t.Fatalf("LoadWithSessionOptions() error = %v", err) + } + store := &manifest.LocalStore{} + m := manifest.New("2026-05-03", nowUTC()) + m.Campaign = "sample-campaign" + m.RunID = "20260521T160000Z-test" + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze"} { + m.MarkStageSucceeded(name, nowUTC(), nil) + } + path := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID) + if err := store.Save(ctx, path, m); err != nil { + t.Fatalf("save manifest: %v", err) + } + runManifestPath := artifacts.SessionRunManifestPathForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, m.RunID) + if err := os.MkdirAll(filepath.Dir(runManifestPath), 0o755); err != nil { + t.Fatalf("mkdir run manifest: %v", err) + } + if err := os.WriteFile(runManifestPath, []byte("{}\n"), 0o644); err != nil { + t.Fatalf("write run manifest: %v", err) + } + return pipelinePath, campaignPath, sessionPath +} diff --git a/internal/app/remote_locks.go b/internal/app/remote_locks.go new file mode 100644 index 0000000..1aebf1e --- /dev/null +++ b/internal/app/remote_locks.go @@ -0,0 +1,157 @@ +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" +) + +type effectiveLocks struct { + Static []config.ArchiveLockRule + Remote []config.ArchiveLockRule + All []config.ArchiveLockRule + Key string +} + +func remoteLocksKey(cfg *config.Config) (string, error) { + if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { + return "", fmt.Errorf("resolved config is required") + } + if cfg.Pipeline.Storage.S3 == nil { + return "", fmt.Errorf("pipeline.storage.s3 configuration is required") + } + sessionPrefix := artifacts.S3SessionPrefix( + cfg.Pipeline.Storage.S3.RootPrefix, + cfg.Session.Campaign, + cfg.Session.SessionID, + ) + return artifacts.S3SessionLocksKey(sessionPrefix), nil +} + +func loadRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*config.ArchiveLockStore, string, error) { + key, err := remoteLocksKey(cfg) + if err != nil { + return nil, "", err + } + exists, err := store.Exists(ctx, key) + if err != nil { + return nil, key, fmt.Errorf("check remote locks %q: %w", key, err) + } + if !exists { + return &config.ArchiveLockStore{}, key, nil + } + tmp, err := downloadObjectToTemp(ctx, store, key, "narratio-locks-*.yml") + if err != nil { + return nil, key, fmt.Errorf("download remote locks %q: %w", key, err) + } + defer func() { _ = os.Remove(tmp) }() + data, err := os.ReadFile(tmp) + if err != nil { + return nil, key, fmt.Errorf("read remote locks %q: %w", key, err) + } + lockStore, err := config.LoadArchiveLockStoreBytes("s3://"+s3BucketName(cfg.Pipeline)+"/"+key, data, cfg.Pipeline.Scriptorium) + if err != nil { + return nil, key, err + } + return lockStore, key, nil +} + +func loadEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*effectiveLocks, error) { + staticLocks := staticArchiveLocks(cfg) + if store == nil { + return &effectiveLocks{ + Static: staticLocks, + All: append([]config.ArchiveLockRule(nil), staticLocks...), + }, nil + } + lockStore, key, err := loadRemoteLockStore(ctx, cfg, store) + if err != nil { + return nil, err + } + remoteLocks := append([]config.ArchiveLockRule(nil), lockStore.Locks...) + return &effectiveLocks{ + Static: staticLocks, + Remote: remoteLocks, + All: config.MergeArchiveLockRules(staticLocks, remoteLocks), + Key: key, + }, nil +} + +func staticArchiveLocks(cfg *config.Config) []config.ArchiveLockRule { + if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Archive == nil { + return nil + } + return append([]config.ArchiveLockRule(nil), cfg.Pipeline.Archive.Locks...) +} + +func applyEffectiveLocks(cfg *config.Config, locks []config.ArchiveLockRule) { + if cfg == nil || cfg.Pipeline == nil { + return + } + if cfg.Pipeline.Archive == nil { + cfg.Pipeline.Archive = &config.ArchiveConfig{} + } + cfg.Pipeline.Archive.Locks = append([]config.ArchiveLockRule(nil), locks...) +} + +func uploadRemoteLockStore(ctx context.Context, store storage.ObjectStore, key string, lockStore *config.ArchiveLockStore) error { + data, err := config.MarshalArchiveLockStore(lockStore) + if err != nil { + return err + } + tmp, err := os.CreateTemp("", "narratio-locks-upload-*.yml") + if err != nil { + return fmt.Errorf("create lock store temp file: %w", err) + } + tmpPath := tmp.Name() + defer func() { _ = os.Remove(tmpPath) }() + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return fmt.Errorf("write lock store temp file: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("close lock store temp file: %w", err) + } + if _, err := store.Upload(ctx, tmpPath, key, storage.UploadOptions{ContentType: "application/x-yaml; charset=utf-8"}); err != nil { + return fmt.Errorf("upload remote locks %q: %w", key, err) + } + return nil +} + +func lockSourceSet(locks []config.ArchiveLockRule) map[string]config.ArchiveLockRule { + out := make(map[string]config.ArchiveLockRule, len(locks)) + for _, lock := range locks { + source := strings.TrimSpace(lock.Source) + if source == "" { + continue + } + lock.Source = source + lock.Reason = strings.TrimSpace(lock.Reason) + out[source] = lock + } + return out +} + +func writeLocalFile(path string, data []byte, force bool) error { + cleaned := filepath.Clean(strings.TrimSpace(path)) + if cleaned == "" || cleaned == "." { + return fmt.Errorf("output path is required") + } + if !force { + if _, err := os.Stat(cleaned); err == nil { + return fmt.Errorf("output file %q already exists; pass --force to overwrite", cleaned) + } else if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("check output file %q: %w", cleaned, err) + } + } + if err := os.MkdirAll(filepath.Dir(cleaned), 0o755); err != nil { + return fmt.Errorf("create output directory: %w", err) + } + return os.WriteFile(cleaned, data, 0o644) +} diff --git a/internal/app/runner.go b/internal/app/runner.go index c8f3f4d..805ffa1 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -87,12 +87,19 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage env.Storage = &storage.NoopBackend{} } if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages) { - objectStore, err := storage.NewObjectStoreFromConfig(ctx, env.Config) + objectStore, err := newObjectStoreFromConfigFn(ctx, env.Config) if err != nil { return nil, fmt.Errorf("initialize object store backend: %w", err) } env.ObjectStore = objectStore } + if needsRemoteLocksForRun(env.Config, stages) { + locks, err := loadEffectiveLocks(ctx, env.Config, env.ObjectStore) + if err != nil { + return nil, fmt.Errorf("load remote archive locks: %w", err) + } + applyEffectiveLocks(env.Config, locks.All) + } if env.Notifier == nil { env.Notifier = ¬ify.NoopSender{} } @@ -576,6 +583,32 @@ func needsObjectStoreForRun(cfg *config.Config, stages []stage.Stage) bool { return true } +func needsRemoteLocksForRun(cfg *config.Config, stages []stage.Stage) bool { + if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { + return false + } + archiveRequested := false + for _, s := range stages { + if s != nil && s.Name() == "archive" { + archiveRequested = true + break + } + } + if !archiveRequested { + 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 cfg.Pipeline.Storage.S3 != nil +} + func configuredScriptoriumArtifacts(cfg *config.Config) map[string]config.ScriptoriumArtifactConfig { if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Scriptorium == nil { return nil diff --git a/internal/app/status.go b/internal/app/status.go deleted file mode 100644 index e4afcfb..0000000 --- a/internal/app/status.go +++ /dev/null @@ -1,67 +0,0 @@ -package app - -import ( - "context" - "flag" - "fmt" - "io" - "sort" - - "gitea.maximumdirect.net/eric/narratio/internal/manifest" -) - -// Status reads and prints stage statuses from an existing manifest. -func Status(ctx context.Context, args []string, out io.Writer) error { - fs := flag.NewFlagSet("status", flag.ContinueOnError) - fs.SetOutput(io.Discard) - - var manifestPath string - fs.StringVar(&manifestPath, "manifest", "", "path to manifest.json") - - if err := fs.Parse(args); err != nil { - return fmt.Errorf("status: invalid flags: %w", err) - } - if fs.NArg() != 0 { - return fmt.Errorf("status: unexpected positional arguments") - } - if manifestPath == "" { - return fmt.Errorf("status: --manifest is required") - } - - store := &manifest.LocalStore{} - m, err := store.Load(ctx, manifestPath) - if err != nil { - return fmt.Errorf("status: %w", err) - } - - if _, err := fmt.Fprintf(out, "session_id: %s\n", m.SessionID); err != nil { - return err - } - if _, err := fmt.Fprintf(out, "updated_at: %s\n", m.UpdatedAt.UTC().Format("2006-01-02T15:04:05Z07:00")); err != nil { - return err - } - - if len(m.Stages) == 0 { - _, err := fmt.Fprintln(out, "stages: no stages recorded") - return err - } - - if _, err := fmt.Fprintln(out, "stages:"); err != nil { - return err - } - - names := make([]string, 0, len(m.Stages)) - for name := range m.Stages { - names = append(names, name) - } - sort.Strings(names) - - for _, name := range names { - status := m.Stages[name].Status - if _, err := fmt.Fprintf(out, "- %s: %s\n", name, status); err != nil { - return err - } - } - - return nil -} diff --git a/internal/artifacts/s3_keys.go b/internal/artifacts/s3_keys.go index f86c0e3..1913e92 100644 --- a/internal/artifacts/s3_keys.go +++ b/internal/artifacts/s3_keys.go @@ -40,6 +40,12 @@ func S3SessionConfigKey(sessionPrefix string) string { return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "session.yml") } +// S3SessionLocksKey returns the mutable session lock store key. +// Format: {session_prefix}/locks.yml +func S3SessionLocksKey(sessionPrefix string) string { + return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "locks.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 3946cb2..7000275 100644 --- a/internal/artifacts/s3_keys_test.go +++ b/internal/artifacts/s3_keys_test.go @@ -22,6 +22,11 @@ func TestS3KeyConstruction(t *testing.T) { t.Fatalf("session config key = %q", sessionConfigKey) } + locksKey := S3SessionLocksKey(`dnd\campaigns\forsaken\sessions\2026-04-19\`) + if locksKey != "dnd/campaigns/forsaken/sessions/2026-04-19/locks.yml" { + t.Fatalf("locks key = %q", locksKey) + } + 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 6a7d16f..e7ee159 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -112,6 +112,11 @@ type ArchiveLockRule struct { Reason string `yaml:"reason"` } +// ArchiveLockStore is the mutable per-session remote lock store. +type ArchiveLockStore struct { + Locks []ArchiveLockRule `yaml:"locks"` +} + // WhisperXConfig configures WhisperX adapter settings. type WhisperXConfig struct { TranscribeURL string `yaml:"transcribe_url"` diff --git a/internal/config/load.go b/internal/config/load.go index 9523a1c..19c8f6e 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -85,6 +85,33 @@ func LoadSessionBytesWithOptions(label string, data []byte, opts SessionLoadOpti return &cfg, nil } +// LoadArchiveLockStoreBytes loads a mutable session lock store with strict +// field checking and source validation. +func LoadArchiveLockStoreBytes(label string, data []byte, scriptorium *ScriptoriumConfig) (*ArchiveLockStore, error) { + var store ArchiveLockStore + if err := decodeStrictYAMLFromReader("archive lock store", label, strings.NewReader(string(data)), &store); err != nil { + return nil, fmt.Errorf("load archive lock store: %w", err) + } + locks, err := ValidateArchiveLockRules(store.Locks, scriptorium, "locks") + if err != nil { + return nil, fmt.Errorf("load archive lock store: %w", err) + } + store.Locks = locks + return &store, nil +} + +// MarshalArchiveLockStore serializes a mutable lock store as strict-compatible YAML. +func MarshalArchiveLockStore(store *ArchiveLockStore) ([]byte, error) { + if store == nil { + store = &ArchiveLockStore{} + } + data, err := yaml.Marshal(store) + if err != nil { + return nil, fmt.Errorf("marshal archive lock store: %w", err) + } + return data, nil +} + // Load loads and resolves combined pipeline, campaign, and session configuration. // Passing only a session path is supported for package-internal compatibility; // in that form campaign.yml is expected next to the session file. diff --git a/internal/config/storage_archive_test.go b/internal/config/storage_archive_test.go index 869bbc7..4203556 100644 --- a/internal/config/storage_archive_test.go +++ b/internal/config/storage_archive_test.go @@ -395,6 +395,54 @@ archive: } } +func TestArchiveLockStoreBytesStrictDecodeAndValidation(t *testing.T) { + store, err := LoadArchiveLockStoreBytes("locks.yml", []byte(`locks: + - source: narratio.transcript.trimmed + reason: reviewed +`), nil) + if err != nil { + t.Fatalf("LoadArchiveLockStoreBytes() error = %v", err) + } + if len(store.Locks) != 1 || store.Locks[0].Source != "narratio.transcript.trimmed" || store.Locks[0].Reason != "reviewed" { + t.Fatalf("locks = %#v", store.Locks) + } + + _, err = LoadArchiveLockStoreBytes("locks.yml", []byte(`locks: + - source: narratio.transcript.trimmed + dest: transcripts/trimmed.json +`), nil) + if err == nil || !strings.Contains(err.Error(), "strict decode failed") { + t.Fatalf("unknown field error = %v, want strict decode failed", err) + } + + _, err = LoadArchiveLockStoreBytes("locks.yml", []byte(`locks: + - source: narratio.transcript.trimmed + - source: narratio.transcript.trimmed +`), nil) + if err == nil || !strings.Contains(err.Error(), "duplicates another archive lock source") { + t.Fatalf("duplicate error = %v", err) + } +} + +func TestMergeArchiveLockRulesStaticWins(t *testing.T) { + merged := MergeArchiveLockRules( + []ArchiveLockRule{{Source: "narratio.transcript.trimmed", Reason: "static"}}, + []ArchiveLockRule{ + {Source: "narratio.transcript.trimmed", Reason: "remote"}, + {Source: "narratio.transcript.full", Reason: "remote full"}, + }, + ) + if len(merged) != 2 { + t.Fatalf("merged len = %d, want 2: %#v", len(merged), merged) + } + if merged[0].Source != "narratio.transcript.trimmed" || merged[0].Reason != "static" { + t.Fatalf("merged[0] = %#v, want static lock", merged[0]) + } + if merged[1].Source != "narratio.transcript.full" { + t.Fatalf("merged[1] = %#v, want remote full lock", merged[1]) + } +} + func TestSessionAudioS3Validation(t *testing.T) { tests := []struct { name string diff --git a/internal/config/validate.go b/internal/config/validate.go index af681ff..9df5306 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -152,24 +152,67 @@ func validateArchive(cfg *ArchiveConfig, scriptorium *ScriptoriumConfig) error { } seenDest[normalizedDest] = struct{}{} } + locks, err := ValidateArchiveLockRules(cfg.Locks, scriptorium, "pipeline.archive.locks") + if err != nil { + return err + } + cfg.Locks = locks + return nil +} + +// ValidateArchiveLockRules validates and normalizes source-based archive locks. +func ValidateArchiveLockRules(locks []ArchiveLockRule, scriptorium *ScriptoriumConfig, label string) ([]ArchiveLockRule, error) { seenLocks := map[string]struct{}{} - for i, item := range cfg.Locks { - prefix := fmt.Sprintf("pipeline.archive.locks[%d]", i) + out := make([]ArchiveLockRule, 0, len(locks)) + if strings.TrimSpace(label) == "" { + label = "archive.locks" + } + for i, item := range locks { + prefix := fmt.Sprintf("%s[%d]", label, i) source := strings.TrimSpace(item.Source) if source == "" { - return fmt.Errorf("%s.source is required", prefix) + return nil, fmt.Errorf("%s.source is required", prefix) } if _, err := archiveSourceKnown(source, scriptorium); err != nil { - return fmt.Errorf("%s.source %q is unsupported: %w", prefix, item.Source, err) + return nil, fmt.Errorf("%s.source %q is unsupported: %w", prefix, item.Source, err) } if _, ok := seenLocks[source]; ok { - return fmt.Errorf("%s.source %q duplicates another archive lock source", prefix, source) + return nil, fmt.Errorf("%s.source %q duplicates another archive lock source", prefix, source) } seenLocks[source] = struct{}{} - cfg.Locks[i].Source = source - cfg.Locks[i].Reason = strings.TrimSpace(item.Reason) + out = append(out, ArchiveLockRule{ + Source: source, + Reason: strings.TrimSpace(item.Reason), + }) } - return nil + return out, nil +} + +// MergeArchiveLockRules returns the union of static and remote locks. Static +// locks win when both sources contain the same lock. +func MergeArchiveLockRules(staticLocks, remoteLocks []ArchiveLockRule) []ArchiveLockRule { + out := make([]ArchiveLockRule, 0, len(staticLocks)+len(remoteLocks)) + seen := map[string]struct{}{} + for _, item := range staticLocks { + source := strings.TrimSpace(item.Source) + if source == "" { + continue + } + out = append(out, ArchiveLockRule{Source: source, Reason: strings.TrimSpace(item.Reason)}) + seen[source] = struct{}{} + } + for _, item := range remoteLocks { + source := strings.TrimSpace(item.Source) + if source == "" { + continue + } + if _, ok := seen[source]; ok { + continue + } + out = append(out, ArchiveLockRule{Source: source, Reason: strings.TrimSpace(item.Reason)}) + seen[source] = struct{}{} + } + return out } func archiveSourceKnown(source string, scriptorium *ScriptoriumConfig) (string, error) {