From b797c029e0fb61f4d8e1e996242de64752eb538e Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Mon, 24 Aug 2026 04:45:01 +0530 Subject: [PATCH 1/9] feat(selenium-devtools-py): subscribe to Chrome's pushed screencast frames --- .../src/selenium_devtools/cdp_screencast.py | 175 +++++++++++++++ .../tests/test_cdp_screencast.py | 207 ++++++++++++++++++ 2 files changed, 382 insertions(+) create mode 100644 packages/selenium-devtools-py/src/selenium_devtools/cdp_screencast.py create mode 100644 packages/selenium-devtools-py/tests/test_cdp_screencast.py diff --git a/packages/selenium-devtools-py/src/selenium_devtools/cdp_screencast.py b/packages/selenium-devtools-py/src/selenium_devtools/cdp_screencast.py new file mode 100644 index 00000000..30914a10 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/cdp_screencast.py @@ -0,0 +1,175 @@ +"""Chrome's push-mode screencast, over CDP. + +The per-command recorder in :mod:`.screencast` takes one screenshot per command +on the main thread, which is a real constraint rather than a shortcut: a +Selenium session is not thread-safe, so a poll thread racing the test's own +commands corrupts both the video and the DOM readback. + +Chrome can PUSH frames instead. ``Page.startScreencast`` streams them as CDP +events over a **separate websocket** — `driver.start_devtools()` opens its own +connection and attaches to the current target — so frames arrive without +issuing anything on the session's command channel. That is what makes a real +frame stream safe here where a poll loop was not. + +Two things this has to bound, which per-command capture never needed to: + +* **Rate.** ``every_nth_frame`` throttles at the source, so frames are dropped + by the browser instead of buffered and thrown away by us. +* **Acknowledgement.** Chrome stops sending after an unacknowledged frame, so + every frame is acked. A missed ack does not degrade the stream, it ends it. + +Availability is discovered, never assumed: anything other than a Chromium +driver with a reachable CDP endpoint returns None and the caller keeps its +per-command capture. Nothing here may raise into the user's test. +""" + +from __future__ import annotations + +import logging +import threading +from typing import Any, Callable, Optional + +from .constants import LOGGER_NAME, SCREENCAST_IMAGE_FORMAT + +_log = logging.getLogger(f"{LOGGER_NAME}.screencast") + +#: Frames the browser skips between the ones it sends. Chrome emits on every +#: composite, which for an animation is far more than a filmstrip needs; asking +#: for every 2nd halves the stream at the source rather than downstream. +DEFAULT_EVERY_NTH_FRAME = 2 + +#: Bounds a frame's longest edge. A retina viewport otherwise streams images +#: several times the size the dashboard renders them at. +DEFAULT_MAX_EDGE = 1280 + +#: PNG rather than JPEG so the encoder's on-disk frames keep one extension and +#: one decoder; `SCREENCAST_IMAGE_FORMAT` is the single place that is decided. +_FORMAT = SCREENCAST_IMAGE_FORMAT + +#: Callback taking one base64 image. Returns whether the frame was buffered, +#: which this module does not act on — it exists so the recorder's `add_frame` +#: can be passed directly. +FrameSink = Callable[[str], Any] + + +class PushScreencast: + """A live CDP screencast. Created by :func:`start_push_screencast`.""" + + def __init__(self, connection: Any, devtools: Any, sink: FrameSink) -> None: + self._connection = connection + self._devtools = devtools + self._sink = sink + self._callback_id: Optional[int] = None + self._frames = 0 + self._stopped = False + # The events arrive on the websocket's own reader thread while `stop` + # runs on the test's, and both touch `_stopped` and the connection. + self._lock = threading.Lock() + + @property + def frame_count(self) -> int: + return self._frames + + def _on_frame(self, event: Any) -> None: + """Buffer one pushed frame and acknowledge it. + + Runs on the websocket reader thread. The ack has to happen even when + the sink refuses the frame — an unacked frame is the last one Chrome + sends, so dropping the ack would silently end the recording. + """ + with self._lock: + if self._stopped: + return + session_id = getattr(event, "session_id", None) + try: + data = getattr(event, "data", None) + if isinstance(data, str) and data: + self._frames += 1 + self._sink(data) + except Exception as exc: # noqa: BLE001 — never kill the reader thread + _log.debug("screencast frame dropped: %s", exc) + finally: + if session_id is not None: + self._ack(session_id) + + def _ack(self, session_id: Any) -> None: + try: + self._connection.execute( + self._devtools.page.screencast_frame_ack(session_id=session_id) + ) + except Exception as exc: # noqa: BLE001 — a dead session ends the run anyway + _log.debug("screencast ack failed: %s", exc) + + def stop(self) -> None: + """Stop the stream. Idempotent, and never raises. + + The websocket is left open: `start_devtools` caches one connection per + driver and closing it here would take the session's other CDP users + down with it. + """ + with self._lock: + if self._stopped: + return + self._stopped = True + try: + self._connection.execute(self._devtools.page.stop_screencast()) + except Exception as exc: # noqa: BLE001 + _log.debug("stop_screencast failed: %s", exc) + if self._callback_id is not None: + try: + self._connection.remove_callback( + self._devtools.page.ScreencastFrame, self._callback_id + ) + except Exception as exc: # noqa: BLE001 + _log.debug("screencast unsubscribe failed: %s", exc) + + +def start_push_screencast( + driver: Any, + sink: FrameSink, + *, + every_nth_frame: int = DEFAULT_EVERY_NTH_FRAME, + max_edge: int = DEFAULT_MAX_EDGE, +) -> Optional[PushScreencast]: + """Subscribe to Chrome's frame stream, or return None if it is unavailable. + + None is the ordinary answer on any non-Chromium browser, on a grid session + with no CDP endpoint, and on a Chrome whose CDP version selenium does not + bundle. The caller keeps per-command capture in every one of those cases, + so this reports at debug level rather than warning about a browser doing + nothing wrong. + """ + start = getattr(driver, "start_devtools", None) + if not callable(start): + _log.debug("push screencast unavailable: driver exposes no CDP") + return None + try: + devtools, connection = start() + except Exception as exc: # noqa: BLE001 — no CDP is a normal outcome + _log.debug("push screencast unavailable: %s", exc) + return None + if devtools is None or connection is None: + return None + + recorder = PushScreencast(connection, devtools, sink) + try: + callback_id = connection.add_callback( + devtools.page.ScreencastFrame, recorder._on_frame + ) + # Recorded before the start command, so a failure there still unwinds + # the subscription rather than leaving a live callback behind. + recorder._callback_id = callback_id + connection.execute( + devtools.page.start_screencast( + format_=_FORMAT, + every_nth_frame=every_nth_frame, + max_width=max_edge, + max_height=max_edge, + ) + ) + except Exception as exc: # noqa: BLE001 + _log.debug("push screencast could not start: %s", exc) + recorder.stop() + return None + _log.info("screencast: streaming frames from the browser (CDP push mode)") + return recorder diff --git a/packages/selenium-devtools-py/tests/test_cdp_screencast.py b/packages/selenium-devtools-py/tests/test_cdp_screencast.py new file mode 100644 index 00000000..ef58d27d --- /dev/null +++ b/packages/selenium-devtools-py/tests/test_cdp_screencast.py @@ -0,0 +1,207 @@ +"""Chrome's pushed frame stream. + +No browser here: the CDP surface is a command object passed to +`connection.execute` and an event object handed to a callback, both of which a +fake models exactly. What the tests are really about is the two rules that make +a pushed stream work at all — every frame gets acknowledged, and nothing raises +out of the websocket's reader thread. +""" + +import types +import unittest + +from selenium_devtools import cdp_screencast +from selenium_devtools.cdp_screencast import start_push_screencast + + +class FakePage: + """The generated `devtools.page` module, reduced to what we call.""" + + class ScreencastFrame: + event_class = "Page.screencastFrame" + + def start_screencast(self, **kwargs): + return ("start_screencast", kwargs) + + def stop_screencast(self): + return ("stop_screencast", {}) + + def screencast_frame_ack(self, session_id): + return ("ack", {"session_id": session_id}) + + +class FakeConnection: + def __init__(self, fail_on=None): + self.executed = [] + self.callbacks = {} + self.removed = [] + self._fail_on = fail_on + self._next_id = 1 + + def execute(self, command): + name = command[0] + if self._fail_on and name == self._fail_on: + raise RuntimeError(f"{name} refused") + self.executed.append(command) + return None + + def add_callback(self, event, callback): + self.callbacks[event.event_class] = callback + cid, self._next_id = self._next_id, self._next_id + 1 + return cid + + def remove_callback(self, event, callback_id): + self.removed.append((event.event_class, callback_id)) + + # ── test helpers ────────────────────────────────────────────────────────── + def push(self, data="aGk=", session_id=7): + self.callbacks["Page.screencastFrame"]( + types.SimpleNamespace(data=data, session_id=session_id, metadata=None) + ) + + @property + def commands(self): + return [name for name, _ in self.executed] + + @property + def acks(self): + return [kw["session_id"] for name, kw in self.executed if name == "ack"] + + +def fake_driver(connection=None, *, raises=False, no_cdp=False): + if no_cdp: + return types.SimpleNamespace() + devtools = types.SimpleNamespace(page=FakePage()) + conn = connection if connection is not None else FakeConnection() + + def start_devtools(): + if raises: + raise RuntimeError("no CDP endpoint") + return devtools, conn + + return types.SimpleNamespace(start_devtools=start_devtools) + + +class TestAvailabilityIsDiscovered(unittest.TestCase): + """None is an ordinary answer — every non-Chromium browser gives it — so it + must never raise and never warn.""" + + def test_a_driver_without_cdp_declines(self): + self.assertIsNone(start_push_screencast(fake_driver(no_cdp=True), print)) + + def test_an_unreachable_cdp_endpoint_declines(self): + self.assertIsNone(start_push_screencast(fake_driver(raises=True), print)) + + def test_a_refused_start_declines_and_leaves_no_subscription(self): + # Otherwise a callback stays live on a stream nobody is acking, and the + # sink keeps taking frames the recording will never use. + conn = FakeConnection(fail_on="start_screencast") + + self.assertIsNone(start_push_screencast(fake_driver(conn), print)) + self.assertEqual(len(conn.removed), 1) + self.assertIn("stop_screencast", conn.commands) + + +class TestTheStreamStarts(unittest.TestCase): + def test_it_subscribes_and_asks_the_browser_to_stream(self): + conn = FakeConnection() + + recorder = start_push_screencast(fake_driver(conn), lambda d: None) + + self.assertIsNotNone(recorder) + self.assertIn("Page.screencastFrame", conn.callbacks) + self.assertIn("start_screencast", conn.commands) + + def test_it_throttles_and_bounds_frames_at_the_source(self): + # Dropping frames in the browser beats buffering them here and throwing + # them away later. + conn = FakeConnection() + + start_push_screencast(fake_driver(conn), lambda d: None) + + (_, kwargs) = next(c for c in conn.executed if c[0] == "start_screencast") + self.assertEqual(kwargs["every_nth_frame"], cdp_screencast.DEFAULT_EVERY_NTH_FRAME) + self.assertEqual(kwargs["max_width"], cdp_screencast.DEFAULT_MAX_EDGE) + self.assertEqual(kwargs["format_"], "png") + + +class TestEveryFrameIsAcknowledged(unittest.TestCase): + """Chrome sends nothing more after a frame it was not told about, so a + missed ack does not degrade the recording — it ends it.""" + + def test_a_frame_reaches_the_sink_and_is_acked(self): + conn = FakeConnection() + seen = [] + recorder = start_push_screencast(fake_driver(conn), seen.append) + + conn.push(data="ZnJhbWU=", session_id=42) + + self.assertEqual(seen, ["ZnJhbWU="]) + self.assertEqual(conn.acks, [42]) + self.assertEqual(recorder.frame_count, 1) + + def test_an_empty_frame_is_still_acked(self): + conn = FakeConnection() + seen = [] + recorder = start_push_screencast(fake_driver(conn), seen.append) + + conn.push(data="") + + self.assertEqual(seen, []) + self.assertEqual(recorder.frame_count, 0) + self.assertEqual(len(conn.acks), 1) + + def test_a_sink_that_raises_neither_escapes_nor_stops_the_stream(self): + # This runs on the websocket's reader thread; an exception there takes + # the whole connection down, not just one frame. + conn = FakeConnection() + + def boom(_data): + raise RuntimeError("buffer full") + + start_push_screencast(fake_driver(conn), boom) + + conn.push() # must not raise + + self.assertEqual(len(conn.acks), 1) + + def test_frames_arriving_after_a_stop_are_ignored(self): + conn = FakeConnection() + seen = [] + recorder = start_push_screencast(fake_driver(conn), seen.append) + recorder.stop() + + conn.push() + + self.assertEqual(seen, []) + + +class TestStopping(unittest.TestCase): + def test_stop_ends_the_stream_and_unsubscribes(self): + conn = FakeConnection() + recorder = start_push_screencast(fake_driver(conn), lambda d: None) + + recorder.stop() + + self.assertIn("stop_screencast", conn.commands) + self.assertEqual(len(conn.removed), 1) + + def test_stop_is_idempotent(self): + conn = FakeConnection() + recorder = start_push_screencast(fake_driver(conn), lambda d: None) + + recorder.stop() + recorder.stop() + + self.assertEqual(conn.commands.count("stop_screencast"), 1) + + def test_a_dead_session_does_not_raise_out_of_stop(self): + conn = FakeConnection() + recorder = start_push_screencast(fake_driver(conn), lambda d: None) + conn._fail_on = "stop_screencast" + + recorder.stop() # teardown must survive a session that already went away + + +if __name__ == "__main__": + unittest.main() From 38cb955f339a338b0f2997a0a1bbd4f9f886c4eb Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Mon, 24 Aug 2026 04:45:15 +0530 Subject: [PATCH 2/9] fix(selenium-devtools-py): bound the screencast buffer without biasing its end --- .../src/selenium_devtools/constants.py | 6 ++ .../src/selenium_devtools/screencast.py | 90 +++++++++++++++---- .../tests/test_screencast.py | 57 ++++++++++++ 3 files changed, 135 insertions(+), 18 deletions(-) diff --git a/packages/selenium-devtools-py/src/selenium_devtools/constants.py b/packages/selenium-devtools-py/src/selenium_devtools/constants.py index dfb1ddb8..4a1fd389 100644 --- a/packages/selenium-devtools-py/src/selenium_devtools/constants.py +++ b/packages/selenium-devtools-py/src/selenium_devtools/constants.py @@ -138,6 +138,12 @@ # shot of about:blank taken before the first command, so a short run's "video" # was black plus one real frame. SCREENCAST_MIN_FRAMES = 1 +# Buffered frames before the recorder halves what it holds. Per-command capture +# needs no cap — it takes one frame per command, so the test's own length bounds +# it — but CDP push mode streams from the browser and would grow without one. +# Mirrors core's `maxBufferFrames` default, decimating rather than truncating so +# both ends of the run survive. +SCREENCAST_MAX_BUFFER_FRAMES = 2000 # Output filename stem; the session id + .webm suffix are appended. SCREENCAST_FILENAME_PREFIX = "selenium-py-video" # The `screencast` wire scope is generated into _contract.py (SCOPE_SCREENCAST). diff --git a/packages/selenium-devtools-py/src/selenium_devtools/screencast.py b/packages/selenium-devtools-py/src/selenium_devtools/screencast.py index 120cbfed..d8d6bb9a 100644 --- a/packages/selenium-devtools-py/src/selenium_devtools/screencast.py +++ b/packages/selenium-devtools-py/src/selenium_devtools/screencast.py @@ -1,21 +1,26 @@ """Screencast recorder — the Python analogue of core's ``ScreencastRecorderBase``. -Frames are captured **synchronously on the main thread, one per command** -(``driver.get_screenshot_as_base64()``), driven from the instrumentation hook. -A background poll thread is deliberately NOT used: Selenium's session is not -thread-safe, and a screenshot fired from a daemon thread races the main thread's -commands and DOM-trace readback on the same connection — corrupting both the -video and the snapshot. The reference JS adapters avoid this too (CDP push-mode -screencast / per-command screenshots on the command queue), so we mirror that. +The recorder owns a frame buffer and the encode; it does not care where frames +came from. Two sources feed it: + +* **Chrome, pushed.** :mod:`.cdp_screencast` subscribes to + ``Page.screencastFrame`` and hands each frame to :meth:`add_frame`. Preferred + where it is available, because the browser decides when the picture changed. +* **Everything else, one frame per command.** A screenshot taken + **synchronously on the main thread**, driven from the instrumentation hook. + +A background poll thread is deliberately NOT used for that second path: +Selenium's session is not thread-safe, and a screenshot fired from a daemon +thread races the main thread's commands and DOM-trace readback on the same +connection, corrupting both the video and the snapshot. Push mode escapes that +because CDP events arrive on their own websocket rather than the session's +command channel — which is why it is a different mechanism rather than the same +poll loop moved. On ``stop`` the buffered frames are encoded to a ``.webm`` via ffmpeg *if it's on PATH* — ffmpeg is an optional dependency, so its absence is a one-line warning and a skipped encode, never an error. -A CDP push-mode fast-path (Chrome ``Page.startScreencast`` via -``execute_cdp_cmd``) is a future optimization — noted here, not implemented; -per-command capture works on every browser selenium drives. - Everything is defensive: a transient screenshot failure (e.g. mid-navigation) is skipped and recording continues; a recorder that never captured a frame encodes nothing. Capture never breaks the user's test. @@ -28,12 +33,14 @@ import shutil import subprocess import tempfile +import threading import weakref from typing import Any, Callable, List, Optional from .constants import ( LOGGER_NAME, SCREENCAST_IMAGE_FORMAT, + SCREENCAST_MAX_BUFFER_FRAMES, SCREENCAST_MIN_FRAMES, ) from .output_dir import OUTPUT_SUBDIR, ensure_output_dir @@ -67,12 +74,22 @@ def take() -> Optional[str]: class ScreencastRecorder: - def __init__(self, *, ffmpeg_path: Optional[str] = None) -> None: + def __init__( + self, + *, + ffmpeg_path: Optional[str] = None, + max_frames: int = SCREENCAST_MAX_BUFFER_FRAMES, + ) -> None: # Resolve ffmpeg once; None means "encoding unavailable, skip it". self._ffmpeg = ffmpeg_path or shutil.which("ffmpeg") self._frames: List[ScreencastFrame] = [] self._screenshot: Optional[ScreenshotFn] = None self._active = False + self._max_frames = max(2, max_frames) + self._buffer_lock = threading.Lock() + # Frames offered, and how many are offered per one kept. See `_buffer`. + self._seen = 0 + self._stride = 1 # ── public API ──────────────────────────────────────────────────────────── @@ -121,20 +138,57 @@ def capture(self) -> bool: except Exception: # noqa: BLE001 — transient miss; keep recording return False if isinstance(data, str) and data: - self._frames.append({"data": data, "timestamp": now_ms()}) + self._buffer(data) return True return False def add_frame(self, data: Optional[str]) -> bool: - """Buffer a frame from an ALREADY-captured base64 screenshot — lets the - caller reuse the per-command screenshot it took for the command entry - instead of paying for a second screenshot round-trip. No-op if not armed - or the data is empty. Returns True iff a frame was buffered.""" + """Buffer a frame from an ALREADY-captured base64 image. + + Serves both sources: the per-command path reuses the screenshot it + already took for the command entry rather than paying for a second + round-trip, and CDP push mode hands over frames the browser sent + unasked. No-op if not armed or the data is empty. Returns True iff a + frame was buffered. + + Called from the CDP websocket's reader thread as well as the test's, so + the append is guarded — a list append is atomic under the GIL but the + decimation below is a read-modify-write. + """ if not self._active or not isinstance(data, str) or not data: return False - self._frames.append({"data": data, "timestamp": now_ms()}) + self._buffer(data) return True + def _buffer(self, data: str) -> None: + with self._buffer_lock: + self._seen += 1 + # Thin the INCOMING frames by however often the buffer has been + # halved. Without this the buffer drifts toward holding only the end + # of the run: each decimation halves what is already held while new + # frames keep arriving unthinned. Measured on a 40-frame run at a cap + # of 6, it kept frames 0, 1, 35, 37, 38, 39 — the last second of the + # run and nothing from the middle of it. + if self._seen % self._stride: + return + self._frames.append({"data": data, "timestamp": now_ms()}) + if len(self._frames) > self._max_frames: + self._decimate() + self._stride *= 2 + + def _decimate(self) -> None: + """Halve the buffer, keeping the first and last frames. + + Push mode streams from the browser, so the buffer is not bounded by the + test's own length the way per-command capture is. Dropping every other + middle frame keeps the run evenly covered: the encoder derives each + frame's on-screen duration from the timestamps, so the survivors simply + hold longer. Truncating either end would lose the start or the finish of + the run outright, which is the part someone is usually looking for. + """ + frames = self._frames + self._frames = [frames[0], *frames[1:-1:2], frames[-1]] + def stop(self) -> None: """Disarm the recorder. Idempotent; safe even if start() never ran.""" self._active = False diff --git a/packages/selenium-devtools-py/tests/test_screencast.py b/packages/selenium-devtools-py/tests/test_screencast.py index 5a85158d..c4e76ff6 100644 --- a/packages/selenium-devtools-py/tests/test_screencast.py +++ b/packages/selenium-devtools-py/tests/test_screencast.py @@ -242,3 +242,60 @@ def test_finalize_encodes_and_delivers_frame(self): if __name__ == "__main__": unittest.main() + + +class TestTheBufferIsBounded(unittest.TestCase): + """Per-command capture is bounded by the test's own length — one frame per + command. A pushed stream is not, so the recorder caps what it holds.""" + + def _armed(self, cap): + rec = ScreencastRecorder(max_frames=cap) + rec.start(None, screenshot_fn=lambda: _PNG_FRAME) + return rec + + def test_the_buffer_stops_growing_at_the_cap(self): + rec = self._armed(8) + + for i in range(200): + rec.add_frame(f"frame-{i}") + + self.assertLessEqual(len(rec.frames), 8) + + def test_the_whole_run_stays_covered_not_just_its_end(self): + # The property that matters, and the one endpoints alone cannot check: + # truncating the tail still leaves frame-0 and whatever arrived since the + # last decimation, so it LOOKS like both ends survived. What separates a + # video of the run from a video of its last moments is whether anything + # from the middle is still there. + rec = self._armed(6) + + for i in range(40): + rec.add_frame(f"frame-{i}") + kept = [int(f["data"].split("-")[1]) for f in rec.frames] + + self.assertEqual(kept[0], 0) + middle = [k for k in kept if 10 <= k <= 30] + self.assertTrue(middle, f"nothing from the middle of the run: {kept}") + + def test_the_end_of_a_long_run_survives(self): + # At the real cap the stride only starts thinning after the buffer is + # full, so the newest frame — the state the run ended in — is kept. + rec = self._armed(2000) + + for i in range(12000): + rec.add_frame(f"frame-{i}") + kept = [int(f["data"].split("-")[1]) for f in rec.frames] + + self.assertEqual(kept[0], 0) + self.assertEqual(kept[-1], 11999) + self.assertLessEqual(len(kept), 2000) + + def test_a_run_under_the_cap_keeps_every_frame_in_order(self): + rec = self._armed(100) + + for i in range(10): + rec.add_frame(f"frame-{i}") + + self.assertEqual( + [f["data"] for f in rec.frames], [f"frame-{i}" for i in range(10)] + ) From c3f15680de37b240a076daf8e2bc0d2b8c9ebd28 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Mon, 24 Aug 2026 04:45:29 +0530 Subject: [PATCH 3/9] feat(selenium-devtools-py): prefer the pushed screencast when Chrome offers one --- .../src/selenium_devtools/instrumentation.py | 37 ++++++++++- .../tests/test_instrumentation.py | 65 +++++++++++++++++++ 2 files changed, 99 insertions(+), 3 deletions(-) diff --git a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py index 932dc473..684feab5 100644 --- a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py +++ b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py @@ -22,6 +22,7 @@ from . import assertions, bidi, bidi_preload, frames from .assert_tracer import ScriptAssertionTracer from .capturer import SessionCapturer +from .cdp_screencast import start_push_screencast from .collector_source import reset_cache as reset_collector_cache from .constants import ( BIDI_CAPABILITY, @@ -269,10 +270,29 @@ def _backend_origin(capturer: SessionCapturer) -> Optional[tuple]: return (host, port) if host and port else None +def _stop_push_screencast(entry: dict) -> None: + """End the browser's frame stream for one driver. Never raises.""" + push = entry.pop("screencast_push", None) + if push is None: + return + try: + push.stop() + _log.info("screencast: %d frame(s) streamed from the browser", + push.frame_count) + except Exception as exc: # noqa: BLE001 — teardown must not raise + _log.debug("stopping the push screencast threw: %s", exc) + + def _add_screencast_frame(entry: dict, shot: Optional[str]) -> None: - """Buffer an already-captured screenshot as a frame of ITS OWN session.""" + """Buffer an already-captured screenshot as a frame of ITS OWN session. + + Skipped while the browser is streaming: the pushed frames already cover the + timeline, and interleaving a per-command shot would duplicate it at a + slightly different moment. The screenshot is still taken — the command ROW + carries it — so this only decides what the video is made of. + """ recorder = entry.get("screencast") - if recorder is None or not shot: + if recorder is None or not shot or entry.get("screencast_push") is not None: return try: recorder.add_frame(shot) @@ -339,6 +359,9 @@ def _enable_bidi_capability(params: Any) -> None: def _close_entry(capturer: SessionCapturer, entry: dict) -> None: """Drain and encode one driver's capture, attributed to the session it was recorded under.""" + # Before the encode: a frame still arriving would land after the buffer was + # read, and the browser keeps sending until it is told to stop. + _stop_push_screencast(entry) _flush_mutations(capturer, entry) entry["snapshot"] = None _finalize_screencast(capturer, entry["session_id"], entry) @@ -417,7 +440,14 @@ def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> Optional[di recorder = ScreencastRecorder() recorder.start(driver) entry["screencast"] = recorder - _log.info("screencast recording started") + # Prefer the browser's own frame stream. It returns None on anything but + # a Chromium driver with a reachable CDP endpoint, and then the + # per-command screenshots the command hook already buffers are the + # recording — so this is an upgrade, never a requirement. + push = start_push_screencast(driver, recorder.add_frame) + entry["screencast_push"] = push + if push is None: + _log.info("screencast recording started (one frame per command)") except Exception as exc: # noqa: BLE001 _log.warning("screencast start threw: %s", exc) try: @@ -679,6 +709,7 @@ def uninstall() -> None: reset_collector_cache() # Never leave a recorder running past teardown, for any session still live. for entry in list(_state.get("sessions", {}).values()): + _stop_push_screencast(entry) recorder = entry.get("screencast") if recorder is not None: recorder.stop() diff --git a/packages/selenium-devtools-py/tests/test_instrumentation.py b/packages/selenium-devtools-py/tests/test_instrumentation.py index e820bc31..1b56bf8a 100644 --- a/packages/selenium-devtools-py/tests/test_instrumentation.py +++ b/packages/selenium-devtools-py/tests/test_instrumentation.py @@ -763,3 +763,68 @@ def test_the_recorder_does_not_retain_its_driver(self): gc.collect() self.assertIsNone(ref(), "the recorder is still holding the driver") self.assertEqual(live_session_ids(), set()) + + +class TestPushScreencastWiring(unittest.TestCase): + """Which source the video is made of, and that the stream is turned off. + + Both are decided in `instrumentation`, so a correct `cdp_screencast` proves + nothing about them on its own. + """ + + class Recorder: + def __init__(self): + self.frames = [] + + def add_frame(self, data): + self.frames.append(data) + return True + + class Push: + def __init__(self, *, raises=False): + self.stopped = 0 + self.frame_count = 3 + self._raises = raises + + def stop(self): + self.stopped += 1 + if self._raises: + raise RuntimeError("session already gone") + + def test_per_command_shots_are_skipped_while_the_browser_streams(self): + # The pushed frames already cover the timeline; a per-command shot would + # duplicate one of them at a slightly different moment. + rec = self.Recorder() + entry = {"screencast": rec, "screencast_push": self.Push()} + + instrumentation._add_screencast_frame(entry, "shot") + + self.assertEqual(rec.frames, []) + + def test_per_command_shots_are_the_recording_without_a_stream(self): + rec = self.Recorder() + entry = {"screencast": rec, "screencast_push": None} + + instrumentation._add_screencast_frame(entry, "shot") + + self.assertEqual(rec.frames, ["shot"]) + + def test_the_stream_is_stopped_once_and_forgotten(self): + # Left running, the browser keeps sending frames into a buffer that has + # already been read and encoded. + push = self.Push() + entry = {"screencast_push": push} + + instrumentation._stop_push_screencast(entry) + instrumentation._stop_push_screencast(entry) + + self.assertEqual(push.stopped, 1) + self.assertNotIn("screencast_push", entry) + + def test_a_stream_that_throws_on_stop_does_not_break_teardown(self): + entry = {"screencast_push": self.Push(raises=True)} + + instrumentation._stop_push_screencast(entry) # must not raise + + def test_no_stream_is_a_no_op(self): + instrumentation._stop_push_screencast({}) # must not raise From fda858b77015eea221e5aed1080ea8ec73c2b814 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Mon, 24 Aug 2026 04:45:41 +0530 Subject: [PATCH 4/9] docs: record how the pushed screencast is bounded --- CLAUDE.md | 3 +++ packages/selenium-devtools-py/README.md | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/CLAUDE.md b/CLAUDE.md index 0525422a..fba4f0ea 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -265,6 +265,9 @@ Documented divergences from the conventions above. They exist today as debt to b - **A command row is stamped at COMPLETION, and the DOM anchor carries the document's own birth time.** These two together are what make the replay line up; both adapters got them wrong in the same way and the fix is symmetric. (a) `selenium-devtools/src/driverPatcher.ts` and `nightwatch-devtools/src/helpers/browserProxy.ts` both ran their capture at completion but stamped `timestamp` with the *invocation* clock, keeping the invocation time as `startTime` only after this fix. The page-side mutation stream is on real time, so an invocation-stamped row ended before its own effect landed and replayed the page from before it — the `#username` fill rendered an empty field, the `#password` fill rendered only the username, and a navigation row rendered the page it had just left. Rows also now span their real duration instead of a synthetic 1 ms. (b) `collector.captureCurrentDom` (the only producer of a mutation with a `url`) stamps `performance.timeOrigin`, not the drain clock. A drain is forced from Node whenever a collector might be fresh, which is always after the navigation — a round trip at best, a whole page load at worst — so drain-stamping put the anchor after several later actions (measured: 9/15 Selenium and 8/15 Nightwatch rows on the wrong DOM). With both in place a navigation row ends after its destination document was born, so the anchor needs no repositioning at all. - `core/trace-mutations.ts` `reattributeDomAnchors` remains as a narrow backstop for the one case the stamps can't cover: an anchor born *after* the last logged command, i.e. a click whose navigation commits once the click has already returned. It snaps such an anchor to the newest logged command, but **only when no logged command completed after it** — if one did, that command's row already resolves the anchor and pulling it earlier mis-credits it to a preceding action and steals the new page's DOM from rows still on the old one (measured: a 206 ms pull moved `/login` onto two rows that were on `/add_remove_elements`). Anchors are only pulled earlier, never past the newest timestamp already in the stream, or replay would apply the outgoing document's refs to the incoming tree. - Residual, accepted: Nightwatch's `click` resolves *before* its navigation commits (measured 5 ms), so a submit-click row can still show its pre-navigation page. Selenium is immune — its click waits for page load. Not worth another heuristic; every heuristic tried here regressed a different row. +- **A pushed screencast needs bounding at both ends, and the obvious bound biases toward the end of the run.** Per-command capture is self-limiting — one frame per command, so the test's own length caps it — which is why `selenium-devtools-py` had no frame cap at all. Chrome's `Page.startScreencast` removes that property: `cdp_screencast.py` subscribes over the websocket `driver.start_devtools()` opens, which is a *different connection* from the session's command channel and therefore safe where the poll thread the module's docstring warns about was not. Every frame must be acked (Chrome sends nothing after an unacknowledged one, so a missed ack ends the recording rather than degrading it), and the rate is thinned at the source with `every_nth_frame` rather than buffered and discarded here. + - The buffer cap then needs care. Halving the buffer and keeping first/last — core's documented `maxBufferFrames` shape — drifts toward the run's end, because each decimation thins what is already held while new frames keep arriving unthinned: measured on a 40-frame run at a cap of 6, it kept frames 0, 1, 35, 37, 38, 39, i.e. the last moments and nothing from the middle. `_buffer` therefore thins the INCOMING frames by the same factor it has halved the buffer (`_stride` doubles per decimation), giving 0, 1, 11, 23, 31 for the same run and, at the real 2000 cap over 12000 frames, 1503 frames with 751 from the middle half and the final frame still present. Asserting only the endpoints does not catch this — tail truncation also leaves frame 0 plus whatever arrived since the last decimation, so the test has to assert something from the *middle* survives. + - Per-command screenshots keep being taken for the command ROWS while a stream is live, but stop feeding the video: the pushed frames already cover the timeline and interleaving would duplicate one of them a few milliseconds off. - **A drain must anchor the document it reads, and the flag for that has only ever had one value.** `core/script-loader.ts` `collectorDrainExpression(forceAnchor)` prepends `captureCurrentDom()` so a freshly injected collector's *async* initial anchor is not lost: the collector schedules it after `waitForBody`, so a drain issued right after a navigation beats it, reads an empty buffer, and the destination's buffer then dies with the page — leaving the navigating action with no DOM. Every production caller in both JS adapters passes `true` (selenium's `drainAfterLiveCommand`, its re-inject-after-navigation and teardown paths; nightwatch's five sites), so the `false` default is vestigial. Python's drain read `getTraceData()` with no anchor at all, which is the same missing backstop the preload does not cover; `selenium-devtools-py/src/selenium_devtools/snapshot.py` `_DRAIN_SCRIPT` now forces it **unconditionally and carries no flag** — one setting is not a knob. Forcing is free after the first anchor of a document because `packages/script` guards `captureCurrentDom` with an `#anchored` flag that deliberately survives its `reset()`, which is why selenium anchors on every live command and still emits ~3 anchors across a 16-row run rather than 16. - **Document-start injection is what removes the whole race class; everything else is reconstruction.** `