Harden initial framework skeleton

This commit is contained in:
2026-05-02 11:44:15 -05:00
parent bec784f1ec
commit 96540cebd4
19 changed files with 194 additions and 75 deletions

View File

@@ -2,29 +2,58 @@
`narratio` is a Go-based orchestration application for processing D&D session audio into transcripts and downstream artifacts.
This repository is currently an **initial scaffold** that establishes:
This repository currently contains a **working scaffold** with strict config loading, local workdir/manifest handling, resumable stage control, and placeholder adapters/stages.
- Go module and package boundaries
- Placeholder CLI commands: `run`, `plan`, `status`, `resume`, `run-stage`
- Basic structured logging setup via `log/slog`
- Stub adapter, stage, config, manifest, artifact, and contract packages
## Expected Config Files
## Current State
`narratio` expects two YAML files:
The CLI commands are intentionally placeholders and print `not yet implemented` messages.
- `pipeline.yml`: pipeline/workspace settings (`workspace`, `storage`, `whisperx`, `seriatim`, `audita`, `analyzer`, `notification`)
- `session.yml`: per-session settings (`session_id`, `inputs`, optional metadata)
Example:
Decoding is strict (`KnownFields(true)`), so unknown YAML fields fail fast.
Example minimal files are available under `examples/`:
- `examples/pipeline.minimal.yml`
- `examples/session.minimal.yml`
## Current Scaffold Status
Implemented in scaffold form:
- strict config load + validation
- local artifact/workdir creation and locking
- manifest create/load/save and stage status tracking
- stage framework with real `prepare` stage and placeholder downstream stages
- resumable run control (`run`, `resume`, `run-stage`, `plan` with run/skip decisions)
- fake/no-op adapters for external tool boundaries
Not implemented yet:
- real WhisperX HTTP integration
- real Seriatim execution
- real Audita execution
- real analyzer integration
- real remote archive/storage backend
- real notification backend
## Run Tests
```bash
go run ./cmd/narratio plan
go test ./...
```
## Non-Goals (Current Pass)
## Run Plan
This scaffold does not yet implement:
```bash
go run ./cmd/narratio plan --config examples/pipeline.minimal.yml --session examples/session.minimal.yml
```
- real config loading/validation behavior
- manifest persistence behavior
- stage execution behavior
- real WhisperX/seriatim/audita/analyzer integrations
- real S3 or notification integrations
## Run Placeholder Pipeline
The current `run` command executes available scaffold behavior (`prepare` + placeholder stages) and records progress in `manifest.json`.
```bash
go run ./cmd/narratio run --config examples/pipeline.minimal.yml --session examples/session.minimal.yml
```

View File

@@ -0,0 +1,2 @@
placeholder-flac-bytes

2
examples/autocorrect.yml Normal file
View File

@@ -0,0 +1,2 @@
[]

2
examples/glossary.yml Normal file
View File

@@ -0,0 +1,2 @@
[]

View File

@@ -0,0 +1,23 @@
workspace:
root: ./tmp/narratio-workspace
storage:
backend: local
whisperx:
timeout: 15m
seriatim:
timeout: 30s
audita:
timeout: 1h
analyzer:
timeout: 20m
artifacts:
output_dir: artifacts
notification:
timeout: 10s

View File

@@ -0,0 +1,10 @@
session_id: 2026-05-03
campaign: sample-campaign
date: 2026-05-03
title: Sample Session
inputs:
audio_dir: ./audio
speakers_file: ./speakers.yml
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml

2
examples/speakers.yml Normal file
View File

@@ -0,0 +1,2 @@
sample-speaker: sample-speaker.flac

View File

