diff --git a/examples/selenium/python-test/login.py b/examples/selenium/python-test/login.py
index 6ce9ec58..cf0cd14e 100644
--- a/examples/selenium/python-test/login.py
+++ b/examples/selenium/python-test/login.py
@@ -47,7 +47,7 @@
     # wait.until(EC.visibility_of_element_located((By.ID, "flash")))
     assert "/secure" in driver.current_url, driver.current_url
     flash = driver.find_element(By.ID, "flash").text
-    assert "You logged into a secure area1" in flash, flash
+    assert "You logged into a secure area" in flash, flash
 
     # wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "a.button")))
     driver.find_element(By.CSS_SELECTOR, "a.button").click()
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/bidi_preload.py b/packages/selenium-devtools-py/src/selenium_devtools/bidi_preload.py
new file mode 100644
index 00000000..6e2e40be
--- /dev/null
+++ b/packages/selenium-devtools-py/src/selenium_devtools/bidi_preload.py
@@ -0,0 +1,68 @@
+"""Document-start registration of the collector via a BiDi preload script.
+
+The ``<script>``-append path in ``snapshot.py`` can only instrument the document
+that exists when it runs, and a ``<script>`` dies with its document. So every
+navigation produces a document we learn about afterwards, and everything built on
+that — when to re-inject, when to drain, which action owns the new DOM — is
+reconstruction, and races. A preload registered for the session runs in EVERY
+document before any of that document's own script, so each one instruments itself
+and anchors its own DOM at its own ``performance.timeOrigin``. Nothing to detect,
+nothing to attribute.
+
+Mirrors ``core/bidi-preload.ts``, which the JS adapters use for the same reason.
+
+Requires a session created with ``webSocketUrl: true``; ``driver.script`` raises
+without it, so this reports False and the caller keeps the ``<script>`` path.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from .constants import BIDI_CAPABILITY, LOGGER_NAME
+
+_log = logging.getLogger(f"{LOGGER_NAME}.preload")
+
+
+def as_function_declaration(source: str) -> str:
+    """Wrap raw collector source as the function declaration BiDi expects.
+
+    Not the ``wrap_injectable`` IIFE: a preload script IS a function, so the
+    bundle's top-level await works directly in its body.
+    """
+    return f"async () => {{ {source} }}"
+
+
+def register_collector_preload(driver: Any, source: str) -> bool:
+    """Register the collector to run at document-start in every document of this
+    session. False means the caller must fall back to ``<script>`` injection.
+
+    Registered with NO browsing-context id, which is what makes it global —
+    contexts created later are covered, which is exactly the set this exists to
+    catch. `driver.script.pin` is the public API for it; its docstring says
+    "current browsing context", but it forwards no `contexts` argument, and BiDi
+    treats that as every context. That reliance is pinned by a test.
+    """
+    caps = getattr(driver, "caps", None)
+    if not (isinstance(caps, dict) and caps.get(BIDI_CAPABILITY)):
+        # Not a warning: the caller already reports the missing capability once,
+        # and the `<script>` path still captures DOM — just with the races this
+        # module exists to remove.
+        _log.debug(
+            "%s not set on the session — no document-start preload, using "
+            "per-document injection",
+            BIDI_CAPABILITY,
+        )
+        return False
+    try:
+        script = driver.script
+        script.pin(as_function_declaration(source))
+    except Exception as exc:  # noqa: BLE001 — any selenium/BiDi failure degrades
+        _log.warning(
+            "BiDi preload unavailable, falling back to per-document injection: %s",
+            exc,
+        )
+        return False
+    _log.info("collector registered at document-start (BiDi preload)")
+    return True
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
index 651ced7f..477e0ed3 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
@@ -19,7 +19,7 @@
 import weakref
 from typing import Any, Optional
 
-from . import bidi, frames
+from . import bidi, bidi_preload, frames
 from .capturer import SessionCapturer
 from .collector_source import reset_cache as reset_collector_cache
 from .constants import (
@@ -32,7 +32,11 @@
 )
 from .output_dir import resolve_adapter_output_dir
 from .screencast import ScreencastRecorder
-from .snapshot import SnapshotCapturer, start_snapshot_capture
+from .snapshot import (
+    SnapshotCapturer,
+    collector_source_text,
+    start_snapshot_capture,
+)
 from .sources import read_source
 from .utils import call_source, now_ms
 
@@ -325,7 +329,10 @@ def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> Optional[di
         if entry["session_id"] == session_id:
             return entry
         _close_entry(capturer, entry)  # same driver, new session
-    entry = {"session_id": session_id, "screencast": None, "snapshot": None}
+    entry = {
+        "session_id": session_id, "screencast": None, "snapshot": None,
+        "preloaded": False,
+    }
     try:
         sessions[driver] = entry
     except TypeError:  # not weak-referenceable
@@ -347,16 +354,30 @@ def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> Optional[di
     except Exception as exc:  # noqa: BLE001
         _log.warning("screencast start threw: %s", exc)
     try:
+        # Register the collector at document-start FIRST, while this runs before
+        # the session's first command executes — a preload registered after a
+        # navigation has already missed the document it needed to instrument.
+        # Falls back to per-document `<script>` injection when BiDi is absent.
+        origin = _backend_origin(capturer)
+        source = collector_source_text(backend=origin)
+        preloaded = bool(
+            source and bidi_preload.register_collector_preload(driver, source)
+        )
+        entry["preloaded"] = preloaded
         # Inject the packages/script DOM observer so the snapshot iframe fills.
         # Use a capture-bypassing execute_script so injection/readback scripts
-        # don't pollute the Actions timeline.
+        # don't pollute the Actions timeline. With the preload registered the
+        # capturer injects nothing and exists only to DRAIN the buffer.
         entry["snapshot"] = start_snapshot_capture(
             driver,
             execute_fn=_guarded_execute_script(driver),
-            backend=_backend_origin(capturer),
+            backend=origin,
+            preloaded=preloaded,
         )
         snapshot_cap = entry["snapshot"]
-        if snapshot_cap is not None and snapshot_cap.injected:
+        if preloaded:
+            pass  # already reported by register_collector_preload
+        elif snapshot_cap is not None and snapshot_cap.injected:
             _log.info("DOM snapshot collector injected")
         elif snapshot_cap is not None:
             # Kept rather than dropped, so a later command retries. Saying so
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/snapshot.py b/packages/selenium-devtools-py/src/selenium_devtools/snapshot.py
index 6611d1e7..e55b7fe1 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/snapshot.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/snapshot.py
@@ -50,6 +50,10 @@
 #: Cheap readiness probe used after injection.
 _READY_SCRIPT = 'return typeof window.wdioTraceCollector !== "undefined";'
 
+#: Distinguishes "the read itself failed" from "the collector is absent here",
+#: which the page returns as a plain null and which is the recovery signal.
+_UNREADABLE = object()
+
 #: Session-lost signatures — expected during teardown, so silenced rather than
 #: logged (matches the JS adapter's error filter).
 _QUIET_ERRORS = ("ECONNREFUSED", "no such session", "invalid session id")
@@ -93,12 +97,17 @@ def wrap_injectable(script_content: str) -> str:
     return f"(async function() {{ {script_content} }})()"
 
 
-def load_injectable_script(
+def collector_source_text(
     path: Optional[str] = None,
     *,
     backend: Optional[tuple] = None,
 ) -> Optional[str]:
-    """The IIFE-wrapped collector source, or None if it cannot be obtained.
+    """The collector's RAW source, or None if it cannot be obtained.
+
+    Raw rather than wrapped, because the two ways of installing it need
+    different envelopes: a ``<script>`` body needs the async IIFE
+    (`wrap_injectable`), while a BiDi preload script is itself a function
+    declaration and takes the source in its body.
 
     ``backend`` is the ``(host, port)`` of the connected backend, which serves
     the collector out of the package it depends on — the only route that works
@@ -110,7 +119,7 @@ def load_injectable_script(
     if path is None and backend is not None:
         source = fetch_collector_source(*backend)
         if source is not None:
-            return wrap_injectable(source)
+            return source
     resolved = path or resolve_script_path()
     if not resolved:
         _warn(
@@ -120,11 +129,20 @@ def load_injectable_script(
         return None
     try:
         with open(resolved, "r", encoding="utf-8") as handle:
-            content = handle.read()
+            return handle.read()
     except OSError as exc:
         _warn(f"could not read injected script ({resolved}): {exc}")
         return None
-    return wrap_injectable(content)
+
+
+def load_injectable_script(
+    path: Optional[str] = None,
+    *,
+    backend: Optional[tuple] = None,
+) -> Optional[str]:
+    """The IIFE-wrapped collector source for the ``<script>`` path, or None."""
+    source = collector_source_text(path, backend=backend)
+    return None if source is None else wrap_injectable(source)
 
 
 def normalize_mutations(trace_data: Any) -> List[Any]:
@@ -155,11 +173,16 @@ def __init__(
         *,
         script_path: Optional[str] = None,
         backend: Optional[tuple] = None,
+        preloaded: bool = False,
     ) -> None:
         self._execute = execute_fn
         self._script_path = script_path
         self._backend = backend
-        self._injected = False
+        # A document-start preload already installed the collector in every
+        # document of this session, so there is nothing to inject — but the
+        # capturer is still what DRAINS the buffer, so it is still needed.
+        self._preloaded = preloaded
+        self._injected = preloaded
 
     @property
     def injected(self) -> bool:
@@ -171,10 +194,13 @@ def inject(self) -> bool:
         Navigation wipes the injected collector, so we probe the live page each
         call and re-install if it's gone (matching the JS adapter, which injects
         per navigation) rather than trusting a one-time flag. Failures are logged
-        no-ops."""
-        wrapped = load_injectable_script(self._script_path, backend=self._backend)
-        if wrapped is None:
-            return False
+        no-ops. A no-op when a document-start preload is registered: the
+        collector is present in every document before any of its script runs,
+        which is the whole point of the preload. A document where the preload
+        nonetheless did not take is recovered by `pull_mutations`, whose own null
+        is the only signal that says so."""
+        if self._preloaded:
+            return True
         try:
             if self._execute(_READY_SCRIPT) is True:
                 self._injected = True
@@ -182,6 +208,18 @@ def inject(self) -> bool:
         except BaseException as exc:  # noqa: BLE001 — probe failure → try install
             if not _is_quiet_error(exc):
                 _warn(f"readiness probe failed: {exc}")
+        self._injected = self._install_now()
+        if not self._injected:
+            _warn("collector not detected immediately after injection")
+        return self._injected
+
+    def _install_now(self) -> bool:
+        """Install the collector into the CURRENT document via ``<script>``.
+        Used both by `inject` and by the drain's recovery, which already knows
+        the collector is absent and so skips the readiness probe."""
+        wrapped = load_injectable_script(self._script_path, backend=self._backend)
+        if wrapped is None:
+            return False
         try:
             self._execute(_INJECT_SCRIPT, wrapped)
             ready = self._execute(_READY_SCRIPT)
@@ -190,20 +228,36 @@ def inject(self) -> bool:
                 _warn(f"injection failed: {exc}")
             return False
         self._injected = ready is True
-        if ready is not True:
-            _warn("collector not detected immediately after injection")
         return self._injected
 
-    def pull_mutations(self) -> List[Any]:
-        """Read and drain the buffered mutations (``getTraceData()`` resets the
-        page-side buffer). Returns [] on any failure or when nothing's buffered."""
+    def _read_trace(self) -> Any:
+        """The page's trace payload, None when the collector is absent from this
+        document, or `_UNREADABLE` when the read itself failed."""
         try:
