From 8b4de99861e787474be0eaa91bc96a1eb02b03ce Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:51:46 +0530
Subject: [PATCH 1/9] feat: split session_timeout into access_timeout +
 completion_timeout
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The single session_timeout knob conflated orphan defense (operator
never connects) with the human's work budget (operator connected
but is taking too long). With one number both jobs share, late
clicks silently shrink the work window — invisible to the
operator.

Replace with two timers, each at both ServerConfig (house default)
and per-call (override) layers:

  access_timeout     — start of wait → first WS connect.
                       Defends against orphaned sessions.
  completion_timeout — first WS connect → detection match.
                       Bounds work time.

None per-call inherits from ServerConfig; None on ServerConfig
means truly no timeout. Defaults: 600s access, 1800s completion.

In handoff.py, the single asyncio.wait_for becomes a three-way
race (_await_timeout_cause) between detection-match, access
deadline, and completion deadline. Returns the cause name
("access" / "completion") or None for a clean match — fed
directly into HandoffResult.timeout_cause for diagnostics.

Lazy listener install (substrate-URL leak defense) moves to a
side task armed on first connect; the race itself starts
immediately so access_timeout begins counting from the call,
not from connect.

Handoff.run(timeout=...) renamed to trigger_timeout=...; the new
access_timeout / completion_timeout kwargs are forwarded into
wait_for_completion on match.

WS upgrade rejects late operator clicks with 1008
access_timeout_expired once the access timer fires; the wrapper
handles that close code → expired card.

session_state WS message gains completion_deadline_ms (epoch
millis) anchored on first accept, reused across reconnects so
the wrapper countdown is reconnect-safe.

Breaking: ServerConfig.session_timeout and the deprecated
ServerConfig.completion_timeout alias are removed. Anyone still
passing them gets TypeError from the dataclass at construction.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 browser_handoff/handoff.py                 | 177 ++++++++++++++----
 browser_handoff/server/config.py           |  46 ++---
 browser_handoff/server/session.py          |  15 ++
 browser_handoff/server/streaming.py        |  55 +++++-
 tests/integration/test_server_lifecycle.py | 122 +++++++++++++
 tests/test_handoff.py                      | 200 ++++++++++++++++++++-
 tests/test_server.py                       |  82 +++------
 7 files changed, 563 insertions(+), 134 deletions(-)

diff --git a/browser_handoff/handoff.py b/browser_handoff/handoff.py
index e8f4736..5089d2d 100644
--- a/browser_handoff/handoff.py
+++ b/browser_handoff/handoff.py
@@ -10,7 +10,7 @@
 from contextlib import suppress
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 from .config import load_file, load_json, load_yaml
 from .detection.base import BaseDetection, DetectionResult
