Strengthen transcription concurrency tests

internal/stage/transcribe_test.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"os"
 	"path/filepath"
-	"sort"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -19,7 +20,7 @@ import (
 )
 
 func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) {
-	env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac"})
+	env, m := setupTranscribeEnv(t, []string{"bob.flac", "alice.flac"})
 	env.WhisperX = &whisperx.FakeClient{
 		Result: whisperx.TranscribeResult{
 			Attempts:   2,
@@ -51,13 +52,11 @@ func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) {
 		gotPaths = append(gotPaths, out.AbsolutePath)
 		verifyJSONFile(t, out.AbsolutePath)
 	}
-	sort.Strings(gotPaths)
 	wantPaths := []string{
 		filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json"),
 		filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "bob.json"),
 	}
-	sort.Strings(wantPaths)
+	if !slices.Equal(gotPaths, wantPaths) {
-	if strings.Join(gotPaths, "|") != strings.Join(wantPaths, "|") {
 		t.Fatalf("output paths = %#v, want %#v", gotPaths, wantPaths)
 	}
 
@@ -87,15 +86,25 @@ func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) {
 	if len(perFile) != 2 {
 		t.Fatalf("metadata per_file len = %d, want 2", len(perFile))
 	}
+	outputPaths, ok := meta["output_paths"].([]string)
+	if !ok || !slices.Equal(outputPaths, wantPaths) {
+		t.Fatalf("metadata output_paths = %#v, want %#v", meta["output_paths"], wantPaths)
+	}
+	runOutputPaths, ok := meta["run_output_paths"].([]string)
+	if !ok || len(runOutputPaths) != 2 || filepath.Base(runOutputPaths[0]) != "alice.json" || filepath.Base(runOutputPaths[1]) != "bob.json" {
+		t.Fatalf("metadata run_output_paths = %#v, want stable alice/bob order", meta["run_output_paths"])
+	}
 }
 
 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
+	started := make(chan struct{}, 2)
+	release := make(chan struct{})
 
 	env.WhisperX = &whisperx.FakeClient{
-		TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
+		TranscribeFn: func(ctx context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
 			n := atomic.AddInt32(&inFlight, 1)
 			for {
 				cur := atomic.LoadInt32(&maxInFlight)
@@ -104,8 +113,15 @@ func TestTranscribeStageConcurrencyBounded(t *testing.T) {
 				}
 			}
 			defer atomic.AddInt32(&inFlight, -1)
-
+			select {
-			time.Sleep(25 * time.Millisecond)
+			case started <- struct{}{}:
+			default:
+			}
+			select {
+			case <-release:
+			case <-ctx.Done():
+				return whisperx.TranscribeResult{}, ctx.Err()
+			}
 			if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil {
 				return whisperx.TranscribeResult{}, err
 			}
@@ -122,42 +138,81 @@ func TestTranscribeStageConcurrencyBounded(t *testing.T) {
 		t.Fatalf("prepare.Run() error = %v", err)
 	}
 
-	if _, err := (transcribeStage{}).Run(context.Background(), env, m); err != nil {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() {
+		_, err := (transcribeStage{}).Run(ctx, env, m)
+		done <- err
+	}()
+	for range 2 {
+		select {
+		case <-started:
+		case <-time.After(time.Second):
+			t.Fatal("configured concurrent transcription work did not start")
+		}
+	}
+	if got := atomic.LoadInt32(&maxInFlight); got != 2 {
+		t.Fatalf("max in-flight = %d, want configured limit 2", got)
+	}
+	close(release)
+	select {
+	case err := <-done:
+		if err != nil {
 			t.Fatalf("transcribe.Run() error = %v", err)
 		}
-
+	case <-time.After(2 * time.Second):
-	if got := atomic.LoadInt32(&maxInFlight); got > 2 {
+		t.Fatal("transcribe.Run() did not complete after releasing work")
-		t.Fatalf("max in-flight = %d, want <= 2", got)
 	}
 }
 
-func TestTranscribeStageFailsWhenOneFileFails(t *testing.T) {
+func TestTranscribeStageFailureCancelsRemainingWork(t *testing.T) {
-	env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac"})
+	env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac", "cara.flac", "dan.flac"})
+	aliceStarted := make(chan struct{})
+	upstreamErr := errors.New("upstream failed")
 	env.WhisperX = &whisperx.FakeClient{
-		TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
+		TranscribeFn: func(ctx context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
-			if req.SpeakerID == "bob" {
+			switch req.SpeakerID {
-				return whisperx.TranscribeResult{}, errors.New("upstream failed")
+			case "alice":
+				close(aliceStarted)
+				<-ctx.Done()
+				return whisperx.TranscribeResult{}, ctx.Err()
+			case "bob":
+				<-aliceStarted
+				return whisperx.TranscribeResult{}, upstreamErr
+			default:
+				return whisperx.TranscribeResult{}, fmt.Errorf("unexpected adapter call for %s", req.SpeakerID)
 			}
-			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)
+	ctx, cancel := context.WithCancel(context.Background())
-	if err == nil {
+	defer cancel()
-		t.Fatal("expected transcribe error, got nil")
+	done := make(chan error, 1)
+	go func() {
+		_, err := (transcribeStage{}).Run(ctx, env, m)
+		done <- err
+	}()
+	var err error
+	select {
+	case err = <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("transcribe.Run() did not return after a speaker failure")
 	}
-	if !strings.Contains(err.Error(), "bob") {
+	if !errors.Is(err, upstreamErr) || !strings.Contains(err.Error(), "bob") {
-		t.Fatalf("error = %q, want speaker context", err.Error())
+		t.Fatalf("error = %v, want wrapped upstream failure with bob speaker context", err)
+	}
+	requests := env.WhisperX.(*whisperx.FakeClient).RequestsSnapshot()
+	requestedSpeakers := make([]string, 0, len(requests))
+	for _, req := range requests {
+		requestedSpeakers = append(requestedSpeakers, req.SpeakerID)
+	}
+	slices.Sort(requestedSpeakers)
+	if !slices.Equal(requestedSpeakers, []string{"alice", "bob"}) {
+		t.Fatalf("adapter speakers = %#v, want only in-flight alice and failing bob", requestedSpeakers)
 	}
 }
 
@@ -186,8 +241,25 @@ func TestTranscribeStageInvalidJSONFails(t *testing.T) {
 	if err == nil {
 		t.Fatal("expected transcribe error, got nil")
 	}
-	if !strings.Contains(strings.ToLower(err.Error()), "json") {
+	if !strings.Contains(err.Error(), "alice") || !strings.Contains(strings.ToLower(err.Error()), "json") {
-		t.Fatalf("error = %q, want json validation context", err.Error())
+		t.Fatalf("error = %q, want speaker-qualified json validation context", err.Error())
+	}
+}
+
+func TestTranscribeStageRejectsAdapterOutputPathWithSpeakerContext(t *testing.T) {
+	env, m := setupTranscribeEnv(t, []string{"alice.flac"})
+	env.WhisperX = &whisperx.FakeClient{
+		TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
+			return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath + ".other"}, 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 || !strings.Contains(err.Error(), "alice") || !strings.Contains(err.Error(), "did not match requested path") {
+		t.Fatalf("transcribe.Run() error = %v, want speaker-qualified output-path mismatch", err)
 	}
 }
 
@@ -284,8 +356,12 @@ func TestTranscribeStageCancellationDoesNotMaterializePartialOutputs(t *testing.
 		cancel()
 		t.Fatal("no transcription completed before cancellation")
 	}
-	if err := <-done; err == nil || !errors.Is(err, context.Canceled) {
+	runErr := <-done
-		t.Fatalf("transcribe.Run() error = %v, want context cancellation", err)
+	if runErr == nil || !errors.Is(runErr, context.Canceled) {
+		t.Fatalf("transcribe.Run() error = %v, want context cancellation", runErr)
+	}
+	if !strings.Contains(runErr.Error(), "1 of 3") {
+		t.Fatalf("transcribe.Run() error = %v, want completed/planned context", runErr)
 	}
 
 	canonical := filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json")