Implement prepare stage input resolution

2026-05-02 11:26:43 -05:00
parent 2854da48c2
commit 1f9d8d0ef7
6 changed files with 602 additions and 8 deletions
--- a/internal/app/commands_test.go
+++ b/internal/app/commands_test.go
@@ -172,6 +172,11 @@ inputs:
 		t.Fatalf("write session config: %v", err)
 	}

+	mustWriteTestFile(t, filepath.Join(dir, "speakers.yml"), "alice: alice.flac\n")
+	mustWriteTestFile(t, filepath.Join(dir, "autocorrect.yml"), "[]\n")
+	mustWriteTestFile(t, filepath.Join(dir, "glossary.yml"), "[]\n")
+	mustWriteTestFile(t, filepath.Join(dir, "audio", "alice.flac"), "audio-bytes")
+
 	return pipelinePath, sessionPath
 }

@@ -189,3 +194,13 @@ func writeManifestPathForExecute(t *testing.T) string {

 	return path
 }
+
+func mustWriteTestFile(t *testing.T, path, contents string) {
+	t.Helper()
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		t.Fatalf("mkdir %q: %v", path, err)
+	}
+	if err := os.WriteFile(path, []byte(contents), 0o644); err != nil {
+		t.Fatalf("write %q: %v", path, err)
+	}
+}
--- a/internal/app/runner_test.go
+++ b/internal/app/runner_test.go
@@ -57,10 +57,19 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) {
 		if sr.Status != manifest.StatusSucceeded {
 			t.Fatalf("stage %q status = %q, want %q", name, sr.Status, manifest.StatusSucceeded)
 		}
+		if name == "prepare" {
+			if sr.Metadata == nil || sr.Metadata["prepared"] != true {
+				t.Fatalf("prepare metadata missing prepared=true: %#v", sr.Metadata)
+			}
+			continue
+		}
 		if sr.Metadata == nil || sr.Metadata["placeholder"] != true {
 			t.Fatalf("stage %q missing placeholder metadata", name)
 		}
 	}
+	if len(m.Inputs) == 0 {
+		t.Fatalf("manifest inputs should be recorded by prepare")
+	}

 	if _, err := os.Stat(summary.ManifestPath); err != nil {
 		t.Fatalf("manifest file missing at %q: %v", summary.ManifestPath, err)
@@ -186,8 +195,21 @@ func testConfig(t *testing.T) *config.Config {
 	t.Helper()

 	workspace := t.TempDir()
+	cfgDir := t.TempDir()
+	sessionPath := filepath.Join(cfgDir, "session.yml")
+	pipelinePath := filepath.Join(cfgDir, "pipeline.yml")
+
+	mustWriteFile(t, pipelinePath, "workspace:\n  root: "+workspace+"\n")
+	mustWriteFile(t, sessionPath, "session_id: 2026-05-03\n")
+	mustWriteFile(t, filepath.Join(cfgDir, "speakers.yml"), "alice: alice.flac\n")
+	mustWriteFile(t, filepath.Join(cfgDir, "autocorrect.yml"), "[]\n")
+	mustWriteFile(t, filepath.Join(cfgDir, "glossary.yml"), "[]\n")
+	mustWriteFile(t, filepath.Join(cfgDir, "audio", "alice.flac"), "audio")
+
 	return &config.Config{
-		Pipeline: &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: workspace}},
+		Pipeline:     &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: workspace}},
+		PipelinePath: pipelinePath,
+		SessionPath:  sessionPath,
 		Session: &config.SessionConfig{
 			SessionID: "2026-05-03",
 			Inputs: config.SessionInputsConfig{
@@ -199,3 +221,13 @@ func testConfig(t *testing.T) *config.Config {
 		},
 	}
 }
