6 Commits

Author SHA1 Message Date
539601bd16 Updated the merge stage to normalize the per-speaker transcripts before merging them
All checks were successful
ci/woodpecker/tag/release Pipeline was successful
2026-05-16 23:30:13 -05:00
6ca1c8d6b0 The backend S3 client now resolves credentials from user-configurable environment variables 2026-05-16 23:22:21 -05:00
4b7b50981b Add .gocache to .gitignore and minor documentation cleanup 2026-05-16 23:21:45 -05:00
33f7ae8f2e Simplify downstream tool configuration
All checks were successful
ci/woodpecker/tag/release Pipeline was successful
2026-05-16 23:30:40 +00:00
d5a9ad38f8 Add post-archive cleanup policies 2026-05-16 23:09:39 +00:00
6fbefb9867 Add session discovery and template support 2026-05-16 22:57:42 +00:00
38 changed files with 1773 additions and 272 deletions

3
.gitignore vendored
View File

@@ -22,6 +22,9 @@ AGENTS.md
# Dependency directories (remove the comment below to include it)
# vendor/
# Go cache
.gocache
# Go workspace file
go.work
go.work.sum

View File

@@ -34,6 +34,25 @@ Pipeline config lookup for CLI commands:
- `/usr/local/etc/narratio/pipeline.yml`
- `/etc/narratio/pipeline.yml`
Session config lookup for CLI commands:
- if `--session <path>` is provided, Narratio uses that path
- if `--session` is omitted, Narratio searches in this order:
- `./session.yml`
- `/usr/local/etc/narratio/session.yml`
- `/etc/narratio/session.yml`
Session template support:
- Narratio renders `session.yml` templates before strict YAML decode.
- `--session-id <value>` provides the `session_id` template variable.
- Supported placeholder forms:
- `{{session_id}}`
- `{{ session_id }}`
- unresolved template placeholders fail with a clear error.
- strict YAML validation still runs after rendering.
- concrete `session.yml` files without templates remain fully supported.
Optional secrets-from-files config:
- `pipeline.secrets.env_dir` may point to a directory of secret files
@@ -52,7 +71,7 @@ Narratio now includes configuration and path-model foundations for archive suppo
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)
@@ -64,6 +83,9 @@ 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`
- `pipeline.archive.enabled`: `true`
@@ -85,6 +107,13 @@ Current boundaries:
- archive uploads `current/run_id.txt` last as the effective commit marker
- required missing promotions fail archive
- optional missing promotions are skipped and recorded
- cleanup remains conservative and opt-in:
- `pipeline.spool.delete_audio_after_archive: true` removes only the run-scoped spool audio directory after successful archive commit
- `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
- 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).
@@ -136,21 +165,40 @@ Archive run-upload details and boundaries are documented in [docs/archive-storag
- `transcripts/normalized.json`: Seriatim-normalized transcript from the normalize stage
- `transcripts/trimmed.json`: gameplay-only normalized polished transcript from trim stage
## Seriatim Configuration
`pipeline.seriatim` configures the Seriatim subprocess adapter used by `merge`, `normalize`, and `trim`.
Minimal behavior:
- `pipeline.seriatim` may be omitted entirely.
- when omitted, Narratio defaults to:
- `binary: seriatim`
- `timeout: 10m`
- `output_schema: seriatim-intermediate`
- `coalesce_gap: 3.0`
- `report: true`
Optional overrides in `pipeline.seriatim` continue to work, including explicit binary paths and advanced `env` tuning values.
## Audita Configuration
`pipeline.audita` configures the real Audita subprocess adapter used by `polish`.
Required:
Minimal behavior:
- `binary`
- `timeout`
- `base_url`
- `model`
- `pipeline.audita` may be omitted entirely.
- when omitted, Narratio defaults to:
- `binary: audita`
- `timeout: 3h`
- `report: true`
Optional:
- `llm_api_key_env` (when set, Narratio requires that env var and passes it to Audita as `AUDITA_LLM_API_KEY`)
- `modules` override list (when empty/omitted, Narratio does not pass `--modules`)
- `base_url` (when omitted, Narratio does not pass `--base-url`; Audita runtime defaults/config may apply)
- `model` (when omitted, Narratio does not pass `--model`; Audita runtime defaults/config may apply)
- `transcript_description`
- `config_path`
- `output_schema` (`bare-segments` or `audita-v1`)
@@ -161,7 +209,7 @@ Optional:
- `validation_llm_concurrency` (> 0 when provided)
- `report` (defaults to `true`)
Narratio passes only configured optional Audita flags. Omitted optional values are left to Audita runtime defaults.
Narratio passes only configured optional Audita flags. Omitted optional values are left to Audita runtime defaults/config.
## Normalize Configuration
@@ -256,7 +304,7 @@ Render-debug files are diagnostics and are not treated as canonical stage output
Key points:
- `scriptorium.binary` is required when section is present
- `scriptorium.binary` defaults to `scriptorium` when section is present
- `scriptorium.config_path` is optional
- `scriptorium.timeout` defaults to `10m` when omitted
- `scriptorium.render_debug` enables render diagnostics globally
@@ -344,7 +392,9 @@ Expected session output paths:
Starter files:
- `examples/pipeline.minimal.yml`
- `examples/pipeline.audita-overrides.yml`
- `examples/session.minimal.yml`
- `examples/session.template.yml`
- `examples/speakers.yml`
## Commands
@@ -363,6 +413,12 @@ go run ./cmd/narratio plan --session examples/session.minimal.yml
Use `--config <path>` to override default pipeline lookup when needed.
Run with a discoverable session template:
```bash
go run ./cmd/narratio run --session-id 2026-04-04
```
Run full pipeline:
```bash
@@ -375,6 +431,12 @@ Run analyze only:
go run ./cmd/narratio run-stage --config examples/pipeline.minimal.yml --session examples/session.minimal.yml analyze
```
Resume with a template session ID:
```bash
go run ./cmd/narratio resume --config examples/pipeline.minimal.yml --session examples/session.template.yml --session-id 2026-04-04
```
## Operational Note
Checksum-based stale detection is not implemented yet.

View File

@@ -115,6 +115,25 @@ CLI pipeline config path resolution:
- `/usr/local/etc/narratio/pipeline.yml`
- `/etc/narratio/pipeline.yml`
CLI session config path resolution:
- when `--session <path>` is provided, that path is used
- when `--session` is omitted, Narratio searches defaults in order:
- `./session.yml`
- `/usr/local/etc/narratio/session.yml`
- `/etc/narratio/session.yml`
Session template rendering:
- session templates are rendered before strict YAML decode
- `--session-id <value>` provides the `session_id` template variable
- supported placeholders:
- `{{session_id}}`
- `{{ session_id }}`
- unresolved placeholders fail clearly
- strict `KnownFields(true)` YAML validation still applies after rendering
- if rendered `session.session_id` conflicts with `--session-id`, load fails clearly
Optional pipeline secrets directory:
- `pipeline.secrets.env_dir` enables loading environment variables from local files before command execution
@@ -132,8 +151,11 @@ 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.spool.delete_audio_after_archive` defaults to `false` (cleanup behavior not implemented yet)
- `pipeline.workspace.cleanup_after_archive` defaults to `false`
- `pipeline.spool.delete_audio_after_archive` defaults to `false`
- `pipeline.archive` is optional and defaults to:
- `enabled: true`
- `upload_run: true`
@@ -156,7 +178,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:
@@ -211,6 +235,13 @@ Archive publishing behavior (implemented):
- `current/run_id.txt` is the effective commit marker
- if promotion or current-manifest upload fails, archive returns failure and does not write `current/run_id.txt`
- failed/incomplete runs remain local and are not uploaded
- post-archive local cleanup (implemented, opt-in):
- cleanup runs only after archive succeeded and wrote `current/run_id.txt`
- cleanup is executed after all selected stages in the command invocation succeed (for example, a later `notify` failure leaves local files intact)
- `pipeline.spool.delete_audio_after_archive: true` removes only `{spool.root}/{campaign}/{session_id}/{run_id}/audio/`
- `pipeline.workspace.cleanup_after_archive: true` removes only `{workspace.root}/work/{campaign}/{session_id}/{run_id}/`
- cleanup does not run when archive is skipped/disabled/fails or when run upload is disabled
- local development `audio_dir`/`audio_files` inputs are never removed by spool cleanup
`pipeline.scriptorium` is optional. Existing pipelines without Scriptorium continue to work.
@@ -220,17 +251,18 @@ Archive publishing behavior (implemented):
`pipeline.audita` drives the real Audita subprocess adapter for the `polish` stage.
Audita required fields:
Audita defaulted fields:
- `binary`
- `timeout`
- `base_url`
- `model`
- `binary` defaults to `audita`
- `timeout` defaults to `3h`
- `report` defaults to `true`
Audita optional fields:
- `llm_api_key_env` (enforced only when configured)
- `modules` override list (when omitted/empty, Narratio does not pass `--modules`)
- `base_url` (when omitted, Narratio does not pass `--base-url`)
- `model` (when omitted, Narratio does not pass `--model`)
- `transcript_description`
- `config_path`
- `output_schema` (`bare-segments` or `audita-v1`)
@@ -239,9 +271,18 @@ Audita optional fields:
- `proposal_llm_concurrency` (> 0 when provided)
- `validation_model`
- `validation_llm_concurrency` (> 0 when provided)
- `report` (default `true`)
- `report` override
Narratio passes only configured optional Audita flags; omitted optional values defer to Audita defaults.
Narratio passes only configured optional Audita flags; omitted optional values defer to Audita runtime defaults/config.
Seriatim defaults:
- `pipeline.seriatim` may be omitted
- `binary` defaults to `seriatim`
- `timeout` defaults to `10m`
- `output_schema` defaults to `seriatim-intermediate`
- `coalesce_gap` defaults to `3.0`
- `report` defaults to `true`
When `pipeline.normalize` is omitted, defaults are applied:
@@ -275,7 +316,7 @@ When `pipeline.trim.enabled: true`:
When `pipeline.scriptorium` is present:
- `binary` is required and non-empty
- `binary` defaults to `scriptorium` when omitted
- `config_path` is optional; when provided it must be non-empty
- `timeout` is optional; when provided it must parse as a Go duration
- default `timeout` is `10m`

View File

@@ -20,11 +20,13 @@ Implemented:
- archive uploads configured promoted outputs to session-level keys.
- archive uploads `current/manifest.json`.
- archive uploads `current/run_id.txt` last as the effective commit marker.
- optional post-archive local cleanup:
- `pipeline.spool.delete_audio_after_archive: true` removes only the run-scoped spool audio directory
- `pipeline.workspace.cleanup_after_archive: true` removes only the run-scoped local workdir
- tests use fake storage and do not require live S3.
Future work:
- spool audio cleanup/deletion behavior
- `notify` stage behavior
- stale detection
- optional future source-audio upload mode
@@ -89,6 +91,7 @@ Archive writes:
Writing `current/run_id.txt` last makes it the effective commit marker for published session state.
If any required run upload, promotion upload, or current-manifest upload fails, archive returns failure and does not write `current/run_id.txt`.
Cleanup runs only after this commit-marker write has succeeded.
## Audio Upload Policy
@@ -99,6 +102,7 @@ Original audio is expected at the session-level audio prefix and is not duplicat
- `archive.enabled: false` skips archive cleanly.
- `archive.upload_run: false` skips run upload cleanly.
- both skip cases also skip post-archive local cleanup.
## Metadata

View File

@@ -95,6 +95,11 @@ Implemented in repository:
- campaign/session/run local work and spool path helpers
- manifest run/path identity fields
- examples and tests for the above foundations
- session template operator UX:
- default session config discovery (`./session.yml`, `/usr/local/etc/narratio/session.yml`, `/etc/narratio/session.yml`)
- `--session-id` template injection for `session_id`
- session template rendering before strict YAML decode
- unresolved template placeholders and `session_id` mismatches fail clearly
- remote storage backend layer:
- object-store abstraction with `List`, `Download`, `Upload`, and `Exists`
- fake storage backend for deterministic, no-network testing
@@ -115,10 +120,13 @@ Implemented in repository:
- `current/run_id.txt` is uploaded last as the effective commit marker
- current pointer content is `{run_id}` plus trailing newline
- if promotion/current manifest upload fails, current pointer is not written
- post-archive local cleanup behavior:
- `pipeline.spool.delete_audio_after_archive` removes run-scoped spool audio only after successful archive commit
- `pipeline.workspace.cleanup_after_archive` removes run-scoped workdir only after successful archive commit
- cleanup is skipped for failed/incomplete/skipped/unarchived runs
Not implemented yet:
- spool audio cleanup / deletion behavior
- `notify` stage behavior
- generic stale detection based on input/config checksums
- optional future mode for uploading source audio from local workspace/spool

View File

@@ -12,7 +12,6 @@ Implemented:
Not implemented:
- spool cleanup/deletion behavior
- uploads of failed runs
## Required Configuration
@@ -21,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`:

View File

@@ -30,6 +30,8 @@ Construction:
- region
- endpoint
- force_path_style
- access_key_id_env
- secret_access_key_env
## Key Invariant
@@ -42,7 +44,9 @@ S3 session/run key builders remain separate and continue to live outside backend
## Security Boundary
- do not store AWS credentials in Narratio config
- AWS credentials are resolved through standard AWS SDK credential chains
- Narratio first checks configured env-var names (`access_key_id_env`, `secret_access_key_env`);
when both are present and non-empty, it uses static credentials from those values
- when either configured credential value is missing, Narratio falls back to the standard AWS SDK credential chain
- AWS SDK-specific types remain isolated to the storage adapter package
## Testing

View File

@@ -0,0 +1,26 @@
workspace:
root: ./tmp/narratio-workspace
whisperx:
transcribe_url: "https://transcription.example.com/transcribe"
seriatim:
binary: "seriatim"
timeout: "10m"
output_schema: "seriatim-intermediate"
coalesce_gap: 3.0
audita:
binary: "audita"
timeout: "3h"
base_url: "https://openrouter.ai/api/v1"
model: "openrouter/google/gemma-4-31b-it"
llm_api_key_env: "AUDITA_LLM_API_KEY"
modules: ["glossary", "homophones", "spoken_word", "grammar"]
output_schema: "audita-v1"
work_dir_retention: "auto"
total_llm_concurrency: 2
proposal_llm_concurrency: 1
validation_model: "openrouter/google/gemma-4-31b-it"
validation_llm_concurrency: 1
report: true

View File

@@ -1,14 +1,16 @@
workspace:
root: ./tmp/narratio-workspace
cleanup_after_archive: false
storage:
backend: local
backend: s3
s3:
bucket: "my-dnd-archive"
root_prefix: "dnd"
region: "us-east-1"
endpoint: ""
force_path_style: false
# 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"
@@ -17,106 +19,33 @@ spool:
archive:
enabled: true
upload_run: true
promote_artifacts:
- from: "transcripts/trimmed.json"
to: "transcripts/trimmed.json"
required: true
- from: "artifacts/session_recap.md"
to: "artifacts/session_recap.md"
required: true
secrets:
# Optional: load environment variables from files in this directory.
# File name = env var name; file contents = env var value.
env_dir: /var/local/narratio/secrets
whisperx:
transcribe_url: "https://transcription.example.com/transcribe"
language: "en"
timeout: "30m"
retries: 3
retry_delay: "2s"
concurrency: 2
seriatim:
binary: "seriatim"
timeout: "10m"
output_schema: "seriatim-intermediate"
coalesce_gap: 3.0
report: true
env:
overlap_word_run_gap: 1.0
overlap_word_run_reorder_window: 1.0
backchannel_max_duration: 2.0
filler_max_duration: 1.25
# Optional. When omitted entirely, Narratio defaults to seriatim binary + runtime defaults.
seriatim: {}
# Optional runtime overrides. Model/provider can be owned by Audita runtime config.
audita:
binary: "audita"
timeout: "3h"
config_path: "/usr/local/etc/audita/config.yml"
llm_api_key_env: "AUDITA_LLM_API_KEY"
# Optional: pass only when overriding Audita's default module sequence.
modules: []
base_url: "https://openrouter.ai/api/v1"
model: "openrouter/google/gemma-4-31b-it"
transcript_description: ""
config_path: ""
output_schema: "audita-v1"
work_dir_retention: "auto"
total_llm_concurrency: 1
proposal_llm_concurrency: 1
validation_model: ""
validation_llm_concurrency: 1
report: true
normalize:
# Session-workdir-relative when not absolute.
output_path: "transcripts/normalized.json"
output_schema: "seriatim-intermediate"
report: true
trim:
enabled: true
# Session-workdir-relative when not absolute.
output_path: "transcripts/trimmed.json"
bounds:
prompt_id: "dnd_session.bounds"
# Empty means use prompt default profile.
profile_id: ""
transcript_input_name: "transcript"
output_path: "artifacts/session_bounds.json"
timeout: "10m"
render_debug: false
render_output_path: "artifacts/session_bounds.render.json"
seriatim:
report: false
# Optional Scriptorium integration for analyze artifacts.
scriptorium:
binary: "scriptorium"
config_path: "/etc/scriptorium/config.yml"
timeout: "10m"
render_debug: false
config_path: "/usr/local/etc/scriptorium/config.yml"
artifacts:
session_recap:
enabled: true
prompt_id: "dnd.session_recap"
profile_id: "local-quality"
output_path: "artifacts/session_recap.md"
timeout: "10m"
# Optional per-artifact override of global scriptorium.render_debug.
# render_debug: true
inputs:
transcript:
# Available transcript sources:
# - trimmed_transcript (recommended for session_recap)
# - normalized_transcript (recommended for future full-session analysis)
# - processed_transcript (raw Audita-polished output)
source: "trimmed_transcript"
required: true
previous_recap:
source: "previous_session_artifact"
artifact: "session_recap"
# Optional: set when previous recap is available.
path: ""
required: false
vars:
session_id: true
@@ -124,11 +53,3 @@ scriptorium:
campaign_name: true
previous_session_id: true
output_kind: "session_recap"
analyzer:
timeout: 20m
artifacts:
output_dir: artifacts
notification:
timeout: 10s

View File

@@ -0,0 +1,12 @@
session_id: "{{ session_id }}"
campaign: sample-campaign
date: ""
title: ""
inputs:
audio_dir: ./audio
# Optional S3 input alternative. Do not configure with audio_dir/audio_files.
# audio_s3:
# prefix: "audio/{{ session_id }}/"
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml

View File

@@ -108,9 +108,7 @@ func NewSubprocessRunner(cfg SubprocessRunnerConfig) (*SubprocessRunner, error)
return nil, fmt.Errorf("audita module at index %d is empty", i)
}
}
if strings.TrimSpace(cfg.BaseURL) == "" {
return nil, fmt.Errorf("audita base url is required")
}
if strings.TrimSpace(cfg.BaseURL) != "" {
u, err := url.Parse(cfg.BaseURL)
if err != nil || u.Scheme == "" || u.Host == "" {
if err != nil {
@@ -118,8 +116,6 @@ func NewSubprocessRunner(cfg SubprocessRunnerConfig) (*SubprocessRunner, error)
}
return nil, fmt.Errorf("audita base url %q is invalid", cfg.BaseURL)
}
if strings.TrimSpace(cfg.Model) == "" {
return nil, fmt.Errorf("audita model is required")
}
if cfg.TotalLLMConcurrency != nil && *cfg.TotalLLMConcurrency <= 0 {
return nil, fmt.Errorf("audita total llm concurrency must be > 0 when provided")
@@ -307,10 +303,14 @@ func (r *SubprocessRunner) buildArgs(req PolishRequest, modules []string) []stri
req.MergedTranscriptPath,
"--glossary", req.GlossaryPath,
"--output", req.OutputProcessedPath,
"--base-url", r.baseURL,
"--model", r.model,
"--work-dir", req.WorkDir,
}
if r.baseURL != "" {
args = append(args, "--base-url", r.baseURL)
}
if r.model != "" {
args = append(args, "--model", r.model)
}
if len(modules) > 0 {
args = append(args, "--modules", strings.Join(modules, ","))
}

View File

@@ -99,9 +99,9 @@ func TestSubprocessRunnerSuccessArgsEnvAndValidation(t *testing.T) {
"process", req.MergedTranscriptPath,
"--glossary", req.GlossaryPath,
"--output", req.OutputProcessedPath,
"--work-dir", req.WorkDir,
"--base-url", "https://openrouter.ai/api/v1",
"--model", "openrouter/google/gemma-4-31b-it",
"--work-dir", req.WorkDir,
"--modules", "glossary,homophones,glossary",
"--report-json", req.ReportPath,
"--transcript-description", "Campaign Session 42",
@@ -248,6 +248,36 @@ func TestSubprocessRunnerOmitsModulesFlagWhenNotConfigured(t *testing.T) {
}
}
func TestSubprocessRunnerOmitsBaseURLAndModelFlagsWhenNotConfigured(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("helper wrapper script uses /bin/sh")
}
t.Setenv("GO_WANT_AUDITA_HELPER", "1")
t.Setenv("AUDITA_HELPER_MODE", "success")
recordPath := filepath.Join(t.TempDir(), "record.json")
t.Setenv("AUDITA_HELPER_RECORD_PATH", recordPath)
runner := mustAuditaRunner(t, SubprocessRunnerConfig{
Binary: writeAuditaHelperWrapper(t),
Timeout: mustParseAuditaDuration(t, "2s"),
LLMAPIKeyEnv: "",
Report: false,
})
req := auditaReqForTest(t, false)
if _, err := runner.Run(context.Background(), req); err != nil {
t.Fatalf("Run() error = %v", err)
}
rec := readAuditaHelperRecord(t, recordPath)
for i := 0; i < len(rec.Args); i++ {
if rec.Args[i] == "--base-url" {
t.Fatalf("args contained --base-url unexpectedly: %#v", rec.Args)
}
if rec.Args[i] == "--model" {
t.Fatalf("args contained --model unexpectedly: %#v", rec.Args)
}
}
}
func TestSubprocessRunnerSubprocessFailure(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("helper wrapper script uses /bin/sh")

View File

@@ -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)

