From 2438a0b7a52881f5e2a82d6f1d94204e40d3b7d0 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 3 May 2026 15:42:44 -0500 Subject: [PATCH] Implement real transcribe stage --- internal/adapters/whisperx/fake.go | 34 +- .../whisperx/transcripts/raw/alice.json | 1 + internal/app/runner_test.go | 35 ++ internal/stage/placeholders.go | 21 +- internal/stage/placeholders_test.go | 12 +- internal/stage/transcribe.go | 316 ++++++++++++++++++ internal/stage/transcribe_test.go | 267 +++++++++++++++ 7 files changed, 660 insertions(+), 26 deletions(-) create mode 100644 internal/adapters/whisperx/transcripts/raw/alice.json create mode 100644 internal/stage/transcribe.go create mode 100644 internal/stage/transcribe_test.go diff --git a/internal/adapters/whisperx/fake.go b/internal/adapters/whisperx/fake.go index bd8c2bf..509b967 100644 --- a/internal/adapters/whisperx/fake.go +++ b/internal/adapters/whisperx/fake.go @@ -1,6 +1,12 @@ package whisperx -import "context" +import ( + "context" + "os" + "path/filepath" +) + +var minimalTranscriptJSON = []byte(`{"schema":"speaker_transcript.v1","segments":[]}`) // NoopClient is a deterministic no-op WhisperX adapter. type NoopClient struct{} @@ -10,6 +16,9 @@ func (n *NoopClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra if err := ctx.Err(); err != nil { return TranscribeResult{}, err } + if err := writeMinimalJSON(req.OutputRawTranscriptPath); err != nil { + return TranscribeResult{}, err + } return TranscribeResult{ OutputRawTranscriptPath: req.OutputRawTranscriptPath, Attempts: 1, @@ -22,9 +31,10 @@ func (n *NoopClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra // FakeClient captures requests and returns deterministic responses for tests. type FakeClient struct { - Requests []TranscribeRequest - Err error - Result TranscribeResult + Requests []TranscribeRequest + Err error + Result TranscribeResult + TranscribeFn func(ctx context.Context, req TranscribeRequest) (TranscribeResult, error) } // Transcribe records the request and returns either configured error or result. @@ -33,6 +43,9 @@ func (f *FakeClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra return TranscribeResult{}, err } f.Requests = append(f.Requests, req) + if f.TranscribeFn != nil { + return f.TranscribeFn(ctx, req) + } if f.Err != nil { return TranscribeResult{}, f.Err } @@ -46,5 +59,18 @@ func (f *FakeClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra if res.Metadata == nil { res.Metadata = map[string]any{"fake": true} } + if err := writeMinimalJSON(res.OutputRawTranscriptPath); err != nil { + return TranscribeResult{}, err + } return res, nil } + +func writeMinimalJSON(path string) error { + if path == "" { + return nil + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + return os.WriteFile(path, minimalTranscriptJSON, 0o644) +} diff --git a/internal/adapters/whisperx/transcripts/raw/alice.json b/internal/adapters/whisperx/transcripts/raw/alice.json new file mode 100644 index 0000000..18be13c --- /dev/null +++ b/internal/adapters/whisperx/transcripts/raw/alice.json @@ -0,0 +1 @@ +{"schema":"speaker_transcript.v1","segments":[]} \ No newline at end of file diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index fffefe7..94631f2 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -75,6 +75,15 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { } continue } + if name == "transcribe" { + if sr.Metadata == nil || sr.Metadata["stage"] != "transcribe" { + t.Fatalf("transcribe metadata missing stage=transcribe: %#v", sr.Metadata) + } + if len(sr.Outputs) == 0 { + t.Fatalf("transcribe outputs missing") + } + continue + } if sr.Metadata == nil || sr.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", name) } @@ -202,6 +211,17 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) { existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) existing.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) + audioPath := filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.SessionID, "audio", "alice.flac") + if err := os.MkdirAll(filepath.Dir(audioPath), 0o755); err != nil { + t.Fatalf("MkdirAll() error = %v", err) + } + if err := os.WriteFile(audioPath, []byte("audio"), 0o644); err != nil { + t.Fatalf("WriteFile() error = %v", err) + } + existing.Inputs = append(existing.Inputs, manifest.InputRecord{ + Kind: "audio", + Path: audioPath, + }) if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } @@ -251,6 +271,21 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { tc.env.Config = cfg tc.env.ArtifactStore = artifactStore tc.env.ManifestStore = &manifest.LocalStore{} + if tc.name == "transcribe" { + paths, ensureErr := artifactStore.EnsureLayout(cfg.Session.SessionID) + if ensureErr != nil { + t.Fatalf("EnsureLayout() error = %v", ensureErr) + } + audioPath := filepath.Join(paths.AudioDir, "alice.flac") + if err := os.WriteFile(audioPath, []byte("audio"), 0o644); err != nil { + t.Fatalf("write transcribe fixture audio: %v", err) + } + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Inputs = append(m.Inputs, manifest.InputRecord{Kind: "audio", Path: audioPath}) + if err := tc.env.ManifestStore.Save(context.Background(), manifestPathFor(cfg), m); err != nil { + t.Fatalf("seed manifest: %v", err) + } + } _, runErr := executeStages(context.Background(), cfg, []stage.Stage{selected}, RunOptions{Env: tc.env}) if runErr == nil { diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index b27dc6d..6aa4ab9 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -10,7 +10,6 @@ import ( "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/manifest" ) @@ -54,24 +53,6 @@ func (s placeholderStage) Run(ctx context.Context, env *Env, m *manifest.Manifes paths := env.ArtifactStore.SessionPaths(sessionID) switch s.name { - case "transcribe": - if env.WhisperX != nil { - req := whisperx.TranscribeRequest{ - SpeakerID: "placeholder-speaker", - AudioPath: filepath.Join(paths.AudioDir, "placeholder-speaker.flac"), - OutputRawTranscriptPath: filepath.Join(paths.TranscriptsRawDir, "placeholder-speaker.json"), - } - resp, err := env.WhisperX.Transcribe(ctx, req) - if err != nil { - return nil, fmt.Errorf("placeholder transcribe adapter call failed: %w", err) - } - result.Outputs = append(result.Outputs, artifacts.Ref{ - Kind: "transcript_raw", - Category: "transcripts", - SessionID: sessionID, - AbsolutePath: resp.OutputRawTranscriptPath, - }) - } case "merge": if env.Seriatim != nil { req := seriatim.MergeRequest{ @@ -164,7 +145,7 @@ func (s placeholderStage) Run(ctx context.Context, env *Env, m *manifest.Manifes func All() []Stage { return []Stage{ prepareStage{}, - placeholderStage{name: "transcribe"}, + transcribeStage{}, placeholderStage{name: "normalize"}, placeholderStage{name: "merge"}, placeholderStage{name: "polish"}, diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 9c01a1b..90a7976 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -36,7 +36,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { 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") + writeStageTestFile(t, filepath.Join(cfgDir, "audio", "alice.flac"), "a") wf := &whisperx.FakeClient{} sf := &seriatim.FakeRunner{} @@ -84,6 +84,15 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } continue } + if s.Name() == "transcribe" { + if result.Metadata["stage"] != "transcribe" { + t.Fatalf("transcribe metadata = %#v, want stage=transcribe", result.Metadata) + } + if len(result.Outputs) == 0 { + t.Fatalf("transcribe outputs = %#v, want non-empty", result.Outputs) + } + continue + } if result.Metadata["placeholder"] != true { t.Fatalf("stage %q missing placeholder metadata", s.Name()) } @@ -115,7 +124,6 @@ func TestPlaceholderAdapterErrorPropagation(t *testing.T) { env *Env wantErr string }{ - {stageName: "transcribe", env: &Env{WhisperX: &whisperx.FakeClient{Err: errors.New("werr")}}, wantErr: "transcribe"}, {stageName: "merge", env: &Env{Seriatim: &seriatim.FakeRunner{Err: errors.New("serr")}}, wantErr: "merge"}, {stageName: "polish", env: &Env{Audita: &audita.FakeRunner{Err: errors.New("aerr")}}, wantErr: "polish"}, {stageName: "analyze", env: &Env{Analyzer: &analyzer.FakeRunner{Err: errors.New("anerr")}}, wantErr: "analyze"}, diff --git a/internal/stage/transcribe.go b/internal/stage/transcribe.go new file mode 100644 index 0000000..bb1589f --- /dev/null +++ b/internal/stage/transcribe.go @@ -0,0 +1,316 @@ +package stage + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +type transcribeStage struct{} + +func (transcribeStage) Name() string { return "transcribe" } + +func (transcribeStage) Declares() IODecl { + return IODecl{ + Inputs: []artifacts.Ref{ + {Kind: "audio", Category: "audio", RelativePath: "audio/*.flac"}, + }, + Outputs: []artifacts.Ref{ + {Kind: "transcript_raw", Category: "transcripts", RelativePath: "transcripts/raw/*.json"}, + }, + } +} + +func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { + if env == nil || env.Config == nil { + return nil, fmt.Errorf("transcribe: stage environment config is required") + } + if env.ArtifactStore == nil { + return nil, fmt.Errorf("transcribe: artifact store is required") + } + if env.Config.Pipeline == nil || env.Config.Session == nil { + return nil, fmt.Errorf("transcribe: resolved config must include pipeline and session") + } + if env.WhisperX == nil { + return nil, fmt.Errorf("transcribe: whisperx adapter is required") + } + + var sessionID string + if m != nil { + sessionID = strings.TrimSpace(m.SessionID) + } + if sessionID == "" { + sessionID = strings.TrimSpace(env.Config.Session.SessionID) + } + if sessionID == "" { + return nil, fmt.Errorf("transcribe: session id is required") + } + + paths := env.ArtifactStore.SessionPaths(sessionID) + audioFiles, err := discoverPreparedAudio(m, paths.AudioDir) + if err != nil { + return nil, fmt.Errorf("transcribe: resolve audio inputs: %w", err) + } + if len(audioFiles) == 0 { + return nil, fmt.Errorf("transcribe: no prepared audio files found") + } + + concurrency := 1 + if env.Config.Pipeline.WhisperX.Concurrency != nil && *env.Config.Pipeline.WhisperX.Concurrency > 0 { + concurrency = *env.Config.Pipeline.WhisperX.Concurrency + } + if concurrency > len(audioFiles) { + concurrency = len(audioFiles) + } + + type job struct { + speakerID string + audioPath string + outPath string + } + jobs := make([]job, 0, len(audioFiles)) + seenSpeaker := map[string]string{} + for _, audioPath := range audioFiles { + base := strings.TrimSuffix(filepath.Base(audioPath), filepath.Ext(audioPath)) + if base == "" { + return nil, fmt.Errorf("transcribe: could not derive speaker id from %q", audioPath) + } + if prev, ok := seenSpeaker[base]; ok { + return nil, fmt.Errorf("transcribe: duplicate speaker/audio basename %q from %q and %q", base, prev, audioPath) + } + seenSpeaker[base] = audioPath + jobs = append(jobs, job{ + speakerID: base, + audioPath: audioPath, + outPath: filepath.Join(paths.TranscriptsRawDir, base+".json"), + }) + } + + stageCtx, cancel := context.WithCancel(ctx) + defer cancel() + + var ( + wg sync.WaitGroup + jobCh = make(chan job) + mu sync.Mutex + firstErr error + perFile = map[string]map[string]any{} + outputRef = map[string]artifacts.Ref{} + ) + + worker := func() { + defer wg.Done() + for j := range jobCh { + if stageCtx.Err() != nil { + return + } + + res, runErr := env.WhisperX.Transcribe(stageCtx, whisperx.TranscribeRequest{ + SpeakerID: j.speakerID, + AudioPath: j.audioPath, + OutputRawTranscriptPath: j.outPath, + }) + if runErr != nil { + mu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("speaker %q (%s): %w", j.speakerID, filepath.Base(j.audioPath), runErr) + cancel() + } + mu.Unlock() + continue + } + + if strings.TrimSpace(res.OutputRawTranscriptPath) == "" { + res.OutputRawTranscriptPath = j.outPath + } + if filepath.Clean(res.OutputRawTranscriptPath) != filepath.Clean(j.outPath) { + mu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("speaker %q: adapter output path %q did not match expected %q", j.speakerID, res.OutputRawTranscriptPath, j.outPath) + cancel() + } + mu.Unlock() + continue + } + if err := validateTranscriptJSONFile(j.outPath); err != nil { + mu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("speaker %q output %q invalid: %w", j.speakerID, j.outPath, err) + cancel() + } + mu.Unlock() + continue + } + + meta := map[string]any{ + "attempts": res.Attempts, + "http_status": res.HTTPStatus, + "duration_ms": res.Duration.Milliseconds(), + "output_path": j.outPath, + } + + mu.Lock() + perFile[j.speakerID] = meta + outputRef[j.speakerID] = artifacts.Ref{ + Kind: "transcript_raw", + Category: "transcripts", + SessionID: sessionID, + AbsolutePath: j.outPath, + } + mu.Unlock() + } + } + + wg.Add(concurrency) + for i := 0; i < concurrency; i++ { + go worker() + } + +dispatch: + for _, j := range jobs { + select { + case <-stageCtx.Done(): + break dispatch + case jobCh <- j: + } + } + close(jobCh) + wg.Wait() + + if firstErr != nil { + return nil, firstErr + } + + speakers := make([]string, 0, len(perFile)) + for speaker := range perFile { + speakers = append(speakers, speaker) + } + sort.Strings(speakers) + + outputs := make([]artifacts.Ref, 0, len(speakers)) + outputPaths := make([]string, 0, len(speakers)) + orderedPerFile := make(map[string]any, len(speakers)) + for _, speaker := range speakers { + ref := outputRef[speaker] + outputs = append(outputs, ref) + outputPaths = append(outputPaths, ref.AbsolutePath) + orderedPerFile[speaker] = perFile[speaker] + } + + retries := 0 + if env.Config.Pipeline.WhisperX.Retries != nil { + retries = *env.Config.Pipeline.WhisperX.Retries + } + + return &StageResult{ + Outputs: outputs, + Metadata: map[string]any{ + "stage": "transcribe", + "audio_files_count": len(audioFiles), + "language": env.Config.Pipeline.WhisperX.Language, + "concurrency": concurrency, + "retries": retries, + "retry_delay": env.Config.Pipeline.WhisperX.RetryDelay, + "timeout": env.Config.Pipeline.WhisperX.Timeout, + "output_paths": outputPaths, + "per_file": orderedPerFile, + }, + }, nil +} + +func discoverPreparedAudio(m *manifest.Manifest, audioDir string) ([]string, error) { + fromManifest := make([]string, 0) + if m != nil { + for _, in := range m.Inputs { + if in.Kind != "audio" { + continue + } + p := strings.TrimSpace(in.Path) + if p == "" { + continue + } + fromManifest = append(fromManifest, filepath.Clean(p)) + } + } + if len(fromManifest) > 0 { + sort.Strings(fromManifest) + if err := validateAudioFiles(fromManifest); err != nil { + return nil, err + } + return fromManifest, nil + } + + 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 + } + p := filepath.Join(audioDir, entry.Name()) + if !isFlac(p) { + continue + } + out = append(out, p) + } + sort.Strings(out) + if err := validateAudioFiles(out); err != nil { + return nil, err + } + return out, nil +} + +func validateAudioFiles(paths []string) error { + if len(paths) == 0 { + return nil + } + seen := map[string]struct{}{} + for _, p := range paths { + clean := filepath.Clean(strings.TrimSpace(p)) + if clean == "" { + return fmt.Errorf("audio path is required") + } + if !isFlac(clean) { + return fmt.Errorf("audio file %q must have .flac extension", clean) + } + info, err := os.Stat(clean) + if err != nil { + return fmt.Errorf("audio file %q not found: %w", clean, err) + } + if !info.Mode().IsRegular() { + return fmt.Errorf("audio file %q is not a regular file", clean) + } + if _, exists := seen[clean]; exists { + return fmt.Errorf("duplicate audio file path %q", clean) + } + seen[clean] = struct{}{} + } + return nil +} + +func validateTranscriptJSONFile(path string) error { + if err := requireFile(path, "raw transcript"); err != nil { + return err + } + data, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read transcript: %w", err) + } + var payload any + if err := json.Unmarshal(data, &payload); err != nil { + return fmt.Errorf("decode json: %w", err) + } + return nil +} diff --git a/internal/stage/transcribe_test.go b/internal/stage/transcribe_test.go new file mode 100644 index 0000000..0796dd1 --- /dev/null +++ b/internal/stage/transcribe_test.go @@ -0,0 +1,267 @@ +package stage + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "sort" + "strings" + "sync/atomic" + "testing" + "time" + + "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" +) + +func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac"}) + env.WhisperX = &whisperx.FakeClient{ + Result: whisperx.TranscribeResult{ + Attempts: 2, + HTTPStatus: 200, + Duration: 1200 * time.Millisecond, + }, + } + + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + + result, err := (transcribeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("transcribe.Run() error = %v", err) + } + if result == nil { + t.Fatal("transcribe result is nil") + } + if len(result.Outputs) != 2 { + t.Fatalf("outputs len = %d, want 2", len(result.Outputs)) + } + + gotPaths := make([]string, 0, len(result.Outputs)) + for _, out := range result.Outputs { + if out.Kind != "transcript_raw" { + t.Fatalf("output kind = %q, want transcript_raw", out.Kind) + } + gotPaths = append(gotPaths, out.AbsolutePath) + verifyJSONFile(t, out.AbsolutePath) + } + sort.Strings(gotPaths) + wantPaths := []string{ + filepath.Join(env.ArtifactStore.SessionPaths(m.SessionID).TranscriptsRawDir, "alice.json"), + filepath.Join(env.ArtifactStore.SessionPaths(m.SessionID).TranscriptsRawDir, "bob.json"), + } + sort.Strings(wantPaths) + if strings.Join(gotPaths, "|") != strings.Join(wantPaths, "|") { + t.Fatalf("output paths = %#v, want %#v", gotPaths, wantPaths) + } + + meta := result.Metadata + if meta == nil { + t.Fatal("metadata is nil") + } + if meta["stage"] != "transcribe" { + t.Fatalf("metadata stage = %#v, want transcribe", meta["stage"]) + } + if meta["audio_files_count"] != 2 { + t.Fatalf("metadata audio_files_count = %#v, want 2", meta["audio_files_count"]) + } + if meta["language"] != "en" { + t.Fatalf("metadata language = %#v, want en", meta["language"]) + } + if meta["concurrency"] != 2 { + t.Fatalf("metadata concurrency = %#v, want 2", meta["concurrency"]) + } + if meta["retries"] != 3 { + t.Fatalf("metadata retries = %#v, want 3", meta["retries"]) + } + perFile, ok := meta["per_file"].(map[string]any) + if !ok { + t.Fatalf("metadata per_file type = %T, want map[string]any", meta["per_file"]) + } + if len(perFile) != 2 { + t.Fatalf("metadata per_file len = %d, want 2", len(perFile)) + } +} + +func TestTranscribeStageConcurrencyBounded(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"a.flac", "b.flac", "c.flac", "d.flac", "e.flac"}) + var inFlight int32 + var maxInFlight int32 + + env.WhisperX = &whisperx.FakeClient{ + TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + n := atomic.AddInt32(&inFlight, 1) + for { + cur := atomic.LoadInt32(&maxInFlight) + if n <= cur || atomic.CompareAndSwapInt32(&maxInFlight, cur, n) { + break + } + } + defer atomic.AddInt32(&inFlight, -1) + + time.Sleep(25 * time.Millisecond) + if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil { + return whisperx.TranscribeResult{}, err + } + return whisperx.TranscribeResult{ + OutputRawTranscriptPath: req.OutputRawTranscriptPath, + Attempts: 1, + HTTPStatus: 200, + Duration: 25 * time.Millisecond, + }, nil + }, + } + + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + + if _, err := (transcribeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("transcribe.Run() error = %v", err) + } + + if got := atomic.LoadInt32(&maxInFlight); got > 2 { + t.Fatalf("max in-flight = %d, want <= 2", got) + } +} + +func TestTranscribeStageFailsWhenOneFileFails(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac"}) + env.WhisperX = &whisperx.FakeClient{ + TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + if req.SpeakerID == "bob" { + return whisperx.TranscribeResult{}, errors.New("upstream failed") + } + if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil { + return whisperx.TranscribeResult{}, err + } + return whisperx.TranscribeResult{ + OutputRawTranscriptPath: req.OutputRawTranscriptPath, + Attempts: 1, + HTTPStatus: 200, + }, nil + }, + } + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + + _, err := (transcribeStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected transcribe error, got nil") + } + if !strings.Contains(err.Error(), "bob") { + t.Fatalf("error = %q, want speaker context", err.Error()) + } +} + +func TestTranscribeStageInvalidJSONFails(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac"}) + env.WhisperX = &whisperx.FakeClient{ + TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + if err := os.MkdirAll(filepath.Dir(req.OutputRawTranscriptPath), 0o755); err != nil { + return whisperx.TranscribeResult{}, err + } + if err := os.WriteFile(req.OutputRawTranscriptPath, []byte("not-json"), 0o644); err != nil { + return whisperx.TranscribeResult{}, err + } + return whisperx.TranscribeResult{ + OutputRawTranscriptPath: req.OutputRawTranscriptPath, + Attempts: 1, + HTTPStatus: 200, + }, nil + }, + } + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + + _, err := (transcribeStage{}).Run(context.Background(), env, m) + if err == nil { + t.Fatal("expected transcribe error, got nil") + } + if !strings.Contains(strings.ToLower(err.Error()), "json") { + t.Fatalf("error = %q, want json validation context", err.Error()) + } +} + +func setupTranscribeEnv(t *testing.T, audioFiles []string) (*Env, *manifest.Manifest) { + t.Helper() + + workspace := t.TempDir() + cfgDir := t.TempDir() + audioDir := filepath.Join(cfgDir, "audio") + for _, name := range audioFiles { + writeFile(t, filepath.Join(audioDir, name), "audio-"+name) + } + + sessionPath := filepath.Join(cfgDir, "session.yml") + pipelinePath := filepath.Join(cfgDir, "pipeline.yml") + writeFile(t, sessionPath, "session_id: 2026-05-03\n") + writeFile(t, pipelinePath, "workspace:\n root: "+workspace+"\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") + + retries := 3 + concurrency := 2 + cfg := &config.Config{ + PipelinePath: pipelinePath, + SessionPath: sessionPath, + Pipeline: &config.PipelineConfig{ + Workspace: config.WorkspaceConfig{Root: workspace}, + WhisperX: config.WhisperXConfig{ + TranscribeURL: "https://transcription.ai.rakestrawhome.com/transcribe", + Language: "en", + Timeout: "30m", + Retries: &retries, + RetryDelay: "2s", + Concurrency: &concurrency, + }, + }, + Session: &config.SessionConfig{ + SessionID: "2026-05-03", + Inputs: config.SessionInputsConfig{ + AudioDir: "./audio", + SpeakersFile: "./speakers.yml", + AutocorrectFile: "./autocorrect.yml", + GlossaryFile: "./glossary.yml", + }, + }, + } + + env := &Env{ + Config: cfg, + ArtifactStore: artifacts.NewLocalStore(workspace), + } + return env, manifest.New("2026-05-03", time.Now().UTC()) +} + +func writeJSONFile(path string, payload any) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + data, err := json.Marshal(payload) + if err != nil { + return err + } + return os.WriteFile(path, data, 0o644) +} + +func verifyJSONFile(t *testing.T, path string) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %q: %v", path, err) + } + var v any + if err := json.Unmarshal(data, &v); err != nil { + t.Fatalf("json decode %q: %v", path, err) + } +}