@@ -125,6 +125,17 @@ async def _capture_crop_metrics(
     return None
 
 
+def _resolve_timeout(
+    per_call: float | None, default: float | None
+) -> float | None:
+    """Resolve a per-call timeout against the ServerConfig default.
+
+    `None` per-call inherits the default; any other value (including
+    `math.inf`) overrides. `None` on both layers means truly disabled.
+    """
+    return default if per_call is None else per_call
+
+
 def _detection_tree_has_llm(detection: BaseDetection) -> bool:
     """True if `detection` or any nested child is an LLMDetection.
 
@@ -150,16 +161,19 @@ class HandoffResult:
     """Outcome of a Handoff.run() call.
 
     Three terminal states:
-      - was_blocked=False                  → no trigger fired within timeout
+      - was_blocked=False                  → no trigger fired within trigger_timeout
       - was_blocked=True, timed_out=False  → human completed the task
-      - was_blocked=True, timed_out=True   → human exceeded session_timeout
+      - was_blocked=True, timed_out=True   → one of the handoff timers fired
     """
 
     was_blocked: bool
     """Whether a trigger fired and a human handoff ran."""
 
     timed_out: bool = False
-    """Only meaningful if `was_blocked`: human exceeded session_timeout."""
+    """Only meaningful if `was_blocked`: either timer fired."""
+
+    timeout_cause: Literal["access", "completion"] | None = None
+    """Which timer fired. None unless `timed_out`."""
 
     scenario_name: str | None = None
     """Name of the scenario that fired."""
@@ -298,28 +312,34 @@ async def run(
         page: "Page",
         *,
         scenarios: list[Scenario] | None = None,
-        timeout: float = 30.0,
+        trigger_timeout: float = 30.0,
+        access_timeout: float | None = None,
+        completion_timeout: float | None = None,
         stream_url: str | None = None,
     ) -> "HandoffResult":
         """Watch for triggers; on match, run a handoff and await completion.
 
         Registers listeners on every scenario's trigger and waits for
-        one to fire (within `timeout`) or for `timeout` to elapse.
+        one to fire (within `trigger_timeout`) or for it to elapse.
 
         Args:
             page: Playwright page to monitor.
             scenarios: Trigger-completion pairs to watch. Falls back to
                 the scenarios set on the instance. ValueError if neither.
-            timeout: Max seconds to wait for any trigger. Does NOT bound
-                the human-completion phase — that uses
-                `ServerConfig.session_timeout` (default 600s).
+            trigger_timeout: Max seconds to wait for any trigger. Does
+                NOT bound the human-completion phase.
+            access_timeout: Per-call override for the pre-connect bound.
+                None inherits `ServerConfig.access_timeout`.
+            completion_timeout: Per-call override for the post-connect
+                work budget. None inherits `ServerConfig.completion_timeout`.
             stream_url: Optional substrate viewer URL. When set, the
                 handoff runs in passthrough mode and `stream_url` is
                 forwarded to `wait_for_completion`.
 
         Returns:
             HandoffResult describing what happened. Never raises on
-            completion-phase timeout — check `result.timed_out`.
+            handoff-phase timeout — check `result.timed_out` and
+            `result.timeout_cause`.
         """
         scenarios = scenarios if scenarios is not None else self.scenarios
         if not scenarios:
@@ -377,12 +397,14 @@ async def on_trigger(detection: BaseDetection) -> None:
 
             if not trigger_event.is_set():
                 with suppress(asyncio.TimeoutError):
-                    await asyncio.wait_for(trigger_event.wait(), timeout=timeout)
+                    await asyncio.wait_for(
+                        trigger_event.wait(), timeout=trigger_timeout
+                    )
 
             if matched_scenario is None or matched_result is None:
                 logger.info(
                     "handoff.run: no trigger matched within %.1fs (page url=%s)",
-                    timeout, page.url,
+                    trigger_timeout, page.url,
                 )
                 return HandoffResult(was_blocked=False)
 
@@ -396,6 +418,8 @@ async def on_trigger(detection: BaseDetection) -> None:
                 reason=matched_result.reason,
                 name=matched_scenario.name,
                 stream_url=stream_url,
+                access_timeout=access_timeout,
+                completion_timeout=completion_timeout,
             )
         finally:
             for cleanup in cleanups:
@@ -410,6 +434,8 @@ async def wait_for_completion(
         reason: str = "Human intervention required",
         name: str = "handoff",
         stream_url: str | None = None,
+        access_timeout: float | None = None,
+        completion_timeout: float | None = None,
     ) -> "HandoffResult":
         """Stream the page to a human *now* and wait until `on` matches.
 
@@ -426,11 +452,14 @@ async def wait_for_completion(
             stream_url: Optional substrate viewer URL. When set, the
                 wrapper iframes this URL; bh still owns detection,
                 notification, and lifecycle.
+            access_timeout: Per-call override of the pre-connect bound.
+                None inherits `ServerConfig.access_timeout`.
+            completion_timeout: Per-call override of the post-connect
+                work budget. None inherits `ServerConfig.completion_timeout`.
 
         Returns:
-            HandoffResult with `was_blocked=True`. Check `timed_out` for
-            whether the human finished within session_timeout. Never
-            raises on timeout.
+            HandoffResult with `was_blocked=True`. Check `timed_out` /
+            `timeout_cause` for which timer fired. Never raises on timeout.
         """
         context = page.context
         start_time = time.time()
@@ -439,6 +468,11 @@ async def wait_for_completion(
         completion_event = asyncio.Event()
         completion_reason: str | None = None
 
+        resolved_access = _resolve_timeout(access_timeout, self.server.access_timeout)
+        resolved_completion = _resolve_timeout(
+            completion_timeout, self.server.completion_timeout
+        )
+
         # Closure cell — the gated callback is defined before the
         # session exists; we patch it in once register_session returns.
         session_ref: dict[str, Any] = {"session": None}
@@ -493,6 +527,8 @@ async def on_completion_detected(detection: BaseDetection) -> None:
                 viewport_size=viewport_size,
                 stream_url=stream_url,
                 crop_metrics=crop_metrics,
+                access_timeout=resolved_access,
+                completion_timeout=resolved_completion,
             )
             session_ref["session"] = session
 
@@ -517,33 +553,28 @@ async def on_completion_detected(detection: BaseDetection) -> None:
                     completion_reason = initial.reason
                     completion_event.set()
 
-            # Lazy install: defer register_listeners until an operator
-            # opens the wrapper. Closes the substrate-URL leak — the
-            # in-page watcher only runs after wrapper auth via the
-            # access token. session_timeout bounds the wait.
-            timed_out = False
-            try:
-                if not completion_event.is_set():
-                    await asyncio.wait_for(
-                        session.presence.wait_until_connected(),
-                        timeout=self.server.session_timeout,
-                    )
+            # Lazy install gates listener registration on first connect
+            # (substrate-URL leak defense). The three-way race below
+            # starts immediately so access_timeout can fire before any
+            # connect; a separate side task arms listeners on connect.
+            timeout_cause: Literal["access", "completion"] | None = None
+
+            async def install_listeners_after_connect() -> None:
+                await session.presence.wait_until_connected()
+                if completion_event.is_set():
+                    return
+                listener_cleanups.append(
+                    on.register_listeners(page, on_completion_detected)
+                )
 
+            listener_install_task = asyncio.create_task(
+                install_listeners_after_connect()
+            )
+            try:
                 if not completion_event.is_set():
-                    listener_cleanups.append(
-                        on.register_listeners(page, on_completion_detected)
+                    timeout_cause = await self._await_timeout_cause(
+                        session, completion_event
                     )
-
-                await asyncio.wait_for(
-                    completion_event.wait(),
-                    timeout=self.server.session_timeout,
-                )
-            except asyncio.TimeoutError:
-                timed_out = True
-                logger.warning(
-                    "Handoff session_timeout: human did not finish "
-                    "within %.0fs", self.server.session_timeout,
-                )
             except asyncio.CancelledError:
                 # Caller (per-step timeout, ctrl-c, explicit cancel)
                 # gave up. Push a task_cancelled event so the wrapper
@@ -554,6 +585,16 @@ async def on_completion_detected(detection: BaseDetection) -> None:
                 with suppress(Exception):
                     await server.notify_task_cancelled(session_id)
                 raise
+            finally:
+                listener_install_task.cancel()
+                with suppress(asyncio.CancelledError, Exception):
+                    await listener_install_task
+
+            timed_out = timeout_cause is not None
+            if timed_out:
+                logger.warning(
+                    "Handoff %s_timeout fired", timeout_cause,
+                )
 
             await server.stop_screencast(session_id)
 
@@ -565,6 +606,7 @@ async def on_completion_detected(detection: BaseDetection) -> None:
             return HandoffResult(
                 was_blocked=True,
                 timed_out=timed_out,
+                timeout_cause=timeout_cause,
                 scenario_name=name,
                 trigger_reason=reason,
                 completion_reason=None if timed_out else completion_reason,
@@ -579,6 +621,63 @@ async def on_completion_detected(detection: BaseDetection) -> None:
                 await server.unregister_session(session_id)
             await self._release_server()
 
+    @staticmethod
+    async def _await_timeout_cause(
+        session: "Any", completion_event: asyncio.Event
+    ) -> Literal["access", "completion"] | None:
+        """Return the timer that fired, or None if detection matched first.
+
+        - "access": pre-first-connect window expired.
+        - "completion": post-first-connect work budget expired.
+        - None: detection matched (completion_event set).
+
+        Sets `session.access_timer_fired` right before returning "access"
+        so the WS guard can reject late operator clicks.
+        """
+        async def access_timeout_branch() -> Literal["access"]:
+            if session.access_timeout is None:
+                await asyncio.Event().wait()  # never fires
+            try:
+                await asyncio.wait_for(
+                    session.presence.wait_until_connected(),
+                    timeout=session.access_timeout,
+                )
+                # Connect won; retire. Block until outer cancels us.
+                await asyncio.Event().wait()
+            except asyncio.TimeoutError:
+                session.access_timer_fired = True
+                return "access"
+            # unreachable
+            return "access"
+
+        async def completion_timeout_branch() -> Literal["completion"]:
+            await session.presence.wait_until_connected()
+            if session.completion_timeout is None:
+                await asyncio.Event().wait()  # never fires
+            await asyncio.sleep(session.completion_timeout)
+            return "completion"
+
+        async def match_branch() -> None:
+            await completion_event.wait()
+            return None
+
+        tasks = [
+            asyncio.create_task(access_timeout_branch()),
+            asyncio.create_task(completion_timeout_branch()),
+            asyncio.create_task(match_branch()),
+        ]
+        try:
+            done, _pending = await asyncio.wait(
+                tasks, return_when=asyncio.FIRST_COMPLETED
+            )
+            return done.pop().result()
+        finally:
+            for t in tasks:
+                t.cancel()
+            for t in tasks:
+                with suppress(asyncio.CancelledError, Exception):
+                    await t
+
     async def _acquire_server(self) -> StreamingServer:
         """Return the shared streaming server, starting it on first use.
 
diff --git a/browser_handoff/server/config.py b/browser_handoff/server/config.py
index 37ef13c..54ee071 100644
--- a/browser_handoff/server/config.py
+++ b/browser_handoff/server/config.py
@@ -2,8 +2,7 @@
 
 from __future__ import annotations
 
-import warnings
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Any
 
 
@@ -21,41 +20,25 @@ class ServerConfig:
         public_base: Public base URL used in notification links
             (e.g. "https://my-tunnel.example.com"). Falls back to
             host:port, with wildcard binds rewritten to localhost.
-        session_timeout: Max seconds a session may live — the human's
-            budget and the lifetime of the stream-URL token.
+        access_timeout: Default seconds from `wait_for_completion()`
+            call to first operator WS connect. Bounds orphan sessions.
+            None disables this layer entirely.
+        completion_timeout: Default seconds from first operator WS
+            connect to detection match. Bounds the human's work budget.
+            None disables this layer entirely.
         jpeg_quality: JPEG quality for screencast frames (1-100).
         every_nth_frame: Capture 1 of every N frames Chrome produces;
             higher values reduce CPU/bandwidth at the cost of smoothness.
-
-    Deprecated:
-        completion_timeout: old name for `session_timeout`; accepted
-            with a DeprecationWarning and mirrored for read access.
     """
 
     host: str = "127.0.0.1"
     port: int = 8080
     public_base: str | None = None
-    session_timeout: float = 600.0
-    # None means "not supplied"; __post_init__ reconciles with session_timeout.
-    completion_timeout: float | None = None
+    access_timeout: float | None = 600.0
+    completion_timeout: float | None = 1800.0
     jpeg_quality: int = 75
     every_nth_frame: int = 1
 
-    def __post_init__(self) -> None:
-        # Frozen dataclass — assign through object.__setattr__.
-        if self.completion_timeout is not None:
-            warnings.warn(
-                "ServerConfig.completion_timeout is deprecated; use "
-                "session_timeout instead. It will be removed in a future "
-                "major release.",
-                DeprecationWarning,
-                stacklevel=2,
-            )
-            object.__setattr__(self, "session_timeout", self.completion_timeout)
-        # Mirror so code reading `config.completion_timeout` keeps working,
-        # silently.
-        object.__setattr__(self, "completion_timeout", self.session_timeout)
-
     def get_base_url(self) -> str:
         """Return the base URL for stream URLs.
 
@@ -72,7 +55,8 @@ def to_dict(self) -> dict[str, Any]:
             "host": self.host,
             "port": self.port,
             "public_base": self.public_base,
-            "session_timeout": self.session_timeout,
+            "access_timeout": self.access_timeout,
+            "completion_timeout": self.completion_timeout,
             "jpeg_quality": self.jpeg_quality,
             "every_nth_frame": self.every_nth_frame,
         }
@@ -86,10 +70,8 @@ def from_dict(cls, data: dict[str, Any]) -> "ServerConfig":
             jpeg_quality=data.get("jpeg_quality", 75),
             every_nth_frame=data.get("every_nth_frame", 1),
         )
