diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 55088f3..0e1c4fd 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -72,7 +72,9 @@ jobs:
       # and the marketing site drift into looking like two products.
       - name: Check the vendored design tokens against openadapt-web
         env:
-          GITHUB_TOKEN: ${{ github.token }}
+          # github.token cannot read private OpenAdaptAI/openadapt-web.
+          # ADMIN_TOKEN can; fall back so local/fork runs still try github.token.
+          GITHUB_TOKEN: ${{ secrets.ADMIN_TOKEN || github.token }}
         run: npm run tokens:check
 
   python-distribution:
diff --git a/engine/auth/runner_bind.py b/engine/auth/runner_bind.py
new file mode 100644
index 0000000..567d39f
--- /dev/null
+++ b/engine/auth/runner_bind.py
@@ -0,0 +1,127 @@
+"""Parse-only grammar for ``openadapt://runner`` authoring bind URIs.
+
+Tauri validates the same fields first. Python parses again so neither IPC nor
+an operating-system protocol invocation can become a general command. This
+module does not claim, store, or poll.
+"""
+
+from __future__ import annotations
+
+import re
+from urllib.parse import parse_qs, urlparse, urlsplit
+
+AUTHORING_ORIGIN = "https://openadapt.ai"
+MAX_URI_BYTES = 2048
+ALLOWED_FIELDS = frozenset({"pack", "bind", "origin"})
+
+BIND_TOKEN_RE = re.compile(r"^oab_[A-Za-z0-9_-]{43}$")
+LEASE_SECRET_RE = re.compile(r"^oals_[a-f0-9]{64}$")
+PACK_ALIAS_RE = re.compile(r"^p\.[A-Za-z0-9_-]{12}$")
+PACK_CIPHER_RE = re.compile(r"^v1\.[A-Za-z0-9_-]{32,2000}$")
+CLOUD_RUNNER_TOKEN_RE = re.compile(r"^oar_[a-f0-9]{64}$")
+PAIRING_SECRET_RE = re.compile(r"^oap_[A-Za-z0-9_-]{43}$")
+BIND_HEX_BODY_RE = re.compile(r"^oab_[a-f0-9]{64}$")
+LEASE_BASE64URL_BODY_RE = re.compile(r"^oals_[A-Za-z0-9_-]{43}$")
+
+
+class RunnerBindError(RuntimeError):
+    """A safe, user-facing runner-link failure with no secret-bearing text."""
+
+
+def valid_bind_token(value: object) -> bool:
+    """Return whether ``value`` is exactly one ``oab_`` bind token."""
+
+    if not isinstance(value, str):
+        return False
+    if (
+        CLOUD_RUNNER_TOKEN_RE.fullmatch(value) is not None
+        or PAIRING_SECRET_RE.fullmatch(value) is not None
+        or BIND_HEX_BODY_RE.fullmatch(value) is not None
+    ):
+        return False
+    return BIND_TOKEN_RE.fullmatch(value) is not None
+
+
+def valid_lease_secret(value: object) -> bool:
+    """Return whether ``value`` is exactly one ``oals_`` mailbox lease secret."""
+
+    if not isinstance(value, str):
+        return False
+    if (
+        CLOUD_RUNNER_TOKEN_RE.fullmatch(value) is not None
+        or PAIRING_SECRET_RE.fullmatch(value) is not None
+        or LEASE_BASE64URL_BODY_RE.fullmatch(value) is not None
+    ):
+        return False
+    return LEASE_SECRET_RE.fullmatch(value) is not None
+
+
+def valid_pack_id(value: object) -> bool:
+    """Return whether ``value`` is a ``p.`` alias or ``v1.`` ciphertext id."""
+
+    if not isinstance(value, str):
+        return False
+    return PACK_ALIAS_RE.fullmatch(value) is not None or PACK_CIPHER_RE.fullmatch(value) is not None
+
+
+def canonical_authoring_origin(value: object) -> str:
+    """Return the pinned production authoring origin, or raise."""
+
+    if not isinstance(value, str):
+        raise RunnerBindError("Runner link does not name the OpenAdapt authoring origin")
+    try:
+        parsed = urlsplit(value)
+        port = parsed.port
+    except ValueError as exc:
+        raise RunnerBindError("Runner link does not name the OpenAdapt authoring origin") from exc
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != "openadapt.ai"
+        or parsed.netloc != "openadapt.ai"
+        or parsed.username
+        or parsed.password
+        or parsed.path not in ("",)
+        or parsed.query
+        or parsed.fragment
+        or port is not None
+        or value != AUTHORING_ORIGIN
+    ):
+        raise RunnerBindError("Runner link does not name the OpenAdapt authoring origin")
+    return AUTHORING_ORIGIN
+
+
+def parse_runner_uri(uri: object) -> dict[str, str]:
+    """Parse the fixed runner action and reject ambiguity or extra fields."""
+
+    if not isinstance(uri, str) or not uri or len(uri) > MAX_URI_BYTES:
+        raise RunnerBindError("Invalid OpenAdapt runner link")
+    parsed = urlparse(uri)
+    if (
+        parsed.scheme != "openadapt"
+        or parsed.netloc != "runner"
+        or parsed.path not in ("", "/")
+        or parsed.params
+        or parsed.fragment
+        or parsed.username
+        or parsed.password
+    ):
+        raise RunnerBindError("Invalid OpenAdapt runner link")
+    try:
+        query = parse_qs(parsed.query, keep_blank_values=True, strict_parsing=True)
+    except ValueError as exc:
+        raise RunnerBindError("Invalid OpenAdapt runner link") from exc
+    if set(query) - ALLOWED_FIELDS or any(len(values) != 1 for values in query.values()):
+        raise RunnerBindError("Runner link contains unknown or duplicate fields")
+    if set(query) != ALLOWED_FIELDS:
+        raise RunnerBindError("Runner link is missing pack, bind, or origin")
+
+    pack = query["pack"][0]
+    bind = query["bind"][0]
+    origin = canonical_authoring_origin(query["origin"][0])
+    if not valid_pack_id(pack):
+        raise RunnerBindError("Pack id is malformed")
+    if CLOUD_RUNNER_TOKEN_RE.fullmatch(bind) or PAIRING_SECRET_RE.fullmatch(bind):
+        raise RunnerBindError("Bind token is malformed")
+    if not valid_bind_token(bind):
+        raise RunnerBindError("Bind token is malformed")
+    return {"pack": pack, "bind": bind, "origin": origin}
diff --git a/engine/auth/store.py b/engine/auth/store.py
index 1f9d2ce..ebfa9ee 100644
--- a/engine/auth/store.py
+++ b/engine/auth/store.py
@@ -815,6 +815,89 @@ def clear_runner_credential(host: str) -> None:
     _kr_delete(_keyring(), host + _RUNNER_SUFFIX)
 
 
