Stream WhisperX uploads safely

2026-08-10 21:49:44 +00:00
parent 32653f54f9
commit 9da2c1e144
12 changed files with 542 additions and 56 deletions
--- a/docs/config.md
+++ b/docs/config.md
@@ -178,7 +178,7 @@ Rules:
 | `pipeline.publish.locks[]` | list | No | empty |
 | `pipeline.publish.locks[].source` | string | Yes (per lock) | must reference supported publish source |
 | `pipeline.publish.locks[].reason` | string | No | empty |
-| `pipeline.whisperx.transcribe_url` | string | Yes | valid URL |
+| `pipeline.whisperx.transcribe_url` | string | Yes | absolute `http` or `https` URL |
 | `pipeline.whisperx.language` | string | No | `en` |
 | `pipeline.whisperx.timeout` | duration | No | `30m` |
 | `pipeline.whisperx.retries` | int | No | `3` |
--- a/docs/integrations/whisperx.md
+++ b/docs/integrations/whisperx.md
@@ -17,6 +17,11 @@ Narratio sends an HTTP `POST` to the configured transcription URL using
 The server must return a `2xx` response whose body is valid JSON. Narratio does
 not currently require a more specific response schema at this boundary.

+The transcription URL must be an absolute `http` or `https` URL. The audio body
+is streamed through a fresh multipart writer for every attempt, so its memory
+use is bounded by the transport buffer rather than by the complete audio file.
+WhisperX response acquisition is capped at 10 MiB.
+
 ## Request And Result Contract

 Each adapter request identifies a speaker, a readable audio file, and the
@@ -40,7 +45,7 @@ transcript output.

 ## Validation And Failure Semantics

-Client construction rejects a missing or invalid absolute transcription URL,
+Client construction rejects a missing or non-HTTP(S) absolute transcription URL,
 a missing language, a non-positive timeout, negative retries, or a negative
 retry delay. A request fails before transmission when its audio or output path
 is missing.
--- a/docs/roadmap/implementation.md
+++ b/docs/roadmap/implementation.md
@@ -810,6 +810,8 @@ server early close, cancellation, blocked producer, exact payloads, bounded
 responses, and goroutine completion. Run Whisper packages and their callers under
 `go test -race`; this stage unblocks the full race suite in Stage 31.

+**Status:** Completed.
+
 ## Stage 24 — Correct prepare/transcribe transition semantics

 **Read first:** `audit-findings.md` lines 2282–2379 (COR-017 through COR-019)
--- a/internal/adapters/whisperx/fake.go
+++ b/internal/adapters/whisperx/fake.go
@@ -3,6 +3,7 @@ package whisperx
 import (
 	"context"
 	"path/filepath"
+	"sync"

 	"gitea.maximumdirect.net/eric/narratio/internal/fileops"
 )
