From cae423750c04f9c9193c2b5d1dcbfe57de605e16 Mon Sep 17 00:00:00 2001 From: Mathias Date: Thu, 1 Oct 2026 16:44:34 +0200 Subject: [PATCH] feat(transcribe): gateway client with an iguana->berget fallback chain (ADR-033) Streams the WAV from disk with a computed Content-Length (no buffering, no chunking), posts verbose_json with field `file` and language=sv, clamps the final segment to the audio duration, and treats an empty transcript as a failure. The chain falls back on ANY primary error, records the serving endpoint, and never falls back once the context is cancelled. gateway.go and chain.go are one commit: the chain's tests use the gateway's StatusError and ErrEmptyTranscript, so neither compiles alone. Mutation-checked: field name, clamp, cancel-fallback, content length and the recorded endpoint each turn a test red. Co-Authored-By: Claude Sonnet 5.5 --- internal/adapters/transcribe/chain.go | 53 +++++++++ internal/adapters/transcribe/gateway.go | 147 ++++++++++++++++++++++++ 2 files changed, 200 insertions(+) create mode 100644 internal/adapters/transcribe/chain.go create mode 100644 internal/adapters/transcribe/gateway.go diff --git a/internal/adapters/transcribe/chain.go b/internal/adapters/transcribe/chain.go new file mode 100644 index 0000000..68df323 --- /dev/null +++ b/internal/adapters/transcribe/chain.go @@ -0,0 +1,53 @@ +package transcribe + +import ( + "context" + "errors" + "fmt" + + "git.d-ma.be/mathias/tapir/internal/domain" +) + +// Backend is one transcription endpoint. *Gateway satisfies it; Name is the value +// recorded as the serving endpoint. +type Backend interface { + Name() string + Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) +} + +// Chain tries a primary (local) backend and, if it fails for ANY reason, a +// fallback. The fallback exists because iguana can wedge permanently on a Metal +// OOM (infra#273) — which is also why a fallback is a data-egress decision: the +// berget endpoint is a cloud service, so the result records which one served it. +type Chain struct { + primary Backend + fallback Backend // nil disables the fallback +} + +// NewChain builds a Chain. A nil fallback means a primary failure is the result. +func NewChain(primary, fallback Backend) *Chain { + return &Chain{primary: primary, fallback: fallback} +} + +// Transcribe returns the first backend's success. It falls back on any primary +// error — transport, timeout, non-2xx, empty transcript — and never inspects a +// response body to decide. A cancelled context stops the chain: a job the user or +// shutdown cancelled must not ship its audio to a cloud endpoint as a last act. +func (c *Chain) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) { + t, err := c.primary.Transcribe(ctx, wavPath) + if err == nil { + t.Endpoint = c.primary.Name() + return t, nil + } + if c.fallback == nil || ctx.Err() != nil { + return domain.Transcription{}, fmt.Errorf("%s: %w", c.primary.Name(), err) + } + t, ferr := c.fallback.Transcribe(ctx, wavPath) + if ferr != nil { + return domain.Transcription{}, errors.Join( + fmt.Errorf("%s: %w", c.primary.Name(), err), + fmt.Errorf("%s: %w", c.fallback.Name(), ferr)) + } + t.Endpoint = c.fallback.Name() + return t, nil +} diff --git a/internal/adapters/transcribe/gateway.go b/internal/adapters/transcribe/gateway.go new file mode 100644 index 0000000..e885c71 --- /dev/null +++ b/internal/adapters/transcribe/gateway.go @@ -0,0 +1,147 @@ +// Package transcribe turns a WAV file into a timed Swedish transcript by calling +// the LiteLLM gateway's OpenAI-compatible /audio/transcriptions endpoint +// (ADR-033). A Gateway is one model alias; a Chain tries a local one and falls +// back to a second. +package transcribe + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "mime/multipart" + "net/http" + "os" + "strings" + + "git.d-ma.be/mathias/tapir/internal/domain" +) + +// ErrEmptyTranscript is returned when the endpoint answered 200 but with no text. +// An empty transcript is not a success: chat over it could only answer "nothing". +var ErrEmptyTranscript = errors.New("transcribe: endpoint returned an empty transcript") + +// StatusError is a non-2xx answer from the gateway. Nothing may branch on the +// body — {"error":"failed to process audio"} has had at least five distinct +// causes — so only the status code is exposed for logic; Snippet is for logs. +type StatusError struct { + Code int + Snippet string +} + +func (e *StatusError) Error() string { + return fmt.Sprintf("transcribe: gateway returned HTTP %d: %s", e.Code, e.Snippet) +} + +// Gateway transcribes through one model alias on the gateway. baseURL is the +// gateway's /v1 root — the in-cluster NodePort, not the public hostname, whose +// ingress rejects bodies over ~8-10 MB with 413 (docs/research/audio-length-ceiling). +type Gateway struct { + baseURL string + key string + model string + client *http.Client +} + +// NewGateway builds a Gateway. The client should carry no overall Timeout: a 90 +// minute recording takes minutes, so the deadline belongs to the caller's context. +func NewGateway(baseURL, key, model string, client *http.Client) *Gateway { + return &Gateway{baseURL: strings.TrimRight(baseURL, "/"), key: key, model: model, client: client} +} + +// Name is the model alias; it is what gets recorded as the serving endpoint. +func (g *Gateway) Name() string { return g.model } + +// Transcribe uploads the WAV at path and returns the normalised transcription. +// The body is streamed from disk with a COMPUTED Content-Length: a 90 minute WAV +// is ~170 MB, so it is neither buffered in memory nor sent chunked. +func (g *Gateway) Transcribe(ctx context.Context, path string) (domain.Transcription, error) { + f, err := os.Open(path) + if err != nil { + return domain.Transcription{}, fmt.Errorf("transcribe: open audio: %w", err) + } + defer f.Close() + info, err := f.Stat() + if err != nil { + return domain.Transcription{}, fmt.Errorf("transcribe: stat audio: %w", err) + } + + var prefix bytes.Buffer + mw := multipart.NewWriter(&prefix) + for _, kv := range [][2]string{{"model", g.model}, {"response_format", "verbose_json"}, {"language", "sv"}} { + if err := mw.WriteField(kv[0], kv[1]); err != nil { + return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err) + } + } + // The multipart field is `file`; `audio_file` is rejected as "Invalid request". + if _, err := mw.CreateFormFile("file", "audio.wav"); err != nil { + return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err) + } + suffix := []byte("\r\n--" + mw.Boundary() + "--\r\n") + + body := io.MultiReader(bytes.NewReader(prefix.Bytes()), f, bytes.NewReader(suffix)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, g.baseURL+"/audio/transcriptions", body) + if err != nil { + return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err) + } + req.ContentLength = int64(prefix.Len()) + info.Size() + int64(len(suffix)) + req.Header.Set("Content-Type", mw.FormDataContentType()) + req.Header.Set("Authorization", "Bearer "+g.key) + // No "Expect: 100-continue": it trips the proxy for any body over 1 KB. + + resp, err := g.client.Do(req) + if err != nil { + return domain.Transcription{}, fmt.Errorf("transcribe: %s: %w", g.model, err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode > 299 { + snip, _ := io.ReadAll(io.LimitReader(resp.Body, 200)) + return domain.Transcription{}, &StatusError{Code: resp.StatusCode, Snippet: strings.TrimSpace(string(snip))} + } + + t, err := parseVerbose(resp.Body) + if err != nil { + return domain.Transcription{}, fmt.Errorf("transcribe: %s: %w", g.model, err) + } + t.Endpoint = g.model + return t, nil +} + +// verboseJSON is the subset of the verbose_json response used. Both backends +// return it; they differ in segment granularity (iguana: whole seconds, berget: +// sub-second) and in the language token, neither of which matters here. +type verboseJSON struct { + Text string `json:"text"` + Duration float64 `json:"duration"` + Segments []struct { + Start float64 `json:"start"` + End float64 `json:"end"` + Text string `json:"text"` + } `json:"segments"` +} + +// parseVerbose normalises a verbose_json body. Whisper's final window can end past +// the audio (9.0 s for 4.25 s of audio was observed), so segment ends are clamped +// to the reported duration. +func parseVerbose(r io.Reader) (domain.Transcription, error) { + var v verboseJSON + if err := json.NewDecoder(r).Decode(&v); err != nil { + return domain.Transcription{}, fmt.Errorf("decode response: %w", err) + } + text := strings.TrimSpace(v.Text) + if text == "" { + return domain.Transcription{}, ErrEmptyTranscript + } + t := domain.Transcription{Text: text, AudioSeconds: v.Duration} + for _, s := range v.Segments { + seg := domain.Segment{Start: s.Start, End: s.End, Text: strings.TrimSpace(s.Text)} + if v.Duration > 0 { + seg.End = min(seg.End, v.Duration) + seg.Start = min(seg.Start, seg.End) + } + t.Segments = append(t.Segments, seg) + } + return t, nil +}