Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e58e545686 | |||
| 6ff54c5a0f | |||
| 924b5d15c6 | |||
| b065663180 | |||
| 3ba564b00f | |||
| a3986cf0d6 | |||
| 539601bd16 | |||
| 6ca1c8d6b0 | |||
| 4b7b50981b |
5
.gitignore
vendored
5
.gitignore
vendored
@@ -2,6 +2,8 @@
|
||||
.codex
|
||||
AGENTS.md
|
||||
|
||||
.DS_Store
|
||||
|
||||
# ---> Go
|
||||
# If you prefer the allow list template instead of the deny list, see community template:
|
||||
# https://github.com/github/gitignore/blob/main/community/Golang/Go.AllowList.gitignore
|
||||
@@ -22,6 +24,9 @@ AGENTS.md
|
||||
# Dependency directories (remove the comment below to include it)
|
||||
# vendor/
|
||||
|
||||
# Go cache
|
||||
.gocache
|
||||
|
||||
# Go workspace file
|
||||
go.work
|
||||
go.work.sum
|
||||
|
||||
@@ -65,13 +65,15 @@ Optional secrets-from-files config:
|
||||
|
||||
YAML decoding is strict (`KnownFields(true)`), so unknown fields fail fast.
|
||||
|
||||
Maintainer note: application defaults are centralized in [`internal/config/defaults.go`](internal/config/defaults.go).
|
||||
|
||||
## Storage And Archive Foundations
|
||||
|
||||
Narratio now includes configuration and path-model foundations for archive support, plus implemented prepare-stage S3 audio input.
|
||||
|
||||
Implemented foundations:
|
||||
|
||||
- `pipeline.storage.s3` config shape (`bucket`, `root_prefix`, `region`, `endpoint`, `force_path_style`)
|
||||
- `pipeline.storage.s3` config shape (`bucket`, `root_prefix`, `region`, `endpoint`, `force_path_style`, `access_key_id_env`, `secret_access_key_env`)
|
||||
- `pipeline.spool` config shape (`root`, `delete_audio_after_archive`)
|
||||
- `pipeline.archive` config shape (`enabled`, `upload_run`, `promote_artifacts`)
|
||||
- promotion-rule validation (`from`/`to` required, relative-only paths, traversal rejected)
|
||||
@@ -83,6 +85,8 @@ Implemented foundations:
|
||||
Current defaults:
|
||||
|
||||
- `pipeline.storage.s3.root_prefix`: `dnd`
|
||||
- `pipeline.storage.s3.access_key_id_env`: `OBJECT_STORAGE_KEY_ID`
|
||||
- `pipeline.storage.s3.secret_access_key_env`: `OBJECT_STORAGE_KEY`
|
||||
- `pipeline.workspace.cleanup_after_archive`: `false`
|
||||
- `pipeline.spool.root`: `/var/spool/narratio`
|
||||
- `pipeline.spool.delete_audio_after_archive`: `false`
|
||||
@@ -110,7 +114,8 @@ Current boundaries:
|
||||
- `pipeline.workspace.cleanup_after_archive: true` removes only the run-scoped local workdir after successful archive commit
|
||||
- cleanup executes only after all selected stages for the command invocation succeed
|
||||
- cleanup does not run for failed, incomplete, skipped, or unarchived runs
|
||||
- local development `audio_dir`/`audio_files` source inputs are never deleted by spool cleanup
|
||||
- local development `audio_dir`/`audio_files` source inputs are never deleted by spool cleanup
|
||||
- S3 credentials are resolved from configured env-var names when both are present; if either is missing, Narratio falls back to the AWS SDK default credential chain
|
||||
|
||||
S3 input details and current boundaries are documented in [docs/s3-audio-input.md](docs/s3-audio-input.md).
|
||||
|
||||
|
||||
@@ -114,6 +114,7 @@ CLI pipeline config path resolution:
|
||||
- when `--config` is omitted, Narratio searches defaults in order:
|
||||
- `/usr/local/etc/narratio/pipeline.yml`
|
||||
- `/etc/narratio/pipeline.yml`
|
||||
- default values are centralized in `internal/config/defaults.go`
|
||||
|
||||
CLI session config path resolution:
|
||||
|
||||
@@ -151,6 +152,8 @@ Storage and archive foundations:
|
||||
- `region`
|
||||
- `endpoint`
|
||||
- `force_path_style` (default `false`)
|
||||
- `access_key_id_env` (default `OBJECT_STORAGE_KEY_ID`)
|
||||
- `secret_access_key_env` (default `OBJECT_STORAGE_KEY`)
|
||||
- `pipeline.spool.root` defaults to `/var/spool/narratio`
|
||||
- `pipeline.workspace.cleanup_after_archive` defaults to `false`
|
||||
- `pipeline.spool.delete_audio_after_archive` defaults to `false`
|
||||
@@ -176,7 +179,9 @@ Session input foundations:
|
||||
Cross-config validation scope:
|
||||
|
||||
- `pipeline.storage.s3.bucket` is required only when an S3-dependent feature is explicitly configured (for current foundations, that includes `session.inputs.audio_s3`, and archive upload intent when using `storage.backend: s3`)
|
||||
- no AWS credentials are stored in Narratio config; credential resolution remains an external runtime concern
|
||||
- no AWS credential values are stored in Narratio config; only env-var names are configured
|
||||
- when both configured credential env vars resolve to non-empty values, the S3 backend uses them as static credentials
|
||||
- when either configured credential value is missing, the S3 backend falls back to the AWS SDK default credential chain
|
||||
|
||||
Remote object-store backend scope:
|
||||
|
||||
|
||||
507
docs/development/workspace-implementation-plan.md
Normal file
507
docs/development/workspace-implementation-plan.md
Normal file
@@ -0,0 +1,507 @@
|
||||
# Workspace Architecture Implementation Plan (Audit)
|
||||
|
||||
## 1. Executive Summary
|
||||
|
||||
**Complexity assessment:** **heavy**.
|
||||
|
||||
This is not a single path-helper refactor. The current codebase has a hybrid session/run model that works for current behavior, but diverges from `docs/development/workspace.md` in foundational places (workspace root shape, manifest responsibilities, stage output placement, and archive symmetry).
|
||||
|
||||
Highest-risk areas:
|
||||
|
||||
1. Splitting the current single manifest model into durable **session manifest** vs per-invocation **run manifest** without regressing skip/force/resume UX.
|
||||
2. Migrating path helpers and artifact-store interfaces from session-only roots (`work/{session}`) to campaign-aware roots (`work/{campaign}/{session}`) while preserving existing runs.
|
||||
3. Introducing run-local stage outputs + immediate promotion while keeping stage tests and archive behavior stable.
|
||||
4. Avoiding stale downstream skips after forced upstream reruns.
|
||||
|
||||
Surprising findings:
|
||||
|
||||
1. Code already has campaign/run-aware helpers (`SessionRunWorkDir`, `SessionSpoolAudioDir`) but core session helpers and manifest pathing remain campaign-unaware.
|
||||
2. Archive recently gained run/session fallback behavior for manifest/promotion sources, which confirms an existing hybrid-layout pressure point.
|
||||
3. Analyze input resolution is functional but ad hoc and stage-local; there is no centralized artifact registry/resolver.
|
||||
|
||||
---
|
||||
|
||||
## 2. Current-State Map
|
||||
|
||||
## 2.1 Workspace Path Construction
|
||||
|
||||
Primary path model:
|
||||
|
||||
- [`internal/artifacts/paths.go`](../../internal/artifacts/paths.go)
|
||||
- `SessionWorkDir(rootDir, sessionID)` => `{root}/work/{session_id}`
|
||||
- `buildSessionPaths(workspaceRoot, sessionID)` roots all canonical paths under `{root}/work/{session_id}`
|
||||
- `SessionRunWorkDir(rootDir, campaign, sessionID, runID)` exists, but is not the default session root helper.
|
||||
|
||||
Artifact store abstraction:
|
||||
|
||||
- [`internal/artifacts/store.go`](../../internal/artifacts/store.go)
|
||||
- `SessionPaths(sessionID string)` / `EnsureLayout(sessionID string)` are session-id-only (no campaign argument).
|
||||
- [`internal/artifacts/local.go`](../../internal/artifacts/local.go)
|
||||
- `EnsureLayout` creates session-level folders under `SessionWorkDir`.
|
||||
|
||||
Path normalization helper:
|
||||
|
||||
- [`internal/artifacts/resolve.go`](../../internal/artifacts/resolve.go)
|
||||
- `ResolveSessionLocalPathForRead` accepts absolute/workspace/session-relative values and probes filesystem.
|
||||
|
||||
Where campaign-aware/run-aware support exists:
|
||||
|
||||
- Local run path helpers: `SessionRunWorkDir`, `SessionSpoolAudioDir`.
|
||||
- S3 key helpers: [`internal/artifacts/s3_keys.go`](../../internal/artifacts/s3_keys.go) (`campaigns/sessions/runs/current`).
|
||||
|
||||
Where session-root assumptions remain `{workspace}/work/{session}`:
|
||||
|
||||
- Manifest path computation: [`internal/app/runner.go`](../../internal/app/runner.go) `manifestPathFor`.
|
||||
- Artifact store layout and most stage `paths := env.ArtifactStore.SessionPaths(sessionID)` calls.
|
||||
- Many tests hardcode `workspace/work/<session>/...` (examples below).
|
||||
|
||||
Manual/ad hoc path construction (not through a single resolver API):
|
||||
|
||||
- Common stage patterns: `filepath.Join(paths.TranscriptsDir, "...")`, `filepath.Join(paths.ArtifactsDir, "...")`, etc.
|
||||
- `prepare` run/work selection: [`internal/stage/prepare.go`](../../internal/stage/prepare.go) `pathsWorkDirForManifest`.
|
||||
- Analyze input fallback path resolution: `resolveInputPathForRead` in [`internal/stage/analyze.go`](../../internal/stage/analyze.go).
|
||||
|
||||
## 2.2 Manifest and Run Identity
|
||||
|
||||
Current manifest model:
|
||||
|
||||
- [`internal/manifest/manifest.go`](../../internal/manifest/manifest.go) `Manifest` includes both session and run fields:
|
||||
- `SessionID`, `Campaign`
|
||||
- `RunID`, `LocalWorkDir`, `LocalSpoolDir`
|
||||
- `S3Bucket`, `S3SessionPrefix`, `S3RunPrefix`
|
||||
- `Stages`, `Inputs`, stage outputs/logs/generated configs/metadata.
|
||||
|
||||
Current persistence:
|
||||
|
||||
- [`internal/manifest/store.go`](../../internal/manifest/store.go) `LocalStore` reads/writes one JSON manifest path.
|
||||
- Runner always loads/saves one manifest path via `manifestPathFor(cfg)` (session-root path under current layout).
|
||||
|
||||
Identity initialization:
|
||||
|
||||
- [`internal/app/runner.go`](../../internal/app/runner.go) `ensureManifestIdentity` populates run fields if absent.
|
||||
- `RunID` is generated once for an empty manifest and reused thereafter (hybrid semantics).
|
||||
|
||||
Interpretation today:
|
||||
|
||||
- Best described as a **hybrid session manifest** with run identity fields, not as distinct session + run manifests.
|
||||
|
||||
What this means for redesign:
|
||||
|
||||
- Session-vs-run split is not just file relocation; it requires new responsibilities and write flows.
|
||||
- A backward-compatible evolution path is possible by:
|
||||
- preserving current fields in session manifest for migration/read-compat,
|
||||
- adding explicit run-manifest type + path,
|
||||
- gradually moving invocation-specific details to run manifests.
|
||||
|
||||
## 2.3 Stage Output Paths (Current)
|
||||
|
||||
All implemented stages currently write canonical artifacts directly into session-level `paths.*` roots (under current session root), with logs/configs typically also session-level.
|
||||
|
||||
1. `prepare` ([`internal/stage/prepare.go`](../../internal/stage/prepare.go))
|
||||
- Inputs copied to `inputs/` (`session.yml`, `pipeline.resolved.yml`, `speakers.yml`, `autocorrect.yml`, `glossary.yml`)
|
||||
- Audio copied to `audio/`
|
||||
- Manifest input provenance recorded in `m.Inputs`
|
||||
- S3 audio uses run-scoped spool/work helpers when `RunID` is present.
|
||||
|
||||
2. `transcribe` ([`internal/stage/transcribe.go`](../../internal/stage/transcribe.go))
|
||||
- Outputs: `transcripts/raw/{speaker}.json`
|
||||
- Stage metadata only (no stage logs/config files produced here).
|
||||
|
||||
3. `merge` ([`internal/stage/merge.go`](../../internal/stage/merge.go))
|
||||
- Pre-normalize intermediates: `transcripts/raw/normalized/{basename}.normalized.json`
|
||||
- Merge output: `transcripts/merged.json`
|
||||
- Report: `artifacts/seriatim.report.json` (if enabled)
|
||||
- Logs/configs:
|
||||
- per-input normalize logs/configs in session-level `logs/` + `config/`
|
||||
- merge logs/config in session-level `logs/` + `config/`
|
||||
|
||||
4. `polish` ([`internal/stage/polish.go`](../../internal/stage/polish.go))
|
||||
- Output: `transcripts/processed.json`
|
||||
- Report: `artifacts/audita.report.json` (if enabled)
|
||||
- Work dir: `artifacts/audita-work`
|
||||
- Logs/config: `logs/audita.*`, `config/audita.generated.yml`
|
||||
|
||||
5. `normalize` ([`internal/stage/normalize.go`](../../internal/stage/normalize.go))
|
||||
- Output: `transcripts/normalized.json` (configurable)
|
||||
- Report: `artifacts/seriatim.normalize.report.json` (if enabled)
|
||||
- Logs/config: `logs/seriatim.normalize.*`, `config/seriatim.normalize.generated.yml`
|
||||
|
||||
6. `trim` ([`internal/stage/trim.go`](../../internal/stage/trim.go))
|
||||
- Output: `transcripts/trimmed.json` (configurable)
|
||||
- Bounds output path from config (default examples use `artifacts/session_bounds.json`)
|
||||
- Logs/configs in session-level `logs/` and `config/` for scriptorium + seriatim invocations
|
||||
|
||||
7. `analyze` ([`internal/stage/analyze.go`](../../internal/stage/analyze.go))
|
||||
- Current implemented artifact: `artifacts/session_recap.md`
|
||||
- Logs/config in session-level `logs/` + `config/`
|
||||
- Optional render diagnostics under session-level artifacts/logs/config.
|
||||
|
||||
8. `archive` ([`internal/stage/archive.go`](../../internal/stage/archive.go))
|
||||
- Reads from a "workDir" derived by `archiveWorkDir`:
|
||||
- prefers `m.LocalWorkDir` if it exists,
|
||||
- else tries run-scoped campaign/session/run path,
|
||||
- else falls back to legacy session path.
|
||||
- Uploads run record and promotions.
|
||||
- Current code now resolves manifest and promotion sources via run/session fallback.
|
||||
|
||||
9. `notify`
|
||||
- Placeholder only in [`internal/stage/placeholders.go`](../../internal/stage/placeholders.go); no durable outputs.
|
||||
|
||||
Durable-vs-diagnostic split today:
|
||||
|
||||
- Durable artifacts and diagnostics are mixed at session level.
|
||||
- No run-local stage directories exist yet.
|
||||
|
||||
## 2.4 Idempotency / Force / Resume / Sparse Runs
|
||||
|
||||
Run control implementation:
|
||||
|
||||
- [`internal/app/run_control.go`](../../internal/app/run_control.go)
|
||||
- skip rule: `!force && stageSucceeded(manifest, stage)`
|
||||
- stale detection TODO only; no invalidation logic.
|
||||
|
||||
Command behavior:
|
||||
|
||||
- `run`: full plan through `executeStages`.
|
||||
- `run-stage`: single selected stage through `executeStages`.
|
||||
- `resume`: starts at first non-succeeded stage unless `--force`.
|
||||
|
||||
Current assumptions:
|
||||
|
||||
- Command invocation mutates a single durable workspace + single manifest for the session path model.
|
||||
- There is no per-invocation run manifest write path.
|
||||
|
||||
Smallest safe UX-preserving invariant to keep during migration:
|
||||
|
||||
- Session manifest remains source-of-truth for skip decisions across invocations.
|
||||
|
||||
## 2.5 Archive and S3 Alignment
|
||||
|
||||
S3 semantics are relatively mature:
|
||||
|
||||
- [`internal/artifacts/s3_keys.go`](../../internal/artifacts/s3_keys.go)
|
||||
- session prefix + run prefix + `current/manifest.json` + `current/run_id.txt`.
|
||||
|
||||
Archive stage behavior:
|
||||
|
||||
- uploads run records under run prefix,
|
||||
- uploads promoted session outputs,
|
||||
- uploads current manifest then current run pointer,
|
||||
- current pointer is commit marker,
|
||||
- required promotions fail; optional promotions skipped.
|
||||
|
||||
Local-vs-remote mismatch still present:
|
||||
|
||||
- Local canonical session root currently defaults to `work/{session}` (artifact store),
|
||||
- while archive/run helpers expect campaign-aware run locations (`work/{campaign}/{session}/{run}`),
|
||||
- causing fallback logic and hybrid handling in `archive`.
|
||||
|
||||
## 2.6 Artifact Source Resolution in Analyze
|
||||
|
||||
Current implementation is stage-local and ad hoc:
|
||||
|
||||
- Transcript discoverers:
|
||||
- `discoverProcessedTranscript`
|
||||
- `discoverNormalizedTranscript`
|
||||
- `discoverTrimmedTranscript`
|
||||
- Input source switch in `resolveScriptoriumInput` supports:
|
||||
- `processed_transcript`
|
||||
- `normalized_transcript`
|
||||
- `trimmed_transcript`
|
||||
- `previous_session_artifact`
|
||||
|
||||
No first-class artifact registry exists yet. Aliases and canonical IDs are not modeled.
|
||||
|
||||
## 2.7 Stale/Invalidation
|
||||
|
||||
- `manifest.StageStatus` already defines `stale` (`internal/manifest/status.go`), but no stage uses it.
|
||||
- Skip logic ignores stale state and only checks `succeeded`.
|
||||
- Forced upstream rerun does not invalidate downstream stage success markers.
|
||||
|
||||
## 2.8 Test Coverage Relevant to Redesign
|
||||
|
||||
High-value existing coverage:
|
||||
|
||||
- Workspace helpers:
|
||||
- [`internal/artifacts/paths_model_test.go`](../../internal/artifacts/paths_model_test.go)
|
||||
- [`internal/artifacts/resolve_test.go`](../../internal/artifacts/resolve_test.go)
|
||||
- Runner semantics:
|
||||
- [`internal/app/runner_test.go`](../../internal/app/runner_test.go)
|
||||
- [`internal/app/resume_run_stage_test.go`](../../internal/app/resume_run_stage_test.go)
|
||||
- Stage path/output behavior:
|
||||
- `internal/stage/*_test.go` for prepare/transcribe/merge/polish/normalize/trim/analyze/archive
|
||||
- Archive behavior:
|
||||
- [`internal/stage/archive_test.go`](../../internal/stage/archive_test.go)
|
||||
- [`internal/app/post_archive_cleanup_test.go`](../../internal/app/post_archive_cleanup_test.go)
|
||||
|
||||
Tests likely to fail during workspace redesign:
|
||||
|
||||
- Any tests hardcoding `work/{session}` manifest and transcript paths (many in `internal/app/*test.go`, `internal/stage/*test.go`).
|
||||
- Archive tests assuming current hybrid fallback behavior.
|
||||
|
||||
---
|
||||
|
||||
## 3. Gap Analysis Against `docs/development/workspace.md`
|
||||
|
||||
| Intended concept | Current status | Notes |
|
||||
|---|---|---|
|
||||
| Session root at `work/{campaign}/{session}` | **Partial / mostly absent** | `SessionWorkDir` and artifact store still use `work/{session}`. Campaign-aware run helper exists separately. |
|
||||
| Distinct session manifest vs run manifest | **Absent** | One hybrid manifest model/file is used. |
|
||||
| Run-local stage dirs under `runs/{run_id}/{stage}` | **Absent** | Stages write canonical outputs/logs/config directly at session level. |
|
||||
| Immediate promotion run-local -> session canonical | **Absent** | No run-local staging area to promote from today. |
|
||||
| Session manifest skip source of truth | **Present** | Skip/resume use single manifest stage statuses. |
|
||||
| Sparse runs represented under `runs/{run_id}` | **Absent** | No run-manifest/per-run stage records yet. |
|
||||
| Artifact resolver with canonical IDs + aliases | **Absent** | Analyze resolves via stage-local source-name switch and fallback helpers. |
|
||||
| Downstream invalidation for forced upstream reruns | **Absent** | TODO only; no stale propagation or status clearing. |
|
||||
| Local semantics mirror archive semantics | **Partial** | Archive/S3 side models runs/current, local workspace core still session-layout-centric. |
|
||||
| Safe path helpers centralization | **Partial** | Good helper base exists, but many stage-level manual joins still encode conventions. |
|
||||
|
||||
---
|
||||
|
||||
## 4. Recommended Implementation Sequence
|
||||
|
||||
## Step 1: Introduce campaign-aware session path model without behavior break
|
||||
|
||||
Purpose:
|
||||
|
||||
- Add first-class helpers for `work/{campaign}/{session}` and make them available everywhere.
|
||||
|
||||
Expected changes:
|
||||
|
||||
- `internal/artifacts`: add/extend path helpers and `SessionPaths` constructor variants that accept campaign.
|
||||
- `internal/app`: pass campaign into path-model entrypoints where available.
|
||||
|
||||
Behavior change:
|
||||
|
||||
- None initially (can keep legacy fallback reads).
|
||||
|
||||
Tests:
|
||||
|
||||
- Add campaign-aware path-model tests.
|
||||
- Keep legacy-path compatibility tests.
|
||||
|
||||
Risks:
|
||||
|
||||
- Wide compile-time touch due `Store` interface signatures.
|
||||
|
||||
Rollback:
|
||||
|
||||
- Keep legacy helper wrappers until full migration lands.
|
||||
|
||||
## Step 2: Split manifest responsibilities (session manifest + run manifest scaffolding)
|
||||
|
||||
Purpose:
|
||||
|
||||
- Preserve current UX while introducing explicit run execution records.
|
||||
|
||||
Expected changes:
|
||||
|
||||
- `internal/manifest`: add run manifest type/store helpers.
|
||||
- `internal/app/runner.go`: create/load session manifest and initialize per-invocation run manifest path.
|
||||
|
||||
Behavior change:
|
||||
|
||||
- Session manifest remains skip truth source.
|
||||
- Run manifest begins recording invocation metadata/stage actions.
|
||||
|
||||
Tests:
|
||||
|
||||
- New tests for both manifest files existing and being updated correctly.
|
||||
|
||||
Risks:
|
||||
|
||||
- Incorrect ordering of saves can regress crash consistency.
|
||||
|
||||
Rollback:
|
||||
|
||||
- Keep session-manifest-only decision logic until run manifest proves stable.
|
||||
|
||||
## Step 3: Move stage execution products to run-local directories with promotion
|
||||
|
||||
Purpose:
|
||||
|
||||
- Align with workspace architecture (`runs/{run_id}/{stage}/...`) while preserving canonical outputs.
|
||||
|
||||
Expected changes:
|
||||
|
||||
- `internal/stage`: each implemented stage writes outputs/logs/config/reports to run-local paths.
|
||||
- Introduce shared promotion helpers (atomic copy/rename + output validation + manifest provenance).
|
||||
|
||||
Behavior change:
|
||||
|
||||
- Canonical outputs remain session-level; run-local diagnostics now preserved per run.
|
||||
|
||||
Tests:
|
||||
|
||||
- Stage tests updated to assert run-local outputs + promoted canonical outputs.
|
||||
- New tests for immediate promotion and producer run metadata.
|
||||
|
||||
Risks:
|
||||
|
||||
- Highest regression risk (all implemented stages touched).
|
||||
|
||||
Rollback:
|
||||
|
||||
- Stage-by-stage migration flag or phased rollout by stage order.
|
||||
|
||||
## Step 4: Archive alignment pass
|
||||
|
||||
Purpose:
|
||||
|
||||
- Remove hybrid fallback complexity once local layout is canonical.
|
||||
|
||||
Expected changes:
|
||||
|
||||
- `internal/stage/archive.go`: resolve sources from canonical session outputs and run manifests deterministically.
|
||||
- Keep `current/manifest.json` + `current/run_id.txt` publication semantics.
|
||||
|
||||
Behavior change:
|
||||
|
||||
- Simpler source selection; fewer cross-layout heuristics.
|
||||
|
||||
Tests:
|
||||
|
||||
- Archive tests for run uploads, promotions, current pointers, and fallback removal/compat gates.
|
||||
|
||||
Risks:
|
||||
|
||||
- Breaking current mixed-layout compatibility too early.
|
||||
|
||||
Rollback:
|
||||
|
||||
- Retain fallback compatibility for one migration window.
|
||||
|
||||
## Step 5: Artifact registry/resolver (analyze first consumer)
|
||||
|
||||
Purpose:
|
||||
|
||||
- Replace ad hoc analyze source resolution with canonical artifact IDs and aliases.
|
||||
|
||||
Expected changes:
|
||||
|
||||
- New resolver package (for example `internal/artifacts/registry` or `internal/stage/artifactresolve`) with IDs:
|
||||
- `narratio.transcript.merged`
|
||||
- `narratio.transcript.polished`
|
||||
- `narratio.transcript.full`
|
||||
- `narratio.transcript.trimmed`
|
||||
- `narratio.bounds.session`
|
||||
- `narratio.artifact.session_recap`
|
||||
- Backward-compatible alias map for current source names.
|
||||
|
||||
Behavior change:
|
||||
|
||||
- Analyze input resolution becomes centralized and consistent.
|
||||
|
||||
Tests:
|
||||
|
||||
- Resolver unit tests for canonical IDs + aliases + missing-input error clarity.
|
||||
- Analyze tests updated to assert resolver usage.
|
||||
|
||||
Risks:
|
||||
|
||||
- Input resolution edge cases for `previous_session_artifact` and required/optional handling.
|
||||
|
||||
Rollback:
|
||||
|
||||
- Keep old resolver path behind a temporary compatibility function.
|
||||
|
||||
## Step 6: Minimal downstream invalidation after forced upstream reruns
|
||||
|
||||
Purpose:
|
||||
|
||||
- Prevent stale downstream skips before full checksum stale detection exists.
|
||||
|
||||
Expected changes:
|
||||
|
||||
- `internal/app/run_control.go` + manifest transition helpers.
|
||||
- On forced rerun success of stage `X`, clear/mark downstream success states.
|
||||
|
||||
Behavior change:
|
||||
|
||||
- Subsequent runs no longer skip stale downstream stages.
|
||||
|
||||
Tests:
|
||||
|
||||
- New run-control tests for force-induced downstream invalidation.
|
||||
- Resume tests with forced sparse runs.
|
||||
|
||||
Risks:
|
||||
|
||||
- Over-invalidating too broadly and degrading UX.
|
||||
|
||||
Rollback:
|
||||
|
||||
- Start with deterministic downstream stage list based on pipeline order only.
|
||||
|
||||
## Step 7: Legacy layout migration strategy
|
||||
|
||||
Purpose:
|
||||
|
||||
- Handle existing `work/{session}` data safely.
|
||||
|
||||
Expected changes:
|
||||
|
||||
- Startup detection/migration path in app layer.
|
||||
- Clear failure messages for ambiguous legacy states.
|
||||
|
||||
Behavior change:
|
||||
|
||||
- Explicit migration semantics instead of implicit fallback drift.
|
||||
|
||||
Tests:
|
||||
|
||||
- Migration detection tests for legacy-only, new-only, and ambiguous layouts.
|
||||
|
||||
Risks:
|
||||
|
||||
- Silent duplication if both layouts are partially populated.
|
||||
|
||||
Rollback:
|
||||
|
||||
- Prefer fail-fast ambiguity policy over auto-merge.
|
||||
|
||||
---
|
||||
|
||||
## 5. Minimal Viable v1.0 Scope
|
||||
|
||||
Must-have for v1.0:
|
||||
|
||||
1. Campaign-aware canonical session roots.
|
||||
2. Session manifest remains skip/resume authority.
|
||||
3. Introduce run manifests + run-local stage records.
|
||||
4. Immediate promotion from run-local outputs to canonical session outputs.
|
||||
5. Minimal downstream invalidation on forced upstream rerun.
|
||||
6. Archive/local semantic alignment with `current/*` behavior preserved.
|
||||
7. Backward-compatible analyze aliases (`processed_transcript`, `normalized_transcript`, `trimmed_transcript`).
|
||||
|
||||
Nice-to-have / defer if risky:
|
||||
|
||||
1. Full checksum-based stale detection graph.
|
||||
2. Broad manifest schema-version migration framework.
|
||||
3. Full artifact-registry rollout beyond analyze’s initial needs.
|
||||
4. Aggressive cleanup of all legacy fallback branches in one release.
|
||||
|
||||
---
|
||||
|
||||
## 6. Open Questions (with Recommendations)
|
||||
|
||||
1. **Should campaign be mandatory for all local path derivation immediately?**
|
||||
- Recommendation: yes for new layout writes; keep controlled read compatibility for legacy session-only roots during migration window.
|
||||
|
||||
2. **Session manifest location transition policy:** auto-migrate vs explicit migrate command?
|
||||
- Recommendation: if user base is small, prefer explicit fail-fast with actionable migration instructions to avoid silent split-brain state.
|
||||
|
||||
3. **Run manifest granularity:** per-stage detailed records vs summary + references?
|
||||
- Recommendation: start with summary + stage status/paths; avoid duplicating full session artifact state to keep write path simple.
|
||||
|
||||
4. **Invalidation marking:** use `stale` status now or clear `succeeded` markers?
|
||||
- Recommendation: if invasive to propagate new status semantics quickly, clear/overwrite downstream success states first; add formal `stale` usage in follow-up.
|
||||
|
||||
5. **Promotion timing:** promote every stage immediately vs delayed at end of run?
|
||||
- Recommendation: immediate per-stage promotion after validation (matches current idempotent skip expectations and simplifies resume behavior).
|
||||
|
||||
6. **Docs consistency order after implementation starts:**
|
||||
- Recommendation: update `README.md` and `architecture.md` in lockstep with each migration step. Current conflict is explicit:
|
||||
- code/README/architecture still primarily describe session roots at `work/{session}`
|
||||
- `docs/development/workspace.md` defines `work/{campaign}/{session}` plus run manifests.
|
||||
|
||||
753
docs/development/workspace.md
Normal file
753
docs/development/workspace.md
Normal file
@@ -0,0 +1,753 @@
|
||||
# Narratio Workspace, Run History, and Artifact Resolution Architecture
|
||||
|
||||
## 1. Purpose
|
||||
|
||||
This document defines the intended v1.0 architecture for Narratio's local workspace layout, run history model, durable session outputs, manifest responsibilities, and artifact resolution contract.
|
||||
|
||||
Narratio is an idempotent session orchestrator. The command:
|
||||
|
||||
```bash
|
||||
narratio run --session-id 2026-05-07
|
||||
```
|
||||
|
||||
means "bring the identified session to its desired completed state." It does **not** mean "always create an entirely new independent output tree and ignore prior session state."
|
||||
|
||||
This distinction drives the architecture:
|
||||
|
||||
* A **session** is the durable domain object and idempotency boundary.
|
||||
* A **run** is an execution attempt that may update the session's durable state.
|
||||
* Durable outputs live at the session level.
|
||||
* Run-specific outputs, logs, generated configs, scratch files, and diagnostics live under `runs/{run_id}/`.
|
||||
* Successful stage outputs are promoted from run-local locations into canonical session-level locations.
|
||||
* The session manifest records current durable state.
|
||||
* Run manifests record execution history and debugging/provenance details.
|
||||
|
||||
This model intentionally mirrors the S3 archive model: session-level current artifacts are distinct from run-record history.
|
||||
|
||||
## 2. Core Concepts
|
||||
|
||||
### 2.1 Session
|
||||
|
||||
A session is the stable unit of work identified by `campaign_id` and `session_id`.
|
||||
|
||||
Examples:
|
||||
|
||||
```text
|
||||
campaign_id = dilfs
|
||||
session_id = 2026-05-07
|
||||
```
|
||||
|
||||
The session directory represents the current durable local state for that session. Re-running Narratio for the same session should consult this state, skip already-completed stages by default, and produce no changes unless work is incomplete, stale, forced, or explicitly selected.
|
||||
|
||||
### 2.2 Run
|
||||
|
||||
A run is a particular execution attempt identified by a generated `run_id`, for example:
|
||||
|
||||
```text
|
||||
20260517T174748Z-abcd1234
|
||||
```
|
||||
|
||||
A run may execute all stages or only a sparse subset of stages. Sparse runs are expected and desirable when the user invokes `--force`, `run-stage`, or a stage-limited command.
|
||||
|
||||
Run directories are provenance/debug records. They should reflect what actually happened during that invocation, not a synthetic complete pipeline layout.
|
||||
|
||||
### 2.3 Durable Output
|
||||
|
||||
A durable output is a canonical session-level artifact intended for later stages, user consumption, archive promotion, or future idempotency decisions.
|
||||
|
||||
Examples:
|
||||
|
||||
```text
|
||||
transcripts/merged.json
|
||||
transcripts/processed.json
|
||||
transcripts/normalized.json
|
||||
transcripts/trimmed.json
|
||||
artifacts/session_recap.md
|
||||
```
|
||||
|
||||
Durable outputs live directly under the session directory, not under a particular run directory.
|
||||
|
||||
### 2.4 Run-Local Output
|
||||
|
||||
A run-local output is the file initially produced by a stage during a specific run. After validation, durable outputs are promoted from run-local paths to session-level canonical paths.
|
||||
|
||||
Run-local outputs, logs, generated configs, reports, and scratch files should remain under:
|
||||
|
||||
```text
|
||||
runs/{run_id}/{stage}/...
|
||||
```
|
||||
|
||||
## 3. Local Workspace Layout
|
||||
|
||||
The canonical local workspace layout is:
|
||||
|
||||
```text
|
||||
{workspace.root}/work/{campaign_id}/{session_id}/
|
||||
manifest.json
|
||||
current/
|
||||
manifest.json
|
||||
run_id.txt
|
||||
inputs/
|
||||
transcripts/
|
||||
artifacts/
|
||||
reports/
|
||||
logs/
|
||||
config/
|
||||
runs/
|
||||
{run_id}/
|
||||
manifest.json
|
||||
prepare/
|
||||
transcribe/
|
||||
merge/
|
||||
polish/
|
||||
normalize/
|
||||
trim/
|
||||
analyze/
|
||||
archive/
|
||||
notify/
|
||||
```
|
||||
|
||||
Not every directory must exist at all times. Directories should be created idempotently when needed.
|
||||
|
||||
### 3.1 Session Root
|
||||
|
||||
The session root is:
|
||||
|
||||
```text
|
||||
{workspace.root}/work/{campaign_id}/{session_id}/
|
||||
```
|
||||
|
||||
The session root is the stable local home for the session. It is the default base for resolving canonical artifact paths.
|
||||
|
||||
The only files that should live directly in the session root are core session-state files, primarily:
|
||||
|
||||
```text
|
||||
manifest.json
|
||||
```
|
||||
|
||||
Lock files may also be session-root scoped if the implementation uses file locks there, but transient locks should not be treated as durable artifacts.
|
||||
|
||||
### 3.2 Session-Level Canonical Directories
|
||||
|
||||
The following directories contain current durable session state:
|
||||
|
||||
```text
|
||||
inputs/
|
||||
transcripts/
|
||||
artifacts/
|
||||
reports/
|
||||
logs/
|
||||
config/
|
||||
current/
|
||||
```
|
||||
|
||||
Recommended meanings:
|
||||
|
||||
| Directory | Purpose |
|
||||
| -------------- | ----------------------------------------------------------------------------- |
|
||||
| `inputs/` | Materialized or copied input files used by the current durable session state. |
|
||||
| `transcripts/` | Canonical transcript tiers. |
|
||||
| `artifacts/` | User-facing and machine-readable generated artifacts. |
|
||||
| `reports/` | Canonical stage reports worth preserving at the session level. |
|
||||
| `logs/` | Optional session-level logs or promoted/latest logs. |
|
||||
| `config/` | Optional session-level generated config snapshots or promoted/latest configs. |
|
||||
| `current/` | Current published session pointers, mirroring the archive backend. |
|
||||
|
||||
Canonical durable outputs should use stable paths under these directories.
|
||||
|
||||
### 3.3 Run History Directory
|
||||
|
||||
Run history lives under:
|
||||
|
||||
```text
|
||||
{workspace.root}/work/{campaign_id}/{session_id}/runs/{run_id}/
|
||||
```
|
||||
|
||||
Each run directory records what happened during that invocation. A run may contain all stage directories or only a sparse subset.
|
||||
|
||||
Example full run:
|
||||
|
||||
```text
|
||||
runs/20260517T174748Z-abcd1234/
|
||||
manifest.json
|
||||
prepare/
|
||||
transcribe/
|
||||
merge/
|
||||
polish/
|
||||
normalize/
|
||||
trim/
|
||||
analyze/
|
||||
archive/
|
||||
notify/
|
||||
```
|
||||
|
||||
Example sparse forced analyze run:
|
||||
|
||||
```text
|
||||
runs/20260518T030000Z-efgh5678/
|
||||
manifest.json
|
||||
analyze/
|
||||
```
|
||||
|
||||
Example sparse polish-through-analyze rerun:
|
||||
|
||||
```text
|
||||
runs/20260518T041500Z-a1b2c3d4/
|
||||
manifest.json
|
||||
polish/
|
||||
normalize/
|
||||
trim/
|
||||
analyze/
|
||||
```
|
||||
|
||||
Run directories should not create stage folders for stages that were not selected, executed, skipped, or otherwise considered during that run unless there is a clear diagnostic reason to do so.
|
||||
|
||||
### 3.4 Stage Run-Local Directories
|
||||
|
||||
Each stage receives a run-local directory:
|
||||
|
||||
```text
|
||||
runs/{run_id}/{stage}/
|
||||
```
|
||||
|
||||
Within that stage directory, the stage may use subdirectories such as:
|
||||
|
||||
```text
|
||||
outputs/
|
||||
logs/
|
||||
reports/
|
||||
config/
|
||||
scratch/
|
||||
```
|
||||
|
||||
For example:
|
||||
|
||||
```text
|
||||
runs/{run_id}/polish/
|
||||
outputs/transcripts/processed.json
|
||||
reports/audita.polish.report.json
|
||||
logs/stdout.log
|
||||
logs/stderr.log
|
||||
config/audita.polish.generated.yml
|
||||
scratch/
|
||||
```
|
||||
|
||||
The exact internal layout of a stage directory may vary by stage, but it should be deterministic, documented, and generated through centralized path helpers rather than ad hoc path joins.
|
||||
|
||||
## 4. Promotion Model
|
||||
|
||||
Narratio uses stage-level promotion with immediate promotion after successful validation.
|
||||
|
||||
The stage lifecycle is:
|
||||
|
||||
1. Resolve required inputs from the current session state and/or run-local context.
|
||||
2. Create the run-local stage directory.
|
||||
3. Execute the stage, writing outputs under `runs/{run_id}/{stage}/...`.
|
||||
4. Validate run-local outputs.
|
||||
5. Promote durable outputs into session-level canonical paths.
|
||||
6. Update the session manifest.
|
||||
7. Update the run manifest.
|
||||
|
||||
Promotion means an atomic or effectively atomic copy/rename from a run-local path to a session-level canonical path.
|
||||
|
||||
Example:
|
||||
|
||||
```text
|
||||
runs/{run_id}/polish/outputs/transcripts/processed.json
|
||||
```
|
||||
|
||||
is promoted to:
|
||||
|
||||
```text
|
||||
transcripts/processed.json
|
||||
```
|
||||
|
||||
Promotion should be safe and deterministic:
|
||||
|
||||
* Validate before promotion.
|
||||
* Write promoted files atomically where possible.
|
||||
* Never leave partially written durable outputs.
|
||||
* Record the producing `run_id` in the session manifest.
|
||||
* Preserve run-local files for debugging unless retention policy deletes them.
|
||||
|
||||
## 5. Promotion Policy: Option A
|
||||
|
||||
Narratio uses immediate stage-level promotion.
|
||||
|
||||
If a selected stage succeeds, its durable outputs are promoted immediately, even if a later selected stage fails.
|
||||
|
||||
Example:
|
||||
|
||||
```bash
|
||||
narratio run --session-id 2026-05-07 --force --stages polish,normalize,trim,analyze
|
||||
```
|
||||
|
||||
If `polish` succeeds and `normalize` fails:
|
||||
|
||||
* `transcripts/processed.json` may be updated from the new run.
|
||||
* `normalize`, `trim`, and `analyze` should not be marked succeeded for the new input state.
|
||||
* Downstream outputs may now be stale relative to the newly promoted polished transcript.
|
||||
|
||||
This policy is simpler, transparent, and consistent with stage-level resumability. It does require explicit stale/invalidation handling.
|
||||
|
||||
## 6. Stale and Invalidation Semantics
|
||||
|
||||
Full checksum-based stale detection may be implemented later. Before that exists, Narratio should still use a simple deterministic invalidation rule for forced or explicit upstream reruns.
|
||||
|
||||
When a stage is successfully re-executed and promoted, downstream stages should be marked stale unless they are also re-executed successfully in the same command invocation.
|
||||
|
||||
Example stage order:
|
||||
|
||||
```text
|
||||
prepare -> transcribe -> merge -> polish -> normalize -> trim -> analyze -> archive -> notify
|
||||
```
|
||||
|
||||
If `polish` is forced and promoted, then the following downstream stages should be invalidated unless rerun successfully:
|
||||
|
||||
```text
|
||||
normalize
|
||||
trim
|
||||
analyze
|
||||
archive
|
||||
notify
|
||||
```
|
||||
|
||||
A stale stage is not equivalent to a failed stage. It means its current durable outputs may no longer correspond to current upstream inputs or configuration.
|
||||
|
||||
Minimum manifest state model:
|
||||
|
||||
```text
|
||||
pending
|
||||
running
|
||||
succeeded
|
||||
failed
|
||||
skipped
|
||||
stale
|
||||
```
|
||||
|
||||
If adding a new `stale` state is too invasive for v1.0, the implementation should at least record stale metadata or clear downstream success markers in a way that prevents accidental idempotent skips based on obsolete outputs.
|
||||
|
||||
## 7. Manifest Responsibilities
|
||||
|
||||
Narratio should distinguish between session manifests and run manifests.
|
||||
|
||||
The same underlying Go types may be reused where practical, but the concepts should remain separate.
|
||||
|
||||
### 7.1 Session Manifest
|
||||
|
||||
Path:
|
||||
|
||||
```text
|
||||
{workspace.root}/work/{campaign_id}/{session_id}/manifest.json
|
||||
```
|
||||
|
||||
The session manifest answers:
|
||||
|
||||
```text
|
||||
What is the current durable state of this session?
|
||||
```
|
||||
|
||||
It should record:
|
||||
|
||||
* campaign ID
|
||||
* session ID
|
||||
* current or latest run ID
|
||||
* current stage states
|
||||
* canonical durable output refs
|
||||
* artifact IDs and paths
|
||||
* producing run ID for each current stage output
|
||||
* relevant input/config checksums when available
|
||||
* stale/invalidated stage information
|
||||
* archive/current publication metadata
|
||||
|
||||
A session's durable state may be a composite of multiple runs.
|
||||
|
||||
For example:
|
||||
|
||||
```text
|
||||
transcripts/merged.json produced by run A
|
||||
transcripts/processed.json produced by run B
|
||||
transcripts/normalized.json produced by run B
|
||||
transcripts/trimmed.json produced by run B
|
||||
artifacts/session_recap.md produced by run C
|
||||
```
|
||||
|
||||
This is valid and expected.
|
||||
|
||||
### 7.2 Run Manifest
|
||||
|
||||
Path:
|
||||
|
||||
```text
|
||||
{workspace.root}/work/{campaign_id}/{session_id}/runs/{run_id}/manifest.json
|
||||
```
|
||||
|
||||
The run manifest answers:
|
||||
|
||||
```text
|
||||
What happened during this specific execution attempt?
|
||||
```
|
||||
|
||||
It should record:
|
||||
|
||||
* run ID
|
||||
* campaign ID
|
||||
* session ID
|
||||
* command mode and selected stages
|
||||
* force flags or stage selection flags
|
||||
* stages considered during this run
|
||||
* stages executed during this run
|
||||
* stages skipped during this run and reasons
|
||||
* run-local output paths
|
||||
* promoted output paths
|
||||
* logs
|
||||
* reports
|
||||
* generated configs
|
||||
* timings
|
||||
* errors
|
||||
* non-secret subprocess invocation metadata
|
||||
|
||||
Run manifests are primarily for debugging, auditability, and archive history.
|
||||
|
||||
## 8. Idempotency and Resume Behavior
|
||||
|
||||
The idempotency boundary is the session, not the run.
|
||||
|
||||
By default:
|
||||
|
||||
```bash
|
||||
narratio run --session-id 2026-05-07
|
||||
```
|
||||
|
||||
should consult the session manifest and skip stages that are already succeeded and not stale.
|
||||
|
||||
If all stages are already complete, the command should execute zero stages and report that the session is already complete.
|
||||
|
||||
Forced execution creates a new run record but updates session-level durable state only for stages that actually succeed and promote outputs.
|
||||
|
||||
Examples:
|
||||
|
||||
```bash
|
||||
narratio run --session-id 2026-05-07 --force
|
||||
```
|
||||
|
||||
Creates a new run and attempts to re-execute the selected/default stage set.
|
||||
|
||||
```bash
|
||||
narratio run-stage --session-id 2026-05-07 analyze --force
|
||||
```
|
||||
|
||||
Creates a sparse run that executes only `analyze`, then promotes updated analysis artifacts if successful.
|
||||
|
||||
```bash
|
||||
narratio resume --session-id 2026-05-07
|
||||
```
|
||||
|
||||
Uses the session manifest to determine what remains incomplete or stale. Resume does not need to resume the same `run_id` unless the implementation explicitly supports resuming an interrupted active run.
|
||||
|
||||
## 9. Artifact Resolution Contract
|
||||
|
||||
Narratio should provide a first-class artifact registry and resolver.
|
||||
|
||||
The resolver maps symbolic artifact source names to canonical session-level paths and manifest output kinds.
|
||||
|
||||
Stages and adapters should not hardcode path fragments when resolving cross-stage inputs. They should ask the artifact resolver for the current durable artifact by ID.
|
||||
|
||||
### 9.1 Canonical Artifact IDs
|
||||
|
||||
Preferred artifact IDs should be namespaced:
|
||||
|
||||
```text
|
||||
narratio.transcript.merged
|
||||
narratio.transcript.polished
|
||||
narratio.transcript.full
|
||||
narratio.transcript.trimmed
|
||||
narratio.bounds.session
|
||||
narratio.artifact.session_recap
|
||||
```
|
||||
|
||||
Recommended initial registry:
|
||||
|
||||
| Artifact ID | Canonical Path | Producer Stage | Output Kind | Meaning |
|
||||
| --------------------------------- | ------------------------------- | -------------- | ------------------------ | ------------------------------------- |
|
||||
| `narratio.transcript.merged` | `transcripts/merged.json` | `merge` | `transcript_merged` | Deterministic Seriatim merge. |
|
||||
| `narratio.transcript.polished` | `transcripts/processed.json` | `polish` | `transcript_processed` | Full Audita-polished transcript. |
|
||||
| `narratio.transcript.full` | `transcripts/normalized.json` | `normalize` | `transcript_normalized` | Preferred full normalized transcript. |
|
||||
| `narratio.transcript.trimmed` | `transcripts/trimmed.json` | `trim` | `transcript_trimmed` | Gameplay-only transcript. |
|
||||
| `narratio.bounds.session` | `artifacts/session_bounds.json` | `trim` | `session_bounds` | Trim bounds selected for the session. |
|
||||
| `narratio.artifact.session_recap` | `artifacts/session_recap.md` | `analyze` | `artifact_session_recap` | Generated session recap. |
|
||||
|
||||
### 9.2 Backward-Compatible Aliases
|
||||
|
||||
Existing source names should remain supported:
|
||||
|
||||
| Legacy Source | Preferred Artifact ID |
|
||||
| ----------------------- | ------------------------------ |
|
||||
| `processed_transcript` | `narratio.transcript.polished` |
|
||||
| `normalized_transcript` | `narratio.transcript.full` |
|
||||
| `trimmed_transcript` | `narratio.transcript.trimmed` |
|
||||
|
||||
These aliases may be supported silently for v1.0. Documentation should prefer namespaced IDs.
|
||||
|
||||
### 9.3 Resolver Behavior
|
||||
|
||||
Artifact resolution should follow this order:
|
||||
|
||||
1. Normalize aliases to canonical artifact IDs.
|
||||
2. Look for a current output reference in the session manifest.
|
||||
3. Fall back to the canonical session-level path.
|
||||
4. If the artifact is required, fail clearly if missing.
|
||||
5. If the artifact is optional and missing, omit it from the downstream invocation.
|
||||
6. Validate the artifact using the expected content validator.
|
||||
7. Return a resolved artifact record containing ID, path, producer stage, output kind, and provenance.
|
||||
|
||||
Example conceptual result:
|
||||
|
||||
```json
|
||||
{
|
||||
"id": "narratio.transcript.trimmed",
|
||||
"path": "/var/lib/narratio/work/dilfs/2026-05-07/transcripts/trimmed.json",
|
||||
"producer_stage": "trim",
|
||||
"producer_run_id": "20260517T174748Z-abcd1234",
|
||||
"output_kind": "transcript_trimmed",
|
||||
"content_type": "application/json"
|
||||
}
|
||||
```
|
||||
|
||||
### 9.4 Artifact Validation
|
||||
|
||||
Transcript artifacts must be valid JSON with a top-level `segments` array.
|
||||
|
||||
Markdown/text artifacts must exist and be non-empty when required.
|
||||
|
||||
Bounds artifacts must match the expected bounds schema and refer to segment IDs in the same transcript ID space used by the trim stage.
|
||||
|
||||
Validation should happen before a resolved artifact is passed to another stage or external subprocess.
|
||||
|
||||
## 10. Analyze Stage Implications
|
||||
|
||||
The analyze stage should consume artifacts through the artifact resolver.
|
||||
|
||||
Preferred Scriptorium config shape:
|
||||
|
||||
```yaml
|
||||
scriptorium:
|
||||
artifacts:
|
||||
session_recap:
|
||||
enabled: true
|
||||
prompt_id: "dnd.session_recap"
|
||||
output_path: "artifacts/session_recap.md"
|
||||
inputs:
|
||||
transcript:
|
||||
source: "narratio.transcript.trimmed"
|
||||
required: true
|
||||
```
|
||||
|
||||
Additional artifacts can choose different transcript tiers:
|
||||
|
||||
```yaml
|
||||
scriptorium:
|
||||
artifacts:
|
||||
table_summary:
|
||||
enabled: true
|
||||
prompt_id: "dnd.table_summary"
|
||||
output_path: "artifacts/table_summary.md"
|
||||
inputs:
|
||||
transcript:
|
||||
source: "narratio.transcript.full"
|
||||
required: true
|
||||
```
|
||||
|
||||
For v1.0, Narratio does not need a generic DAG engine. It may execute configured analyze artifacts in deterministic order and allow later artifacts to consume earlier artifacts only when that relationship is explicit and unambiguous.
|
||||
|
||||
Rules:
|
||||
|
||||
* Artifact inputs resolve from current session-level durable state.
|
||||
* Outputs are first written run-locally.
|
||||
* Successful analyze outputs are promoted to session-level `artifacts/` paths.
|
||||
* Manifest output refs record the producing run ID.
|
||||
* Optional inputs are omitted when unavailable.
|
||||
* Required missing inputs fail before invoking Scriptorium.
|
||||
|
||||
## 11. Archive Alignment
|
||||
|
||||
Local workspace semantics should mirror archive semantics.
|
||||
|
||||
Local session-level durable paths:
|
||||
|
||||
```text
|
||||
work/{campaign}/{session}/transcripts/trimmed.json
|
||||
work/{campaign}/{session}/artifacts/session_recap.md
|
||||
work/{campaign}/{session}/current/manifest.json
|
||||
work/{campaign}/{session}/current/run_id.txt
|
||||
work/{campaign}/{session}/runs/{run_id}/...
|
||||
```
|
||||
|
||||
should map naturally to remote archive paths:
|
||||
|
||||
```text
|
||||
{root_prefix}/campaigns/{campaign}/sessions/{session}/transcripts/trimmed.json
|
||||
{root_prefix}/campaigns/{campaign}/sessions/{session}/artifacts/session_recap.md
|
||||
{root_prefix}/campaigns/{campaign}/sessions/{session}/current/manifest.json
|
||||
{root_prefix}/campaigns/{campaign}/sessions/{session}/current/run_id.txt
|
||||
{root_prefix}/campaigns/{campaign}/sessions/{session}/runs/{run_id}/...
|
||||
```
|
||||
|
||||
The archive stage should publish run records and promoted current artifacts consistently with the local model.
|
||||
|
||||
`current/run_id.txt` remains the effective commit marker for the archived current session state.
|
||||
|
||||
## 12. Path Helper Requirements
|
||||
|
||||
All code should use centralized path helpers for workspace paths.
|
||||
|
||||
Stage code should not manually assemble durable cross-stage paths using raw string joins except through the path model.
|
||||
|
||||
Recommended helper surface:
|
||||
|
||||
```text
|
||||
SessionRoot(campaignID, sessionID)
|
||||
SessionManifestPath(campaignID, sessionID)
|
||||
SessionCurrentDir(campaignID, sessionID)
|
||||
SessionTranscriptsDir(campaignID, sessionID)
|
||||
SessionArtifactsDir(campaignID, sessionID)
|
||||
SessionReportsDir(campaignID, sessionID)
|
||||
SessionLogsDir(campaignID, sessionID)
|
||||
SessionConfigDir(campaignID, sessionID)
|
||||
RunsDir(campaignID, sessionID)
|
||||
RunRoot(campaignID, sessionID, runID)
|
||||
RunManifestPath(campaignID, sessionID, runID)
|
||||
RunStageDir(campaignID, sessionID, runID, stage)
|
||||
RunStageOutputsDir(campaignID, sessionID, runID, stage)
|
||||
RunStageLogsDir(campaignID, sessionID, runID, stage)
|
||||
RunStageReportsDir(campaignID, sessionID, runID, stage)
|
||||
RunStageConfigDir(campaignID, sessionID, runID, stage)
|
||||
CanonicalArtifactPath(campaignID, sessionID, artifactID)
|
||||
```
|
||||
|
||||
Path helpers should enforce safe relative paths for configured output paths:
|
||||
|
||||
* reject absolute paths unless explicitly allowed for a particular config field
|
||||
* reject `..` traversal
|
||||
* normalize separators
|
||||
* preserve deterministic output paths
|
||||
|
||||
## 13. Directory Creation Policy
|
||||
|
||||
Directory creation should be centralized and idempotent.
|
||||
|
||||
Recommended policy:
|
||||
|
||||
* `prepare` ensures the baseline session directory structure exists.
|
||||
* Every stage also calls shared layout helpers to ensure its required run-local directories exist before writing.
|
||||
* `run-stage` should not depend on a prior `prepare` invocation merely to create folders.
|
||||
* Missing directories should be created with appropriate permissions.
|
||||
* Directory creation should not imply stage success.
|
||||
|
||||
This provides consistent layout while keeping direct stage execution robust.
|
||||
|
||||
## 14. Cleanup and Retention
|
||||
|
||||
Cleanup must preserve the distinction between durable session state and run history.
|
||||
|
||||
Workspace cleanup after successful archive may remove selected local directories only according to explicit configuration.
|
||||
|
||||
Potential retention policies:
|
||||
|
||||
```text
|
||||
keep_all_runs
|
||||
keep_failed_runs
|
||||
keep_last_n_runs
|
||||
delete_run_after_success
|
||||
```
|
||||
|
||||
For v1.0, conservative retention is preferred:
|
||||
|
||||
* Do not delete durable session-level outputs unless explicitly requested.
|
||||
* Do not delete failed run directories by default.
|
||||
* If cleanup is enabled, remove only documented run-scoped or spool-scoped paths.
|
||||
* Local development audio inputs must never be deleted by workspace cleanup.
|
||||
|
||||
## 15. Migration From Existing Layout
|
||||
|
||||
Existing installations may currently use a simpler path such as:
|
||||
|
||||
```text
|
||||
{workspace.root}/work/{session_id}/manifest.json
|
||||
```
|
||||
|
||||
The v1.0 layout introduces campaign-aware session roots:
|
||||
|
||||
```text
|
||||
{workspace.root}/work/{campaign_id}/{session_id}/manifest.json
|
||||
```
|
||||
|
||||
Migration options:
|
||||
|
||||
1. Best-effort automatic discovery of legacy session manifests.
|
||||
2. A one-time migration command.
|
||||
3. Clear release notes requiring users to move or regenerate workspace state.
|
||||
|
||||
For v1.0, it is acceptable to require explicit migration if the user base is small and the archive contains the authoritative durable outputs. However, the application should fail clearly when it detects an ambiguous legacy layout rather than silently creating duplicate state.
|
||||
|
||||
## 16. Documentation Updates Required
|
||||
|
||||
The following documentation should be updated to reflect this architecture:
|
||||
|
||||
* `README.md`
|
||||
* `docs/architecture.md`
|
||||
* a dedicated workspace/run-history document, such as this file
|
||||
* S3/archive documentation
|
||||
* analyze/artifact configuration documentation
|
||||
* example pipeline files
|
||||
|
||||
Documentation should consistently use the following terms:
|
||||
|
||||
| Term | Meaning |
|
||||
| ---------------- | ---------------------------------------------------------------------- |
|
||||
| Session | Durable domain object and idempotency boundary. |
|
||||
| Run | Execution attempt that may update session state. |
|
||||
| Durable output | Canonical current session-level output. |
|
||||
| Run-local output | Output produced inside a specific run directory before promotion. |
|
||||
| Promotion | Validated copy/rename from run-local output to durable session output. |
|
||||
| Session manifest | Current durable state of the session. |
|
||||
| Run manifest | Execution record for a particular run. |
|
||||
| Artifact ID | Symbolic source name resolved by the artifact registry. |
|
||||
|
||||
## 17. Architectural Invariants
|
||||
|
||||
The following invariants should hold after implementation:
|
||||
|
||||
1. `session_id` remains the idempotency boundary for normal operator commands.
|
||||
2. `run_id` identifies an execution attempt, not the primary durable workspace.
|
||||
3. Session-level canonical artifacts are the default inputs for downstream stages.
|
||||
4. Run-local outputs are promoted only after validation.
|
||||
5. A session's current durable state may be composed of outputs from multiple runs.
|
||||
6. Sparse run directories are valid and expected.
|
||||
7. The session manifest records current stage/artifact state and producer run IDs.
|
||||
8. The run manifest records what happened during one invocation.
|
||||
9. Artifact consumers resolve symbolic artifact IDs through a registry/resolver.
|
||||
10. Local workspace semantics mirror S3 archive semantics.
|
||||
11. Directory creation is centralized and idempotent.
|
||||
12. Stage code uses path helpers rather than ad hoc path construction.
|
||||
13. Forced upstream reruns invalidate downstream stage success unless downstream stages are rerun successfully.
|
||||
14. Cleanup never removes durable session outputs or local development inputs unless explicitly configured to do so.
|
||||
|
||||
## 18. Implementation Guidance
|
||||
|
||||
A practical implementation sequence is:
|
||||
|
||||
1. Add this architecture document.
|
||||
2. Add or revise path model helpers for session roots, run roots, stage directories, and canonical artifact paths.
|
||||
3. Introduce session manifest versus run manifest concepts.
|
||||
4. Route stage outputs through run-local directories.
|
||||
5. Add promotion helpers with validation and atomic writes.
|
||||
6. Update existing stages to promote durable outputs to session-level canonical paths.
|
||||
7. Add artifact registry and resolver.
|
||||
8. Update analyze to use artifact IDs and aliases.
|
||||
9. Add simple downstream stale invalidation for forced upstream reruns.
|
||||
10. Align archive/local path behavior and documentation.
|
||||
11. Update examples and README.
|
||||
12. Add tests for idempotency, sparse forced runs, promotion, manifest provenance, and artifact resolution.
|
||||
|
||||
This sequence intentionally avoids introducing a generic DAG engine. The v1.0 goal is a clear, deterministic, stage-oriented orchestrator with stable session-level outputs and inspectable run history.
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,35 +0,0 @@
|
||||
# S3 Archive Foundations Runbook
|
||||
|
||||
This runbook documents the currently implemented storage/archive foundations and the boundaries of current behavior.
|
||||
|
||||
## Implemented Now
|
||||
|
||||
- config modeling for:
|
||||
- `pipeline.storage.s3`
|
||||
- `pipeline.spool`
|
||||
- `pipeline.archive`
|
||||
- `session.inputs.audio_s3`
|
||||
- promotion rule validation for safe relative paths
|
||||
- run ID generation and path/key helper functions
|
||||
- manifest run/path identity fields
|
||||
- remote storage backend layer:
|
||||
- object-store interface (`List`, `Download`, `Upload`, `Exists`)
|
||||
- fake backend for deterministic tests
|
||||
- S3-compatible backend using AWS SDK v2
|
||||
- config-based backend construction helper
|
||||
|
||||
## Not Implemented Yet
|
||||
|
||||
- prepare-stage S3 object listing or download
|
||||
- archive-stage S3 upload or promotion writes
|
||||
- writing `current/manifest.json` or `current/run_id.txt` in S3
|
||||
|
||||
## Operational Notes
|
||||
|
||||
- local audio workflows remain the active development path (`audio_dir` or `audio_files`)
|
||||
- `audio_s3` and local audio config are mutually exclusive
|
||||
- do not place AWS credentials in Narratio config files
|
||||
|
||||
## Next Implementation Target
|
||||
|
||||
Use the storage backend layer in prepare-stage session audio discovery/download flow, while preserving local audio input support.
|
||||
@@ -20,6 +20,8 @@ Not implemented:
|
||||
|
||||
- `storage.s3.bucket` must be set when S3 audio input is used.
|
||||
- `storage.s3.root_prefix` defaults to `dnd`.
|
||||
- `storage.s3.access_key_id_env` defaults to `OBJECT_STORAGE_KEY_ID`.
|
||||
- `storage.s3.secret_access_key_env` defaults to `OBJECT_STORAGE_KEY`.
|
||||
- `spool.root` defaults to `/var/spool/narratio`.
|
||||
|
||||
`session.yml`:
|
||||
@@ -1,58 +0,0 @@
|
||||
# Storage Backends
|
||||
|
||||
This document describes the currently implemented remote object storage backend layer used by Narratio, and its intended role in later prepare/archive work.
|
||||
|
||||
## Implemented
|
||||
|
||||
Remote object store abstraction:
|
||||
|
||||
- `List(ctx, prefix)`
|
||||
- `Download(ctx, key, localPath)`
|
||||
- `Upload(ctx, localPath, key, opts)`
|
||||
- `Exists(ctx, key)`
|
||||
|
||||
Object metadata model includes:
|
||||
|
||||
- key
|
||||
- size
|
||||
- ETag (provider metadata only)
|
||||
- last modified time when available
|
||||
|
||||
Backends:
|
||||
|
||||
- fake storage backend for deterministic tests
|
||||
- S3-compatible backend implemented with AWS SDK for Go v2
|
||||
|
||||
Construction:
|
||||
|
||||
- config-based constructor builds S3 backend from `pipeline.storage.s3` values:
|
||||
- bucket
|
||||
- region
|
||||
- endpoint
|
||||
- force_path_style
|
||||
|
||||
## Key Invariant
|
||||
|
||||
- callers pass full bucket-relative object keys
|
||||
- storage backends do not prepend `root_prefix`
|
||||
- storage backends do not infer campaign/session/run paths
|
||||
|
||||
S3 session/run key builders remain separate and continue to live outside backend implementations.
|
||||
|
||||
## Security Boundary
|
||||
|
||||
- do not store AWS credentials in Narratio config
|
||||
- AWS credentials are resolved through standard AWS SDK credential chains
|
||||
- AWS SDK-specific types remain isolated to the storage adapter package
|
||||
|
||||
## Testing
|
||||
|
||||
- fake storage tests cover list/download/upload/exists and error paths
|
||||
- S3 backend tests use injected fake S3 API clients
|
||||
- tests do not require live S3 services, AWS credentials, or network access
|
||||
|
||||
## Not Implemented Yet
|
||||
|
||||
- prepare-stage S3 object listing or downloads
|
||||
- archive-stage S3 uploads or promotion writes
|
||||
- writing `current/manifest.json` or `current/run_id.txt` to S3
|
||||
@@ -8,6 +8,9 @@ storage:
|
||||
bucket: "my-dnd-archive"
|
||||
root_prefix: "dnd"
|
||||
region: "us-east-1"
|
||||
# Optional credential env-var names (defaulted when omitted):
|
||||
# access_key_id_env: "OBJECT_STORAGE_KEY_ID"
|
||||
# secret_access_key_env: "OBJECT_STORAGE_KEY"
|
||||
|
||||
spool:
|
||||
root: "/var/spool/narratio"
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"time"
|
||||
|
||||
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/aws/smithy-go"
|
||||
@@ -35,6 +36,8 @@ type s3ClientOptions struct {
|
||||
Region string
|
||||
Endpoint string
|
||||
ForcePathStyle bool
|
||||
AccessKeyID string
|
||||
SecretKey string
|
||||
}
|
||||
|
||||
var newS3Client = func(ctx context.Context, opts s3ClientOptions) (s3API, error) {
|
||||
@@ -42,6 +45,15 @@ var newS3Client = func(ctx context.Context, opts s3ClientOptions) (s3API, error)
|
||||
if strings.TrimSpace(opts.Region) != "" {
|
||||
loadOpts = append(loadOpts, awsconfig.WithRegion(strings.TrimSpace(opts.Region)))
|
||||
}
|
||||
if strings.TrimSpace(opts.AccessKeyID) != "" && strings.TrimSpace(opts.SecretKey) != "" {
|
||||
loadOpts = append(loadOpts, awsconfig.WithCredentialsProvider(
|
||||
credentials.NewStaticCredentialsProvider(
|
||||
strings.TrimSpace(opts.AccessKeyID),
|
||||
strings.TrimSpace(opts.SecretKey),
|
||||
"",
|
||||
),
|
||||
))
|
||||
}
|
||||
awsCfg, err := awsconfig.LoadDefaultConfig(ctx, loadOpts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load aws config: %w", err)
|
||||
@@ -67,6 +79,8 @@ func NewS3BackendFromConfig(ctx context.Context, cfg config.StorageS3Config) (*S
|
||||
Region: cfg.Region,
|
||||
Endpoint: cfg.Endpoint,
|
||||
ForcePathStyle: cfg.ForcePathStyle,
|
||||
AccessKeyID: s3CredentialFromEnv(orDefaultEnvName(cfg.AccessKeyIDEnv, config.DefaultS3AccessKeyIDEnv)),
|
||||
SecretKey: s3CredentialFromEnv(orDefaultEnvName(cfg.SecretKeyEnv, config.DefaultS3SecretAccessKeyEnv)),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("build s3 client: %w", err)
|
||||
@@ -78,6 +92,26 @@ func NewS3BackendFromConfig(ctx context.Context, cfg config.StorageS3Config) (*S
|
||||
}, nil
|
||||
}
|
||||
|
||||
func s3CredentialFromEnv(envVarName string) string {
|
||||
name := strings.TrimSpace(envVarName)
|
||||
if name == "" {
|
||||
return ""
|
||||
}
|
||||
value, ok := os.LookupEnv(name)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
return strings.TrimSpace(value)
|
||||
}
|
||||
|
||||
func orDefaultEnvName(name, fallback string) string {
|
||||
trimmed := strings.TrimSpace(name)
|
||||
if trimmed == "" {
|
||||
return fallback
|
||||
}
|
||||
return trimmed
|
||||
}
|
||||
|
||||
// List returns objects under prefix.
|
||||
func (b *S3Backend) List(ctx context.Context, prefix string) ([]ObjectInfo, error) {
|
||||
normalizedPrefix := normalizeObjectKey(prefix)
|
||||
|
||||
@@ -187,6 +187,8 @@ func TestS3BackendExistsNotFound(t *testing.T) {
|
||||
func TestNewS3BackendFromConfigUsesClientOptions(t *testing.T) {
|
||||
original := newS3Client
|
||||
t.Cleanup(func() { newS3Client = original })
|
||||
t.Setenv("OBJECT_STORAGE_KEY_ID", "id-123")
|
||||
t.Setenv("OBJECT_STORAGE_KEY", "secret-abc")
|
||||
|
||||
var got s3ClientOptions
|
||||
newS3Client = func(_ context.Context, opts s3ClientOptions) (s3API, error) {
|
||||
@@ -209,6 +211,9 @@ func TestNewS3BackendFromConfigUsesClientOptions(t *testing.T) {
|
||||
if got.Region != "us-east-1" || got.Endpoint != "http://localhost:9000" || !got.ForcePathStyle {
|
||||
t.Fatalf("client options = %#v, want region/endpoint/path-style values", got)
|
||||
}
|
||||
if got.AccessKeyID != "id-123" || got.SecretKey != "secret-abc" {
|
||||
t.Fatalf("client options credentials = %#v, want env-resolved static credentials", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewS3BackendFromConfigRequiresBucket(t *testing.T) {
|
||||
@@ -218,6 +223,30 @@ func TestNewS3BackendFromConfigRequiresBucket(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewS3BackendFromConfigFallsBackWhenCredentialEnvMissing(t *testing.T) {
|
||||
original := newS3Client
|
||||
t.Cleanup(func() { newS3Client = original })
|
||||
|
||||
var got s3ClientOptions
|
||||
newS3Client = func(_ context.Context, opts s3ClientOptions) (s3API, error) {
|
||||
got = opts
|
||||
return &fakeS3API{}, nil
|
||||
}
|
||||
|
||||
_, err := NewS3BackendFromConfig(context.Background(), config.StorageS3Config{
|
||||
Bucket: "my-archive",
|
||||
Region: "us-east-1",
|
||||
AccessKeyIDEnv: "MISSING_ACCESS_KEY_ID",
|
||||
SecretKeyEnv: "MISSING_SECRET_KEY",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewS3BackendFromConfig() error = %v", err)
|
||||
}
|
||||
if got.AccessKeyID != "" || got.SecretKey != "" {
|
||||
t.Fatalf("client options credentials = %#v, want empty fallback values", got)
|
||||
}
|
||||
}
|
||||
|
||||
func strPtr(v string) *string { return &v }
|
||||
func int64Ptr(v int64) *int64 { return &v }
|
||||
|
||||
|
||||
@@ -2,6 +2,8 @@ package artifacts
|
||||
|
||||
import (
|
||||
"path/filepath"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||
)
|
||||
|
||||
// SessionPaths contains canonical local paths for one session work directory.
|
||||
@@ -22,34 +24,33 @@ type SessionPaths struct {
|
||||
|
||||
// SessionWorkDir returns the work directory for one session.
|
||||
func SessionWorkDir(rootDir, sessionID string) string {
|
||||
return filepath.Join(rootDir, "work", sessionID)
|
||||
return filepath.Join(rootDir, config.PathWorkDirSegment, sessionID)
|
||||
}
|
||||
|
||||
// SessionRunWorkDir returns the campaign/session/run scoped local work directory.
|
||||
func SessionRunWorkDir(rootDir, campaign, sessionID, runID string) string {
|
||||
return filepath.Join(rootDir, "work", campaign, sessionID, runID)
|
||||
return filepath.Join(rootDir, config.PathWorkDirSegment, campaign, sessionID, runID)
|
||||
}
|
||||
|
||||
// SessionSpoolAudioDir returns the campaign/session/run scoped local spool audio path.
|
||||
func SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID string) string {
|
||||
return filepath.Join(spoolRoot, campaign, sessionID, runID, "audio")
|
||||
return filepath.Join(spoolRoot, campaign, sessionID, runID, config.PathAudioDirSegment)
|
||||
}
|
||||
|
||||
func buildSessionPaths(workspaceRoot, sessionID string) SessionPaths {
|
||||
root := SessionWorkDir(workspaceRoot, sessionID)
|
||||
transcripts := filepath.Join(root, "transcripts")
|
||||
return SessionPaths{
|
||||
WorkspaceRoot: workspaceRoot,
|
||||
Root: root,
|
||||
InputsDir: filepath.Join(root, "inputs"),
|
||||
AudioDir: filepath.Join(root, "audio"),
|
||||
TranscriptsDir: transcripts,
|
||||
TranscriptsRawDir: filepath.Join(transcripts, "raw"),
|
||||
TranscriptsTrimmedDir: filepath.Join(transcripts, "trimmed"),
|
||||
ArtifactsDir: filepath.Join(root, "artifacts"),
|
||||
ConfigDir: filepath.Join(root, "config"),
|
||||
LogsDir: filepath.Join(root, "logs"),
|
||||
ManifestPath: filepath.Join(root, "manifest.json"),
|
||||
LockPath: filepath.Join(root, ".lock"),
|
||||
InputsDir: filepath.Join(root, config.PathInputsDirSegment),
|
||||
AudioDir: filepath.Join(root, config.PathAudioDirSegment),
|
||||
TranscriptsDir: filepath.Join(root, config.PathTranscriptsSegment),
|
||||
TranscriptsRawDir: filepath.Join(root, filepath.FromSlash(config.PathTranscriptsRaw)),
|
||||
TranscriptsTrimmedDir: filepath.Join(root, filepath.FromSlash(config.PathTranscriptsTrimmed)),
|
||||
ArtifactsDir: filepath.Join(root, config.PathArtifactsDirSegment),
|
||||
ConfigDir: filepath.Join(root, config.PathConfigDirSegment),
|
||||
LogsDir: filepath.Join(root, config.PathLogsDirSegment),
|
||||
ManifestPath: filepath.Join(root, config.PathManifestFile),
|
||||
LockPath: filepath.Join(root, config.PathLockFile),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,8 @@ package artifacts
|
||||
import (
|
||||
"path"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||
)
|
||||
|
||||
// S3SessionPrefix builds the canonical S3 session prefix.
|
||||
@@ -10,9 +12,9 @@ import (
|
||||
func S3SessionPrefix(rootPrefix, campaign, sessionID string) string {
|
||||
prefix := path.Join(
|
||||
cleanS3PathPart(rootPrefix),
|
||||
"campaigns",
|
||||
config.S3CampaignsSegment,
|
||||
cleanS3PathPart(campaign),
|
||||
"sessions",
|
||||
config.S3SessionsSegment,
|
||||
cleanS3PathPart(sessionID),
|
||||
)
|
||||
return ensureS3TrailingSlash(prefix)
|
||||
@@ -21,7 +23,7 @@ func S3SessionPrefix(rootPrefix, campaign, sessionID string) string {
|
||||
// S3RunPrefix builds the canonical S3 run prefix.
|
||||
// Format: {session_prefix}/runs/{run_id}/
|
||||
func S3RunPrefix(sessionPrefix, runID string) string {
|
||||
prefix := path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "runs", cleanS3PathPart(runID))
|
||||
prefix := path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), config.S3RunsSegment, cleanS3PathPart(runID))
|
||||
return ensureS3TrailingSlash(prefix)
|
||||
}
|
||||
|
||||
@@ -35,13 +37,13 @@ func S3AudioPrefix(sessionPrefix, audioPrefix string) string {
|
||||
// S3CurrentManifestKey returns the current manifest pointer key.
|
||||
// Format: {session_prefix}/current/manifest.json
|
||||
func S3CurrentManifestKey(sessionPrefix string) string {
|
||||
return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "current", "manifest.json")
|
||||
return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), config.S3CurrentSegment, config.S3ManifestFile)
|
||||
}
|
||||
|
||||
// S3CurrentRunPointerKey returns the current run pointer key.
|
||||
// Format: {session_prefix}/current/run_id.txt
|
||||
func S3CurrentRunPointerKey(sessionPrefix string) string {
|
||||
return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), "current", "run_id.txt")
|
||||
return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), config.S3CurrentSegment, config.S3RunIDFile)
|
||||
}
|
||||
|
||||
// S3PromotedArtifactKey returns the destination key for one promoted artifact.
|
||||
|
||||
@@ -60,6 +60,8 @@ type StorageS3Config struct {
|
||||
Region string `yaml:"region"`
|
||||
Endpoint string `yaml:"endpoint"`
|
||||
ForcePathStyle bool `yaml:"force_path_style"`
|
||||
AccessKeyIDEnv string `yaml:"access_key_id_env"`
|
||||
SecretKeyEnv string `yaml:"secret_access_key_env"`
|
||||
}
|
||||
|
||||
// SpoolConfig configures local spool storage for staged data.
|
||||
|
||||
@@ -8,8 +8,71 @@ const (
|
||||
DefaultSessionConfigPathLocal = "./session.yml"
|
||||
DefaultSessionConfigPathUsrLocal = "/usr/local/etc/narratio/session.yml"
|
||||
DefaultSessionConfigPathEtc = "/etc/narratio/session.yml"
|
||||
DefaultS3AccessKeyIDEnv = "OBJECT_STORAGE_KEY_ID"
|
||||
DefaultS3SecretAccessKeyEnv = "OBJECT_STORAGE_KEY"
|
||||
DefaultStorageS3RootPrefix = "dnd"
|
||||
DefaultSpoolRoot = "/var/spool/narratio"
|
||||
|
||||
DefaultWhisperXLanguage = "en"
|
||||
DefaultWhisperXTimeout = "30m"
|
||||
DefaultWhisperXRetryDelay = "2s"
|
||||
DefaultWhisperXConcurrency = 2
|
||||
DefaultWhisperXRetries = 3
|
||||
|
||||
DefaultSeriatimBinary = "seriatim"
|
||||
DefaultSeriatimTimeout = "10m"
|
||||
DefaultSeriatimOutputSchema = "seriatim-intermediate"
|
||||
DefaultSeriatimCoalesceGap = 3.0
|
||||
DefaultSeriatimReport = true
|
||||
|
||||
DefaultAuditaBinary = "audita"
|
||||
DefaultAuditaTimeout = "3h"
|
||||
DefaultAuditaReport = true
|
||||
|
||||
DefaultScriptoriumBinary = "scriptorium"
|
||||
DefaultScriptoriumTimeout = "10m"
|
||||
|
||||
DefaultTrimBoundsTimeout = "10m"
|
||||
DefaultTrimSeriatimReport = false
|
||||
|
||||
DefaultNormalizeOutputPath = "transcripts/normalized.json"
|
||||
DefaultNormalizeOutputSchema = "seriatim-intermediate"
|
||||
DefaultNormalizeReport = true
|
||||
|
||||
DefaultArchiveEnabled = true
|
||||
DefaultArchiveUploadRun = true
|
||||
|
||||
PathWorkDirSegment = "work"
|
||||
PathInputsDirSegment = "inputs"
|
||||
PathAudioDirSegment = "audio"
|
||||
PathTranscriptsSegment = "transcripts"
|
||||
PathTranscriptsRaw = "transcripts/raw"
|
||||
PathTranscriptsTrimmed = "transcripts/trimmed"
|
||||
PathArtifactsDirSegment = "artifacts"
|
||||
PathConfigDirSegment = "config"
|
||||
PathLogsDirSegment = "logs"
|
||||
PathManifestFile = "manifest.json"
|
||||
PathLockFile = ".lock"
|
||||
PathTranscriptMerged = "transcripts/merged.json"
|
||||
PathTranscriptProcessed = "transcripts/processed.json"
|
||||
PathTranscriptNormalized = "transcripts/normalized.json"
|
||||
PathTranscriptTrimmed = "transcripts/trimmed.json"
|
||||
|
||||
S3CampaignsSegment = "campaigns"
|
||||
S3SessionsSegment = "sessions"
|
||||
S3RunsSegment = "runs"
|
||||
S3CurrentSegment = "current"
|
||||
S3ManifestFile = "manifest.json"
|
||||
S3RunIDFile = "run_id.txt"
|
||||
)
|
||||
|
||||
// DefaultArchivePromoteArtifacts defines the default archive promotion rules.
|
||||
// Callers should copy this slice before mutating.
|
||||
var DefaultArchivePromoteArtifacts = []ArchivePromotionRule{
|
||||
{From: PathTranscriptTrimmed, To: PathTranscriptTrimmed},
|
||||
{From: "artifacts/session_recap.md", To: "artifacts/session_recap.md"},
|
||||
}
|
||||
|
||||
// DefaultPipelineConfigSearchPaths defines the default search order for
|
||||
// pipeline.yml when callers do not provide an explicit path.
|
||||
//
|
||||
|
||||
@@ -174,7 +174,13 @@ func applyStorageDefaults(cfg *StorageConfig) {
|
||||
cfg.S3 = &StorageS3Config{}
|
||||
}
|
||||
if cfg.S3.RootPrefix == "" {
|
||||
cfg.S3.RootPrefix = "dnd"
|
||||
cfg.S3.RootPrefix = DefaultStorageS3RootPrefix
|
||||
}
|
||||
if cfg.S3.AccessKeyIDEnv == "" {
|
||||
cfg.S3.AccessKeyIDEnv = DefaultS3AccessKeyIDEnv
|
||||
}
|
||||
if cfg.S3.SecretKeyEnv == "" {
|
||||
cfg.S3.SecretKeyEnv = DefaultS3SecretAccessKeyEnv
|
||||
}
|
||||
}
|
||||
|
||||
@@ -183,7 +189,7 @@ func applySpoolDefaults(cfg *SpoolConfig) {
|
||||
return
|
||||
}
|
||||
if cfg.Root == "" {
|
||||
cfg.Root = "/var/spool/narratio"
|
||||
cfg.Root = DefaultSpoolRoot
|
||||
}
|
||||
}
|
||||
|
||||
@@ -196,16 +202,13 @@ func applyArchiveDefaults(cfg **ArchiveConfig) {
|
||||
}
|
||||
|
||||
if (*cfg).Enabled == nil {
|
||||
(*cfg).Enabled = boolPtr(true)
|
||||
(*cfg).Enabled = boolPtr(DefaultArchiveEnabled)
|
||||
}
|
||||
if (*cfg).UploadRun == nil {
|
||||
(*cfg).UploadRun = boolPtr(true)
|
||||
(*cfg).UploadRun = boolPtr(DefaultArchiveUploadRun)
|
||||
}
|
||||
if len((*cfg).PromoteArtifacts) == 0 {
|
||||
(*cfg).PromoteArtifacts = []ArchivePromotionRule{
|
||||
{From: "transcripts/trimmed.json", To: "transcripts/trimmed.json", Required: boolPtr(true)},
|
||||
{From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)},
|
||||
}
|
||||
(*cfg).PromoteArtifacts = append([]ArchivePromotionRule(nil), DefaultArchivePromoteArtifacts...)
|
||||
}
|
||||
for i := range (*cfg).PromoteArtifacts {
|
||||
if (*cfg).PromoteArtifacts[i].Required == nil {
|
||||
@@ -219,19 +222,19 @@ func applyWhisperXDefaults(cfg *WhisperXConfig) {
|
||||
return
|
||||
}
|
||||
if cfg.Language == "" {
|
||||
cfg.Language = "en"
|
||||
cfg.Language = DefaultWhisperXLanguage
|
||||
}
|
||||
if cfg.Timeout == "" {
|
||||
cfg.Timeout = "30m"
|
||||
cfg.Timeout = DefaultWhisperXTimeout
|
||||
}
|
||||
if cfg.RetryDelay == "" {
|
||||
cfg.RetryDelay = "2s"
|
||||
cfg.RetryDelay = DefaultWhisperXRetryDelay
|
||||
}
|
||||
if cfg.Concurrency == nil {
|
||||
cfg.Concurrency = intPtr(2)
|
||||
cfg.Concurrency = intPtr(DefaultWhisperXConcurrency)
|
||||
}
|
||||
if cfg.Retries == nil {
|
||||
cfg.Retries = intPtr(3)
|
||||
cfg.Retries = intPtr(DefaultWhisperXRetries)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -245,19 +248,19 @@ func applySeriatimDefaults(cfg *SeriatimConfig) {
|
||||
return
|
||||
}
|
||||
if cfg.Binary == "" {
|
||||
cfg.Binary = "seriatim"
|
||||
cfg.Binary = DefaultSeriatimBinary
|
||||
}
|
||||
if cfg.Timeout == "" {
|
||||
cfg.Timeout = "10m"
|
||||
cfg.Timeout = DefaultSeriatimTimeout
|
||||
}
|
||||
if cfg.OutputSchema == "" {
|
||||
cfg.OutputSchema = "seriatim-intermediate"
|
||||
cfg.OutputSchema = DefaultSeriatimOutputSchema
|
||||
}
|
||||
if cfg.CoalesceGap == nil {
|
||||
cfg.CoalesceGap = float64Ptr(3.0)
|
||||
cfg.CoalesceGap = float64Ptr(DefaultSeriatimCoalesceGap)
|
||||
}
|
||||
if cfg.Report == nil {
|
||||
cfg.Report = boolPtr(true)
|
||||
cfg.Report = boolPtr(DefaultSeriatimReport)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -266,13 +269,13 @@ func applyAuditaDefaults(cfg *AuditaConfig) {
|
||||
return
|
||||
}
|
||||
if cfg.Binary == "" {
|
||||
cfg.Binary = "audita"
|
||||
cfg.Binary = DefaultAuditaBinary
|
||||
}
|
||||
if cfg.Timeout == "" {
|
||||
cfg.Timeout = "3h"
|
||||
cfg.Timeout = DefaultAuditaTimeout
|
||||
}
|
||||
if cfg.Report == nil {
|
||||
cfg.Report = boolPtr(true)
|
||||
cfg.Report = boolPtr(DefaultAuditaReport)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -281,10 +284,10 @@ func applyScriptoriumDefaults(cfg *ScriptoriumConfig) {
|
||||
return
|
||||
}
|
||||
if cfg.Binary == "" {
|
||||
cfg.Binary = "scriptorium"
|
||||
cfg.Binary = DefaultScriptoriumBinary
|
||||
}
|
||||
if cfg.Timeout == "" {
|
||||
cfg.Timeout = "10m"
|
||||
cfg.Timeout = DefaultScriptoriumTimeout
|
||||
}
|
||||
}
|
||||
|
||||
@@ -293,10 +296,10 @@ func applyTrimDefaults(cfg *TrimConfig) {
|
||||
return
|
||||
}
|
||||
if cfg.Bounds.Timeout == "" {
|
||||
cfg.Bounds.Timeout = "10m"
|
||||
cfg.Bounds.Timeout = DefaultTrimBoundsTimeout
|
||||
}
|
||||
if cfg.Seriatim.Report == nil {
|
||||
cfg.Seriatim.Report = boolPtr(false)
|
||||
cfg.Seriatim.Report = boolPtr(DefaultTrimSeriatimReport)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -305,13 +308,13 @@ func applyNormalizeDefaults(cfg *NormalizeConfig) {
|
||||
return
|
||||
}
|
||||
if cfg.OutputSchema == "" {
|
||||
cfg.OutputSchema = defaultNormalizeOutputSchema
|
||||
cfg.OutputSchema = DefaultNormalizeOutputSchema
|
||||
}
|
||||
if cfg.OutputPath == "" && !cfg.outputPathWasSet() {
|
||||
cfg.OutputPath = defaultNormalizeOutputPath
|
||||
cfg.OutputPath = DefaultNormalizeOutputPath
|
||||
}
|
||||
if cfg.Report == nil {
|
||||
cfg.Report = boolPtr(true)
|
||||
cfg.Report = boolPtr(DefaultNormalizeReport)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -6,11 +6,6 @@ import (
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultNormalizeOutputPath = "transcripts/normalized.json"
|
||||
defaultNormalizeOutputSchema = "seriatim-intermediate"
|
||||
)
|
||||
|
||||
// UnmarshalYAML tracks explicit normalize.output_path presence so validation can
|
||||
// distinguish omitted vs explicitly empty values.
|
||||
func (cfg *NormalizeConfig) UnmarshalYAML(node *yaml.Node) error {
|
||||
|
||||
@@ -24,6 +24,12 @@ storage:
|
||||
if cfg.Pipeline.Storage.S3.RootPrefix != "dnd" {
|
||||
t.Fatalf("storage.s3.root_prefix = %q, want dnd", cfg.Pipeline.Storage.S3.RootPrefix)
|
||||
}
|
||||
if cfg.Pipeline.Storage.S3.AccessKeyIDEnv != DefaultS3AccessKeyIDEnv {
|
||||
t.Fatalf("storage.s3.access_key_id_env = %q, want %q", cfg.Pipeline.Storage.S3.AccessKeyIDEnv, DefaultS3AccessKeyIDEnv)
|
||||
}
|
||||
if cfg.Pipeline.Storage.S3.SecretKeyEnv != DefaultS3SecretAccessKeyEnv {
|
||||
t.Fatalf("storage.s3.secret_access_key_env = %q, want %q", cfg.Pipeline.Storage.S3.SecretKeyEnv, DefaultS3SecretAccessKeyEnv)
|
||||
}
|
||||
if cfg.Pipeline.Storage.S3.ForcePathStyle {
|
||||
t.Fatalf("storage.s3.force_path_style = true, want false default")
|
||||
}
|
||||
@@ -33,6 +39,77 @@ storage:
|
||||
}
|
||||
}
|
||||
|
||||
func TestStorageS3CredentialEnvNamesLoadAndValidate(t *testing.T) {
|
||||
pipelineYAML := testPipelineBaseYAML + `
|
||||
storage:
|
||||
backend: s3
|
||||
s3:
|
||||
bucket: my-dnd-archive
|
||||
access_key_id_env: CUSTOM_KEY_ID
|
||||
secret_access_key_env: CUSTOM_SECRET
|
||||
`
|
||||
pipelinePath, sessionPath := writeConfigFiles(t, pipelineYAML, testSessionBaseYAML)
|
||||
|
||||
cfg, err := Load(pipelinePath, sessionPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error = %v", err)
|
||||
}
|
||||
if cfg.Pipeline.Storage.S3.AccessKeyIDEnv != "CUSTOM_KEY_ID" {
|
||||
t.Fatalf("storage.s3.access_key_id_env = %q, want CUSTOM_KEY_ID", cfg.Pipeline.Storage.S3.AccessKeyIDEnv)
|
||||
}
|
||||
if cfg.Pipeline.Storage.S3.SecretKeyEnv != "CUSTOM_SECRET" {
|
||||
t.Fatalf("storage.s3.secret_access_key_env = %q, want CUSTOM_SECRET", cfg.Pipeline.Storage.S3.SecretKeyEnv)
|
||||
}
|
||||
if err := Validate(cfg); err != nil {
|
||||
t.Fatalf("Validate() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStorageS3CredentialEnvValidation(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
pipelineYML string
|
||||
wantErr string
|
||||
}{
|
||||
{
|
||||
name: "invalid access key env name",
|
||||
pipelineYML: testPipelineBaseYAML + `
|
||||
storage:
|
||||
backend: s3
|
||||
s3:
|
||||
bucket: my-dnd-archive
|
||||
access_key_id_env: "123BAD"
|
||||
`,
|
||||
wantErr: "pipeline.storage.s3.access_key_id_env must be a valid environment variable name",
|
||||
},
|
||||
{
|
||||
name: "invalid secret key env name",
|
||||
pipelineYML: testPipelineBaseYAML + `
|
||||
storage:
|
||||
backend: s3
|
||||
s3:
|
||||
bucket: my-dnd-archive
|
||||
secret_access_key_env: "bad-name"
|
||||
`,
|
||||
wantErr: "pipeline.storage.s3.secret_access_key_env must be a valid environment variable name",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
pipelinePath, sessionPath := writeConfigFiles(t, tt.pipelineYML, testSessionBaseYAML)
|
||||
cfg, err := Load(pipelinePath, sessionPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Load() error = %v", err)
|
||||
}
|
||||
err = Validate(cfg)
|
||||
if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
|
||||
t.Fatalf("Validate() error = %v, want to contain %q", err, tt.wantErr)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpoolAndArchiveDefaults(t *testing.T) {
|
||||
pipelinePath, sessionPath := writeConfigFiles(t, testPipelineBaseYAML, testSessionBaseYAML)
|
||||
|
||||
|
||||
@@ -91,6 +91,12 @@ func validateStorage(cfg StorageConfig) error {
|
||||
if cfg.S3.Endpoint != "" && strings.TrimSpace(cfg.S3.Endpoint) == "" {
|
||||
return fmt.Errorf("pipeline.storage.s3.endpoint must be non-empty when provided")
|
||||
}
|
||||
if err := validateEnvVarNameField("pipeline.storage.s3.access_key_id_env", cfg.S3.AccessKeyIDEnv); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := validateEnvVarNameField("pipeline.storage.s3.secret_access_key_env", cfg.S3.SecretKeyEnv); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -430,6 +436,18 @@ func archiveUploadConfiguredForS3(pipeline *PipelineConfig) bool {
|
||||
}
|
||||
|
||||
var windowsAbsPathRE = regexp.MustCompile(`^[A-Za-z]:[\\/].*`)
|
||||
var envVarNameRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
|
||||
|
||||
func validateEnvVarNameField(fieldName, value string) error {
|
||||
trimmed := strings.TrimSpace(value)
|
||||
if trimmed == "" {
|
||||
return fmt.Errorf("%s must be non-empty", fieldName)
|
||||
}
|
||||
if !envVarNameRE.MatchString(trimmed) {
|
||||
return fmt.Errorf("%s must be a valid environment variable name", fieldName)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateRelativeSafePath(fieldName, value string) error {
|
||||
trimmed := strings.TrimSpace(value)
|
||||
|
||||
@@ -18,6 +18,10 @@ import (
|
||||
)
|
||||
|
||||
type archiveStage struct{}
|
||||
type archiveUploadFile struct {
|
||||
RelativePath string
|
||||
LocalPath string
|
||||
}
|
||||
|
||||
var archivePrerequisiteStages = []string{
|
||||
"prepare",
|
||||
@@ -112,29 +116,27 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
||||
return nil, fmt.Errorf("archive: run id is required")
|
||||
}
|
||||
|
||||
runFiles, err := collectArchiveRunFiles(workDir)
|
||||
manifestSource, err := resolveArchiveManifestSource(env, m, workDir)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("archive: resolve manifest source: %w", err)
|
||||
}
|
||||
|
||||
runFiles, err := collectArchiveRunFiles(workDir, manifestSource)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
||||
}
|
||||
promotions, err := resolveArchivePromotions(workDir, env.Config.Pipeline.Archive.PromoteArtifacts)
|
||||
workDirCandidates := archiveSourceWorkDirs(env, m, workDir)
|
||||
promotions, err := resolveArchivePromotions(workDirCandidates, env.Config.Pipeline.Archive.PromoteArtifacts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
||||
}
|
||||
currentManifestSource := filepath.Join(workDir, "manifest.json")
|
||||
if info, err := os.Stat(currentManifestSource); err != nil {
|
||||
return nil, fmt.Errorf("archive: current manifest source %q: %w", currentManifestSource, err)
|
||||
} else if info.IsDir() {
|
||||
return nil, fmt.Errorf("archive: current manifest source %q is a directory", currentManifestSource)
|
||||
}
|
||||
|
||||
runUploaded := make([]string, 0, len(runFiles))
|
||||
for _, rel := range runFiles {
|
||||
localPath := filepath.Join(workDir, filepath.FromSlash(rel))
|
||||
key := artifacts.S3RunRelativeDestinationKey(runPrefix, rel)
|
||||
if _, err := env.ObjectStore.Upload(ctx, localPath, key, storage.UploadOptions{}); err != nil {
|
||||
return nil, fmt.Errorf("archive: upload run file %q to %q: %w", rel, key, err)
|
||||
for _, file := range runFiles {
|
||||
key := artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath)
|
||||
if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil {
|
||||
return nil, fmt.Errorf("archive: upload run file %q to %q: %w", file.RelativePath, key, err)
|
||||
}
|
||||
runUploaded = append(runUploaded, rel)
|
||||
runUploaded = append(runUploaded, file.RelativePath)
|
||||
}
|
||||
|
||||
promotedUploaded := make([]string, 0, len(promotions))
|
||||
@@ -329,21 +331,37 @@ func archiveBucket(env *Env, m *manifest.Manifest) string {
|
||||
return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
|
||||
}
|
||||
|
||||
func resolveArchivePromotions(workDir string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
||||
func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
||||
if len(workDirs) == 0 {
|
||||
return nil, fmt.Errorf("at least one workdir candidate is required")
|
||||
}
|
||||
out := make([]archivePromotion, 0, len(rules))
|
||||
for _, rule := range rules {
|
||||
from := strings.TrimSpace(rule.From)
|
||||
to := strings.TrimSpace(rule.To)
|
||||
required := rule.Required == nil || *rule.Required
|
||||
|
||||
localPath, err := resolveWorkDirRelativePath(workDir, from)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
||||
}
|
||||
info, err := os.Stat(localPath)
|
||||
exists := err == nil && !info.IsDir()
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
||||
var (
|
||||
localPath string
|
||||
exists bool
|
||||
)
|
||||
for _, candidateRoot := range workDirs {
|
||||
resolvedPath, err := resolveWorkDirRelativePath(candidateRoot, from)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
||||
}
|
||||
info, err := os.Stat(resolvedPath)
|
||||
if err == nil && !info.IsDir() {
|
||||
localPath = resolvedPath
|
||||
exists = true
|
||||
break
|
||||
}
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
||||
}
|
||||
if localPath == "" {
|
||||
localPath = resolvedPath
|
||||
}
|
||||
}
|
||||
|
||||
out = append(out, archivePromotion{
|
||||
@@ -357,6 +375,33 @@ func resolveArchivePromotions(workDir string, rules []config.ArchivePromotionRul
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func archiveSourceWorkDirs(env *Env, m *manifest.Manifest, runWorkDir string) []string {
|
||||
candidates := make([]string, 0, 2)
|
||||
if strings.TrimSpace(runWorkDir) != "" {
|
||||
candidates = append(candidates, filepath.Clean(runWorkDir))
|
||||
}
|
||||
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
||||
if sessionID == "" && m != nil {
|
||||
sessionID = strings.TrimSpace(m.SessionID)
|
||||
}
|
||||
if sessionID != "" {
|
||||
sessionWorkDir := filepath.Clean(artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID))
|
||||
if sessionWorkDir != "" {
|
||||
candidates = append(candidates, sessionWorkDir)
|
||||
}
|
||||
}
|
||||
seen := make(map[string]struct{}, len(candidates))
|
||||
out := make([]string, 0, len(candidates))
|
||||
for _, c := range candidates {
|
||||
if _, ok := seen[c]; ok {
|
||||
continue
|
||||
}
|
||||
seen[c] = struct{}{}
|
||||
out = append(out, c)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
||||
rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
|
||||
if rel == "." || rel == "" {
|
||||
@@ -375,8 +420,8 @@ func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
||||
return cleanedFull, nil
|
||||
}
|
||||
|
||||
func collectArchiveRunFiles(workDir string) ([]string, error) {
|
||||
files := make([]string, 0, 64)
|
||||
func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile, error) {
|
||||
files := make([]archiveUploadFile, 0, 64)
|
||||
|
||||
for _, dirName := range archiveRunUploadDirs {
|
||||
fullDir := filepath.Join(workDir, dirName)
|
||||
@@ -403,30 +448,72 @@ func collectArchiveRunFiles(workDir string) ([]string, error) {
|
||||
return fmt.Errorf("relative path from %q to %q: %w", workDir, path, err)
|
||||
}
|
||||
rel = filepath.ToSlash(rel)
|
||||
files = append(files, rel)
|
||||
files = append(files, archiveUploadFile{
|
||||
RelativePath: rel,
|
||||
LocalPath: path,
|
||||
})
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, fmt.Errorf("walk %q: %w", fullDir, err)
|
||||
}
|
||||
}
|
||||
|
||||
manifestPath := filepath.Join(workDir, "manifest.json")
|
||||
manifestInfo, err := os.Stat(manifestPath)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil, fmt.Errorf("manifest.json not found in workdir %q", workDir)
|
||||
return nil, fmt.Errorf("manifest.json not found (checked path %q)", manifestPath)
|
||||
}
|
||||
return nil, fmt.Errorf("stat %q: %w", manifestPath, err)
|
||||
}
|
||||
if manifestInfo.IsDir() {
|
||||
return nil, fmt.Errorf("manifest path %q is a directory", manifestPath)
|
||||
}
|
||||
files = append(files, "manifest.json")
|
||||
files = append(files, archiveUploadFile{
|
||||
RelativePath: "manifest.json",
|
||||
LocalPath: manifestPath,
|
||||
})
|
||||
|
||||
sort.Strings(files)
|
||||
sort.Slice(files, func(i, j int) bool {
|
||||
return files[i].RelativePath < files[j].RelativePath
|
||||
})
|
||||
return files, nil
|
||||
}
|
||||
|
||||
func resolveArchiveManifestSource(env *Env, m *manifest.Manifest, workDir string) (string, error) {
|
||||
candidates := make([]string, 0, 3)
|
||||
candidates = append(candidates, filepath.Join(workDir, "manifest.json"))
|
||||
|
||||
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
||||
if sessionID == "" && m != nil {
|
||||
sessionID = strings.TrimSpace(m.SessionID)
|
||||
}
|
||||
if sessionID != "" {
|
||||
sessionManifest := filepath.Join(artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID), "manifest.json")
|
||||
candidates = append(candidates, sessionManifest)
|
||||
}
|
||||
|
||||
seen := make(map[string]struct{}, len(candidates))
|
||||
for _, candidate := range candidates {
|
||||
clean := filepath.Clean(candidate)
|
||||
if _, ok := seen[clean]; ok {
|
||||
continue
|
||||
}
|
||||
seen[clean] = struct{}{}
|
||||
info, err := os.Stat(clean)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
continue
|
||||
}
|
||||
return "", fmt.Errorf("stat %q: %w", clean, err)
|
||||
}
|
||||
if info.IsDir() {
|
||||
continue
|
||||
}
|
||||
return clean, nil
|
||||
}
|
||||
return "", fmt.Errorf("manifest.json not found in run workdir or session workdir")
|
||||
}
|
||||
|
||||
func writeCurrentManifestSnapshot(m *manifest.Manifest, archiveMetadata map[string]any) (string, error) {
|
||||
if m == nil {
|
||||
return "", fmt.Errorf("manifest is required")
|
||||
|
||||
@@ -3,6 +3,7 @@ package stage
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
@@ -182,6 +183,29 @@ func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchivePromotionFallsBackToSessionWorkDir(t *testing.T) {
|
||||
env, m, runWorkDir := archiveFixture(t)
|
||||
sessionWorkDir := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, m.SessionID)
|
||||
sessionTrimmed := filepath.Join(sessionWorkDir, "transcripts", "trimmed.json")
|
||||
|
||||
if err := os.Remove(filepath.Join(runWorkDir, "transcripts", "trimmed.json")); err != nil {
|
||||
t.Fatalf("remove run scoped trimmed transcript: %v", err)
|
||||
}
|
||||
writeStageTestFile(t, sessionTrimmed, "{}\n")
|
||||
|
||||
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v", err)
|
||||
}
|
||||
if result.Metadata["promoted_files_uploaded"] != 2 {
|
||||
t.Fatalf("metadata promoted_files_uploaded = %#v, want 2", result.Metadata["promoted_files_uploaded"])
|
||||
}
|
||||
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||
if _, ok := fake.Objects[m.S3SessionPrefix+"transcripts/trimmed.json"]; !ok {
|
||||
t.Fatalf("missing promoted trimmed key from session fallback")
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchiveDoesNotWriteCurrentPointerWhenPromotionUploadFails(t *testing.T) {
|
||||
env, m, _ := archiveFixture(t)
|
||||
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
@@ -86,10 +87,15 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
|
||||
stderrPath := filepath.Join(paths.LogsDir, "seriatim.stderr.log")
|
||||
genCfgPath := filepath.Join(paths.ConfigDir, "seriatim.generated.yml")
|
||||
|
||||
normalizedInputs, normalizeLogs, normalizeConfigs, normalizeMeta, err := normalizeMergeInputs(ctx, env, inputs, paths)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
reportEnabled := env.Config.Pipeline.Seriatim.Report != nil && *env.Config.Pipeline.Seriatim.Report
|
||||
req := seriatim.MergeRequest{
|
||||
GeneratedConfigPath: genCfgPath,
|
||||
InputTranscriptPaths: inputs,
|
||||
InputTranscriptPaths: normalizedInputs,
|
||||
OutputMergedTranscriptPath: mergedPath,
|
||||
ReportPath: "",
|
||||
SpeakersPath: speakersPath,
|
||||
@@ -146,8 +152,11 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
|
||||
|
||||
meta := map[string]any{
|
||||
"stage": "merge",
|
||||
"input_transcripts_count": len(inputs),
|
||||
"input_transcripts_count": len(normalizedInputs),
|
||||
"input_transcript_paths": inputs,
|
||||
"normalized_inputs_count": len(normalizedInputs),
|
||||
"normalized_input_paths": normalizedInputs,
|
||||
"normalize_inputs": normalizeMeta,
|
||||
"output_schema": env.Config.Pipeline.Seriatim.OutputSchema,
|
||||
"coalesce_gap": coalesceGap,
|
||||
"report_enabled": reportEnabled,
|
||||
@@ -172,12 +181,97 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
|
||||
|
||||
return &StageResult{
|
||||
Outputs: outputs,
|
||||
Logs: []string{stdoutPath, stderrPath},
|
||||
GeneratedConfigs: []string{genCfgPath},
|
||||
Logs: append(normalizeLogs, stdoutPath, stderrPath),
|
||||
GeneratedConfigs: append(normalizeConfigs, genCfgPath),
|
||||
Metadata: meta,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type normalizeMergeInputMeta struct {
|
||||
InputPath string `json:"input_path"`
|
||||
OutputPath string `json:"output_path"`
|
||||
StdoutLogPath string `json:"stdout_log_path"`
|
||||
StderrLogPath string `json:"stderr_log_path"`
|
||||
GeneratedConfig string `json:"generated_config_path"`
|
||||
DurationMs int64 `json:"duration_ms"`
|
||||
ExitCode int `json:"exit_code"`
|
||||
InvokedBinary string `json:"invoked_binary"`
|
||||
OutputSchema string `json:"output_schema"`
|
||||
AdapterReportPath string `json:"adapter_report_path,omitempty"`
|
||||
AdapterOutputPath string `json:"adapter_output_path,omitempty"`
|
||||
}
|
||||
|
||||
func normalizeMergeInputs(ctx context.Context, env *Env, rawInputs []string, paths artifacts.SessionPaths) ([]string, []string, []string, []normalizeMergeInputMeta, error) {
|
||||
normalizedInputs := make([]string, 0, len(rawInputs))
|
||||
logs := make([]string, 0, len(rawInputs)*2)
|
||||
configs := make([]string, 0, len(rawInputs))
|
||||
meta := make([]normalizeMergeInputMeta, 0, len(rawInputs))
|
||||
normalizedDir := filepath.Join(paths.TranscriptsRawDir, "normalized")
|
||||
if err := os.MkdirAll(normalizedDir, 0o755); err != nil {
|
||||
return nil, nil, nil, nil, fmt.Errorf("merge: ensure normalized transcripts directory %q: %w", normalizedDir, err)
|
||||
}
|
||||
|
||||
var timeout time.Duration
|
||||
timeoutRaw := strings.TrimSpace(env.Config.Pipeline.Seriatim.Timeout)
|
||||
if timeoutRaw != "" {
|
||||
parsed, err := time.ParseDuration(timeoutRaw)
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, fmt.Errorf("merge: parse seriatim timeout %q: %w", env.Config.Pipeline.Seriatim.Timeout, err)
|
||||
}
|
||||
timeout = parsed
|
||||
}
|
||||
for _, input := range rawInputs {
|
||||
base := strings.TrimSuffix(filepath.Base(input), filepath.Ext(input))
|
||||
outPath := filepath.Join(normalizedDir, base+".normalized.json")
|
||||
stdoutPath := filepath.Join(paths.LogsDir, "seriatim.normalize."+base+".stdout.log")
|
||||
stderrPath := filepath.Join(paths.LogsDir, "seriatim.normalize."+base+".stderr.log")
|
||||
cfgPath := filepath.Join(paths.ConfigDir, "seriatim.normalize."+base+".generated.yml")
|
||||
|
||||
req := seriatim.NormalizeRequest{
|
||||
Binary: env.Config.Pipeline.Seriatim.Binary,
|
||||
InputTranscriptPath: input,
|
||||
OutputNormalizedPath: outPath,
|
||||
OutputSchema: env.Config.Pipeline.Seriatim.OutputSchema,
|
||||
ReportPath: "",
|
||||
StdoutLogPath: stdoutPath,
|
||||
StderrLogPath: stderrPath,
|
||||
GeneratedConfigPath: cfgPath,
|
||||
Timeout: timeout,
|
||||
}
|
||||
res, err := env.Seriatim.Normalize(ctx, req)
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, fmt.Errorf("merge: normalize input %q failed: %w", input, err)
|
||||
}
|
||||
|
||||
finalOutputPath := outPath
|
||||
if strings.TrimSpace(res.OutputNormalizedPath) != "" {
|
||||
finalOutputPath = strings.TrimSpace(res.OutputNormalizedPath)
|
||||
}
|
||||
if err := validateTranscriptJSONFile(finalOutputPath); err != nil {
|
||||
return nil, nil, nil, nil, fmt.Errorf("merge: normalized transcript %q invalid (input %q): %w", finalOutputPath, input, err)
|
||||
}
|
||||
|
||||
normalizedInputs = append(normalizedInputs, finalOutputPath)
|
||||
logs = append(logs, stdoutPath, stderrPath)
|
||||
configs = append(configs, cfgPath)
|
||||
meta = append(meta, normalizeMergeInputMeta{
|
||||
InputPath: input,
|
||||
OutputPath: finalOutputPath,
|
||||
StdoutLogPath: stdoutPath,
|
||||
StderrLogPath: stderrPath,
|
||||
GeneratedConfig: cfgPath,
|
||||
DurationMs: res.Duration.Milliseconds(),
|
||||
ExitCode: res.ExitCode,
|
||||
InvokedBinary: res.InvokedBinary,
|
||||
OutputSchema: res.OutputSchema,
|
||||
AdapterReportPath: res.ReportPath,
|
||||
AdapterOutputPath: res.OutputNormalizedPath,
|
||||
})
|
||||
}
|
||||
|
||||
return normalizedInputs, logs, configs, meta, nil
|
||||
}
|
||||
|
||||
func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) {
|
||||
fromManifest := make([]string, 0)
|
||||
if m != nil && m.Stages != nil {
|
||||
|
||||
@@ -2,6 +2,7 @@ package stage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -14,6 +15,22 @@ import (
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||
)
|
||||
|
||||
type normalizeDirAssertingRunner struct {
|
||||
*seriatim.FakeRunner
|
||||
ExpectedDir string
|
||||
}
|
||||
|
||||
func (r *normalizeDirAssertingRunner) Normalize(ctx context.Context, req seriatim.NormalizeRequest) (seriatim.NormalizeResult, error) {
|
||||
info, err := os.Stat(r.ExpectedDir)
|
||||
if err != nil {
|
||||
return seriatim.NormalizeResult{}, fmt.Errorf("normalized directory check failed for %q: %w", r.ExpectedDir, err)
|
||||
}
|
||||
if !info.IsDir() {
|
||||
return seriatim.NormalizeResult{}, fmt.Errorf("normalized path %q exists but is not a directory", r.ExpectedDir)
|
||||
}
|
||||
return r.FakeRunner.Normalize(ctx, req)
|
||||
}
|
||||
|
||||
func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
|
||||
env, m := setupMergeEnv(t)
|
||||
paths := env.ArtifactStore.SessionPaths(m.SessionID)
|
||||
@@ -54,6 +71,17 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
|
||||
if len(req.InputTranscriptPaths) != 2 {
|
||||
t.Fatalf("input transcripts = %#v, want 2", req.InputTranscriptPaths)
|
||||
}
|
||||
if len(fake.NormalizeRequests) != 2 {
|
||||
t.Fatalf("normalize requests = %#v, want 2", fake.NormalizeRequests)
|
||||
}
|
||||
if fake.NormalizeRequests[0].InputTranscriptPath != inA || fake.NormalizeRequests[1].InputTranscriptPath != inB {
|
||||
t.Fatalf("normalize request inputs = %#v", fake.NormalizeRequests)
|
||||
}
|
||||
for _, mergeIn := range req.InputTranscriptPaths {
|
||||
if !strings.Contains(mergeIn, filepath.Join("transcripts", "raw", "normalized")) {
|
||||
t.Fatalf("merge input path = %q, want normalized input path", mergeIn)
|
||||
}
|
||||
}
|
||||
|
||||
if len(result.Outputs) != 2 {
|
||||
t.Fatalf("outputs len = %d, want 2", len(result.Outputs))
|
||||
@@ -64,11 +92,11 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
|
||||
if result.Outputs[1].Kind != "seriatim_report" {
|
||||
t.Fatalf("output[1] kind = %q, want seriatim_report", result.Outputs[1].Kind)
|
||||
}
|
||||
if len(result.Logs) != 2 {
|
||||
t.Fatalf("logs = %#v, want 2 paths", result.Logs)
|
||||
if len(result.Logs) != 6 {
|
||||
t.Fatalf("logs = %#v, want 6 paths (4 normalize + 2 merge)", result.Logs)
|
||||
}
|
||||
if len(result.GeneratedConfigs) != 1 {
|
||||
t.Fatalf("generated configs = %#v, want 1 path", result.GeneratedConfigs)
|
||||
if len(result.GeneratedConfigs) != 3 {
|
||||
t.Fatalf("generated configs = %#v, want 3 paths (2 normalize + 1 merge)", result.GeneratedConfigs)
|
||||
}
|
||||
|
||||
meta := result.Metadata
|
||||
@@ -84,6 +112,12 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
|
||||
if meta["input_transcripts_count"] != 2 {
|
||||
t.Fatalf("metadata input_transcripts_count = %#v, want 2", meta["input_transcripts_count"])
|
||||
}
|
||||
if meta["normalized_inputs_count"] != 2 {
|
||||
t.Fatalf("metadata normalized_inputs_count = %#v, want 2", meta["normalized_inputs_count"])
|
||||
}
|
||||
if _, ok := meta["normalize_inputs"]; !ok {
|
||||
t.Fatalf("metadata normalize_inputs missing: %#v", meta)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeStageFailsWhenNoRawTranscripts(t *testing.T) {
|
||||
@@ -152,6 +186,9 @@ func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testin
|
||||
if len(fake.Requests) != 1 || len(fake.Requests[0].InputTranscriptPaths) != 1 {
|
||||
t.Fatalf("fallback inputs = %#v", fake.Requests)
|
||||
}
|
||||
if len(fake.NormalizeRequests) != 1 {
|
||||
t.Fatalf("normalize requests = %#v, want 1", fake.NormalizeRequests)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t *testing.T) {
|
||||
@@ -180,8 +217,75 @@ func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("input transcript paths = %#v, want len 1", got)
|
||||
}
|
||||
if got[0] != filepath.Clean(rawPath) {
|
||||
t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath))
|
||||
if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) {
|
||||
t.Fatalf("resolved transcript path = %q, want normalized path under transcripts/raw/normalized", got[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeStageCreatesNormalizedRawDirectoryBeforeNormalize(t *testing.T) {
|
||||
env, m := setupMergeEnv(t)
|
||||
paths := env.ArtifactStore.SessionPaths(m.SessionID)
|
||||
|
||||
rawPath := filepath.Join(paths.TranscriptsRawDir, "alice.json")
|
||||
writeFile(t, rawPath, `{"segments":[]}`)
|
||||
writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
|
||||
writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
|
||||
|
||||
normalizedDir := filepath.Join(paths.TranscriptsRawDir, "normalized")
|
||||
if err := os.RemoveAll(normalizedDir); err != nil {
|
||||
t.Fatalf("remove normalized dir: %v", err)
|
||||
}
|
||||
|
||||
env.Seriatim = &normalizeDirAssertingRunner{
|
||||
FakeRunner: &seriatim.FakeRunner{},
|
||||
ExpectedDir: normalizedDir,
|
||||
}
|
||||
|
||||
if _, err := (mergeStage{}).Run(context.Background(), env, m); err != nil {
|
||||
t.Fatalf("merge.Run() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeStageFailsWhenNormalizeAdapterFails(t *testing.T) {
|
||||
env, m := setupMergeEnv(t)
|
||||
paths := env.ArtifactStore.SessionPaths(m.SessionID)
|
||||
in := filepath.Join(paths.TranscriptsRawDir, "alice.json")
|
||||
writeFile(t, in, `{"segments":[]}`)
|
||||
writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
|
||||
writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
|
||||
env.Seriatim = &seriatim.FakeRunner{NormalizeErr: context.DeadlineExceeded}
|
||||
|
||||
_, err := (mergeStage{}).Run(context.Background(), env, m)
|
||||
if err == nil {
|
||||
t.Fatal("expected error, got nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "normalize input") {
|
||||
t.Fatalf("error = %q", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeStageFailsWhenNormalizedOutputInvalid(t *testing.T) {
|
||||
env, m := setupMergeEnv(t)
|
||||
paths := env.ArtifactStore.SessionPaths(m.SessionID)
|
||||
in := filepath.Join(paths.TranscriptsRawDir, "alice.json")
|
||||
writeFile(t, in, `{"segments":[]}`)
|
||||
writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
|
||||
writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
|
||||
|
||||
badNormalized := filepath.Join(paths.ArtifactsDir, "bad.normalized.json")
|
||||
writeFile(t, badNormalized, "not-json")
|
||||
env.Seriatim = &seriatim.FakeRunner{
|
||||
NormalizeResult: seriatim.NormalizeResult{
|
||||
OutputNormalizedPath: badNormalized,
|
||||
},
|
||||
}
|
||||
|
||||
_, err := (mergeStage{}).Run(context.Background(), env, m)
|
||||
if err == nil {
|
||||
t.Fatal("expected error, got nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "normalized transcript") {
|
||||
t.Fatalf("error = %q", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -211,8 +315,8 @@ func TestMergeStageResolvesSessionRelativeManifestOutputs(t *testing.T) {
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("input transcript paths = %#v, want len 1", got)
|
||||
}
|
||||
if got[0] != filepath.Clean(rawPath) {
|
||||
t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath))
|
||||
if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) {
|
||||
t.Fatalf("resolved transcript path = %q, want normalized path under transcripts/raw/normalized", got[0])
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,31 +0,0 @@
|
||||
# Narratio UX Evaluation Report
|
||||
|
||||
## 1. Executive Summary
|
||||
Narratio has a functional core pipeline with robust S3 integration for input and output, but it currently falls short of the intended "minimalist" operator UX. The primary gaps are the lack of session configuration discovery, the absence of session template support (and the `--session-id` flag), and the missing local cleanup logic. While the pipeline runs successfully, the operator must currently provide explicit session file paths for every run.
|
||||
|
||||
## 2. Feature Matrix
|
||||
|
||||
| Feature | Status | Evidence | Tests | Documentation | Notes |
|
||||
| :--- | :--- | :--- | :--- | :--- | :--- |
|
||||
| **Pipeline Config Discovery** | Implemented | `internal/app/pipeline_config_path_test.go` | Yes | Accurate | Checks `/usr/local/etc` and `/etc`. |
|
||||
| **Session Config Discovery** | Missing | `internal/app/run.go:30` | N/A | Stale | `--session` is mandatory. |
|
||||
| **Session Templates** | Missing | `internal/config/load.go` | N/A | Missing | No variable interpolation in `session.yml`. |
|
||||
| **`--session-id` CLI Flag** | Missing | `cmd/narratio` | N/A | Missing | Not implemented in CLI. |
|
||||
| **Minimal Seriatim Config** | Implemented | `internal/config/load.go` | Yes | Accurate | Defaults for timeout/schema provided. |
|
||||
| **Minimal Audita Config** | Implemented | `internal/config/load.go` | Yes | Accurate | Defaults for base_url/model provided. |
|
||||
| **S3 Audio Input** | Implemented | `internal/stage/prepare.go` | Yes | Accurate | Supports `.flac` downloads from S3. |
|
||||
| **S3 Archive & Promotion** | Implemented | `internal/stage/archive.go` | Yes | Accurate | Correct paths and commit markers. |
|
||||
| **Local Cleanup** | Missing | `architecture.md:136` | No | Stale | Config exists, logic is not implemented. |
|
||||
|
||||
## 3. Current Happy Path
|
||||
The shortest command that works today is:
|
||||
`narratio run --session <path_to_session.yml>`
|
||||
*(Assuming `pipeline.yml` is present in `/etc/narratio/` or `/usr/local/etc/narratio/`)*.
|
||||
|
||||
## 4. Gaps to Intended UX
|
||||
1. **Session Discovery & Templates (High):** The requirement to pass `--session` and the inability to use `--session-id` with a template is the largest friction point for operators.
|
||||
2. **Local Cleanup (Medium):** Spool and work directories are not cleaned up after successful archival, leading to local disk growth.
|
||||
3. **Local Pipeline Config (Low):** Narratio does not check `./pipeline.yml`, requiring users to use `--config` or move files to system directories.
|
||||
|
||||
## 5. Recommended Next Implementation Prompt
|
||||
"Implement session configuration discovery and template support. Specifically: 1) Add a search order for `session.yml` (e.g., `./session.yml`, `/etc/narratio/session.yml`) if `--session` is omitted. 2) Implement the `--session-id` CLI flag. 3) Add variable interpolation to `session.yml` so that `{{session_id}}` can be replaced by the value from the flag or the discovered session config before YAML decoding."
|
||||
54
report.md
54
report.md
@@ -1,54 +0,0 @@
|
||||
## 1. Executive Summary
|
||||
Narratio is close on S3 input/archive mechanics but not yet close on the intended minimal operator UX.
|
||||
Core S3 workflow is implemented (prepare S3 audio download, archive run upload, promotions, current pointers), but key UX items are missing: no `--session-id` flag, no session auto-discovery, and no session template variable injection. Cleanup/retention for spool/workdirs after archive is also still future work.
|
||||
|
||||
## 2. Feature Matrix
|
||||
|
||||
| Feature | Status | Evidence | Tests | Documentation status | Notes |
|
||||
|---|---|---|---|---|---|
|
||||
| Pipeline config auto-discovery when `--config` omitted | Implemented | `internal/app/pipeline_config_path.go`, `internal/config/defaults.go` | `internal/app/pipeline_config_path_test.go`, `internal/app/commands_test.go` | Accurate in `README.md`, `architecture.md` | Order: `/usr/local/etc/narratio/pipeline.yml`, then `/etc/narratio/pipeline.yml`; no `./pipeline.yml` default |
|
||||
| Session config auto-discovery when `--session` omitted | Missing | `--session` required in `internal/app/run.go`, `plan.go`, `resume.go`, `run_stage.go` | Covered by missing-flag tests in `internal/app/commands_test.go` | Accurate (docs do not claim auto-discovery) | No precedence order exists for session file search |
|
||||
| Session template variables in `session.yml` | Missing | Strict decode path in `internal/config/load.go` + strict YAML behavior | No template tests found | Not documented as implemented | No render-before-decode templating mechanism found |
|
||||
| `--session-id` CLI injection | Missing | No `--session-id` flag in command parsers (`run/plan/resume/run-stage`) | No tests for `--session-id` | Not documented as implemented | Intended minimal UX command not currently supported |
|
||||
| Campaign/run-aware work+spool paths | Implemented | `internal/artifacts/paths.go`, usage in prepare/archive | Path/helper tests in `internal/artifacts` + stage tests | Documented in README/architecture/roadmap | Layout includes `{campaign}/{session_id}/{run_id}` |
|
||||
| Run ID generation format | Implemented | `internal/artifacts/run_id.go` | Run ID tests in `internal/artifacts` | Documented | UTC timestamp + random suffix format present |
|
||||
| Storage backend abstraction | Implemented | `internal/adapters/storage/object_store.go` | Storage backend tests in `internal/adapters/storage` | Documented in README/architecture | Narrow interface (`List/Download/Upload/Exists`) |
|
||||
| S3 backend + fake backend | Implemented | `internal/adapters/storage/s3_backend.go`, `fake.go` | Adapter tests pass without live S3 | Documented | No AWS creds in config schema/examples |
|
||||
| Prepare S3 audio input (`inputs.audio_s3`) | Implemented | `internal/stage/prepare.go` | `internal/stage/prepare_test.go` | Documented in `docs/s3-audio-input.md`, README, architecture | Lists prefix, filters `.flac`, downloads/materializes, fails on none |
|
||||
| Local audio workflow | Implemented | Prepare logic still supports `audio_dir`/`audio_files` | Prepare tests cover local behavior and conflict with `audio_s3` | Documented | Local+S3 conflict is enforced |
|
||||
| Manifest provenance for S3 audio | Implemented | S3 source metadata assignment in prepare stage | Covered by S3 prepare tests | Documented | ETag recorded as metadata, not checksum |
|
||||
| Archive run upload under `runs/{run_id}` | Implemented | `internal/stage/archive.go` | `internal/stage/archive_test.go` | Documented in `docs/archive-storage.md`, README, architecture | Successful/completed runs only |
|
||||
| Archive promotion rules | Implemented | Archive stage promotion handling | Archive tests cover required/optional/mapping behavior | Documented | Default promoted outputs: `transcripts/trimmed.json`, `artifacts/session_recap.md` |
|
||||
| `current/manifest.json` + `current/run_id.txt` last | Implemented | Archive stage upload order logic | Archive tests verify ordering and pointer content | Documented | `current/run_id.txt` is commit marker; written last |
|
||||
| Avoid upload of failed/incomplete runs | Implemented | Archive prerequisite checks | Archive tests cover prerequisite failure path | Documented | Failed runs stay local |
|
||||
| Spool/workdir cleanup after successful archive | Missing | `spool.delete_audio_after_archive` exists but no cleanup behavior in stages/app | No cleanup behavior tests found | Docs accurately call cleanup future work | Gap vs intended UX item 12 |
|
||||
| Minimal Seriatim config | Partial | Validation requires `seriatim.binary`; defaults fill timeout/schema/gap | Config load/validate tests | Docs mostly accurate | “Binary-only” works after defaults, but still validated post-defaults |
|
||||
| Minimal Audita config | Partial | Validation requires `audita.binary` and `audita.model`; defaults for timeout/base_url/etc in loader | Config tests in `internal/config` | Docs currently list `timeout`/`base_url` as required in README section | UX expectation “binary + llm_api_key_env only” does not hold because model is required |
|
||||
|
||||
## 3. Current Happy Path
|
||||
Shortest realistic command today is:
|
||||
|
||||
`narratio run --session /path/to/session.yml`
|
||||
|
||||
That works only if pipeline config is discoverable at `/usr/local/etc/narratio/pipeline.yml` or `/etc/narratio/pipeline.yml`.
|
||||
Otherwise minimum is:
|
||||
|
||||
`narratio run --config /path/to/pipeline.yml --session /path/to/session.yml`
|
||||
|
||||
`narratio run --session-id 2026-04-04` does not work today (flag not implemented).
|
||||
|
||||
## 4. Gaps to Intended UX
|
||||
1. Missing `--session-id` flow with session template injection (largest UX gap).
|
||||
2. No session config auto-discovery order when `--session` is omitted.
|
||||
3. No session template rendering engine / unresolved-variable handling.
|
||||
4. Cleanup policy not implemented (`spool.delete_audio_after_archive` is modeled only).
|
||||
5. Audita minimal config UX still stricter than intended (model required).
|
||||
6. Optional doc refinement: explicitly call out that `./pipeline.yml` is not in current default search order.
|
||||
|
||||
## 5. Recommended Next Implementation Prompt
|
||||
Implement session template and `--session-id` UX only:
|
||||
|
||||
> Add session discovery and template rendering support so `narratio run --session-id <id>` works with no `--session` in normal setups.
|
||||
> Requirements: define deterministic session discovery order; support rendering template variables in `session.yml` before strict YAML decode; inject CLI `--session-id` into template variables; fail clearly on unresolved variables; preserve strict field validation after render; keep existing `--session` explicit path behavior; add tests for discovery precedence, render success/failure, and CLI integration; update README/architecture/examples accordingly; do not change archive/prepare storage behavior.
|
||||
|
||||
Validation note: `go test ./...` passes for the inspected state.
|
||||
Reference in New Issue
Block a user