+
+func mustWriteFile(t *testing.T, path, contents string) {
+	t.Helper()
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		t.Fatalf("MkdirAll(%q): %v", path, err)
+	}
+	if err := os.WriteFile(path, []byte(contents), 0o644); err != nil {
+		t.Fatalf("WriteFile(%q): %v", path, err)
+	}
+}
--- a/internal/stage/placeholders.go
+++ b/internal/stage/placeholders.go
@@ -163,7 +163,7 @@ func (s placeholderStage) Run(ctx context.Context, env *Env, m *manifest.Manifes
 // All returns the canonical ordered stage list for full pipeline execution.
 func All() []Stage {
 	return []Stage{
-		placeholderStage{name: "prepare"},
+		prepareStage{},
 		placeholderStage{name: "transcribe"},
 		placeholderStage{name: "normalize"},
 		placeholderStage{name: "merge"},
--- a/internal/stage/placeholders_test.go
+++ b/internal/stage/placeholders_test.go
@@ -3,6 +3,8 @@ package stage
 import (
 	"context"
 	"errors"
+	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -18,7 +20,7 @@ import (
 	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
 )

-func TestPlaceholderStagesReturnSuccessMetadata(t *testing.T) {
+func TestStagesReturnExpectedMetadata(t *testing.T) {
 	stages := All()
 	if len(stages) == 0 {
 		t.Fatal("expected non-empty stage list")
@@ -26,10 +28,15 @@ func TestPlaceholderStagesReturnSuccessMetadata(t *testing.T) {

 	root := t.TempDir()
 	store := artifacts.NewLocalStore(root)
-	_, err := store.EnsureLayout("2026-05-03")
-	if err != nil {
-		t.Fatalf("EnsureLayout() error = %v", err)
-	}
+	cfgDir := t.TempDir()
+	sessionPath := filepath.Join(cfgDir, "session.yml")
+	pipelinePath := filepath.Join(cfgDir, "pipeline.yml")
+	writeStageTestFile(t, sessionPath, "session_id: 2026-05-03\n")
+	writeStageTestFile(t, pipelinePath, "workspace:\n  root: "+root+"\n")
+	writeStageTestFile(t, filepath.Join(cfgDir, "speakers.yml"), "alice: alice.flac\n")
+	writeStageTestFile(t, filepath.Join(cfgDir, "autocorrect.yml"), "[]\n")
+	writeStageTestFile(t, filepath.Join(cfgDir, "glossary.yml"), "[]\n")
+	writeStageTestFile(t, filepath.Join(cfgDir, "audio", "placeholder-speaker.flac"), "a")

 	wf := &whisperx.FakeClient{}
 	sf := &seriatim.FakeRunner{}
@@ -39,7 +46,20 @@ func TestPlaceholderStagesReturnSuccessMetadata(t *testing.T) {
 	nf := &notify.FakeSender{}

 	env := &Env{
-		Config:        &config.Config{Session: &config.SessionConfig{SessionID: "2026-05-03"}},
+		Config: &config.Config{
+			SessionPath:  sessionPath,
+			PipelinePath: pipelinePath,
+			Pipeline:     &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: root}},
+			Session: &config.SessionConfig{
+				SessionID: "2026-05-03",
+				Inputs: config.SessionInputsConfig{
+					AudioDir:        "./audio",
+					SpeakersFile:    "./speakers.yml",
+					AutocorrectFile: "./autocorrect.yml",
+					GlossaryFile:    "./glossary.yml",
+				},
+			},
+		},
 		ArtifactStore: store,
 		WhisperX:      wf,
 		Seriatim:      sf,
@@ -58,6 +78,12 @@ func TestPlaceholderStagesReturnSuccessMetadata(t *testing.T) {
 		if result == nil {
 			t.Fatalf("stage %q returned nil result", s.Name())
 		}
+		if s.Name() == "prepare" {
+			if result.Metadata["prepared"] != true {
+				t.Fatalf("prepare metadata = %#v, want prepared=true", result.Metadata)
+			}
+			continue
+		}
 		if result.Metadata["placeholder"] != true {
 			t.Fatalf("stage %q missing placeholder metadata", s.Name())
 		}
@@ -124,3 +150,13 @@ func TestPlaceholderAdapterErrorPropagation(t *testing.T) {
 		})
 	}
 }
+
+func writeStageTestFile(t *testing.T, path, contents string) {
+	t.Helper()
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		t.Fatalf("MkdirAll(%q): %v", path, err)
+	}
+	if err := os.WriteFile(path, []byte(contents), 0o644); err != nil {
+		t.Fatalf("WriteFile(%q): %v", path, err)
+	}
+}
--- a/internal/stage/prepare.go
+++ b/internal/stage/prepare.go
@@ -0,0 +1,307 @@
+package stage
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+	"gopkg.in/yaml.v3"
+)
+
+type prepareStage struct{}
+
+func (prepareStage) Name() string { return "prepare" }
+
+func (prepareStage) Declares() IODecl {
+	return IODecl{
+		Inputs: []artifacts.Ref{
+			{Kind: "config", Category: "inputs", RelativePath: "session.yml"},
+			{Kind: "config", Category: "inputs", RelativePath: "pipeline.resolved.yml"},
+			{Kind: "config", Category: "inputs", RelativePath: "speakers.yml"},
+			{Kind: "config", Category: "inputs", RelativePath: "autocorrect.yml"},
+			{Kind: "config", Category: "inputs", RelativePath: "glossary.yml"},
+			{Kind: "audio", Category: "audio", RelativePath: "*.flac"},
+		},
+	}
+}
+
+func (prepareStage) Run(_ context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
+	if env == nil || env.Config == nil {
+		return nil, fmt.Errorf("prepare: stage environment config is required")
+	}
+	if env.ArtifactStore == nil {
+		return nil, fmt.Errorf("prepare: artifact store is required")
+	}
+	if env.Config.Session == nil || env.Config.Pipeline == nil {
+		return nil, fmt.Errorf("prepare: resolved config must include pipeline and session")
+	}
+
+	sessionID := strings.TrimSpace(m.SessionID)
+	if sessionID == "" {
+		sessionID = strings.TrimSpace(env.Config.Session.SessionID)
+	}
+	if sessionID == "" {
+		return nil, fmt.Errorf("prepare: session id is required")
+	}
+
+	paths, err := env.ArtifactStore.EnsureLayout(sessionID)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: ensure workdir layout: %w", err)
+	}
+
+	sessionSrc := env.Config.SessionPath
+	if err := requireFile(sessionSrc, "session.yml"); err != nil {
+		return nil, fmt.Errorf("prepare: %w", err)
+	}
+	sessionDir := filepath.Dir(sessionSrc)
+
+	speakersSrc, err := resolvePath(sessionDir, env.Config.Session.Inputs.SpeakersFile)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: speakers path: %w", err)
+	}
+	autocorrectSrc, err := resolvePath(sessionDir, env.Config.Session.Inputs.AutocorrectFile)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: autocorrect path: %w", err)
+	}
+	glossarySrc, err := resolvePath(sessionDir, env.Config.Session.Inputs.GlossaryFile)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: glossary path: %w", err)
+	}
+
+	for _, required := range []struct {
+		path string
+		name string
+	}{
+		{path: speakersSrc, name: "speakers.yml"},
+		{path: autocorrectSrc, name: "autocorrect.yml"},
+		{path: glossarySrc, name: "glossary.yml"},
+	} {
+		if err := requireFile(required.path, required.name); err != nil {
+			return nil, fmt.Errorf("prepare: %w", err)
+		}
+	}
+
+	resolvedAudio, err := resolveAudioFiles(sessionDir, env.Config.Session.Inputs)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: resolve audio inputs: %w", err)
+	}
+
+	copiedByDest := map[string]string{}
+	inputs := make([]manifest.InputRecord, 0, 5+len(resolvedAudio))
+	registerInput := func(kind, path, checksum string) {
+		inputs = append(inputs, manifest.InputRecord{Kind: kind, Path: path, Checksum: checksum})
+	}
+
+	sessionDst := filepath.Join(paths.InputsDir, "session.yml")
+	sessionChecksum, err := copyFileIfChanged(env.ArtifactStore, sessionSrc, sessionDst)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: materialize session.yml: %w", err)
+	}
+	registerInput("session_config", sessionDst, sessionChecksum)
+
+	pipelineResolvedBytes, err := renderResolvedPipeline(env.Config.Pipeline)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: render pipeline.resolved.yml: %w", err)
+	}
+	pipelineDst := filepath.Join(paths.InputsDir, "pipeline.resolved.yml")
+	pipelineChecksum, err := writeBytesIfChanged(env.ArtifactStore, pipelineDst, pipelineResolvedBytes)
+	if err != nil {
+		return nil, fmt.Errorf("prepare: materialize pipeline.resolved.yml: %w", err)
+	}
+	registerInput("pipeline_resolved", pipelineDst, pipelineChecksum)
+
+	for _, cfgFile := range []struct {
+		kind string
+		src  string
+		dst  string
+	}{
+		{kind: "speakers", src: speakersSrc, dst: filepath.Join(paths.InputsDir, "speakers.yml")},
+		{kind: "autocorrect", src: autocorrectSrc, dst: filepath.Join(paths.InputsDir, "autocorrect.yml")},
+		{kind: "glossary", src: glossarySrc, dst: filepath.Join(paths.InputsDir, "glossary.yml")},
+	} {
+		checksum, err := copyFileIfChanged(env.ArtifactStore, cfgFile.src, cfgFile.dst)
+		if err != nil {
+			return nil, fmt.Errorf("prepare: materialize %s: %w", cfgFile.kind, err)
+		}
+		registerInput(cfgFile.kind, cfgFile.dst, checksum)
+	}
+
+	for _, src := range resolvedAudio {
+		base := filepath.Base(src)
+		if prev, exists := copiedByDest[base]; exists && prev != src {
+			return nil, fmt.Errorf("prepare: duplicate audio basename %q from %q and %q", base, prev, src)
+		}
+		copiedByDest[base] = src
+
+		dst := filepath.Join(paths.AudioDir, base)
+		checksum, err := copyFileIfChanged(env.ArtifactStore, src, dst)
+		if err != nil {
+			return nil, fmt.Errorf("prepare: materialize audio %q: %w", base, err)
+		}
+		registerInput("audio", dst, checksum)
+	}
+
+	sort.Slice(inputs, func(i, j int) bool {
+		if inputs[i].Kind != inputs[j].Kind {
+			return inputs[i].Kind < inputs[j].Kind
+		}
+		return inputs[i].Path < inputs[j].Path
+	})
+	m.Inputs = inputs
+
+	return &StageResult{
+		Metadata: map[string]any{
+			"prepared":             true,
+			"stage":                "prepare",
+			"inputs_count":         len(inputs),
+			"audio_files_resolved": len(resolvedAudio),
+		},
+	}, nil
+}
+
+func renderResolvedPipeline(cfg *config.PipelineConfig) ([]byte, error) {
+	return yaml.Marshal(cfg)
+}
+
+func resolveAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) {
+	if len(inputs.AudioFiles) > 0 {
+		out := make([]string, 0, len(inputs.AudioFiles))
+		for _, p := range inputs.AudioFiles {
+			resolved, err := resolvePath(sessionDir, p)
+			if err != nil {
+				return nil, err
+			}
+			if !isFlac(resolved) {
+				return nil, fmt.Errorf("audio file %q must have .flac extension", resolved)
+			}
+			if err := requireFile(resolved, "audio file"); err != nil {
+				return nil, err
+			}
+			out = append(out, resolved)
+		}
+		sort.Strings(out)
+		return out, nil
+	}
+
+	audioDir, err := resolvePath(sessionDir, inputs.AudioDir)
+	if err != nil {
+		return nil, err
+	}
+	entries, err := os.ReadDir(audioDir)
+	if err != nil {
+		return nil, fmt.Errorf("read audio directory %q: %w", audioDir, err)
+	}
+
+	out := make([]string, 0)
+	for _, entry := range entries {
+		if entry.IsDir() {
+			continue
+		}
+		name := entry.Name()
+		full := filepath.Join(audioDir, name)
+		if !isFlac(full) {
+			continue
+		}
+		if err := requireFile(full, "audio file"); err != nil {
+			return nil, err
+		}
+		out = append(out, full)
+	}
+
+	if len(out) == 0 {
+		return nil, fmt.Errorf("no .flac files found in audio directory %q", audioDir)
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+func resolvePath(baseDir, p string) (string, error) {
+	trimmed := strings.TrimSpace(p)
+	if trimmed == "" {
+		return "", fmt.Errorf("path is required")
+	}
+	if filepath.IsAbs(trimmed) {
+		return filepath.Clean(trimmed), nil
+	}
+	return filepath.Clean(filepath.Join(baseDir, trimmed)), nil
+}
+
+func requireFile(path string, label string) error {
+	if strings.TrimSpace(path) == "" {
+		return fmt.Errorf("%s path is required", label)
+	}
+	info, err := os.Stat(path)
+	if err != nil {
+		return fmt.Errorf("%s %q not found: %w", label, path, err)
+	}
+	if info.IsDir() {
+		return fmt.Errorf("%s %q is a directory", label, path)
+	}
+	return nil
+}
+
+func isFlac(path string) bool {
+	return strings.EqualFold(filepath.Ext(path), ".flac")
+}
+
+func copyFileIfChanged(store artifacts.Store, src, dst string) (string, error) {
+	srcChecksum, err := store.Checksum(src)
+	if err != nil {
+		return "", err
+	}
+
+	exists, err := store.Exists(dst)
+	if err != nil {
+		return "", err
+	}
+	if exists {
+		dstChecksum, err := store.Checksum(dst)
+		if err != nil {
+			return "", err
+		}
+		if dstChecksum == srcChecksum {
+			return srcChecksum, nil
+		}
+	}
+
+	data, err := os.ReadFile(src)
+	if err != nil {
+		return "", err
+	}
+	if err := store.WriteFileAtomic(dst, data, 0o644); err != nil {
+		return "", err
+	}
+	return srcChecksum, nil
+}
+
+func writeBytesIfChanged(store artifacts.Store, dst string, data []byte) (string, error) {
+	digest := sha256.Sum256(data)
+	targetChecksum := hex.EncodeToString(digest[:])
+
+	exists, err := store.Exists(dst)
+	if err != nil {
+		return "", err
+	}
+	if exists {
+		dstChecksum, err := store.Checksum(dst)
+		if err != nil {
+			return "", err
+		}
+		if dstChecksum == targetChecksum {
+			return targetChecksum, nil
+		}
+	}
+
+	if err := store.WriteFileAtomic(dst, data, 0o644); err != nil {
+		return "", err
+	}
+	return targetChecksum, nil
+}
--- a/internal/stage/prepare_test.go
+++ b/internal/stage/prepare_test.go
@@ -0,0 +1,204 @@
+package stage
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+func TestPrepareStageExplicitAudioFiles(t *testing.T) {
+	env, m := setupPrepareEnv(t)
+
+	root := filepath.Dir(env.Config.SessionPath)
+	audioA := filepath.Join(root, "audio", "alice.flac")
+	audioB := filepath.Join(root, "audio", "bob.flac")
+	writeFile(t, audioA, "a")
+	writeFile(t, audioB, "b")
+	env.Config.Session.Inputs.AudioFiles = []string{"./audio/alice.flac", "./audio/bob.flac"}
+
+	s := prepareStage{}
+	result, err := s.Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("prepare.Run() error = %v", err)
+	}
+	if result == nil || result.Metadata["prepared"] != true {
+		t.Fatalf("result metadata = %#v, want prepared=true", result)
+	}
+
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	for _, p := range []string{
+		filepath.Join(paths.InputsDir, "session.yml"),
+		filepath.Join(paths.InputsDir, "pipeline.resolved.yml"),
+		filepath.Join(paths.InputsDir, "speakers.yml"),
+		filepath.Join(paths.InputsDir, "autocorrect.yml"),
+		filepath.Join(paths.InputsDir, "glossary.yml"),
+		filepath.Join(paths.AudioDir, "alice.flac"),
+		filepath.Join(paths.AudioDir, "bob.flac"),
+	} {
+		if _, err := os.Stat(p); err != nil {
+			t.Fatalf("expected materialized file %q: %v", p, err)
+		}
+	}
+
+	if len(m.Inputs) != 7 {
+		t.Fatalf("manifest inputs len = %d, want 7", len(m.Inputs))
+	}
+	for _, in := range m.Inputs {
+		if in.Checksum == "" {
+			t.Fatalf("input %#v missing checksum", in)
+		}
+	}
+}
+
+func TestPrepareStageAudioDirEnumeration(t *testing.T) {
+	env, m := setupPrepareEnv(t)
+
+	root := filepath.Dir(env.Config.SessionPath)
+	writeFile(t, filepath.Join(root, "audio", "a.flac"), "a")
+	writeFile(t, filepath.Join(root, "audio", "ignore.txt"), "x")
+	env.Config.Session.Inputs.AudioFiles = nil
+	env.Config.Session.Inputs.AudioDir = "./audio"
+
+	s := prepareStage{}
+	result, err := s.Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("prepare.Run() error = %v", err)
+	}
+	if got := result.Metadata["audio_files_resolved"]; got != 1 {
+		t.Fatalf("audio_files_resolved = %#v, want 1", got)
+	}
+
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	if _, err := os.Stat(filepath.Join(paths.AudioDir, "a.flac")); err != nil {
+		t.Fatalf("expected copied flac: %v", err)
+	}
+	if _, err := os.Stat(filepath.Join(paths.AudioDir, "ignore.txt")); err == nil {
+		t.Fatalf("unexpected non-flac copied")
+	}
+}
+
+func TestPrepareStageValidationFailures(t *testing.T) {
+	t.Run("missing speakers", func(t *testing.T) {
+		env, m := setupPrepareEnv(t)
+		root := filepath.Dir(env.Config.SessionPath)
+		writeFile(t, filepath.Join(root, "audio", "a.flac"), "a")
+		_ = os.Remove(filepath.Join(root, "speakers.yml"))
+
+		_, err := (prepareStage{}).Run(context.Background(), env, m)
+		if err == nil || !strings.Contains(err.Error(), "speakers") {
+			t.Fatalf("error = %v, want speakers error", err)
+		}
+	})
+
+	t.Run("missing audio file", func(t *testing.T) {
+		env, m := setupPrepareEnv(t)
+		env.Config.Session.Inputs.AudioFiles = []string{"./audio/missing.flac"}
+
+		_, err := (prepareStage{}).Run(context.Background(), env, m)
+		if err == nil || !strings.Contains(err.Error(), "audio") {
+			t.Fatalf("error = %v, want audio error", err)
+		}
+	})
+
+	t.Run("invalid audio extension", func(t *testing.T) {
+		env, m := setupPrepareEnv(t)
+		root := filepath.Dir(env.Config.SessionPath)
+		writeFile(t, filepath.Join(root, "audio", "a.mp3"), "x")
+		env.Config.Session.Inputs.AudioFiles = []string{"./audio/a.mp3"}
+
+		_, err := (prepareStage{}).Run(context.Background(), env, m)
+		if err == nil || !strings.Contains(strings.ToLower(err.Error()), ".flac") {
+			t.Fatalf("error = %v, want .flac error", err)
+		}
+	})
+}
+
+func TestPrepareStageIdempotent(t *testing.T) {
+	env, m := setupPrepareEnv(t)
+	root := filepath.Dir(env.Config.SessionPath)
+	writeFile(t, filepath.Join(root, "audio", "a.flac"), "a")
+	env.Config.Session.Inputs.AudioFiles = []string{"./audio/a.flac"}
+
+	s := prepareStage{}
+	_, err := s.Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("first run error = %v", err)
+	}
+	first := snapshotInputs(m.Inputs)
+
+	time.Sleep(5 * time.Millisecond)
+	_, err = s.Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("second run error = %v", err)
+	}
+	second := snapshotInputs(m.Inputs)
+
+	if len(first) != len(second) {
+		t.Fatalf("input lengths differ: %d vs %d", len(first), len(second))
+	}
+	for k, v := range first {
+		if second[k] != v {
+			t.Fatalf("checksum for %q changed: %q vs %q", k, v, second[k])
+		}
+	}
+}
+
+func setupPrepareEnv(t *testing.T) (*Env, *manifest.Manifest) {
+	t.Helper()
+	workspace := t.TempDir()
+	cfgDir := t.TempDir()
+
+	sessionPath := filepath.Join(cfgDir, "session.yml")
+	pipelinePath := filepath.Join(cfgDir, "pipeline.yml")
+
+	writeFile(t, pipelinePath, "workspace:\n  root: "+workspace+"\n")
+	writeFile(t, sessionPath, "session_id: 2026-05-03\n")
+	writeFile(t, filepath.Join(cfgDir, "speakers.yml"), "alice: alice.flac\n")
+	writeFile(t, filepath.Join(cfgDir, "autocorrect.yml"), "[]\n")
+	writeFile(t, filepath.Join(cfgDir, "glossary.yml"), "[]\n")
+
+	cfg := &config.Config{
+		Pipeline:     &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: workspace}},
+		SessionPath:  sessionPath,
+		PipelinePath: pipelinePath,
+		Session: &config.SessionConfig{
+			SessionID: "2026-05-03",
+			Inputs: config.SessionInputsConfig{
+				AudioDir:        "./audio",
+				SpeakersFile:    "./speakers.yml",
+				AutocorrectFile: "./autocorrect.yml",
+				GlossaryFile:    "./glossary.yml",
+			},
+		},
+	}
+
+	store := artifacts.NewLocalStore(workspace)
+	env := &Env{Config: cfg, ArtifactStore: store}
+	m := manifest.New("2026-05-03", time.Now().UTC())
+	return env, m
+}
+
+func writeFile(t *testing.T, path, contents string) {
+	t.Helper()
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		t.Fatalf("MkdirAll(%q): %v", path, err)
+	}
+	if err := os.WriteFile(path, []byte(contents), 0o644); err != nil {
+		t.Fatalf("WriteFile(%q): %v", path, err)
+	}
+}
+
+func snapshotInputs(inputs []manifest.InputRecord) map[string]string {
+	out := make(map[string]string, len(inputs))
+	for _, in := range inputs {
+		out[in.Kind+"|"+in.Path] = in.Checksum
+	}
+	return out
+}