-        # Prefer the new key; fall back to the deprecated one (warns via
-        # __post_init__).
-        if "session_timeout" in data:
-            kwargs["session_timeout"] = data["session_timeout"]
-        elif "completion_timeout" in data:
+        if "access_timeout" in data:
+            kwargs["access_timeout"] = data["access_timeout"]
+        if "completion_timeout" in data:
             kwargs["completion_timeout"] = data["completion_timeout"]
         return cls(**kwargs)
diff --git a/browser_handoff/server/session.py b/browser_handoff/server/session.py
index 762d2e7..830feb6 100644
--- a/browser_handoff/server/session.py
+++ b/browser_handoff/server/session.py
@@ -114,6 +114,21 @@ class HandoffSession:
     # iframe to just the page content. None when not in passthrough mode
     # or when the JS evaluate returned degenerate values.
     crop_metrics: dict[str, int] | None = None
+    # Per-session resolved timeouts. None at either layer means "no
+    # bound at this layer." access_timeout fires before first connect;
+    # completion_timeout fires after.
+    access_timeout: float | None = None
+    completion_timeout: float | None = None
+    # Set by the access-deadline task right before it returns. The WS
+    # upgrade handler reads this to reject late operator clicks with
+    # 1008. First-connect-gated, not wall-clock: a connect-then-drop
+    # before access_timeout retires the timer and leaves this False.
+    access_timer_fired: bool = False
+    # Wall-clock epoch (time.time) at which completion_timeout fires.
+    # Anchored on first WS connect so reconnects see the same deadline;
+    # the wrapper's countdown banner reads this via the session_state
+    # WS message. None when completion_timeout is None or no connect yet.
+    completion_deadline_ts: float | None = None
 
     @property
     def is_passthrough(self) -> bool:
diff --git a/browser_handoff/server/streaming.py b/browser_handoff/server/streaming.py
index 9775469..e86463c 100644
--- a/browser_handoff/server/streaming.py
+++ b/browser_handoff/server/streaming.py
@@ -352,12 +352,46 @@ async def websocket_endpoint(websocket: WebSocket, t: str | None = None):
                 await websocket.close()
                 return
 
+            # Access-timer guard: a late operator clicking past
+            # access_timeout must not get a live session. Close with
+            # 1008 + a known reason so the wrapper renders the expired
+            # card instead of the generic ended card.
+            if session_state.access_timer_fired:
+                await websocket.close(code=1008, reason="access_timeout_expired")
+                return
+
             session_state.websockets.append(websocket)
+            # First connect anchors the completion deadline against
+            # wall-clock so subsequent reconnects see the same banner
+            # countdown. Set BEFORE bump() so observers reading the
+            # deadline off a connected session always see a value.
+            if (
+                session_state.completion_deadline_ts is None
+                and session_state.completion_timeout is not None
+            ):
+                session_state.completion_deadline_ts = (
+                    time.time() + session_state.completion_timeout
+                )
             # One bump on accept flips both presence signals at once:
             # the connect gate for lazy install AND the freshness
             # timestamp the orchestration callback reads.
             session_state.presence.bump()
 
+            # Push completion deadline so the wrapper's countdown banner
+            # can render local time-remaining against an absolute epoch
+            # (drift-safe across reconnects). Omitted when there's no
+            # bound — wrapper hides the banner.
+            deadline_ms = (
+                int(session_state.completion_deadline_ts * 1000)
+                if session_state.completion_deadline_ts is not None
+                else None
+            )
+            payload: dict[str, Any] = {"type": "session_state"}
+            if deadline_ms is not None:
+                payload["completion_deadline_ms"] = deadline_ms
+            with suppress(Exception):
+                await websocket.send_json(payload)
+
             if session_state.is_passthrough:
                 try:
                     await self._handle_passthrough_websocket(websocket, session_state)
@@ -500,6 +534,8 @@ async def register_session(
         viewport_size: dict[str, int] | None = None,
         stream_url: str | None = None,
         crop_metrics: dict[str, int] | None = None,
+        access_timeout: float | None = None,
+        completion_timeout: float | None = None,
     ) -> HandoffSession:
         """Register a new Page for streaming.
 
@@ -515,6 +551,12 @@ async def register_session(
             crop_metrics: Six ints (screen_w/h, page_x/y, page_w/h) used
                 to crop the iframe in passthrough mode. Only meaningful
                 with `stream_url`.
+            access_timeout: Resolved bound for this session's pre-connect
+                wait. Stored on the session for the access-deadline task
+                to read.
+            completion_timeout: Resolved bound for this session's
+                post-connect work. Stored on the session; the WS handler
+                turns it into a wall-clock deadline on first connect.
         """
         cdp = await context.new_cdp_session(page)
         await cdp.send("Page.enable")
@@ -529,10 +571,17 @@ async def register_session(
             viewport_size=viewport_size or DEFAULT_VIEWPORT.copy(),
             stream_url=stream_url,
             crop_metrics=crop_metrics,
+            access_timeout=access_timeout,
+            completion_timeout=completion_timeout,
         )
-        # Token lifetime is bounded by the handoff's session_timeout —
-        # leaked links die with the handoff.
-        session.expires_at = time.time() + self.config.session_timeout
+        # Token-resolution cap: worst-case session lifetime is
+        # access + completion. If either layer is unbounded, the token
+        # has no wall-clock expiry — orchestration's unregister still
+        # drops it on handoff end.
+        if access_timeout is not None and completion_timeout is not None:
+            session.expires_at = time.time() + access_timeout + completion_timeout
+        else:
+            session.expires_at = None
         self.sessions[session_id] = session
         self._token_to_session[session.access_token] = session_id
 
