diff --git a/CLAUDE.md b/CLAUDE.md
index ce03621..d7a652c 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -48,7 +48,7 @@ aai.settings.api_key = "your-key"
 - `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines; `open_live()` returns an `AsyncLiveSession`. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one
 - `aai.SyncTranscriptionConfig` — Sync options: `model` (default `universal-3-5-pro`), `prompt`, `keyterms_prompt`, `conversation_context`, `language_codes`, `timestamps`, `sample_rate`, `channels`
 - `aai.SyncTranscriptResponse` — Sync result: `.text`, `.words` (`SyncWord` with `confidence` always, `start`/`end` only when `timestamps=True`), `.confidence`, `.audio_duration_ms`, `.session_id`, `.request_time_ms`
-- `aai.DictationTranscriber` — Dictation API: audio uploaded as it is spoken over one live request, transcript (and optional LLM pass) out. Methods: `open_live()` (push-style `DictationLiveSession`: `write()`/`close()`/`result()`/`abort()`), `transcribe_live()` (iterable, file object, bytes or path), `warm()`. There is no buffered `transcribe()`: every entry point posts to `/v1/transcribe/live`
+- `aai.DictationTranscriber` — Dictation API: audio uploaded as it is spoken over one live request, transcript (and optional LLM pass) out. Methods: `open_live()` (push-style `DictationLiveSession`: `write()`/`close()`/`result()`/`abort()`), `transcribe_live()` (iterable, file object, bytes or path), `warm()`. There is no buffered `transcribe()`: every entry point starts on `/v1/transcribe/live`, which caps audio at 120 s. Past the cap (a 413 `audio_too_large`), `fallback_to_buffered=True` (the default) reads the source to its end and posts the whole recording once to `/v1/transcribe` (up to 20 min / 256 MB): slower result, and an in-memory copy of the audio (~32 KB/s at 16 kHz mono). `fallback_to_buffered=False` keeps no copy and raises the 413
 - `aai.AsyncDictationTranscriber` — Asyncio counterpart of `DictationTranscriber`; `open_live()` returns an `AsyncDictationLiveSession`. Owns an HTTP pool: use `async with` or `await aclose()`
 - `aai.DictationConfig` — Dictation options: `sample_rate`, `channels`, `language_codes`, `keyterms_prompt`, `llm_instruction`, `rewrite_mode` (`aai.DictationRewriteMode.fast` default | `smart`). Rejects unknown fields
 - `aai.DictationResponse` — Dictation result: `.text`, `.words` (`DictationWord` with `text` + `confidence`), `.confidence`, `.llm_response`, `.llm_error`, `.audio_duration_ms`, `.session_id`, `.request_time_ms`, `.sync_time_ms`, and `.final_text` (the LLM rewrite, falling back to `.text`)
diff --git a/README.md b/README.md
index 525f5d6..871ffc2 100644
--- a/README.md
+++ b/README.md
@@ -682,7 +682,7 @@ except aai.SyncTranscriptError as error:
 
 ### **Dictation Examples**
 
-`aai.DictationTranscriber` is for dictation: a person speaks a short note and gets it back as text, optionally cleaned up or reformatted by an LLM. It has one connection, the live upload — audio is sent as it is spoken, the service transcribes each speech segment as it lands, and when the speaker stops what remains is the final segment and the LLM pass. There is no job id and no polling. Audio must be WAV or raw 16-bit PCM, up to 120 s per request.
+`aai.DictationTranscriber` is for dictation: a person speaks a short note and gets it back as text, optionally cleaned up or reformatted by an LLM. Every dictation starts on the live upload — audio is sent as it is spoken, the service transcribes each speech segment as it lands, and when the speaker stops what remains is the final segment and the LLM pass. There is no job id and no polling. Audio must be WAV or raw 16-bit PCM. The live upload takes up to 120 s; a longer dictation, up to 20 minutes, falls back to the buffered endpoint on its own (see below).
 
 The result is a `DictationResponse`: `text` is the raw transcript, `llm_response` the rewritten text when the LLM pass ran, and `final_text` is the one to show the user — the rewrite when there is one, the raw transcript otherwise.
 
@@ -716,7 +716,7 @@ Leaving the `with` block calls `close()`. If the block raises, the session is ab
 
 This is **not** the streaming API. A session returns one finished transcript when the audio ends; use `aai.RealTimeTranscriber` when you need words back while the speaker is still talking.
 
-The server aborts an upload that goes silent for long, so keep writing until the speaker is done rather than pausing. Errors that would normally arrive at the end (bad key, rate limit, capacity, audio too large) can surface part-way through the upload; `result()` raises them as `DictationError`.
+The server aborts an upload that goes silent for long, so keep writing until the speaker is done rather than pausing. Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload; `result()` raises them as `DictationError`. Audio past the 120 s live cap is not an error by default: see "Dictate for longer than 2 minutes" below.
 
 </details>
 
@@ -869,6 +869,29 @@ The transcriber owns an HTTP connection pool: use `async with`, or call `await t
 
 </details>
 
+<details>
+  <summary>Dictate for longer than 2 minutes</summary>
+
+The live upload is capped at 120 s of audio. Past it the server answers `413` with `error_code="audio_too_large"`, and by default the transcriber falls back to the buffered endpoint (`/v1/transcribe`, up to 20 minutes / 256 MB) for you: it keeps reading your audio source until it ends — the iterator is exhausted, or the session is closed — then posts the whole recording once, with the same config, and returns that result like a live one. Your code does not change.
+
+The trade-offs:
+
+- **Slower result.** 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, where only the final segment is left once the speaker stops.
+- **Memory.** To have the recording to resend, the transcriber keeps an in-memory copy of all the audio it uploads: about 32 KB per second at 16 kHz mono, so about 38 MB for 20 minutes. Past 256 MB the copy is dropped and a `413` `DictationError` is raised.
+
+Only a `413` `audio_too_large` from the live endpoint falls back; every other error is raised as before. If the buffered request fails too (past 20 minutes, say), its own error is raised. `abort()`, or an exception from your producer, drops the dictation with no fallback request.
+
+To keep dictations live-only and hold no copy of the audio, turn it off:
+
+```python
+transcriber = aai.DictationTranscriber(fallback_to_buffered=False)
+# a dictation past 120 s now raises DictationError(status_code=413, error_code="audio_too_large")
+```
+
+`aai.AsyncDictationTranscriber` takes the same option.
+
+</details>
+
 <details>
   <summary>Handle errors</summary>
 
diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py
index 0a0a43a..fcfdf38 100644
--- a/assemblyai/__version__.py
+++ b/assemblyai/__version__.py
@@ -1 +1 @@
-__version__ = "1.9.0"
+__version__ = "1.10.0"
diff --git a/assemblyai/dictation/v1/_base.py b/assemblyai/dictation/v1/_base.py
index 9ff2b25..038e133 100644
--- a/assemblyai/dictation/v1/_base.py
+++ b/assemblyai/dictation/v1/_base.py
@@ -1,7 +1,16 @@
 from __future__ import annotations
 
 import os
-from typing import AsyncIterable, BinaryIO, Iterable, Mapping, Optional, Tuple, Union
+from typing import (
+    AsyncIterable,
+    BinaryIO,
+    Callable,
+    Iterable,
+    Mapping,
+    Optional,
+    Tuple,
+    Union,
+)
 from urllib.parse import urlparse
 
 from ... import _audio
@@ -19,11 +28,11 @@
     "resolve_source",
 ]
 
-# Audio for a dictation request. The live endpoint is the only one this
-# client opens, so it takes every shape audio comes in: a producer that is
-# still recording (an iterable of chunks, or a file object read as it fills)
-# and audio that is already complete (bytes, or a local path), which goes out
-# over the same connection as a single chunk.
+# Audio for a dictation request. Every request starts on the live endpoint, so
+# it takes every shape audio comes in: a producer that is still recording (an
+# iterable of chunks, or a file object read as it fills) and audio that is
+# already complete (bytes, or a local path), which goes out over the same
+# connection as a single chunk.
 AudioSource = Union[
     str, bytes, bytearray, "os.PathLike[str]", BinaryIO, Iterable[bytes]
 ]
@@ -162,16 +171,19 @@ def __init__(
         client: _client.Client,
         config: DictationConfig,
         owner: str,
+        fallback_to_buffered: bool,
     ) -> None:
         self._client = client
         self.config = config
         self._owner = owner
+        self.fallback_to_buffered = fallback_to_buffered
 
     def transcribe_live(
         self,
         *,
         data: AudioSource,
         config: Optional[DictationConfig],
+        cancelled: Optional[Callable[[], bool]] = None,
     ) -> DictationResponse:
         config = config or self.config
         chunks, filename, content_type = resolve_source(data, config, self._owner)
@@ -184,4 +196,6 @@ def transcribe_live(
             audio_content_type=content_type,
             config=_config_to_json(config),
             timeout=self._client.settings.dictation_http_timeout,
+            fallback_to_buffered=self.fallback_to_buffered,
+            cancelled=cancelled,
         )
diff --git a/assemblyai/dictation/v1/api.py b/assemblyai/dictation/v1/api.py
index 65934bf..7f598bf 100644
--- a/assemblyai/dictation/v1/api.py
+++ b/assemblyai/dictation/v1/api.py
@@ -1,18 +1,44 @@
-from typing import Iterator, Optional
+import logging
+from typing import Callable, Iterator, Optional, Tuple
 
 import httpx
 
 from ... import types
