diff --git a/packages/backend/src/baseline/types.ts b/packages/backend/src/baseline/types.ts
index b361b883..8c25a45b 100644
--- a/packages/backend/src/baseline/types.ts
+++ b/packages/backend/src/baseline/types.ts
@@ -3,6 +3,7 @@ import type {
   ConsoleLog,
   Metadata,
   NetworkRequest,
+  ScreencastFrame,
   TestError,
   TestStatus,
   TraceMutation
@@ -54,4 +55,9 @@ export interface ActiveRun {
   /** Raw `logs` frames, the trace's transcript source. Only the JS adapters
    *  send these, so this is routinely empty. */
   traceLogs: string[]
+  /** Dense screencast frames for the trace filmstrip. The JS adapters hand
+   *  their recorder's buffer straight to the exporter in-process; an adapter
+   *  that exports through here has to send them, so they accumulate like any
+   *  other stream. Empty unless the adapter asked for a filmstrip. */
+  screencastFrames: ScreencastFrame[]
 }
diff --git a/packages/backend/src/baseline/utils.ts b/packages/backend/src/baseline/utils.ts
index 6f5825cf..4ad48326 100644
--- a/packages/backend/src/baseline/utils.ts
+++ b/packages/backend/src/baseline/utils.ts
@@ -9,7 +9,8 @@ export function freshRun(): ActiveRun {
     sources: {},
     nodes: new Map(),
     startedAt: Date.now(),
-    traceLogs: []
+    traceLogs: [],
+    screencastFrames: []
   }
 }
 