diff --git a/tests/integration/test_server_lifecycle.py b/tests/integration/test_server_lifecycle.py
index bab46d0..9d7356c 100644
--- a/tests/integration/test_server_lifecycle.py
+++ b/tests/integration/test_server_lifecycle.py
@@ -135,6 +135,128 @@ async def test_concurrent_handoffs_share_one_server(
         await ctx2.close()
 
 
+async def test_access_timeout_fires_without_connect(
+    browser: Browser, base_url: str
+) -> None:
+    """Operator never connects → access timer fires, result carries
+    timeout_cause='access'."""
+    port = _free_port()
+    handoff = Handoff(
+        server=ServerConfig(
+            host="127.0.0.1",
+            port=port,
+            access_timeout=0.5,
+            completion_timeout=10.0,
+        ),
+    )
+
+    ctx = await browser.new_context()
+    page = await ctx.new_page()
+    try:
+        await page.goto(f"{base_url}/login")
+        # Never bump presence; access timer should fire.
+        result = await asyncio.wait_for(
+            handoff.wait_for_completion(
+                page, on=Detection.url(path_contains=["/dashboard"]),
+            ),
+            timeout=10,
+        )
+        assert result.was_blocked is True
+        assert result.timed_out is True
+        assert result.timeout_cause == "access"
+    finally:
+        await ctx.close()
+
+
+async def test_completion_timeout_fires_after_connect(
+    browser: Browser, base_url: str
+) -> None:
+    """Operator connects but never satisfies detection → completion
+    timer fires, result carries timeout_cause='completion'."""
+    port = _free_port()
+    handoff = Handoff(
+        server=ServerConfig(
+            host="127.0.0.1",
+            port=port,
+            access_timeout=10.0,
+            completion_timeout=0.5,
+        ),
+    )
+
+    ctx = await browser.new_context()
+    page = await ctx.new_page()
+    h = None
+    try:
+        await page.goto(f"{base_url}/login")
+        complete = Detection.url(path_contains=["/dashboard"])
+        h = asyncio.create_task(handoff.wait_for_completion(page, on=complete))
+
+        # Simulate the operator opening the wrapper. Once registered,
+        # bump presence so the access timer retires and the completion
+        # timer starts. The completion timer (0.5s) then fires.
+        await _wait_until(
+            lambda: handoff.is_serving and bool(handoff._server.sessions)
+        )
+        session = next(iter(handoff._server.sessions.values()))
+        session.presence.bump()
+
+        result = await asyncio.wait_for(h, timeout=10)
+        assert result.was_blocked is True
+        assert result.timed_out is True
+        assert result.timeout_cause == "completion"
+    finally:
+        if h is not None and not h.done():
+            h.cancel()
+            await asyncio.gather(h, return_exceptions=True)
+        await ctx.close()
+
+
+async def test_completion_timer_anchors_on_first_connect(
+    browser: Browser, base_url: str
+) -> None:
+    """A reconnect (second bump) must not reset the completion timer."""
+    port = _free_port()
+    handoff = Handoff(
+        server=ServerConfig(
+            host="127.0.0.1",
+            port=port,
+            access_timeout=10.0,
+            completion_timeout=1.5,
+        ),
+    )
+
+    ctx = await browser.new_context()
+    page = await ctx.new_page()
+    h = None
+    try:
+        await page.goto(f"{base_url}/login")
+        complete = Detection.url(path_contains=["/dashboard"])
+        h = asyncio.create_task(handoff.wait_for_completion(page, on=complete))
+
+        await _wait_until(
+            lambda: handoff.is_serving and bool(handoff._server.sessions)
+        )
+        session = next(iter(handoff._server.sessions.values()))
+        start = asyncio.get_running_loop().time()
+        session.presence.bump()
+        await asyncio.sleep(0.8)
+        session.presence.bump()  # reconnect — must not reset
+
+        result = await asyncio.wait_for(h, timeout=10)
+        elapsed = asyncio.get_running_loop().time() - start
+        assert result.timeout_cause == "completion"
+        # Anchored to first bump: fires ~1.5s elapsed.
+        # Reset bug: fires ~0.8+1.5 = 2.3s elapsed.
+        # Threshold 2.0 sits midway with ~0.5s slack on each side —
+        # enough headroom for scheduler jitter on slow runners.
+        assert elapsed < 2.0, f"completion timer may have reset (elapsed={elapsed:.3f})"
+    finally:
+        if h is not None and not h.done():
+            h.cancel()
+            await asyncio.gather(h, return_exceptions=True)
+        await ctx.close()
+
+
 async def test_sequential_handoffs_restart_server_on_same_port(
     browser: Browser, base_url: str
 ) -> None:
diff --git a/tests/test_handoff.py b/tests/test_handoff.py
index 26f465f..af53db1 100644
--- a/tests/test_handoff.py
+++ b/tests/test_handoff.py
@@ -120,7 +120,8 @@ def test_from_dict(self):
             ],
             "server": {
                 "port": 8080,
-                "session_timeout": 300,
+                "access_timeout": 120,
+                "completion_timeout": 300,
             },
             "notifiers": [
                 {"type": "slack", "webhook_url": "https://test.com/webhook"},
@@ -130,7 +131,8 @@ def test_from_dict(self):
         assert len(handoff.scenarios) == 1
         assert handoff.scenarios[0].name == "challenge"
         assert handoff.server.port == 8080
-        assert handoff.server.session_timeout == 300
+        assert handoff.server.access_timeout == 120
+        assert handoff.server.completion_timeout == 300
         assert len(handoff.notifiers) == 1
 
     def test_from_dict_without_scenarios_allowed(self):
@@ -267,6 +269,7 @@ def test_not_blocked(self):
         result = HandoffResult(was_blocked=False)
         assert result.was_blocked is False
         assert result.timed_out is False
+        assert result.timeout_cause is None
         assert result.scenario_name is None
         assert result.trigger_reason is None
         assert result.completion_reason is None
@@ -283,15 +286,17 @@ def test_completed(self):
         )
         assert result.was_blocked is True
         assert result.timed_out is False
+        assert result.timeout_cause is None
         assert result.scenario_name == "login_required"
         assert result.trigger_reason == "Login form detected"
         assert result.completion_reason == "URL matched /dashboard"
         assert result.duration == 10.5
 
-    def test_timed_out(self):
+    def test_timed_out_access(self):
         result = HandoffResult(
             was_blocked=True,
             timed_out=True,
+            timeout_cause="access",
             scenario_name="login_required",
             trigger_reason="Login form detected",
             completion_reason=None,
@@ -299,8 +304,157 @@ def test_timed_out(self):
         )
         assert result.was_blocked is True
         assert result.timed_out is True
+        assert result.timeout_cause == "access"
         assert result.completion_reason is None
 