-from ..._multipart import AudioChunks, StreamingMultipartEncoder, iter_chunks
+from ..._multipart import (
+    AudioChunks,
+    StreamingMultipartEncoder,
+    _Aborted,
+    iter_chunks,
+)
 from ...sync.v1.api import _error_from_response as _shared_error_from_response
 from .models import DictationError, DictationResponse
 
-# The one endpoint this client posts audio to. The Dictation API serves it
-# under /v1 only (no unprefixed alias), and also at `/v1/transcribe/stream`,
-# the path it shipped under — but `live` is its name.
+logger = logging.getLogger(__name__)
+
+# The endpoint every dictation starts on. The Dictation API serves it under /v1
+# only (no unprefixed alias), and also at `/v1/transcribe/stream`, the path it
+# shipped under — but `live` is its name. It caps a request at 120 s of audio.
 ENDPOINT_TRANSCRIBE_LIVE = "/v1/transcribe/live"
+# The buffered endpoint: the whole recording in one body, up to 20 minutes /
+# 256 MB. Requested only as the fallback for a live request that ran past the
+# live cap; see `transcribe_live(fallback_to_buffered=True)`.
+ENDPOINT_TRANSCRIBE = "/v1/transcribe"
 ENDPOINT_WARM = "/v1/warm"
 
+# The most audio the fallback holds in memory, matching the buffered
+# endpoint's 256 MB body limit. Past it the recording could not be accepted
+# anyway, so the copy is dropped and the request fails without being sent.
+BUFFERED_MAX_BYTES = 256 * 1024 * 1024
+
+# The live endpoint's answer to audio past its 120 s cap — the one failure the
+# buffered fallback retries.
+_LIVE_CAP_STATUS = 413
+_LIVE_CAP_ERROR_CODE = "audio_too_large"
+
+# Piece size the buffered body is handed to the transport in, so sending the
+# copy never makes a second full-size copy of it.
+_BUFFERED_SEND_SIZE = 1024 * 1024
+
 
 def _error_from_response(response: httpx.Response) -> types.AssemblyAIError:
     """
@@ -31,6 +57,170 @@ def _error_from_response(response: httpx.Response) -> types.AssemblyAIError:
     )
 
 
+def _is_live_cap(error: types.AssemblyAIError) -> bool:
+    """Whether a live request failed only because it ran past the 120 s cap."""
+
+    return (
+        isinstance(error, DictationError)
+        and error.status_code == _LIVE_CAP_STATUS
+        and error.error_code == _LIVE_CAP_ERROR_CODE
+    )
+
+
+def _too_large_for_fallback() -> DictationError:
+    """The error raised when the recording outgrows `BUFFERED_MAX_BYTES`."""
+
+    return DictationError(
+        f"audio exceeds {BUFFERED_MAX_BYTES // (1024 * 1024)} MB, the most the "
+        "buffered fallback for a dictation past the live 120 s cap can send",
+        status_code=_LIVE_CAP_STATUS,
+        error_code=_LIVE_CAP_ERROR_CODE,
+    )
+
+
+class _Recording:
+    """
+    The copy of the audio kept for the buffered fallback.
+
+    Only audio bytes are copied, never the multipart framing. The copy is
+    capped at `BUFFERED_MAX_BYTES`: once it passes that it is dropped, the
+    producer is not read further, and a 413 `audio_too_large` is raised. The
+    cap holds while the live request is still uploading too — the Dictation
+    API reads a live body to its end before answering 413, so most of a long
+    recording arrives before the drain — and the live endpoint could never
+    accept that much audio anyway (120 s is at most ~23 MB of 16-bit audio).
+    """
+
+    def __init__(self) -> None:
+        self.audio = bytearray()
+
+    def _keep(self, chunk: bytes) -> None:
+        self.audio += chunk
+        if len(self.audio) > BUFFERED_MAX_BYTES:
+            self.audio = bytearray()
+            raise _too_large_for_fallback()
+
+
+class _RecordedChunks(_Recording):
+    """
+    The live upload's audio source, keeping a copy of every byte it hands out.
+
+    The transport pulls from it while the live request runs. When that request
+    fails on the 120 s cap, `drain()` goes on reading the same source — which
+    the transport no longer does — until the producer ends, so the copy holds
+    the whole recording for the buffered retry.
+    """
+
+    def __init__(self, chunks: AudioChunks) -> None:
+        super().__init__()
+        self._chunks = iter_chunks(chunks)
+
+    def __iter__(self) -> "_RecordedChunks":
+        return self
+
+    def __next__(self) -> bytes:
+        chunk = next(self._chunks)
+        self._keep(chunk)
+        return chunk
+
+    def drain(self) -> None:
+        """
+        Reads the rest of the source into the copy, until the producer ends.
+
+        Raises:
+            DictationError: once the copy passes `BUFFERED_MAX_BYTES`; the copy
+                is dropped and the producer is not read further.
+            Exception: anything the producer raises, unchanged.
+        """
+        for chunk in self._chunks:
+            self._keep(chunk)
+
+
+def _buffered_body(
+    audio: bytearray,
+    *,
+    filename: str,
+    audio_content_type: str,
+    config: Optional[dict],
+) -> Tuple[str, int, Iterator[bytes]]:
+    """
+    Frames the complete recording as the buffered endpoint's body.
+
+    Same parts, same order, same audio filename and Content-Type as the live
+    upload. The body is handed out in bounded pieces with its length declared
+    up front, so the request carries a `Content-Length` rather than chunked
+    framing and the recording is never copied whole a second time.
+
+    Returns: `(content_type, content_length, pieces)`.
+    """
+    encoder = StreamingMultipartEncoder()
+    head = encoder.config_part(config) + encoder.audio_header(
+        filename, audio_content_type
+    )
+    tail = encoder.closing()
+
+    def pieces() -> Iterator[bytes]:
+        yield head
+        for start in range(0, len(audio), _BUFFERED_SEND_SIZE):
+            yield bytes(audio[start : start + _BUFFERED_SEND_SIZE])
+        yield tail
+
+    return encoder.content_type, len(head) + len(audio) + len(tail), pieces()
+
+
+def _parse(response: httpx.Response) -> DictationResponse:
+    if response.status_code != httpx.codes.OK:
+        raise _error_from_response(response)
+
+    return DictationResponse.parse_obj(response.json())
+
+
+def transcribe_buffered(
+    client: httpx.Client,
+    *,
+    base_url: str,
+    audio: bytearray,
+    filename: str,
+    audio_content_type: str,
+    config: Optional[dict],
+    timeout: float,
+) -> DictationResponse:
+    """
+    Posts a complete recording to the buffered endpoint, `/v1/transcribe`.
+
+    The fallback for a live request that ran past the 120 s cap: the buffered
+    endpoint takes up to 20 minutes / 256 MB, the same config fields, and
+    answers with the same response shape.
+
+    Args:
+        client: the HTTP client (carries the `Authorization` header).
+        base_url: the Dictation API base URL.
+        audio: the whole recording (WAV container or S16LE PCM).
+        filename: name for the audio multipart part.
+        audio_content_type: the audio part's Content-Type; selects the decoder.
+        config: the JSON `config` part, or None for an empty one.
+        timeout: per-operation timeout in seconds.
+
+    Returns: the parsed dictation response.
+
+    Raises: `DictationError` on any non-200 response.
+    """
+    content_type, length, pieces = _buffered_body(
+        audio,
+        filename=filename,
+        audio_content_type=audio_content_type,
+        config=config,
+    )
+    response = client.post(
+        base_url.rstrip("/") + ENDPOINT_TRANSCRIBE,
+        content=pieces,
+        headers={"Content-Type": content_type, "Content-Length": str(length)},
+        timeout=timeout,
+    )
+
+    return _parse(response)
+
+
 def transcribe_live(
     client: httpx.Client,
     *,
@@ -40,6 +230,8 @@ def transcribe_live(
     audio_content_type: str,
     config: Optional[dict],
     timeout: float,
+    fallback_to_buffered: bool = False,
+    cancelled: Optional[Callable[[], bool]] = None,
 ) -> DictationResponse:
     """
     Posts a dictation request whose audio is uploaded as it arrives.