View File

@@ -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 }

View File

@@ -64,12 +64,12 @@ func TestExecuteMissingRequiredFlags(t *testing.T) {
args []string
want string
}{
{name: "run missing flags", args: []string{"run"}, want: "run: --session is required"},
{name: "plan missing flags", args: []string{"plan"}, want: "plan: --session is required"},
{name: "run missing flags", args: []string{"run"}, want: "run: no pipeline config path provided and no default pipeline config found; searched:"},
{name: "plan missing flags", args: []string{"plan"}, want: "plan: no pipeline config path provided and no default pipeline config found; searched:"},
{name: "status missing flags", args: []string{"status"}, want: "status: --manifest is required"},
{name: "resume missing flags", args: []string{"resume"}, want: "resume: --session is required"},
{name: "resume missing flags", args: []string{"resume"}, want: "resume: no pipeline config path provided and no default pipeline config found; searched:"},
{name: "run-stage missing name", args: []string{"run-stage", "--config", "a", "--session", "b"}, want: "run-stage: expected exactly one stage name"},
{name: "run-stage missing config flags", args: []string{"run-stage", "polish"}, want: "run-stage: --session is required"},
{name: "run-stage missing config flags", args: []string{"run-stage", "polish"}, want: "run-stage: no pipeline config path provided and no default pipeline config found; searched:"},
{name: "run missing config uses defaults", args: []string{"run", "--session", "session.yml"}, want: "run: no pipeline config path provided and no default pipeline config found; searched:"},
}