diff --git a/packages/backend/src/baselineStore.ts b/packages/backend/src/baselineStore.ts
index c23420cf..9b89c948 100644
--- a/packages/backend/src/baselineStore.ts
+++ b/packages/backend/src/baselineStore.ts
@@ -82,6 +82,11 @@ class BaselineStore {
       case 'logs':
         appendArray(this.#activeRun.traceLogs, data)
         return
+      case 'screencastFrames':
+        // Sent in batches: a run's buffer can reach the recorder's cap, and one
+        // message carrying all of it would sit near the socket's payload limit.
+        appendArray(this.#activeRun.screencastFrames, data)
+        return
       case 'suites':
         this.#ingestSuites(data)
         return
diff --git a/packages/backend/src/trace-export.ts b/packages/backend/src/trace-export.ts
index 872cc0f1..91c4686d 100644
--- a/packages/backend/src/trace-export.ts
+++ b/packages/backend/src/trace-export.ts
@@ -82,6 +82,12 @@ export async function exportActiveRunTrace(
     sessionId: request.sessionId,
     ...(request.format ? { format: request.format } : {}),
     ...(request.fileStem ? { fileStem: request.fileStem } : {}),
+    // Omitted when empty rather than passed as []: the exporter treats absence
+    // as "no dense filmstrip" and keeps the sparse per-action one, which is
+    // what an adapter that did not ask for frames should still get.
+    ...(run.screencastFrames.length
+      ? { screencastFrames: run.screencastFrames }
+      : {}),
     testMetadata: testMetadataFromNodes(run.nodes)
   })
 }
diff --git a/packages/backend/tests/trace-export.test.ts b/packages/backend/tests/trace-export.test.ts
index 9e992a3f..ad4017d6 100644
--- a/packages/backend/tests/trace-export.test.ts
+++ b/packages/backend/tests/trace-export.test.ts
@@ -26,6 +26,13 @@ import {
 import { freshRun } from '../src/baseline/utils.js'
 import type { ActiveRun, TimeWindowNode } from '../src/baseline/types.js'
 
+/** Smallest valid JPEG — the exporter content-addresses frame bytes, so they
+ *  have to be real image data rather than a placeholder string. */
+const JPEG_1PX =
+  '/9j/4AAQSkZJRgABAQEAYABgAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0a' +
+  'HBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/wAALCAABAAEBAREA/8QAFAABAAAAAAAA' +
+  'AAAAAAAAAAAACf/EABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEAAD8AKp//2Q=='
+
 const dirs: string[] = []
 
 afterEach(async () => {
@@ -248,6 +255,43 @@ describe('exportActiveRunTrace', () => {
     expect(JSON.stringify(options[0])).toContain('chrome')
   })
 
+  // The JS adapters hand their recorder's buffer straight to the exporter
+  // in-process; an adapter exporting through the backend has to send it, so the
+  // frames arrive as a stream and have to survive the round trip into the zip.
+  it('writes the dense filmstrip when frames were streamed', async () => {
+    const outputDir = await tmpDir()
+    const zipPath = await exportActiveRunTrace(
+      run({
+        screencastFrames: [
+          { data: JPEG_1PX, timestamp: 1100 },
+          { data: JPEG_1PX, timestamp: 1200 }
+        ]
+      }),
+      { outputDir, sessionId: 'sess-film' }
+    )
+
+    const events = await traceEvents(zipPath)
+    const frames = events.filter(
+      (e) => e.type === TRACE_EVENT_TYPES.screencastFrame
+    )
+    expect(frames.length).toBeGreaterThan(0)
+  })
+
+  // Absence means "no dense filmstrip", which keeps the sparse per-action one.
+  // Passing an empty array instead would be a different thing to the exporter.
+  it('writes no screencast-frame events when none were streamed', async () => {
+    const outputDir = await tmpDir()
+    const zipPath = await exportActiveRunTrace(run(), {
+      outputDir,
+      sessionId: 'sess-nofilm'
+    })
+
+    const frames = (await traceEvents(zipPath)).filter(
+      (e) => e.type === TRACE_EVENT_TYPES.screencastFrame
+    )
+    expect(frames).toEqual([])
+  })
+
   it('honours fileStem so a per-test slice can name its own artifact', async () => {
     const outputDir = await tmpDir()
     const zipPath = await exportActiveRunTrace(run(), {
diff --git a/packages/selenium-devtools-py/scripts/gen_contract.py b/packages/selenium-devtools-py/scripts/gen_contract.py
index 8620d984..b54452f4 100644
--- a/packages/selenium-devtools-py/scripts/gen_contract.py
+++ b/packages/selenium-devtools-py/scripts/gen_contract.py
@@ -42,6 +42,9 @@
     "SCOPE_SCREENCAST": "screencast",
     "SCOPE_SOURCES": "sources",
     "SCOPE_MUTATIONS": "mutations",
+    # The dense filmstrip. The JS adapters hand their recorder's buffer to the
+    # exporter in-process; an adapter exporting through the backend sends it.
+    "SCOPE_SCREENCAST_FRAMES": "screencastFrames",
 }
 
 
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/__init__.py b/packages/selenium-devtools-py/src/selenium_devtools/__init__.py
index ecce0f22..8f70c70b 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/__init__.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/__init__.py
@@ -30,6 +30,7 @@
     DEFAULT_HOST,
     DEFAULT_PORT,
     ENV_HOST,
+    ENV_FILMSTRIP,
     ENV_PORT,
     ENV_TRACE,
     LOGGER_NAME,
@@ -75,10 +76,22 @@ def _restore_excepthook() -> None:
 _active: dict = {
     "capturer": None, "transport": None, "process": None, "url": None,
     "handle": None, "terminal": None, "logs": None, "excepthook": None,
-    "trace": False, "traced": False,
+    "trace": False, "traced": False, "filmstrip_mark": None,
 }
 
 
+def _filmstrip_enabled(filmstrip: Optional[bool]) -> bool:
+    """Whether trace mode records a dense filmstrip. Default ON, as in the JS
+    adapters (`BaseDevToolsOptions.filmstrip`), opt-out only — the argument
+    wins, then the environment."""
+    if filmstrip is not None:
+        return filmstrip
+    value = os.environ.get(ENV_FILMSTRIP)
+    if value is None:
+        return True
+    return value.lower() not in ("0", "false", "no", "off", "")
+
+
 def _trace_enabled(trace: Optional[bool]) -> bool:
     """Whether this run writes a trace archive. The argument wins over the
     environment so a script can opt out of an exported default."""
@@ -148,6 +161,40 @@ def _export_trace(
         session_id = (
             getattr(capturer, "session_id", None) or resolve_run_id()
         )
+        # Before the request, not with it: the buffer holds up to a couple of
+        # thousand JPEGs, and the backend accumulates them like any other
+        # stream.
+        #
+        # Only what has not gone out already. A failed export is retried at
+        # teardown, and the backend APPENDS these — it has no key to replace or
+        # dedupe on — so resending the buffer would put every frame in the trace
+        # twice.
+        #
+        # Keyed on the newest timestamp sent, NOT on how many were sent. A live
+        # recorder decimates its bounded buffer in place
+        # (`screencast._decimate` halves it, keeping the ends), so between two
+        # attempts the list can SHRINK and every index shift — an offset would
+        # then skip frames it never sent. Decimation drops frames but never
+        # renumbers the survivors, and timestamps are monotonic per recorder, so
+        # the watermark stays meaningful however the buffer is rewritten.
+        # None, not 0: "nothing sent yet" is not a timestamp, and a frame
+        # stamped 0 would be filtered out by one.
+        #
+        # The boundary is INCLUSIVE, so a retry may resend the frame the last
+        # attempt ended on. Timestamps are milliseconds and two frames can share
+        # one, and an exclusive boundary would drop the unsent twin — a gap in
+        # the filmstrip. A resend costs nothing much instead: the exporter
+        # content-addresses frame bytes, so the duplicate shares one resource.
+        # Losing a frame beats duplicating one only if you never look at it.
+        mark = _active["filmstrip_mark"]
+        pending = [
+            f
+            for f in instrumentation.screencast_frames()
+            if mark is None or f.get("timestamp", 0) >= mark
+        ]
+        sent = trace_export.send_frames(_active["transport"], pending)
+        if sent:
+            _active["filmstrip_mark"] = pending[sent - 1].get("timestamp", mark)
         return trace_export.export(
             _active["transport"],
             output_dir=output_dir or resolve_adapter_output_dir(),
@@ -164,6 +211,7 @@ def enable(
     *,
     webdriver_cls: Optional[type] = None,
     trace: Optional[bool] = None,
+    filmstrip: Optional[bool] = None,
 ) -> Optional[SessionCapturer]:
     """Connect to the backend and instrument Selenium. Idempotent.
 
@@ -178,6 +226,7 @@ def enable(
     # Decided before anything reads it: the screencast recorder, the dashboard
     # window and the teardown export all branch on this.
     trace_mode = _trace_enabled(trace)
+    filmstrip_mode = trace_mode and _filmstrip_enabled(filmstrip)
 
     # Before the backend is launched: the directory a rerun spawns in travels
     # through the environment the backend process inherits. A framework plugin
@@ -212,7 +261,9 @@ def enable(
         return None
 
     capturer = SessionCapturer(transport)
-    instrumentation.install(capturer, webdriver_cls, trace=trace_mode)
+    instrumentation.install(
+        capturer, webdriver_cls, trace=trace_mode, filmstrip=filmstrip_mode
+    )
     # Plain scripts only: a framework plugin calls
     # `set_external_suites`, which turns this back off.
     instrumentation.start_assertion_tracing(capturer)
@@ -226,7 +277,7 @@ def enable(
     url = f"http://{host}:{port}"
     _active.update(
         capturer=capturer, transport=transport, process=process, url=url,
-        terminal=term, logs=logs, trace=trace_mode, traced=False,
+        terminal=term, logs=logs, trace=trace_mode, traced=False, filmstrip_mark=None,
     )
 
     # Open the dashboard window and wire exit/signal + control-frame teardown so
@@ -289,7 +340,7 @@ def disable() -> None:
     trace_export.reset()
     _active.update(
         capturer=None, transport=None, process=None, url=None, handle=None,
-        terminal=None, logs=None, excepthook=None, trace=False, traced=False,
+        terminal=None, logs=None, excepthook=None, trace=False, traced=False, filmstrip_mark=None,
     )
 
 
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/_contract.py b/packages/selenium-devtools-py/src/selenium_devtools/_contract.py
index 8857e937..74ea76c0 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/_contract.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/_contract.py
@@ -10,6 +10,7 @@
 SCOPE_SCREENCAST = "screencast"
 SCOPE_SOURCES = "sources"
 SCOPE_MUTATIONS = "mutations"
+SCOPE_SCREENCAST_FRAMES = "screencastFrames"
 SCOPE_REPLACE_COMMAND = "replaceCommand"
 
 DATA_SCOPES = frozenset(['actionSnapshots', 'commands', 'config', 'consoleLogs', 'logs', 'metadata', 'mutations', 'networkRequests', 'screencast', 'screencastFrames', 'sources', 'suites'])
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/constants.py b/packages/selenium-devtools-py/src/selenium_devtools/constants.py
index b3ead544..0e281dbd 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/constants.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/constants.py
@@ -85,6 +85,16 @@
 #: Opt in to writing a trace archive at the end of the run.
 ENV_TRACE = "DEVTOOLS_TRACE"
 
+#: Opt OUT of the dense filmstrip in trace mode (on by default, as in the JS
+#: adapters). Falsy values disable it.
+ENV_FILMSTRIP = "DEVTOOLS_FILMSTRIP"
+
+#: Filmstrip frames per websocket message. The buffer can hold
+#: SCREENCAST_MAX_BUFFER_FRAMES JPEGs, which in one message would approach the
+#: socket's payload limit; the transport also masks payloads in a per-byte
+#: Python loop (~57 MB/s measured), so smaller messages keep the stall short.
+SCREENCAST_FRAME_BATCH = 50
+
 MIN_NODE_MAJOR = 18
 NODE_VERSION_TIMEOUT_S = 5.0
 
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
index 3edea737..a3d13ec0 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
@@ -246,10 +246,15 @@ def _capture_source(capturer: SessionCapturer, call_src: Optional[str]) -> None:
     # Set by enable()'s excepthook when an exception reaches top level. The
     # synthetic suite's final state reads this rather than assuming success.
     "run_failed": False,
-    # Trace mode. The archive carries per-command screenshots, not the
-    # screencast — `screencastFrames` does not cross the wire yet — so
-    # recording one writes a .webm nothing reads. Set by install().
+    # Trace mode. Set by install().
     "trace": False,
+    # Record a dense filmstrip into the trace. Only meaningful in trace mode;
+    # the recorder runs in live mode regardless, for the dashboard video.
+    "filmstrip": False,
+    # Frames kept from every session's recorder as it finalizes, because that
+    # is the only moment they are reachable. Run-scoped: a run may replace its
+    # driver, and the filmstrip is the whole run.
+    "filmstrip_frames": [],
 }
 
 
@@ -325,6 +330,34 @@ def _attach_performance(
         _log.debug("could not replace the navigation row: %s", exc)
 
 
+def screencast_frames() -> list:
+    """Every frame this run's recorders buffered, in time order.
+
+    Two sources, because a frame is only reachable from one of them at a time:
+
+    * sessions that already quit — kept as they finalized, since
+      `_finalize_screencast` pops the recorder off the entry and `sessions` is
+      keyed weakly by a driver about to be collected;
+    * sessions still live — read straight off their recorder, because an export
+      can run before the last driver quits, and `uninstall` then stops those
+      recorders without keeping anything.
+
+    Sorted rather than concatenated: sessions are appended in quit order, and a
+    replaced session can finish after a later one started.
+    """
+    frames: list = list(_state["filmstrip_frames"])
+    for entry in list(_state.get("sessions", {}).values()):
+        recorder = entry.get("screencast")
+        if recorder is None:
+            continue
+        try:
+            frames.extend(recorder.frames)
+        except Exception as exc:  # noqa: BLE001 — a poorer filmstrip, not a failed run
+            _log.debug("could not read a live recorder's frames: %s", exc)
+    frames.sort(key=lambda f: f.get("timestamp", 0))
+    return frames
+
+
 def resolved_output_dir() -> Optional[str]:
     """The ``test-results`` dir this run resolved from its first test file, or
     None if no command carried a user call source. Screencast videos already
@@ -458,6 +491,21 @@ def _finalize_screencast(
     recorder = entry.pop("screencast", None)
     if recorder is None:
         return
+    # Take the buffer BEFORE anything else. This is the last moment it exists:
+    # the entry drops its recorder on the line above, and `sessions` is keyed
+    # weakly by a driver that quit and is about to be collected — so reading it
+    # at export time, as this first did, always found nothing.
+    if _state["filmstrip"]:
+        try:
+            _state["filmstrip_frames"].extend(recorder.frames)
+        except Exception as exc:  # noqa: BLE001 — a poorer filmstrip, not a failed run
+            _log.debug("could not keep the filmstrip frames: %s", exc)
+    if _state["trace"]:
+        # No video in trace mode: the frames ARE the filmstrip, and a .webm is
+        # a live-dashboard artifact with no dashboard to play it. Stopping
+        # rather than finalizing is what skips the encode.
+        recorder.stop()
+        return
     try:
         info = recorder.finalize(session_id, output_dir=_state.get("output_dir"))
     except Exception as exc:  # noqa: BLE001
@@ -517,9 +565,11 @@ def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> Optional[di
     except Exception as exc:  # noqa: BLE001 — capture must never break the test
         _log.warning("BiDi attach threw: %s", exc)
     try:
-        if _state["trace"]:
-            # The archive's frames are the per-command screenshots; the video is
-            # a live-dashboard artifact, and trace mode opens no dashboard.
+        if _state["trace"] and not _state["filmstrip"]:
+            # With no filmstrip the archive's frames are the per-command
+            # screenshots alone, and the video is a live-dashboard artifact —
+            # trace mode opens no dashboard, so a recorder here writes a .webm
+            # nothing reads.
             raise _SkipScreencast
         recorder = ScreencastRecorder()
         recorder.start(driver)
@@ -727,6 +777,7 @@ def install(
     webdriver_cls: Optional[type] = None,
     *,
     trace: bool = False,
+    filmstrip: bool = False,
 ) -> None:
     if _state["installed"]:
         return
@@ -807,7 +858,7 @@ def patched_execute(self, driver_command: str, params: Any = None):  # noqa: ANN
     _state.update(
         installed=True, cls=webdriver_cls, orig=orig_execute,
         sessions=weakref.WeakKeyDictionary(), output_dir=None, default_suite=None,
-        trace=trace,
+        trace=trace, filmstrip=filmstrip, filmstrip_frames=[],
     )
 
 
@@ -820,11 +871,20 @@ def uninstall() -> None:
     # per-run state.
     reset_collector_cache()
     # Never leave a recorder running past teardown, for any session still live.
+    # Keep the buffer first: `disable()` uninstalls BEFORE its fallback export,
+    # and `sessions` is replaced below, so a session that never quit would
+    # otherwise have its frames dropped here — the plain-script path exactly.
     for entry in list(_state.get("sessions", {}).values()):
         _stop_push_screencast(entry)
         recorder = entry.get("screencast")
-        if recorder is not None:
-            recorder.stop()
+        if recorder is None:
+            continue
+        if _state["filmstrip"]:
+            try:
+                _state["filmstrip_frames"].extend(recorder.frames)
+            except Exception as exc:  # noqa: BLE001 — a poorer filmstrip, not a failed run
+                _log.debug("could not keep a live recorder's frames: %s", exc)
+        recorder.stop()
     if not _state["installed"]:
         _state.update(
             sessions=weakref.WeakKeyDictionary(),
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/trace_export.py b/packages/selenium-devtools-py/src/selenium_devtools/trace_export.py
index 1a3026d9..a6fbb335 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/trace_export.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/trace_export.py
@@ -28,8 +28,12 @@
 from dataclasses import dataclass
 from typing import Any, Optional
 
-from ._contract import SCOPE_TRACE_EXPORT
-from .constants import LOGGER_NAME, TRACE_EXPORT_TIMEOUT_S
+from ._contract import SCOPE_SCREENCAST_FRAMES, SCOPE_TRACE_EXPORT
+from .constants import (
+    LOGGER_NAME,
+    SCREENCAST_FRAME_BATCH,
+    TRACE_EXPORT_TIMEOUT_S,
+)
 
 _log = logging.getLogger(f"{LOGGER_NAME}.trace")
 
@@ -78,6 +82,36 @@ def reset() -> None:
         _pending = None
 
 
+def send_frames(transport: Any, frames: list) -> int:
+    """Stream the filmstrip to the backend ahead of the export request.
+
+    In batches: a run's buffer reaches the recorder's cap of a couple of
+    thousand JPEG frames, and one message carrying all of them would sit near
+    the socket's payload limit — and the transport masks its payload in a
+    per-byte Python loop, measured at ~57 MB/s, so a single huge frame is a
+    visible stall as well as a risk.
+
+    Returns how many frames were accepted. A partial send is not an error worth
+    failing the run over: the exporter thins the filmstrip anyway, and fewer
+    frames is a poorer video rather than a broken trace.
+    """
+    if not frames or transport is None:
+        return 0
+    sent = 0
+    for start in range(0, len(frames), SCREENCAST_FRAME_BATCH):
+        batch = frames[start : start + SCREENCAST_FRAME_BATCH]
+        try:
+            if not transport.send_json(SCOPE_SCREENCAST_FRAMES, batch):
+                break
+        except Exception as exc:  # noqa: BLE001 — never break the test
+            _log.debug("filmstrip batch dropped: %s", exc)
+            break
+        sent += len(batch)
+    if sent:
+        _log.debug("streamed %d filmstrip frame(s) for the trace", sent)
+    return sent
+
+
 def export(
     transport: Any,
     *,
diff --git a/packages/selenium-devtools-py/tests/test_filmstrip.py b/packages/selenium-devtools-py/tests/test_filmstrip.py
new file mode 100644
index 00000000..c4aa65aa
--- /dev/null
+++ b/packages/selenium-devtools-py/tests/test_filmstrip.py
@@ -0,0 +1,433 @@
+"""The dense filmstrip in a Python trace.
+
+The JS adapters hand their recorder's buffer straight to the exporter
+in-process. An adapter that exports through the backend has to send it, so the
+filmstrip becomes a stream like any other — and the recorder, which trace mode
+otherwise switches off, has to run for it to exist at all.
+"""
+
+import unittest
+from unittest import mock
+
+from selenium_devtools import instrumentation, trace_export
+from selenium_devtools._contract import SCOPE_SCREENCAST_FRAMES
+from selenium_devtools.constants import SCREENCAST_FRAME_BATCH
+
+
+class FakeTransport:
+    def __init__(self, *, sends=True, raises_after=None):
+        self.connected = True
+        self.sent = []
+        self._sends = sends
+        self._raises_after = raises_after
+
+    def send_json(self, scope, data):
+        if self._raises_after is not None and len(self.sent) >= self._raises_after:
+            raise OSError("socket gone")
+        self.sent.append((scope, data))
+        return self._sends
+
+    def close(self):
+        self.connected = False
+
+
+def frames(n, *, start=0):
+    return [{"data": f"f{i}", "timestamp": start + i} for i in range(n)]
+
+
+class TestStreamingTheFilmstrip(unittest.TestCase):
+    def test_frames_go_out_under_the_screencast_frames_scope(self):
+        tx = FakeTransport()
+
+        self.assertEqual(trace_export.send_frames(tx, frames(3)), 3)
+
+        scope, batch = tx.sent[0]
+        self.assertEqual(scope, SCOPE_SCREENCAST_FRAMES)
+        self.assertEqual([f["data"] for f in batch], ["f0", "f1", "f2"])
+
+    # One message carrying a full buffer would sit near the socket's payload
+    # limit, and the transport masks payloads in a per-byte Python loop.
+    def test_a_large_buffer_is_batched(self):
+        tx = FakeTransport()
+        total = SCREENCAST_FRAME_BATCH * 2 + 7
+
+        self.assertEqual(trace_export.send_frames(tx, frames(total)), total)
+
+        self.assertEqual(len(tx.sent), 3)
+        self.assertEqual(len(tx.sent[0][1]), SCREENCAST_FRAME_BATCH)
+        self.assertEqual(len(tx.sent[-1][1]), 7)
+        # Every frame reaches the backend exactly once, in order.
+        streamed = [f["data"] for _, batch in tx.sent for f in batch]
+        self.assertEqual(streamed, [f["data"] for f in frames(total)])
+
+    # Fewer frames is a poorer video, not a broken trace — the run must not fail.
+    def test_a_refused_or_throwing_socket_stops_without_raising(self):
+        refused = FakeTransport(sends=False)
+        self.assertEqual(trace_export.send_frames(refused, frames(120)), 0)
+
+        throwing = FakeTransport(raises_after=1)
+        sent = trace_export.send_frames(throwing, frames(120))
+        self.assertEqual(sent, SCREENCAST_FRAME_BATCH)
+
+    def test_nothing_to_send_is_not_a_message(self):
+        tx = FakeTransport()
+        self.assertEqual(trace_export.send_frames(tx, []), 0)
+        self.assertEqual(trace_export.send_frames(None, frames(3)), 0)
+        self.assertEqual(tx.sent, [])
+
+
+class TestWhereTheFramesComeFrom(unittest.TestCase):
+    """Collected as each session finalizes, because that is the only moment
+    they are reachable.
+
+    The first version read them at export time from `_state["sessions"]`, which
+    is a WeakKeyDictionary keyed by driver — and `_finalize_screencast` pops the
+    recorder off the entry at quit anyway. So a real run streamed nothing and
+    the trace fell back to the sparse per-action strip; the unit test passed
+    because it populated `sessions` by hand with live recorders.
+    """
+
+    class Recorder:
+        def __init__(self, buf):
+            self._buf = buf
+            self.stopped = 0
+            self.finalized = 0
+
+        @property
+        def frames(self):
+            return list(self._buf)
+
+        def stop(self):
+            self.stopped += 1
+
+        def finalize(self, *a, **k):
+            self.finalized += 1
+            return None
+
+    def _finalize(self, recorder, *, trace, filmstrip):
+        from selenium_devtools.capturer import SessionCapturer
+
+        entry = {"screencast": recorder}
+        with mock.patch.dict(
+            instrumentation._state,
+            {"trace": trace, "filmstrip": filmstrip, "filmstrip_frames": []},
+        ):
+            instrumentation._finalize_screencast(
+                SessionCapturer(FakeTransport()), "sess", entry
+            )
+            return instrumentation.screencast_frames()
+
+    def test_the_buffer_is_kept_as_the_session_finalizes(self):
+        rec = self.Recorder(frames(3))
+        kept = self._finalize(rec, trace=True, filmstrip=True)
+        self.assertEqual([f["data"] for f in kept], ["f0", "f1", "f2"])
+
+    # The recorder is popped off the entry here, so anything not taken now is
+    # unreachable afterwards.
+    def test_nothing_is_kept_without_a_filmstrip(self):
+        rec = self.Recorder(frames(3))
+        self.assertEqual(self._finalize(rec, trace=True, filmstrip=False), [])
+
+    def test_trace_mode_stops_the_recorder_instead_of_encoding(self):
+        rec = self.Recorder(frames(3))
+        self._finalize(rec, trace=True, filmstrip=True)
+        self.assertEqual(rec.finalized, 0, "encoded a .webm in trace mode")
+        self.assertEqual(rec.stopped, 1)
+
+    def test_live_mode_still_encodes_the_video(self):
+        rec = self.Recorder(frames(3))
+        self._finalize(rec, trace=False, filmstrip=False)
+        self.assertEqual(rec.finalized, 1)
+
+    def test_sessions_accumulate_in_time_order(self):
+        # A replaced session can finish after a later one started, so quit
+        # order is not frame order.
+        with mock.patch.dict(
+            instrumentation._state,
+            {"filmstrip_frames": frames(2, start=100) + frames(2, start=0)},
+        ):
+            collected = instrumentation.screencast_frames()
+        self.assertEqual([f["timestamp"] for f in collected], [0, 1, 100, 101])
+
+
+class TestWhenTheRecorderRuns(unittest.TestCase):
+    """Trace mode switches the recorder off — except for the filmstrip, which
+    is the one thing that needs it."""
+
+    def test_the_option_defaults_on_like_the_js_adapters(self):
+        import selenium_devtools as pkg
+
+        with mock.patch.dict("os.environ", {}, clear=True):
+            self.assertTrue(pkg._filmstrip_enabled(None))
+
+    def test_it_is_opt_out(self):
+        import selenium_devtools as pkg
+
+        for value, expected in [
+            ("0", False), ("false", False), ("no", False), ("off", False),
+            ("", False), ("1", True), ("true", True),
+        ]:
+            with self.subTest(value=value):
+                with mock.patch.dict(
+                    "os.environ", {"DEVTOOLS_FILMSTRIP": value}, clear=True
+                ):
+                    self.assertIs(pkg._filmstrip_enabled(None), expected)
+
+    def test_the_argument_wins(self):
+        import selenium_devtools as pkg
+
+        with mock.patch.dict("os.environ", {"DEVTOOLS_FILMSTRIP": "0"}, clear=True):
+            self.assertTrue(pkg._filmstrip_enabled(True))
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestTheRecorderRunsForTheFilmstrip(unittest.TestCase):
+    """Drives the real session bring-up. The gate added with the trace export
+    switched the recorder off in trace mode outright; without the filmstrip
+    exception there are no frames to send and every test above passes on an
+    empty buffer."""
+
+    class Driver:
+        def __init__(self):
+            self.session_id = "sess-film"
+            self.caps = {"browserName": "chrome"}
+
+        def execute(self, command, params=None):
+            return {"value": None}
+
+        def get_screenshot_as_base64(self):
+            return "c2hvdA=="
+
+    def setUp(self):
+        from selenium_devtools.capturer import SessionCapturer
+
+        instrumentation.uninstall()
+        self.cap = SessionCapturer(FakeTransport())
+
+    def tearDown(self):
+        instrumentation.uninstall()
+
+    def _bring_up(self, *, trace, filmstrip):
+        instrumentation.install(
+            self.cap, self.Driver, trace=trace, filmstrip=filmstrip
+        )
+        with mock.patch.object(instrumentation, "ScreencastRecorder") as rec, \
+                mock.patch.object(
+                    instrumentation, "start_push_screencast", return_value=None
+                ), \
+                mock.patch.object(instrumentation, "bidi"), \
+                mock.patch.object(instrumentation, "bidi_preload"), \
+                mock.patch.object(
+                    instrumentation, "collector_source_text", return_value=None
+                ):
+            instrumentation._ensure_session_setup(self.Driver(), self.cap)
+        return rec
+
+    def test_trace_mode_with_a_filmstrip_records(self):
+        self._bring_up(trace=True, filmstrip=True).assert_called_once()
+
+    def test_trace_mode_without_one_does_not(self):
+        self._bring_up(trace=True, filmstrip=False).assert_not_called()
+
+    def test_live_mode_records_regardless(self):
+        # The dashboard video does not depend on the filmstrip option.
+        self._bring_up(trace=False, filmstrip=False).assert_called_once()
+
+
+class TestALiveSessionStillContributes(unittest.TestCase):
+    """An export can run before the last driver quits.
+
+    Frames are only reachable from one place at a time: a finalized session's
+    buffer was kept as it finalized, a live one's is still on its recorder — and
+    `uninstall` stops live recorders without keeping anything, so reading only
+    the finalized accumulator drops whatever the last session recorded.
+    """
+
+    class Recorder:
+        def __init__(self, buf):
+            self._buf = buf
+
+        @property
+        def frames(self):
+            return list(self._buf)
+
+    def test_a_live_recorder_is_read_as_well(self):
+        with mock.patch.dict(
+            instrumentation._state,
+            {
+                "filmstrip_frames": frames(2, start=0),
+                "sessions": {"d": {"screencast": self.Recorder(frames(2, start=100))}},
+            },
+        ):
+            collected = instrumentation.screencast_frames()
+        self.assertEqual([f["timestamp"] for f in collected], [0, 1, 100, 101])
+
+    def test_a_live_session_alone_is_enough(self):
+        with mock.patch.dict(
+            instrumentation._state,
+            {
+                "filmstrip_frames": [],
+                "sessions": {"d": {"screencast": self.Recorder(frames(3))}},
+            },
+        ):
+            self.assertEqual(len(instrumentation.screencast_frames()), 3)
+
+    def test_a_recorder_that_raises_does_not_break_the_export(self):
+        class Broken:
+            @property
+            def frames(self):
+                raise RuntimeError("session gone")
+
+        with mock.patch.dict(
+            instrumentation._state,
+            {"filmstrip_frames": frames(2), "sessions": {"d": {"screencast": Broken()}}},
+        ):
+            self.assertEqual(len(instrumentation.screencast_frames()), 2)
+
+
+class TestRetriesDoNotDuplicateFrames(unittest.TestCase):
+    """A failed export is retried at teardown (#340), and the backend APPENDS
+    these — it has no key to replace or dedupe on — so resending the buffer
+    would put every frame in the trace twice."""
+
+    def setUp(self):
+        import selenium_devtools as pkg
+
+        self.pkg = pkg
+        self.saved = dict(pkg._active)
+
+    def tearDown(self):
+        self.pkg._active.clear()
+        self.pkg._active.update(self.saved)
+
+    def _arm(self):
+        self.pkg._active.update(
+            capturer=None, transport=FakeTransport(), process=None, url=None,
+            handle=None, terminal=None, logs=None, excepthook=None,
+            trace=True, traced=False, filmstrip_mark=None,
+        )
+
+    def _export_with(self, buf, *, accepted=None):
+        """Run one export attempt over `buf`; returns the send_frames mock."""
+        take = (lambda tx, f: len(f)) if accepted is None else (lambda tx, f: accepted)
+        with mock.patch.object(
+            instrumentation, "screencast_frames", return_value=buf
+        ), mock.patch.object(
+            instrumentation, "resolved_output_dir", return_value="/out"
+        ), mock.patch.object(
+            trace_export, "send_frames", side_effect=take
+        ) as send, mock.patch.object(trace_export, "export", return_value=None):
+            self.pkg.export_trace()
+        return send
+
+    # The boundary is inclusive, so a retry re-sends the one frame the last
+    # attempt ended on — never the buffer. The exporter content-addresses frame
+    # bytes, so that duplicate shares a resource with the original.
+    def test_a_retry_resends_only_the_boundary_frame(self):
+        self._arm()
+        buf = frames(10)
+        self._export_with(buf)
+        send = self._export_with(buf)
+        self.assertEqual(
+            [f["timestamp"] for f in send.call_args[0][1]], [9],
+            "resent more than the boundary frame",
+        )
+
+    def test_a_retry_sends_what_a_partial_send_missed(self):
+        self._arm()
+        buf = frames(10)
+        self._export_with(buf, accepted=4)   # 0..3 accepted, watermark 3
+        send = self._export_with(buf)
+        self.assertEqual(
+            [f["timestamp"] for f in send.call_args[0][1]], [3, 4, 5, 6, 7, 8, 9]
+        )
+
+    # Two frames can share a millisecond. An exclusive boundary dropped the
+    # unsent twin — a hole in the filmstrip — because the watermark had already
+    # advanced past its timestamp.
+    def test_a_frame_sharing_the_boundary_millisecond_is_not_lost(self):
+        self._arm()
+        twin_a = {"data": "a", "timestamp": 5}
+        twin_b = {"data": "b", "timestamp": 5}
+        buf = frames(5) + [twin_a, twin_b] + frames(2, start=6)
+
+        self._export_with(buf, accepted=6)   # ends on twin_a, watermark 5
+        send = self._export_with(buf)
+
+        resent = [f["data"] for f in send.call_args[0][1]]
+        self.assertIn("b", resent, "dropped the unsent twin")
+
+    # The buffer is BOUNDED and decimated in place — `screencast._decimate`
+    # halves it, keeping the ends — so between two attempts the list can shrink
+    # and every index shift. An offset would skip frames it never sent.
+    def test_a_decimated_buffer_between_attempts_loses_nothing(self):
+        self._arm()
+        first = frames(10)                       # timestamps 0..9
+        self._export_with(first, accepted=6)     # 0..5 accepted, watermark 5
+
+        # The live recorder halves its buffer, then records more.
+        decimated = [first[0], *first[1:-1:2], first[-1]] + frames(3, start=10)
+        send = self._export_with(decimated)
+
+        resent = [f["timestamp"] for f in send.call_args[0][1]]
+        self.assertNotIn(0, resent, "resent a frame the backend already has")
+        # 6 and 8 are absent because decimation REMOVED them — they no longer
+        # exist to send. 5 is the inclusive boundary.
+        self.assertEqual(resent, [5, 7, 9, 10, 11, 12])
+
+    def test_frames_older_than_the_watermark_are_never_resent(self):
+        self._arm()
+        self._export_with(frames(5))             # watermark 4
+        send = self._export_with(frames(5, start=0) + frames(2, start=5))
+        self.assertEqual([f["timestamp"] for f in send.call_args[0][1]], [4, 5, 6])
+
+
+class TestUninstallKeepsLiveFrames(unittest.TestCase):
+    """`disable()` uninstalls BEFORE its fallback export, and uninstall replaces
+    `sessions` — so a session that never quit would have its frames dropped
+    before anything could read them. That is the plain-script path exactly: no
+    per-test fixture quits the driver, so the last session is always live."""
+
+    class Recorder:
+        def __init__(self, buf):
+            self._buf = buf
+            self.stopped = 0
+
+        @property
+        def frames(self):
+            return list(self._buf)
+
+        def stop(self):
+            self.stopped += 1
+
+    def tearDown(self):
+        instrumentation.uninstall()
+
+    def _uninstall_with(self, *, filmstrip):
+        rec = self.Recorder(frames(4))
+        sessions = {"driver": {"screencast": rec}}
+        with mock.patch.dict(
+            instrumentation._state,
+            {
+                "sessions": sessions,
+                "filmstrip": filmstrip,
+                "filmstrip_frames": [],
+                "installed": False,
+            },
+        ):
+            instrumentation.uninstall()
+            kept = list(instrumentation._state["filmstrip_frames"])
+        return rec, kept
+
+    def test_a_live_recorders_buffer_survives_uninstall(self):
+        rec, kept = self._uninstall_with(filmstrip=True)
+        self.assertEqual(len(kept), 4)
+        self.assertEqual(rec.stopped, 1, "recorder left running past teardown")
+
+    def test_nothing_is_kept_without_a_filmstrip(self):
+        rec, kept = self._uninstall_with(filmstrip=False)
+        self.assertEqual(kept, [])
+        self.assertEqual(rec.stopped, 1)