+    def test_timed_out_completion(self):
+        result = HandoffResult(
+            was_blocked=True,
+            timed_out=True,
+            timeout_cause="completion",
+            scenario_name="login_required",
+            trigger_reason="Login form detected",
+            completion_reason=None,
+            duration=1800.0,
+        )
+        assert result.timeout_cause == "completion"
+
+
+class TestResolveTimeout:
+    """`_resolve_timeout` picks between per-call and config default.
+
+    None per-call inherits the default; any other value (including
+    math.inf) overrides. Pinning the layering rule directly because
+    it shapes every call-site override semantics.
+    """
+
+    def test_per_call_none_inherits_default(self):
+        from browser_handoff.handoff import _resolve_timeout
+
+        assert _resolve_timeout(None, 600.0) == 600.0
+        assert _resolve_timeout(None, None) is None
+
+    def test_per_call_value_overrides_default(self):
+        from browser_handoff.handoff import _resolve_timeout
+
+        assert _resolve_timeout(5.0, 600.0) == 5.0
+        # Per-call wins even when it's a "disable" sentinel.
+        import math
+
+        assert _resolve_timeout(math.inf, 600.0) == math.inf
+        # Per-call wins even when the default is unbounded.
+        assert _resolve_timeout(10.0, None) == 10.0
+
+
+class TestAwaitTimeoutCause:
+    """The three-way race returning the timeout cause (or None on match).
+
+    Drives `Handoff._await_timeout_cause` with a stand-in session so the
+    decision logic can be tested without a real browser or WS.
+    """
+
+    @staticmethod
+    def _session(
+        *,
+        access_timeout: float | None,
+        completion_timeout: float | None,
+    ):
+        from browser_handoff.server import SessionPresence
+
+        class _Session:
+            pass
+
+        s = _Session()
+        s.access_timeout = access_timeout
+        s.completion_timeout = completion_timeout
+        s.access_timer_fired = False
+        s.presence = SessionPresence()
+        return s
+
+    async def test_detection_match_wins(self):
+        import asyncio
+
+        from browser_handoff import Handoff
+
+        session = self._session(access_timeout=5.0, completion_timeout=5.0)
+        completion_event = asyncio.Event()
+        session.presence.bump()  # connect immediately
+        completion_event.set()  # detection already matched
+        result = await Handoff._await_timeout_cause(session, completion_event)
+        assert result is None
+        assert session.access_timer_fired is False
+
+    async def test_access_timeout_fires_without_connect(self):
+        import asyncio
+
+        from browser_handoff import Handoff
+
+        session = self._session(access_timeout=0.05, completion_timeout=10.0)
+        completion_event = asyncio.Event()
+        # Never bump presence — operator never connects.
+        result = await Handoff._await_timeout_cause(session, completion_event)
+        assert result == "access"
+        assert session.access_timer_fired is True
+
+    async def test_completion_timeout_fires_after_connect(self):
+        import asyncio
+
+        from browser_handoff import Handoff
+
+        session = self._session(access_timeout=10.0, completion_timeout=0.05)
+        completion_event = asyncio.Event()
+        session.presence.bump()  # connected
+        result = await Handoff._await_timeout_cause(session, completion_event)
+        assert result == "completion"
+        # Access timer was retired by the connect, not fired.
+        assert session.access_timer_fired is False
+
+    async def test_none_access_timeout_disables_access_timeout_branch(self):
+        import asyncio
+
+        from browser_handoff import Handoff
+
+        # Access disabled at this layer; completion still bounded so the
+        # race can resolve. Without a connect, completion never starts —
+        # use completion_event to terminate the race.
+        session = self._session(access_timeout=None, completion_timeout=None)
+        completion_event = asyncio.Event()
+        async def trip() -> None:
+            await asyncio.sleep(0.02)
+            completion_event.set()
+
+        asyncio.create_task(trip())
+        result = await asyncio.wait_for(
+            Handoff._await_timeout_cause(session, completion_event),
+            timeout=1.0,
+        )
+        assert result is None
+
+    async def test_completion_timer_anchors_on_first_connect_only(self):
+        """Bump twice; the completion_timeout sleep should start with the
+        first connect and not reset on the second."""
+        import asyncio
+        import time
+
+        from browser_handoff import Handoff
+
+        session = self._session(access_timeout=10.0, completion_timeout=0.15)
+        completion_event = asyncio.Event()
+
+        async def bump_twice() -> None:
+            session.presence.bump()
+            await asyncio.sleep(0.05)
+            session.presence.bump()  # reconnect — must not reset
+
+        asyncio.create_task(bump_twice())
+        start = time.monotonic()
+        result = await Handoff._await_timeout_cause(session, completion_event)
+        elapsed = time.monotonic() - start
+        assert result == "completion"
+        # Should fire ~0.15s after the first bump (≈0s), not after the
+        # second (≈0.05s). Allow generous slack for scheduler jitter.
+        assert elapsed < 0.30, f"completion timer appears to have reset (elapsed={elapsed:.3f})"
+
 
 @pytest.mark.filterwarnings("ignore::DeprecationWarning")
 class TestComplexConfig:
@@ -509,7 +663,7 @@ def test_handoff_with_scenarios_from_yaml(self):
         - ".otp-input"
 server:
   port: 8080
-  session_timeout: 600
+  completion_timeout: 600
 """
         handoff = Handoff.from_yaml(yaml_str)
         assert len(handoff.scenarios) == 2
@@ -551,7 +705,7 @@ def test_handoff_with_scenarios_from_file(self, tmp_path):
 
 server:
   port: 8080
-  session_timeout: 300
+  completion_timeout: 300
 
 notifiers:
   - type: slack
@@ -562,7 +716,7 @@ def test_handoff_with_scenarios_from_file(self, tmp_path):
         assert handoff.scenarios[0].name == "login_with_consent"
         assert handoff.scenarios[1].name == "google_oauth"
         assert handoff.server.port == 8080
-        assert handoff.server.session_timeout == 300
+        assert handoff.server.completion_timeout == 300
         assert len(handoff.notifiers) == 1
 
 
@@ -743,3 +897,37 @@ def test_run_accepts_stream_url(self):
         assert "stream_url" in sig.parameters
         assert sig.parameters["stream_url"].kind == inspect.Parameter.KEYWORD_ONLY
         assert sig.parameters["stream_url"].default is None
+
+
+class TestTimeoutKwargSignatures:
+    """Lock the renamed and added timeout kwargs at the call sites.
+
+    `Handoff.run(timeout=...)` was renamed to `trigger_timeout`. Both
+    entry points gained `access_timeout` and `completion_timeout`
+    overrides that default to None (= inherit ServerConfig).
+    """
+
+    def test_run_uses_trigger_timeout_not_timeout(self):
+        import inspect
+
+        sig = inspect.signature(Handoff.run)
+        assert "trigger_timeout" in sig.parameters
+        assert sig.parameters["trigger_timeout"].default == 30.0
+        # Old name must be gone — passing it should fail.
+        assert "timeout" not in sig.parameters
+
+    def test_run_has_timeout_overrides(self):
+        import inspect
+
+        sig = inspect.signature(Handoff.run)
+        for name in ("access_timeout", "completion_timeout"):
+            assert name in sig.parameters, name
+            assert sig.parameters[name].default is None
+
+    def test_wait_for_completion_has_timeout_overrides(self):
+        import inspect
+
+        sig = inspect.signature(Handoff.wait_for_completion)
+        for name in ("access_timeout", "completion_timeout"):
+            assert name in sig.parameters, name
+            assert sig.parameters[name].default is None
diff --git a/tests/test_server.py b/tests/test_server.py
index 964fbaa..d6ced2b 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -38,7 +38,8 @@ def test_default_values(self):
         assert config.host == "127.0.0.1"
         assert config.port == 8080
         assert config.public_base is None
-        assert config.session_timeout == 600.0
+        assert config.access_timeout == 600.0
+        assert config.completion_timeout == 1800.0
         assert config.jpeg_quality == 75
         assert config.every_nth_frame == 1
 
@@ -48,12 +49,26 @@ def test_custom_values(self):
             host="localhost",
             port=3000,
             public_base="https://proxy.example.com",
-            session_timeout=300.0,
+            access_timeout=120.0,
+            completion_timeout=300.0,
         )
         assert config.host == "localhost"
         assert config.port == 3000
         assert config.public_base == "https://proxy.example.com"
-        assert config.session_timeout == 300.0
+        assert config.access_timeout == 120.0
+        assert config.completion_timeout == 300.0
+
+    def test_timeout_layers_accept_none(self):
+        # None at the config layer means truly no bound at that layer.
+        config = ServerConfig(access_timeout=None, completion_timeout=None)
+        assert config.access_timeout is None
+        assert config.completion_timeout is None
+
+    def test_old_session_timeout_rejected(self):
+        # Clean break — the deprecated knob is gone; the dataclass
+        # raises immediately so misuse fails at construction.
+        with pytest.raises(TypeError):
+            ServerConfig(session_timeout=300.0)
 
     def test_get_base_url_with_public_base(self):
         """Test get_base_url with public_base set."""
@@ -81,14 +96,16 @@ def test_to_dict(self):
             host="0.0.0.0",
             port=8080,
             public_base="https://example.com",
-            session_timeout=120.0,
+            access_timeout=120.0,
+            completion_timeout=240.0,
         )
         data = config.to_dict()
         assert data == {
             "host": "0.0.0.0",
             "port": 8080,
             "public_base": "https://example.com",
-            "session_timeout": 120.0,
+            "access_timeout": 120.0,
+            "completion_timeout": 240.0,
             "jpeg_quality": 75,
             "every_nth_frame": 1,
         }
@@ -99,13 +116,15 @@ def test_from_dict(self):
             "host": "127.0.0.1",
             "port": 9000,
             "public_base": "https://proxy.test.com",
-            "session_timeout": 60.0,
+            "access_timeout": 30.0,
+            "completion_timeout": 60.0,
         }
         config = ServerConfig.from_dict(data)
         assert config.host == "127.0.0.1"
         assert config.port == 9000
         assert config.public_base == "https://proxy.test.com"
-        assert config.session_timeout == 60.0
+        assert config.access_timeout == 30.0
+        assert config.completion_timeout == 60.0
 
     def test_from_dict_defaults(self):
         """Test from_dict with missing values uses defaults."""
@@ -113,7 +132,8 @@ def test_from_dict_defaults(self):
         assert config.host == "127.0.0.1"
         assert config.port == 8080
         assert config.public_base is None
-        assert config.session_timeout == 600.0
+        assert config.access_timeout == 600.0
+        assert config.completion_timeout == 1800.0
 
     def test_from_dict_partial(self):
         """Test from_dict with partial values."""
@@ -244,52 +264,6 @@ def test_get_stream_url_is_deprecated_alias(self):
         assert any(issubclass(w.category, DeprecationWarning) for w in caught)
 
 
-class TestSessionTimeoutDeprecation:
-    """`completion_timeout` is renamed to `session_timeout` (deprecated alias).
-
-    The value bounds the whole session/token lifetime now, not just the
-    completion wait, so the name follows the meaning. The old name keeps
-    working (with a warning when set) for one major cycle.
-    """
-
-    def test_session_timeout_is_canonical(self):
-        import warnings
-
-        with warnings.catch_warnings():
-            warnings.simplefilter("error")  # any deprecation here would fail
-            assert ServerConfig().session_timeout == 600.0
-            assert ServerConfig(session_timeout=300.0).session_timeout == 300.0
-
-    def test_completion_timeout_still_readable_without_warning(self):
-        # Reading the alias on a config built the new way must not warn and
-        # must mirror session_timeout (old code keeps working).
-        import warnings
-
-        with warnings.catch_warnings():
-            warnings.simplefilter("error")
-            assert ServerConfig(session_timeout=120.0).completion_timeout == 120.0
-
-    def test_passing_completion_timeout_warns_and_applies(self):
-        with pytest.warns(DeprecationWarning, match="session_timeout"):
-            config = ServerConfig(completion_timeout=42.0)
-        assert config.session_timeout == 42.0
-        assert config.completion_timeout == 42.0  # alias mirrors it
-
-    def test_to_dict_uses_new_name(self):
-        data = ServerConfig(session_timeout=300.0).to_dict()
-        assert data["session_timeout"] == 300.0
-        assert "completion_timeout" not in data
-
-    def test_from_dict_new_name(self):
-        config = ServerConfig.from_dict({"session_timeout": 200.0})
-        assert config.session_timeout == 200.0
-
-    def test_from_dict_old_name_warns(self):
-        with pytest.warns(DeprecationWarning, match="session_timeout"):
-            config = ServerConfig.from_dict({"completion_timeout": 200.0})
-        assert config.session_timeout == 200.0
-
-
 class TestPassthroughSession:
     """HandoffSession.is_passthrough is derived from stream_url.
 