View File

@@ -21,9 +21,11 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
var pipelinePath string
var sessionPath string
var sessionID string
var force bool
fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)")
fs.StringVar(&sessionPath, "session", "", "path to session.yml")
fs.StringVar(&sessionID, "session-id", "", "session identifier for session.yml templates")
fs.BoolVar(&force, "force", false, "force stage execution (reserved for future behavior)")
if err := fs.Parse(args); err != nil {
@@ -32,16 +34,18 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
if fs.NArg() != 0 {
return fmt.Errorf("plan: unexpected positional arguments")
}
if sessionPath == "" {
return fmt.Errorf("plan: --session is required")
}
resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
if err != nil {
return fmt.Errorf("plan: %w", err)
}
resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
if err != nil {
return fmt.Errorf("plan: %w", err)
}
cfg, err := config.Load(resolvedPipelinePath, sessionPath)
cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedSessionPath, config.SessionLoadOptions{
SessionID: sessionID,
})
if err != nil {
return fmt.Errorf("plan: %w", err)
}

View File

@@ -0,0 +1,208 @@
package app
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m *manifest.Manifest, executed []string) error {
if env == nil || env.Config == nil || env.Config.Pipeline == nil || m == nil {
return nil
}
spoolRequested := env.Config.Pipeline.Spool.DeleteAudioAfterArchive
workRequested := env.Config.Pipeline.Workspace.CleanupAfterArchive
if !spoolRequested && !workRequested {
return nil
}
sr := archiveStageRecordForCleanup(m, executed)
if sr == nil {
return nil
}
if sr.Metadata == nil {
sr.Metadata = map[string]any{}
}
sr.Metadata["spool_cleanup_requested"] = spoolRequested
sr.Metadata["workdir_cleanup_requested"] = workRequested
eligible, reason := archiveCleanupEligible(env.Config, sr)
if !eligible {
sr.Metadata["cleanup_skipped"] = true
sr.Metadata["cleanup_skipped_reason"] = reason
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return fmt.Errorf("save manifest cleanup skip metadata %q: %w", manifestPath, err)
}
return nil
}
spoolDir := strings.TrimSpace(m.LocalSpoolDir)
if spoolDir == "" {
spoolDir = artifacts.SessionSpoolAudioDir(
env.Config.Pipeline.Spool.Root,
strings.TrimSpace(env.Config.Session.Campaign),
strings.TrimSpace(env.Config.Session.SessionID),
strings.TrimSpace(m.RunID),
)
}
workDir := strings.TrimSpace(m.LocalWorkDir)
if workDir == "" {
workDir = artifacts.SessionRunWorkDir(
env.Config.Pipeline.Workspace.Root,
strings.TrimSpace(env.Config.Session.Campaign),
strings.TrimSpace(env.Config.Session.SessionID),
strings.TrimSpace(m.RunID),
)
}
if spoolRequested {
if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Spool.Root), spoolDir, "pipeline.spool.delete_audio_after_archive"); err != nil {
sr.Metadata["cleanup_failed"] = true
sr.Metadata["cleanup_failed_policy"] = "pipeline.spool.delete_audio_after_archive"
sr.Metadata["cleanup_failed_path"] = spoolDir
_ = env.ManifestStore.Save(ctx, manifestPath, m)
return err
}
sr.Metadata["spool_cleanup_deleted"] = filepath.Clean(spoolDir)
}
if !workRequested {
sr.Metadata["cleanup_completed"] = true
sr.Metadata["cleanup_skipped"] = false
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return fmt.Errorf("save manifest cleanup metadata %q: %w", manifestPath, err)
}
return nil
}
if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Workspace.Root), workDir, "pipeline.workspace.cleanup_after_archive"); err != nil {
sr.Metadata["cleanup_failed"] = true
sr.Metadata["cleanup_failed_policy"] = "pipeline.workspace.cleanup_after_archive"
sr.Metadata["cleanup_failed_path"] = workDir
_ = env.ManifestStore.Save(ctx, manifestPath, m)
return err
}
sr.Metadata["workdir_cleanup_deleted"] = filepath.Clean(workDir)
sr.Metadata["cleanup_completed"] = true
sr.Metadata["cleanup_skipped"] = false
return nil
}
func archiveStageRecordForCleanup(m *manifest.Manifest, executed []string) *manifest.StageRecord {
if m == nil {
return nil
}
archiveRan := false
for _, name := range executed {
if name == "archive" {
archiveRan = true
break
}
}
if !archiveRan {
return nil
}
sr := m.Stages["archive"]
if sr == nil || sr.Status != manifest.StatusSucceeded {
return nil
}
return sr
}
func archiveCleanupEligible(cfg *config.Config, sr *manifest.StageRecord) (bool, string) {
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Archive == nil {
return false, "archive configuration is missing"
}
enabled := true
if cfg.Pipeline.Archive.Enabled != nil {
enabled = *cfg.Pipeline.Archive.Enabled
}
if !enabled {
return false, "archive.enabled is false"
}
uploadRun := true
if cfg.Pipeline.Archive.UploadRun != nil {
uploadRun = *cfg.Pipeline.Archive.UploadRun
}
if !uploadRun {
return false, "archive.upload_run is false"
}
if sr == nil || sr.Metadata == nil {
return false, "archive metadata is missing"
}
if skipped, _ := sr.Metadata["skipped"].(bool); skipped {
return false, "archive stage was skipped"
}
if uploaded, _ := sr.Metadata["uploaded"].(bool); !uploaded {
return false, "archive did not upload run record"
}
if pointer, _ := sr.Metadata["current_pointer_written"].(bool); !pointer {
return false, "archive did not write current pointer"
}
if strings.TrimSpace(asString(sr.Metadata["current_run_id_key"])) == "" {
return false, "archive current run pointer key is missing"
}
return true, ""
}
func removeRunScopedDir(root, target, policy string) error {
cleanRoot := strings.TrimSpace(root)
cleanTarget := strings.TrimSpace(target)
if cleanRoot == "" {
return fmt.Errorf("cleanup policy %s: root path is required", policy)
}
if cleanTarget == "" {
return fmt.Errorf("cleanup policy %s: target path is required", policy)
}
rootAbs, err := filepath.Abs(cleanRoot)
if err != nil {
return fmt.Errorf("cleanup policy %s: resolve root %q: %w", policy, cleanRoot, err)
}
targetAbs, err := filepath.Abs(cleanTarget)
if err != nil {
return fmt.Errorf("cleanup policy %s: resolve target %q: %w", policy, cleanTarget, err)
}
rel, err := filepath.Rel(rootAbs, targetAbs)
if err != nil {
return fmt.Errorf("cleanup policy %s: relative path from %q to %q: %w", policy, rootAbs, targetAbs, err)
}
if rel == "." {
return fmt.Errorf("cleanup policy %s: refusing to delete root directory %q", policy, rootAbs)
}
if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return fmt.Errorf("cleanup policy %s: refusing to delete path outside root: root=%q target=%q", policy, rootAbs, targetAbs)
}
info, err := os.Lstat(targetAbs)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return fmt.Errorf("cleanup policy %s: stat target %q: %w", policy, targetAbs, err)
}
if info.Mode()&os.ModeSymlink != 0 {
return fmt.Errorf("cleanup policy %s: refusing to delete symlink path %q", policy, targetAbs)
}
if !info.IsDir() {
return fmt.Errorf("cleanup policy %s: target %q is not a directory", policy, targetAbs)
}
if err := os.RemoveAll(targetAbs); err != nil {
return fmt.Errorf("cleanup policy %s: remove %q: %w", policy, targetAbs, err)
}
return nil
}
func asString(v any) string {
s, _ := v.(string)
return s
}