@@ -32,7 +33,8 @@ func (n *NoopClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra

 // FakeClient captures requests and returns deterministic responses for tests.
 type FakeClient struct {
-	Requests     []TranscribeRequest
+	requestsMu   sync.RWMutex
+	requests     []TranscribeRequest
 	Err          error
 	Result       TranscribeResult
 	TranscribeFn func(ctx context.Context, req TranscribeRequest) (TranscribeResult, error)
@@ -43,7 +45,9 @@ func (f *FakeClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra
 	if err := ctx.Err(); err != nil {
 		return TranscribeResult{}, err
 	}
-	f.Requests = append(f.Requests, req)
+	f.requestsMu.Lock()
+	f.requests = append(f.requests, req)
+	f.requestsMu.Unlock()
 	if f.TranscribeFn != nil {
 		return f.TranscribeFn(ctx, req)
 	}
@@ -66,6 +70,13 @@ func (f *FakeClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra
 	return res, nil
 }

+// RequestsSnapshot returns a copy of captured requests safe for concurrent test assertions.
+func (f *FakeClient) RequestsSnapshot() []TranscribeRequest {
+	f.requestsMu.RLock()
+	defer f.requestsMu.RUnlock()
+	return append([]TranscribeRequest(nil), f.requests...)
+}
+
 func writeMinimalJSON(path string) error {
 	if path == "" {
 		return nil
--- a/internal/adapters/whisperx/fake_test.go
+++ b/internal/adapters/whisperx/fake_test.go
@@ -3,6 +3,7 @@ package whisperx
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 )

@@ -14,8 +15,9 @@ func TestFakeClientCapturesRequestAndReturnsPath(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Transcribe() error = %v", err)
 	}
-	if len(fake.Requests) != 1 || fake.Requests[0].SpeakerID != "alice" {
-		t.Fatalf("requests = %#v, want one alice request", fake.Requests)
+	requests := fake.RequestsSnapshot()
+	if len(requests) != 1 || requests[0].SpeakerID != "alice" {
+		t.Fatalf("requests = %#v, want one alice request", requests)
 	}
 	if res.OutputRawTranscriptPath != req.OutputRawTranscriptPath {
 		t.Fatalf("output path = %q, want %q", res.OutputRawTranscriptPath, req.OutputRawTranscriptPath)
@@ -29,3 +31,22 @@ func TestFakeClientError(t *testing.T) {
 		t.Fatal("expected error, got nil")
 	}
 }
+
+func TestFakeClientRequestsSnapshotSupportsConcurrentCalls(t *testing.T) {
+	fake := &FakeClient{}
+	const callers = 16
+	var group sync.WaitGroup
+	group.Add(callers)
+	for i := 0; i < callers; i++ {
+		go func() {
+			defer group.Done()
+			if _, err := fake.Transcribe(context.Background(), TranscribeRequest{}); err != nil {
+				t.Errorf("Transcribe() error = %v", err)
+			}
+		}()
+	}
+	group.Wait()
+	if got := len(fake.RequestsSnapshot()); got != callers {
+		t.Fatalf("captured requests = %d, want %d", got, callers)
+	}
+}
--- a/internal/adapters/whisperx/http.go
+++ b/internal/adapters/whisperx/http.go
@@ -1,7 +1,6 @@
 package whisperx

 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -14,12 +13,16 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"

 	"gitea.maximumdirect.net/eric/narratio/internal/fileops"
 )

-const defaultMaxResponseBytes int64 = 10 * 1024 * 1024
+const (
+	defaultMaxWhisperXResponseBytes int64 = 10 * 1024 * 1024
+	whisperXUploadBufferSize              = 32 * 1024
+)

 // HTTPClientConfig contains parsed, deterministic WhisperX HTTP client settings.
 type HTTPClientConfig struct {
@@ -41,6 +44,7 @@ type HTTPClient struct {
 	retryDelay       time.Duration
 	httpClient       *http.Client
 	maxResponseBytes int64
+	openAudio        func(string) (io.ReadCloser, error)
 }

 // NewHTTPClientFromConfigValues builds a client from config values and parses durations once.
@@ -74,11 +78,11 @@ func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) {
 		return nil, fmt.Errorf("whisperx transcribe_url is required")
 	}
 	u, err := url.Parse(cfg.TranscribeURL)
-	if err != nil || u.Scheme == "" || u.Host == "" {
+	if err != nil || !u.IsAbs() || u.Host == "" || !isHTTPURLScheme(u.Scheme) {
 		if err != nil {
 			return nil, fmt.Errorf("invalid whisperx transcribe_url %q: %w", cfg.TranscribeURL, err)
 		}
-		return nil, fmt.Errorf("invalid whisperx transcribe_url %q", cfg.TranscribeURL)
+		return nil, fmt.Errorf("invalid whisperx transcribe_url %q: must be an absolute http or https URL", cfg.TranscribeURL)
 	}
 	if cfg.Timeout <= 0 {
 		return nil, fmt.Errorf("whisperx timeout must be > 0")
@@ -100,7 +104,7 @@ func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) {

 	maxBytes := cfg.MaxResponseBytes
 	if maxBytes <= 0 {
-		maxBytes = defaultMaxResponseBytes
+		maxBytes = defaultMaxWhisperXResponseBytes
 	}

 	return &HTTPClient{
@@ -111,6 +115,7 @@ func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) {
 		retryDelay:       cfg.RetryDelay,
 		httpClient:       client,
 		maxResponseBytes: maxBytes,
+		openAudio:        func(path string) (io.ReadCloser, error) { return os.Open(path) },
 	}, nil
 }

@@ -185,52 +190,40 @@ func (c *HTTPClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra
 }

 func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) (int, []byte, error) {
-	bodyBuf := &bytes.Buffer{}
-	writer := multipart.NewWriter(bodyBuf)
+	upload := newMultipartUpload(ctx, audioPath, c.language, c.openAudio)
+	defer upload.Close()

-	fileWriter, err := writer.CreateFormFile("file", filepath.Base(audioPath))
-	if err != nil {
-		return 0, nil, fmt.Errorf("create multipart file field: %w", err)
-	}
-
-	audioFile, err := os.Open(audioPath)
-	if err != nil {
-		return 0, nil, fmt.Errorf("open audio file %q: %w", audioPath, err)
-	}
-	if _, err := io.Copy(fileWriter, audioFile); err != nil {
-		_ = audioFile.Close()
-		return 0, nil, fmt.Errorf("copy audio file %q: %w", audioPath, err)
-	}
-	if err := audioFile.Close(); err != nil {
-		return 0, nil, fmt.Errorf("close audio file %q: %w", audioPath, err)
-	}
-
-	if err := writer.WriteField("language", c.language); err != nil {
-		return 0, nil, fmt.Errorf("write language form field: %w", err)
-	}
-	if err := writer.Close(); err != nil {
-		return 0, nil, fmt.Errorf("close multipart writer: %w", err)
-	}
-
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), bodyBuf)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), upload)
 	if err != nil {
 		return 0, nil, fmt.Errorf("build whisperx request: %w", err)
 	}
-	req.Header.Set("Content-Type", writer.FormDataContentType())
+	req.Header.Set("Content-Type", upload.contentType)

 	resp, err := c.httpClient.Do(req)
 	if err != nil {
+		_ = upload.Close()
+		if producerErr := upload.Wait(); producerErr != nil {
+			return 0, nil, fmt.Errorf("stream whisperx request body: %w", producerErr)
+		}
 		return 0, nil, fmt.Errorf("perform whisperx request: %w", err)
 	}
 	defer resp.Body.Close()

-	data, err := readBounded(resp.Body, c.maxResponseBytes)
-	if err != nil {
-		return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err)
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		_ = upload.Close()
+		if _, err := readWhisperXResponse(resp.Body, c.maxResponseBytes); err != nil {
+			return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err)
+		}
+		return resp.StatusCode, nil, fmt.Errorf("whisperx returned status %d", resp.StatusCode)
 	}