From 544d482e8d4a3b6bf0520986a1cce3771c65674d Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:57:49 +0530
Subject: [PATCH 2/9] feat(templates): completion countdown banner in
 intervention.html
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Renders the remaining completion_timeout in the toolbar against
the absolute completion_deadline_ms from the session_state WS
message. Switches to a warning style under 60s; hides when no
deadline is set; reconnect-safe because the deadline is absolute.

Handles the new WS close code (1008 access_timeout_expired) by
rendering the expired card on load — the operator clicked past
the access window.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 browser_handoff/templates/intervention.html | 80 +++++++++++++++++++--
 1 file changed, 76 insertions(+), 4 deletions(-)

diff --git a/browser_handoff/templates/intervention.html b/browser_handoff/templates/intervention.html
index 26e3fb5..2a267fe 100644
--- a/browser_handoff/templates/intervention.html
+++ b/browser_handoff/templates/intervention.html
@@ -436,6 +436,32 @@
         50%      { opacity: 0.4; }
       }
 
+      /* Completion-deadline countdown banner. Hidden until the server
+         pushes a deadline; switches to a warning style in the last 60s. */
+      .countdown-pill {
+        display: none;
+        align-items: center;
+        gap: 7px;
+        height: 26px;
+        padding: 0 10px;
+        background: rgba(255, 255, 255, 0.035);
+        border: 1px solid var(--border);
+        border-radius: 999px;
+        font-family: var(--font-mono);
+        font-size: 11px;
+        font-variant-numeric: tabular-nums;
+        color: var(--text-dim);
+        flex-shrink: 0;
+        text-transform: uppercase;
+        letter-spacing: 1.2px;
+      }
+      .countdown-pill.visible { display: inline-flex; }
+      .countdown-pill.warn {
+        color: var(--amber);
+        border-color: rgba(255, 149, 0, 0.4);
+        background: rgba(255, 149, 0, 0.08);
+      }
+
       #stream-container {
         position: relative;
         width: 100%;
@@ -776,8 +802,9 @@
       }
 
       /* Expired / ended overlays — same shape as the completion card,
-         different glyph + accent color. Expired (red) means session_timeout
-         fired; ended (gray) is the catchall for any other non-success exit:
+         different glyph + accent color. Expired (red) means a handoff
+         timeout fired (access or completion); ended (gray) is the
+         catchall for any other non-success exit:
          caller cancellation, abrupt process death, lost WS without a prior
          end event. The operator doesn't need the cause split out — "this is
          over and you can close the tab" is the only actionable framing. */
@@ -945,6 +972,10 @@
           />
         </div>
 
+        <div class="countdown-pill" id="countdown-pill">
+          <span id="countdown-value">--:--</span>
+        </div>
+
         <div class="latency-pill connecting" id="latency-pill">
           <span class="lat-dot"></span>
           <span class="lat-value" id="latency-value">Connecting</span>
@@ -1073,6 +1104,8 @@ <h3>Keyboard shortcuts</h3>
       const reasonExpandLabel = document.getElementById('reason-expand-label');
       const reasonDropdown = document.getElementById('reason-dropdown');
       const reconnectPill = document.getElementById('reconnect-pill');
+      const countdownPill = document.getElementById('countdown-pill');
+      const countdownValue = document.getElementById('countdown-value');
 
       // Show more / Show less for long reasons. Truncates in the header
       // with a right-edge fade; click opens an overlay panel (z-index 60)
@@ -1205,11 +1238,18 @@ <h3>Keyboard shortcuts</h3>
         lastPongTs = Date.now();
       };
 
-      ws.onclose = () => {
+      ws.onclose = (event) => {
         if (isCompleted) return;
+        streamConnected = false;
+        // Late-click rejection: server closed us with 1008 because
+        // access_timeout already fired. Surface the expired card
+        // directly instead of the generic ended one.
+        if (event && event.code === 1008 && event.reason === 'access_timeout_expired') {
+          showExpiredOverlay();
+          return;
+        }
         // No clean end event landed before close — the session is
         // unrecoverable; the cause doesn't matter to the operator.
-        streamConnected = false;
         showEndedOverlay();
       };
 
@@ -1248,6 +1288,10 @@ <h3>Keyboard shortcuts</h3>
           }
         } else if (message.type === 'url_changed') {
           setUrl(message.url);
+        } else if (message.type === 'session_state') {
+          if (typeof message.completion_deadline_ms === 'number') {
+            startCountdown(message.completion_deadline_ms);
+          }
         } else if (message.type === 'pong') {
           // Echo'd timestamp → RTT = now() - ts.
           if (typeof message.ts === 'number') {
@@ -1259,6 +1303,32 @@ <h3>Keyboard shortcuts</h3>
         }
       };
 