@@ -3,6 +3,8 @@ package analyzer
import "context"
// TODO: implement analyzer integration once the analyzer contract is finalized.
// Runner is the adapter boundary for analyzer invocations.
type Runner interface {
Run(ctx context.Context, req AnalyzeRequest) (AnalyzeResult, error)

View File

@@ -3,6 +3,8 @@ package audita
import "context"
// TODO: implement a real Audita subprocess/service adapter.
// Runner is the adapter boundary for audita polish invocations.
type Runner interface {
Run(ctx context.Context, req PolishRequest) (PolishResult, error)

View File

@@ -3,6 +3,8 @@ package notify
import "context"
// TODO: implement a real notification backend (email/webhook/etc.).
// Sender is the adapter boundary for notifications.
type Sender interface {
Send(ctx context.Context, req SendRequest) (SendResult, error)

View File

@@ -3,6 +3,8 @@ package seriatim
import "context"
// TODO: implement a real Seriatim subprocess adapter.
// Runner is the adapter boundary for seriatim merge invocations.
type Runner interface {
Run(ctx context.Context, req MergeRequest) (MergeResult, error)

View File

@@ -3,6 +3,8 @@ package storage
import "context"
// TODO: implement remote storage/archive backends (S3/SFTP/etc.).
// Backend is the adapter boundary for archive/storage operations.
type Backend interface {
Archive(ctx context.Context, req ArchiveRequest) (ArchiveResult, error)

View File

@@ -3,6 +3,8 @@ package whisperx
import "context"
// TODO: implement a real WhisperX HTTP client adapter.
// Client is the adapter boundary for WhisperX transcription jobs.
type Client interface {
Transcribe(ctx context.Context, req TranscribeRequest) (TranscribeResult, error)

View File

@@ -1,50 +1,7 @@
package app
import (
"log/slog"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/analyzer"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/audita"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/notify"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
"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"
)
import "gitea.maximumdirect.net/eric/narratio/internal/stage"
// Env is the shared dependency container passed to orchestrator components.
type Env struct {
Config *config.Config
ArtifactStore artifacts.Store
ManifestStore manifest.Store
Logger *slog.Logger
WhisperX whisperx.Client
Seriatim seriatim.Runner
Audita audita.Runner
Analyzer analyzer.Runner
Storage storage.Backend
Notifier notify.Sender
}
func toStageEnv(env *Env) *stage.Env {
if env == nil {
return nil
}
return &stage.Env{
Config: env.Config,
ArtifactStore: env.ArtifactStore,
ManifestStore: env.ManifestStore,
Logger: env.Logger,
WhisperX: env.WhisperX,
Seriatim: env.Seriatim,
Audita: env.Audita,
Analyzer: env.Analyzer,
Storage: env.Storage,
Notifier: env.Notifier,
}
}
// It aliases stage.Env so orchestration and stages stay on one source of truth.
type Env = stage.Env

View File

@@ -89,7 +89,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
return nil, err
}
stageEnv := toStageEnv(env)
stageEnv := env
decisions := decideStageActions(stages, m, opts.Force)
@@ -102,15 +102,18 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if d.Action == stageActionSkip {
skipped = append(skipped, s.Name())
env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force)
continue
}
executed = append(executed, s.Name())
now := nowUTC()
m.MarkStageRunning(s.Name(), now)
env.Logger.Info("starting stage", "stage", s.Name())
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest before stage %q: %w", s.Name(), err)
}
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
result, err := s.Run(ctx, stageEnv, m)
if err != nil {
@@ -118,6 +121,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if saveErr := env.ManifestStore.Save(ctx, manifestPath, m); saveErr != nil {
return nil, fmt.Errorf("stage %q failed (%v) and manifest save failed (%v)", s.Name(), err, saveErr)
}
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err)
}
@@ -128,6 +132,8 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)
}
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath)
env.Logger.Info("stage succeeded", "stage", s.Name())
}
return &RunSummary{

View File

@@ -4,6 +4,7 @@ import (
"fmt"
"io"
"os"
"path/filepath"
"gopkg.in/yaml.v3"
)
@@ -11,7 +12,7 @@ import (
// LoadPipeline loads pipeline configuration from a YAML file with strict field checking.
func LoadPipeline(path string) (*PipelineConfig, error) {
var cfg PipelineConfig
if err := decodeStrictYAML(path, &cfg); err != nil {
if err := decodeStrictYAML("pipeline", path, &cfg); err != nil {
return nil, fmt.Errorf("load pipeline config: %w", err)
}
return &cfg, nil
@@ -20,7 +21,7 @@ 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(path, &cfg); err != nil {
if err := decodeStrictYAML("session", path, &cfg); err != nil {
return nil, fmt.Errorf("load session config: %w", err)
}
return &cfg, nil
@@ -46,23 +47,31 @@ func Load(pipelinePath, sessionPath string) (*Config, error) {
}, nil
}
func decodeStrictYAML(path string, out any) error {
func decodeStrictYAML(kind, path string, out any) error {
f, err := os.Open(path)
if err != nil {
return fmt.Errorf("open %q: %w", path, err)
return fmt.Errorf("%s file %q: open: %w", kind, path, err)
}
defer f.Close()
dec := yaml.NewDecoder(f)
dec.KnownFields(true)
if err := dec.Decode(out); err != nil {
return fmt.Errorf("decode %q: %w", path, err)
return fmt.Errorf("%s file %q: strict decode failed: %w", kind, path, err)
}
var extra any
if err := dec.Decode(&extra); err != nil && err != io.EOF {
return fmt.Errorf("decode trailing content in %q: %w", path, err)
return fmt.Errorf("%s file %q: trailing content decode failed: %w", kind, path, err)
}
return nil
}
func shortName(path, fallback string) string {
base := filepath.Base(path)
if base == "." || base == string(filepath.Separator) {
return fallback
}
return base
}

View File

@@ -51,7 +51,7 @@ inputs:
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`,
wantLoadErr: "field bogus not found",
wantLoadErr: "pipeline file",
},
{
name: "unknown session field fails",
@@ -66,7 +66,7 @@ inputs:
glossary_file: ./glossary.yml
unknown_field: true
`,
wantLoadErr: "field unknown_field not found",
wantLoadErr: "session file",
},
{
name: "missing required field fails",
@@ -80,7 +80,7 @@ inputs:
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`,
wantValidate: "session.session_id is required",
wantValidate: "session config \"session.yml\" invalid: session.session_id is required",
},
{
name: "invalid timeout fails",
@@ -96,7 +96,7 @@ inputs:
autocorrect_file: ./autocorrect.yml
glossary_file: ./glossary.yml
`,
wantValidate: "pipeline.whisperx.timeout must be a valid duration",
wantValidate: "pipeline config \"pipeline.yml\" invalid: pipeline.whisperx.timeout must be a valid duration",
},
}
@@ -112,6 +112,9 @@ inputs:
if !strings.Contains(err.Error(), tt.wantLoadErr) {
t.Fatalf("load error = %q, want to contain %q", err.Error(), tt.wantLoadErr)
}
if !strings.Contains(err.Error(), "strict decode failed") {
t.Fatalf("load error = %q, want strict decode context", err.Error())
}
return
}
@@ -162,6 +165,22 @@ func TestValidateMissingAudioSource(t *testing.T) {
if !strings.Contains(err.Error(), "audio_dir or at least one audio_files") {
t.Fatalf("error = %q, want audio source guidance", err.Error())
}
if !strings.Contains(err.Error(), "session config") {
t.Fatalf("error = %q, want session config context", err.Error())
}
}
func TestExamplesLoadAndValidate(t *testing.T) {
pipelinePath := filepath.Join("..", "..", "examples", "pipeline.minimal.yml")
sessionPath := filepath.Join("..", "..", "examples", "session.minimal.yml")
cfg, err := Load(pipelinePath, sessionPath)
if err != nil {
t.Fatalf("Load(examples) error = %v", err)
}
if err := Validate(cfg); err != nil {
t.Fatalf("Validate(examples) error = %v", err)
}
}
func writeConfigFiles(t *testing.T, pipelineYAML, sessionYAML string) (string, string) {

View File

@@ -19,10 +19,10 @@ func Validate(cfg *Config) error {
}
if err := validatePipeline(cfg.Pipeline); err != nil {
return err
return fmt.Errorf("pipeline config %q invalid: %w", shortName(cfg.PipelinePath, "pipeline.yml"), err)
}
if err := validateSession(cfg.Session); err != nil {
return err
return fmt.Errorf("session config %q invalid: %w", shortName(cfg.SessionPath, "session.yml"), err)
}
return nil

View File

@@ -34,6 +34,10 @@ func TestStageMarkHelpers(t *testing.T) {
if len(stage.Outputs) != 1 {
t.Fatalf("outputs len = %d, want 1", len(stage.Outputs))
}
outputs[0].LocalPath = "mutated.json"
if stage.Outputs[0].LocalPath != "transcripts/processed.json" {
t.Fatalf("stage outputs should be copied, got %#v", stage.Outputs)
}
failedAt := succeededAt.Add(1 * time.Minute)
m.MarkStageFailed("analyze", failedAt, "analyzer crashed")
@@ -64,3 +68,45 @@ func TestStageMarkHelpers(t *testing.T) {
t.Fatalf("error code = %q, want %q", skipped.Error.Code, "skipped")
}
}
func TestMarkStageRunningClearsPriorCompletionAndError(t *testing.T) {
m := New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
failedAt := time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC)
m.MarkStageFailed("merge", failedAt, "boom")
runAt := failedAt.Add(1 * time.Minute)
m.MarkStageRunning("merge", runAt)
stage := m.Stages["merge"]
if stage == nil {
t.Fatal("missing stage record")
}
if stage.Status != StatusRunning {
t.Fatalf("status = %q, want %q", stage.Status, StatusRunning)
}
if stage.CompletedAt != nil {
t.Fatalf("completed_at = %v, want nil while running", stage.CompletedAt)
}
if stage.Error != nil {
t.Fatalf("error = %#v, want nil while running", stage.Error)
}
}
func TestMarkStageSucceededClearsError(t *testing.T) {
m := New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
m.MarkStageFailed("analyze", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), "failed once")
m.MarkStageSucceeded("analyze", time.Date(2026, 5, 3, 10, 2, 0, 0, time.UTC), nil)
stage := m.Stages["analyze"]
if stage == nil {
t.Fatal("missing stage record")
}
if stage.Status != StatusSucceeded {
t.Fatalf("status = %q, want %q", stage.Status, StatusSucceeded)
}
if stage.Error != nil {
t.Fatalf("error = %#v, want nil on success", stage.Error)
}
}