-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return resp.StatusCode, nil, fmt.Errorf("whisperx returned status %d", resp.StatusCode)
+	if err := upload.Wait(); err != nil {
+		return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", err)
+	}
+
+	data, err := readWhisperXResponse(resp.Body, c.maxResponseBytes)
+	if err != nil {
+		return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err)
 	}
 	return resp.StatusCode, data, nil
 }
@@ -266,18 +259,183 @@ func (c *HTTPClient) shouldRetry(parent context.Context, err error, status int)
 	return false
 }

-func readBounded(r io.Reader, maxBytes int64) ([]byte, error) {
+func readWhisperXResponse(r io.Reader, maxBytes int64) ([]byte, error) {
 	limited := io.LimitReader(r, maxBytes+1)
 	data, err := io.ReadAll(limited)
 	if err != nil {
 		return nil, err
 	}
 	if int64(len(data)) > maxBytes {
-		return nil, fmt.Errorf("response exceeds max size %d bytes", maxBytes)
+		return nil, fmt.Errorf("whisperx response exceeds configured limit of %d bytes", maxBytes)
 	}
 	return data, nil
 }

+func isHTTPURLScheme(scheme string) bool {
+	switch strings.ToLower(scheme) {
+	case "http", "https":
+		return true
+	default:
+		return false
+	}
+}
+
+type multipartUpload struct {
+	reader      *io.PipeReader
+	writer      *io.PipeWriter
+	contentType string
+	done        chan struct{}
+
+	mu      sync.Mutex
+	audio   io.Closer
+	err     error
+	aborted bool
+}
+
+func newMultipartUpload(ctx context.Context, audioPath, language string, openAudio func(string) (io.ReadCloser, error)) *multipartUpload {
+	reader, writer := io.Pipe()
+	multipartWriter := multipart.NewWriter(writer)
+	upload := &multipartUpload{
+		reader:      reader,
+		writer:      writer,
+		contentType: multipartWriter.FormDataContentType(),
+		done:        make(chan struct{}),
+	}
+
+	go func() {
+		err := upload.write(ctx, multipartWriter, audioPath, language, openAudio)
+		if err != nil {
+			_ = writer.CloseWithError(err)
+		} else {
+			_ = writer.Close()
+		}
+		upload.mu.Lock()
+		upload.err = err
+		upload.audio = nil
+		upload.mu.Unlock()
+		close(upload.done)
+	}()
+
+	go func() {
+		select {
+		case <-ctx.Done():
+			upload.abort()
+		case <-upload.done:
+		}
+	}()
+
+	return upload
+}
+
+func (u *multipartUpload) Read(p []byte) (int, error) {
+	return u.reader.Read(p)
+}
+
+func (u *multipartUpload) Close() error {
+	u.abort()
+	<-u.done
+	return nil
+}
+
+func (u *multipartUpload) Wait() error {
+	<-u.done
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	return u.err
+}
+
+func (u *multipartUpload) write(ctx context.Context, writer *multipart.Writer, audioPath, language string, openAudio func(string) (io.ReadCloser, error)) error {
+	fileWriter, err := writer.CreateFormFile("file", filepath.Base(audioPath))
+	if err != nil {
+		return u.producerError(ctx, fmt.Errorf("create multipart file field: %w", err))
+	}
+
+	audioFile, err := openAudio(audioPath)
+	if err != nil {
+		return u.producerError(ctx, fmt.Errorf("open audio file %q: %w", audioPath, err))
+	}
+	u.setAudio(audioFile)
+
+	_, copyErr := io.CopyBuffer(fileWriter, &contextReader{ctx: ctx, reader: audioFile}, make([]byte, whisperXUploadBufferSize))
+	closeErr := audioFile.Close()
+	u.clearAudio(audioFile)
+	if copyErr != nil {
+		return u.producerError(ctx, fmt.Errorf("copy audio file %q: %w", audioPath, copyErr))
+	}
+	if closeErr != nil {
+		return u.producerError(ctx, fmt.Errorf("close audio file %q: %w", audioPath, closeErr))
+	}
+
+	if err := writer.WriteField("language", language); err != nil {
+		return u.producerError(ctx, fmt.Errorf("write language form field: %w", err))
+	}
+	if err := writer.Close(); err != nil {
+		return u.producerError(ctx, fmt.Errorf("close multipart writer: %w", err))
+	}
+	return nil
+}
+
+func (u *multipartUpload) producerError(ctx context.Context, err error) error {
+	if ctx.Err() != nil {
+		return ctx.Err()
+	}
+	u.mu.Lock()
+	aborted := u.aborted
+	u.mu.Unlock()
+	if aborted {
+		return nil
+	}
+	return err
+}
+
+func (u *multipartUpload) setAudio(audio io.Closer) {
+	u.mu.Lock()
+	u.audio = audio
+	aborted := u.aborted
+	u.mu.Unlock()
+	if aborted {
+		_ = audio.Close()
+	}
+}
+
+func (u *multipartUpload) clearAudio(audio io.Closer) {
+	u.mu.Lock()
+	if u.audio == audio {
+		u.audio = nil
+	}
+	u.mu.Unlock()
+}
+
+func (u *multipartUpload) abort() {
+	u.mu.Lock()
+	if u.aborted {
+		u.mu.Unlock()
+		return
+	}
+	u.aborted = true
+	audio := u.audio
+	u.mu.Unlock()
+
+	_ = u.reader.Close()
+	if audio != nil {
+		_ = audio.Close()
+	}
+}
+
+type contextReader struct {
+	ctx    context.Context
+	reader io.Reader
+}
+
+func (r *contextReader) Read(p []byte) (int, error) {
+	select {
+	case <-r.ctx.Done():
+		return 0, r.ctx.Err()
+	default:
+		return r.reader.Read(p)
+	}
+}
+
 func writeFileAtomic(path string, data []byte, perm os.FileMode) error {
 	if strings.TrimSpace(path) == "" {
 		return fmt.Errorf("path is required")
--- a/internal/adapters/whisperx/http_test.go
+++ b/internal/adapters/whisperx/http_test.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"io"
+	"mime/multipart"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -19,6 +20,7 @@ func TestHTTPClientTranscribeSuccess(t *testing.T) {
 	var gotLanguage string
 	var gotFileField string
 	var gotFileSize int
+	var gotFileData string

 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodPost {
@@ -40,6 +42,7 @@ func TestHTTPClientTranscribeSuccess(t *testing.T) {
 			t.Fatalf("ReadAll(file) error = %v", err)
 		}
 		gotFileSize = len(data)
+		gotFileData = string(data)

 		w.Header().Set("Content-Type", "application/json")
 		_, _ = w.Write([]byte(`{"schema":"speaker_transcript.v1","segments":[]}`))
@@ -77,13 +80,27 @@ func TestHTTPClientTranscribeSuccess(t *testing.T) {
 	if gotFileSize == 0 {
 		t.Fatal("file size = 0, want >0")
 	}
+	if gotFileData != "audio-data" {
+		t.Fatalf("file data = %q, want exact payload", gotFileData)
+	}
 	verifyJSONFile(t, outPath)
 }

 func TestHTTPClientRetriesOnTransientAndSucceeds(t *testing.T) {
 	var calls atomic.Int32
+	var payloads []string
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		n := calls.Add(1)
+		file, _, err := r.FormFile("file")
+		if err != nil {
+			t.Fatalf("FormFile(file) error = %v", err)
+		}
+		data, err := io.ReadAll(file)
+		_ = file.Close()
+		if err != nil {
+			t.Fatalf("ReadAll(file) error = %v", err)
+		}
+		payloads = append(payloads, string(data))
 		if n == 1 {
 			http.Error(w, "temporary", http.StatusInternalServerError)
 			return
@@ -112,6 +129,9 @@ func TestHTTPClientRetriesOnTransientAndSucceeds(t *testing.T) {
 	if calls.Load() != 2 {
 		t.Fatalf("calls = %d, want 2", calls.Load())
 	}
+	if len(payloads) != 2 || payloads[0] != "audio-data" || payloads[1] != "audio-data" {
+		t.Fatalf("retry payloads = %#v, want two exact audio payloads", payloads)
+	}
 	verifyJSONFile(t, outPath)
 }

@@ -247,8 +267,247 @@ func TestHTTPClientConstructorValidation(t *testing.T) {
 	if err == nil {
 		t.Fatal("expected bad retry_delay error")
 	}
+	for _, endpoint := range []string{"ftp://example.com/transcribe", "file:///tmp/transcribe", "//example.com/transcribe", "https:/missing-host"} {
+		if _, err := NewHTTPClientFromConfigValues(endpoint, "en", "30m", "2s", 1); err == nil {
+			t.Errorf("NewHTTPClientFromConfigValues(%q) error = nil, want endpoint validation error", endpoint)
+		}
+	}
+	for _, endpoint := range []string{"http://example.com/transcribe", "https://example.com/transcribe"} {
+		if _, err := NewHTTPClientFromConfigValues(endpoint, "en", "30m", "2s", 1); err != nil {
+			t.Errorf("NewHTTPClientFromConfigValues(%q) error = %v", endpoint, err)
+		}
+	}
 }

+func TestHTTPClientStreamsUploadBeforeSourceCompletes(t *testing.T) {
+	release := make(chan struct{})
+	source := newGatedReadCloser([]byte("audio-data"), release)
+	firstByteReceived := make(chan struct{})
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		part := firstMultipartFilePart(t, r)
+		buf := make([]byte, 1)
+		if _, err := part.Read(buf); err != nil {
+			t.Errorf("Read(file) error = %v", err)
+			return
+		}
+		close(firstByteReceived)
+		if _, err := io.Copy(io.Discard, part); err != nil {
+			t.Errorf("discard remaining file data: %v", err)
+			return
+		}
+		_, _ = w.Write([]byte(`{"ok":true}`))
+	}))
+	defer srv.Close()
+
+	client := newTestHTTPClient(t, srv.URL)
+	client.openAudio = func(string) (io.ReadCloser, error) { return source, nil }
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := client.Transcribe(context.Background(), TranscribeRequest{AudioPath: "audio.flac", OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")})
+		done <- err
+	}()
+
+	select {
+	case <-firstByteReceived:
+		close(release)
+	case <-time.After(time.Second):
+		t.Fatal("server did not receive streamed audio before source completed")
+	}
+	if err := <-done; err != nil {
+		t.Fatalf("Transcribe() error = %v", err)
+	}
+}
+
+func TestHTTPClientSourceReadFailureReachesCaller(t *testing.T) {
+	sourceErr := errors.New("source read failed")
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+	}))
+	defer srv.Close()
+
+	client := newTestHTTPClient(t, srv.URL)
+	client.openAudio = func(string) (io.ReadCloser, error) {
+		return &failingReadCloser{first: []byte("partial"), err: sourceErr}, nil
+	}
+
+	_, err := client.Transcribe(context.Background(), TranscribeRequest{AudioPath: "audio.flac", OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")})
+	if !errors.Is(err, sourceErr) {
+		t.Fatalf("Transcribe() error = %v, want source read failure", err)
+	}
+}
+
+func TestHTTPClientEarlyServerResponseReleasesBlockedProducer(t *testing.T) {
+	release := make(chan struct{})
+	source := newGatedReadCloser([]byte("audio-data"), release)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "bad request", http.StatusBadRequest)
+	}))
+	defer srv.Close()
+
+	client := newTestHTTPClient(t, srv.URL)
+	client.openAudio = func(string) (io.ReadCloser, error) { return source, nil }
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := client.Transcribe(context.Background(), TranscribeRequest{AudioPath: "audio.flac", OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")})
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("Transcribe() error = nil, want HTTP status error")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Transcribe() did not finish after server closed the request early")
+	}
+	select {
+	case <-source.closed:
+	case <-time.After(time.Second):
+		t.Fatal("blocked audio source was not closed")
+	}
+}
+
+func TestHTTPClientCancellationReleasesBlockedProducer(t *testing.T) {
+	release := make(chan struct{})
+	source := newGatedReadCloser([]byte("audio-data"), release)
+	firstByteReceived := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		part := firstMultipartFilePart(t, r)
+		buf := make([]byte, 1)
+		if _, err := part.Read(buf); err != nil {
+			t.Errorf("Read(file) error = %v", err)
+			return
+		}
+		close(firstByteReceived)
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	client := newTestHTTPClient(t, srv.URL)
+	client.openAudio = func(string) (io.ReadCloser, error) { return source, nil }
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		_, err := client.Transcribe(ctx, TranscribeRequest{AudioPath: "audio.flac", OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")})
+		done <- err
+	}()
+
+	select {
+	case <-firstByteReceived:
+		cancel()
+	case <-time.After(time.Second):
+		cancel()
+		t.Fatal("server did not receive initial streamed audio")
+	}
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("Transcribe() error = %v, want context cancellation", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Transcribe() did not finish after cancellation")
+	}
+	select {
+	case <-source.closed:
+	case <-time.After(time.Second):
+		t.Fatal("blocked audio source was not closed on cancellation")
+	}
+}
+
+func TestHTTPClientBoundsWhisperXResponse(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		_, _ = w.Write([]byte(`{"ok":true}`))
+	}))
+	defer srv.Close()
+
+	client, err := NewHTTPClient(HTTPClientConfig{TranscribeURL: srv.URL, Language: "en", Timeout: time.Second, MaxResponseBytes: 4})
+	if err != nil {
+		t.Fatalf("NewHTTPClient() error = %v", err)
+	}
+	audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data")
+	_, err = client.Transcribe(context.Background(), TranscribeRequest{AudioPath: audioPath, OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")})
+	if err == nil || !strings.Contains(err.Error(), "whisperx response exceeds configured limit") {
+		t.Fatalf("Transcribe() error = %v, want bounded WhisperX response error", err)
+	}
+}
+
+func newTestHTTPClient(t *testing.T, endpoint string) *HTTPClient {
+	t.Helper()
+	client, err := NewHTTPClientFromConfigValues(endpoint, "en", "2s", "1ms", 0)
+	if err != nil {
+		t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err)
+	}
+	return client
+}
+
+func firstMultipartFilePart(t *testing.T, r *http.Request) *multipart.Part {
+	t.Helper()
+	reader, err := r.MultipartReader()
+	if err != nil {
+		t.Fatalf("MultipartReader() error = %v", err)
+	}
+	part, err := reader.NextPart()
+	if err != nil {
+		t.Fatalf("NextPart() error = %v", err)
+	}
+	if part.FormName() != "file" {
+		t.Fatalf("first form field = %q, want file", part.FormName())
+	}
+	return part
+}
+
+type gatedReadCloser struct {
+	first   []byte
+	release <-chan struct{}
+	closed  chan struct{}
+	sent    bool
+	once    atomic.Bool
+}
+
+func newGatedReadCloser(first []byte, release <-chan struct{}) *gatedReadCloser {
+	return &gatedReadCloser{first: first, release: release, closed: make(chan struct{})}
+}
+
+func (r *gatedReadCloser) Read(p []byte) (int, error) {
+	if !r.sent {
+		r.sent = true
+		return copy(p, r.first), nil
+	}
+	select {
+	case <-r.release:
+		return 0, io.EOF
+	case <-r.closed:
+		return 0, errors.New("audio source closed")
+	}
+}
+
+func (r *gatedReadCloser) Close() error {
+	if r.once.CompareAndSwap(false, true) {
+		close(r.closed)
+	}
+	return nil
+}
+
+type failingReadCloser struct {
+	first []byte
+	err   error
+	sent  bool
+}
+
+func (r *failingReadCloser) Read(p []byte) (int, error) {
+	if !r.sent {
+		r.sent = true
+		return copy(p, r.first), nil
+	}
+	return 0, r.err
+}
+
+func (r *failingReadCloser) Close() error { return nil }
+
 func writeWhisperXTestFile(t *testing.T, name, contents string) string {
 	t.Helper()
 	path := filepath.Join(t.TempDir(), name)
--- a/internal/app/whisperx_wiring_test.go
+++ b/internal/app/whisperx_wiring_test.go
@@ -132,8 +132,9 @@ func TestExecuteStagesExplicitWhisperXInjectionOverridesDefaultWiring(t *testing
 	if err != nil {
 		t.Fatalf("executeStages() error = %v", err)
 	}
-	if len(fake.Requests) != 1 {
-		t.Fatalf("fake whisperx requests = %d, want 1", len(fake.Requests))
+	requests := fake.RequestsSnapshot()
+	if len(requests) != 1 {
+		t.Fatalf("fake whisperx requests = %d, want 1", len(requests))
 	}
 }

--- a/internal/config/validate.go
+++ b/internal/config/validate.go
@@ -395,11 +395,11 @@ func validateWhisperX(cfg WhisperXConfig) error {
 		return fmt.Errorf("pipeline.whisperx.transcribe_url is required")
 	}
 	u, err := url.Parse(cfg.TranscribeURL)
-	if err != nil || u.Scheme == "" || u.Host == "" {
+	if err != nil || !u.IsAbs() || u.Host == "" || (strings.ToLower(u.Scheme) != "http" && strings.ToLower(u.Scheme) != "https") {
 		if err != nil {
-			return fmt.Errorf("pipeline.whisperx.transcribe_url must be a valid URL: %w", err)
+			return fmt.Errorf("pipeline.whisperx.transcribe_url must be an absolute http or https URL: %w", err)
 		}
-		return fmt.Errorf("pipeline.whisperx.transcribe_url must be a valid URL")
+		return fmt.Errorf("pipeline.whisperx.transcribe_url must be an absolute http or https URL")
 	}
 	if err := validateDuration("pipeline.whisperx.timeout", cfg.Timeout); err != nil {
 		return err
--- a/internal/config/validation_duration_test.go
+++ b/internal/config/validation_duration_test.go
@@ -39,6 +39,33 @@ func TestValidateDurationsRequirePositiveValues(t *testing.T) {
 	}
 }

+func TestValidateWhisperXRequiresAbsoluteHTTPSEndpoint(t *testing.T) {
+	retries := 0
+	concurrency := 1
+	base := WhisperXConfig{
+		Language:    "en",
+		Timeout:     "1s",
+		Retries:     &retries,
+		RetryDelay:  "0s",
+		Concurrency: &concurrency,
+	}
+
+	for _, endpoint := range []string{"ftp://example.com/transcribe", "file:///tmp/transcribe", "//example.com/transcribe", "https:/missing-host"} {
+		cfg := base
+		cfg.TranscribeURL = endpoint
+		if err := validateWhisperX(cfg); err == nil || !strings.Contains(err.Error(), "absolute http or https URL") {
+			t.Errorf("validateWhisperX(%q) error = %v, want HTTP(S) endpoint validation", endpoint, err)
+		}
+	}
+	for _, endpoint := range []string{"http://example.com/transcribe", "https://example.com/transcribe"} {
+		cfg := base
+		cfg.TranscribeURL = endpoint
+		if err := validateWhisperX(cfg); err != nil {
+			t.Errorf("validateWhisperX(%q) error = %v", endpoint, err)
+		}
+	}
+}
+
 func TestValidateNotariusTimeoutRequiresPositiveValue(t *testing.T) {
 	for _, value := range []string{"0s", "-1ms"} {
 		t.Run(value, func(t *testing.T) {
--- a/internal/stage/placeholders_test.go
+++ b/internal/stage/placeholders_test.go
@@ -235,8 +235,9 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
 		}
 	}

-	if len(wf.Requests) != 1 {
-		t.Fatalf("whisperx calls = %d, want 1", len(wf.Requests))
+	whisperXRequests := wf.RequestsSnapshot()
+	if len(whisperXRequests) != 1 {
+		t.Fatalf("whisperx calls = %d, want 1", len(whisperXRequests))
 	}
 	if len(sf.Requests) != 1 {
 		t.Fatalf("seriatim calls = %d, want 1", len(sf.Requests))
--- a/internal/stage/transcribe_test.go
+++ b/internal/stage/transcribe_test.go
@@ -206,10 +206,11 @@ func TestTranscribeStageUsesRunLocalOutputAndMaterializesCanonical(t *testing.T)
 	if err != nil {
 		t.Fatalf("transcribe.Run() error = %v", err)
 	}
-	if len(fake.Requests) != 1 {
-		t.Fatalf("requests = %d, want 1", len(fake.Requests))
+	requests := fake.RequestsSnapshot()
+	if len(requests) != 1 {
+		t.Fatalf("requests = %d, want 1", len(requests))
 	}
-	runOut := fake.Requests[0].OutputRawTranscriptPath
+	runOut := requests[0].OutputRawTranscriptPath
 	if !strings.Contains(runOut, filepath.Join("runs", m.RunID, "transcribe", "outputs")) {
 		t.Fatalf("run-local output path = %q, want runs/{run_id}/transcribe/outputs path", runOut)
 	}