@@ -63,18 +255,29 @@ def transcribe_live(
             it bounds connecting, each socket write and each read while waiting
             for the response, not the request end to end. Time blocked in the
             caller's producer is not counted, so it need not cover the
-            recording.
+            recording. The buffered fallback request gets the same timeout;
+            draining the producer before it is not bounded by any timeout.
+        fallback_to_buffered: keep a copy of the audio and, if the live request
+            fails with 413 `audio_too_large` (the 120 s cap), read the producer
+            to its end and post the whole recording to `/v1/transcribe`.
+        cancelled: polled once the producer has ended, before the fallback
+            request; when it returns True the request is dropped instead.
 
-    Returns: the parsed dictation response.
+    Returns: the parsed dictation response, from the live endpoint or, after
+        a fallback, from the buffered one.
 
     Raises: `DictationError` on any non-200 response — including one the
         server sends while the upload is still in flight, which it may do for
-        auth, rate-limit, size and capacity failures. `TypeError` if the
-        producer yields something other than bytes. Any other exception the
+        auth, rate-limit, size and capacity failures. After a fallback it is
+        the buffered request's own error, or a 413 `audio_too_large` raised
+        here once the recording passes `BUFFERED_MAX_BYTES`. `TypeError` if
+        the producer yields something other than bytes. Any other exception the
         producer raises propagates unchanged; the connection is dropped and no
         transcript is returned.
     """
     encoder = StreamingMultipartEncoder()
+    recorded = _RecordedChunks(chunks) if fallback_to_buffered else None
+    source: Iterator[bytes] = recorded if recorded is not None else iter_chunks(chunks)
 
     def body() -> Iterator[bytes]:
         # config first: the server needs sample_rate and channels before it can
@@ -83,7 +286,7 @@ def body() -> Iterator[bytes]:
             filename, audio_content_type
         )
 
-        for chunk in iter_chunks(chunks):
+        for chunk in source:
             if chunk:
                 yield chunk
 
@@ -96,7 +299,35 @@ def body() -> Iterator[bytes]:
         timeout=timeout,
     )
 
-    if response.status_code != httpx.codes.OK:
-        raise _error_from_response(response)
+    if response.status_code == httpx.codes.OK:
+        return DictationResponse.parse_obj(response.json())
 
-    return DictationResponse.parse_obj(response.json())
+    error = _error_from_response(response)
+    if recorded is None or not _is_live_cap(error):
+        raise error
+
+    # Past the live cap. The 413 may come back while the producer is still
+    # recording — the transport stops reading the body once it has the
+    # response — so keep reading it until the speaker stops. That wait is
+    # recording time, not network time, and no timeout bounds it.
+    recorded.drain()
+    if cancelled is not None and cancelled():
+        raise _Aborted()
+
+    audio = recorded.audio
+    logger.info(
+        "Dictation passed the live endpoint's 120 s cap; posting the whole "
+        "recording (%d bytes) to %s",
+        len(audio),
+        ENDPOINT_TRANSCRIBE,
+    )
+
+    return transcribe_buffered(
+        client,
+        base_url=base_url,
+        audio=audio,
+        filename=filename,
+        audio_content_type=audio_content_type,
+        config=config,
+        timeout=timeout,
+    )
diff --git a/assemblyai/dictation/v1/async_api.py b/assemblyai/dictation/v1/async_api.py
index 426eded..6350b52 100644
--- a/assemblyai/dictation/v1/async_api.py
+++ b/assemblyai/dictation/v1/async_api.py
@@ -4,15 +4,98 @@
 through `api._error_from_response`.
 """
 
-from typing import AsyncIterator, Optional
+from typing import AsyncIterator, Callable, Optional
 
 import httpx
 
-from ..._multipart import AsyncAudioChunks, StreamingMultipartEncoder, aiter_chunks
-from .api import ENDPOINT_TRANSCRIBE_LIVE, _error_from_response
+from ..._multipart import (
+    AsyncAudioChunks,
+    StreamingMultipartEncoder,
+    _Aborted,
+    aiter_chunks,
+)
+from .api import (
+    ENDPOINT_TRANSCRIBE,
+    ENDPOINT_TRANSCRIBE_LIVE,
+    _buffered_body,
+    _error_from_response,
+    _is_live_cap,
+    _parse,
+    _Recording,
+    logger,
+)
 from .models import DictationResponse
 
-__all__ = ["transcribe_live"]
+__all__ = ["transcribe_buffered", "transcribe_live"]
+
+
+class _AsyncRecordedChunks(_Recording):
+    """
+    The asyncio counterpart of `api._RecordedChunks`: the live upload's audio
+    source, keeping a copy of every byte it hands out, and able to go on
+    reading the source once the transport has stopped.
+    """
+
+    def __init__(self, chunks: AsyncAudioChunks) -> None:
+        super().__init__()
+        self._chunks = aiter_chunks(chunks)
+
+    def __aiter__(self) -> "_AsyncRecordedChunks":
+        return self
+
+    async def __anext__(self) -> bytes:
+        chunk = await self._chunks.__anext__()
+        self._keep(chunk)
+        return chunk
+
+    async def drain(self) -> None:
+        """
+        Reads the rest of the source into the copy, until the producer ends.
+
+        Raises:
+            DictationError: once the copy passes `BUFFERED_MAX_BYTES`; the copy
+                is dropped and the producer is not read further.
+            Exception: anything the producer raises, unchanged.
+        """
+        async for chunk in self._chunks:
+            self._keep(chunk)
+
+
+async def transcribe_buffered(
+    client: httpx.AsyncClient,
+    *,
+    base_url: str,
+    audio: bytearray,
+    filename: str,
+    audio_content_type: str,
+    config: Optional[dict],
+    timeout: float,
+) -> DictationResponse:
+    """
+    Posts a complete recording to the buffered endpoint, `/v1/transcribe`.
+
+    The asyncio counterpart of `api.transcribe_buffered`; same endpoint, same
+    framing, same errors.
+    """
+    content_type, length, pieces = _buffered_body(
+        audio,
+        filename=filename,
+        audio_content_type=audio_content_type,
+        config=config,
+    )
+
+    async def body() -> AsyncIterator[bytes]:
+        for piece in pieces:
+            yield piece
+
+    response = await client.post(
+        base_url.rstrip("/") + ENDPOINT_TRANSCRIBE,
+        content=body(),
+        headers={"Content-Type": content_type, "Content-Length": str(length)},
+        timeout=timeout,
+    )
+
+    return _parse(response)
 
 
 async def transcribe_live(
@@ -24,12 +107,14 @@ async def transcribe_live(
     audio_content_type: str,
     config: Optional[dict],
     timeout: float,
+    fallback_to_buffered: bool = False,
+    cancelled: Optional[Callable[[], bool]] = None,
 ) -> DictationResponse:
     """
     Posts a dictation request whose audio is uploaded as it arrives.
 
     The asyncio counterpart of `api.transcribe_live`; same endpoint, same
-    chunked framing, same errors.
+    chunked framing, same errors, same buffered fallback.
 
     Args:
         client: the HTTP client (carries the `Authorization` header).
@@ -40,13 +125,22 @@ async def transcribe_live(
         config: the JSON `config` part. None sends an empty object: the live
             endpoint requires the part ahead of the audio.
         timeout: per-operation timeout in seconds; see `api.transcribe_live`.
+        fallback_to_buffered: see `api.transcribe_live`.
+        cancelled: see `api.transcribe_live`.
 
-    Returns: the parsed dictation response.
+    Returns: the parsed dictation response, from the live endpoint or, after
+        a fallback, from the buffered one.
 
     Raises: `DictationError` on any non-200 response — including one the
-        server sends while the upload is still in flight.
+        server sends while the upload is still in flight. After a fallback it
+        is the buffered request's own error, or a 413 `audio_too_large` raised
+        here once the recording passes `BUFFERED_MAX_BYTES`.
     """
     encoder = StreamingMultipartEncoder()
+    recorded = _AsyncRecordedChunks(chunks) if fallback_to_buffered else None
+    source: AsyncIterator[bytes] = (
+        recorded if recorded is not None else aiter_chunks(chunks)
+    )
 
     async def body() -> AsyncIterator[bytes]:
         # config first: the server needs sample_rate and channels before it can
@@ -55,7 +149,7 @@ async def body() -> AsyncIterator[bytes]:
             filename, audio_content_type
         )
 
-        async for chunk in aiter_chunks(chunks):
+        async for chunk in source:
             if chunk:
                 yield chunk
 
@@ -68,7 +162,33 @@ async def body() -> AsyncIterator[bytes]:
         timeout=timeout,
     )
 
-    if response.status_code != httpx.codes.OK:
-        raise _error_from_response(response)
+    if response.status_code == httpx.codes.OK:
+        return DictationResponse.parse_obj(response.json())
+
+    error = _error_from_response(response)
+    if recorded is None or not _is_live_cap(error):
+        raise error
+
+    # Past the live cap: read the producer until the speaker stops, then send
+    # the whole recording buffered. See `api.transcribe_live`.
+    await recorded.drain()
+    if cancelled is not None and cancelled():
+        raise _Aborted()
+
+    audio = recorded.audio
+    logger.info(
+        "Dictation passed the live endpoint's 120 s cap; posting the whole "
+        "recording (%d bytes) to %s",
+        len(audio),
+        ENDPOINT_TRANSCRIBE,
+    )
 
-    return DictationResponse.parse_obj(response.json())
+    return await transcribe_buffered(
+        client,
+        base_url=base_url,
+        audio=audio,
+        filename=filename,
+        audio_content_type=audio_content_type,
+        config=config,
+        timeout=timeout,
+    )
diff --git a/assemblyai/dictation/v1/async_client.py b/assemblyai/dictation/v1/async_client.py
index b4a381a..a017214 100644
--- a/assemblyai/dictation/v1/async_client.py
+++ b/assemblyai/dictation/v1/async_client.py
@@ -44,9 +44,10 @@ class AsyncDictationLiveSession:
     library's capture thread, say — schedule them with
     `loop.call_soon_threadsafe(session.write, chunk)`.
 