+_AUTHORING_LEASE_PREFIX = "openadapt-authoring-lease|"
+_AUTHORING_LEASE_KEYS = frozenset(
+    {
+        "pack",
+        "origin",
+        "lease_secret",
+        "lease_s",
+        "claimed_at",
+        "allowed_sub",
+        "allowed_client_id",
+        "allowed_at",
+    }
+)
+_SHA256_HEX_VALUE = re.compile(r"^[a-f0-9]{64}$")
+
+
+def _authoring_lease_account(pack_id: str) -> str:
+    from engine.auth.runner_bind import valid_pack_id
+
+    if not valid_pack_id(pack_id):
+        raise ValueError("pack id is malformed")
+    return _AUTHORING_LEASE_PREFIX + pack_id
+
+
+def store_authoring_lease(pack_id: str, payload: dict) -> bool:
+    """Persist one authoring mailbox lease in the OS keychain."""
+
+    from engine.auth.runner_bind import (
+        AUTHORING_ORIGIN,
+        valid_lease_secret,
+        valid_pack_id,
+    )
+
+    if (
+        not isinstance(payload, dict)
+        or set(payload) != _AUTHORING_LEASE_KEYS
+        or not valid_pack_id(payload.get("pack"))
+        or payload.get("pack") != pack_id
+        or payload.get("origin") != AUTHORING_ORIGIN
+        or not valid_lease_secret(payload.get("lease_secret"))
+        or not isinstance(payload.get("lease_s"), int)
+        or isinstance(payload.get("lease_s"), bool)
+        or payload.get("lease_s") <= 0
+        or not isinstance(payload.get("claimed_at"), str)
+        or payload.get("claimed_at") == ""
+    ):
+        return False
+    for key in ("allowed_sub", "allowed_client_id", "allowed_at"):
+        value = payload.get(key)
+        if value is None:
+            continue
+        if key == "allowed_at" and isinstance(value, str) and value:
+            continue
+        if key != "allowed_at" and isinstance(value, str) and _SHA256_HEX_VALUE.fullmatch(value):
+            continue
+        return False
+    account = _authoring_lease_account(pack_id)
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+    return _apply_exact(_keyring(), account, encoded)
+
+
+def load_authoring_lease(pack_id: str) -> dict | None:
+    """Load the authoring mailbox lease for ``pack_id``, or None."""
+
+    account = _authoring_lease_account(pack_id)
+    readable, raw = _strict_get(_keyring(), account)
+    if not readable or raw is None:
+        return None
+    try:
+        payload = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return None
+    if not isinstance(payload, dict) or set(payload) != _AUTHORING_LEASE_KEYS:
+        return None
+    return payload
+
+
+def clear_authoring_lease(pack_id: str) -> None:
+    """Delete the authoring mailbox lease for ``pack_id``."""
+
+    _kr_delete(_keyring(), _authoring_lease_account(pack_id))
+
+
 def canonical_host_origin(host: str) -> str:
     """Return a safe web origin for credential binding, or ``""``.
 
diff --git a/engine/authoring_runner.py b/engine/authoring_runner.py
new file mode 100644
index 0000000..256b87e
--- /dev/null
+++ b/engine/authoring_runner.py
@@ -0,0 +1,1220 @@
+"""Outbound authoring mailbox client for ChatGPT.com / Claude.ai drive-once.
+
+Copy poll / lease / TTL / kill-as-command / metadata-callback *shape* from
+:mod:`engine.hosted_runner`. Do not copy org, Stripe, ``oar_``, a 25s poll wait,
+trust-manifest, or journal dispatch. Windows native is COACH_ONLY: this module
+must not spawn ``win_agent`` or call ``parallels_vm.launch_agent``.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import re
+import stat
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+from urllib.parse import quote
+
+import httpx
+from loguru import logger
+
+from engine.auth.runner_bind import (
+    AUTHORING_ORIGIN,
+    parse_runner_uri,
+    valid_lease_secret,
+    valid_pack_id,
+)
+from engine.auth.store import load_authoring_lease, store_authoring_lease
+from engine.config import EngineConfig
+
+API_TIMEOUT_S = 10.0
+DEFAULT_LEASE_S = 900
+POLL_WAIT_S = 0
+LOCAL_POLL_SLEEP_S = 1.0
+NODE_TABLE_LIFETIME_S = 15 * 60
+COMMAND_ENVELOPE_SCHEMA = "openadapt.authoring.command/v1"
+OBSERVE_SCHEMA = "openadapt.authoring.observe/v1"
+CLIENT_DISPLAYS = frozenset({"ChatGPT", "Claude"})
+PAUSE_PROMPT = "Type in the application. Continue here when done."
+ENQUEUE_REQUIRING_ALLOW = frozenset(
+    {
+        "observe",
+        "click",
+        "start_record",
+        "pause_for_input",
+        "stop_record",
+        "compile",
+        "set_coach",
+        "get_coach",
+        "halt",
+    }
+)
+COACH_ONLY_BACKENDS = frozenset({"windows", "rdp", "citrix"})
+UNIQUE_WINDOW_BACKENDS = frozenset({"macos", "linux"})
+PROCESS_NAME_RE = re.compile(r"^[A-Za-z0-9 ._-]{1,64}$")
+SIX_DIGITS_RE = re.compile(r"\d{6,}")
+_SHA256_HEX = re.compile(r"^[a-f0-9]{64}$")
+_SAFE_PARAM = re.compile(r"^[A-Za-z0-9_]{1,40}$")
+_NODE_ID = re.compile(r"^n_[a-f0-9]{8}$")
+_COMMAND_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,200}$")
+FORBIDDEN_RESULT_KEYS = frozenset(
+    {
+        "value",
+        "text",
+        "title",
+        "screenshot",
+        "png",
+        "ocr",
+        "backend_pixels",
+        "pixels",
+        "events",
+        "window_title",
+        "image",
+        "raw",
+        "leaseSecret",
+        "lease_secret",
+        "bind",
+    }
+)
+MAILBOX_ACTIONS = frozenset({"claim", "poll", "callback", "allow"})
+
+
+class AuthoringError(RuntimeError):
+    """A safe, user-facing authoring failure with no secret-bearing text."""
+
+
+class AuthoringCoachOnly(AuthoringError):
+    """This substrate cannot agent-drive in v1."""
+
+
+class AuthoringTransportError(AuthoringError):
+    """The mailbox HTTPS transport did not confirm an operation."""
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _sha256_hex(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _lease_hmac_key(lease_secret: str) -> bytes:
+    """Use the lease secret body as HMAC key material. Do not hash it as a password."""
+
+    if not valid_lease_secret(lease_secret):
+        raise AuthoringError("The authoring mailbox credential is malformed.")
+    return bytes.fromhex(lease_secret[5:])
+
+
+def _sanitize_result(value: Any) -> Any:
+    """Drop titles, values, pixels, and other vendor-forbidden keys."""
+
+    if isinstance(value, dict):
+        return {
+            key: _sanitize_result(child)
+            for key, child in value.items()
+            if key not in FORBIDDEN_RESULT_KEYS
+        }
+    if isinstance(value, list):
+        return [_sanitize_result(item) for item in value]
+    return value
+
+
+def _require_empty_cookies(browser: Any) -> None:
+    cookies_fn = getattr(browser, "cookies", None)
+    if cookies_fn is None:
+        context = getattr(browser, "context", None)
+        cookies_fn = getattr(context, "cookies", None)
+    if not callable(cookies_fn):
+        raise AuthoringError("Playwright Chromium did not start with empty cookies.")
+    cookies = cookies_fn()
+    if cookies:
+        raise AuthoringError("Playwright Chromium did not start with empty cookies.")
+
+
+def launch_empty_playwright_chromium(url: str) -> Any:
+    """Launch a fresh Chromium with empty cookies. Never attach to logged-in Chrome."""
+
+    if not isinstance(url, str) or not url.startswith("https://"):
+        raise AuthoringError("A Playwright job needs a URL typed into Desktop.")
+    try:
+        from playwright.sync_api import sync_playwright
+    except ImportError as exc:
+        raise AuthoringError("Playwright Chromium is unavailable.") from exc
+    playwright = sync_playwright().start()
+    browser = playwright.chromium.launch(headless=False)
+    context = browser.new_context()
+    cookies = context.cookies()
+    if cookies:
+        browser.close()
+        playwright.stop()
+        raise AuthoringError("Playwright Chromium did not start with empty cookies.")
+    page = context.new_page()
+    page.goto(url)
+    return page
+
+
+def _pack_dir(data_dir: Path, pack_id: str) -> Path:
+    return Path(data_dir) / "authoring" / _sha256_hex(pack_id)[:16]
+
+
+def filter_coach_hint(text: object) -> str | None:
+    """Apply the 80-character / no-URL / no-``@`` / no-6-digits coach filter."""
+
+    if not isinstance(text, str):
+        return None
+    collapsed = " ".join(text.split())
+    if not collapsed or len(collapsed) > 80:
+        return None
+    if "://" in collapsed or "@" in collapsed or SIX_DIGITS_RE.search(collapsed):
+        return None
+    return collapsed
+
+
+def _client_display(value: object) -> str:
+    if value in CLIENT_DISPLAYS:
+        return str(value)
+    return "ChatGPT"
+
+
+class NodeTable:
+    """Laptop-only node table. Mode 0600, 15-minute lifetime."""
+
+    def __init__(self, path: Path, hmac_key: bytes) -> None:
+        self._path = Path(path)
+        self._hmac_key = hmac_key
+        self._lock = threading.Lock()
+
+    def clear(self) -> None:
+        with self._lock:
+            try:
+                self._path.unlink()
+            except FileNotFoundError:
+                return
+
+    def mint_node_id(self, provider_runtime_id: str) -> str:
+        digest = hmac.new(
+            self._hmac_key,
+            provider_runtime_id.encode("utf-8"),
+            hashlib.sha256,
+        ).hexdigest()
+        return f"n_{digest[:8]}"
+
+    def replace(self, rows: list[dict[str, Any]]) -> None:
+        payload = {
+            "updated_at": time.time(),
+            "rows": rows,
+        }
+        encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+        self._path.parent.mkdir(parents=True, mode=0o700, exist_ok=True)
+        if not os.name == "nt":
+            os.chmod(self._path.parent, 0o700)
+        flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
+        descriptor = os.open(self._path, flags, 0o600)
+        try:
+            if os.name != "nt":
+                os.fchmod(descriptor, 0o600)
+            os.write(descriptor, encoded.encode("utf-8"))
+        finally:
+            os.close(descriptor)
+
+    def get(self, node_id: str) -> dict[str, Any] | None:
+        with self._lock:
+            try:
+                raw = self._path.read_text(encoding="utf-8")
+                details = self._path.stat()
+            except OSError:
+                return None
+            if os.name != "nt" and stat.S_IMODE(details.st_mode) != 0o600:
+                return None
+            try:
+                payload = json.loads(raw)
+            except json.JSONDecodeError:
+                return None
+            if not isinstance(payload, dict) or not isinstance(payload.get("rows"), list):
+                return None
+            updated = payload.get("updated_at")
+            stale = not isinstance(updated, (int, float)) or (
+                time.time() - updated > NODE_TABLE_LIFETIME_S
+            )
+            if stale:
+                return None
+            for row in payload["rows"]:
+                if isinstance(row, dict) and row.get("node_id") == node_id:
+                    observed = row.get("observed_at")
+                    if (
+                        isinstance(observed, (int, float))
+                        and time.time() - (observed / 1000.0) > NODE_TABLE_LIFETIME_S
+                    ):
+                        return None
+                    return row
+            return None
+
+
+def project_observe(
+    *,
+    backend: str,
+    provider: str,
+    recording: bool,
+    agent_drive: bool,
+    coach_only: bool,
+    process_name: str | None,
+    raw_nodes: list[dict[str, Any]],
+    node_table: NodeTable,
+) -> dict[str, Any]:
+    """PHI-safe observe projection. Fail closed; never a raw fallback."""
+
+    window = {
+        "process_name": (
+            process_name if process_name and PROCESS_NAME_RE.fullmatch(process_name) else None
+        ),
+        "role": "window",
+        "bounds": {"x": 0.0, "y": 0.0, "w": 1.0, "h": 1.0},
+    }
+    if window["process_name"] is None:
+        window.pop("process_name")
+    tree: list[dict[str, Any]] = []
+    rows: list[dict[str, Any]] = []
+    if agent_drive and not coach_only:
+        for raw in raw_nodes:
+            projected, row = _project_node(raw, node_table)
+            if projected is None or row is None:
+                continue
+            tree.append(projected)
+            rows.append(row)
+            if len(tree) >= 200:
+                break
+    node_table.replace(rows)
+    payload: dict[str, Any] = {
+        "schema_version": OBSERVE_SCHEMA,
+        "backend": backend,
+        "provider": provider,
+        "mode": "authoring",
+        "agent_drive": agent_drive and not coach_only,
+        "coach_only": coach_only or not agent_drive,
+        "recording": recording,
+        "window": window,
+        "tree": tree,
+        "truncated": len(raw_nodes) > len(tree),
+        "node_count": len(tree),
+    }
+    encoded = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+    if len(encoded) > 32 * 1024:
+        payload["tree"] = []
+        payload["truncated"] = True
+        payload["node_count"] = 0
+        payload["reason"] = "empty_projection"
+        node_table.replace([])
+    elif not tree:
+        payload["reason"] = "empty_projection"
+    return payload
+
+
+def _project_node(
+    raw: dict[str, Any],
+    node_table: NodeTable,
+) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
+    if not isinstance(raw, dict):
+        return None, None
+    runtime_id = raw.get("provider_runtime_id")
+    pixels = raw.get("backend_pixels")
+    bounds = raw.get("bounds")
+    if not isinstance(runtime_id, str) or not runtime_id:
+        return None, None
+    if not isinstance(pixels, dict) or not isinstance(bounds, dict):
+        return None, None
+    try:
+        pixel_box = {key: int(pixels[key]) for key in ("x", "y", "w", "h")}
+        normalized = {key: float(bounds[key]) for key in ("x", "y", "w", "h")}
+    except (KeyError, TypeError, ValueError):
+        return None, None
+    node_id = node_table.mint_node_id(runtime_id)
+    projected: dict[str, Any] = {
+        "node_id": node_id,
+        "role": str(raw.get("role") or "unknown")[:40],
+        "control_type": str(raw.get("control_type") or "")[:40],
+        "enabled": bool(raw.get("enabled", True)),
+        "focused": bool(raw.get("focused", False)),
+        "bounds": normalized,
+    }
+    automation_id = _project_label(raw.get("automation_id"))
+    if automation_id:
+        projected["automation_id"] = automation_id
+    name = _project_label(raw.get("name"))
+    if name:
+        projected["name"] = name
+    row = {
+        "node_id": node_id,
+        "backend_pixels": pixel_box,
+        "normalized": normalized,
+        "provider_runtime_id": runtime_id,
+        "observed_at": int(time.time() * 1000),
+    }
+    return projected, row
+
+
+def _project_label(value: object) -> str | None:
+    if not isinstance(value, str):
+        return None
+    collapsed = " ".join(value.split())
+    if not collapsed or len(collapsed) > 80:
+        return None
+    if "://" in collapsed or "@" in collapsed or SIX_DIGITS_RE.search(collapsed):
+        return None
+    return collapsed
+
+
+class AuthoringMailboxTransport:
+    """Outbound HTTPS bind/poll/callback. Wait is always 0."""
+
+    def __init__(
+        self,
+        *,
+        origin: str,
+        audit: Any,
+        client: httpx.Client | None = None,
+    ) -> None:
+        if origin != AUTHORING_ORIGIN:
+            raise AuthoringTransportError("The authoring origin is not pinned.")
+        self.origin = origin
+        self._audit = audit
+        if client is not None:
+            base = str(client.base_url).removesuffix("/")
+            if base != origin:
+                raise AuthoringTransportError("The authoring HTTP client differs from its origin.")
+        self._client = client or httpx.Client(
+            base_url=origin,
+            timeout=API_TIMEOUT_S,
+            follow_redirects=False,
+        )
+        self._owns_client = client is None
+
+    def close(self) -> None:
+        if self._owns_client:
+            self._client.close()
+
+    def _path(self, pack_id: str, action: str) -> str:
+        if not valid_pack_id(pack_id) or action not in MAILBOX_ACTIONS:
+            raise AuthoringTransportError("The authoring mailbox path is invalid.")
+        return f"/j/{quote(pack_id, safe='._-')}/runner/{action}"
+
+    def _post(
+        self,
+        path: str,
+        body: dict[str, Any],
+        *,
+        headers: dict[str, str],
+        expected: tuple[int, ...],
+        allow_empty: bool = False,
+    ) -> tuple[int, dict[str, Any] | None]:
+        operation = path.rsplit("/", 1)[-1]
+        self._audit.log(
+            "authoring_request",
+            operation=operation,
+            destination=self.origin,
+            path=path.rsplit("/", 3)[0] + "/runner/" + operation,
+        )
+        try:
+            response = self._client.post(
+                path,
+                json=body,
+                headers=headers,
+                follow_redirects=False,
+            )
+        except (httpx.HTTPError, OSError) as exc:
+            self._audit.log(
+                "authoring_transport_failed",
+                operation=operation,
+                destination=self.origin,
+                error_type=type(exc).__name__,
+            )
+            raise AuthoringTransportError(
+                f"The authoring {operation} request did not complete."
+            ) from exc
+        self._audit.log(
+            "authoring_response",
+            operation=operation,
+            destination=self.origin,
+            status_code=response.status_code,
+        )
+        if response.status_code == 401:
+            raise AuthoringTransportError("The authoring mailbox credential was rejected.")
+        if allow_empty and response.status_code == 204:
+            return 204, None
+        if response.status_code not in expected:
+            raise AuthoringTransportError(
+                f"The authoring {operation} request returned HTTP {response.status_code}."
+            )
+        cache = (response.headers.get("cache-control") or "").strip().lower()
+        if cache != "no-store":
+            raise AuthoringTransportError(
+                f"The authoring {operation} response was not marked no-store."
+            )
+        try:
+            parsed = response.json()
+        except (ValueError, json.JSONDecodeError) as exc:
+            raise AuthoringTransportError(
+                f"The authoring {operation} response was not valid JSON."
+            ) from exc
+        if not isinstance(parsed, dict):
+            raise AuthoringTransportError(f"The authoring {operation} response was not an object.")
+        return response.status_code, parsed
+
+    def claim(self, pack_id: str, bind: str) -> dict[str, Any]:
+        path = self._path(pack_id, "claim")
+        status, body = self._post(
+            path,
+            {"bind": bind},
+            headers={"Content-Type": "application/json"},
+            expected=(201,),
+        )
+        assert status == 201
+        assert body is not None
+        secret = body.get("leaseSecret")
+        lease_s = body.get("lease_s", DEFAULT_LEASE_S)
+        if not valid_lease_secret(secret) or not isinstance(lease_s, int) or lease_s <= 0:
+            raise AuthoringTransportError("The authoring claim response was not a mailbox lease.")
+        return {"leaseSecret": secret, "lease_s": lease_s}
+
+    def poll(self, pack_id: str, lease_secret: str) -> dict[str, Any] | None:
+        if not valid_lease_secret(lease_secret):
+            raise AuthoringTransportError("The authoring mailbox credential is malformed.")
+        path = self._path(pack_id, "poll")
+        _, body = self._post(
+            path,
+            {"wait_seconds": POLL_WAIT_S, "lease_seconds": DEFAULT_LEASE_S},
+            headers={
+                "Authorization": f"Bearer {lease_secret}",
+                "Content-Type": "application/json",
+            },
+            expected=(200,),
+            allow_empty=True,
+        )
+        return body
+
+    def callback(
+        self,
+        pack_id: str,
+        lease_secret: str,
+        payload: dict[str, Any],
+    ) -> None:
+        if not valid_lease_secret(lease_secret):
+            raise AuthoringTransportError("The authoring mailbox credential is malformed.")
+        path = self._path(pack_id, "callback")
+        self._post(
+            path,
+            payload,
+            headers={
+                "Authorization": f"Bearer {lease_secret}",
+                "Content-Type": "application/json",
+            },
+            expected=(200, 202),
+        )
+
+    def allow(self, pack_id: str, lease_secret: str, command_id: str) -> None:
+        if not valid_lease_secret(lease_secret):
+            raise AuthoringTransportError("The authoring mailbox credential is malformed.")
+        if not isinstance(command_id, str) or _COMMAND_ID.fullmatch(command_id) is None:
+            raise AuthoringTransportError("The authoring Allow request is malformed.")
+        path = self._path(pack_id, "allow")
+        self._post(
+            path,
+            {"command_id": command_id},
+            headers={
+                "Authorization": f"Bearer {lease_secret}",
+                "Content-Type": "application/json",
+            },
+            expected=(200, 202),
+        )
+
+
+class AuthoringRunner:
+    """Claim, Allow-per-sub, wait=0 poll, and Flow record_observed session."""
+
+    def __init__(
+        self,
+        config: EngineConfig,
+        *,
+        emit: Callable[[str, dict[str, Any]], None] | None = None,
+        audit: Any | None = None,
+        client: httpx.Client | None = None,
+        sleep: Callable[[float], None] = time.sleep,
+        observe_nodes: Callable[[], list[dict[str, Any]]] | None = None,
+        recorder_factory: Callable[..., Any] | None = None,
+        compile_recording: Callable[..., Any] | None = None,
+        playwright_launcher: Callable[[str], Any] | None = None,
+        text_value_at: Callable[[dict[str, int]], str | None] | None = None,
+        unique_window: Callable[[], dict[str, Any] | None] | None = None,
+    ) -> None:
+        self.config = config
+        self.emit = emit or (lambda _event, _data: None)
+        self.audit = audit
+        self._client = client
+        self._sleep = sleep
+        self._observe_nodes = observe_nodes or (lambda: [])
+        self._recorder_factory = recorder_factory
+        self._compile_recording = compile_recording
+        self._playwright_launcher = playwright_launcher
+        self._text_value_at = text_value_at
+        self._unique_window = unique_window
+        self._transport: AuthoringMailboxTransport | None = None
+        self._lock = threading.RLock()
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._pack: str | None = None
+        self._lease_secret: str | None = None
+        self._allowed_sub: str | None = None
+        self._allowed_client_id: str | None = None
+        self._pending_allow: dict[str, Any] | None = None
+        self._pin: dict[str, Any] = {"backend": "macos", "window_title_unique": False}
+        self._node_table: NodeTable | None = None
+        self._recorder: Any | None = None
+        self._recording = False
+        self._paused = False
+        self._pause_target: dict[str, Any] | None = None
+        self._pause_command_id: str | None = None
+        self._active_command_id: str | None = None
+        self._secret_pause = False
+        self._secret_type_recorded = False
+        self._actuation_started = False
+        self._uncertain = False
+        self._coach_hint: str | None = None
+        self._out_dir: Path | None = None
+        self._playwright: Any | None = None
+
+    def is_bound(self) -> bool:
+        return self._pack is not None and self._lease_secret is not None
+
+    def has_pause(self) -> bool:
+        return self._paused and self._recorder is not None
+
+    def status_dict(self) -> dict[str, Any] | None:
+        if not self.is_bound():
+            return None
+        if self._paused:
+            return {
+                "recording": True,
+                "paused": True,
+                "capture_id": None,
+                "pause_prompt": PAUSE_PROMPT,
+                "controls": {"pause": False, "resume": True, "stop": True},
+            }
+        if self._recording:
+            return {
+                "recording": True,
+                "paused": False,
+                "capture_id": None,
+                "controls": {"pause": False, "resume": False, "stop": True},
+            }
+        return {
+            "recording": False,
+            "paused": False,
+            "capture_id": None,
+            "controls": {"pause": False, "resume": False, "stop": self.is_bound()},
+        }
+
+    def status(self) -> dict[str, Any]:
+        pending = self._pending_allow
+        if pending and self._allowed_sub and pending.get("oauth_sub_sha256") != self._allowed_sub:
+            state = "replace_allow"
+        elif pending:
+            state = "pending_allow"
+        elif self.is_bound():
+            state = "bound"
+        else:
+            state = "idle"
+        return {
+            "status": state,
+            "pack_bound": bool(self._pack),
+            "allowed": bool(self._allowed_sub),
+            "client_display": pending.get("client_display") if pending else None,
+            "coach_only": self._pin.get("backend") in COACH_ONLY_BACKENDS,
+        }
+
+    def pin_target(self, **fields: Any) -> dict[str, Any]:
+        backend = str(fields.get("backend") or self._pin.get("backend") or "macos")
+        unique = fields.get("window_title_unique")
+        macos_app = fields.get("macos_app")
+        macos_title = fields.get("macos_window_title")
+        linux_app = fields.get("linux_app")
+        linux_title = fields.get("linux_window_title")
+        if fields.get("use_frontmost") is True:
+            probe = self._unique_window() if self._unique_window is not None else None
+            if not isinstance(probe, dict) or probe.get("unique") is not True:
+                unique = False
+            else:
+                unique = True
+                backend = str(probe.get("backend") or backend)
+                macos_app = probe.get("process_name") or macos_app
+                macos_title = probe.get("window_title") or macos_title
+                linux_app = probe.get("process_name") or linux_app
+                linux_title = probe.get("window_title") or linux_title
+        if unique is None:
+            unique = False if backend in UNIQUE_WINDOW_BACKENDS else True
+        pin = {
+            "backend": backend,
+            "url": fields.get("url") if backend == "web" else None,
+            "macos_app": macos_app,
+            "macos_window_title": macos_title,
+            "linux_app": linux_app,
+            "linux_window_title": linux_title,
+            "window_title_unique": bool(unique),
+        }
+        self._pin = pin
+        return {
+            "ok": True,
+            "backend": backend,
+            "coach_only": self._coach_only(),
+        }
+
+    def claim_uri(self, uri: str, *, start_loop: bool = True) -> dict[str, Any]:
+        parsed = parse_runner_uri(uri)
+        pack = parsed["pack"]
+        bind = parsed["bind"]
+        origin = parsed["origin"]
+        transport = AuthoringMailboxTransport(
+            origin=origin,
+            audit=self.audit or _NullAudit(),
+            client=self._client,
+        )
+        try:
+            claimed = transport.claim(pack, bind)
+        except AuthoringTransportError:
+            close = getattr(transport, "close", None)
+            if self._client is None and callable(close):
+                close()
+            raise
+        lease_secret = claimed["leaseSecret"]
+        payload = {
+            "pack": pack,
+            "origin": origin,
+            "lease_secret": lease_secret,
+            "lease_s": int(claimed["lease_s"]),
+            "claimed_at": _utc_now(),
+            "allowed_sub": None,
+            "allowed_client_id": None,
+            "allowed_at": None,
+        }
+        if not store_authoring_lease(pack, payload):
+            close = getattr(transport, "close", None)
+            if self._client is None and callable(close):
+                close()
+            raise AuthoringError(
+                "Desktop could not store the authoring lease in the OS keychain."
+            )
+        with self._lock:
+            self._transport = transport
+            self._pack = pack
+            self._lease_secret = lease_secret
+            hmac_key = _lease_hmac_key(lease_secret)
+            self._node_table = NodeTable(
+                _pack_dir(self.config.data_dir, pack) / "nodes.json",
+                hmac_key,
+            )
+            self._out_dir = _pack_dir(self.config.data_dir, pack) / "recording"
+        if self.audit:
+            self.audit.log("authoring_bind_claimed", pack_hash=_sha256_hex(pack)[:16])
+        if start_loop:
+            self.start()
+        self.emit("authoring_state", self.status())
+        return {"bound": True, "origin": origin, "pack_prefix": pack[:2]}
+
+    def start(self) -> None:
+        with self._lock:
+            if self._thread and self._thread.is_alive():
+                return
+            self._stop.clear()
+            self._thread = threading.Thread(target=self._loop, name="authoring-poll", daemon=True)
+            self._thread.start()
+
+    def stop_loop(self) -> None:
+        self._stop.set()
+        thread = self._thread
+        if thread is not None:
+            thread.join(timeout=2.0)
+
+    def _loop(self) -> None:
+        while not self._stop.is_set():
+            try:
+                self.poll_once()
+            except AuthoringError:
+                logger.warning("authoring poll failed")
+            self._sleep(LOCAL_POLL_SLEEP_S)
+
+    def poll_once(self) -> None:
+        transport = self._transport
+        pack = self._pack
+        secret = self._lease_secret
+        if transport is None or pack is None or secret is None:
+            return
+        body = transport.poll(pack, secret)
+        if body is None:
+            return
+        if body.get("halted") is True or (
+            isinstance(body.get("head"), dict) and body["head"].get("halted") is True
+        ):
+            self._halt(unsigned=True, command_id=None)
+            return
+        envelope = body.get("command") if isinstance(body.get("command"), dict) else body
+        if isinstance(envelope, dict) and envelope.get("tool"):
+            self.handle_envelope(envelope)
+
+    def handle_envelope(self, envelope: dict[str, Any]) -> None:
+        tool = envelope.get("tool")
+        command_id = envelope.get("command_id")
+        pack_id = envelope.get("pack_id")
+        if tool not in ENQUEUE_REQUIRING_ALLOW | {"bind_pack"}:
+            self._callback_error(command_id, "unknown_tool")
+            return
+        if not isinstance(command_id, str) or _COMMAND_ID.fullmatch(command_id) is None:
+            return
+        if pack_id != self._pack:
+            self._callback_error(command_id, "pack_mismatch")
+            return
+        if command_id in {self._active_command_id, self._pause_command_id}:
+            return
+        if (
+            tool == "bind_pack"
+            and self._pending_allow
+            and self._pending_allow.get("command_id") == command_id
+        ):
+            return
+        sub = envelope.get("oauth_sub_sha256")
+        if tool == "bind_pack":
+            self._queue_allow(envelope)
+            return
+        if not self._allowed_sub or sub != self._allowed_sub:
+            self._callback_error(command_id, "not_allowed")
+            return
+        args = envelope.get("args") if isinstance(envelope.get("args"), dict) else {}
+        self._active_command_id = command_id
+        try:
+            result = self._dispatch_tool(str(tool), args)
+        except AuthoringCoachOnly:
+            self._active_command_id = None
+            self._callback(
+                {
+                    "command_id": command_id,
+                    "status": "done",
+                    "result": {"error": "COACH_ONLY", "agent_drive": False, "coach_only": True},
+                }
+            )
+            return
+        except AuthoringError as exc:
+            self._active_command_id = None
+            self._callback_error(command_id, str(exc))
+            return
+        if tool == "pause_for_input":
+            self._pause_command_id = command_id
+            return
+        self._active_command_id = None
+        self._callback({"command_id": command_id, "status": "done", "result": result})
+
+    def _queue_allow(self, envelope: dict[str, Any]) -> None:
+        sub = envelope.get("oauth_sub_sha256")
+        client = envelope.get("client_id_sha256")
+        if not isinstance(sub, str) or _SHA256_HEX.fullmatch(sub) is None:
+            self._callback_error(envelope.get("command_id"), "invalid_allow")
+            return
+        if client is not None and (
+            not isinstance(client, str) or _SHA256_HEX.fullmatch(client) is None
+        ):
+            self._callback_error(envelope.get("command_id"), "invalid_allow")
+            return
+        display = _client_display(envelope.get("client_display"))
+        self._pending_allow = {
+            "command_id": envelope.get("command_id"),
+            "oauth_sub_sha256": sub,
+            "client_id_sha256": client,
+            "client_display": display,
+        }
+        status = (
+            "replace_allow"
+            if self._allowed_sub and self._allowed_sub != sub
+            else "pending_allow"
+        )
+        copy = (
+            f"A different {display} account is asking. Allow it to replace the current one?"
+            if status == "replace_allow"
+            else f"Allow {display} to drive this job"
+        )
+        self.emit(
+            "authoring_state",
+            {"status": status, "client_display": display, "prompt": copy},
+        )
+
+    def allow(self, *, replace: bool = False) -> dict[str, Any]:
+        pending = self._pending_allow
+        if pending is None:
+            raise AuthoringError("There is no pending Allow request.")
+        if (
+            self._allowed_sub
+            and self._allowed_sub != pending["oauth_sub_sha256"]
+            and not replace
+        ):
+            return self.status()
+        command_id = pending.get("command_id")
+        display = pending.get("client_display")
+        if (
+            isinstance(command_id, str)
+            and self._transport is not None
+            and self._pack
+            and self._lease_secret
+        ):
+            self._transport.allow(self._pack, self._lease_secret, command_id)
+        granted_at = _utc_now()
+        self._allowed_sub = pending["oauth_sub_sha256"]
+        self._allowed_client_id = pending.get("client_id_sha256")
+        stored = load_authoring_lease(self._pack or "") if self._pack else None
+        if stored is not None:
+            stored["allowed_sub"] = self._allowed_sub
+            stored["allowed_client_id"] = self._allowed_client_id
+            stored["allowed_at"] = granted_at
+            store_authoring_lease(self._pack or "", stored)
+        self._pending_allow = None
+        if self.audit:
+            self.audit.log(
+                "authoring_allowed",
+                pack_hash=_sha256_hex(self._pack or "")[:16],
+                allowed_sub_prefix=(self._allowed_sub or "")[:8],
+                client_display=display,
+            )
+        self.emit("authoring_state", self.status())
+        return {"allowed": True, "client_display": display}
+
+    def deny(self) -> dict[str, Any]:
+        pending = self._pending_allow
+        self._pending_allow = None
+        if pending and isinstance(pending.get("command_id"), str):
+            self._callback_error(pending["command_id"], "denied")
+        self.emit("authoring_state", {"status": "bound"})
+        return {"allowed": False}
+
+    def continue_pause(self) -> dict[str, Any]:
+        if not self.has_pause() or self._pause_target is None or self._recorder is None:
+            return self.status_dict() or {"recording": False, "paused": False}
+        recorder = self._recorder
+        target = self._pause_target
+        if hasattr(recorder, "type_text"):
+            original = recorder.type_text
+
+            def _forbidden(*_args: Any, **_kwargs: Any) -> None:
+                raise AuthoringError("Continue must not type")
+
+            recorder.type_text = _forbidden
+        else:
+            original = None
+        try:
+            text = None
+            if self._text_value_at is not None:
+                text = self._text_value_at(target["backend_pixels"])
+            if target.get("secret"):
+                if text is not None and not text:
+                    return self.status_dict() or {"recording": True, "paused": True}
+                recorder.record_observed(
+                    event={"kind": "type"},
+                    param=target.get("param"),
+                    secret=True,
+                    redact_region=target.get("backend_pixels"),
+                )
+                self._secret_type_recorded = True
+            else:
+                recorder.record_observed(
+                    event={"kind": "type"},
+                    param=target.get("param"),
+                    text=text,
+                )
+        finally:
+            if original is not None:
+                recorder.type_text = original
+        command_id = self._pause_command_id
+        self._paused = False
+        self._pause_target = None
+        self._pause_command_id = None
+        self._active_command_id = None
+        self.emit("status_update", self.status_dict() or {})
+        if self.audit:
+            self.audit.log(
+                "authoring_pause_typed",
+                param=target.get("param"),
+                secret=bool(target.get("secret")),
+            )
+        if isinstance(command_id, str):
+            self._callback(
+                {
+                    "command_id": command_id,
+                    "status": "done",
+                    "result": {"recorded": True, "param": target.get("param")},
+                }
+            )
+        return self.status_dict() or {}
+
+    def operator_stop(self) -> dict[str, Any]:
+        self._halt(unsigned=True, command_id=None)
+        return {"recording": False, "paused": False, "halted": True}
+
+    def _dispatch_tool(self, tool: str, args: dict[str, Any]) -> dict[str, Any]:
+        if tool == "observe":
+            return self._observe()
+        if tool == "start_record":
+            return self._start_record()
+        if tool == "click":
+            return self._click(args)
+        if tool == "pause_for_input":
+            return self._pause_for_input(args)
+        if tool == "stop_record":
+            return self._stop_record()
+        if tool == "compile":
+            return self._compile()
+        if tool == "halt":
+            self._halt(unsigned=False, command_id=None)
+            return {"halted": True}
+        if tool == "set_coach":
+            hint = filter_coach_hint(args.get("hint") or args.get("text"))
+            self._coach_hint = hint
+            return {"ok": hint is not None}
+        if tool == "get_coach":
+            return {"hint": self._coach_hint}
+        raise AuthoringError("unknown_tool")
+
+    def _coach_only(self) -> bool:
+        backend = str(self._pin.get("backend") or "")
+        if backend in COACH_ONLY_BACKENDS:
+            return True
+        if backend in UNIQUE_WINDOW_BACKENDS and not self._pin.get("window_title_unique"):
+            return True
+        return False
+
+    def _observe(self) -> dict[str, Any]:
+        backend = str(self._pin.get("backend") or "macos")
+        coach_only = self._coach_only()
+        agent_drive = not coach_only
+        if self._node_table is None:
+            raise AuthoringError("not_bound")
+        raw = [] if coach_only else list(self._observe_nodes())
+        provider = {
+            "web": "playwright_ax",
+            "macos": "ax",
+            "linux": "atspi",
+        }.get(backend, "none")
+        process_name = None
+        if backend == "web":
+            process_name = "Chromium"
+        elif backend == "macos":
+            process_name = self._pin.get("macos_app")
+        elif backend == "linux":
+            process_name = self._pin.get("linux_app")
+        return project_observe(
+            backend=backend,
+            provider=provider,
+            recording=self._recording,
+            agent_drive=agent_drive,
+            coach_only=coach_only,
+            process_name=process_name if isinstance(process_name, str) else None,
+            raw_nodes=raw,
+            node_table=self._node_table,
+        )
+
+    def _start_record(self) -> dict[str, Any]:
+        if self._coach_only():
+            raise AuthoringCoachOnly("COACH_ONLY")
+        backend = str(self._pin.get("backend") or "macos")
+        if backend == "windows":
+            raise AuthoringCoachOnly("COACH_ONLY")
+        if backend == "web":
+            url = self._pin.get("url")
+            if not isinstance(url, str) or not url.startswith("https://"):
+                raise AuthoringError("A Playwright job needs a URL typed into Desktop.")
+            launcher = self._playwright_launcher or launch_empty_playwright_chromium
+            self._playwright = launcher(url)
+            _require_empty_cookies(self._playwright)
+        factory = self._recorder_factory
+        if factory is None:
+            raise AuthoringError("Authoring recorder is unavailable.")
+        out_dir = self._out_dir or _pack_dir(self.config.data_dir, self._pack or "p.invalidpackid")
+        out_dir.mkdir(parents=True, exist_ok=True)
+        self._recorder = factory(out_dir)
+        self._recording = True
+        self._secret_pause = False
+        self._secret_type_recorded = False
+        self.emit("status_update", self.status_dict() or {})
+        return {"recording": True}
+
+    def _click(self, args: dict[str, Any]) -> dict[str, Any]:
+        if self._coach_only():
+            raise AuthoringCoachOnly("COACH_ONLY")
+        with self._lock:
+            if self._uncertain:
+                raise AuthoringError("RECONCILIATION_REQUIRED")
+        node_id = args.get("node_id")
+        if not isinstance(node_id, str) or _NODE_ID.fullmatch(node_id) is None:
+            raise AuthoringError("stale_node")
+        if self._node_table is None or self._recorder is None:
+            raise AuthoringError("stale_node")
+        row = self._node_table.get(node_id)
+        if row is None:
+            raise AuthoringError("stale_node")
+        pixels = row["backend_pixels"]
+        x = int(pixels["x"] + pixels["w"] / 2)
+        y = int(pixels["y"] + pixels["h"] / 2)
+        with self._lock:
+            self._actuation_started = True
+        try:
+            self._recorder.click(x, y)
+        except Exception:
+            with self._lock:
+                self._uncertain = True
+            raise AuthoringError("RECONCILIATION_REQUIRED") from None
+        finally:
+            with self._lock:
+                self._actuation_started = False
+        return {"clicked": True}
+
+    def _pause_for_input(self, args: dict[str, Any]) -> dict[str, Any]:
+        if self._recorder is None:
+            raise AuthoringError("not_recording")
+        node_id = args.get("node_id")
+        param = args.get("param") or "note"
+        if not isinstance(param, str) or _SAFE_PARAM.fullmatch(param) is None:
+            raise AuthoringError("invalid_param")
+        secret = bool(args.get("secret"))
+        row = None
+        if isinstance(node_id, str) and self._node_table is not None:
+            row = self._node_table.get(node_id)
+        if row is None:
+            raise AuthoringError("stale_node")
+        self._pause_target = {
+            "node_id": row["node_id"],
+            "backend_pixels": row["backend_pixels"],
+            "param": param,
+            "secret": secret,
+        }
+        if secret:
+            self._secret_pause = True
+        self._paused = True
+        self.emit("status_update", self.status_dict() or {})
+        return {"paused": True, "param": param}
+
+    def _stop_record(self) -> dict[str, Any]:
+        recorder = self._recorder
+        if recorder is None:
+            return {"recording": False}
+        finish = getattr(recorder, "finish", None)
+        if callable(finish):
+            finish()
+        self._recording = False
+        self._paused = False
+        self.emit("status_update", self.status_dict() or {})
+        return {"recording": False}
+
+    def _compile(self) -> dict[str, Any]:
+        if self._secret_pause and not self._secret_type_recorded:
+            if self.audit:
+                self.audit.log("authoring_compile_refused_missing_type")
+            raise AuthoringError("secret_type_missing")
+        compile_recording = self._compile_recording
+        workflow_id = "wf_local"
+        if callable(compile_recording) and self._out_dir is not None:
+            workflow = compile_recording(self._out_dir)
+            workflow_id = str(
+                getattr(workflow, "id", None) or getattr(workflow, "workflow_id", workflow_id)
+            )
+        if self._node_table is not None:
+            self._node_table.clear()
+        return {
+            "status": "needs_human_admit",
+            "workflow_id": workflow_id,
+            "recording_retained": True,
+        }
+
+    def _halt(self, *, unsigned: bool, command_id: str | None) -> None:
+        with self._lock:
+            if self._actuation_started:
+                self._uncertain = True
+                uncertain = True
+            else:
+                uncertain = False
+                self._recording = False
+                self._paused = False
+                self._recorder = None
+                self._pause_target = None
+                self._pause_command_id = None
+                self._active_command_id = None
+        if uncertain:
+            self._callback(
+                {
+                    "command_id": command_id,
+                    "status": "error",
+                    "result": {"error": "RECONCILIATION_REQUIRED"},
+                }
+            )
+            return
+        if self._node_table is not None:
+            self._node_table.clear()
+        if unsigned and self._pack and self._lease_secret and self._transport:
+            self._transport.callback(
+                self._pack,
+                self._lease_secret,
+                {"halted": True, "status": "halted"},
+            )
+        self.emit("status_update", {"recording": False, "paused": False, "halted": True})
+
+    def _callback(self, payload: dict[str, Any]) -> None:
+        if self._transport is None or self._pack is None or self._lease_secret is None:
+            return
+        closed = {
+            key: payload[key]
+            for key in ("command_id", "status", "result", "halted")
+            if key in payload
+        }
+        if "result" in closed:
+            closed["result"] = _sanitize_result(closed["result"])
+        self._transport.callback(self._pack, self._lease_secret, closed)
+
+    def _callback_error(self, command_id: object, error: str) -> None:
+        if not isinstance(command_id, str):
+            return
+        self._callback(
+            {
+                "command_id": command_id,
+                "status": "error",
+                "result": {"error": error},
+            }
+        )
+
+
+class _NullAudit:
+    def log(self, *_args: Any, **_kwargs: Any) -> None:
+        return
+
+
+def restore_authoring_runner(
+    config: EngineConfig, pack_id: str, **kwargs: Any
+) -> AuthoringRunner | None:
+    """Rebuild an authoring runner from a stored lease without re-claiming."""
+
+    if not valid_pack_id(pack_id):
+        return None
+    stored = load_authoring_lease(pack_id)
+    if stored is None:
+        return None
+    runner = AuthoringRunner(config, **kwargs)
+    runner._pack = stored["pack"]
+    runner._lease_secret = stored["lease_secret"]
+    runner._allowed_sub = stored.get("allowed_sub")
+    runner._allowed_client_id = stored.get("allowed_client_id")
+    hmac_key = _lease_hmac_key(stored["lease_secret"])
+    runner._node_table = NodeTable(_pack_dir(config.data_dir, pack_id) / "nodes.json", hmac_key)
+    runner._transport = AuthoringMailboxTransport(
+        origin=stored["origin"],
+        audit=kwargs.get("audit") or _NullAudit(),
+        client=kwargs.get("client"),
+    )
+    return runner
diff --git a/engine/dispatch.py b/engine/dispatch.py
index cbec8c3..fc91fa9 100644
--- a/engine/dispatch.py
+++ b/engine/dispatch.py
@@ -121,6 +121,7 @@ def __init__(
         flow_bridge: Any = None,
         runner: Any = None,
         portal: Any = None,
+        authoring: Any = None,
     ) -> None:
         self.config = config
         self._db = db
@@ -135,6 +136,7 @@ def __init__(
         # The mobile decision portal is likewise built on first use so the
         # engine never binds a socket or spawns a console it was not asked for.
         self.portal = portal
+        self.authoring = authoring
 
     @property
     def db(self) -> Any:
@@ -284,6 +286,11 @@ def _register(self) -> None:
             "login_browser": self.login_browser,
             "login_paste": self.login_paste,
             "connect_uri": self.connect_uri,
+            "claim_runner_uri": self.claim_runner_uri,
+            "authoring_allow": self.authoring_allow,
+            "authoring_deny": self.authoring_deny,
+            "authoring_status": self.authoring_status,
+            "authoring_pin_target": self.authoring_pin_target,
             "logout": self.logout,
             "get_auth_status": self.get_auth_status,
             # config / settings
@@ -567,6 +574,9 @@ def _remember_first_workflow_recording(
 
     def stop_recording(self, **params: Any) -> dict:
         """Stop the active recording, retain it, and compile it automatically."""
+        authoring = self.services.authoring
+        if authoring is not None and authoring.is_bound():
+            return authoring.operator_stop()
         controller = self.services.controller
         active = self._flow_recording
         if active is not None:
@@ -654,11 +664,19 @@ def pause_recording(self, **params: Any) -> dict:
         return self.get_status()
 
     def resume_recording(self, **params: Any) -> dict:
-        """Resume is not supported (stop/start instead); report current status."""
+        """Overlay Resume during an authoring pause records the typed field."""
+        authoring = self.services.authoring
+        if authoring is not None and authoring.has_pause():
+            return authoring.continue_pause()
         return self.get_status()
 
     def get_status(self, **params: Any) -> dict:
         """Return the current :class:`EngineStatus`-shaped recording status."""
+        authoring = self.services.authoring
+        if authoring is not None:
+            status = authoring.status_dict()
+            if status is not None and (status.get("recording") or status.get("halted")):
+                return status
         return self._status_dict(self.services.controller)
 
     def _status_dict(self, controller: Any) -> dict:
@@ -3261,6 +3279,42 @@ def connect_uri(self, **params: Any) -> dict:
         )
         return result
 
+    def _authoring_service(self) -> Any:
+        if self.services.authoring is None:
+            from engine.authoring_runner import AuthoringRunner
+
+            self.services.authoring = AuthoringRunner(
+                self.config,
+                emit=self.emit,
+                audit=self.services.audit,
+            )
+        return self.services.authoring
+
+    def claim_runner_uri(self, **params: Any) -> dict:
+        """Claim one validated ``openadapt://runner`` bind URI."""
+        uri = params.get("uri")
+        if not isinstance(uri, str):
+            raise ValueError("uri is required")
+        result = self._authoring_service().claim_uri(uri)
+        self.emit("authoring_state", self._authoring_service().status())
+        return result
+
+    def authoring_allow(self, **params: Any) -> dict:
+        """Allow the pending connector ``sub`` to drive this job."""
+        return self._authoring_service().allow(replace=params.get("replace") is True)
+
+    def authoring_deny(self, **params: Any) -> dict:
+        """Refuse the pending Allow request."""
+        return self._authoring_service().deny()
+
+    def authoring_status(self, **params: Any) -> dict:
+        """Return the local authoring bind / Allow state."""
+        return self._authoring_service().status()
+
+    def authoring_pin_target(self, **params: Any) -> dict:
+        """Pin the local authoring backend. Titles never go to MCP."""
+        return self._authoring_service().pin_target(**params)
+
     def logout(self, **params: Any) -> dict:
         """Clear only the credential for the selected safe hosted origin."""
         from engine.auth.store import (
diff --git a/scripts/vendor-design-tokens.mjs b/scripts/vendor-design-tokens.mjs
index ddfea31..b8698d2 100755
--- a/scripts/vendor-design-tokens.mjs
+++ b/scripts/vendor-design-tokens.mjs
@@ -25,10 +25,29 @@ const write = process.argv.includes('--write');
 
 const sha256 = (bytes) => crypto.createHash('sha256').update(bytes).digest('hex');
 
-async function fetchCanonical(url) {
-  const response = await fetch(url, { headers: { accept: 'text/plain' } });
+async function fetchCanonical(entry) {
+  // openadapt-web is private, so raw.githubusercontent.com 404s. CI already
+  // passes GITHUB_TOKEN; use the Contents API the same way --write does.
+  const token = process.env.GITHUB_TOKEN;
+  if (token) {
+    const url =
+      `https://api.github.com/repos/${provenance.canonical_repository}` +
+      `/contents/${entry.canonical_path}` +
+      `?ref=${encodeURIComponent(provenance.canonical_branch)}`;
+    const response = await fetch(url, {
+      headers: {
+        accept: 'application/vnd.github.raw',
+        authorization: `Bearer ${token}`,
+      },
+    });
+    if (!response.ok) {
+      throw new Error(`GET ${url} -> HTTP ${response.status}`);
+    }
+    return Buffer.from(await response.arrayBuffer());
+  }
+  const response = await fetch(entry.raw_url, { headers: { accept: 'text/plain' } });
   if (!response.ok) {
-    throw new Error(`GET ${url} -> HTTP ${response.status}`);
+    throw new Error(`GET ${entry.raw_url} -> HTTP ${response.status}`);
   }
   return Buffer.from(await response.arrayBuffer());
 }
@@ -59,7 +78,7 @@ for (const [name, entry] of Object.entries(provenance.files)) {
     );
   }
 
-  const canonical = await fetchCanonical(entry.raw_url);
+  const canonical = await fetchCanonical(entry);
   const canonicalSha = sha256(canonical);
 
   if (write) {
diff --git a/src-tauri/src/pairing.rs b/src-tauri/src/pairing.rs
index 12d22fe..c83557c 100644
--- a/src-tauri/src/pairing.rs
+++ b/src-tauri/src/pairing.rs
@@ -1,9 +1,14 @@
-//! Strict operating-system deep-link boundary for one-click Cloud pairing.
+//! Strict operating-system deep-link boundary for Cloud pairing and authoring.
 //!
 //! The protocol handler never opens a URL or constructs a process command. It
-//! accepts one fixed `openadapt://connect` URI, validates every field, and
-//! forwards the original URI as one JSON string to the fixed Python
-//! `connect_uri` sidecar action.
+//! accepts two fixed schemes, validates every field, and forwards the original
+//! URI as one JSON string to one sidecar action:
+//!
+//! - `openadapt://connect` → `connect_uri`
+//! - `openadapt://runner` → `claim_runner_uri`
+//!
+//! Connect is not widened to accept runner fields, and runner is not widened
+//! to accept connect fields.
 
 use std::collections::{HashMap, HashSet};
 use std::error::Error;
@@ -18,6 +23,7 @@ use url::Url;
 use crate::sidecar::SidecarInner;
 
 const MANAGED_HOST: &str = "app.openadapt.ai";
+const AUTHORING_ORIGIN: &str = "https://openadapt.ai";
 const MAX_URI_BYTES: usize = 2048;
 const MAX_RECENT_LINKS: usize = 64;
 
@@ -72,7 +78,12 @@ fn route_urls(
     let action = match single_action(urls) {
         Ok(action) => action,
         Err(error) => {
-            emit_state(&app, "error", Some(error));
+            let event = if urls.iter().any(|url| url.host_str() == Some("runner")) {
+                "engine://authoring_state"
+            } else {
+                "engine://pairing_state"
+            };
+            emit_state(&app, event, "error", Some(error));
             return;
         }
     };
@@ -92,32 +103,48 @@ fn route_urls(
         handled.insert(fingerprint);
     }
 
-    emit_state(&app, "connecting", None);
+    let event = status_event(action.command);
+    let connecting = if action.command == "claim_runner_uri" {
+        "claiming"
+    } else {
+        "connecting"
+    };
+    emit_state(&app, event, connecting, None);
     tauri::async_runtime::spawn(async move {
         let result = engine
             .send_command(action.command, json!({ "uri": action.uri }))
             .await;
         match result {
             Ok(data) => {
-                let _ = app.emit(
-                    "engine://pairing_state",
-                    json!({ "status": "connected", "data": data }),
-                );
+                let connected = if action.command == "claim_runner_uri" {
+                    "bound"
+                } else {
+                    "connected"
+                };
+                let _ = app.emit(event, json!({ "status": connected, "data": data }));
             }
             Err(error) => {
                 eprintln!("[pairing] connection failed: {error}");
-                emit_state(&app, "error", Some(&error));
+                emit_state(&app, event, "error", Some(&error));
             }
         }
     });
 }
 
-fn emit_state(app: &AppHandle, status: &str, error: Option<&str>) {
+fn status_event(command: &str) -> &'static str {
+    if command == "claim_runner_uri" {
+        "engine://authoring_state"
+    } else {
+        "engine://pairing_state"
+    }
+}
+
+fn emit_state(app: &AppHandle, event: &str, status: &str, error: Option<&str>) {
     let payload = match error {
         Some(error) => json!({ "status": status, "error": error }),
         None => json!({ "status": status }),
     };
-    let _ = app.emit("engine://pairing_state", payload);
+    let _ = app.emit(event, payload);
 }
 
 fn fingerprint(uri: &str) -> u64 {
@@ -135,18 +162,31 @@ fn single_action(urls: &[Url]) -> Result<PairingAction, &'static str> {
 
 fn action_for_url(url: &Url) -> Result<PairingAction, &'static str> {
     let uri = url.as_str();
+    let runner = url.host_str() == Some("runner");
+    let invalid = if runner {
+        "Invalid OpenAdapt runner link"
+    } else {
+        "Invalid OpenAdapt connect link"
+    };
     if uri.len() > MAX_URI_BYTES
         || url.scheme() != "openadapt"
-        || url.host_str() != Some("connect")
         || !url.username().is_empty()
         || url.password().is_some()
         || url.port().is_some()
         || !matches!(url.path(), "" | "/")
         || url.fragment().is_some()
     {
-        return Err("Invalid OpenAdapt connect link");
+        return Err(invalid);
     }
 
+    match url.host_str() {
+        Some("connect") => connect_action(url, uri),
+        Some("runner") => runner_action(url, uri),
+        _ => Err("Invalid OpenAdapt connect link"),
+    }
+}
+
+fn connect_action(url: &Url, uri: &str) -> Result<PairingAction, &'static str> {
     let mut fields: HashMap<String, String> = HashMap::new();
     for (key, value) in url.query_pairs() {
         if !matches!(key.as_ref(), "pairing" | "host" | "destination_kind")
@@ -175,12 +215,78 @@ fn action_for_url(url: &Url) -> Result<PairingAction, &'static str> {
     })
 }
 
+fn runner_action(url: &Url, uri: &str) -> Result<PairingAction, &'static str> {
+    let mut fields: HashMap<String, String> = HashMap::new();
+    for (key, value) in url.query_pairs() {
+        if !matches!(key.as_ref(), "pack" | "bind" | "origin")
+            || fields
+                .insert(key.into_owned(), value.into_owned())
+                .is_some()
+        {
+            return Err("Runner link contains unknown or duplicate fields");
+        }
+    }
+
+    let pack = fields
+        .get("pack")
+        .ok_or("Runner link is missing pack, bind, or origin")?;
+    let bind = fields
+        .get("bind")
+        .ok_or("Runner link is missing pack, bind, or origin")?;
+    let origin = fields
+        .get("origin")
+        .ok_or("Runner link is missing pack, bind, or origin")?;
+    if !valid_pack_id(pack) {
+        return Err("Pack id is malformed");
+    }
+    if !valid_bind_token(bind) {
+        return Err("Bind token is malformed");
+    }
+    if origin != AUTHORING_ORIGIN {
+        return Err("Runner link does not name the OpenAdapt authoring origin");
+    }
+
+    Ok(PairingAction {
+        command: "claim_runner_uri",
+        uri: uri.to_owned(),
+    })
+}
+
 fn valid_pairing_secret(value: &str) -> bool {
-    value.len() == 47
-        && value.starts_with("oap_")
-        && value[4..]
-            .bytes()
-            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
+    value.len() == 47 && value.starts_with("oap_") && unreserved_body(&value[4..])
+}
+
+fn valid_bind_token(value: &str) -> bool {
+    if value.starts_with("oar_") || value.starts_with("oap_") {
+        return false;
+    }
+    // Cloud runner bodies are 64 hex. That encoding is never a bind token.
+    if value.len() == 68 && value.starts_with("oab_") && hex_body(&value[4..]) {
+        return false;
+    }
+    value.len() == 47 && value.starts_with("oab_") && unreserved_body(&value[4..])
+}
+
+fn hex_body(value: &str) -> bool {
+    value
+        .bytes()
+        .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
+}
+
+fn valid_pack_id(value: &str) -> bool {
+    if let Some(body) = value.strip_prefix("p.") {
+        return body.len() == 12 && unreserved_body(body);
+    }
+    if let Some(body) = value.strip_prefix("v1.") {
+        return (32..=2000).contains(&body.len()) && unreserved_body(body);
+    }
+    false
+}
+
+fn unreserved_body(value: &str) -> bool {
+    value
+        .bytes()
+        .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
 }
 
 fn validate_destination(host: &str, destination_kind: Option<&str>) -> Result<(), &'static str> {
@@ -230,6 +336,13 @@ mod tests {
         Url::parse(raw).unwrap()
     }
 
+    const BIND: &str = "oab_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+    const PACK: &str = "p.abcdefghijkl";
+
+    fn runner_uri() -> String {
+        format!("openadapt://runner?pack={PACK}&bind={BIND}&origin=https%3A%2F%2Fopenadapt.ai")
+    }
+
     #[test]
     fn accepts_only_fixed_connect_action() {
         let url = parse(&format!(
@@ -244,8 +357,30 @@ mod tests {
             format!("https://connect?pairing={SECRET}&host=https://app.openadapt.ai"),
             format!("openadapt://connect/run?pairing={SECRET}&host=https://app.openadapt.ai"),
             format!("openadapt://connect?pairing={SECRET}&host=https://app.openadapt.ai#x"),
+            runner_uri(),
+            format!("openadapt://connect?pack={PACK}&bind={BIND}&origin=https://openadapt.ai"),
         ] {
-            assert!(action_for_url(&parse(&raw)).is_err());
+            assert!(action_for_url(&parse(&raw)).is_err(), "{raw}");
+        }
+    }
+
+    #[test]
+    fn accepts_only_fixed_runner_action() {
+        let url = parse(&runner_uri());
+        let action = action_for_url(&url).unwrap();
+        assert_eq!(action.command, "claim_runner_uri");
+        assert_eq!(action.uri, url.as_str());
+
+        for raw in [
+            format!("openadapt://run?pack={PACK}&bind={BIND}&origin=https://openadapt.ai"),
+            format!(
+                "openadapt://connect/runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai"
+            ),
+            format!("{}#x", runner_uri()),
+            format!("openadapt://connect?pairing={SECRET}&host=https://app.openadapt.ai"),
+            format!("openadapt://runner?pairing={SECRET}&host=https://app.openadapt.ai"),
+        ] {
+            assert!(action_for_url(&parse(&raw)).is_err(), "{raw}");
         }
     }
 
@@ -265,6 +400,38 @@ mod tests {
         }
     }
 
+    #[test]
+    fn runner_rejects_malformed_duplicate_unknown_and_foreign_tokens() {
+        for raw in [
+            format!("openadapt://runner?pack=short&bind={BIND}&origin=https://openadapt.ai"),
+            format!("openadapt://runner?pack={PACK}&bind={BIND}"),
+            format!(
+                "openadapt://runner?pack={PACK}&bind={BIND}&bind={BIND}&origin=https://openadapt.ai"
+            ),
+            format!(
+                "openadapt://runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai&command=run"
+            ),
+            format!(
+                "openadapt://runner?pack={PACK}&bind=oar_{}&origin=https://openadapt.ai",
+                "a".repeat(64)
+            ),
+            format!("openadapt://runner?pack={PACK}&bind={SECRET}&origin=https://openadapt.ai"),
+            format!(
+                "openadapt://runner?pack={PACK}&bind=oab_{}&origin=https://openadapt.ai",
+                "a".repeat(64)
+            ),
+            format!(
+                "openadapt://runner?pack={PACK}&bind=oals_{}&origin=https://openadapt.ai",
+                "A".repeat(43)
+            ),
+            format!(
+                "openadapt://runner?pack={PACK}&bind={BIND}&origin=https://preview.openadapt.ai"
+            ),
+        ] {
+            assert!(action_for_url(&parse(&raw)).is_err(), "{raw}");
+        }
+    }
+
     #[test]
     fn argument_shaped_data_never_changes_the_fixed_action() {
         let encoded_argument = "%2D%2Dhost%3Dhttps%3A%2F%2Fevil.example";
diff --git a/src/App.firstWorkflowNavigation.test.tsx b/src/App.firstWorkflowNavigation.test.tsx
index d97f51d..700299f 100644
--- a/src/App.firstWorkflowNavigation.test.tsx
+++ b/src/App.firstWorkflowNavigation.test.tsx
@@ -68,6 +68,16 @@ import App from "./App";
 import { CMD, engineTry, EVT } from "./lib/engine";
 import type { FirstWorkflowState } from "./lib/types";
 
+function bootDefaults(command: string): unknown {
+  if (command === CMD.GET_AUTH_STATUS) return { authenticated: true };
+  if (command === CMD.GET_NEEDS_ATTENTION) {
+    return { count: 0, open_halts: 0, failed_runs: 0 };
+  }
+  if (command === CMD.GET_SYNC_STATE) return { state: "synced", queued: 0 };
+  if (command === CMD.AUTHORING_STATUS) return { status: "idle" };
+  return undefined;
+}
+
 afterEach(() => {
   cleanup();
   appEventMocks.handlers.clear();
@@ -76,21 +86,18 @@ afterEach(() => {
 
 it("stays in onboarding when its durable stage cannot be saved", async () => {
   let stageAttempts = 0;
-  vi.mocked(engineTry).mockImplementation(async (command) => {
-    if (command === CMD.GET_AUTH_STATUS) return { authenticated: true };
+  vi.mocked(engineTry).mockImplementation(async (command, _params, fallback) => {
+    const boot = bootDefaults(command);
+    if (boot !== undefined) return boot;
     if (command === CMD.GET_WORKFLOWS) return [];
     if (command === CMD.GET_FIRST_WORKFLOW_STATE) {
       return { ok: true, state: null };
     }
-    if (command === CMD.GET_NEEDS_ATTENTION) {
-      return { count: 0, open_halts: 0, failed_runs: 0 };
-    }
-    if (command === CMD.GET_SYNC_STATE) return { state: "synced", queued: 0 };
     if (command === CMD.SET_FIRST_WORKFLOW_STAGE) {
       stageAttempts += 1;
       return { ok: stageAttempts > 1 };
     }
-    return null;
+    return fallback ?? null;
   });
 
   render(<App />);
@@ -121,23 +128,20 @@ it("returns the pre-run action review to the supervised replay", async () => {
     task: "Read one test record",
     updated_at: "2026-08-27T00:00:00Z",
   };
-  vi.mocked(engineTry).mockImplementation(async (command, params) => {
-    if (command === CMD.GET_AUTH_STATUS) return { authenticated: true };
+  vi.mocked(engineTry).mockImplementation(async (command, params, fallback) => {
+    const boot = bootDefaults(command);
+    if (boot !== undefined) return boot;
     if (command === CMD.GET_WORKFLOWS) {
       return [{ id: "workflow-1", name: "Test", steps: 1 }];
     }
     if (command === CMD.GET_FIRST_WORKFLOW_STATE) {
       return { ok: true, state };
     }
-    if (command === CMD.GET_NEEDS_ATTENTION) {
-      return { count: 0, open_halts: 0, failed_runs: 0 };
-    }
-    if (command === CMD.GET_SYNC_STATE) return { state: "synced", queued: 0 };
     if (command === CMD.SET_FIRST_WORKFLOW_STAGE) {
       expect(params).toEqual({ stage: "review", workflow_id: "workflow-1" });
       return { ok: true, state: { ...state, stage: "review" } };
     }
-    return null;
+    return fallback ?? null;
   });
 
   render(<App />);
@@ -157,19 +161,16 @@ it("keeps the first-workflow context after a library visit", async () => {
     task: "Read one test record",
     updated_at: "2026-08-27T00:00:00Z",
   };
-  vi.mocked(engineTry).mockImplementation(async (command) => {
-    if (command === CMD.GET_AUTH_STATUS) return { authenticated: true };
+  vi.mocked(engineTry).mockImplementation(async (command, _params, fallback) => {
+    const boot = bootDefaults(command);
+    if (boot !== undefined) return boot;
     if (command === CMD.GET_WORKFLOWS) {
       return [{ id: "workflow-1", name: "Test workflow", steps: 1 }];
     }
     if (command === CMD.GET_FIRST_WORKFLOW_STATE) {
       return { ok: true, state };
     }
-    if (command === CMD.GET_NEEDS_ATTENTION) {
-      return { count: 0, open_halts: 0, failed_runs: 0 };
-    }
-    if (command === CMD.GET_SYNC_STATE) return { state: "synced", queued: 0 };
-    return null;
+    return fallback ?? null;
   });
 
   render(<App />);
@@ -192,19 +193,16 @@ it("keeps navigation locked while the supervised replay is running", async () =>
     task: "Read one test record",
     updated_at: "2026-08-27T00:00:00Z",
   };
-  vi.mocked(engineTry).mockImplementation(async (command) => {
-    if (command === CMD.GET_AUTH_STATUS) return { authenticated: true };
+  vi.mocked(engineTry).mockImplementation(async (command, _params, fallback) => {
+    const boot = bootDefaults(command);
+    if (boot !== undefined) return boot;
     if (command === CMD.GET_WORKFLOWS) {
       return [{ id: "workflow-1", name: "Test workflow", steps: 1 }];
     }
     if (command === CMD.GET_FIRST_WORKFLOW_STATE) {
       return { ok: true, state };
     }
-    if (command === CMD.GET_NEEDS_ATTENTION) {
-      return { count: 0, open_halts: 0, failed_runs: 0 };
-    }
-    if (command === CMD.GET_SYNC_STATE) return { state: "synced", queued: 0 };
-    return null;
+    return fallback ?? null;
   });
 
   render(<App />);
@@ -229,17 +227,14 @@ it("keeps navigation locked from recording stop through review initialization",
     task: "Read one test record",
     updated_at: "2026-08-27T00:00:00Z",
   };
-  vi.mocked(engineTry).mockImplementation(async (command) => {
-    if (command === CMD.GET_AUTH_STATUS) return { authenticated: true };
+  vi.mocked(engineTry).mockImplementation(async (command, _params, fallback) => {
+    const boot = bootDefaults(command);
+    if (boot !== undefined) return boot;
     if (command === CMD.GET_WORKFLOWS) return [];
     if (command === CMD.GET_FIRST_WORKFLOW_STATE) {
       return { ok: true, state: firstState };
     }
-    if (command === CMD.GET_NEEDS_ATTENTION) {
-      return { count: 0, open_halts: 0, failed_runs: 0 };
-    }
-    if (command === CMD.GET_SYNC_STATE) return { state: "synced", queued: 0 };
-    return null;
+    return fallback ?? null;
   });
 
   render(<App />);
@@ -275,3 +270,25 @@ it("keeps navigation locked from recording stop through review initialization",
     expect((workflows as HTMLButtonElement).disabled).toBe(false),
   );
 });
+
+it("leaves loading when authoring status is missing", async () => {
+  vi.mocked(engineTry).mockImplementation(async (command) => {
+    if (command === CMD.GET_AUTH_STATUS) return { authenticated: true };
+    if (command === CMD.GET_WORKFLOWS) return [];
+    if (command === CMD.GET_FIRST_WORKFLOW_STATE) {
+      return { ok: true, state: null };
+    }
+    if (command === CMD.GET_NEEDS_ATTENTION) {
+      return { count: 0, open_halts: 0, failed_runs: 0 };
+    }
+    if (command === CMD.GET_SYNC_STATE) return { state: "synced", queued: 0 };
+    if (command === CMD.AUTHORING_STATUS) return null;
+    return null;
+  });
+
+  render(<App />);
+  expect(
+    await screen.findByRole("button", { name: "Start first workflow" }),
+  ).toBeTruthy();
+  expect(screen.queryByText("Loading…")).toBeNull();
+});
diff --git a/src/App.tsx b/src/App.tsx
index 398bf8d..8668bf8 100644
--- a/src/App.tsx
+++ b/src/App.tsx
@@ -170,6 +170,15 @@ export default function App() {
   const [sync, setSync] = useState<SyncState>({ state: "synced", queued: 0 });
   const [breaks, setBreaks] = useState(0);
   const [pairing, setPairing] = useState<PairingState | null>(null);
+  const [authoring, setAuthoring] = useState<{
+    status: string;
+    client_display?: string;
+    prompt?: string;
+    error?: string;
+    allowed?: boolean;
+    coach_only?: boolean;
+  } | null>(null);
+  const [authoringUrl, setAuthoringUrl] = useState("");
   const [firstRunPersistencePending, setFirstRunPersistencePending] =
     useState(false);
   const [firstWorkflowRunning, setFirstWorkflowRunning] = useState(false);
@@ -222,36 +231,52 @@ export default function App() {
   }
 
   // Bootstrap: auth status, sidecar liveness, and the status channels.
+  // Authoring bind is optional — a missing status must not leave the shell
+  // on Loading, or first-workflow navigation never mounts.
   useEffect(() => {
     (async () => {
-      setEngineUp(await sidecarRunning());
-      const a = await engineTry<AuthStatus>(
-        CMD.GET_AUTH_STATUS,
-        {},
-        { authenticated: false },
-      );
-      setAuth(a);
-      const [wf, firstWorkflow] = await Promise.all([
-        engineTry<Workflow[]>(CMD.GET_WORKFLOWS, {}, []),
-        engineTry<FirstWorkflowStateResponse>(
-          CMD.GET_FIRST_WORKFLOW_STATE,
+      try {
+        setEngineUp(await sidecarRunning());
+        const a = await engineTry<AuthStatus>(
+          CMD.GET_AUTH_STATUS,
           {},
-          { ok: true, state: null },
-        ),
-      ]);
-      const resumedRoute = routeForFirstWorkflow(firstWorkflow.state);
-      setFirstWorkflowState(firstWorkflow.state);
-      setOnboarded(wf.length > 0 || resumedRoute !== null);
-      if (resumedRoute) setRoute(resumedRoute);
-      const na = await engineTry<NeedsAttention>(
-        CMD.GET_NEEDS_ATTENTION,
-        {},
-        { count: 0, open_halts: 0, failed_runs: 0 },
-      );
-      setBreaks(na.count);
-      const ss = await engineTry<SyncState>(CMD.GET_SYNC_STATE, {}, sync);
-      setSync(ss);
-      setCheckedAuth(true);
+          { authenticated: false },
+        );
+        setAuth(a);
+        const [wf, firstWorkflow] = await Promise.all([
+          engineTry<Workflow[]>(CMD.GET_WORKFLOWS, {}, []),
+          engineTry<FirstWorkflowStateResponse>(
+            CMD.GET_FIRST_WORKFLOW_STATE,
+            {},
+            { ok: true, state: null },
+          ),
+        ]);
+        const resumedRoute = routeForFirstWorkflow(firstWorkflow.state);
+        setFirstWorkflowState(firstWorkflow.state);
+        setOnboarded(wf.length > 0 || resumedRoute !== null);
+        if (resumedRoute) setRoute(resumedRoute);
+        const na = await engineTry<NeedsAttention>(
+          CMD.GET_NEEDS_ATTENTION,
+          {},
+          { count: 0, open_halts: 0, failed_runs: 0 },
+        );
+        setBreaks(na.count);
+        const ss = await engineTry<SyncState>(CMD.GET_SYNC_STATE, {}, sync);
+        setSync(ss);
+        const authoringStatus = await engineTry<{
+          status: string;
+          client_display?: string;
+          prompt?: string;
+          error?: string;
+          allowed?: boolean;
+          coach_only?: boolean;
+        } | null>(CMD.AUTHORING_STATUS, {}, { status: "idle" });
+        if (authoringStatus?.status && authoringStatus.status !== "idle") {
+          setAuthoring(authoringStatus);
+        }
+      } finally {
+        setCheckedAuth(true);
+      }
     })();
 
     const unsubs = [
@@ -312,6 +337,19 @@ export default function App() {
           });
         }
       }),
+      onEngineEvent(
+        EVT.AUTHORING_STATE,
+        (state: {
+          status: string;
+          client_display?: string;
+          prompt?: string;
+          error?: string;
+          allowed?: boolean;
+          coach_only?: boolean;
+        }) => {
+          setAuthoring(state);
+        },
+      ),
     ];
     return () => unsubs.forEach((p) => p.then((u) => u()).catch(() => {}));
     // eslint-disable-next-line react-hooks/exhaustive-deps
@@ -341,6 +379,145 @@ export default function App() {
       )}
     </div>
   ) : null;
+  const authoringNotice =
+    authoring &&
+    (authoring.status === "pending_allow" ||
+      authoring.status === "replace_allow" ||
+      authoring.status === "error" ||
+      authoring.status === "bound") ? (
+      <div
+        className={`pairing-notice ${authoring.status === "error" ? "error" : ""}`}
+        role={authoring.status === "error" ? "alert" : "status"}
+      >
+        <span>
+          {authoring.status === "error"
+            ? authoring.error || "The authoring bind could not be completed."
+            : authoring.status === "bound"
+              ? authoring.coach_only
+                ? "This job is coach-only on this window. ChatGPT can suggest; you click."
+                : authoring.allowed
+                  ? "Pin the browser URL or use this window. Titles stay on this computer."
+                  : "This computer is bound. Pin the window, then Allow ChatGPT to drive this job."
+              : authoring.prompt ||
+                (authoring.status === "replace_allow"
+                  ? `A different ${authoring.client_display || "ChatGPT"} account is asking. Allow it to replace the current one?`
+                  : `Allow ${authoring.client_display || "ChatGPT"} to drive this job`)}
+        </span>
+        {(authoring.status === "pending_allow" ||
+          authoring.status === "replace_allow") && (
+          <span className="allow-actions">
+            <button
+              type="button"
+              onClick={() => {
+                void engineTry(
+                  CMD.AUTHORING_ALLOW,
+                  { replace: authoring.status === "replace_allow" },
+                  {},
+                ).then(() =>
+                  engineTry(
+                    CMD.AUTHORING_STATUS,
+                    {},
+                    { status: "bound", allowed: true },
+                  ).then((next) => setAuthoring(next)),
+                );
+              }}
+            >
+              Allow
+            </button>
+            <button
+              type="button"
+              onClick={() => {
+                void engineTry(CMD.AUTHORING_DENY, {}, {}).then(() =>
+                  engineTry(
+                    CMD.AUTHORING_STATUS,
+                    {},
+                    { status: "bound", allowed: false },
+                  ).then((next) => setAuthoring(next)),
+                );
+              }}
+            >
+              Not now
+            </button>
+          </span>
+        )}
+        {authoring.status === "bound" && (
+          <span className="allow-actions">
+            <input
+              aria-label="Playwright URL"
+              onChange={(event) => setAuthoringUrl(event.target.value)}
+              placeholder="https://"
+              type="url"
+              value={authoringUrl}
+            />
+            <button
+              type="button"
+              onClick={() => {
+                void engineTry(
+                  CMD.AUTHORING_PIN_TARGET,
+                  { backend: "web", url: authoringUrl },
+                  {},
+                ).then((next) =>
+                  setAuthoring((current) =>
+                    current
+                      ? {
+                          ...current,
+                          coach_only:
+                            typeof next === "object" &&
+                            next !== null &&
+                            "coach_only" in next
+                              ? Boolean(
+                                  (next as { coach_only?: boolean }).coach_only,
+                                )
+                              : current.coach_only,
+                        }
+                      : current,
+                  ),
+                );
+              }}
+            >
+              Pin browser
+            </button>
+            <button
+              type="button"
+              onClick={() => {
+                void engineTry(
+                  CMD.AUTHORING_PIN_TARGET,
+                  { use_frontmost: true },
+                  {},
+                ).then((next) =>
+                  setAuthoring((current) =>
+                    current
+                      ? {
+                          ...current,
+                          coach_only:
+                            typeof next === "object" &&
+                            next !== null &&
+                            "coach_only" in next
+                              ? Boolean(
+                                  (next as { coach_only?: boolean }).coach_only,
+                                )
+                              : true,
+                        }
+                      : current,
+                  ),
+                );
+              }}
+            >
+              Use this window
+            </button>
+          </span>
+        )}
+        {authoring.status === "error" && (
+          <button
+            type="button"
+            aria-label="Dismiss authoring notice"
+            onClick={() => setAuthoring(null)}
+          >
+            ×
+          </button>
+        )}
+      </div>
+    ) : null;
   const firstWorkflowStageNotice = firstWorkflowStageError ? (
     <div className="pairing-notice error" role="alert">
       <span>{firstWorkflowStageError}</span>
@@ -358,6 +535,7 @@ export default function App() {
     return (
       <>
         {pairingNotice}
+        {authoringNotice}
         {firstWorkflowStageNotice}
         <DesktopEntryShell>
           <div className="center-stage"><span className="page-sub">Loading…</span></div>
@@ -370,6 +548,7 @@ export default function App() {
     return (
       <>
         {pairingNotice}
+        {authoringNotice}
         {firstWorkflowStageNotice}
         <DesktopEntryShell>
           <Login
@@ -390,6 +569,7 @@ export default function App() {
     return (
       <>
         {pairingNotice}
+        {authoringNotice}
         {firstWorkflowStageNotice}
         <DesktopEntryShell>
           <Onboarding
@@ -430,6 +610,7 @@ export default function App() {
   return (
     <>
       {pairingNotice}
+      {authoringNotice}
       {firstWorkflowStageNotice}
       <div className="app">
         <header className="desktop-shell">
diff --git a/src/lib/engine.ts b/src/lib/engine.ts
index 438bcef..6859ac9 100644
--- a/src/lib/engine.ts
+++ b/src/lib/engine.ts
@@ -70,6 +70,11 @@ export const CMD = {
   LOGIN_PASTE: "login_paste",
   LOGOUT: "logout",
   GET_AUTH_STATUS: "get_auth_status",
+  CLAIM_RUNNER_URI: "claim_runner_uri",
+  AUTHORING_ALLOW: "authoring_allow",
+  AUTHORING_DENY: "authoring_deny",
+  AUTHORING_STATUS: "authoring_status",
+  AUTHORING_PIN_TARGET: "authoring_pin_target",
   // config / settings (lane, phi_mode, hosted host)
   GET_CONFIG: "get_config",
   SET_CONFIG: "set_config",
@@ -119,6 +124,7 @@ export const EVT = {
   BREAK_COUNT: "break_count",
   SIDECAR_STATE: "sidecar_state",
   PAIRING_STATE: "pairing_state",
+  AUTHORING_STATE: "authoring_state",
   RUNNER_STATE: "runner_state",
   PORTAL_STATE: "portal_state",
   // Carries only {title, body, open_count, route}; see attentionNotification.ts.
diff --git a/src/lib/types.ts b/src/lib/types.ts
index 2df2738..9ccd996 100644
--- a/src/lib/types.ts
+++ b/src/lib/types.ts
@@ -80,6 +80,7 @@ export interface EngineStatus {
   paused: boolean;
   duration_secs?: number | null;
   capture_id?: string | null;
+  pause_prompt?: string | null;
   controls?: {
     pause: boolean;
     resume: boolean;
diff --git a/src/overlay/ControlOverlay.tsx b/src/overlay/ControlOverlay.tsx
index 983b793..ba83919 100644
--- a/src/overlay/ControlOverlay.tsx
+++ b/src/overlay/ControlOverlay.tsx
@@ -251,7 +251,9 @@ export function ControlOverlay() {
         <div className="overlay-copy" data-tauri-drag-region>
           <div className="overlay-primary" role="status" aria-live="polite">
             <span className="overlay-pulse" aria-hidden="true" />
-            <strong>{overlayPlainStatus(state.phase)}</strong>
+            <strong>
+              {state.pausePrompt || overlayPlainStatus(state.phase)}
+            </strong>
             <span className="overlay-step">{stepLabel}</span>
             {state.profile && (
               <span className="overlay-profile">{state.profile}</span>
@@ -293,10 +295,16 @@ export function ControlOverlay() {
             disabled={busy || !pauseAvailable}
             title={controlHelp}
             aria-describedby={!pauseAvailable ? "pause-unavailable" : undefined}
-            aria-label={showResume ? "Resume OpenAdapt" : "Pause OpenAdapt"}
+            aria-label={
+              showResume
+                ? state.pausePrompt
+                  ? "Continue OpenAdapt"
+                  : "Resume OpenAdapt"
+                : "Pause OpenAdapt"
+            }
             onClick={() => control(showResume ? "resume" : "pause")}
           >
-            {showResume ? "Resume" : "Pause"}
+            {showResume ? (state.pausePrompt ? "Continue" : "Resume") : "Pause"}
           </button>
           <button
             type="button"
diff --git a/src/overlay/state.test.ts b/src/overlay/state.test.ts
index 84c76ed..9c75709 100644
--- a/src/overlay/state.test.ts
+++ b/src/overlay/state.test.ts
@@ -11,6 +11,24 @@ import {
   CONTROL_OVERLAY_FRAME_VERSION,
 } from "./contract";
 
+it("shows the authoring pause card copy without sending titles", () => {
+  const state = reduceControlOverlay(EMPTY_OVERLAY_STATE, {
+    kind: "recording-status",
+    status: {
+      recording: true,
+      paused: true,
+      pause_prompt: "Type in the application. Continue here when done.",
+      controls: { pause: false, resume: true, stop: true },
+    },
+  });
+
+  expect(state.phase).toBe("paused");
+  expect(state.pausePrompt).toBe(
+    "Type in the application. Continue here when done.",
+  );
+  expect(JSON.stringify(state)).not.toContain("title");
+});
+
 it("projects only advertised recording controls", () => {
   const state = reduceControlOverlay(EMPTY_OVERLAY_STATE, {
     kind: "recording-status",
diff --git a/src/overlay/state.ts b/src/overlay/state.ts
index f2386ce..f2be905 100644
--- a/src/overlay/state.ts
+++ b/src/overlay/state.ts
@@ -41,6 +41,7 @@ export interface ControlOverlayState {
   evidenceClasses: string[];
   modelCalls: number | null;
   externalNetworkCalls: "none" | "observed" | "unknown" | null;
+  pausePrompt: string | null;
 }
 
 export type ControlOverlayInput =
@@ -75,6 +76,7 @@ export const EMPTY_OVERLAY_STATE: ControlOverlayState = {
   evidenceClasses: [],
   modelCalls: null,
   externalNetworkCalls: null,
+  pausePrompt: null,
 };
 
 /** Only states with no in-flight observation or actuation may receive input. */
@@ -105,6 +107,10 @@ function recordingState(status: EngineStatus): ControlOverlayState {
     controls: capabilities,
     elapsedSeconds:
       typeof status.duration_secs === "number" ? status.duration_secs : null,
+    pausePrompt:
+      status.paused && typeof status.pause_prompt === "string"
+        ? status.pause_prompt
+        : null,
   };
 }
 
diff --git a/src/styles/app.css b/src/styles/app.css
index facdde4..b2325a5 100644
--- a/src/styles/app.css
+++ b/src/styles/app.css
@@ -1515,6 +1515,30 @@ select.input:focus {
   font-size: 20px;
 }
 
+.pairing-notice .allow-actions {
+  display: flex;
+  gap: 8px;
+  flex-shrink: 0;
+}
+
+.pairing-notice .allow-actions button {
+  border: 1px solid var(--border);
+  border-radius: 8px;
+  padding: 6px 12px;
+  font-size: 13px;
+}
+
+.pairing-notice .allow-actions input[type="url"] {
+  min-width: 12rem;
+  flex: 1;
+  border: 1px solid var(--border);
+  border-radius: 8px;
+  padding: 6px 10px;
+  background: var(--bg);
+  color: var(--text);
+  font: inherit;
+}
+
 /* decision portal — pairing panel and paired-device list */
 .pairing-panel {
   display: flex;
diff --git a/tests/test_engine/test_auth_runner_bind.py b/tests/test_engine/test_auth_runner_bind.py
new file mode 100644
index 0000000..0e0080d
--- /dev/null
+++ b/tests/test_engine/test_auth_runner_bind.py
@@ -0,0 +1,107 @@
+"""Parse-only tests for ``openadapt://runner`` bind grammar."""
+
+from __future__ import annotations
+
+import pytest
+
+from engine.auth import pairing
+from engine.auth.runner_bind import (
+    RunnerBindError,
+    parse_runner_uri,
+    valid_bind_token,
+    valid_lease_secret,
+    valid_pack_id,
+)
+
+BIND = "oab_" + "A" * 43
+PACK = "p.abcdefghijkl"
+ORIGIN = "https://openadapt.ai"
+VALID_URI = f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https%3A%2F%2Fopenadapt.ai"
+CONNECT_SECRET = "oap_" + "A" * 43
+CONNECT_URI = (
+    f"openadapt://connect?pairing={CONNECT_SECRET}&host=https%3A%2F%2Fapp.openadapt.ai"
+)
+
+
+def test_parser_accepts_only_the_fixed_runner_action() -> None:
+    assert parse_runner_uri(VALID_URI) == {
+        "pack": PACK,
+        "bind": BIND,
+        "origin": ORIGIN,
+    }
+    for uri in (
+        VALID_URI.replace("://runner?", "://run?"),
+        VALID_URI.replace("://runner?", "://connect?"),
+        VALID_URI.replace("openadapt:", "https:"),
+        VALID_URI.replace("runner?", "runner/claim?"),
+        VALID_URI + "#fragment",
+        f"openadapt://user@runner?pack={PACK}&bind={BIND}&origin={ORIGIN}",
+        CONNECT_URI,
+    ):
+        with pytest.raises(RunnerBindError, match="Invalid OpenAdapt runner link"):
+            parse_runner_uri(uri)
+
+
+def test_connect_parser_rejects_runner_shapes() -> None:
+    with pytest.raises(pairing.PairingError, match="Invalid OpenAdapt connect link"):
+        pairing.parse_connect_uri(VALID_URI)
+    with pytest.raises(pairing.PairingError):
+        pairing.parse_connect_uri(
+            f"openadapt://connect?pack={PACK}&bind={BIND}&origin={ORIGIN}"
+        )
+
+
+def test_parser_rejects_malformed_missing_duplicate_and_unknown_fields() -> None:
+    bad = (
+        "",
+        "openadapt://runner?pack",
+        f"openadapt://runner?pack=short&bind={BIND}&origin={ORIGIN}",
+        f"openadapt://runner?pack={PACK}&bind={BIND}",
+        f"openadapt://runner?pack={PACK}&bind={BIND}&bind={BIND}&origin={ORIGIN}",
+        f"openadapt://runner?pack={PACK}&bind={BIND}&origin={ORIGIN}&command=whoami",
+        f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https://preview.openadapt.ai",
+        f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai/",
+        f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai:443",
+        f"openadapt://runner?pack={PACK}&bind={BIND}&origin=http://openadapt.ai",
+    )
+    for uri in bad:
+        with pytest.raises(RunnerBindError):
+            parse_runner_uri(uri)
+
+
+def test_prefix_parsers_reject_foreign_and_swapped_encodings() -> None:
+    oar = "oar_" + "a" * 64
+    oap = "oap_" + "A" * 43
+    oab_hex = "oab_" + "a" * 64
+    oals_b64 = "oals_" + "A" * 43
+    oa_prefix = "oa" + "A" * 43
+    oab = BIND
+    oals = "oals_" + "a" * 64
+
+    assert valid_bind_token(oab) is True
+    assert valid_lease_secret(oals) is True
+    assert valid_pack_id(PACK) is True
+    assert valid_pack_id("v1." + "A" * 48) is True
+
+    for value in (oar, oap, oab_hex, oals_b64, oals, oa_prefix, PACK):
+        assert valid_bind_token(value) is False, value
+    for value in (oar, oap, oab, oab_hex, oals_b64, oa_prefix, PACK):
+        assert valid_lease_secret(value) is False, value
+    for value in (oar, oap, oab, oals, oa_prefix, "p.short", "v1.short"):
+        assert valid_pack_id(value) is False, value
+
+    for bind in (oar, oap, oab_hex, oals_b64, oa_prefix):
+        uri = f"openadapt://runner?pack={PACK}&bind={bind}&origin={ORIGIN}"
+        with pytest.raises(RunnerBindError, match="Bind token is malformed"):
+            parse_runner_uri(uri)
+
+
+def test_argument_shaped_values_remain_data_and_cannot_select_an_action() -> None:
+    for payload in (
+        "--origin=https://evil.example",
+        "%2D%2Dorigin%3Dhttps%3A%2F%2Fevil.example",
+        "oab_" + "A" * 42 + ";",
+    ):
+        uri = f"openadapt://runner?pack={PACK}&bind={payload}&origin={ORIGIN}"
+        with pytest.raises(RunnerBindError, match="Bind token is malformed"):
+            parse_runner_uri(uri)
diff --git a/tests/test_engine/test_authoring_runner.py b/tests/test_engine/test_authoring_runner.py
new file mode 100644
index 0000000..7f8585b
--- /dev/null
+++ b/tests/test_engine/test_authoring_runner.py
@@ -0,0 +1,556 @@
+"""Authoring mailbox claim, Allow-per-sub, wait=0 poll, and record_observed."""
+
+from __future__ import annotations
+
+import json
+import stat
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import httpx
+import pytest
+
+from engine.auth.store import load_authoring_lease
+from engine.authoring_runner import (
+    POLL_WAIT_S,
+    AuthoringCoachOnly,
+    AuthoringError,
+    AuthoringRunner,
+    project_observe,
+)
+from engine.config import EngineConfig
+
+BIND = "oab_" + "A" * 43
+PACK = "p.abcdefghijkl"
+LEASE = "oals_" + "a" * 64
+ORIGIN = "https://openadapt.ai"
+URI = f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https%3A%2F%2Fopenadapt.ai"
+SUB = "b" * 64
+OTHER_SUB = "d" * 64
+CLIENT = "c" * 64
+
+
+class FakeAudit:
+    def __init__(self) -> None:
+        self.events: list[tuple[str, dict[str, Any]]] = []
+
+    def log(self, event: str, **data: Any) -> None:
+        self.events.append((event, data))
+
+
+class FakeRecorder:
+    def __init__(self) -> None:
+        self.clicks: list[tuple[int, int]] = []
+        self.observed: list[dict[str, Any]] = []
+        self.typed: list[Any] = []
+        self.finished = False
+
+    def click(self, x: int, y: int) -> None:
+        self.clicks.append((x, y))
+
+    def type_text(self, *args: Any, **kwargs: Any) -> None:
+        self.typed.append((args, kwargs))
+        raise AssertionError("Continue must not call type_text")
+
+    def record_observed(self, **kwargs: Any) -> None:
+        self.observed.append(kwargs)
+
+    def finish(self) -> SimpleNamespace:
+        self.finished = True
+        return SimpleNamespace(ok=True)
+
+
+def _nodes() -> list[dict[str, Any]]:
+    return [
+        {
+            "provider_runtime_id": "ax-elem-1",
+            "role": "button",
+            "control_type": "button",
+            "automation_id": "btnContinue",
+            "enabled": True,
+            "focused": False,
+            "bounds": {"x": 0.72, "y": 0.88, "w": 0.14, "h": 0.05},
+            "backend_pixels": {"x": 920, "y": 640, "w": 180, "h": 36},
+            "value": "SSN-SECRET",
+            "title": "Patient chart",
+            "name": "Continue",
+        },
+        {
+            "provider_runtime_id": "ax-note",
+            "role": "text_input",
+            "control_type": "edit",
+            "automation_id": "note",
+            "enabled": True,
+            "focused": True,
+            "bounds": {"x": 0.2, "y": 0.4, "w": 0.5, "h": 0.1},
+            "backend_pixels": {"x": 200, "y": 400, "w": 500, "h": 40},
+            "name": "note",
+        },
+    ]
+
+
+def _mailbox(
+    tmp_path: Path,
+    *,
+    claim_status: int = 201,
+    claim_body: dict[str, Any] | None = None,
+    poll_bodies: list[dict[str, Any] | None] | None = None,
+) -> tuple[AuthoringRunner, list[httpx.Request], FakeRecorder, FakeAudit]:
+    requests: list[httpx.Request] = []
+    polls = list(poll_bodies or [])
+    recorder = FakeRecorder()
+    audit = FakeAudit()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        path = request.url.path
+        if path.endswith("/runner/claim"):
+            if claim_status == 201:
+                return httpx.Response(
+                    201,
+                    headers={"Cache-Control": "no-store"},
+                    json=claim_body
+                    or {"leaseSecret": LEASE, "lease_s": 900},
+                )
+            return httpx.Response(
+                claim_status,
+                headers={"Cache-Control": "no-store"},
+                json={"error": "rejected"},
+            )
+        if path.endswith("/runner/poll"):
+            body = json.loads(request.content)
+            assert body["wait_seconds"] == 0
+            assert body["lease_seconds"] == 900
+            assert request.headers["Authorization"] == f"Bearer {LEASE}"
+            if not polls:
+                return httpx.Response(204)
+            next_body = polls.pop(0)
+            if next_body is None:
+                return httpx.Response(204)
+            return httpx.Response(
+                200,
+                headers={"Cache-Control": "no-store"},
+                json=next_body,
+            )
+        if path.endswith("/runner/callback"):
+            assert request.headers["Authorization"] == f"Bearer {LEASE}"
+            return httpx.Response(
+                202,
+                headers={"Cache-Control": "no-store"},
+                json={"accepted": True},
+            )
+        if path.endswith("/runner/allow"):
+            assert request.headers["Authorization"] == f"Bearer {LEASE}"
+            body = json.loads(request.content)
+            assert body["command_id"]
+            return httpx.Response(
+                200,
+                headers={"Cache-Control": "no-store"},
+                json={"allowedAt": "2026-09-01T00:00:00Z", "client_display": "ChatGPT"},
+            )
+        return httpx.Response(404)
+
+    client = httpx.Client(
+        base_url=ORIGIN,
+        transport=httpx.MockTransport(handler),
+        follow_redirects=False,
+    )
+    config = EngineConfig(data_dir=tmp_path / ".openadapt", log_level="WARNING")
+    config.data_dir.mkdir(parents=True, exist_ok=True)
+    runner = AuthoringRunner(
+        config,
+        audit=audit,
+        client=client,
+        sleep=lambda _seconds: None,
+        observe_nodes=_nodes,
+        recorder_factory=lambda _out_dir: recorder,
+        compile_recording=lambda _out_dir: SimpleNamespace(id="wf_mockmed"),
+        playwright_launcher=lambda url: SimpleNamespace(url=url, cookies=lambda: []),
+        text_value_at=lambda _pixels: "follow up in two weeks",
+    )
+    return runner, requests, recorder, audit
+
+
+def _envelope(tool: str, *, sub: str = SUB, args: dict[str, Any] | None = None) -> dict[str, Any]:
+    return {
+        "schema_version": "openadapt.authoring.command/v1",
+        "command_id": f"cmd_{tool}",
+        "pack_id": PACK,
+        "tool": tool,
+        "args": args or {},
+        "oauth_sub_sha256": sub,
+        "client_id_sha256": CLIENT,
+        "client_display": "ChatGPT",
+    }
+
+
+def test_claim_stores_lease_without_returning_the_secret(tmp_path: Path) -> None:
+    runner, requests, _recorder, _audit = _mailbox(tmp_path)
+    result = runner.claim_uri(URI, start_loop=False)
+    assert result["bound"] is True
+    assert "leaseSecret" not in result
+    assert "lease_secret" not in result
+    stored = load_authoring_lease(PACK)
+    assert stored is not None
+    assert stored["lease_secret"] == LEASE
+    assert stored["origin"] == ORIGIN
+    assert requests[0].url.path == f"/j/{PACK}/runner/claim"
+    assert json.loads(requests[0].content) == {"bind": BIND}
+
+
+@pytest.mark.parametrize("status", [409, 410, 404, 401])
+def test_claim_maps_mailbox_failures(tmp_path: Path, status: int) -> None:
+    runner, _requests, _recorder, _audit = _mailbox(tmp_path, claim_status=status)
+    with pytest.raises(AuthoringError):
+        runner.claim_uri(URI, start_loop=False)
+    assert load_authoring_lease(PACK) is None
+
+
+def test_poll_wait_is_zero_not_twenty_five(tmp_path: Path) -> None:
+    assert POLL_WAIT_S == 0
+    source = Path("engine/authoring_runner.py").read_text(encoding="utf-8")
+    assert "DEFAULT_WAIT_S" not in source
+    assert '"wait_seconds": POLL_WAIT_S' in source
+    assert "from openadapt_flow.backends.win_agent" not in source
+    assert "launch_agent(" not in source
+    assert "import parallels_vm" not in source
+    assert "hashlib.sha256(lease_secret" not in source
+    assert "hashlib.sha256(stored[\"lease_secret\"]" not in source
+    assert "bytes.fromhex(lease_secret[5:])" in source
+    assert '"allow"' in source
+    assert "PAUSE_PROMPT" in source
+    runner, requests, _recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.poll_once()
+    poll = next(item for item in requests if item.url.path.endswith("/poll"))
+    assert json.loads(poll.content)["wait_seconds"] == 0
+
+
+def test_bind_pack_allow_is_per_sub_and_required_for_halt(tmp_path: Path) -> None:
+    runner, requests, recorder, _audit = _mailbox(tmp_path)
+    events: list[tuple[str, dict[str, Any]]] = []
+    runner.emit = lambda event, data: events.append((event, data))
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    assert any(
+        event == "authoring_state" and data["status"] == "pending_allow"
+        for event, data in events
+    )
+    runner.handle_envelope(_envelope("observe"))
+    runner.handle_envelope(_envelope("halt"))
+    denied = [
+        json.loads(item.content)
+        for item in requests
+        if item.url.path.endswith("/callback")
+    ]
+    assert {item["result"]["error"] for item in denied} == {"not_allowed"}
+    assert runner.allow()["allowed"] is True
+    allow_req = next(item for item in requests if item.url.path.endswith("/allow"))
+    assert json.loads(allow_req.content) == {"command_id": "cmd_bind_pack"}
+    assert allow_req.headers["Authorization"] == f"Bearer {LEASE}"
+    runner.handle_envelope(_envelope("observe"))
+    observe = json.loads(requests[-1].content)["result"]
+    assert observe["schema_version"] == "openadapt.authoring.observe/v1"
+    assert "value" not in json.dumps(observe)
+    assert "title" not in json.dumps(observe)
+    assert "SSN-SECRET" not in json.dumps(observe)
+    runner.handle_envelope(_envelope("halt", sub=OTHER_SUB))
+    assert json.loads(requests[-1].content)["result"]["error"] == "not_allowed"
+    runner.handle_envelope(_envelope("halt"))
+    assert json.loads(requests[-1].content)["result"] == {"halted": True}
+    assert recorder.clicks == []
+
+
+def test_unsigned_stop_uses_lease_bearer_not_mcp_jwt(tmp_path: Path) -> None:
+    runner, requests, _recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.operator_stop()
+    callback = next(item for item in reversed(requests) if item.url.path.endswith("/callback"))
+    body = json.loads(callback.content)
+    assert body["halted"] is True
+    assert callback.headers["Authorization"] == f"Bearer {LEASE}"
+
+
+def test_continue_records_observed_on_pause_target_never_type_text(tmp_path: Path) -> None:
+    runner, requests, recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="web", url="https://openadapt.ai/mockmed")
+    runner.handle_envelope(_envelope("start_record"))
+    observe = runner._observe()
+    note = next(node for node in observe["tree"] if node.get("automation_id") == "note")
+    before = [item for item in requests if item.url.path.endswith("/callback")]
+    runner.handle_envelope(
+        _envelope("pause_for_input", args={"node_id": note["node_id"], "param": "note"})
+    )
+    after_pause = [item for item in requests if item.url.path.endswith("/callback")]
+    assert after_pause == before
+    assert runner.status_dict()["pause_prompt"] == (
+        "Type in the application. Continue here when done."
+    )
+    runner.continue_pause()
+    assert recorder.typed == []
+    assert recorder.observed[0]["event"] == {"kind": "type"}
+    assert recorder.observed[0]["text"] == "follow up in two weeks"
+    assert "secret" not in recorder.observed[0] or recorder.observed[0].get("secret") is not True
+    continued = json.loads(requests[-1].content)
+    assert continued["result"] == {"recorded": True, "param": "note"}
+    assert "text" not in continued["result"]
+    assert "value" not in continued["result"]
+
+
+def test_secret_continue_has_no_text_and_compile_refuses_if_missing(tmp_path: Path) -> None:
+    runner, _requests, recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="macos", window_title_unique=True)
+    runner.handle_envelope(_envelope("start_record"))
+    observe = runner._observe()
+    note = next(node for node in observe["tree"] if node.get("automation_id") == "note")
+    runner.handle_envelope(
+        _envelope(
+            "pause_for_input",
+            args={"node_id": note["node_id"], "param": "ssn", "secret": True},
+        )
+    )
+    with pytest.raises(AuthoringError, match="secret_type_missing"):
+        runner._compile()
+    runner.continue_pause()
+    assert "text" not in recorder.observed[0]
+    assert recorder.observed[0]["secret"] is True
+    compiled = runner._compile()
+    assert compiled == {
+        "status": "needs_human_admit",
+        "workflow_id": "wf_mockmed",
+        "recording_retained": True,
+    }
+    assert compiled.get("success") is not True
+
+
+def test_click_uses_backend_pixels_and_stale_node_does_not_retry(tmp_path: Path) -> None:
+    runner, _requests, recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="macos", window_title_unique=True)
+    runner.handle_envelope(_envelope("start_record"))
+    observe = runner._observe()
+    button = next(node for node in observe["tree"] if node.get("automation_id") == "btnContinue")
+    runner.handle_envelope(_envelope("click", args={"node_id": button["node_id"]}))
+    assert recorder.clicks == [(1010, 658)]
+    with pytest.raises(AuthoringError, match="stale_node"):
+        runner._click({"node_id": "n_deadbeef"})
+    runner._uncertain = True
+    with pytest.raises(AuthoringError, match="RECONCILIATION_REQUIRED"):
+        runner._click({"node_id": button["node_id"]})
+    assert recorder.clicks == [(1010, 658)]
+
+
+def test_linux_without_a_unique_title_is_coach_only(tmp_path: Path) -> None:
+    runner, _requests, recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="linux", window_title_unique=False)
+    observe = runner._observe()
+    assert observe["coach_only"] is True
+    with pytest.raises(AuthoringCoachOnly):
+        runner._start_record()
+    assert recorder.clicks == []
+
+
+def test_windows_native_is_coach_only_and_does_not_start_recorder(tmp_path: Path) -> None:
+    runner, _requests, recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="windows")
+    observe = runner._observe()
+    assert observe["coach_only"] is True
+    assert observe["agent_drive"] is False
+    assert observe["tree"] == []
+    with pytest.raises(AuthoringCoachOnly):
+        runner._start_record()
+    with pytest.raises(AuthoringCoachOnly):
+        runner._click({"node_id": "n_00000000"})
+    assert recorder.clicks == []
+    assert recorder.finished is False
+
+
+def test_playwright_launcher_gets_the_desktop_url_not_an_mcp_string(tmp_path: Path) -> None:
+    launched: list[str] = []
+    runner, _requests, _recorder, _audit = _mailbox(tmp_path)
+    runner._playwright_launcher = lambda url: launched.append(url) or SimpleNamespace(
+        cookies=lambda: []
+    )
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="web", url="https://openadapt.ai/mockmed")
+    runner._start_record()
+    assert launched == ["https://openadapt.ai/mockmed"]
+
+
+def test_node_table_is_mode_0600(tmp_path: Path) -> None:
+    runner, _requests, _recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner._observe()
+    assert runner._node_table is not None
+    mode = stat.S_IMODE(runner._node_table._path.stat().st_mode)
+    assert mode == 0o600
+
+
+def test_projector_drops_value_title_and_six_digit_names(tmp_path: Path) -> None:
+    from engine.authoring_runner import NodeTable
+
+    table = NodeTable(tmp_path / "nodes.json", b"key")
+    payload = project_observe(
+        backend="web",
+        provider="playwright_ax",
+        recording=False,
+        agent_drive=True,
+        coach_only=False,
+        process_name="Chromium",
+        raw_nodes=[
+            {
+                "provider_runtime_id": "x",
+                "role": "field",
+                "control_type": "edit",
+                "automation_id": "ssn-123456",
+                "name": "acct 1234567",
+                "value": "000-00-0000",
+                "title": "Secret",
+                "bounds": {"x": 0.1, "y": 0.1, "w": 0.1, "h": 0.1},
+                "backend_pixels": {"x": 1, "y": 1, "w": 2, "h": 2},
+            }
+        ],
+        node_table=table,
+    )
+    dumped = json.dumps(payload)
+    assert "000-00-0000" not in dumped
+    assert "Secret" not in dumped
+    assert "ssn-123456" not in dumped
+    assert "acct 1234567" not in dumped
+    assert "value" not in dumped
+    assert payload["tree"][0]["node_id"].startswith("n_")
+
+
+def test_dispatch_resume_and_stop_use_authoring_when_bound(tmp_path: Path) -> None:
+    from engine.db import IndexDB
+    from engine.dispatch import EngineDispatcher, EngineServices
+
+    runner, _requests, recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="macos", window_title_unique=True)
+    runner.handle_envelope(_envelope("start_record"))
+    observe = runner._observe()
+    note = next(node for node in observe["tree"] if node.get("automation_id") == "note")
+    runner.handle_envelope(
+        _envelope("pause_for_input", args={"node_id": note["node_id"], "param": "note"})
+    )
+    db = IndexDB(tmp_path / "index.db")
+    db.initialize()
+    disp = EngineDispatcher(
+        runner.config,
+        services=EngineServices(runner.config, db=db, authoring=runner, audit=FakeAudit()),
+    )
+    resumed = disp.dispatch("resume_recording", {})
+    assert resumed["paused"] is False
+    assert recorder.typed == []
+    assert recorder.observed
+    stopped = disp.dispatch("stop_recording", {})
+    assert stopped["halted"] is True
+    db.close()
+
+
+def test_replace_allow_required_for_a_second_sub(tmp_path: Path) -> None:
+    runner, _requests, _recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.handle_envelope(_envelope("bind_pack", sub=OTHER_SUB))
+    assert runner.status()["status"] == "replace_allow"
+    assert runner.allow()["allowed"] is True
+    assert runner._allowed_sub == SUB
+    runner.allow(replace=True)
+    assert runner._allowed_sub == OTHER_SUB
+
+
+def test_macos_without_unique_window_is_coach_only(tmp_path: Path) -> None:
+    runner, _requests, recorder, _audit = _mailbox(tmp_path)
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    pinned = runner.pin_target(backend="macos", macos_window_title="Patient Jane Doe")
+    assert pinned["coach_only"] is True
+    observe = runner._observe()
+    assert observe["coach_only"] is True
+    assert "Patient Jane Doe" not in json.dumps(observe)
+    with pytest.raises(AuthoringCoachOnly):
+        runner._start_record()
+    assert recorder.clicks == []
+
+
+def test_use_frontmost_unique_window_stays_local(tmp_path: Path) -> None:
+    runner, requests, _recorder, _audit = _mailbox(tmp_path)
+    runner._unique_window = lambda: {
+        "backend": "macos",
+        "unique": True,
+        "process_name": "Notes",
+        "window_title": "Patient Jane Doe",
+    }
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    pinned = runner.pin_target(use_frontmost=True)
+    assert pinned == {"ok": True, "backend": "macos", "coach_only": False}
+    assert "Patient Jane Doe" not in json.dumps(pinned)
+    observe = runner._observe()
+    assert observe["window"].get("process_name") == "Notes"
+    assert "Patient Jane Doe" not in json.dumps(observe)
+    assert all("Patient Jane Doe" not in item.content.decode("utf-8") for item in requests)
+
+
+def test_playwright_rejects_nonempty_cookies(tmp_path: Path) -> None:
+    runner, _requests, _recorder, _audit = _mailbox(tmp_path)
+    runner._playwright_launcher = lambda _url: SimpleNamespace(cookies=lambda: [{"name": "sid"}])
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="web", url="https://openadapt.ai/mockmed")
+    with pytest.raises(AuthoringError, match="empty cookies"):
+        runner._start_record()
+
+
+def test_callback_strips_titles_and_never_retries_uncertain_click(tmp_path: Path) -> None:
+    runner, requests, recorder, _audit = _mailbox(tmp_path)
+
+    def _boom(_x: int, _y: int) -> None:
+        raise RuntimeError("delivery unknown")
+
+    runner.claim_uri(URI, start_loop=False)
+    runner.handle_envelope(_envelope("bind_pack"))
+    runner.allow()
+    runner.pin_target(backend="macos", window_title_unique=True)
+    runner.handle_envelope(_envelope("start_record"))
+    observe = runner._observe()
+    button = next(node for node in observe["tree"] if node.get("automation_id") == "btnContinue")
+    recorder.click = _boom
+    runner.handle_envelope(_envelope("click", args={"node_id": button["node_id"]}))
+    body = json.loads(requests[-1].content)
+    assert body["result"]["error"] == "RECONCILIATION_REQUIRED"
+    assert "title" not in json.dumps(body)
+    assert "backend_pixels" not in json.dumps(body)
+    recorder.click = lambda x, y: recorder.clicks.append((x, y))
+    with pytest.raises(AuthoringError, match="RECONCILIATION_REQUIRED"):
+        runner._click({"node_id": button["node_id"]})
+    assert recorder.clicks == []
diff --git a/tests/test_pairing_protocol_boundary.py b/tests/test_pairing_protocol_boundary.py
index c30736a..ff4ab6f 100644
--- a/tests/test_pairing_protocol_boundary.py
+++ b/tests/test_pairing_protocol_boundary.py
@@ -21,10 +21,14 @@ def test_single_instance_precedes_deep_link_and_handoff_is_fixed() -> None:
     deep_link = main.index(".plugin(tauri_plugin_deep_link::init())")
     assert single < deep_link
     assert 'command: "connect_uri"' in pairing
+    assert 'command: "claim_runner_uri"' in pairing
     assert 'json!({ "uri": action.uri })' in pairing
     assert "std::process::Command" not in pairing
     assert "open_external" not in pairing
     assert "ShellExt" not in pairing
+    assert 'Some("connect") => connect_action' in pairing
+    assert 'Some("runner") => runner_action' in pairing
+    assert "pack" not in pairing.split("fn connect_action", 1)[1].split("fn runner_action", 1)[0]
 
 
 def test_python_pairing_action_has_no_shell_or_navigation_escape_hatch() -> None:
@@ -32,6 +36,7 @@ def test_python_pairing_action_has_no_shell_or_navigation_escape_hatch() -> None
     store = (ROOT / "engine/auth/store.py").read_text()
     dispatch = (ROOT / "engine/dispatch.py").read_text()
     assert '"connect_uri": self.connect_uri' in dispatch
+    assert '"claim_runner_uri": self.claim_runner_uri' in dispatch
     assert "subprocess" not in pairing
     assert "shell=" not in pairing
     assert "webbrowser" not in pairing
@@ -43,3 +48,26 @@ def test_python_pairing_action_has_no_shell_or_navigation_escape_hatch() -> None
     assert "write_text" not in pairing + store
     assert "open(" not in pairing + store
     assert "logger." not in pairing
+
+
+def test_python_runner_bind_is_parse_only() -> None:
+    runner_bind = (ROOT / "engine/auth/runner_bind.py").read_text()
+    assert "def parse_runner_uri" in runner_bind
+    assert "httpx" not in runner_bind
+    assert "keyring" not in runner_bind
+    assert "subprocess" not in runner_bind
+    assert "webbrowser" not in runner_bind
+    assert "os.system" not in runner_bind
+    assert "wait_seconds" not in runner_bind
+    assert "DEFAULT_WAIT_S" not in runner_bind
+    assert "oar_" in runner_bind
+    assert "oap_" in runner_bind
+    assert "oab_" in runner_bind
+    assert "oals_" in runner_bind
+    assert "BIND_HEX_BODY_RE" in runner_bind
+    assert "LEASE_BASE64URL_BODY_RE" in runner_bind
+    authoring = (ROOT / "engine/authoring_runner.py").read_text(encoding="utf-8")
+    assert 'action not in MAILBOX_ACTIONS' in authoring
+    assert '"allow"' in authoring
+    assert "win_agent" in authoring
+    assert "from openadapt_flow.backends.win_agent" not in authoring