-            trace_data = self._execute(_READ_TRACE_SCRIPT)
+            return self._execute(_READ_TRACE_SCRIPT)
         except BaseException as exc:  # noqa: BLE001
             if not _is_quiet_error(exc):
                 _warn(f"trace read failed: {exc}")
+            return _UNREADABLE
+
+    def pull_mutations(self) -> List[Any]:
+        """Read and drain the buffered mutations (``getTraceData()`` resets the
+        page-side buffer). Returns [] on any failure or when nothing's buffered.
+
+        A null payload means the collector is not in THIS document, and that is
+        the only signal which says so — a preload can still miss one (its script
+        can throw, or a document can predate registration), and with the preload
+        registered nothing else probes. Recovered once here, mirroring core's
+        `drainCollectorWithRecovery`, which costs nothing on the happy path
+        because the drain happens either way."""
+        data = self._read_trace()
+        if data is _UNREADABLE:
             return []
-        return normalize_mutations(trace_data)
+        if data is None and self._install_now():
+            data = self._read_trace()
+            if data is _UNREADABLE:
+                return []
+        return normalize_mutations(data)
 
 
 def start_snapshot_capture(
@@ -212,6 +266,7 @@ def start_snapshot_capture(
     script_path: Optional[str] = None,
     execute_fn: Optional[ExecuteFn] = None,
     backend: Optional[tuple] = None,
+    preloaded: bool = False,
 ) -> Optional[SnapshotCapturer]:
     """Build a ``SnapshotCapturer`` and attempt the first injection. Returns the
     capturer — including when that injection fails, so a later command can retry
@@ -223,7 +278,9 @@ def start_snapshot_capture(
     if not callable(run):
         _warn("driver has no execute_script — snapshot capture skipped")
         return None
-    capturer = SnapshotCapturer(run, script_path=script_path, backend=backend)
+    capturer = SnapshotCapturer(
+        run, script_path=script_path, backend=backend, preloaded=preloaded
+    )
     # The capturer is returned even when this first injection fails. It used to
     # return None, which made the failure terminal: the caller stores None, its
     # post-command refresh skips a missing capturer, and `inject()` is never
diff --git a/packages/selenium-devtools-py/tests/test_bidi_preload.py b/packages/selenium-devtools-py/tests/test_bidi_preload.py
new file mode 100644
index 00000000..e6471ae6
--- /dev/null
+++ b/packages/selenium-devtools-py/tests/test_bidi_preload.py
@@ -0,0 +1,182 @@
+"""Document-start registration of the collector.
+
+A `<script>`-append can only instrument the document that exists when it runs,
+and dies with it — so every navigation yields a document we learn about after the
+fact. A preload runs in EVERY document before that document's own script, which
+removes the whole class of "when do we re-inject / who owns this DOM" races.
+"""
+
+import importlib.util
+import unittest
+from unittest import mock
+
+from selenium_devtools import bidi_preload
+from selenium_devtools.snapshot import SnapshotCapturer
+
+
+class FakeScript:
+    def __init__(self, raises=None):
+        self.pinned = []
+        self._raises = raises
+
+    def pin(self, script):
+        if self._raises:
+            raise self._raises
+        self.pinned.append(script)
+        return "preload-id"
+
+
+class FakeDriver:
+    def __init__(self, caps=None, script=None):
+        self.caps = caps if caps is not None else {"webSocketUrl": "ws://x"}
+        self._script = script or FakeScript()
+
+    @property
+    def script(self):
+        return self._script
+
+
+class TestRegistering(unittest.TestCase):
+    def test_registers_the_collector_as_a_function_declaration(self):
+        # A preload script IS a function, so the bundle's top-level await works
+        # in its body — unlike the `<script>` path, which needs the async IIFE.
+        driver = FakeDriver()
+
+        self.assertTrue(
+            bidi_preload.register_collector_preload(driver, "COLLECTOR")
+        )
+        self.assertEqual(driver.script.pinned, ["async () => { COLLECTOR }"])
+
+    def test_a_session_without_the_bidi_capability_falls_back(self):
+        # No webSocketUrl means driver.script cannot open. The `<script>` path
+        # still captures DOM, just with the races the preload removes.
+        driver = FakeDriver(caps={})
+
+        self.assertFalse(
+            bidi_preload.register_collector_preload(driver, "COLLECTOR")
+        )
+
+    def test_a_selenium_failure_falls_back_with_a_reason(self):
+        driver = FakeDriver(script=FakeScript(raises=RuntimeError("no bidi")))
+
+        with self.assertLogs("selenium_devtools.preload", level="WARNING") as logs:
+            self.assertFalse(
+                bidi_preload.register_collector_preload(driver, "COLLECTOR")
+            )
+
+        self.assertIn("falling back to per-document injection", "\n".join(logs.output))
+
+
+class TestTheScriptPathIsGatedNotDeleted(unittest.TestCase):
+    """The `<script>` path is the only capture without BiDi, so it stays — it is
+    simply not used when a preload already installed the collector everywhere."""
+
+    def test_a_preloaded_capturer_injects_nothing(self):
+        calls = {"n": 0}
+
+        def execute(*_args):
+            calls["n"] += 1
+            return True
+
+        capturer = SnapshotCapturer(execute, preloaded=True)
+
+        self.assertTrue(capturer.injected)  # present from document-start
+        self.assertTrue(capturer.inject())
+        self.assertEqual(calls["n"], 0)  # no probe, no injection
+
+    def test_without_a_preload_the_script_path_still_runs(self):
+        calls = {"n": 0}
+
+        def execute(*_args):
+            calls["n"] += 1
+            return True
+
+        capturer = SnapshotCapturer(execute, script_path=None, preloaded=False)
+        with mock.patch(
+            "selenium_devtools.snapshot.load_injectable_script", return_value="SRC"
+        ):
+            self.assertTrue(capturer.inject())
+
+        self.assertGreater(calls["n"], 0)
+
+
+class TestADocumentThePreloadMissed(unittest.TestCase):
+    """A preload covers every document in principle, but its script can throw,
+    and a document can predate registration. With the preload registered nothing
+    probes any more, so the drain's own null is the only signal that a document
+    has no collector — and it has to act on it."""
+
+    def test_the_drain_recovers_a_document_without_a_collector(self):
+        state = {"present": False}
+        seen = []
+
+        def execute(script, *_args):
+            seen.append(script)
+            if "createElement" in script:
+                state["present"] = True
+                return True
+            if "getTraceData" in script:
+                return {"mutations": [{"type": "childList"}]} if state["present"] else None
+            return state["present"]  # readiness probe
+
+        capturer = SnapshotCapturer(execute, preloaded=True)
+        with mock.patch(
+            "selenium_devtools.snapshot.load_injectable_script", return_value="SRC"
+        ):
+            mutations = capturer.pull_mutations()
+
+        self.assertEqual(len(mutations), 1)  # recovered rather than silently empty
+        self.assertTrue(any("createElement" in s for s in seen))
+
+    def test_a_healthy_preloaded_document_is_not_reinstalled(self):
+        # Recovery must cost nothing on the happy path — the drain already
+        # happens, and a document that answers needs no <script> at all.
+        seen = []
+
+        def execute(script, *_args):
+            seen.append(script)
+            return {"mutations": []}
+
+        capturer = SnapshotCapturer(execute, preloaded=True)
+        capturer.pull_mutations()
+
+        self.assertEqual([s for s in seen if "createElement" in s], [])
+
+
+#: selenium is the adapter's only runtime requirement, but the package is
+#: deliberately importable and unit-testable without it — the CI job installs
+#: nothing and runs `PYTHONPATH=src python -m unittest`. Mirrors the
+#: `resolve_script_path()` guard in test_snapshot.py on the same principle.
+_HAS_SELENIUM = importlib.util.find_spec("selenium") is not None
+
+
+@unittest.skipUnless(_HAS_SELENIUM, "selenium is not installed")
+class TestTheSeleniumSurfaceWeDependOn(unittest.TestCase):
+    """`pin()` is public, but its docstring says "current browsing context" while
+    we depend on it registering GLOBALLY so documents created later are covered.
+    That reliance is undocumented, so it is pinned here: if selenium ever scopes
+    `pin` to one context, this fails instead of the preload silently covering
+    only the first document."""
+
+    def test_pin_registers_without_a_browsing_context(self):
+        from selenium.webdriver.common.bidi.script import Script
+
+        seen = {}
+
+        def fake_add(self, function_declaration, *args, **kwargs):
+            seen["args"] = args
+            seen["kwargs"] = kwargs
+            return "id"
+
+        # A bare instance: constructing a real Script needs a live driver, and
+        # only the dispatch from pin() to _add_preload_script is under test.
+        script = Script.__new__(Script)
+        with mock.patch.object(Script, "_add_preload_script", fake_add):
+            script.pin("async () => {}")
+
+        self.assertEqual(seen["args"], ())
+        self.assertIsNone(seen["kwargs"].get("contexts"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/packages/selenium-devtools-py/tests/test_snapshot.py b/packages/selenium-devtools-py/tests/test_snapshot.py
index 63d23466..734fe07c 100644
--- a/packages/selenium-devtools-py/tests/test_snapshot.py
+++ b/packages/selenium-devtools-py/tests/test_snapshot.py
@@ -118,12 +118,16 @@ def test_inject_reinjects_after_navigation(self):
         installs = [c for c in exec_fn.calls if "createElement" in c[0]]
         self.assertEqual(len(installs), 2)
 
-    def test_inject_missing_script_is_noop(self):
+    def test_inject_without_a_source_installs_nothing(self):
+        # The readiness probe comes first now, so an already-instrumented page
+        # costs no source load at all. With no source AND no collector present,
+        # nothing is installed and the capturer stays uninjected.
         exec_fn = FakeExec()
         cap = SnapshotCapturer(exec_fn, script_path="/no/such.js")
+
         self.assertFalse(cap.inject())
         self.assertFalse(cap.injected)
-        self.assertEqual(exec_fn.calls, [])  # script missing → no execute at all
+        self.assertEqual([c for c in exec_fn.calls if "createElement" in c[0]], [])
 
     def test_inject_swallows_execute_errors(self):
         def boom(script, *args):
@@ -196,19 +200,21 @@ def execute_script(self, script, *args):
 
     def test_a_later_command_can_still_install_the_collector(self):
         # The retry path only exists if the capturer is still around to use it.
-        calls = {"n": 0}
+        ready = {"present": False}
 
         class Driver:
             def execute_script(self, script, *args):
-                calls["n"] += 1
-                return True  # the page reports the collector present
+                if "createElement" in script:
+                    ready["present"] = True  # the install lands
+                    return True
+                return ready["present"]  # the readiness probe
 
         with mock.patch(
             "selenium_devtools.snapshot.load_injectable_script",
-            side_effect=[None, "SOURCE"],  # first attempt has no source, then it does
+            side_effect=[None, "SOURCE"],  # no source first, then it arrives
         ):
             capturer = start_snapshot_capture(Driver(), script_path="/no/such.js")
-            self.assertFalse(capturer.injected)
+            self.assertFalse(capturer.injected)  # nothing to install yet
 
             self.assertTrue(capturer.inject())  # the next command retries