diff --git a/CHANGELOG.md b/CHANGELOG.md index afb1bfb..a03f5de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Changelog +## [4.43.0] + +- `client.dictation` now handles dictations past the live endpoint's 120 s cap. When `transcribeLive()` or `openLive()` fails with 413 `audio_too_large`, the SDK keeps reading the audio until it ends (the stream or iterable ends, or `close()` / `result()` on a session), posts the whole recording once to the buffered endpoint `POST /v1/transcribe` (up to 20 minutes or 256 MB, same config, file name and content type), and resolves with that result exactly as a live one, `final_text` included. To have the recording to send, an in-memory copy of every uploaded audio byte is kept from the first one, about 32 KB per second at 16 kHz mono (about 38 MB at 20 minutes); a copy past 256 MB is dropped and the call throws `DictationError` (413, `audio_too_large`). A fallback result waits for the whole recording to upload and be transcribed, so it returns more slowly than a live one +- Add `fallbackToBuffered` to `DictationLiveOptions` (default `true`). Set it to `false` to keep no copy and get the live 413 as before. Only a 413 with `audio_too_large` falls back; every other failure is thrown unchanged, a failed buffered request throws its own error, and `abort()` or an abort `signal` drops everything with no fallback request. Once the fallback engages, `timeout` no longer cuts off the rest of the recording, and the buffered request gets a fresh `timeout` deadline + ## [4.42.0] - Add Text-to-Speech: `client.tts.synthesize(request)` returns one text's complete audio (`TextToSpeechResponse`: `audio` plus the response headers as `requestId`, `region`, `model`, `voiceId`, `encoding`, `sampleRate`, `billedCharacters`, `audioDurationMs`, ...), `client.tts.stream(request)` returns it as the API generates it (`TextToSpeechStream`, an async iterable over a web `ReadableStream`), and `client.tts.session(params)` opens a WebSocket `TextToSpeechSession` for voice agents: `connect()`, `generate()` / `flush()` / `audio()`, `speak()`, `cancel()`, `keepAlive()`, `close()`, with `on("audio" | "flushDone" | "wordBoundaries" | "cancelled" | "termination" | "error" | "close", ...)`. Requests take the API's fields (`text`, `voice_id`, `model_id`, `language`, `sample_rate`, `encoding`, `container`, `bit_rate`); failures throw `TextToSpeechError` (`status`, `errorCode`, `param`, `requestId`, `retryable`, `retryAfter`, `errors`, `closeCode`, `body`). New client option `ttsBaseUrl` (default `https://tts.us.assemblyai.com`); `client.tts.createTemporaryToken()` mints a token for browser sessions. Text sent while `connect()` is pending is held and sent the moment the socket opens, so it need not wait for `Begin` diff --git a/CLAUDE.md b/CLAUDE.md index 7084cc3..e898e46 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -328,7 +328,8 @@ transcript when the audio ends. iterable, or a web `ReadableStream` — or audio already held whole: a local file path (Node/Bun/Deno), a data URL, raw bytes (`Uint8Array`/`ArrayBuffer`), or a Blob/File, which is sent as a single chunk over the same connection. **Not** a URL — that throws, pointing at -`client.transcripts`. Audio must be WAV or raw 16-bit PCM, ≤120 s per request. +`client.transcripts`. Audio must be WAV or raw 16-bit PCM. The live upload takes ≤120 s; longer +dictations (≤20 min) fall back to the buffered endpoint by default (below). ```typescript await client.dictation.warm(); // open the connection ahead of time (does not validate the key) @@ -386,15 +387,28 @@ timestamps), `confidence`, `llm_response` (`string | null`, the rewrite when the 429/503. Auth, rate-limit, size and capacity failures can surface part-way through the upload rather than only at the end. +**Past 120 s — buffered fallback** (`DictationLiveOptions.fallbackToBuffered`, default `true`, applies +to `transcribeLive()` and `openLive()`): the SDK keeps an in-memory copy of every audio byte it +uploads. When the live request fails with 413 `audio_too_large`, it keeps reading the source until +it ends (stream/iterable ends, or `close()`/`result()`), then posts the whole copy once to `POST +/v1/transcribe` (≤20 min / 256 MB, same config, file name and content type) and returns that result +like a live one, `final_text` included. A failed buffered request throws its own error; any other +live failure is thrown unchanged. `abort()`/`signal` drop everything, with no fallback request. The +live deadline stops applying once the fallback engages; the buffered request gets a fresh `timeout`. +Trade-offs: a fallback result waits for the whole recording to upload and be transcribed (slower +than live), and the copy costs ~32 KB/s at 16 kHz mono (~38 MB at 20 min); a copy past 256 MB is +dropped and the call throws `DictationError` (413, `audio_too_large`). `fallbackToBuffered: false` +keeps no copy and throws the live 413 as before. + **Pre-warming**: `await client.dictation.warm()` issues a `GET /v1/warm` to take DNS + TCP + TLS off the critical path. It returns `true` once the socket is open — any HTTP response counts, even a non-200 one, because it does not validate the key — and `false` on a transport failure. Call it shortly before the request; the pooled connection idles out after a few seconds. -**Client-side timeout**: third argument, `DictationLiveOptions` — `{ timeout, signal }`. -`timeout` (default 300 000 ms) is a **total** deadline from the start of the request, spanning the -upload, the transcription and the LLM pass; the service caps audio at 120 s. `signal` cancels a -`transcribeLive()` from outside; sessions manage their own. +**Client-side timeout**: third argument, `DictationLiveOptions` — `{ timeout, signal, +fallbackToBuffered }`. `timeout` (default 300 000 ms) is a **total** deadline from the start of the +request, spanning the upload, the transcription and the LLM pass; the live upload caps audio at +120 s. `signal` cancels a `transcribeLive()` from outside; sessions manage their own. **Exports** from the package root: `DictationTranscriber`, `DictationLiveSession`, `DictationError`, and the `DictationAudioInput`, `DictationConfig`, `DictationLiveOptions`, @@ -464,7 +478,7 @@ Gotchas: `stream()` gives up a body that sends no audio for `readTimeout` ms (30 - **Streaming uses universal-3-5-pro** as the speech model - **Never expose API keys client-side** — use temporary auth tokens for browser streaming - **Sync timeout is a total deadline** — `transcribe()`, `transcribeLive()` and `openLive()` all default to a 180 s deadline spanning the whole request (upload plus transcription), and the sync API still caps audio at 120 s -- **Dictation timeout is a total 300 s deadline** — `client.dictation` spans the upload, the transcription and the LLM pass in one deadline; the service caps audio at 120 s and accepts WAV or raw 16-bit PCM only +- **Dictation timeout is a total 300 s deadline** — `client.dictation` spans the upload, the transcription and the LLM pass in one deadline; the live upload caps audio at 120 s (longer dictations fall back to the buffered endpoint, up to 20 min, unless `fallbackToBuffered: false`) and accepts WAV or raw 16-bit PCM only - **Node >= 18 required** - **Only runtime dependency**: ws (WebSocket library) - **Multi-runtime support**: Works in Node.js, Deno, Bun, Cloudflare Workers, and browsers diff --git a/README.md b/README.md index 27b3db0..8539e0f 100644 --- a/README.md +++ b/README.md @@ -557,8 +557,9 @@ console.log(result.final_text, result.session_id); `result.final_text` is the one to show the user: the LLM rewrite when an `llm_instruction` ran, the raw transcript otherwise. `result.text` always holds -the raw transcript. Audio must be WAV or raw 16-bit PCM, up to 120 seconds per -request. +the raw transcript. Audio must be WAV or raw 16-bit PCM. The live upload takes +up to 120 seconds; a longer dictation, up to 20 minutes, is sent again whole +(see "Dictations longer than 2 minutes" below). > This is not the streaming API. `client.streaming` returns words while the > speaker is still talking; dictation returns one finished transcript when the @@ -638,8 +639,8 @@ A URL is rejected with an error pointing at `client.transcripts`, which is where URL ingestion lives. `options.timeout` (default 300 000 ms) is a **total** deadline from the start of -the request, spanning the upload, the transcription and the LLM pass; the -service caps the audio itself at 120 seconds. `options.signal` drops the request +the request, spanning the upload, the transcription and the LLM pass; the live +upload caps the audio itself at 120 seconds. `options.signal` drops the request from outside, the way `abort()` does for a session. Caveats, for both entry points: @@ -654,6 +655,50 @@ Caveats, for both entry points: +
+Dictations longer than 2 minutes + +The live upload takes up to 120 seconds of audio. Past that, the dictation API +refuses it with a 413 (`audio_too_large`), and by default the SDK falls back to +the buffered endpoint, `POST /v1/transcribe`, which takes up to 20 minutes or +256 MB: + +1. From the first byte, the SDK keeps an in-memory copy of the audio it uploads. +2. When the live request is refused for its length, the SDK keeps reading your + audio until it ends: the stream or iterable ends, or you call `close()` / + `result()` on a session. +3. It posts the whole recording once, with the same config, and resolves with + that transcript exactly as it would a live one. + +Nothing changes in your code: `transcribeLive()` and `openLive()` work the same +for a 10 second note and a 15 minute one. Two trade-offs come with it: + +- **Latency.** After a fallback, the result waits for the whole recording to + upload and be transcribed, so a long dictation returns more slowly than a live + one. +- **Memory.** The copy costs about 32 KB per second of 16 kHz mono 16-bit PCM, + about 38 MB at 20 minutes. A copy that grows past 256 MB is dropped, and the + call throws a `DictationError` (413, `audio_too_large`). + +Only a 413 with `audio_too_large` falls back; every other failure is thrown as +it is. If the buffered request fails, its own error is thrown (a 413 past 20 +minutes, say). `abort()` or an abort `signal` drops everything with no fallback +request. `options.timeout` stops applying to the live request once the fallback +engages (the rest of the recording takes as long as it takes), and the buffered +request gets a fresh `timeout` of its own. + +To keep no copy and get the 413 instead, turn it off: + +```typescript +const result = await client.dictation.transcribeLive( + micStream, + { sample_rate: 16_000, channels: 1 }, + { fallbackToBuffered: false }, +); +``` + +
+
Rewrite the transcript with an LLM diff --git a/package.json b/package.json index f4507f2..dd645b4 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "assemblyai", - "version": "4.42.0", + "version": "4.43.0", "description": "The AssemblyAI JavaScript SDK provides an easy-to-use interface for interacting with the AssemblyAI API, which supports async and real-time transcription and text-to-speech.", "engines": { "node": ">=18" diff --git a/src/services/dictation/live.ts b/src/services/dictation/live.ts index 9428b2c..2490bc5 100644 --- a/src/services/dictation/live.ts +++ b/src/services/dictation/live.ts @@ -12,9 +12,11 @@ import { LiveSession } from "../../utils/live"; * iterables, and audio held whole, go to `client.dictation.transcribeLive()` * instead. * - * Everything `transcribeLive()` says about the 120 s audio cap, the - * server-side silence abort and errors surfacing mid-upload applies here - * unchanged. + * Everything `transcribeLive()` says about the 120 s audio cap and the + * buffered fallback past it, the server-side silence abort and errors + * surfacing mid-upload applies here unchanged. When the fallback engages, + * the session keeps taking writes until `close()` or `result()`, then sends + * the whole recording at once; `abort()` drops it with no fallback request. * * @example * ```ts diff --git a/src/services/dictation/recording.ts b/src/services/dictation/recording.ts new file mode 100644 index 0000000..514f258 --- /dev/null +++ b/src/services/dictation/recording.ts @@ -0,0 +1,187 @@ +import { DictationError } from "../../utils/errors/dictation"; + +/** + * The most audio the buffered fallback keeps in memory: the size cap of the + * buffered endpoint, `POST /v1/transcribe` (256 MB). A copy past it could + * never be accepted there, so it is dropped rather than grown further. + */ +export const maxBufferedAudioBytes = 256 * 1024 * 1024; + +/** + * The error a fallback throws when the recording outgrows + * `maxBufferedAudioBytes`: the same status and code the buffered endpoint + * itself answers with for audio over its size cap. + */ +function overflowError(): DictationError { + return new DictationError( + `audio exceeds ${maxBufferedAudioBytes} bytes (256 MB), the most the ` + + "buffered dictation fallback keeps", + 413, + "audio_too_large", + ); +} + +/** + * Rejects with the signal's reason once it aborts; never settles otherwise. + * The listener is removed by `dispose()`. + */ +function abortion(signal: AbortSignal): { + promise: Promise; + dispose: () => void; +} { + let dispose = () => undefined as void; + const promise = new Promise((_, reject) => { + const onAbort = () => + reject(signal.reason ?? new Error("The operation was aborted.")); + signal.addEventListener("abort", onAbort, { once: true }); + dispose = () => signal.removeEventListener("abort", onAbort); + }); + // Only ever observed through Promise.race; never an unhandled rejection. + promise.catch(() => undefined); + return { promise, dispose }; +} + +/** + * The caller's audio source, with a copy kept of every chunk it produces. + * + * The live upload reads the source through `chunks`; the copy holds the same + * bytes from the first one on, so that when the live request is refused for + * being too long the whole recording can be sent again in one buffered + * request. Only the audio is kept, never the multipart framing. + * + * The source has exactly one reader at a time: a chunk the live upload asked + * for but gave up waiting on (when its body is ended early) is still + * recorded, and is the first one `drain()` sees. + * @internal + */ +export class AudioRecording { + private readonly iterator: AsyncIterator; + private recorded: Uint8Array[] = []; + private recordedBytes = 0; + private pending: Promise> | undefined; + private ended = false; + private overflowed = false; + private released = false; + + /** + * The source as the live upload reads it. Ending this view early (the body + * does so when the server answers mid-upload) leaves the source itself + * open: whether it is released or drained is decided once the response is + * known, by `release()` or `drain()`. + */ + readonly chunks: AsyncIterable; + + /** + * @param source - The caller's audio, already coerced to bytes. + */ + constructor(source: AsyncIterable) { + this.iterator = source[Symbol.asyncIterator](); + this.chunks = { + [Symbol.asyncIterator]: () => ({ + next: () => this.next(), + return: async () => ({ done: true as const, value: undefined }), + }), + }; + } + + /** + * The next chunk from the source, recorded before it is handed out. Calls + * made while one is in flight share it, so no chunk is read twice or lost. + */ + private next(): Promise> { + if (this.ended) return Promise.resolve({ done: true, value: undefined }); + if (!this.pending) { + this.pending = this.iterator.next().then( + (result) => { + this.pending = undefined; + if (result.done) this.ended = true; + else this.record(result.value); + return result; + }, + (error: unknown) => { + this.pending = undefined; + this.ended = true; + throw error; + }, + ); + } + return this.pending; + } + + private record(chunk: Uint8Array): void { + if (chunk.byteLength === 0 || this.released) return; + this.recordedBytes += chunk.byteLength; + if (this.overflowed) return; + if (this.recordedBytes > maxBufferedAudioBytes) { + // Past the buffered endpoint's cap the copy is useless; let it go now + // rather than holding hundreds of megabytes until the request ends. + this.overflowed = true; + this.recorded = []; + return; + } + // Keep a private copy: a producer may reuse its buffer for the next chunk. + this.recorded.push(chunk.slice()); + } + + /** + * Read the source to its end, recording every chunk: the rest of the + * recording, after the live request was refused. Called once the live + * upload has stopped reading. + * @param signal - Aborts the drain; the source is released and the copy + * dropped. + * @returns A promise that resolves once the source has ended. + * @throws DictationError (413, `audio_too_large`) when the copy passes + * `maxBufferedAudioBytes`. + * @throws Whatever the source throws, unchanged, or the signal's reason. + */ + async drain(signal?: AbortSignal): Promise { + const aborted = signal ? abortion(signal) : undefined; + try { + for (;;) { + if (signal?.aborted) { + throw signal.reason ?? new Error("The operation was aborted."); + } + if (this.overflowed) throw overflowError(); + const next = this.next(); + const result = aborted + ? await Promise.race([next, aborted.promise]) + : await next; + if (result.done) break; + } + // An abort can end the source (a session's abort() ends its queue) + // in the same breath as it fires the signal; it is still an abort. + if (signal?.aborted) { + throw signal.reason ?? new Error("The operation was aborted."); + } + if (this.overflowed) throw overflowError(); + } catch (error) { + this.release(); + throw error; + } finally { + aborted?.dispose(); + } + } + + /** + * Hand over the recorded audio, in order, and forget it here. + * @returns The recorded chunks. + * @throws DictationError (413, `audio_too_large`) when the copy passed + * `maxBufferedAudioBytes`. + */ + take(): Uint8Array[] { + if (this.overflowed) throw overflowError(); + const recorded = this.recorded; + this.recorded = []; + return recorded; + } + + /** + * Let go of the source and drop the copy. A source still producing is + * stopped, as the live upload would have stopped it. + */ + release(): void { + this.released = true; + this.recorded = []; + void this.iterator.return?.()?.catch(() => undefined); + } +} diff --git a/src/services/dictation/service.ts b/src/services/dictation/service.ts index 53eb229..159ca9d 100644 --- a/src/services/dictation/service.ts +++ b/src/services/dictation/service.ts @@ -11,6 +11,7 @@ import { DictationError } from "../../utils/errors/dictation"; import { getPath } from "../../utils/path"; import { basename, + concatChunks, dataUrlToBytes, deadlineSignal, describe, @@ -27,14 +28,20 @@ import { resolveFormat, } from "../../utils/live"; import { DictationLiveSession } from "./live"; +import { AudioRecording } from "./recording"; -// The one endpoint this service posts audio to; the dictation API serves it -// under /v1 only. +// The endpoint this service posts audio to; the dictation API serves it under +// /v1 only. It caps a request at 120 s of audio. const transcribeLiveEndpoint = "/v1/transcribe/live"; +// The buffered endpoint: the whole recording in one body, up to 20 minutes or +// 256 MB. Used only by the fallback, once the live endpoint has refused a +// dictation for running past its cap. +const transcribeBufferedEndpoint = "/v1/transcribe"; const warmEndpoint = "/v1/warm"; // A fetch deadline spans the whole request, so this covers the upload, the // transcription and the LLM pass together. It matches the service's own -// total request budget; the audio itself is capped at 120 s. +// total request budget; the live audio itself is capped at 120 s. A buffered +// fallback request gets a fresh deadline of the same length. const defaultLiveTimeoutMs = 300_000; const warmTimeoutMs = 10_000; // Caps mirror the dictation service's `config` part; a field over its cap is @@ -72,7 +79,10 @@ const contentTypes: Record = { * `transcribeLive()`, so the upload and every speech segment but the last are * done by the time the speaker stops. Audio already held whole — bytes, a * Blob, a local path — goes to `transcribeLive()` too and travels the same - * connection as a single chunk: there is one request shape here. + * connection as a single chunk: there is one request shape here. The one + * exception is a dictation that runs past the live endpoint's 120 s cap, + * which by default is sent again whole to the buffered endpoint (see + * `DictationLiveOptions.fallbackToBuffered`). * * Unlike `client.transcripts`, which submits a job to the async API and polls * for completion, there is no job id, no status to poll and no URL ingestion. @@ -103,16 +113,25 @@ export class DictationTranscriber extends BaseService { * * A producer must keep producing: an upload that goes silent for long * enough is aborted server-side, so stop by ending the stream rather than - * pausing it. The service caps a request at 120 s of audio. For sources - * that deliver audio through a callback rather than a stream, see - * `openLive()`. + * pausing it. For sources that deliver audio through a callback rather than + * a stream, see `openLive()`. + * + * The live endpoint caps a request at 120 s of audio. Past it, by default, + * the SDK falls back to the buffered endpoint: it keeps reading the source + * until it ends, then posts the whole recording once to `/v1/transcribe` + * (up to 20 minutes or 256 MB) and resolves with that result. A fallback + * result waits for the whole recording to upload and be transcribed, so a + * long dictation returns more slowly than a live one. Set + * `options.fallbackToBuffered` to `false` to get the 413 instead; see + * `DictationLiveOptions.fallbackToBuffered` for the memory the copy costs. * @param audio - Audio still being produced (an async iterable, Node * streams included; a sync iterable; a web `ReadableStream`) or audio held * whole (a local file path, a data URL, raw bytes, a Blob). Raw PCM also * requires `sample_rate` and `channels` on the config. * @param config - Options for this dictation request. - * @param options - Client-side options: the total request deadline and an - * optional abort signal. + * @param options - Client-side options: the request deadline, an optional + * abort signal, and whether to fall back to the buffered endpoint past the + * live cap. * @returns A promise that resolves to the finished transcript, whose * `final_text` carries the LLM rewrite when one was asked for. * @throws Error when `audio` is a URL, or when raw PCM is missing @@ -120,8 +139,10 @@ export class DictationTranscriber extends BaseService { * @throws TypeError when `audio` is of an unsupported type, or when a * produced chunk is not bytes. * @throws DictationError when the request fails. Auth, rate-limit, size and - * capacity failures can surface part-way through the upload. Anything the - * producer throws propagates unchanged; the connection is dropped. + * capacity failures can surface part-way through the upload. After a + * fallback, the error is the buffered request's own (a 413 past 20 minutes + * or 256 MB, say). Anything the producer throws propagates unchanged; the + * connection is dropped. * @example * ```ts * const result = await client.dictation.transcribeLive(mic, { @@ -145,16 +166,12 @@ export class DictationTranscriber extends BaseService { contentTypes, configName, ); - const signal = deadlineSignal( - options.timeout ?? defaultLiveTimeoutMs, - options.signal, - ); - return await this.postLive( + return await this.transcribe( source.chunks, filename, contentType, config, - signal, + options, ); } @@ -166,10 +183,19 @@ export class DictationTranscriber extends BaseService { * immediately; call `session.write(chunk)` from the callback, then * `await session.result()` for the transcript once the speaker stops. See * `DictationLiveSession`. + * + * Past the live endpoint's 120 s cap the session falls back to the + * buffered endpoint, as `transcribeLive()` does: it keeps taking writes + * until `close()` or `result()`, then posts the whole recording once, and + * `result()` resolves with that transcript — more slowly than a live one, + * since it waits for the whole recording to upload and be transcribed. + * `abort()` still drops everything, with no fallback request. * @param config - Options for this dictation request. Raw PCM requires * `sample_rate` and `channels`. - * @param options - Client-side options: the total request deadline, which - * must cover the dictation. + * @param options - Client-side options: the request deadline, which must + * cover the live part of the dictation, and whether to fall back to the + * buffered endpoint past the live cap. The session manages its own abort + * signal; use `session.abort()`. * @returns The open session. * @example * ```ts @@ -193,14 +219,70 @@ export class DictationTranscriber extends BaseService { contentTypes, configName, ); + return await this.transcribe(chunks, filename, contentType, config, { + timeout: options.timeout, + fallbackToBuffered: options.fallbackToBuffered, + signal: abortSignal, + }); + }); + } + + /** + * Run one dictation: the live request, and the buffered fallback when the + * live endpoint refuses the audio for running past its cap. + * + * With the fallback on, a copy of every audio byte the live upload reads is + * kept. When the live request fails with 413 `audio_too_large` — mid-upload + * or after the body was sent — the source is read on to its end (that is + * still recording time, so the live deadline no longer applies; the abort + * signal does), and the whole copy is posted once to the buffered + * endpoint under a fresh deadline. Every other outcome is the live + * request's own, and with the fallback off no copy is kept at all. + */ + private async transcribe( + chunks: AsyncIterable, + filename: string, + contentType: string, + config: DictationConfig, + options: DictationLiveOptions, + ): Promise { + const configJson = buildConfigJson(config); + const timeoutMs = options.timeout ?? defaultLiveTimeoutMs; + const format = { configJson, filename, contentType }; + if (options.fallbackToBuffered === false) { return await this.postLive( chunks, - filename, - contentType, - config, - deadlineSignal(options.timeout ?? defaultLiveTimeoutMs, abortSignal), + format, + deadlineSignal(timeoutMs, options.signal), ); - }); + } + + const recording = new AudioRecording(chunks); + try { + // A 200 arrives once the whole body was sent, so the source has ended. + return await this.postLive( + recording.chunks, + format, + deadlineSignal(timeoutMs, options.signal), + ); + } catch (error) { + if (!isAudioTooLarge(error)) { + // Exactly what the live request does on its own: stop the producer + // and report the failure. + recording.release(); + throw error; + } + } + + // The live endpoint refused the dictation for its length. The live + // upload has stopped reading; read the rest of the recording here, up to + // key-up, then send all of it at once. + await recording.drain(options.signal); + return await this.postBuffered( + recording.take(), + format, + deadlineSignal(timeoutMs, options.signal), + ); } /** @@ -212,18 +294,16 @@ export class DictationTranscriber extends BaseService { */ private async postLive( chunks: AsyncIterable, - filename: string, - contentType: string, - config: DictationConfig, + format: AudioFormat, signal: AbortSignal, ): Promise { const boundary = multipartBoundary(); const body = liveBody({ head: multipartHead({ boundary, - config: buildConfigJson(config), - filename, - contentType, + config: format.configJson, + filename: format.filename, + contentType: format.contentType, }), chunks, closing: multipartClosing(boundary), @@ -251,18 +331,48 @@ export class DictationTranscriber extends BaseService { // The server can reject while the upload is still in flight (auth, // rate limit, size, capacity). fetch resolves with the status at once // but withholds the response body until the request body ends, so end - // it now rather than streaming the rest of a rejected dictation. + // it now rather than streaming the rest of a rejected dictation. With + // the buffered fallback on, this only stops the upload: the source + // stays open, for the fallback to read to its end. body.finish(); - throw await errorFromResponse( - response, - DictationError, - "dictation transcription", - ); } - const json = (await response.json()) as DictationResponse; - // The LLM rewrite is the text to show when there is one; a pass that was - // never asked for, or that failed, leaves `llm_response` null. - return { ...json, final_text: json.llm_response ?? json.text }; + return await dictationResult(response); + } + + /** + * Post a whole recording to the buffered endpoint in one request: the same + * `config` part and audio part as the live request, with the audio already + * complete. + * @param audio - The recorded chunks, in order. Emptied here once they are + * copied into the body, so the recording is held once, not twice, while + * the request is in flight. + */ + private async postBuffered( + audio: Uint8Array[], + format: AudioFormat, + signal: AbortSignal, + ): Promise { + const boundary = multipartBoundary(); + const body = concatChunks([ + multipartHead({ + boundary, + config: format.configJson, + filename: format.filename, + contentType: format.contentType, + }), + ...audio, + multipartClosing(boundary), + ]); + audio.length = 0; + const response = await this.fetchResponse(transcribeBufferedEndpoint, { + method: "POST", + body, + headers: { + "Content-Type": `multipart/form-data; boundary=${boundary}`, + }, + signal, + }); + return await dictationResult(response); } /** @@ -293,6 +403,49 @@ export class DictationTranscriber extends BaseService { } } +/** What a request's `config` and `audio` parts carry besides the audio. */ +type AudioFormat = { + configJson: Record; + filename: string; + contentType: string; +}; + +/** + * Read a dictation response: the transcript on a 200, a `DictationError` + * otherwise. Live and buffered requests answer in the same shape. + * @param response - The response; its body is read here. + * @returns The transcript, with `final_text` derived. + * @throws DictationError when the status is not 200. + */ +async function dictationResult(response: Response): Promise { + if (response.status !== 200) { + throw await errorFromResponse( + response, + DictationError, + "dictation transcription", + ); + } + const json = (await response.json()) as DictationResponse; + // The LLM rewrite is the text to show when there is one; a pass that was + // never asked for, or that failed, leaves `llm_response` null. + return { ...json, final_text: json.llm_response ?? json.text }; +} + +/** + * Whether a failed live request is the one the buffered fallback answers: + * the live endpoint's 413 for audio past its 120 s cap. Any other 413 (a + * body too large in bytes, say) and every other failure is not. + * @param error - What the live request threw. + * @returns `true` for a 413 with error code `audio_too_large`. + */ +function isAudioTooLarge(error: unknown): boolean { + return ( + error instanceof DictationError && + error.status === 413 && + error.errorCode === "audio_too_large" + ); +} + type ResolvedSource = { chunks: AsyncIterable; filename?: string; diff --git a/src/types/dictation/index.ts b/src/types/dictation/index.ts index 3607d6c..2af808b 100644 --- a/src/types/dictation/index.ts +++ b/src/types/dictation/index.ts @@ -99,15 +99,46 @@ export type DictationLiveOptions = { * The total request deadline in milliseconds, measured from the start of * the request: it spans the upload, the transcription and the LLM pass. * Defaults to 300 000, matching the service's total request budget. The - * audio itself is capped at 120 seconds. + * live audio itself is capped at 120 seconds. + * + * When the buffered fallback engages, this deadline stops applying to the + * live request: the rest of the recording is read for as long as it lasts + * (that is recording time, not request time), and the buffered request + * then gets a fresh deadline of the same length. `signal` applies + * throughout. */ timeout?: number; /** * An `AbortSignal` that drops the request when aborted. `openLive()` * sessions manage their own; pass one here to cancel a `transcribeLive()` - * call from outside. + * call from outside. Aborting while a fallback is reading the recording or + * posting it drops everything; no fallback request is sent after an abort. */ signal?: AbortSignal; + /** + * Fall back to the buffered endpoint when a dictation runs past the live + * endpoint's 120 s cap. Defaults to `true`. + * + * When the live request fails with 413 `audio_too_large`, the SDK keeps + * reading the audio until it ends (the stream or iterable ends, or + * `close()` / `result()` on a session), then posts the whole recording + * once to `POST /v1/transcribe`, which takes up to 20 minutes or 256 MB, + * and returns that result exactly as a live one. If that request fails, + * its own error is thrown — a 413 past 20 minutes or 256 MB, say — not the + * live 413. Every other live failure is thrown as it is. + * + * The trade-offs: after a fallback the result waits for the whole + * recording to upload and be transcribed, so a long dictation returns more + * slowly than a live one. And to have the recording to send, the SDK keeps + * an in-memory copy of every audio byte from the first one on, about 32 KB + * per second of 16 kHz mono 16-bit PCM (about 38 MB at 20 minutes). A copy + * that grows past 256 MB is dropped and the call throws a `DictationError` + * (413, `audio_too_large`). + * + * Set to `false` to keep no copy and get the live 413 instead, exactly as + * before the fallback existed. + */ + fallbackToBuffered?: boolean; }; /** diff --git a/tests/unit/dictation.test.ts b/tests/unit/dictation.test.ts index 018156b..635e2fc 100644 --- a/tests/unit/dictation.test.ts +++ b/tests/unit/dictation.test.ts @@ -1,4 +1,10 @@ -import { createReadStream, mkdtempSync, rmSync, writeFileSync } from "fs"; +import { + createReadStream, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from "fs"; import fetchMock from "jest-fetch-mock"; import { tmpdir } from "os"; import path from "path"; @@ -625,8 +631,10 @@ describe("dictation errors", () => { title: "Audio Too Large", detail: "the audio exceeds the size limit", }); + // This 413 is the one the buffered fallback answers; switch it off to see + // the live error itself. const error = await assembly.dictation - .transcribeLive(chunks(bytes("RIFF"))) + .transcribeLive(chunks(bytes("RIFF")), {}, { fallbackToBuffered: false }) .catch((e) => e); expect(error).toBeInstanceOf(DictationError); expect(error.errorCode).toBe("audio_too_large"); @@ -960,7 +968,7 @@ describe("dictation session", () => { }); describe("dictation routing", () => { - it("should never request the buffered transcription endpoint", async () => { + it("should not request the buffered endpoint when the live one succeeds", async () => { mockOk(); mockOk(); await assembly.dictation.transcribeLive( @@ -973,3 +981,460 @@ describe("dictation routing", () => { } }); }); + +describe("dictation buffered fallback", () => { + const bufferedUrl = "/v1/transcribe"; + // The body production returns for a dictation past the live 120 s cap. + const tooLong = { + error: "audio exceeds 120000 ms", + error_code: "audio_too_large", + }; + const pcm = { sample_rate: 16000, channels: 1 }; + const sleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + + function deferred() { + let resolve: () => void = () => undefined; + const promise = new Promise((r) => (resolve = r)); + return { promise, resolve }; + } + + /** + * Reads `count` pieces of the streamed live body (the framing head is the + * first), as a server part-way through the upload would, then lets go. + */ + async function readSome(count: number, index = 0): Promise { + const reader = ( + requestInit(index).body as ReadableStream + ).getReader(); + for (let i = 0; i < count; i++) { + if ((await reader.read()).done) break; + } + reader.releaseLock(); + } + + /** + * The live endpoint refusing the dictation for its length: after reading + * `count` pieces of the body (mid-upload), or all of it when unset. + */ + function mockTooLong(count?: number, onRefuse?: () => void) { + fetchMock.doMockOnceIf( + requestMatches({ url: liveUrl, method: "POST" }), + async () => { + if (count === undefined) await requestBodyBytes(); + else await readSome(count); + onRefuse?.(); + return { body: JSON.stringify(tooLong), status: 413 }; + }, + ); + } + + function mockBuffered(body: object = okResponse, status = 200, headers = {}) { + fetchMock.doMockOnceIf( + requestMatches({ url: bufferedUrl, method: "POST" }), + JSON.stringify(body), + { status, headers }, + ); + } + + /** The parts of the buffered request, whose body is sent whole. */ + function bufferedParts(index = 1): Part[] { + return parseParts(requestInit(index).body as Uint8Array, index); + } + + const urls = () => fetchMock.mock.calls.map(([url]) => url); + + /** + * A recording that is still going: `count` two-byte chunks `[i, i]`, one + * every `delayMs`. Tracks how much it produced and whether it was let go. + */ + function microphone(count: number, delayMs = 2) { + const state = { + produced: 0, + producedAfterRefusal: 0, + refused: false, + finished: false, + }; + async function* source() { + try { + for (let i = 0; i < count; i++) { + await sleep(delayMs); + state.produced++; + if (state.refused) state.producedAfterRefusal++; + yield new Uint8Array([i, i]); + } + } finally { + state.finished = true; + } + } + return { state, source: source() }; + } + + const audioOf = (count: number) => + Array.from({ length: count }, (_, i) => [i, i]).flat(); + + it("should fall back mid-upload and send all the audio, including what was produced after the 413", async () => { + const mic = microphone(20); + mockTooLong(4, () => (mic.state.refused = true)); + mockBuffered({ ...okResponse, llm_response: "Take 2 tablets daily." }); + + const result = await assembly.dictation.transcribeLive(mic.source, { + ...pcm, + llm_instruction: "Clean this up.", + }); + + expect(urls()).toEqual([ + defaultBaseUrl + liveUrl, + defaultBaseUrl + bufferedUrl, + ]); + // Returned exactly like a live result, final_text included. + expect(result.text).toBe("take two tablets daily"); + expect(result.final_text).toBe("Take 2 tablets daily."); + expect(result.session_id).toBe(okResponse.session_id); + + // The recording went on after the 413, and the buffered request carries + // every byte of it, from the first chunk to the last, once each. + expect(mic.state.producedAfterRefusal).toBeGreaterThan(0); + expect(mic.state.produced).toBe(20); + const parts = bufferedParts(); + expect(parts.map((p) => p.name)).toEqual(["config", "audio"]); + expect(JSON.parse(decode(parts[0].body))).toEqual({ + ...pcm, + llm_instruction: "Clean this up.", + }); + expect(parts[0].type).toBe("application/json"); + expect(parts[1].filename).toBe("audio.pcm"); + expect(parts[1].type).toBe("audio/pcm"); + expect(Array.from(parts[1].body)).toEqual(audioOf(20)); + expect(requestHeaders(1)["Content-Type"]).toMatch( + /^multipart\/form-data; boundary=----assemblyai-[0-9a-z]{32}$/, + ); + expect(requestHeaders(1)["Authorization"]).toBe(defaultApiKey); + expect(requestInit(1).body).toBeInstanceOf(Uint8Array); + + // The live upload itself was ended at the 413, not streamed on. + const live = decode(await requestBodyBytes(0)); + expect(live.trimEnd().endsWith("--")).toBe(true); + }); + + it("should fall back when the 413 arrives after the whole body was sent", async () => { + mockTooLong(); + mockBuffered(); + const filePath = path.join(testDir, "gore-short.wav"); + const result = await assembly.dictation.transcribeLive( + createReadStream(filePath), + ); + + expect(result.final_text).toBe("take two tablets daily"); + expect(urls()).toEqual([ + defaultBaseUrl + liveUrl, + defaultBaseUrl + bufferedUrl, + ]); + const parts = bufferedParts(); + expect(decode(parts[0].body)).toBe("{}"); + // The same file name and content type as the live request. + expect(parts[1].filename).toBe("gore-short.wav"); + expect(parts[1].type).toBe("audio/wav"); + // The whole file, byte for byte (parseParts is not binary-exact, so look + // for it in the raw body). + const file = readFileSync(filePath); + const raw = Buffer.from(requestInit(1).body as Uint8Array); + expect(parts[1].body.length).toBe(file.length); + expect(raw.indexOf(file)).toBeGreaterThan(0); + }); + + it("should fall back for audio held whole, sent as one chunk", async () => { + mockTooLong(); + mockBuffered(); + await assembly.dictation.transcribeLive(bytes("\x00\x01\x02\x03"), pcm); + const parts = bufferedParts(); + expect(Array.from(parts[1].body)).toEqual([0, 1, 2, 3]); + expect(JSON.parse(decode(parts[0].body))).toEqual(pcm); + }); + + it("should throw the live 413 as before when fallbackToBuffered is false", async () => { + const mic = microphone(200); + mockTooLong(3); + const error = await assembly.dictation + .transcribeLive(mic.source, pcm, { fallbackToBuffered: false }) + .catch((e) => e); + + expect(error).toBeInstanceOf(DictationError); + expect(error.status).toBe(413); + expect(error.errorCode).toBe("audio_too_large"); + expect(error.message).toBe("audio exceeds 120000 ms"); + expect(fetchMock.mock.calls).toHaveLength(1); + // The producer was let go of, as before. + await sleep(10); + expect(mic.state.finished).toBe(true); + expect(mic.state.produced).toBeLessThan(200); + }); + + it.each([ + [ + "another 413 code", + 413, + { error: "request body too large", error_code: "payload_too_large" }, + ], + [ + "a 413 problem-details body for another reason", + 413, + { status: 413, title: "Payload Too Large", detail: "too many bytes" }, + ], + ["a 429", 429, { error: "slow down", error_code: "rate_limited" }], + ["a 500", 500, { error: "boom", error_code: "internal_error" }], + ["a 401", 401, { error: "Invalid API key", error_code: "unauthorized" }], + ])("should not fall back on %s", async (_, status, body) => { + const mic = microphone(200); + fetchMock.doMockOnceIf( + requestMatches({ url: liveUrl, method: "POST" }), + async () => { + await readSome(3); + return { body: JSON.stringify(body), status }; + }, + ); + const error = await assembly.dictation + .transcribeLive(mic.source, pcm) + .catch((e) => e); + + expect(error).toBeInstanceOf(DictationError); + expect(error.status).toBe(status); + expect(fetchMock.mock.calls).toHaveLength(1); + await sleep(10); + expect(mic.state.finished).toBe(true); + }); + + it("should not fall back when the live request fails without a response", async () => { + fetchMock.mockRejectOnce(new TypeError("fetch failed")); + const error = await assembly.dictation + .transcribeLive(chunks(bytes("RIFF"))) + .catch((e) => e); + expect(error).toBeInstanceOf(TypeError); + expect(fetchMock.mock.calls).toHaveLength(1); + }); + + it("should throw the buffered request's own error when the fallback fails", async () => { + mockTooLong(); + mockBuffered( + { error: "audio exceeds 1200000 ms", error_code: "audio_too_large" }, + 413, + ); + const error = await assembly.dictation + .transcribeLive(chunks(bytes("\x00\x00")), pcm) + .catch((e) => e); + + expect(error).toBeInstanceOf(DictationError); + expect(error.status).toBe(413); + expect(error.message).toBe("audio exceeds 1200000 ms"); + // One fallback, not a loop. + expect(fetchMock.mock.calls).toHaveLength(2); + }); + + it("should throw a failed fallback's status and Retry-After", async () => { + mockTooLong(); + mockBuffered( + { status: 503, title: "Capacity Exceeded", detail: "no capacity" }, + 503, + { "Retry-After": "7" }, + ); + const error = await assembly.dictation + .transcribeLive(chunks(bytes("\x00\x00")), pcm) + .catch((e) => e); + expect(error).toBeInstanceOf(DictationError); + expect(error.status).toBe(503); + expect(error.errorCode).toBe("capacity_exceeded"); + expect(error.retryAfter).toBe(7); + }); + + it("should rethrow a producer error while draining, with no fallback request", async () => { + mockTooLong(2); + async function* failing() { + yield bytes("\x00\x00"); + await sleep(10); + yield bytes("\x00\x00"); + throw new Error("microphone unplugged"); + } + await expect( + assembly.dictation.transcribeLive(failing(), pcm), + ).rejects.toThrow("microphone unplugged"); + expect(fetchMock.mock.calls).toHaveLength(1); + }); + + it("should make no fallback request after an abort signal while draining", async () => { + const controller = new AbortController(); + const refused = deferred(); + const mic = microphone(10_000); + mockTooLong(2, refused.resolve); + const pending = assembly.dictation + .transcribeLive(mic.source, pcm, { signal: controller.signal }) + .catch((e) => e); + + await refused.promise; + await sleep(20); // the fallback is reading the rest of the recording + expect(mic.state.finished).toBe(false); + controller.abort(); + + const error = await pending; + expect(String(error)).toMatch(/abort/i); + expect(fetchMock.mock.calls).toHaveLength(1); + await sleep(10); + expect(mic.state.finished).toBe(true); // the producer was let go of + }); + + it("should drop the copy and throw past 256 MB, with no fallback request", async () => { + mockTooLong(2); + // 17 x 16 MB: the copy passes 256 MB on the last chunk. + const block = new Uint8Array(16 * 1024 * 1024); + let yielded = 0; + let finished = false; + async function* huge() { + try { + for (let i = 0; i < 100; i++) { + yielded++; + yield block; + } + } finally { + finished = true; + } + } + const error = await assembly.dictation + .transcribeLive(huge(), pcm) + .catch((e) => e); + + expect(error).toBeInstanceOf(DictationError); + expect(error.status).toBe(413); + expect(error.errorCode).toBe("audio_too_large"); + expect(error.message).toContain("256 MB"); + expect(fetchMock.mock.calls).toHaveLength(1); + expect(yielded).toBe(17); // stopped reading at the cap + await sleep(5); + expect(finished).toBe(true); + }, 10_000); + + it("should not let the live deadline cut off the rest of the recording", async () => { + // ~200 ms of recording against a 50 ms deadline: the 413 comes early, + // and draining the rest is recording time, not request time. + const mic = microphone(40, 5); + mockTooLong(2); + mockBuffered(); + const result = await assembly.dictation.transcribeLive(mic.source, pcm, { + timeout: 50, + }); + expect(result.final_text).toBe("take two tablets daily"); + expect(Array.from(bufferedParts()[1].body)).toEqual(audioOf(40)); + }); + + it("should give the buffered request a fresh deadline", async () => { + const mic = microphone(20, 5); // ~100 ms, past the 60 ms deadline + mockTooLong(2); + let startedAt = 0; + let abortedAt = 0; + fetchMock.doMockOnceIf( + requestMatches({ url: bufferedUrl, method: "POST" }), + async (request) => { + startedAt = Date.now(); + await new Promise((resolve) => { + request.signal.addEventListener("abort", () => resolve(), { + once: true, + }); + }); + abortedAt = Date.now(); + throw Object.assign(new Error("The operation was aborted."), { + name: "AbortError", + }); + }, + ); + const startedCall = Date.now(); + await expect( + assembly.dictation.transcribeLive(mic.source, pcm, { timeout: 60 }), + ).rejects.toThrow("aborted"); + // The buffered request started after the live deadline had passed, and + // still had its own deadline to run against. + expect(startedAt - startedCall).toBeGreaterThanOrEqual(60); + expect(abortedAt - startedAt).toBeGreaterThanOrEqual(50); + }); + + describe("with openLive()", () => { + it("should keep taking writes after the 413 and send them all on result()", async () => { + const refused = deferred(); + mockTooLong(3, refused.resolve); + mockBuffered(); + const session = assembly.dictation.openLive(pcm); + session.write(new Uint8Array([0, 0])); + session.write(new Uint8Array([1, 1])); + + await refused.promise; + await sleep(5); + // Still recording: the session is open and takes more audio. + expect(session.closed).toBe(false); + session.write(new Uint8Array([2, 2])); + await sleep(5); + session.write(new Uint8Array([3, 3]).buffer); + + const result = await session.result(); + expect(result.final_text).toBe("take two tablets daily"); + expect(urls()).toEqual([ + defaultBaseUrl + liveUrl, + defaultBaseUrl + bufferedUrl, + ]); + const parts = bufferedParts(); + expect(JSON.parse(decode(parts[0].body))).toEqual(pcm); + expect(parts[1].filename).toBe("audio.pcm"); + expect(parts[1].type).toBe("audio/pcm"); + expect(Array.from(parts[1].body)).toEqual([0, 0, 1, 1, 2, 2, 3, 3]); + }); + + it("should fall back after close() when the 413 comes at the end", async () => { + mockTooLong(); + mockBuffered(); + const session = assembly.dictation.openLive(pcm); + session.write(new Uint8Array([0, 0])); + session.close(); + await session.result(); + expect(Array.from(bufferedParts()[1].body)).toEqual([0, 0]); + }); + + it("should make no fallback request after abort()", async () => { + const refused = deferred(); + mockTooLong(2, refused.resolve); + const session = assembly.dictation.openLive(pcm); + session.write(new Uint8Array([0, 0])); + await refused.promise; + await sleep(5); + session.write(new Uint8Array([1, 1])); + + await session.abort(); + expect(session.closed).toBe(true); + await expect(session.result()).rejects.toThrow("aborted"); + await sleep(10); + expect(fetchMock.mock.calls).toHaveLength(1); + }); + + it("should reject with the live 413 when fallbackToBuffered is false", async () => { + mockTooLong(2); + const session = assembly.dictation.openLive(pcm, { + fallbackToBuffered: false, + }); + session.write(new Uint8Array([0, 0])); + const error = await session.result().catch((e) => e); + expect(error).toBeInstanceOf(DictationError); + expect(error.status).toBe(413); + expect(error.errorCode).toBe("audio_too_large"); + expect(fetchMock.mock.calls).toHaveLength(1); + }); + + it("should not let the live deadline cut off a session still recording", async () => { + const refused = deferred(); + mockTooLong(2, refused.resolve); + mockBuffered(); + const session = assembly.dictation.openLive(pcm, { timeout: 30 }); + session.write(new Uint8Array([0, 0])); + await refused.promise; + await sleep(60); // past the live deadline, still recording + session.write(new Uint8Array([1, 1])); + const result = await session.result(); + expect(result.final_text).toBe("take two tablets daily"); + expect(Array.from(bufferedParts()[1].body)).toEqual([0, 0, 1, 1]); + }); + }); +});