View File

@@ -0,0 +1,396 @@
package app
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
type archiveSuccessStage struct {
metadata map[string]any
}
func (archiveSuccessStage) Name() string { return "archive" }
func (archiveSuccessStage) Declares() stage.IODecl { return stage.IODecl{} }
func (s archiveSuccessStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
md := map[string]any{
"stage": "archive",
"uploaded": true,
"current_pointer_written": true,
"current_run_id_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/run_id.txt",
}
for k, v := range s.metadata {
md[k] = v
}
return &stage.StageResult{Metadata: md}, nil
}
type notifyFailStage struct{}
func (notifyFailStage) Name() string { return "notify" }
func (notifyFailStage) Declares() stage.IODecl { return stage.IODecl{} }
func (notifyFailStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
return nil, errors.New("notify failed")
}
func TestPostArchiveCleanupDisabledKeepsLocalDirs(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = false
cfg.Pipeline.Workspace.CleanupAfterArchive = false
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
assertExists(t, seed.localSourceAudio)
}
func TestPostArchiveCleanupSpoolOnly(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = false
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
assertMissing(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
assertExists(t, seed.localSourceAudio)
}
func TestPostArchiveCleanupWorkdirOnly(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = false
cfg.Pipeline.Workspace.CleanupAfterArchive = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
assertExists(t, cfg.Pipeline.Workspace.Root)
assertExists(t, seed.otherRunDir)
assertMissing(t, seed.runWorkDir)
assertExists(t, seed.spoolAudioDir)
}
func TestPostArchiveCleanupBothPolicies(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
assertMissing(t, seed.spoolAudioDir)
assertMissing(t, seed.runWorkDir)
assertExists(t, seed.otherRunDir)
}
func TestPostArchiveCleanupNotRunWhenArchiveFails(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
_, err := executeStages(context.Background(), cfg, []stage.Stage{failingStage{name: "archive", err: errors.New("archive failed")}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}})
if err == nil || !strings.Contains(err.Error(), "stage \"archive\" failed") {
t.Fatalf("executeStages() error = %v, want archive failure", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
func TestPostArchiveCleanupNotRunWhenArchiveSkipped(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"skipped": true}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
func TestPostArchiveCleanupNotRunWhenCurrentPointerMissing(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"current_pointer_written": false}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
func TestPostArchiveCleanupNotRunWhenArchiveUploadDisabled(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
cfg.Pipeline.Archive.UploadRun = boolPtr(false)
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
func TestPostArchiveCleanupWaitsUntilAllStagesSucceed(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
_, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}, notifyFailStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}})
if err == nil || !strings.Contains(err.Error(), "stage \"notify\" failed") {
t.Fatalf("executeStages() error = %v, want notify failure", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
func TestPostArchiveCleanupFailsOnUnsafePath(t *testing.T) {
cfg, _ := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = false
manifestPath := manifestPathFor(cfg)
store := &manifest.LocalStore{}
m, err := store.Load(context.Background(), manifestPath)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
m.LocalSpoolDir = filepath.Join(filepath.Dir(cfg.Pipeline.Spool.Root), "outside-spool")
if err := store.Save(context.Background(), manifestPath, m); err != nil {
t.Fatalf("Save() error = %v", err)
}
_, err = executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}})
if err == nil || !strings.Contains(err.Error(), "refusing to delete path outside root") {
t.Fatalf("executeStages() error = %v, want safe-path failure", err)
}
}
func TestPostArchiveCleanupNotRunWhenPromotionIsMissing(t *testing.T) {
cfg, seed, runID := archiveStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
cfg.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
{From: "artifacts/missing.md", To: "artifacts/missing.md", Required: boolPtr(true)},
}
archiveStageImpl, err := stage.Select("archive")
if err != nil {
t.Fatalf("Select(archive) error = %v", err)
}
_, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}})
if err == nil || !strings.Contains(err.Error(), "required promotion source missing") {
t.Fatalf("executeStages() error = %v, want promotion-missing failure", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
assertExists(t, filepath.Join(seed.runWorkDir, "manifest.json"))
assertExists(t, artifacts.SessionRunWorkDir(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID))
}
func TestPostArchiveCleanupNotRunWhenCurrentManifestUploadFails(t *testing.T) {
cfg, seed, _ := archiveStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
failKey := seed.sessionPrefix + "current/manifest.json"
archiveStageImpl, err := stage.Select("archive")
if err != nil {
t.Fatalf("Select(archive) error = %v", err)
}
_, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{
Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}},
})
if err == nil || !strings.Contains(err.Error(), "current manifest") {
t.Fatalf("executeStages() error = %v, want current-manifest failure", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
func TestPostArchiveCleanupNotRunWhenCurrentPointerUploadFails(t *testing.T) {
cfg, seed, _ := archiveStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true
failKey := seed.sessionPrefix + "current/run_id.txt"
archiveStageImpl, err := stage.Select("archive")
if err != nil {
t.Fatalf("Select(archive) error = %v", err)
}
_, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{
Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}},
})
if err == nil || !strings.Contains(err.Error(), "current run pointer") {
t.Fatalf("executeStages() error = %v, want current-run-pointer failure", err)
}
assertExists(t, seed.spoolAudioDir)
assertExists(t, seed.runWorkDir)
}
type cleanupSeed struct {
runWorkDir string
otherRunDir string
spoolAudioDir string
localSourceAudio string
sessionPrefix string
}
func cleanupFixtureConfig(t *testing.T) (*config.Config, cleanupSeed) {
t.Helper()
cfg := testConfig(t)
cfg.Pipeline.Archive = &config.ArchiveConfig{Enabled: boolPtr(true), UploadRun: boolPtr(true)}
cfg.Pipeline.Spool.Root = filepath.Join(t.TempDir(), "spool")
runID := "20260516T010203Z-1a2b3c4d"
runWorkDir := artifacts.SessionRunWorkDir(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID)
otherRunDir := artifacts.SessionRunWorkDir(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, "20260516T010204Z-5e6f7a8b")
spoolAudioDir := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID)
mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n")
mustWriteFile(t, filepath.Join(runWorkDir, "logs", "stage.log"), "log\n")
mustWriteFile(t, filepath.Join(otherRunDir, "logs", "stage.log"), "other\n")
mustWriteFile(t, filepath.Join(spoolAudioDir, "speaker.flac"), "flac\n")
localSourceAudio := filepath.Join(filepath.Dir(cfg.SessionPath), "audio", "alice.flac")
mustWriteFile(t, localSourceAudio, "source\n")
seed := manifest.New(cfg.Session.SessionID, time.Now().UTC())
seed.Campaign = cfg.Session.Campaign
seed.RunID = runID
seed.LocalWorkDir = runWorkDir
seed.LocalSpoolDir = spoolAudioDir
seed.S3Bucket = "my-dnd-archive"
seed.S3SessionPrefix = "dnd/campaigns/sample-campaign/sessions/2026-05-03/"
seed.S3RunPrefix = seed.S3SessionPrefix + "runs/" + runID + "/"
store := &manifest.LocalStore{}
if err := os.MkdirAll(filepath.Dir(manifestPathFor(cfg)), 0o755); err != nil {
t.Fatalf("MkdirAll() error = %v", err)
}
if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil {
t.Fatalf("seed manifest save error = %v", err)
}
return cfg, cleanupSeed{
runWorkDir: runWorkDir,
otherRunDir: otherRunDir,
spoolAudioDir: spoolAudioDir,
localSourceAudio: localSourceAudio,
sessionPrefix: seed.S3SessionPrefix,
}
}
func archiveStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, string) {
t.Helper()
cfg, seed := cleanupFixtureConfig(t)
runID := "20260516T010203Z-1a2b3c4d"
cfg.Pipeline.Storage.S3 = &config.StorageS3Config{
Bucket: "my-dnd-archive",
RootPrefix: "dnd",
}
cfg.Pipeline.Archive = &config.ArchiveConfig{
Enabled: boolPtr(true),
UploadRun: boolPtr(true),
PromoteArtifacts: []config.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)},
},
}
writeArchiveFixtureRunFiles(t, seed.runWorkDir)
store := &manifest.LocalStore{}
seedManifest, err := store.Load(context.Background(), manifestPathFor(cfg))
if err != nil {
t.Fatalf("Load() error = %v", err)
}
for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze"} {
seedManifest.MarkStageSucceeded(name, time.Now().UTC(), nil)
}
seedManifest.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", cfg.Session.Campaign, cfg.Session.SessionID)
seedManifest.S3RunPrefix = artifacts.S3RunPrefix(seedManifest.S3SessionPrefix, runID)
if err := store.Save(context.Background(), manifestPathFor(cfg), seedManifest); err != nil {
t.Fatalf("Save() error = %v", err)
}
seed.sessionPrefix = seedManifest.S3SessionPrefix
return cfg, seed, runID
}
func writeArchiveFixtureRunFiles(t *testing.T, runWorkDir string) {
t.Helper()
mustWriteFile(t, filepath.Join(runWorkDir, "inputs", "session.yml"), "session_id: 2026-05-03\n")
mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "raw", "speaker.json"), "{}\n")
mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "trimmed.json"), "{}\n")
mustWriteFile(t, filepath.Join(runWorkDir, "artifacts", "session_recap.md"), "# recap\n")
mustWriteFile(t, filepath.Join(runWorkDir, "reports", "audita.report.json"), "{}\n")
mustWriteFile(t, filepath.Join(runWorkDir, "config", "audita.generated.yml"), "key: value\n")
mustWriteFile(t, filepath.Join(runWorkDir, "logs", "audita.stderr.log"), "stderr\n")
mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n")
}
type failKeyStore struct {
delegate *storage.FakeBackend
failKey string
}
func (s *failKeyStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) {
return s.delegate.List(ctx, prefix)
}
func (s *failKeyStore) Download(ctx context.Context, key, localPath string) error {
return s.delegate.Download(ctx, key, localPath)
}
func (s *failKeyStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) {
if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) {
return storage.ObjectInfo{}, errors.New("forced upload failure")
}
return s.delegate.Upload(ctx, localPath, key, opts)
}
func (s *failKeyStore) Exists(ctx context.Context, key string) (bool, error) {
return s.delegate.Exists(ctx, key)
}
func assertExists(t *testing.T, path string) {
t.Helper()
if _, err := os.Stat(path); err != nil {
t.Fatalf("expected path to exist %q: %v", path, err)
}
}
func assertMissing(t *testing.T, path string) {
t.Helper()
if _, err := os.Stat(path); !os.IsNotExist(err) {
t.Fatalf("expected path to be removed %q, stat err=%v", path, err)
}
}

