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
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
 
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/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');
       }
 
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.
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"]
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"]
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
             )
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 "
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.