-    Everything `transcribe_live()` says about the 120 s audio cap, the
-    server-side silence abort and errors surfacing mid-upload applies here
-    unchanged.
+    Everything `transcribe_live()` 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. After a fallback, `result()`
+    waits for the whole recording to upload and be transcribed.
 
     Example:
         ```python
@@ -60,17 +61,22 @@ class AsyncDictationLiveSession:
 
     def __init__(
         self,
-        start: Callable[[AsyncIterator[bytes]], asyncio.Task[DictationResponse]],
+        start: Callable[
+            [AsyncIterator[bytes], Callable[[], bool]],
+            asyncio.Task[DictationResponse],
+        ],
     ) -> None:
         """
         Args:
             start: begins the upload from the given producer and returns the
-                task that will hold its outcome. Supplied by the transcriber.
+                task that will hold its outcome. Its second argument reports
+                whether the session has been aborted, so a buffered fallback is
+                never sent for an aborted session. Supplied by the transcriber.
         """
         self._queue: "asyncio.Queue[Optional[bytes]]" = asyncio.Queue()
         self._closed = False
         self._aborted = False
-        self._task = start(self._chunks())
+        self._task = start(self._chunks(), lambda: self._aborted)
 
     async def _chunks(self) -> AsyncIterator[bytes]:
         while True:
@@ -138,10 +144,12 @@ async def abort(self) -> None:
         Drops the request without a transcript.
 
         The connection is closed and nothing the server may still return is
-        kept; `result()` raises afterwards. Idempotent, and a no-op once the
-        request has completed. Waits until the task has let go of the
-        connection. After `close()` the upload is already complete, so
-        aborting then waits out the in-flight response instead.
+        kept; `result()` raises afterwards. No buffered fallback request is
+        sent after an abort, and the copy of the audio kept for one is
+        dropped. Idempotent, and a no-op once the request has completed.
+        Waits until the task has let go of the connection. After `close()` the
+        upload is already complete, so aborting then waits out the in-flight
+        response instead.
         """
         if self._aborted or self._task.done():
             return
@@ -184,7 +192,9 @@ class AsyncDictationTranscriber:
     `warm()` as coroutines and `open_live()` returning an
     `AsyncDictationLiveSession`. Use it in asyncio code (FastAPI, aiohttp,
     voice agents), where the threaded `DictationTranscriber` would block the
-    loop.
+    loop. Past the live endpoint's 120 s cap it falls back to the buffered
+    endpoint exactly as `DictationTranscriber` does; see
+    `fallback_to_buffered` on `__init__`.
 
     The transcriber owns an HTTP connection pool. Close it with `aclose()`,
     or use the transcriber as an async context manager.
@@ -213,6 +223,7 @@ def __init__(
         client: Optional[_async_client.AsyncClient] = None,
         config: Optional[DictationConfig] = None,
         api_key: Optional[str] = None,
+        fallback_to_buffered: bool = True,
     ) -> None:
         """
         Creates an `AsyncDictationTranscriber`.
@@ -229,6 +240,22 @@ def __init__(
                 transcriber builds and owns a client made from a copy of that
                 client's settings with the key replaced, and the given client is
                 left untouched and stays the caller's to close.
+            fallback_to_buffered: Whether a dictation past the live endpoint's
+                120 s cap falls back to the buffered endpoint. Applies to
+                `transcribe_live()` and `open_live()`. When True (the
+                default), the transcriber keeps an in-memory copy of all the
+                audio it uploads. If the live request is refused with 413
+                `audio_too_large`, it keeps reading the audio source until it
+                ends — the producer's iterator is exhausted, or the session is
+                closed — then posts the whole recording once to
+                `/v1/transcribe` (up to 20 minutes / 256 MB) and returns that
+                result like a live one. 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. The copy costs
+                about 32 KB per second of 16 kHz mono audio (about 38 MB at 20
+                minutes); past 256 MB the copy is dropped and a 413
+                `DictationError` is raised. False keeps no copy and raises the
+                live 413 as before.
 
         Raises:
             TypeError: if `config` is not a `DictationConfig`.
@@ -238,6 +265,13 @@ def __init__(
         self._owns_client = client is None or api_key is not None
         self._client = _async_client._resolve_client(client, api_key)
         self.config = config or DictationConfig()
+        self._fallback_to_buffered = fallback_to_buffered
+
+    @property
+    def fallback_to_buffered(self) -> bool:
+        """Whether audio past the live 120 s cap falls back to `/v1/transcribe`."""
+
+        return self._fallback_to_buffered
 
     @property
     def client(self) -> _async_client.AsyncClient:
@@ -263,7 +297,12 @@ async def transcribe_live(
 
         The caller must keep producing: an upload that goes silent for long
         enough is aborted server-side. Stop by ending the iterator, not by
-        pausing it. The service caps a request at 120 s of audio.
+        pausing it. The live endpoint caps a request at 120 s of audio. Past
+        it, unless the transcriber was built with `fallback_to_buffered=False`,
+        the iterator is read to its end and the whole recording is posted to
+        the buffered endpoint (up to 20 minutes), whose result is returned.
+        That result waits for the whole recording to upload and be
+        transcribed, so a long dictation returns more slowly than a live one.
 
         Args:
             data: An async iterable of audio chunks, a plain iterable, a file
@@ -282,6 +321,9 @@ async def transcribe_live(
                 `channels`.
             DictationError: if the request fails. Auth, rate-limit, size and
                 capacity failures can surface part-way through the upload.
+                After a fallback, it is the buffered request's own error (a
+                413 past 20 minutes, say), or a 413 `audio_too_large` once the
+                recording passes 256 MB.
             Exception: anything the producer raises mid-upload propagates
                 unchanged. The connection is dropped and no transcript is
                 returned.
@@ -298,6 +340,14 @@ async def mic_chunks():
         """
         check_config(type(self).__name__, config)
 
+        return await self._transcribe_live(data, config)
+
+    async def _transcribe_live(
+        self,
+        data: AsyncAudioSource,
+        config: Optional[DictationConfig],
+        cancelled: Optional[Callable[[], bool]] = None,
+    ) -> DictationResponse:
         config = config or self.config
         # Resolving may read a whole file for a path source; keep that off
         # the loop.
@@ -315,6 +365,8 @@ async def mic_chunks():
             audio_content_type=content_type,
             config=_config_to_json(config),
             timeout=self._client.settings.dictation_http_timeout,
+            fallback_to_buffered=self._fallback_to_buffered,
+            cancelled=cancelled,
         )
 
     def open_live(
@@ -331,6 +383,11 @@ def open_live(
         for the transcript once the speaker stops. See
         `AsyncDictationLiveSession`.
 
+        Past the live endpoint's 120 s cap the session falls back the way
+        `transcribe_live()` does: keep writing until the speaker stops, and
+        `result()` returns the buffered endpoint's transcript of the whole
+        recording, more slowly than a live result.
+
         Not a coroutine, so it can be used directly as `async with
         transcriber.open_live(config) as session:`. Must be called while the
         event loop is running.
@@ -348,7 +405,9 @@ def open_live(
         loop = asyncio.get_running_loop()
 
         return AsyncDictationLiveSession(
-            lambda chunks: loop.create_task(self.transcribe_live(chunks, config=config))
+            lambda chunks, cancelled: loop.create_task(
+                self._transcribe_live(chunks, config, cancelled)
+            )
         )
 
     async def warm(self) -> bool:
diff --git a/assemblyai/dictation/v1/client.py b/assemblyai/dictation/v1/client.py
index ec02569..7baca25 100644
--- a/assemblyai/dictation/v1/client.py
+++ b/assemblyai/dictation/v1/client.py
@@ -26,9 +26,10 @@ class DictationLiveSession:
     arrives in a callback rather than from an iterator you can hand to
     `transcribe_live()`.
 
-    Everything `transcribe_live()` says about the 120 s audio cap, the
-    server-side silence abort and errors surfacing mid-upload applies here
-    unchanged.
+    Everything `transcribe_live()` 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. After a fallback, `result()`
+    waits for the whole recording to upload and be transcribed.
 
     Example:
         ```python
@@ -51,18 +52,21 @@ class DictationLiveSession:
     def __init__(
         self,
         start: Callable[
-            [Iterator[bytes]], concurrent.futures.Future[DictationResponse]
+            [Iterator[bytes], Callable[[], bool]],
+            concurrent.futures.Future[DictationResponse],
         ],
     ) -> None:
         """
         Args:
             start: begins the upload from the given producer and returns the
-                future that will hold its outcome. Supplied by the transcriber.
+                future that will hold its outcome. Its second argument reports
+                whether the session has been aborted, so a buffered fallback is
+                never sent for an aborted session. Supplied by the transcriber.
         """
         self._queue: "queue.Queue[Optional[bytes]]" = queue.Queue()
         self._closed = False
         self._aborted = False
-        self._future = start(self._chunks())
+        self._future = start(self._chunks(), lambda: self._aborted)
 
     def _chunks(self) -> Iterator[bytes]:
         while True:
@@ -130,10 +134,12 @@ def abort(self) -> None:
         Drops the request without a transcript.
 
         The connection is closed and nothing the server may still return is
-        kept; `result()` raises afterwards. Idempotent, and a no-op once the
-        request has completed. Blocks until the worker thread has let go of
-        the connection. After `close()` the upload is already complete, so
-        aborting then waits out the in-flight response instead.
+        kept; `result()` raises afterwards. No buffered fallback request is
+        sent after an abort, and the copy of the audio kept for one is
+        dropped. Idempotent, and a no-op once the request has completed.
+        Blocks until the worker thread has let go of the connection. After
+        `close()` the upload is already complete, so aborting then waits out
+        the in-flight response instead.
         """
         if self._aborted or self._future.done():
             return
@@ -176,6 +182,11 @@ class DictationTranscriber:
     a follow-up LLM pass: set `llm_instruction` on the config and read
     `result.final_text`.
 
+    The live endpoint takes up to 120 s of audio. A longer dictation, up to
+    20 minutes, still succeeds: the transcriber falls back to the buffered
+    endpoint and returns its result the same way. See `fallback_to_buffered`
+    on `__init__` for what that costs and how to turn it off.
+
     Example:
         ```python
         import assemblyai as aai
@@ -197,6 +208,7 @@ def __init__(
         config: Optional[DictationConfig] = None,
         max_workers: Optional[int] = None,
         api_key: Optional[str] = None,
+        fallback_to_buffered: bool = True,
     ) -> None:
         """
         Creates a `DictationTranscriber`.