View File

@@ -17,9 +17,11 @@ func Resume(ctx context.Context, args []string, out io.Writer) error {
var pipelinePath string
var sessionPath string
var sessionID string
var force bool
fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)")
fs.StringVar(&sessionPath, "session", "", "path to session.yml")
fs.StringVar(&sessionID, "session-id", "", "session identifier for session.yml templates")
fs.BoolVar(&force, "force", false, "force stage execution")
if err := fs.Parse(args); err != nil {
@@ -28,16 +30,18 @@ func Resume(ctx context.Context, args []string, out io.Writer) error {
if fs.NArg() != 0 {
return fmt.Errorf("resume: unexpected positional arguments")
}
if sessionPath == "" {
return fmt.Errorf("resume: --session is required")
}
resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
if err != nil {
return fmt.Errorf("resume: %w", err)
}
resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
if err != nil {
return fmt.Errorf("resume: %w", err)
}
cfg, err := config.Load(resolvedPipelinePath, sessionPath)
cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedSessionPath, config.SessionLoadOptions{
SessionID: sessionID,
})
if err != nil {
return fmt.Errorf("resume: %w", err)
}

View File

@@ -16,9 +16,11 @@ func Run(ctx context.Context, args []string, out io.Writer) error {
var pipelinePath string
var sessionPath string
var sessionID string
var force bool
fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)")
fs.StringVar(&sessionPath, "session", "", "path to session.yml")
fs.StringVar(&sessionID, "session-id", "", "session identifier for session.yml templates")
fs.BoolVar(&force, "force", false, "force stage execution (reserved for future behavior)")
if err := fs.Parse(args); err != nil {
@@ -27,16 +29,18 @@ func Run(ctx context.Context, args []string, out io.Writer) error {
if fs.NArg() != 0 {
return fmt.Errorf("run: unexpected positional arguments")
}
if sessionPath == "" {
return fmt.Errorf("run: --session is required")
}
resolvedPipelinePath, err := resolvePipelineConfigPath(pipelinePath)
if err != nil {
return fmt.Errorf("run: %w", err)
}
resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
if err != nil {
return fmt.Errorf("run: %w", err)
}
cfg, err := config.Load(resolvedPipelinePath, sessionPath)
cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedSessionPath, config.SessionLoadOptions{
SessionID: sessionID,
})
if err != nil {
return fmt.Errorf("run: %w", err)
}

View File

@@ -16,9 +16,11 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error {
var pipelinePath string
var sessionPath string
var sessionID string
var force bool
fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)")
fs.StringVar(&sessionPath, "session", "", "path to session.yml")
fs.StringVar(&sessionID, "session-id", "", "session identifier for session.yml templates")
fs.BoolVar(&force, "force", false, "force stage execution (reserved for future behavior)")
if err := fs.Parse(args); err != nil {
@@ -27,10 +29,6 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error {
if fs.NArg() != 1 {
return fmt.Errorf("run-stage: expected exactly one stage name")
}
if sessionPath == "" {
return fmt.Errorf("run-stage: --session is required")
}
stageName := fs.Arg(0)
stages, err := BuildSingleStagePlan(stageName)
if err != nil {
@@ -41,8 +39,14 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error {
if err != nil {
return fmt.Errorf("run-stage: %w", err)
}
resolvedSessionPath, err := resolveSessionConfigPath(sessionPath)
if err != nil {
return fmt.Errorf("run-stage: %w", err)
}
cfg, err := config.Load(resolvedPipelinePath, sessionPath)
cfg, err := config.LoadWithSessionOptions(resolvedPipelinePath, resolvedSessionPath, config.SessionLoadOptions{
SessionID: sessionID,
})
if err != nil {
return fmt.Errorf("run-stage: %w", err)
}

View File

@@ -168,6 +168,10 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
env.Logger.Info("stage succeeded", "stage", s.Name())
}
if err := runPostArchiveCleanup(ctx, env, manifestPath, m, executed); err != nil {
return nil, fmt.Errorf("post-archive cleanup: %w", err)
}
return &RunSummary{
SessionID: cfg.Session.SessionID,
ManifestPath: manifestPath,
@@ -246,7 +250,7 @@ func buildDefaultAuditaRunner(cfg *config.Config) (audita.Runner, error) {
}
a := cfg.Pipeline.Audita
if strings.TrimSpace(a.Binary) == "" || strings.TrimSpace(a.Timeout) == "" || strings.TrimSpace(a.BaseURL) == "" || strings.TrimSpace(a.Model) == "" {
if strings.TrimSpace(a.Binary) == "" || strings.TrimSpace(a.Timeout) == "" {
// Compatibility fallback for tests or internal call paths that bypass config validation/defaults.
return &audita.NoopRunner{}, nil
}

View File

@@ -497,6 +497,55 @@ func testConfig(t *testing.T) *config.Config {
}
}
func TestBuildDefaultRunnersWithOmittedToolSections(t *testing.T) {
dir := t.TempDir()
pipelinePath := filepath.Join(dir, "pipeline.yml")
sessionPath := filepath.Join(dir, "session.yml")
pipelineYAML := `workspace:
root: ` + t.TempDir() + `
whisperx:
transcribe_url: https://example.com/transcribe
analyzer:
timeout: 20m
notification:
timeout: 10s
`
sessionYAML := `session_id: 2026-05-03
campaign: sample-campaign
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
mustWriteFile(t, pipelinePath, pipelineYAML)
mustWriteFile(t, sessionPath, sessionYAML)
cfg, err := config.Load(pipelinePath, sessionPath)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if err := config.Validate(cfg); err != nil {
t.Fatalf("Validate() error = %v", err)
}
serRunner, err := buildDefaultSeriatimRunner(cfg)
if err != nil {
t.Fatalf("buildDefaultSeriatimRunner() error = %v", err)
}
if _, ok := serRunner.(*seriatim.SubprocessRunner); !ok {
t.Fatalf("seriatim runner type = %T, want *seriatim.SubprocessRunner", serRunner)
}
audRunner, err := buildDefaultAuditaRunner(cfg)
if err != nil {
t.Fatalf("buildDefaultAuditaRunner() error = %v", err)
}
if _, ok := audRunner.(*audita.SubprocessRunner); !ok {
t.Fatalf("audita runner type = %T, want *audita.SubprocessRunner", audRunner)
}
}
func mustWriteFile(t *testing.T, path, contents string) {
t.Helper()
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {

View File

@@ -0,0 +1,86 @@
package app
import (
"bytes"
"context"
"os"
"path/filepath"
"strings"
"testing"
)
func TestPlanUsesDiscoveredSessionTemplateWithSessionID(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
sessionTemplate := `session_id: "{{ session_id }}"
campaign: sample-campaign
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
if err := os.WriteFile(sessionPath, []byte(sessionTemplate), 0o644); err != nil {
t.Fatalf("write session template: %v", err)
}
cwd := filepath.Dir(sessionPath)
originalWD, err := os.Getwd()
if err != nil {
t.Fatalf("Getwd(): %v", err)
}
if err := os.Chdir(cwd); err != nil {
t.Fatalf("Chdir(%q): %v", cwd, err)
}
t.Cleanup(func() { _ = os.Chdir(originalWD) })
var out bytes.Buffer
if err := Plan(context.Background(), []string{"--config", pipelinePath, "--session-id", "2026-04-04"}, &out); err != nil {
t.Fatalf("Plan() error = %v", err)
}
if !strings.Contains(out.String(), "narratio plan: workdir prepared") {
t.Fatalf("output = %q, want plan output", out.String())
}
}
func TestPlanFailsWhenSessionIDMismatchesConcreteSession(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
var out bytes.Buffer
err := Plan(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath, "--session-id", "2026-04-04"}, &out)
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "session_id mismatch") {
t.Fatalf("error = %q, want mismatch context", err.Error())
}
}
func TestRunStageAcceptsSessionIDFlagAndParsesStageName(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
var out bytes.Buffer
err := RunStage(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath, "--session-id", "2026-05-03", "prepare"}, &out)
if err != nil {
t.Fatalf("RunStage() error = %v", err)
}
if !strings.Contains(out.String(), "stage=prepare") {
t.Fatalf("output = %q, want stage output", out.String())
}
}
func TestResolveSessionConfigPathErrorIncludesSearchedPaths(t *testing.T) {
_, err := resolveSessionConfigPathWithCandidates("", []string{"./session.yml", "/usr/local/etc/narratio/session.yml", "/etc/narratio/session.yml"})
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "searched") {
t.Fatalf("error = %q, want searched paths", err.Error())
}
if !strings.Contains(err.Error(), "pass --session") {
t.Fatalf("error = %q, want explicit-session guidance", err.Error())
}
}

View File

@@ -0,0 +1,49 @@
package app
import (
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/config"
)
func resolveSessionConfigPath(flagValue string) (string, error) {
return resolveSessionConfigPathWithCandidates(flagValue, config.DefaultSessionConfigSearchPaths)
}
func resolveSessionConfigPathWithCandidates(flagValue string, candidates []string) (string, error) {
if explicit := strings.TrimSpace(flagValue); explicit != "" {
return explicit, nil
}
ordered := make([]string, 0, len(candidates))
for _, raw := range candidates {
path := strings.TrimSpace(raw)
if path == "" {
continue
}
ordered = append(ordered, path)
info, err := os.Stat(path)
if err == nil {
if info.IsDir() {
continue
}
return filepath.Clean(path), nil
}
if errors.Is(err, os.ErrNotExist) {
continue
}
return "", fmt.Errorf("check default session config %q: %w", path, err)
}
if len(ordered) == 0 {
return "", fmt.Errorf("no session config path provided and no default locations configured")
}
return "", fmt.Errorf(
"no session config path provided and no default session config found; searched: %s; pass --session to use an explicit path",
strings.Join(ordered, ", "),
)
}

View File

@@ -0,0 +1,68 @@
package app
import (
"os"
"path/filepath"
"strings"
"testing"
)
func TestResolveSessionConfigPathWithCandidatesExplicitWins(t *testing.T) {
got, err := resolveSessionConfigPathWithCandidates(" ./custom/session.yml ", []string{"./session.yml", "/a", "/b"})
if err != nil {
t.Fatalf("resolveSessionConfigPathWithCandidates() error = %v", err)
}
if got != "./custom/session.yml" {
t.Fatalf("resolved path = %q, want explicit path", got)
}
}
func TestResolveSessionConfigPathWithCandidatesUsesFirstExisting(t *testing.T) {
dir := t.TempDir()
first := filepath.Join(dir, "first.yml")
second := filepath.Join(dir, "second.yml")
if err := os.WriteFile(second, []byte("session_id: 2026-05-03\n"), 0o644); err != nil {
t.Fatalf("write second default: %v", err)
}
got, err := resolveSessionConfigPathWithCandidates("", []string{first, second})
if err != nil {
t.Fatalf("resolveSessionConfigPathWithCandidates() error = %v", err)
}
if got != filepath.Clean(second) {
t.Fatalf("resolved path = %q, want %q", got, filepath.Clean(second))
}
}
func TestResolveSessionConfigPathWithCandidatesPrecedence(t *testing.T) {
dir := t.TempDir()
first := filepath.Join(dir, "first.yml")
second := filepath.Join(dir, "second.yml")
if err := os.WriteFile(first, []byte("session_id: 2026-05-03\n"), 0o644); err != nil {
t.Fatalf("write first default: %v", err)
}
if err := os.WriteFile(second, []byte("session_id: 2026-05-03\n"), 0o644); err != nil {
t.Fatalf("write second default: %v", err)
}
got, err := resolveSessionConfigPathWithCandidates("", []string{first, second})
if err != nil {
t.Fatalf("resolveSessionConfigPathWithCandidates() error = %v", err)
}
if got != filepath.Clean(first) {
t.Fatalf("resolved path = %q, want first candidate %q", got, filepath.Clean(first))
}
}
func TestResolveSessionConfigPathWithCandidatesMissing(t *testing.T) {
_, err := resolveSessionConfigPathWithCandidates("", []string{"/does/not/exist/one.yml", "/does/not/exist/two.yml"})
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "no default session config found") {
t.Fatalf("error = %q, want missing-defaults context", err.Error())
}
if !strings.Contains(err.Error(), "pass --session") {
t.Fatalf("error = %q, want explicit-path guidance", err.Error())
}
}

View File

@@ -37,6 +37,7 @@ type SessionConfig struct {
// WorkspaceConfig configures local workspace behavior.
type WorkspaceConfig struct {
Root string `yaml:"root"`
CleanupAfterArchive bool `yaml:"cleanup_after_archive"`
}
// SecretsConfig configures optional local filesystem secret loading.
@@ -59,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.

View File

@@ -5,6 +5,11 @@ package config
const (
DefaultPipelineConfigPathUsrLocal = "/usr/local/etc/narratio/pipeline.yml"
DefaultPipelineConfigPathEtc = "/etc/narratio/pipeline.yml"
DefaultSessionConfigPathLocal = "./session.yml"
DefaultSessionConfigPathUsrLocal = "/usr/local/etc/narratio/session.yml"
DefaultSessionConfigPathEtc = "/etc/narratio/session.yml"
DefaultS3AccessKeyIDEnv = "OBJECT_STORAGE_KEY_ID"
DefaultS3SecretAccessKeyEnv = "OBJECT_STORAGE_KEY"
)
// DefaultPipelineConfigSearchPaths defines the default search order for
@@ -16,3 +21,14 @@ var DefaultPipelineConfigSearchPaths = []string{
DefaultPipelineConfigPathUsrLocal,
DefaultPipelineConfigPathEtc,
}
// DefaultSessionConfigSearchPaths defines the default search order for
// session.yml when callers do not provide an explicit path.
//
// Keep this in a variable so future defaults can be extended without changing
// call sites.
var DefaultSessionConfigSearchPaths = []string{
DefaultSessionConfigPathLocal,
DefaultSessionConfigPathUsrLocal,
DefaultSessionConfigPathEtc,
}

View File

@@ -5,6 +5,8 @@ import (
"io"
"os"
"path/filepath"
"regexp"
"strings"
"gopkg.in/yaml.v3"
)
@@ -21,21 +23,56 @@ func LoadPipeline(path string) (*PipelineConfig, error) {
// LoadSession loads session configuration from a YAML file with strict field checking.
func LoadSession(path string) (*SessionConfig, error) {
var cfg SessionConfig
if err := decodeStrictYAML("session", path, &cfg); err != nil {
return LoadSessionWithOptions(path, SessionLoadOptions{})
}
// SessionLoadOptions configures session template rendering behavior.
type SessionLoadOptions struct {
SessionID string
}
// LoadSessionWithOptions loads session configuration from a YAML file with
// strict field checking after template rendering.
func LoadSessionWithOptions(path string, opts SessionLoadOptions) (*SessionConfig, error) {
sessionBytes, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("load session config: session file %q: open: %w", path, err)
}
rendered, err := renderSessionTemplate(string(sessionBytes), opts)
if err != nil {
return nil, fmt.Errorf("load session config: %w", err)
}
var cfg SessionConfig
if err := decodeStrictYAMLFromReader("session", path, strings.NewReader(rendered), &cfg); err != nil {
return nil, fmt.Errorf("load session config: %w", err)
}
if strings.TrimSpace(opts.SessionID) != "" && strings.TrimSpace(cfg.SessionID) != "" && strings.TrimSpace(cfg.SessionID) != strings.TrimSpace(opts.SessionID) {
return nil, fmt.Errorf(
"load session config: session file %q: session_id mismatch: --session-id %q does not match rendered session_id %q",
path,
strings.TrimSpace(opts.SessionID),
strings.TrimSpace(cfg.SessionID),
)
}
return &cfg, nil
}
// Load loads and resolves combined pipeline and session configuration.
func Load(pipelinePath, sessionPath string) (*Config, error) {
return LoadWithSessionOptions(pipelinePath, sessionPath, SessionLoadOptions{})
}
// LoadWithSessionOptions loads and resolves combined pipeline and session
// configuration with session template options.
func LoadWithSessionOptions(pipelinePath, sessionPath string, sessionOpts SessionLoadOptions) (*Config, error) {
pipelineCfg, err := LoadPipeline(pipelinePath)
if err != nil {
return nil, err
}
sessionCfg, err := LoadSession(sessionPath)
sessionCfg, err := LoadSessionWithOptions(sessionPath, sessionOpts)
if err != nil {
return nil, err
}
@@ -55,7 +92,11 @@ func decodeStrictYAML(kind, path string, out any) error {
}
defer f.Close()
dec := yaml.NewDecoder(f)
return decodeStrictYAMLFromReader(kind, path, f, out)
}
func decodeStrictYAMLFromReader(kind, path string, r io.Reader, out any) error {
dec := yaml.NewDecoder(r)
dec.KnownFields(true)
if err := dec.Decode(out); err != nil {
return fmt.Errorf("%s file %q: strict decode failed: %w", kind, path, err)
@@ -69,6 +110,36 @@ func decodeStrictYAML(kind, path string, out any) error {
return nil
}
var sessionTemplatePattern = regexp.MustCompile(`\{\{\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\}\}`)
func renderSessionTemplate(content string, opts SessionLoadOptions) (string, error) {
sessionID := strings.TrimSpace(opts.SessionID)
rendered := content
if sessionID != "" {
rendered = strings.ReplaceAll(rendered, "{{session_id}}", sessionID)
rendered = strings.ReplaceAll(rendered, "{{ session_id }}", sessionID)
}
unresolved := sessionTemplatePattern.FindAllStringSubmatch(rendered, -1)
if len(unresolved) > 0 {
vars := make([]string, 0, len(unresolved))
for _, m := range unresolved {
if len(m) > 1 {
vars = append(vars, m[1])
}
}
if len(vars) > 0 {
return "", fmt.Errorf(
"session file template rendering failed: unresolved template variable(s): %s; pass --session-id when using {{ session_id }}",
strings.Join(vars, ", "),
)
}
return "", fmt.Errorf("session file template rendering failed: unresolved template placeholders remain")
}
return rendered, nil
}
func shortName(path, fallback string) string {
base := filepath.Base(path)
if base == "." || base == string(filepath.Separator) {
@@ -105,6 +176,12 @@ func applyStorageDefaults(cfg *StorageConfig) {
if cfg.S3.RootPrefix == "" {
cfg.S3.RootPrefix = "dnd"
}
if cfg.S3.AccessKeyIDEnv == "" {
cfg.S3.AccessKeyIDEnv = DefaultS3AccessKeyIDEnv
}
if cfg.S3.SecretKeyEnv == "" {
cfg.S3.SecretKeyEnv = DefaultS3SecretAccessKeyEnv
}
}
func applySpoolDefaults(cfg *SpoolConfig) {
@@ -173,6 +250,9 @@ func applySeriatimDefaults(cfg *SeriatimConfig) {
if cfg == nil {
return
}
if cfg.Binary == "" {
cfg.Binary = "seriatim"
}
if cfg.Timeout == "" {
cfg.Timeout = "10m"
}
@@ -191,15 +271,12 @@ func applyAuditaDefaults(cfg *AuditaConfig) {
if cfg == nil {
return
}
if cfg.Binary == "" {
cfg.Binary = "audita"
}
if cfg.Timeout == "" {
cfg.Timeout = "3h"
}
if cfg.BaseURL == "" {
cfg.BaseURL = "https://openrouter.ai/api/v1"
}
if cfg.Model == "" {
cfg.Model = "openrouter/google/gemma-4-31b-it"
}
if cfg.Report == nil {
cfg.Report = boolPtr(true)
}
@@ -209,6 +286,9 @@ func applyScriptoriumDefaults(cfg *ScriptoriumConfig) {
if cfg == nil {
return
}
if cfg.Binary == "" {
cfg.Binary = "scriptorium"
}
if cfg.Timeout == "" {
cfg.Timeout = "10m"
}

View File

@@ -37,6 +37,26 @@ inputs:
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`,
checkDefault: true,
},
{
name: "seriatim and audita sections can be omitted",
pipelineYAML: `workspace:
root: /tmp/narratio
whisperx:
transcribe_url: https://transcription.ai.rakestrawhome.com/transcribe
analyzer:
timeout: 20m
notification:
timeout: 15s
`,
sessionYAML: `session_id: 2026-05-03
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`,
checkDefault: true,
},
@@ -296,7 +316,7 @@ inputs:
wantLoadErr: "strict decode failed",
},
{
name: "missing seriatim binary fails",
name: "missing seriatim binary uses default",
pipelineYAML: `workspace:
root: /tmp/narratio
whisperx:
@@ -311,7 +331,6 @@ inputs:
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`,
wantValidate: "pipeline config \"pipeline.yml\" invalid: pipeline.seriatim.binary is required",
},
{
name: "invalid seriatim timeout fails",
@@ -412,7 +431,7 @@ inputs:
wantLoadErr: "strict decode failed",
},
{
name: "missing audita binary fails",
name: "missing audita binary uses default",
pipelineYAML: `workspace:
root: /tmp/narratio
whisperx:
@@ -429,7 +448,6 @@ inputs:
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`,
wantValidate: "pipeline config \"pipeline.yml\" invalid: pipeline.audita.binary is required",
},
{
name: "invalid audita timeout fails",
@@ -713,6 +731,9 @@ inputs:
if cfg.Pipeline.Seriatim.Timeout != "10m" {
t.Fatalf("seriatim.timeout = %q, want %q", cfg.Pipeline.Seriatim.Timeout, "10m")
}
if cfg.Pipeline.Seriatim.Binary != "seriatim" {
t.Fatalf("seriatim.binary = %q, want %q", cfg.Pipeline.Seriatim.Binary, "seriatim")
}
if cfg.Pipeline.Seriatim.OutputSchema != "seriatim-intermediate" {
t.Fatalf("seriatim.output_schema = %q, want %q", cfg.Pipeline.Seriatim.OutputSchema, "seriatim-intermediate")
}
@@ -725,17 +746,20 @@ inputs:
if cfg.Pipeline.Audita.Timeout != "3h" {
t.Fatalf("audita.timeout = %q, want %q", cfg.Pipeline.Audita.Timeout, "3h")
}
if cfg.Pipeline.Audita.Binary != "audita" {
t.Fatalf("audita.binary = %q, want %q", cfg.Pipeline.Audita.Binary, "audita")
}
if cfg.Pipeline.Audita.LLMAPIKeyEnv != "" {
t.Fatalf("audita.llm_api_key_env = %q, want empty by default", cfg.Pipeline.Audita.LLMAPIKeyEnv)
}
if cfg.Pipeline.Audita.Modules != nil {
t.Fatalf("audita.modules = %#v, want nil default (optional override)", cfg.Pipeline.Audita.Modules)
}
if cfg.Pipeline.Audita.BaseURL != "https://openrouter.ai/api/v1" {
t.Fatalf("audita.base_url = %q, want %q", cfg.Pipeline.Audita.BaseURL, "https://openrouter.ai/api/v1")
if cfg.Pipeline.Audita.BaseURL != "" {
t.Fatalf("audita.base_url = %q, want empty default", cfg.Pipeline.Audita.BaseURL)
}
if cfg.Pipeline.Audita.Model != "openrouter/google/gemma-4-31b-it" {
t.Fatalf("audita.model = %q, want %q", cfg.Pipeline.Audita.Model, "openrouter/google/gemma-4-31b-it")
if cfg.Pipeline.Audita.Model != "" {
t.Fatalf("audita.model = %q, want empty default", cfg.Pipeline.Audita.Model)
}
if cfg.Pipeline.Audita.ValidationModel != "" {
t.Fatalf("audita.validation_model = %q, want empty default", cfg.Pipeline.Audita.ValidationModel)

View File

@@ -49,11 +49,19 @@ func TestScriptoriumLoadAndValidate(t *testing.T) {
wantLoadErr: "strict decode failed",
},
{
name: "missing binary fails when section present",
name: "missing binary defaults when section present",
scriptoriumYAML: `scriptorium:
timeout: 10m
`,
wantValidateErr: "pipeline.scriptorium.binary is required",
assert: func(t *testing.T, cfg *Config) {
t.Helper()
if cfg.Pipeline.Scriptorium == nil {
t.Fatal("scriptorium config should be present")
}
if cfg.Pipeline.Scriptorium.Binary != "scriptorium" {
t.Fatalf("scriptorium.binary = %q, want scriptorium", cfg.Pipeline.Scriptorium.Binary)
}
},
},
{
name: "enabled artifact missing prompt id fails",

View File

@@ -0,0 +1,156 @@
package config
import (
"os"
"path/filepath"
"strings"
"testing"
)
func TestLoadSessionWithOptionsRendersCompactPlaceholder(t *testing.T) {
dir := t.TempDir()
sessionPath := filepath.Join(dir, "session.yml")
sessionYAML := `session_id: "{{session_id}}"
campaign: sample-campaign
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
if err := os.WriteFile(sessionPath, []byte(sessionYAML), 0o644); err != nil {
t.Fatalf("write session.yml: %v", err)
}
cfg, err := LoadSessionWithOptions(sessionPath, SessionLoadOptions{SessionID: "2026-04-04"})
if err != nil {
t.Fatalf("LoadSessionWithOptions() error = %v", err)
}
if cfg.SessionID != "2026-04-04" {
t.Fatalf("SessionID = %q, want 2026-04-04", cfg.SessionID)
}
}
func TestLoadSessionWithOptionsRendersSpacedPlaceholder(t *testing.T) {
dir := t.TempDir()
sessionPath := filepath.Join(dir, "session.yml")
sessionYAML := `session_id: "{{ session_id }}"
campaign: sample-campaign
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
if err := os.WriteFile(sessionPath, []byte(sessionYAML), 0o644); err != nil {
t.Fatalf("write session.yml: %v", err)
}
cfg, err := LoadSessionWithOptions(sessionPath, SessionLoadOptions{SessionID: "2026-04-04"})
if err != nil {
t.Fatalf("LoadSessionWithOptions() error = %v", err)
}
if cfg.SessionID != "2026-04-04" {
t.Fatalf("SessionID = %q, want 2026-04-04", cfg.SessionID)
}
}
func TestLoadSessionWithOptionsUnresolvedPlaceholderFails(t *testing.T) {
dir := t.TempDir()
sessionPath := filepath.Join(dir, "session.yml")
sessionYAML := `session_id: "{{ session_id }}"
campaign: sample-campaign
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
if err := os.WriteFile(sessionPath, []byte(sessionYAML), 0o644); err != nil {
t.Fatalf("write session.yml: %v", err)
}
_, err := LoadSessionWithOptions(sessionPath, SessionLoadOptions{})
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "unresolved template variable") {
t.Fatalf("error = %q, want unresolved-variable context", err.Error())
}
if !strings.Contains(err.Error(), "session_id") {
t.Fatalf("error = %q, want session_id variable", err.Error())
}
}
func TestLoadSessionWithOptionsMismatchFails(t *testing.T) {
dir := t.TempDir()
sessionPath := filepath.Join(dir, "session.yml")
sessionYAML := `session_id: 2026-05-03
campaign: sample-campaign
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
if err := os.WriteFile(sessionPath, []byte(sessionYAML), 0o644); err != nil {
t.Fatalf("write session.yml: %v", err)
}
_, err := LoadSessionWithOptions(sessionPath, SessionLoadOptions{SessionID: "2026-04-04"})
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "session_id mismatch") {
t.Fatalf("error = %q, want mismatch context", err.Error())
}
}
func TestLoadSessionWithOptionsUnknownFieldStillRejectedAfterRendering(t *testing.T) {
dir := t.TempDir()
sessionPath := filepath.Join(dir, "session.yml")
sessionYAML := `session_id: "{{ session_id }}"
campaign: sample-campaign
unknown_field: true
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
if err := os.WriteFile(sessionPath, []byte(sessionYAML), 0o644); err != nil {
t.Fatalf("write session.yml: %v", err)
}
_, err := LoadSessionWithOptions(sessionPath, SessionLoadOptions{SessionID: "2026-04-04"})
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "strict decode failed") {
t.Fatalf("error = %q, want strict-decode context", err.Error())
}
}
func TestLoadSessionWithOptionsConcreteSessionStillLoads(t *testing.T) {
dir := t.TempDir()
sessionPath := filepath.Join(dir, "session.yml")
sessionYAML := `session_id: 2026-05-03
campaign: sample-campaign
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`
if err := os.WriteFile(sessionPath, []byte(sessionYAML), 0o644); err != nil {
t.Fatalf("write session.yml: %v", err)
}
cfg, err := LoadSessionWithOptions(sessionPath, SessionLoadOptions{})
if err != nil {
t.Fatalf("LoadSessionWithOptions() error = %v", err)
}
if cfg.SessionID != "2026-05-03" {
t.Fatalf("SessionID = %q, want 2026-05-03", cfg.SessionID)
}
}

View File

@@ -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)
@@ -47,6 +124,9 @@ func TestSpoolAndArchiveDefaults(t *testing.T) {
if cfg.Pipeline.Spool.DeleteAudioAfterArchive {
t.Fatalf("spool.delete_audio_after_archive = true, want false")
}
if cfg.Pipeline.Workspace.CleanupAfterArchive {
t.Fatalf("workspace.cleanup_after_archive = true, want false")
}
if cfg.Pipeline.Archive == nil {
t.Fatal("archive should be initialized by defaults")
}

View File

@@ -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
}
@@ -276,9 +282,6 @@ func validateAudita(cfg AuditaConfig) error {
return fmt.Errorf("pipeline.audita.base_url must be a valid URL")
}
}
if strings.TrimSpace(cfg.Model) == "" {
return fmt.Errorf("pipeline.audita.model is required")
}
if cfg.TotalLLMConcurrency != nil && *cfg.TotalLLMConcurrency <= 0 {
return fmt.Errorf("pipeline.audita.total_llm_concurrency must be > 0")
}
@@ -433,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)

View File

@@ -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,93 @@ 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))
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(paths.TranscriptsRawDir, "normalized", 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 {

View File

@@ -54,6 +54,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 +75,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 +95,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 +169,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 +200,51 @@ 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 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 +274,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])
}
}

View File

@@ -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."

View File

@@ -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.