diff --git a/README.md b/README.md index f8b4910..ed3808a 100644 --- a/README.md +++ b/README.md @@ -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 +``` diff --git a/examples/audio/sample-speaker.flac b/examples/audio/sample-speaker.flac new file mode 100644 index 0000000..f79771b --- /dev/null +++ b/examples/audio/sample-speaker.flac @@ -0,0 +1,2 @@ +placeholder-flac-bytes + diff --git a/examples/autocorrect.yml b/examples/autocorrect.yml new file mode 100644 index 0000000..7dd4387 --- /dev/null +++ b/examples/autocorrect.yml @@ -0,0 +1,2 @@ +[] + diff --git a/examples/glossary.yml b/examples/glossary.yml new file mode 100644 index 0000000..7dd4387 --- /dev/null +++ b/examples/glossary.yml @@ -0,0 +1,2 @@ +[] + diff --git a/examples/pipeline.minimal.yml b/examples/pipeline.minimal.yml new file mode 100644 index 0000000..8e05f38 --- /dev/null +++ b/examples/pipeline.minimal.yml @@ -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 + diff --git a/examples/session.minimal.yml b/examples/session.minimal.yml new file mode 100644 index 0000000..a5693f4 --- /dev/null +++ b/examples/session.minimal.yml @@ -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 + diff --git a/examples/speakers.yml b/examples/speakers.yml new file mode 100644 index 0000000..ed4acd7 --- /dev/null +++ b/examples/speakers.yml @@ -0,0 +1,2 @@ +sample-speaker: sample-speaker.flac + diff --git a/internal/adapters/analyzer/runner.go b/internal/adapters/analyzer/runner.go index f5ac1c2..ab2ca2c 100644 --- a/internal/adapters/analyzer/runner.go +++ b/internal/adapters/analyzer/runner.go @@ -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) diff --git a/internal/adapters/audita/runner.go b/internal/adapters/audita/runner.go index bd94533..7f73233 100644 --- a/internal/adapters/audita/runner.go +++ b/internal/adapters/audita/runner.go @@ -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) diff --git a/internal/adapters/notify/sender.go b/internal/adapters/notify/sender.go index 44e9902..520b44b 100644 --- a/internal/adapters/notify/sender.go +++ b/internal/adapters/notify/sender.go @@ -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) diff --git a/internal/adapters/seriatim/runner.go b/internal/adapters/seriatim/runner.go index 1115b16..8754f87 100644 --- a/internal/adapters/seriatim/runner.go +++ b/internal/adapters/seriatim/runner.go @@ -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) diff --git a/internal/adapters/storage/archive.go b/internal/adapters/storage/archive.go index e466fa5..bf083b5 100644 --- a/internal/adapters/storage/archive.go +++ b/internal/adapters/storage/archive.go @@ -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) diff --git a/internal/adapters/whisperx/client.go b/internal/adapters/whisperx/client.go index c835314..40925f8 100644 --- a/internal/adapters/whisperx/client.go +++ b/internal/adapters/whisperx/client.go @@ -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) diff --git a/internal/app/app.go b/internal/app/app.go index b71ee9e..b0f539c 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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 diff --git a/internal/app/runner.go b/internal/app/runner.go index 17d882e..27d137a 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -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{ diff --git a/internal/config/load.go b/internal/config/load.go index cfdcbdd..4dd28cd 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -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 +} diff --git a/internal/config/load_validate_test.go b/internal/config/load_validate_test.go index 626515e..953a085 100644 --- a/internal/config/load_validate_test.go +++ b/internal/config/load_validate_test.go @@ -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) { diff --git a/internal/config/validate.go b/internal/config/validate.go index 965c7d6..b6f42c5 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -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 diff --git a/internal/manifest/manifest_test.go b/internal/manifest/manifest_test.go index 68f612f..88c2d7e 100644 --- a/internal/manifest/manifest_test.go +++ b/internal/manifest/manifest_test.go @@ -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) + } +}