@@ -212,6 +224,22 @@ def __init__(
                 transcriber builds its own client from a copy of that client's
                 settings with the key replaced, and the given client is left
                 untouched.
+            fallback_to_buffered: Whether a dictation past the live endpoint's
+                120 s cap falls back to the buffered endpoint. Applies to
+                `transcribe_live()` and `open_live()`. When True (the
+                default), the transcriber keeps an in-memory copy of all the
+                audio it uploads. If the live request is refused with 413
+                `audio_too_large`, it keeps reading the audio source until it
+                ends — the producer's iterator is exhausted, or the session is
+                closed — then posts the whole recording once to
+                `/v1/transcribe` (up to 20 minutes / 256 MB) and returns that
+                result like a live one. 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. The copy costs
+                about 32 KB per second of 16 kHz mono audio (about 38 MB at 20
+                minutes); past 256 MB the copy is dropped and a 413
+                `DictationError` is raised. False keeps no copy and raises the
+                live 413 as before.
 
         Raises:
             TypeError: if `config` is not a `DictationConfig`.
@@ -223,6 +251,7 @@ def __init__(
             client=self._client,
             config=config or DictationConfig(),
             owner=type(self).__name__,
+            fallback_to_buffered=fallback_to_buffered,
         )
 
         if not max_workers:
@@ -244,6 +273,11 @@ def config(self, config: DictationConfig) -> None:
 
         self._impl.config = config
 
+    @property
+    def fallback_to_buffered(self) -> bool:
+        """Whether audio past the live 120 s cap falls back to `/v1/transcribe`."""
+        return self._impl.fallback_to_buffered
+
     def transcribe_live(
         self,
         data: AudioSource,
@@ -261,7 +295,12 @@ def transcribe_live(
 
         The caller must keep producing: an upload that goes silent for long
         enough is aborted server-side. Stop by ending the iterator, not by
-        pausing it. The service caps a request at 120 s of audio.
+        pausing it. The live endpoint caps a request at 120 s of audio. Past
+        it, unless the transcriber was built with `fallback_to_buffered=False`,
+        the iterator is read to its end and the whole recording is posted to
+        the buffered endpoint (up to 20 minutes), whose result is returned.
+        That result waits for the whole recording to upload and be
+        transcribed, so a long dictation returns more slowly than a live one.
 
         Args:
             data: An iterable of audio chunks, a binary file object read as it
@@ -278,6 +317,9 @@ def transcribe_live(
                 `channels`.
             DictationError: if the request fails. Auth, rate-limit, size and
                 capacity failures can surface part-way through the upload.
+                After a fallback, it is the buffered request's own error (a
+                413 past 20 minutes, say), or a 413 `audio_too_large` once the
+                recording passes 256 MB.
             Exception: anything the producer raises mid-upload propagates
                 unchanged. The connection is dropped and no transcript is
                 returned.
@@ -308,6 +350,11 @@ def open_live(
         the callback, then `session.result()` for the transcript once the
         speaker stops. See `DictationLiveSession`.
 
+        Past the live endpoint's 120 s cap the session falls back the way
+        `transcribe_live()` does: keep writing until the speaker stops, and
+        `result()` returns the buffered endpoint's transcript of the whole
+        recording, more slowly than a live result.
+
         Args:
             config: Options for this call. If `None`, the transcriber's default
                 configuration is used. Raw PCM requires `sample_rate` and
@@ -327,8 +374,11 @@ def open_live(
         check_config(type(self).__name__, config)
 
         return DictationLiveSession(
-            lambda chunks: self._executor.submit(
-                self._impl.transcribe_live, data=chunks, config=config
+            lambda chunks, cancelled: self._executor.submit(
+                self._impl.transcribe_live,
+                data=chunks,
+                config=config,
+                cancelled=cancelled,
             )
         )
 
diff --git a/tests/unit/test_dictation.py b/tests/unit/test_dictation.py
index ba8fafe..947abe3 100644
--- a/tests/unit/test_dictation.py
+++ b/tests/unit/test_dictation.py
@@ -8,21 +8,23 @@
 
 import email
 import io
+import json
 import threading
-from typing import List, Optional
+import time
+from typing import Iterable, List, Optional, cast
 
 import httpx
 import pytest
 from pytest_httpx import HTTPXMock
 
 import assemblyai as aai
+from assemblyai._multipart import _STREAM_READ_SIZE
+from assemblyai.dictation.v1 import api
 from assemblyai.dictation.v1.models import (
     _DICTATION_MAX_KEYTERMS_COUNT,
     _DICTATION_MAX_KEYTERMS_PROMPT_LEN,
     _DICTATION_MAX_STT_PROMPT_LEN,
 )
-from assemblyai._multipart import _STREAM_READ_SIZE
-from assemblyai.dictation.v1 import api
 
 aai.settings.api_key = "test"
 
@@ -111,14 +113,14 @@ def test_public_names_are_the_dictation_models_classes():
 
 def test_there_is_no_buffered_transcribe():
     # Given the two clients
-    # Then neither exposes the buffered entry points: the live route is the
-    # only one the Dictation API markets, and the only one this client opens
+    # Then neither exposes a buffered entry point: every dictation starts on
+    # the live route, and the buffered route is only its fallback past the
+    # 120 s cap
     for cls in (aai.DictationTranscriber, aai.AsyncDictationTranscriber):
         assert not hasattr(cls, "transcribe")
         assert not hasattr(cls, "transcribe_async")
         assert hasattr(cls, "transcribe_live")
         assert hasattr(cls, "open_live")
-    assert not hasattr(api, "ENDPOINT_TRANSCRIBE")
     assert not hasattr(api, "transcribe")
 
 
@@ -681,9 +683,12 @@ def test_problem_details_envelope_maps_to_dictation_error(httpx_mock: HTTPXMock)
         json={"status": 413, "title": "Audio Too Large", "detail": "too long"},
     )
 
-    # When streaming, then the snake_cased title becomes the error_code
+    # When streaming without the buffered fallback (which would retry this
+    # one), then the snake_cased title becomes the error_code
     with pytest.raises(aai.DictationError) as exc_info:
-        aai.DictationTranscriber().transcribe_live(_chunks(b"RIFF"))
+        aai.DictationTranscriber(fallback_to_buffered=False).transcribe_live(
+            _chunks(b"RIFF")
+        )
 
     error = exc_info.value
     assert error.status_code == 413
@@ -819,11 +824,13 @@ def test_dictation_settings_defaults():
     assert settings.dictation_http_timeout == 300.0
 
 
-def test_the_buffered_endpoint_is_never_requested(httpx_mock: HTTPXMock):
-    """No entry point on this client posts to `/v1/transcribe`.
+def test_the_buffered_endpoint_is_not_requested_within_the_cap(
+    httpx_mock: HTTPXMock,
+):
+    """No entry point posts to `/v1/transcribe` while the live request succeeds.
 
-    The service still serves the buffered route, but the live route is the
-    only one this client opens, whichever way audio is submitted.
+    The buffered route is only the fallback for a live 413 past the 120 s cap;
+    whichever way audio is submitted, a dictation within it goes live only.
     """
     for _ in range(4):  # one per request; older pytest-httpx has no is_reusable
         _mock_ok(httpx_mock)
@@ -1051,3 +1058,426 @@ def test_live_session_surfaces_server_errors(httpx_mock: HTTPXMock):
             session.result()
 
     assert exc.value.error_code == "capacity_exceeded"
+
+
+# --- buffered fallback past the live 120 s cap --------------------------------
+
+# The live endpoint's answer to audio past its cap, verbatim from production.
+_LIVE_CAP = {"error": "audio exceeds 120000 ms", "error_code": "audio_too_large"}
+
+_BUFFERED_OK = {
+    **_OK_RESPONSE,
+    "audio_duration_ms": 130000,
+    "session_id": "4f0c1d7e-buffered",
+}
+
+_PCM = aai.DictationConfig(sample_rate=16000, channels=1)
+
+
+class _CapTransport(httpx.BaseTransport):
+    """
+    A transport scripted like the Dictation API around the live 120 s cap.
+
+    The live request's body is read for `live_pieces` pieces — the multipart
+    head is the first — or whole when it is None, and then answered with
+    `live_status`/`live_json`. Stopping part-way is what httpcore does when
+    the server answers while the upload is still in flight: it stops pulling
+    the body and returns the response. The buffered request is read whole and
+    answered with `buffered_status`/`buffered_json`.
+    """
+
+    def __init__(
+        self,
+        *,
+        live_status: int = 413,
+        live_json: Optional[dict] = None,
+        live_pieces: Optional[int] = None,
+        buffered_status: int = 200,
+        buffered_json: Optional[dict] = None,
+        gate: Optional[threading.Event] = None,
+    ) -> None:
+        self.live_status = live_status
+        self.live_json = _LIVE_CAP if live_json is None else live_json
+        self.live_pieces = live_pieces
+        self.buffered_status = buffered_status
+        self.buffered_json = _BUFFERED_OK if buffered_json is None else buffered_json
+        self.gate = gate
+        self.live_request: Optional[httpx.Request] = None
+        self.live_body = b""
+        self.body_read = threading.Event()
+        self.answered = threading.Event()
+        self.buffered: List[tuple] = []
+
+    def handle_request(self, request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/v1/transcribe/live":
+            self.live_request = request
+            pieces: List[bytes] = []
+            for piece in cast(Iterable[bytes], request.stream):
+                pieces.append(piece)
+                if self.live_pieces is not None and len(pieces) >= self.live_pieces:
+                    break
+            self.live_body = b"".join(pieces)
+            self.body_read.set()
+            if self.gate is not None:
+                self.gate.wait(timeout=5.0)
+            self.answered.set()
+            return httpx.Response(self.live_status, json=self.live_json)
+
+        body = b"".join(cast(Iterable[bytes], request.stream))
+        self.buffered.append((request, body))
+        return httpx.Response(self.buffered_status, json=self.buffered_json)
+
+
+def _transcriber_over(transport: httpx.BaseTransport, **kwargs):
+    """A `DictationTranscriber` whose client sends through `transport`."""
+
+    client = aai.Client()
+    client.http_client.close()
+    client._http_client = httpx.Client(
+        transport=transport, headers=dict(client.http_client.headers)
+    )
+    return aai.DictationTranscriber(client=client, **kwargs)
+
+
+def _body_parts(content_type: str, body: bytes):
+    raw = b"Content-Type: " + content_type.encode() + b"\r\n\r\n"
+    message = email.message_from_bytes(raw + body)
+
+    return {
+        part.get_param("name", header="content-disposition"): part
+        for part in message.get_payload()
+    }
+
+
+def _audio_head(body: bytes) -> bytes:
+    """The audio part's headers as the live body carried them."""
+
+    start = body.index(b'name="audio"')
+    return body[start : body.index(b"\r\n\r\n", start)]
+
+
+def _assert_buffered(transport: _CapTransport, audio: bytes, config: dict) -> None:
+    """The fallback posted `audio` whole, framed exactly like the live upload."""
+
+    assert len(transport.buffered) == 1
+    request, body = transport.buffered[0]
+    assert str(request.url) == BUFFERED_URL
+    assert request.headers["content-length"] == str(len(body))
+    assert "transfer-encoding" not in request.headers
+    assert request.headers["authorization"] == "test"
+
+    parts = _body_parts(request.headers["content-type"], body)
+    assert list(parts) == ["config", "audio"]
+    assert json.loads(parts["config"].get_payload(decode=True)) == config
+    assert parts["audio"].get_payload(decode=True) == audio
+    # Same filename and Content-Type on the audio part as the live upload
+    assert _audio_head(body) == _audio_head(transport.live_body)
+
+
+def test_fallback_is_on_by_default_and_can_be_turned_off():
+    assert aai.DictationTranscriber().fallback_to_buffered is True
+    assert (
+        aai.DictationTranscriber(fallback_to_buffered=False).fallback_to_buffered
+        is False
+    )
+
+
+@pytest.mark.parametrize("fallback", [True, False])
+def test_fallback_leaves_the_success_path_unchanged(fallback):
+    # Given a dictation within the cap, which the live endpoint answers 200
+    transport = _CapTransport(live_status=200, live_json=_OK_RESPONSE)
+
+    # When it is sent with the fallback on or off
+    result = _transcriber_over(
+        transport, fallback_to_buffered=fallback
+    ).transcribe_live(_chunks(b"\x01\x02", b"\x03\x04"), config=_PCM)
+
+    # Then it is the live result, from the live route only, with the same
+    # chunked body either way
+    assert result.session_id == _OK_RESPONSE["session_id"]
+    assert transport.buffered == []
+    assert transport.live_request.headers["transfer-encoding"] == "chunked"
+    parts = _body_parts(
+        transport.live_request.headers["content-type"], transport.live_body
+    )
+    assert parts["audio"].get_payload(decode=True) == b"\x01\x02\x03\x04"
+
+
+def test_live_cap_mid_upload_falls_back_with_all_the_audio():
+    # Given the live endpoint answering 413 audio_too_large after the head and
+    # the first chunk, while the producer is still recording
+    transport = _CapTransport(live_pieces=2)
+    pulled = []
+
+    def producer():
+        for chunk in (b"A" * 4, b"B" * 4, b"C" * 4, b"D" * 4):
+            pulled.append((chunk[:1], transport.answered.is_set()))
+            yield chunk
+
+    # When the dictation runs
+    result = _transcriber_over(transport).transcribe_live(producer(), config=_PCM)
+
+    # Then the producer went on being read after the 413 until it ended
+    assert pulled == [(b"A", False), (b"B", True), (b"C", True), (b"D", True)]
+    # And the whole recording went to the buffered endpoint once, with the
+    # same config, and its result came back like a live one
+    _assert_buffered(
+        transport,
+        b"A" * 4 + b"B" * 4 + b"C" * 4 + b"D" * 4,
+        {"sample_rate": 16000, "channels": 1},
+    )
+    assert isinstance(result, aai.DictationResponse)
+    assert result.session_id == _BUFFERED_OK["session_id"]
+    assert result.audio_duration_ms == 130000
+
+
+def test_live_cap_after_the_whole_body_falls_back():
+    # Given a 413 that arrives only once the whole body was sent — what
+    # production does over HTTP/1.1, where it reads the body to the end first
+    transport = _CapTransport()
+
+    # When the dictation runs
+    result = _transcriber_over(transport).transcribe_live(
+        _chunks(b"RIFF", b"wav-1", b"wav-2"),
+        config=aai.DictationConfig(language_codes=["en"]),
+    )
+
+    # Then the buffered request carries the same audio and config
+    _assert_buffered(transport, b"RIFFwav-1wav-2", {"language_codes": ["en"]})
+    assert result.session_id == _BUFFERED_OK["session_id"]
+
+
+def test_live_cap_drains_the_rest_of_a_file_object():
+    # Given a file object the live upload had read only part of
+    audio = bytes(range(256)) * ((3 * _STREAM_READ_SIZE) // 256 + 1)
+    transport = _CapTransport(live_pieces=2)
+
+    # When the dictation runs
+    _transcriber_over(transport).transcribe_live(io.BytesIO(audio), config=_PCM)
+
+    # Then the rest of the file was read into the buffered request
+    _assert_buffered(transport, audio, {"sample_rate": 16000, "channels": 1})
+
+
+def test_fallback_disabled_raises_the_live_413_as_before():
+    # Given a transcriber built without the fallback
+    transport = _CapTransport(live_pieces=2)
+    pulled = []
+
+    def producer():
+        for chunk in (b"A", b"B", b"C"):
+            pulled.append(chunk)
+            yield chunk
+
+    # When the live endpoint answers 413 mid-upload
+    with pytest.raises(aai.DictationError) as exc:
+        _transcriber_over(transport, fallback_to_buffered=False).transcribe_live(
+            producer(), config=_PCM
+        )
+
+    # Then the live error is raised, the producer is not read further and
+    # nothing is retried
+    assert exc.value.status_code == 413
+    assert exc.value.error_code == "audio_too_large"
+    assert "120000 ms" in str(exc.value)
+    assert pulled == [b"A"]
+    assert transport.buffered == []
+
+
+@pytest.mark.parametrize(
+    "status, body",
+    [
+        (413, {"error": "body too large", "error_code": "payload_too_large"}),
+        (413, {"status": 413, "title": "Request Entity Too Large"}),
+        (413, {"detail": "too large"}),
+        (401, {"detail": "Invalid API key"}),
+        (429, {"status": 429, "title": "Too Many Requests", "detail": "slow down"}),
+        (500, {"error": "boom", "error_code": "audio_too_large"}),
+        (503, {"status": 503, "title": "Capacity Exceeded"}),
+    ],
+)
+def test_other_failures_do_not_fall_back(status, body):
+    # Given a live failure other than 413 audio_too_large
+    transport = _CapTransport(live_status=status, live_json=body, live_pieces=2)
+
+    # When the dictation runs, then that error is raised as before and
+    # nothing is retried
+    with pytest.raises(aai.DictationError) as exc:
+        _transcriber_over(transport).transcribe_live(_chunks(b"A", b"B"), config=_PCM)
+
+    assert exc.value.status_code == status
+    assert transport.buffered == []
+
+
+@pytest.mark.parametrize(
+    "status, body, code",
+    [
+        (
+            413,
+            {"error": "audio exceeds 1200000 ms", "error_code": "audio_too_large"},
+            "audio_too_large",
+        ),
+        (500, {"status": 500, "title": "Internal Error"}, "internal_error"),
+    ],
+)
+def test_a_failed_fallback_raises_its_own_error(status, body, code):
+    # Given a buffered request that fails too
+    transport = _CapTransport(buffered_status=status, buffered_json=body)
+
+    # When the dictation runs
+    with pytest.raises(aai.DictationError) as exc:
+        _transcriber_over(transport).transcribe_live(_chunks(b"A"), config=_PCM)
+
+    # Then the buffered request's error is raised, not the live 413
+    assert len(transport.buffered) == 1
+    assert exc.value.status_code == status
+    assert exc.value.error_code == code
+    assert "120000 ms" not in str(exc.value)
+
+
+def test_the_copy_is_capped_while_draining(monkeypatch):
+    # Given a copy cap of 10 bytes and a producer that keeps recording
+    monkeypatch.setattr(api, "BUFFERED_MAX_BYTES", 10)
+    transport = _CapTransport(live_pieces=2)
+    pulled = []
+
+    def producer():
+        for _ in range(100):
+            pulled.append(1)
+            yield b"x" * 4
+
+    # When the live request hits the cap and the producer is drained
+    with pytest.raises(aai.DictationError) as exc:
+        _transcriber_over(transport).transcribe_live(producer(), config=_PCM)
+
+    # Then draining stops once the copy passes the cap, with a 413
+    # audio_too_large, and nothing is sent
+    assert exc.value.status_code == 413
+    assert exc.value.error_code == "audio_too_large"
+    assert len(pulled) == 3
+    assert transport.buffered == []
+
+
+def test_buffered_max_bytes_matches_the_buffered_endpoint():
+    assert api.BUFFERED_MAX_BYTES == 256 * 1024 * 1024
+    assert api.ENDPOINT_TRANSCRIBE == "/v1/transcribe"
+
+
+def test_a_producer_error_while_draining_propagates_without_fallback():
+    # Given a producer that fails after the live 413
+    transport = _CapTransport(live_pieces=2)
+
+    def producer():
+        yield b"A"
+        raise ValueError("microphone unplugged")
+
+    # When the dictation runs, then the producer's error propagates unchanged
+    # and nothing is sent
+    with pytest.raises(ValueError, match="microphone unplugged"):
+        _transcriber_over(transport).transcribe_live(producer(), config=_PCM)
+
+    assert transport.buffered == []
+
+
+def test_open_live_falls_back_with_audio_written_after_the_413():
+    # Given a session whose live request is answered 413 after the first chunk
+    transport = _CapTransport(live_pieces=2)
+
+    with _transcriber_over(transport) as transcriber:
+        session = transcriber.open_live(_PCM)
+        session.write(b"A" * 4)
+        assert transport.answered.wait(timeout=5.0)
+
+        # When the speaker keeps talking, then stops
+        session.write(b"B" * 4)
+        session.write(b"C" * 4)
+        session.close()
+        result = session.result()
+
+    # Then the buffered request carries everything written, and its result is
+    # the session's
+    _assert_buffered(
+        transport, b"A" * 4 + b"B" * 4 + b"C" * 4, {"sample_rate": 16000, "channels": 1}
+    )
+    assert result.session_id == _BUFFERED_OK["session_id"]
+
+
+def test_open_live_abort_after_the_413_sends_no_fallback():
+    # Given a session whose live request was answered 413 mid-recording
+    transport = _CapTransport(live_pieces=2)
+
+    with _transcriber_over(transport) as transcriber:
+        session = transcriber.open_live(_PCM)
+        session.write(b"A")
+        assert transport.answered.wait(timeout=5.0)
+        session.write(b"B")
+
+        # When it is aborted while the producer is being drained
+        session.abort()
+
+        # Then no buffered request is made and there is no result
+        assert transport.buffered == []
+        with pytest.raises(RuntimeError, match="aborted"):
+            session.result()
+
+
+def test_open_live_abort_after_close_sends_no_fallback():
+    # Given a closed session whose live 413 has not arrived yet
+    gate = threading.Event()
+    transport = _CapTransport(gate=gate)
+
+    with _transcriber_over(transport) as transcriber:
+        session = transcriber.open_live(_PCM)
+        session.write(b"A")
+        session.close()
+        assert transport.body_read.wait(timeout=5.0)
+
+        # When it is aborted before the 413 comes back
+        aborter = threading.Thread(target=session.abort)
+        aborter.start()
+        while not session._aborted:
+            time.sleep(0.001)
+        gate.set()
+        aborter.join(timeout=5.0)
+
+        # Then the fallback is not sent
+        assert transport.answered.is_set()
+        assert transport.buffered == []
+        with pytest.raises(RuntimeError, match="aborted"):
+            session.result()
+
+
+def test_the_copy_is_capped_during_the_live_upload_too(monkeypatch):
+    # Given a copy cap of 10 bytes, and a live endpoint that reads the whole
+    # body before answering — as production does
+    monkeypatch.setattr(api, "BUFFERED_MAX_BYTES", 10)
+    transport = _CapTransport()
+    pulled = []
+
+    def producer():
+        for _ in range(100):
+            pulled.append(1)
+            yield b"x" * 4
+
+    # When the copy passes the cap while the live request is still uploading
+    with pytest.raises(aai.DictationError) as exc:
+        _transcriber_over(transport).transcribe_live(producer(), config=_PCM)
+
+    # Then the upload stops there with a 413 audio_too_large; memory stays
+    # bounded rather than growing until the speaker stops
+    assert exc.value.status_code == 413
+    assert exc.value.error_code == "audio_too_large"
+    assert len(pulled) == 3
+    assert transport.buffered == []
+
+
+def test_the_copy_cap_does_not_apply_without_the_fallback(monkeypatch):
+    # Given no fallback, so no copy is kept
+    monkeypatch.setattr(api, "BUFFERED_MAX_BYTES", 10)
+    transport = _CapTransport(live_status=200, live_json=_OK_RESPONSE)
+
+    # When more than the cap is uploaded, then it goes out as before
+    _transcriber_over(transport, fallback_to_buffered=False).transcribe_live(
+        _chunks(*([b"x" * 4] * 10)), config=_PCM
+    )
+    assert transport.live_body.count(b"x") == 40
diff --git a/tests/unit/test_dictation_async.py b/tests/unit/test_dictation_async.py
index ae839a9..3d26374 100644
--- a/tests/unit/test_dictation_async.py
+++ b/tests/unit/test_dictation_async.py
@@ -8,8 +8,9 @@
 import asyncio
 import email
 import io
+import json
 import threading
-from typing import List, Optional
+from typing import AsyncIterable, List, Optional, cast
 
 import httpx
 import pytest
@@ -17,13 +18,14 @@
 
 import assemblyai as aai
 from assemblyai._multipart import _STREAM_READ_SIZE, aiter_chunks
-from assemblyai.dictation.v1 import async_api
+from assemblyai.dictation.v1 import api, async_api
 
 pytestmark = pytest.mark.asyncio
 
 aai.settings.api_key = "test"
 
 LIVE_URL = f"{aai.settings.dictation_base_url}/v1/transcribe/live"
+BUFFERED_URL = f"{aai.settings.dictation_base_url}/v1/transcribe"
 WARM_URL = f"{aai.settings.dictation_base_url}/v1/warm"
 
 _OK_RESPONSE = {
@@ -663,8 +665,10 @@ async def test_live_session_surfaces_server_errors(httpx_mock: HTTPXMock):
     assert exc.value.retry_after == 3
 
 
-async def test_the_buffered_endpoint_is_never_requested(httpx_mock: HTTPXMock):
-    """No entry point on the asyncio client posts to `/v1/transcribe`."""
+async def test_the_buffered_endpoint_is_not_requested_within_the_cap(
+    httpx_mock: HTTPXMock,
+):
+    """No asyncio entry point posts to `/v1/transcribe` while live succeeds."""
     for _ in range(4):  # one per request; older pytest-httpx has no is_reusable
         _mock_ok(httpx_mock)
 
@@ -687,3 +691,338 @@ def test_open_live_needs_a_running_loop():
     # When a session is opened, then the mistake is named rather than deferred
     with pytest.raises(RuntimeError, match="no running event loop"):
         transcriber.open_live()
+
+
+# --- buffered fallback past the live 120 s cap --------------------------------
+
+_LIVE_CAP = {"error": "audio exceeds 120000 ms", "error_code": "audio_too_large"}
+
+_BUFFERED_OK = {
+    **_OK_RESPONSE,
+    "audio_duration_ms": 130000,
+    "session_id": "4f0c1d7e-buffered",
+}
+
+_PCM = aai.DictationConfig(sample_rate=16000, channels=1)
+
+
+class _AsyncCapTransport(httpx.AsyncBaseTransport):
+    """
+    The asyncio twin of `test_dictation._CapTransport`: the live body is read
+    for `live_pieces` pieces (whole when None) and answered with
+    `live_status`/`live_json`, the way httpcore stops pulling a body once the
+    server has answered; the buffered body is read whole and answered with
+    `buffered_status`/`buffered_json`.
+    """
+
+    def __init__(
+        self,
+        *,
+        live_status: int = 413,
+        live_json: Optional[dict] = None,
+        live_pieces: Optional[int] = None,
+        buffered_status: int = 200,
+        buffered_json: Optional[dict] = None,
+        gate: Optional[asyncio.Event] = None,
+    ) -> None:
+        self.live_status = live_status
+        self.live_json = _LIVE_CAP if live_json is None else live_json
+        self.live_pieces = live_pieces
+        self.buffered_status = buffered_status
+        self.buffered_json = _BUFFERED_OK if buffered_json is None else buffered_json
+        self.gate = gate
+        self.live_request: Optional[httpx.Request] = None
+        self.live_body = b""
+        self.body_read = asyncio.Event()
+        self.answered = asyncio.Event()
+        self.buffered: List[tuple] = []
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/v1/transcribe/live":
+            self.live_request = request
+            pieces: List[bytes] = []
+            stream = cast(AsyncIterable[bytes], request.stream).__aiter__()
+            while self.live_pieces is None or len(pieces) < self.live_pieces:
+                try:
+                    pieces.append(await stream.__anext__())
+                except StopAsyncIteration:
+                    break
+            self.live_body = b"".join(pieces)
+            self.body_read.set()
+            if self.gate is not None:
+                await asyncio.wait_for(self.gate.wait(), timeout=5.0)
+            self.answered.set()
+            return httpx.Response(self.live_status, json=self.live_json)
+
+        body = b"".join(
+            [piece async for piece in cast(AsyncIterable[bytes], request.stream)]
+        )
+        self.buffered.append((request, body))
+        return httpx.Response(self.buffered_status, json=self.buffered_json)
+
+
+def _transcriber_over(transport: httpx.AsyncBaseTransport, **kwargs):
+    """An `AsyncDictationTranscriber` whose client sends through `transport`."""
+
+    client = aai.AsyncClient()
+    client._http_client = httpx.AsyncClient(
+        transport=transport, headers=dict(client.http_client.headers)
+    )
+    return aai.AsyncDictationTranscriber(client=client, **kwargs)
+
+
+def _body_parts(content_type: str, body: bytes):
+    raw = b"Content-Type: " + content_type.encode() + b"\r\n\r\n"
+    message = email.message_from_bytes(raw + body)
+
+    return {
+        part.get_param("name", header="content-disposition"): part
+        for part in message.get_payload()
+    }
+
+
+def _audio_head(body: bytes) -> bytes:
+    start = body.index(b'name="audio"')
+    return body[start : body.index(b"\r\n\r\n", start)]
+
+
+def _assert_buffered(transport, audio: bytes, config: dict) -> None:
+    """The fallback posted `audio` whole, framed exactly like the live upload."""
+
+    assert len(transport.buffered) == 1
+    request, body = transport.buffered[0]
+    assert str(request.url) == BUFFERED_URL
+    assert request.headers["content-length"] == str(len(body))
+    assert "transfer-encoding" not in request.headers
+
+    parts = _body_parts(request.headers["content-type"], body)
+    assert list(parts) == ["config", "audio"]
+    assert json.loads(parts["config"].get_payload(decode=True)) == config
+    assert parts["audio"].get_payload(decode=True) == audio
+    assert _audio_head(body) == _audio_head(transport.live_body)
+
+
+async def test_fallback_is_on_by_default_and_can_be_turned_off():
+    assert aai.AsyncDictationTranscriber().fallback_to_buffered is True
+    assert (
+        aai.AsyncDictationTranscriber(fallback_to_buffered=False).fallback_to_buffered
+        is False
+    )
+
+
+@pytest.mark.parametrize("fallback", [True, False])
+async def test_fallback_leaves_the_success_path_unchanged(fallback):
+    # Given a dictation within the cap
+    transport = _AsyncCapTransport(live_status=200, live_json=_OK_RESPONSE)
+
+    # When it is sent with the fallback on or off
+    result = await _transcriber_over(
+        transport, fallback_to_buffered=fallback
+    ).transcribe_live(_achunks(b"\x01\x02", b"\x03\x04"), config=_PCM)
+
+    # Then it is the live result, from the live route only
+    assert result.session_id == _OK_RESPONSE["session_id"]
+    assert transport.buffered == []
+    assert transport.live_request.headers["transfer-encoding"] == "chunked"
+    parts = _body_parts(
+        transport.live_request.headers["content-type"], transport.live_body
+    )
+    assert parts["audio"].get_payload(decode=True) == b"\x01\x02\x03\x04"
+
+
+async def test_live_cap_mid_upload_falls_back_with_all_the_audio():
+    # Given a 413 audio_too_large after the head and the first chunk, while
+    # the async producer is still recording
+    transport = _AsyncCapTransport(live_pieces=2)
+    pulled = []
+
+    async def producer():
+        for chunk in (b"A" * 4, b"B" * 4, b"C" * 4, b"D" * 4):
+            pulled.append((chunk[:1], transport.answered.is_set()))
+            yield chunk
+
+    # When the dictation runs
+    result = await _transcriber_over(transport).transcribe_live(producer(), config=_PCM)
+
+    # Then the producer went on being read after the 413, and the whole
+    # recording went to the buffered endpoint once
+    assert pulled == [(b"A", False), (b"B", True), (b"C", True), (b"D", True)]
+    _assert_buffered(
+        transport,
+        b"A" * 4 + b"B" * 4 + b"C" * 4 + b"D" * 4,
+        {"sample_rate": 16000, "channels": 1},
+    )
+    assert result.session_id == _BUFFERED_OK["session_id"]
+    assert result.audio_duration_ms == 130000
+
+
+async def test_live_cap_after_the_whole_body_falls_back():
+    # Given a 413 that arrives once the whole body was sent
+    transport = _AsyncCapTransport()
+
+    # When the dictation runs from a plain iterable
+    result = await _transcriber_over(transport).transcribe_live(
+        [b"RIFF", b"wav-1", b"wav-2"]
+    )
+
+    # Then the buffered request carries the same audio and the empty config
+    _assert_buffered(transport, b"RIFFwav-1wav-2", {})
+    assert result.session_id == _BUFFERED_OK["session_id"]
+
+
+async def test_fallback_disabled_raises_the_live_413_as_before():
+    transport = _AsyncCapTransport(live_pieces=2)
+    pulled = []
+
+    async def producer():
+        for chunk in (b"A", b"B", b"C"):
+            pulled.append(chunk)
+            yield chunk
+
+    with pytest.raises(aai.DictationError) as exc:
+        await _transcriber_over(transport, fallback_to_buffered=False).transcribe_live(
+            producer(), config=_PCM
+        )
+
+    assert exc.value.status_code == 413
+    assert exc.value.error_code == "audio_too_large"
+    assert pulled == [b"A"]
+    assert transport.buffered == []
+
+
+@pytest.mark.parametrize(
+    "status, body",
+    [
+        (413, {"error": "body too large", "error_code": "payload_too_large"}),
+        (413, {"detail": "too large"}),
+        (429, {"status": 429, "title": "Too Many Requests"}),
+        (500, {"error": "boom", "error_code": "audio_too_large"}),
+    ],
+)
+async def test_other_failures_do_not_fall_back(status, body):
+    transport = _AsyncCapTransport(live_status=status, live_json=body, live_pieces=2)
+
+    with pytest.raises(aai.DictationError) as exc:
+        await _transcriber_over(transport).transcribe_live(
+            _achunks(b"A", b"B"), config=_PCM
+        )
+
+    assert exc.value.status_code == status
+    assert transport.buffered == []
+
+
+async def test_a_failed_fallback_raises_its_own_error():
+    # Given a buffered request refused past its own 20-minute cap
+    transport = _AsyncCapTransport(
+        buffered_status=413,
+        buffered_json={
+            "error": "audio exceeds 1200000 ms",
+            "error_code": "audio_too_large",
+        },
+    )
+
+    with pytest.raises(aai.DictationError) as exc:
+        await _transcriber_over(transport).transcribe_live(_achunks(b"A"), config=_PCM)
+
+    # Then that error is raised, not the live one
+    assert len(transport.buffered) == 1
+    assert "1200000 ms" in str(exc.value)
+
+
+async def test_the_copy_is_capped_while_draining(monkeypatch):
+    monkeypatch.setattr(api, "BUFFERED_MAX_BYTES", 10)
+    transport = _AsyncCapTransport(live_pieces=2)
+    pulled = []
+
+    async def producer():
+        for _ in range(100):
+            pulled.append(1)
+            yield b"x" * 4
+
+    with pytest.raises(aai.DictationError) as exc:
+        await _transcriber_over(transport).transcribe_live(producer(), config=_PCM)
+
+    assert exc.value.status_code == 413
+    assert exc.value.error_code == "audio_too_large"
+    assert len(pulled) == 3
+    assert transport.buffered == []
+
+
+async def test_open_live_falls_back_with_audio_written_after_the_413():
+    transport = _AsyncCapTransport(live_pieces=2)
+
+    async with _transcriber_over(transport) as transcriber:
+        session = transcriber.open_live(_PCM)
+        session.write(b"A" * 4)
+        await asyncio.wait_for(transport.answered.wait(), timeout=5.0)
+
+        # When the speaker keeps talking, then stops
+        session.write(b"B" * 4)
+        session.write(b"C" * 4)
+        session.close()
+        result = await session.result()
+
+    _assert_buffered(
+        transport, b"A" * 4 + b"B" * 4 + b"C" * 4, {"sample_rate": 16000, "channels": 1}
+    )
+    assert result.session_id == _BUFFERED_OK["session_id"]
+
+
+async def test_open_live_abort_after_the_413_sends_no_fallback():
+    transport = _AsyncCapTransport(live_pieces=2)
+
+    async with _transcriber_over(transport) as transcriber:
+        session = transcriber.open_live(_PCM)
+        session.write(b"A")
+        await asyncio.wait_for(transport.answered.wait(), timeout=5.0)
+        session.write(b"B")
+
+        await session.abort()
+
+        assert transport.buffered == []
+        with pytest.raises(RuntimeError, match="aborted"):
+            await session.result()
+
+
+async def test_open_live_abort_after_close_sends_no_fallback():
+    # Given a closed session whose live 413 has not arrived yet
+    gate = asyncio.Event()
+    transport = _AsyncCapTransport(gate=gate)
+
+    async with _transcriber_over(transport) as transcriber:
+        session = transcriber.open_live(_PCM)
+        session.write(b"A")
+        session.close()
+        await asyncio.wait_for(transport.body_read.wait(), timeout=5.0)
+
+        # When it is aborted before the 413 comes back
+        aborter = asyncio.ensure_future(session.abort())
+        await asyncio.sleep(0)
+        assert session._aborted
+        gate.set()
+        await asyncio.wait_for(aborter, timeout=5.0)
+
+        # Then the fallback is not sent
+        assert transport.answered.is_set()
+        assert transport.buffered == []
+        with pytest.raises(RuntimeError, match="aborted"):
+            await session.result()
+
+
+async def test_the_copy_is_capped_during_the_live_upload_too(monkeypatch):
+    monkeypatch.setattr(api, "BUFFERED_MAX_BYTES", 10)
+    transport = _AsyncCapTransport()
+    pulled = []
+
+    async def producer():
+        for _ in range(100):
+            pulled.append(1)
+            yield b"x" * 4
+
+    with pytest.raises(aai.DictationError) as exc:
+        await _transcriber_over(transport).transcribe_live(producer(), config=_PCM)
+
+    assert exc.value.status_code == 413
+    assert exc.value.error_code == "audio_too_large"
+    assert len(pulled) == 3
+    assert transport.buffered == []