+      // Completion-deadline countdown. Anchored on a wall-clock epoch
+      // from the server so reconnects don't reset the visible time. Past
+      // zero, the server's normal expiry path renders the expired card —
+      // we just stop ticking and let the overlay take over.
+      let countdownTimer = null;
+      function startCountdown(deadlineMs) {
+        if (countdownTimer) clearInterval(countdownTimer);
+        countdownPill.classList.add('visible');
+        const tick = () => {
+          if (isCompleted) {
+            clearInterval(countdownTimer);
+            return;
+          }
+          const remaining = Math.max(0, deadlineMs - Date.now());
+          const totalSec = Math.ceil(remaining / 1000);
+          const m = Math.floor(totalSec / 60);
+          const s = totalSec % 60;
+          countdownValue.textContent =
+            String(m).padStart(2, '0') + ':' + String(s).padStart(2, '0');
+          countdownPill.classList.toggle('warn', remaining > 0 && remaining <= 60000);
+          if (remaining <= 0) clearInterval(countdownTimer);
+        };
+        tick();
+        countdownTimer = setInterval(tick, 1000);
+      }
+
       function nowTimeText() {
         return new Date().toLocaleTimeString([], {
           hour: '2-digit', minute: '2-digit', second: '2-digit', hour12: false,
@@ -1271,6 +1341,8 @@ <h3>Keyboard shortcuts</h3>
         // close doesn't flip the pill or re-show the reconnect overlay.
         isCompleted = true;
         latencyPill.style.display = 'none';
+        countdownPill.style.display = 'none';
+        if (countdownTimer) clearInterval(countdownTimer);
         streamContainer.classList.remove('reconnecting');
       }
 

From 1149445c79dacec74220dc18bd9ed2640802990c Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:57:57 +0530
Subject: [PATCH 3/9] feat(templates): completion countdown banner in
 proxy_intervention.html

Same countdown + expired-card handling as the screencast
template, ported to the passthrough proxy wrapper.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../templates/proxy_intervention.html         | 71 ++++++++++++++++++-
 1 file changed, 70 insertions(+), 1 deletion(-)

diff --git a/browser_handoff/templates/proxy_intervention.html b/browser_handoff/templates/proxy_intervention.html
index a8e9008..5ae2716 100644
--- a/browser_handoff/templates/proxy_intervention.html
+++ b/browser_handoff/templates/proxy_intervention.html
@@ -411,6 +411,32 @@
         50%      { opacity: 0.4; }
       }
 
+      /* Completion-deadline countdown. Hidden until the server pushes a
+         deadline; flips to warn in the last 60s. */
+      .countdown-pill {
+        display: none;
+        align-items: center;
+        gap: 7px;
+        height: 26px;
+        padding: 0 10px;
+        background: rgba(255, 255, 255, 0.035);
+        border: 1px solid var(--border);
+        border-radius: 999px;
+        font-family: var(--font-mono);
+        font-size: 11px;
+        font-variant-numeric: tabular-nums;
+        color: var(--text-dim);
+        flex-shrink: 0;
+        text-transform: uppercase;
+        letter-spacing: 1.2px;
+      }
+      .countdown-pill.visible { display: inline-flex; }
+      .countdown-pill.warn {
+        color: var(--amber);
+        border-color: rgba(255, 149, 0, 0.4);
+        background: rgba(255, 149, 0, 0.08);
+      }
+
       /* Stage container: hosts the cropped iframe of the substrate viewer.
          Two CSS profiles depending on whether we have crop metrics:
            - With crop_metrics: container at *page* aspect; iframe extends
@@ -747,6 +773,10 @@
           />
         </div>
 
+        <div class="countdown-pill" id="countdown-pill">
+          <span id="countdown-value">--:--</span>
+        </div>
+
         <div class="connection-pill connecting" id="connection-pill">
           <span class="conn-dot"></span>
           <span class="conn-label" id="connection-label">Connecting</span>
@@ -868,6 +898,8 @@ <h3>Keyboard shortcuts</h3>
       const reasonDropdown = document.getElementById('reason-dropdown');
       const connectionPill = document.getElementById('connection-pill');
       const connectionLabel = document.getElementById('connection-label');
+      const countdownPill = document.getElementById('countdown-pill');
+      const countdownValue = document.getElementById('countdown-value');
 
       // Show more / Show less for long reasons (see intervention.html
       // for the shared pattern).
@@ -981,6 +1013,33 @@ <h3>Keyboard shortcuts</h3>
       function markTerminal() {
         isTerminal = true;
         connectionPill.style.display = 'none';
+        countdownPill.style.display = 'none';
+        if (countdownTimer) clearInterval(countdownTimer);
+      }
+
+      // Completion-deadline countdown. Wall-clock anchored so reconnect
+      // doesn't reset the visible time. Past zero, the server's expiry
+      // path takes over and renders the expired card.
+      let countdownTimer = null;
+      function startCountdown(deadlineMs) {
+        if (countdownTimer) clearInterval(countdownTimer);
+        countdownPill.classList.add('visible');
+        const tick = () => {
+          if (isTerminal) {
+            clearInterval(countdownTimer);
+            return;
+          }
+          const remaining = Math.max(0, deadlineMs - Date.now());
+          const totalSec = Math.ceil(remaining / 1000);
+          const m = Math.floor(totalSec / 60);
+          const s = totalSec % 60;
+          countdownValue.textContent =
+            String(m).padStart(2, '0') + ':' + String(s).padStart(2, '0');
+          countdownPill.classList.toggle('warn', remaining > 0 && remaining <= 60000);
+          if (remaining <= 0) clearInterval(countdownTimer);
+        };
+        tick();
+        countdownTimer = setInterval(tick, 1000);
       }
 
       ws = new WebSocket(WS_URL);
@@ -997,6 +1056,10 @@ <h3>Keyboard shortcuts</h3>
         try { message = JSON.parse(event.data); } catch { return; }
         if (message.type === 'ready') {
           setConnectionState('live');
+        } else if (message.type === 'session_state') {
+          if (typeof message.completion_deadline_ms === 'number') {
+            startCountdown(message.completion_deadline_ms);
+          }
         } else if (message.type === 'url_changed') {
           setUrl(message.url);
         } else if (message.type === 'task_completed') {
@@ -1015,8 +1078,14 @@ <h3>Keyboard shortcuts</h3>
         showEndedOverlay();
       };
 
-      ws.onclose = () => {
+      ws.onclose = (event) => {
         if (isTerminal) return;
+        // Late-click rejection from the server's access-timer guard:
+        // surface the expired card instead of the generic ended one.
+        if (event && event.code === 1008 && event.reason === 'access_timeout_expired') {
+          showExpiredOverlay();
+          return;
+        }
         // No clean end event landed before close → the server is gone.
         // Surface the terminal card immediately instead of waiting on
         // a "Reconnecting" sweep that never reconnects.

From f8e17ce1f3039421b84445bc979d2a3e3eaaeaf3 Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:58:03 +0530
Subject: [PATCH 4/9] docs(readme): refresh timeout config references

Update the How-it-works and ServerConfig snippets to the new
access_timeout / completion_timeout names. trigger_timeout
replaces timeout on Handoff.run.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 README.md | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index d5e9aca..473ce7a 100644
--- a/README.md
+++ b/README.md
@@ -62,7 +62,7 @@ asyncio.run(main())
 
 A `Handoff` holds your transport config — the streaming server and notifiers — and is reusable across pages and runs. You decide *what* to watch for per call, so the same `Handoff` serves any number of scenarios.
 
-**Let the library detect the moment** with `handoff.run(page, scenarios=[...])`. A `Scenario` is a pair: a `trigger` that says "stop, a human is needed" and a `complete` that says "OK, they're done." `run` watches every scenario's trigger. If none fires within `timeout` seconds, it returns `HandoffResult(was_blocked=False)` and your script keeps going. If one fires, it starts a local streaming server, surfaces the URL (printed to logs and pushed to your notifiers), and waits until that scenario's `complete` matches — or until `server.session_timeout` elapses, in which case the result has `timed_out=True`. It never raises on timeout; check the result.
+**Let the library detect the moment** with `handoff.run(page, scenarios=[...])`. A `Scenario` is a pair: a `trigger` that says "stop, a human is needed" and a `complete` that says "OK, they're done." `run` watches every scenario's trigger. If none fires within `trigger_timeout` seconds, it returns `HandoffResult(was_blocked=False)` and your script keeps going. If one fires, it starts a local streaming server, surfaces the URL (printed to logs and pushed to your notifiers), and waits until that scenario's `complete` matches — or until one of the handoff timers fires (`access_timeout` if the operator never opens the link, `completion_timeout` if they open it but don't finish). On timeout the result has `timed_out=True` and `timeout_cause` set to `"access"` or `"completion"`. It never raises on timeout; check the result.
 
 **Already know a human is needed?** Skip trigger detection and stream right away with `handoff.wait_for_completion(page, on=...)`. This is the right call when something upstream already decided — e.g. an AI agent navigated to the payment page itself — so watching for a trigger would be redundant:
 
@@ -194,7 +194,8 @@ Handoff(
         host="127.0.0.1",                             # "0.0.0.0" to expose on LAN
         port=8080,
         public_base="https://my-tunnel.example.com",  # what notifiers link to
-        session_timeout=600,                          # max session lifetime / human wait (s)
+        access_timeout=600,                           # pre-connect bound (s)
+        completion_timeout=1800,                      # post-connect work budget (s)
         jpeg_quality=75,
         every_nth_frame=1,
     ),
@@ -203,7 +204,7 @@ Handoff(
 
 ### Access control
 
-The stream URL carries a high-entropy capability token (`…/?t=<token>`): whoever holds the link can view **and control** the page, so treat it like a password. The token is unguessable, decoupled from internal ids, and expires when the handoff finishes or `session_timeout` elapses — a stale link stops working. When exposing beyond loopback (`0.0.0.0`, a tunnel, or a sandbox preview URL), **serve over HTTPS/WSS** so the token isn't readable in transit; set `public_base` to your public `https://` origin and the operator link is built from it. There is no second factor yet — one leaked, still-active link grants control, so deliver it over a trusted channel.
+The stream URL carries a high-entropy capability token (`…/?t=<token>`): whoever holds the link can view **and control** the page, so treat it like a password. The token is unguessable, decoupled from internal ids, and expires when the handoff finishes or the worst-case session lifetime (`access_timeout` + `completion_timeout`) elapses — a stale link stops working. When exposing beyond loopback (`0.0.0.0`, a tunnel, or a sandbox preview URL), **serve over HTTPS/WSS** so the token isn't readable in transit; set `public_base` to your public `https://` origin and the operator link is built from it. There is no second factor yet — one leaked, still-active link grants control, so deliver it over a trusted channel.
 
 ## Examples
 

From e50a75e988bb26e560877d282cb5ff5c92b8e7f1 Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:58:11 +0530
Subject: [PATCH 5/9] example: rename timeout to trigger_timeout in
 browser_use_assisted_shopping/local.py

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 examples/browser_use_assisted_shopping/local.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/examples/browser_use_assisted_shopping/local.py b/examples/browser_use_assisted_shopping/local.py
index 7c3c177..81d73f6 100644
--- a/examples/browser_use_assisted_shopping/local.py
+++ b/examples/browser_use_assisted_shopping/local.py
@@ -142,7 +142,7 @@ async def request_human_help(reason: str, done_when: str) -> ActionResult:
 
         print(f"\n-> Handoff requested: {reason}\n   done_when: {done_when}\n")
         # Pause the agent while the human works — else browser-use's
-        # step_timeout races the human's session_timeout and the
+        # step_timeout races the human's completion_timeout and the
         # shorter one wins, cancelling the tool call mid-handoff.
         # try/finally guarantees resume on timeout or error.
         agent = agent_ref["agent"]

From acab3683ec38a2e7656558a2bdaf9b44b45f5840 Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:58:17 +0530
Subject: [PATCH 6/9] example: rename timeout to trigger_timeout in
 browser_use_assisted_shopping/using_kernel.py

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 examples/browser_use_assisted_shopping/using_kernel.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/examples/browser_use_assisted_shopping/using_kernel.py b/examples/browser_use_assisted_shopping/using_kernel.py
index 26827b0..8a39761 100644
--- a/examples/browser_use_assisted_shopping/using_kernel.py
+++ b/examples/browser_use_assisted_shopping/using_kernel.py
@@ -154,7 +154,7 @@ async def request_human_help(reason: str, done_when: str) -> ActionResult:
 
         print(f"\n-> Handoff requested: {reason}\n   done_when: {done_when}\n")
         # Pause the agent while the human works — else browser-use's
-        # step_timeout races the human's session_timeout and the
+        # step_timeout races the human's completion_timeout and the
         # shorter one wins, cancelling the tool call mid-handoff.
         # try/finally guarantees resume on timeout or error.
         agent = agent_ref["agent"]

From 06ef85172d04a0b4d40cb0ac1f578cda31670199 Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:58:21 +0530
Subject: [PATCH 7/9] example: rename timeout to trigger_timeout in
 claude_oauth_login_handoff/in_daytona.py

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 examples/claude_oauth_login_handoff/in_daytona.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/examples/claude_oauth_login_handoff/in_daytona.py b/examples/claude_oauth_login_handoff/in_daytona.py
index 42c9dff..e33a5cb 100644
--- a/examples/claude_oauth_login_handoff/in_daytona.py
+++ b/examples/claude_oauth_login_handoff/in_daytona.py
@@ -160,8 +160,8 @@ async def main() -> None:
 
             # Signed preview URL: token is in the URL, so the human
             # opens it without a daytona.io login or a custom header.
-            # 1h validity — well past session_timeout, with slack if
-            # the operator steps away.
+            # 1h validity — well past access + completion timeouts,
+            # with slack if the operator steps away.
             preview = await sandbox.create_signed_preview_url(
                 STREAMING_PORT, expires_in_seconds=3600
             )

From accd42cce9fafe2291a4c5ac127ff2ff584ca255 Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:58:26 +0530
Subject: [PATCH 8/9] example: refresh timeout API in
 claude_oauth_login_handoff/local.py

Rename timeout to trigger_timeout, and rewire the error message
to read result.timeout_cause instead of the removed
handoff.server.completion_timeout.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 examples/claude_oauth_login_handoff/local.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/examples/claude_oauth_login_handoff/local.py b/examples/claude_oauth_login_handoff/local.py
index ab787f4..f17eb7d 100644
--- a/examples/claude_oauth_login_handoff/local.py
+++ b/examples/claude_oauth_login_handoff/local.py
@@ -148,15 +148,15 @@ async def capture_code(authorize_url: str, server: CallbackServer) -> str:
                     # Fresh profile → claude.ai bounces /oauth/authorize → /login.
                     # The Login scenario fires, the human signs in, and run()
                     # returns once they land back on /oauth/authorize.
-                    result = await handoff.run(page, timeout=30)
+                    result = await handoff.run(page, trigger_timeout=30)
                     if result.was_blocked:
                         if result.timed_out:
                             console.print(
                                 "[red]✗ Human did not finish login in time[/red]"
                             )
                             raise TimeoutError(
-                                f"Human did not finish login within "
-                                f"{handoff.server.completion_timeout:.0f}s"
+                                f"Human did not finish login "
+                                f"(timeout_cause={result.timeout_cause})"
                             )
                         console.print(
                             f"[green]✓[/green] Login completed "

From 2d46a8c6019ff69d24bb3326f51a3a488366751e Mon Sep 17 00:00:00 2001
From: "synacktra.work@gmail.com" <synacktra.work@gmail.com>
Date: Tue, 30 Jun 2026 15:58:49 +0530
Subject: [PATCH 9/9] chore(gitignore): ignore temporary example files
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

examples/**/temp* — local scratch test scripts that shouldn't
land in the repo.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .gitignore | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.gitignore b/.gitignore
index 8ed21b4..5aa5c59 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,4 @@
+examples/**/temp*
 .claude
 
 # Byte-compiled / optimized / DLL files