diff --git a/docs/runtime-model.md b/docs/runtime-model.md
index 7384f0d..385fd40 100644
--- a/docs/runtime-model.md
+++ b/docs/runtime-model.md
@@ -133,12 +133,50 @@ exiting; the retained shell is not an agent runtime. The three states:
 |---|---|---|---|---|---|
 | pane dead | false | false | offline | pane_dead | false |
 | retained shell (CLI exited) | true | false | offline | cli_exited | false |
+| anchored member (`remote`) | true | false | offline | cli_exited | false |
 | live CLI | true | true | per runtime | per runtime | per runtime |
 
 Consumers — delivery refuses a retained shell before any native transport
 (the send event stays durable on the bus); idle notify, session-snapshot
 capture, and duo pairing all skip retained shells.
 
+### `remote`
+
+Source — the member pane's `@hive-remote` tag, written at registration.
+
+Meaning — the member's agent process does not live on this pane. `channel`
+is the only value today: the pane is an **anchor** whose channel socket and
+ready marker are symlinks to an external Claude session's own
+`hive-client-<pid>.sock`. The pane exists to hold the member's identity, so
+every pane-keyed authority (routing, tags, `kill-pane` as the kick control,
+doctor) keeps working unchanged.
+
+The honest asymmetry: `alive` no longer implies the member is reachable, and
+`cliAlive` is permanently `false` because no CLI was ever meant to run here.
+
+**An anchored member has no push transport.** A host that launches its
+sessions without `--channels` — the desktop app owns its argv — cannot
+receive channel notifications at all, so nothing is pushed to this pane and
+the channel socket is never written to. Delivery is the durable bus write
+plus the member's own inbox hook, which drains it after each tool call and
+again at the end of the turn. `Agent.send` names that boundary
+`busInboxAccepted`; it deliberately claims less than the channel and daemon
+classifications, because no push was attempted and none may be claimed.
+
+The socket still earns its keep as **liveness evidence**: the external
+session's channel server unlinks the real socket and marker when it exits, so
+the anchor's symlinks dangle and delivery fails closed exactly like a dead
+pane server — the message stays on the bus for its return.
+
+Consumers — delivery gates an anchored member on that liveness rather than
+refusing it as a retained shell; the resume snapshot records the marker and
+resume skips those members instead of spawning a look-alike CLI on the team's
+routing key (an external session reconnects itself by re-running `hive duo
+init --channel <socket>`, which relinks the anchor). Because every session on
+the host runs the same inbox hook, the saved context carries a `session`
+claim: the first session to drain owns the identity and siblings are refused,
+and re-forming the duo clears the claim.
+
 ### `inputState`
 
 Source:
diff --git a/plugins/hive/hooks/hooks.json b/plugins/hive/hooks/hooks.json
index 439cfe0..8453876 100644
--- a/plugins/hive/hooks/hooks.json
+++ b/plugins/hive/hooks/hooks.json
@@ -13,6 +13,29 @@
           }
         ]
       }
+    ],
+    "PostToolUse": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/inbox_hook.py\"",
+            "timeout": 20
+          }
+        ]
+      }
+    ],
+    "Stop": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/inbox_hook.py\"",
+            "timeout": 180,
+            "statusMessage": "hive inbox..."
+          }
+        ]
+      }
     ]
   }
 }
diff --git a/plugins/hive/scripts/inbox_hook.py b/plugins/hive/scripts/inbox_hook.py
new file mode 100644
index 0000000..ff28d1a
--- /dev/null
+++ b/plugins/hive/scripts/inbox_hook.py
@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Deliver a hive member's inbound messages into a session that channels cannot reach.
+
+Claude Code delivers channel notifications only to sessions launched with
+``--channels``. Sessions hosted by the Claude Code desktop app are not: the
+app owns its argv, so a desktop-led duo worker never sees the push (measured
+A/B: same binary, same stream-json transport, flag present -> delivered, flag
+absent -> silently dropped). A hook is the only injection point that lives
+*inside* the session process, so it is the delivery path for those members.
+
+Two events, two jobs:
+
+- ``PostToolUse`` — mid-turn delivery. A member doing real work calls tools
+  constantly, so each tool result is a delivery point; a reply that lands
+  during a long turn arrives within one tool call instead of at turn end.
+  Never blocks, never waits.
+- ``Stop`` — the idle/settle path. Drains what is left, and while one of the
+  member's own messages is still unanswered it holds the turn open waiting
+  for the answer (the duo's "wait for the verdict" semantics).
+
+Silent no-op everywhere else: inside tmux the native channel/app-server
+transports already deliver, and a session bound to no hive team has no inbox.
+Each drain advances a durable cursor, so no message is delivered twice and a
+blocked stop cannot repeat on the same message. ``--session`` scopes the
+drain to the member's own session, so other desktop sessions running the same
+hook never steal its inbox.
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+
+WAIT_SECONDS = 120  # ponytail: fixed budget; make it a setting if a duo outgrows it
+
+
+def _collect(args: list[str]) -> dict:
+    hive = shutil.which("hive")
+    if not hive:
+        return {}
+    try:
+        out = subprocess.run(
+            [hive, "collect", *args],
+            capture_output=True,
+            text=True,
+            timeout=WAIT_SECONDS + 30,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return {}
+    if out.returncode != 0:
+        return {}
+    try:
+        payload = json.loads(out.stdout)
+    except ValueError:
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def _envelope(message: dict) -> str:
+    head = f"<HIVE from={message.get('from', '?')} to={message.get('to', '?')}"
+    if message.get("msgId"):
+        head += f" msgId={message['msgId']}"
+    if message.get("replyTo") or message.get("inReplyTo"):
+        head += f" reply-to={message.get('replyTo') or message.get('inReplyTo')}"
+    if message.get("artifact"):
+        head += f" artifact={message['artifact']}"
+    return f"{head}>\n{message.get('body', '')}\n</HIVE>"
+
+
+def _preamble(count: int) -> str:
+    return (
+        f"{count} hive message{'s' if count > 1 else ''} arrived. Follow the hive "
+        f"protocol for each (reply with `hive reply <agent>`):\n\n"
+    )
+
+
+def main() -> None:
+    if os.environ.get("TMUX_PANE"):
+        return  # native channel / app-server transports own delivery in tmux
+    try:
+        event = json.load(sys.stdin)
+    except (ValueError, OSError):
+        event = {}
+    session = str(event.get("session_id") or "")
+    claim = ["--session", session] if session else []
+
+    if event.get("hook_event_name") == "PostToolUse":
+        # Mid-turn delivery: every tool call is a delivery point, so a message
+        # that lands while the member is working arrives with the next tool
+        # result instead of waiting for the turn to end. Never waits — a turn
+        # in progress must not be stalled by an empty inbox.
+        messages = (_collect(claim).get("messages")) or []
+        if not messages:
+            return
+        body = "\n\n".join(_envelope(m) for m in messages)
+        print(json.dumps({
+            "hookSpecificOutput": {
+                "hookEventName": "PostToolUse",
+                "additionalContext": _preamble(len(messages)) + body,
+            },
+        }))
+        return
+
+    already_blocking = bool(event.get("stop_hook_active"))
+    payload = _collect(claim)
+    if not payload.get("messages") and not already_blocking:
+        # Nothing unread. Wait for the answer only while one is actually owed,
+        # so an idle session ends its turn immediately.
+        payload = _collect([*claim, "--wait", str(WAIT_SECONDS), "--if-awaiting"])
+    messages = payload.get("messages") or []
+    if not messages:
+        return
+
+    body = "\n\n".join(_envelope(m) for m in messages)
+    count = len(messages)
+    print(json.dumps({
+        "decision": "block",
+        "reason": _preamble(count) + body,
+        "systemMessage": f"hive: {count} message{'s' if count > 1 else ''} delivered",
+    }))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/hive/adapters/claude_channel.py b/src/hive/adapters/claude_channel.py
index 51424fb..888564c 100644
--- a/src/hive/adapters/claude_channel.py
+++ b/src/hive/adapters/claude_channel.py
@@ -282,6 +282,64 @@ def prepare_pane(cwd: str) -> list[str]:
     return ["--channels", PLUGIN_SPEC]
 
 
+# --- anchor pane linkage ----------------------------------------------------
+
+def link_client_socket(pane: str, client_sock: str | Path) -> str | None:
+    """Point *pane*'s channel socket + marker at an external client's.
+
+    Used by ``hive duo init --channel``: the anchor pane of a desktop-led duo
+    has no CLI of its own, so its pane-addressed socket/marker become symlinks
+    to the external Claude session's ``hive-client-<pid>.sock``. Delivery
+    (:func:`send_to_pane`) is untouched: ``connect()`` follows symlinks, and
+    when the client's server exits and unlinks its real files the symlinks
+    dangle — ``exists()`` goes False and sends fail closed, exactly like a
+    dead pane server.
+
+    Returns None on success, or a human-readable refusal. Fail-closed: the
+    client socket must exist and its marker must advertise a known version
+    before any link is created.
+    """
+    client_sock = Path(client_sock)
+    client_marker = Path(marker_path_for_socket_path(client_sock))
+    if not client_sock.exists():
+        return f"channel socket {client_sock} does not exist (is the Claude session still running?)"
+    try:
+        version = client_marker.read_text().strip()
+    except OSError:
+        return f"channel marker {client_marker} is unreadable; refusing to link an unready channel"
+    if version not in (MARKER_LEGACY, MARKER_RECEIPT_CAPABLE):
+        return f"channel marker {client_marker} has unknown version {version!r}; refusing to link"
+    sock_link = channel_socket_path(pane)
+    marker_link = ready_marker_path(pane)
+    sock_link.parent.mkdir(parents=True, exist_ok=True)
+    for link, target in ((sock_link, client_sock), (marker_link, client_marker)):
+        try:
+            link.unlink()
+        except OSError:
+            pass
+        os.symlink(target, link)
+    return None
+
+
+def marker_path_for_socket_path(sock_path: str | Path) -> str:
+    """Marker path for a raw socket path (same shape the server writes)."""
+    s = str(sock_path)
+    return s[: -len(".sock")] + ".ready" if s.endswith(".sock") else s + ".ready"
+
+
+def remote_member_alive(pane: str) -> bool:
+    """Whether an anchored member's external session is still running.
+
+    Its channel server unlinks the real socket and marker on exit, so the
+    anchor's symlinks dangle and ``exists()`` (which follows them) goes False.
+    Nothing is ever pushed over this socket — a session the app launched
+    without ``--channels`` cannot receive channel notifications — so this is
+    liveness evidence only, and delivery to a live member is the bus plus the
+    member's own inbox hook.
+    """
+    return channel_socket_path(pane).exists() and ready_marker_path(pane).exists()
+
+
 # --- delivery ---------------------------------------------------------------
 
 def _extract_msg_id(text: str) -> str:
diff --git a/src/hive/adapters/claude_channel_server.py b/src/hive/adapters/claude_channel_server.py
index ba57cfe..4eca283 100644
--- a/src/hive/adapters/claude_channel_server.py
+++ b/src/hive/adapters/claude_channel_server.py
@@ -8,7 +8,11 @@
 
 Inbound seam: a per-pane unix socket under ``$HIVE_HOME/channel`` whose name is
 derived from ``$TMUX_PANE`` (so the plugin's single server entry serves every
-pane). ``claude_channel.send_to_pane`` connects and writes one JSON frame
+pane). Outside tmux (Claude Code desktop) the socket is pid-keyed
+(``hive-client-<pid>.sock``) and its path is appended to the MCP instructions;
+``hive duo init --channel`` symlinks an anchor pane's socket to it so
+pane-addressed delivery reaches the session. ``claude_channel.send_to_pane``
+connects and writes one JSON frame
 ``{"msg_id": ..., "content": ...}``; this server emits it to Claude and then
 answers with a single-byte **local MCP-write receipt** (``b"1"``) on the same
 connection. The receipt only proves the notification was written+flushed to
@@ -42,6 +46,14 @@
     "and follow the hive protocol exactly as if it were injected directly. "
     "This channel is one-way: reply with the hive CLI, not a channel tool."
 )
+# Appended for sessions outside tmux (Claude Code desktop): the agent reads its
+# own socket path from these instructions and hands it to `hive duo init
+# --channel` — the agent itself is the bridge between this MCP server and the
+# hive CLI, so no separate discovery mechanism exists.
+CLIENT_INSTRUCTIONS_SUFFIX = (
+    " This session runs outside tmux; its channel socket is {path}. To form a "
+    "duo from here, run: hive duo init --channel {path}"
+)
 
 _stdout_lock = threading.Lock()
 # The ready marker is published only once BOTH gates are open: the socket is
@@ -77,14 +89,24 @@ def socket_path_for_pane(pane: str) -> str:
     return os.path.join(_hive_home(), "channel", f"hive-pane-{slug}.sock")
 
 
+def socket_path_for_client(pid: int) -> str:
+    """Socket for a Claude session outside tmux (e.g. Claude Code desktop).
+
+    Keyed by this server's pid — unique per session, dies with it. An anchor
+    pane's ``hive-pane-*.sock`` is symlinked here by ``hive duo init
+    --channel`` so pane-addressed delivery reaches the external session.
+    """
+    return os.path.join(_hive_home(), "channel", f"hive-client-{pid}.sock")
+
+
 def marker_path_for_socket(sock_path: str) -> str:
     """Same name shape as ``claude_channel.ready_marker_path``."""
     return sock_path[: -len(".sock")] + ".ready"
 
 
-def _resolve_socket_path() -> str | None:
+def _resolve_socket_path() -> str:
     pane = os.environ.get("TMUX_PANE", "")
-    return socket_path_for_pane(pane) if pane else None
+    return socket_path_for_pane(pane) if pane else socket_path_for_client(os.getpid())
 
 
 def _safe_unlink(path: str) -> None:
@@ -192,11 +214,15 @@ def _maybe_publish_marker(path: str) -> None:
 
 def _handle_request(method: str, params: dict) -> dict:
     if method == "initialize":
+        instructions = INSTRUCTIONS
+        if not os.environ.get("TMUX_PANE", ""):
+            path = _resolve_socket_path()
+            instructions += CLIENT_INSTRUCTIONS_SUFFIX.format(path=path)
         return {
             "protocolVersion": params.get("protocolVersion") or "2025-06-18",
             "capabilities": {"experimental": {"claude/channel": {}}},
             "serverInfo": {"name": SERVER_NAME, "version": "0.1.0"},
-            "instructions": INSTRUCTIONS,
+            "instructions": instructions,
         }
     if method == "ping":
         return {}
@@ -209,29 +235,26 @@ class _MethodNotFound(Exception):
 
 def main() -> None:
     path = _resolve_socket_path()
-    if path:
-        # Signal handlers must be installed from the main thread; unlink the
-        # socket + ready marker on SIGTERM/SIGINT (Claude killing the MCP
-        # child) since atexit does not run on signal termination. A stale
-        # socket is also cleared before bind on the next spawn, and spawn
-        # clears a stale marker before launch, so this is best-effort.
-        def _remove_artifacts() -> None:
-            _safe_unlink(path)
-            _safe_unlink(marker_path_for_socket(path))
-
-        def _cleanup(*_args: object) -> None:
-            _remove_artifacts()
-            os._exit(0)
+    # Signal handlers must be installed from the main thread; unlink the
+    # socket + ready marker on SIGTERM/SIGINT (Claude killing the MCP
+    # child) since atexit does not run on signal termination. A stale
+    # socket is also cleared before bind on the next spawn, and spawn
+    # clears a stale marker before launch, so this is best-effort.
+    def _remove_artifacts() -> None:
+        _safe_unlink(path)
+        _safe_unlink(marker_path_for_socket(path))
+
+    def _cleanup(*_args: object) -> None:
+        _remove_artifacts()
+        os._exit(0)
 
-        try:
-            signal.signal(signal.SIGTERM, _cleanup)
-            signal.signal(signal.SIGINT, _cleanup)
-        except (ValueError, OSError):
-            pass
-        atexit.register(_remove_artifacts)
-        threading.Thread(target=_socket_loop, args=(path,), daemon=True).start()
-    else:
-        _log("no TMUX_PANE; channel socket disabled (MCP handshake only)")
+    try:
+        signal.signal(signal.SIGTERM, _cleanup)
+        signal.signal(signal.SIGINT, _cleanup)
+    except (ValueError, OSError):
+        pass
+    atexit.register(_remove_artifacts)
+    threading.Thread(target=_socket_loop, args=(path,), daemon=True).start()
     for raw in sys.stdin:
         raw = raw.strip()
         if not raw:
@@ -248,7 +271,7 @@ def _cleanup(*_args: object) -> None:
             # Readiness strictly follows the initialize RESPONSE reaching the
             # MCP transport — never the request handling alone. Only now may a
             # sender observe the pane as channel-ready.
-            if msg.get("method") == "initialize" and path and not _initialized.is_set():
+            if msg.get("method") == "initialize" and not _initialized.is_set():
                 _initialized.set()
                 _maybe_publish_marker(path)
         except _MethodNotFound as e:
diff --git a/src/hive/adapters/codex_app_server.py b/src/hive/adapters/codex_app_server.py
index 40d9e36..a32578b 100644
--- a/src/hive/adapters/codex_app_server.py
+++ b/src/hive/adapters/codex_app_server.py
@@ -480,6 +480,14 @@ def spawn_daemon(
     short-lived CLI; the sidecar reaps it via the pidfile when the pane dies.
     Returns False if the daemon fails to bind or dies before becoming ready.
     """
+    if not os.path.isabs(codex_bin):
+        # Resolve the bare name here, once: Popen would otherwise resolve it
+        # against this process's PATH while the pane's TUI resolves against
+        # the pane shell's — a leftover second install silently splits the
+        # two (2026-07-14 version-skew incident).
+        import shutil
+
+        codex_bin = shutil.which(codex_bin) or codex_bin
     sock = pane_socket_path(pane)
     sock.parent.mkdir(parents=True, exist_ok=True)
     if sock.exists():
diff --git a/src/hive/agent.py b/src/hive/agent.py
index a3d0956..dcecfc9 100644
--- a/src/hive/agent.py
+++ b/src/hive/agent.py
@@ -20,6 +20,12 @@
     "codex": "codex",
 }
 
+# Accepted-transport classification for a member with no push transport: the
+# message is durable on the bus and the member's own inbox hook drains it. It
+# claims strictly less than the channel/daemon classifications — no push was
+# attempted, so none can be claimed.
+ACCEPTED_BUS_INBOX = "busInboxAccepted"
+
 
 def _shell_escape(s: str) -> str:
     """Escape a string for safe shell use."""
@@ -206,6 +212,7 @@ def spawn(
         cli: str = "claude",
         workspace: str = "",
         session_mode: str = "fork",
+        allow_outside_tmux: bool = False,
     ) -> Agent:
         """Spawn an agent CLI (claude/codex) in a tmux pane.
 
@@ -225,7 +232,13 @@ def spawn(
             raise ValueError(f"unsupported session_mode '{session_mode}', must be fork or resume")
         cwd = cwd or os.getcwd()
         if not tmux.is_inside_tmux():
-            raise ValueError(_TMUX_REQUIRED_MESSAGE)
+            # A desktop-led duo spawns into a detached session; the opt-in
+            # trades the caller-location gate for positive evidence that the
+            # target pane exists on the server.
+            if not allow_outside_tmux:
+                raise ValueError(_TMUX_REQUIRED_MESSAGE)
+            if not target_pane or not tmux.get_pane_window_target(target_pane):
+                raise ValueError(f"target pane '{target_pane}' not found on the tmux server")
 
         from .agent_cli import get_profile
         profile = get_profile(cli)
@@ -258,7 +271,13 @@ def spawn(
         tmux.set_pane_title(pane_id, f"[{name}]")
         tmux.tag_pane(pane_id, "agent", name, team_name, cli=cli)
 
-        bin_path = CLI_BINS[cli]
+        # Resolve the binary once, here, to an absolute path: the pane shell
+        # and the daemon Popen otherwise each resolve the bare name against
+        # their *own* PATH (spawner env vs pane env), and a version-skewed
+        # leftover install silently splits them (2026-07-14 codex incident).
+        import shutil
+
+        bin_path = shutil.which(CLI_BINS[cli]) or CLI_BINS[cli]
         # No `exec`: the CLI runs as the pane shell's foreground child, so
         # the pane (and a usable shell) survives the CLI exiting.
         cmd_parts = [_shell_escape(bin_path)]
@@ -273,7 +292,7 @@ def spawn(
             # handoff shortcut stays embedded (below).
             if not session_id or session_mode == "resume":
                 from .adapters import codex_app_server
-                if not codex_app_server.spawn_daemon(pane_id):
+                if not codex_app_server.spawn_daemon(pane_id, codex_bin=bin_path):
                     # Codex runtime state is daemon-native only (embedded codex
                     # is unsupported), so a pane without a daemon would join the
                     # team stateless. Undo the pane side effects instead of
@@ -422,6 +441,12 @@ def send(self, text: str) -> str:
         # retained shell can carry a stale title, the declared cli tag, and
         # (for codex) a surviving per-pane daemon with an open thread — none
         # of that may route a message into a pane nobody is watching.
+        #
+        # Exception: an anchor pane (`@hive-remote=channel`) hosts no CLI by
+        # design — its member lives in an external session that pulls its mail
+        # off the bus (the plugin's inbox hook). Nothing is pushed to it, so
+        # this is not a transport at all; the pane's channel link is only
+        # liveness evidence that the external session still exists.
         probe = None
         try:
             from .agent_cli import detect_cli_process_for_pane
@@ -430,6 +455,19 @@ def send(self, text: str) -> str:
         except Exception:
             probe = None
         if probe is None:
+            if tmux.get_pane_option(self.pane_id, "hive-remote") == "channel":
+                from .adapters import claude_channel
+
+                if not claude_channel.remote_member_alive(self.pane_id):
+                    raise DeliveryError(
+                        f"remote member on pane {self.pane_id} is gone "
+                        "(its session unlinked the channel socket); the message "
+                        "stays on the bus for when it comes back"
+                    )
+                # The bus write already happened upstream and the member's
+                # inbox hook drains it. Naming that honestly keeps the delivery
+                # state machine from recording a push that never happens.
+                return ACCEPTED_BUS_INBOX
             raise DeliveryError(
                 f"no live CLI process on pane {self.pane_id} (cli_exited): "
                 "refusing native transport to a retained shell"
diff --git a/src/hive/bus.py b/src/hive/bus.py
index 1f0fc3d..3b4d1a5 100644
--- a/src/hive/bus.py
+++ b/src/hive/bus.py
@@ -355,6 +355,32 @@ def count_events(workspace: str | Path) -> int:
     return int(row["count"]) if row is not None else 0
 
 
+def read_inbound_after(
+    workspace: str | Path,
+    *,
+    recipient: str,
+    after_seq: int,
+) -> list[tuple[int, dict[str, object]]]:
+    """Inbound send events for *recipient* newer than *after_seq*, oldest first.
+
+    The read side of `hive collect`: a member whose session cannot receive
+    channel push (a desktop-led worker) drains its inbox from the bus, keyed
+    by the monotonic seq as a durable cursor.
+    """
+    with _connect(workspace) as conn:
+        rows = conn.execute(
+            """
+            SELECT * FROM messages
+            WHERE intent = 'send'
+              AND to_agent = ?
+              AND seq > ?
+            ORDER BY seq ASC
+            """,
+            (recipient, after_seq),
+        ).fetchall()
+    return [(int(row["seq"]), _row_to_event(row)) for row in rows]
+
+
 def latest_inbound_send_event(
     workspace: str | Path,
     *,
@@ -408,6 +434,76 @@ def latest_unanswered_inbound_send_event(
     return _row_to_event(row) if row is not None else None
 
 
+def is_awaiting_reply(
+    workspace: str | Path,
+    *,
+    sender: str,
+    within_seconds: float,
+    now: float | None = None,
+) -> bool:
+    """True when *sender* just sent an outbound still owed a reply — i.e. it is
+    *sender*'s turn to wait for a peer, not to act.
+
+    Two gates, both required:
+
+    1. *Structure*: the newest message *sender* is party to must be *sender*'s
+       own send, unanswered. Judged from the single latest message, never from
+       "any unanswered send anywhere" — a finished exchange always leaves one
+       trailing unanswered send (whoever spoke last is never replied to), so an
+       any-unanswered test latches True forever.
+    2. *Recency*: that send must be newer than *within_seconds*. Structure alone
+       cannot tell a fresh handoff ("review this", ending the turn to wait) from
+       a stale sign-off ("thanks, shipping") the member left behind turns ago —
+       both are a trailing unanswered outbound. A handoff worth waiting on is
+       seconds old; once it ages past the wait window a reply is not imminent,
+       so later unrelated turns do not re-arm on it.
+
+    *now* is injectable for tests; it defaults to wall-clock time.
+    """
+    with _connect(workspace) as conn:
+        row = conn.execute(
+            """
+            SELECT from_agent, msg_id, created_at FROM messages
+            WHERE intent = 'send' AND (from_agent = ? OR to_agent = ?)
+            ORDER BY seq DESC
+            LIMIT 1
+            """,
+            (sender, sender),
+        ).fetchone()
+        if row is None or row["from_agent"] != sender or not row["msg_id"]:
+            return False  # no history, or the latest move was inbound / anonymous
+        answered = conn.execute(
+            """
+            SELECT 1 FROM messages
+            WHERE intent = 'send' AND in_reply_to = ? AND to_agent = ?
+            LIMIT 1
+            """,
+            (row["msg_id"], sender),
+        ).fetchone()
+    if answered is not None:
+        return False
+    age = _age_seconds(str(row["created_at"] or ""), now=now)
+    return age is not None and age <= within_seconds
+
+
+def _age_seconds(created_at: str, *, now: float | None) -> float | None:
+    """Seconds between an ISO-8601 ``created_at`` and *now* (wall clock if None).
+    None when the timestamp is missing or unparseable — an unknown age is not a
+    fresh one, so callers gate it out."""
+    if not created_at:
+        return None
+    import datetime as _dt
+
+    try:
+        ts = _dt.datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=_dt.timezone.utc)
+    current = now if now is not None else _dt.datetime.now(_dt.timezone.utc).timestamp()
+    return current - ts.timestamp()
+
+
 def has_send_reply_to(
     workspace: str | Path,
     *,
diff --git a/src/hive/cli.py b/src/hive/cli.py
index 743717a..36578f6 100644
--- a/src/hive/cli.py
+++ b/src/hive/cli.py
@@ -12,6 +12,7 @@
 import sys
 import time
 from collections import defaultdict
+from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -34,6 +35,7 @@
     "team": "Daily",
     "send": "Daily",
     "reply": "Daily",
+    "collect": "Daily",
     "notify": "Daily",
     "compact": "Daily",
     "skills": "Daily",
@@ -184,11 +186,23 @@ def _discover_tmux_binding() -> dict[str, str]:
 
 
 def _default_team() -> str | None:
-    return _discover_tmux_binding().get("team")
+    discovered = _discover_tmux_binding().get("team")
+    if discovered:
+        return discovered
+    if not tmux.is_inside_tmux():
+        # Desktop-led worker: outside tmux the identity lives in the saved
+        # default context (written by `hive duo init --channel`).
+        return hive_context.load_current_context().get("team") or None
+    return None
 
 
 def _default_agent() -> str | None:
-    return _discover_tmux_binding().get("agent")
+    discovered = _discover_tmux_binding().get("agent")
+    if discovered:
+        return discovered
+    if not tmux.is_inside_tmux():
+        return hive_context.load_current_context().get("agent") or None
+    return None
 
 
 def _require_team(team: str | None) -> str:
@@ -753,7 +767,12 @@ def cli(ctx: click.Context):
         return
     _require_codex_native(ctx.invoked_subcommand)
     if ctx.invoked_subcommand not in _TMUX_OPTIONAL_ROOT_COMMANDS and ctx.invoked_subcommand is not None and not tmux.is_inside_tmux():
-        _fail(_TMUX_REQUIRED_MESSAGE)
+        # Desktop-led exception: `init`/`duo` form the binding themselves, and
+        # once a saved default context names a team this external session IS a
+        # member (the anchor-pane duo) — its commands resolve against the tmux
+        # server without the caller living inside tmux.
+        if ctx.invoked_subcommand not in ("init", "duo") and not hive_context.load_current_context().get("team"):
+            _fail(_TMUX_REQUIRED_MESSAGE)
 
 
 def _gc_dead_teams() -> None:
@@ -776,6 +795,135 @@ def _gc_dead_teams() -> None:
         hive_context.clear_current_context()
 
 
+@contextmanager
+def _cursor_claim_lock(cursor_path: Path):
+    """Exclusive lock guarding a collect cursor's read-claim-advance step.
+
+    The lock file is a sibling ``.lock`` so it never collides with the cursor
+    value itself; a lock failure degrades to no-lock rather than dropping mail.
+    """
+    import fcntl
+
+    lock_path = cursor_path.with_suffix(".lock")
+    try:
+        fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR, 0o600)
+    except OSError:
+        yield  # best-effort: never let a lock problem swallow delivery
+        return
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX)
+        yield
+    finally:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        finally:
+            os.close(fd)
+
+
+def _claim_context_session(session_id: str) -> bool:
+    """Whether *session_id* owns the saved member identity, claiming it if free.
+
+    A desktop member's identity lives in the shared default context, but every
+    host session runs the same inbox hook. First claim wins: the session that
+    formed the duo is the one working right after, so it claims before an
+    idle sibling can. `hive duo init` clears the claim, so re-forming from
+    another session moves the identity rather than deadlocking it.
+    """
+    ctx = hive_context.load_current_context()
+    owner = ctx.get("session", "")
+    if owner and owner != session_id:
+        return False
+    if not owner:
+        hive_context.save_current_context(
+            team=ctx.get("team", ""),
+            workspace=ctx.get("workspace", ""),
+            agent=ctx.get("agent", ""),
+            session=session_id,
+        )
+    return True
+
+
+@cli.command("collect")
+@click.option(
+    "--wait",
+    "wait_seconds",
+    type=click.IntRange(0, 3600),
+    default=0,
+    help="Block up to N seconds for new inbound messages (0 = return immediately)",
+)
+@click.option(
+    "--if-awaiting",
+    is_flag=True,
+    help="With --wait: only block while one of my own messages is unanswered",
+)
+@click.option(
+    "--session",
+    "session_id",
+    default="",
+    help="Host session id; drains only when this session owns the member identity",
+)
+def collect_cmd(wait_seconds: int, if_awaiting: bool, session_id: str):
+    """Drain this member's inbound messages from the bus (blocking inbox).
+
+    For members whose session cannot receive channel push — a desktop-led
+    worker outside tmux, whose messages arrive through the plugin's Stop hook
+    instead. Returns inbound messages newer than the last collect and advances
+    a durable cursor.
+
+    \b
+    Examples:
+      hive collect                            # drain unread now
+      hive collect --wait 120 --if-awaiting   # block only while a reply is owed
+    """
+    if session_id and not _claim_context_session(session_id):
+        # Another host session owns this member identity. Draining here would
+        # deliver its mail into the wrong conversation, so stay silent.
+        click.echo(json.dumps({"messages": [], "count": 0, "notMine": True}, indent=2))
+        return
+    team_name, t = _resolve_scoped_team(None, required=True)
+    assert t is not None
+    me = _resolve_sender(None)
+    ws = _resolve_workspace(t, required=True)
+    if (
+        wait_seconds
+        and if_awaiting
+        and not bus.is_awaiting_reply(ws, sender=me, within_seconds=wait_seconds)
+    ):
+        # Only hold the turn open when a reply is genuinely owed and recent
+        # (the handoff-then-wait case); a stale trailing send never re-arms.
+        wait_seconds = 0
+    cursor_path = Path(ws) / "state" / f"collect-cursor-{me}"
+    cursor_path.parent.mkdir(parents=True, exist_ok=True)
+    deadline = time.monotonic() + wait_seconds
+    rows: list[tuple[int, dict[str, object]]] = []
+    while True:
+        # Read cursor → claim rows → advance cursor is one atomic step under an
+        # exclusive lock: two hooks firing on parallel tool calls otherwise read
+        # the same cursor and each return the same messages (duplicate delivery).
+        # The lock wraps only the claim, never the wait — a collect blocking for
+        # a reply must not freeze a sibling drain.
+        with _cursor_claim_lock(cursor_path):
+            try:
+                cursor = int(cursor_path.read_text().strip())
+            except (OSError, ValueError):
+                cursor = 0
+            rows = bus.read_inbound_after(ws, recipient=me, after_seq=cursor)
+            if rows:
+                cursor_path.write_text(str(rows[-1][0]))
+        if rows or time.monotonic() >= deadline:
+            break
+        time.sleep(0.5)
+    payload: dict[str, object] = {
+        "agent": me,
+        "team": team_name,
+        "messages": [event for _seq, event in rows],
+        "count": len(rows),
+    }
+    if not rows and wait_seconds:
+        payload["timedOut"] = True
+    click.echo(json.dumps(payload, indent=2, ensure_ascii=False))
+
+
 _FORK_MIN_COLS = 80
 _FORK_MIN_ROWS = 20
 
@@ -1398,10 +1546,16 @@ def _require_codex_daemon_backed(pane: str) -> None:
     )
 
 
-def _run_duo_init(validator_cli: str | None) -> None:
+def _run_duo_init(validator_cli: str | None, channel: str = "") -> None:
     """Shared callback body for the equivalent `hive init` / `hive duo init`."""
     if not tmux.is_inside_tmux():
-        _fail("hive init requires a tmux session. Run `tmux new-session` or `tmux attach` first, then rerun.")
+        # Desktop-led duo: this (external) Claude session is the worker,
+        # reachable over its channel socket; hive owns a detached session
+        # for the anchor + validator panes.
+        sock = channel or _discover_client_channel_socket()
+        result = _create_ccd_duo(channel_socket=sock, validator_cli=validator_cli)
+        click.echo(json.dumps(result, indent=2, ensure_ascii=False))
+        return
     current_pane = tmux.get_current_pane_id() or ""
     if not current_pane:
         _fail("cannot determine current pane")
@@ -1420,14 +1574,20 @@ def _run_duo_init(validator_cli: str | None) -> None:
     default=None,
     help="CLI for validator (default: anti-family of current pane's CLI)",
 )
-def init_cmd(validator_cli: str | None):
+@click.option(
+    "--channel",
+    default="",
+    help="Outside tmux only: this Claude session's channel socket "
+    "(hive-client-<pid>.sock, printed in the hive-channel MCP instructions)",
+)
+def init_cmd(validator_cli: str | None, channel: str):
     """Initialize a duo in this window (equivalent to `hive duo init`).
 
     Worker = this pane, validator = anti-family spawn. Team name and
     workspace derive from the final window. Idempotent: re-running in a
     bound window reports the existing binding.
     """
-    _run_duo_init(validator_cli)
+    _run_duo_init(validator_cli, channel=channel)
 
 
 @cli.command("register")
@@ -2144,8 +2304,11 @@ def team_cmd():
     """
     _gc_dead_teams()
     discovered = _discover_tmux_binding()
-    if discovered.get("team"):
-        _, t = _resolve_scoped_team(str(discovered.get("team")), required=False)
+    team_name = discovered.get("team")
+    if not team_name and not tmux.is_inside_tmux():
+        team_name = hive_context.load_current_context().get("team") or None
+    if team_name:
+        _, t = _resolve_scoped_team(str(team_name), required=False)
         if t is not None:
             click.echo(json.dumps(_team_status_payload(t), indent=2, ensure_ascii=False))
             return
@@ -2523,6 +2686,7 @@ def _spawn_duo_validator(
     cli: str,
     model: str,
     pane_count_after: int,
+    allow_outside_tmux: bool = False,
 ) -> Agent:
     """Spawn a duo validator beside *worker_pane* and wire its identity.
 
@@ -2544,6 +2708,7 @@ def _spawn_duo_validator(
         skill="none",
         prompt=_role_bootstrap_prompt("duo-validator"),
         workspace=ws,
+        allow_outside_tmux=allow_outside_tmux,
     )
     t.agents["validator"] = validator_agent
     tmux.set_pane_option(validator_agent.pane_id, "hive-group", "duo")
@@ -2812,6 +2977,232 @@ def _create_standalone_duo(
     return result
 
 
+_CCD_SESSION_NAME = "hive-ccd"
+
+# Channel push never reaches a session the host launched without `--channels`
+# (the desktop app owns its argv). The plugin's Stop hook delivers instead, so
+# the member is told what to expect rather than handed a polling command.
+_REMOTE_INBOX_NOTE = (
+    "delivered automatically at the end of each turn by the hive plugin's Stop hook"
+)
+
+
+def _probe_unix_socket(path: str) -> bool:
+    """True when something is listening on *path* (connect succeeds)."""
+    import socket as _socket
+
+    s = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+    s.settimeout(1.0)
+    try:
+        s.connect(path)
+        return True
+    except OSError:
+        return False
+    finally:
+        s.close()
+
+
+def _discover_client_channel_socket() -> str:
+    """The single live ``hive-client-*.sock``, or fail with guidance.
+
+    Liveness is a real connect probe — a socket file surviving a kill -9'd
+    session would otherwise shadow the live one forever. Ambiguity (several
+    live desktop sessions) requires an explicit ``--channel``.
+    """
+    from .adapters import claude_channel
+
+    channel_dir = claude_channel.channel_socket_path("").parent
+    candidates = [
+        str(sock)
+        for sock in sorted(channel_dir.glob("hive-client-*.sock"))
+        if _probe_unix_socket(str(sock))
+    ]
+    if len(candidates) == 1:
+        return candidates[0]
+    if not candidates:
+        _fail(
+            "no live client channel socket found; pass --channel <socket> "
+            "(the path is in this session's hive-channel MCP instructions), "
+            "or run inside a tmux session"
+        )
+    _fail(
+        "multiple live client channel sockets found; pass --channel <socket>:\n  "
+        + "\n  ".join(candidates)
+    )
+    raise AssertionError("unreachable")
+
+
+def _existing_ccd_duo(channel_socket: str) -> dict[str, object] | None:
+    """Idempotency for the desktop-led path: the saved default context names a
+    live team whose anchor is a remote-channel member. A stale or divergent
+    symlink (the desktop session restarted and its pid-keyed socket changed)
+    is healed by re-linking the anchor to *this* socket — the team and its
+    validator survive the desktop session's restarts."""
+    ctx = hive_context.load_current_context()
+    team_name = ctx.get("team", "")
+    if not team_name:
+        return None
+    try:
+        t = Team.load(team_name)
+    except (FileNotFoundError, ValueError):
+        return None
+    worker = t.agents.get("worker")
+    if worker is None or tmux.get_pane_option(worker.pane_id, "hive-remote") != "channel":
+        return None
+    from .adapters import claude_channel
+
+    relinked = False
+    link = claude_channel.channel_socket_path(worker.pane_id)
+    try:
+        current_target = os.readlink(link)
+    except OSError:
+        current_target = ""
+    if current_target != channel_socket:
+        if claude_channel.link_client_socket(worker.pane_id, channel_socket) is not None:
+            return None  # this socket is unusable; fall through to a fresh form (which fails loudly)
+        relinked = True
+    validator = t.agents.get("validator")
+    result: dict[str, object] = {
+        "team": t.name,
+        "window": t.tmux_window,
+        "group": "duo",
+        "worker": {"pane": worker.pane_id, "name": "worker", "cli": "claude", "remote": "channel"},
+        "validator": (
+            {"pane": validator.pane_id, "name": "validator", "cli": validator.cli}
+            if validator
+            else None
+        ),
+        "watch": f"tmux attach -t {_CCD_SESSION_NAME}",
+        "inbox": _REMOTE_INBOX_NOTE,
+    }
+    if relinked:
+        result["relinked"] = True
+    return result
+
+
+def _create_ccd_duo(*, channel_socket: str, validator_cli: str | None) -> dict[str, object]:
+    """Desktop-led duo: an external Claude session (outside tmux) is the worker.
+
+    The worker's tmux presence is an *anchor pane* — a plain shell in a
+    detached ``hive-ccd`` session carrying the member tags plus
+    ``@hive-remote=channel``, its channel socket/marker symlinked to the
+    external session's ``hive-client-<pid>.sock``. Everything downstream
+    (identity, routing, delivery, reaping, doctor) keeps working on pane
+    authority; the validator is a normal spawned pane beside the anchor.
+    """
+    from .adapters import claude_channel
+
+    _gc_dead_teams()
+    plugin_manager.cleanup_retired_plugins()
+
+    existing = _existing_ccd_duo(channel_socket)
+    if existing is not None:
+        return existing
+
+    cwd = os.getcwd()
+    if not tmux.has_session(_CCD_SESSION_NAME):
+        tmux.new_session(_CCD_SESSION_NAME)
+    window_name = _duo_window_name(cwd)
+    window, anchor_pane = tmux.new_window(_CCD_SESSION_NAME, name=window_name, cwd=cwd)
+    if not window or not anchor_pane:
+        _fail(f"could not create a window in tmux session '{_CCD_SESSION_NAME}'")
+    try:
+        window_id = tmux.get_window_id(window) or ""
+        final_index = window.rsplit(":", 1)[-1] if ":" in window else "0"
+        team_name = _default_team_name_for_window(_CCD_SESSION_NAME, window_id, final_index)
+        _claim_team_name(team_name, this_window=window, explicit=False)
+
+        ws_path = _default_auto_workspace_path(_CCD_SESSION_NAME, window_id, final_index)
+        from .sidecar import stop_sidecar
+
+        stop_sidecar(str(ws_path))
+        bus.reset_workspace(ws_path)
+
+        t = Team.create_for_window(
+            team_name,
+            window_target=window,
+            lead_pane_id=anchor_pane,
+            lead_name="worker",
+            description=f"desktop-led duo ({cwd})",
+            workspace=str(ws_path),
+            tag_lead=False,
+            allow_outside_tmux=True,
+        )
+
+        tmux.rename_window(window, _unique_duo_window_name(window_name, window))
+        tmux.configure_hive_window(window)
+        tmux.set_pane_title(anchor_pane, "[worker]")
+        tmux.set_pane_option(anchor_pane, "hive-role", "agent")
+        tmux.set_pane_option(anchor_pane, "hive-agent", "worker")
+        tmux.set_pane_option(anchor_pane, "hive-team", team_name)
+        tmux.set_pane_option(anchor_pane, "hive-group", "duo")
+        tmux.set_pane_option(anchor_pane, "hive-cli", "claude")
+        tmux.set_pane_option(anchor_pane, "hive-remote", "channel")
+
+        err = claude_channel.link_client_socket(anchor_pane, channel_socket)
+        if err:
+            raise RuntimeError(f"cannot link channel socket: {err}")
+
+        v_cli, v_model = _resolve_validator_cli_model("anthropic", validator_cli)
+        validator_agent = _spawn_duo_validator(
+            t,
+            window=window,
+            worker_pane=anchor_pane,
+            worker_cwd=cwd,
+            ws=str(ws_path),
+            cli=v_cli,
+            model=v_model,
+            pane_count_after=2,
+            allow_outside_tmux=True,
+        )
+    except Exception as e:  # noqa: BLE001 — undo the half-built window
+        tmux.kill_window(window)
+        _fail(str(e))
+        raise AssertionError("unreachable")
+
+    # Identity is written only once the duo actually exists: a failed link or
+    # spawn above must not leave the saved default context pointing at a
+    # window that was just torn down (it poisons `hive collect` and the next
+    # init's idempotency check).
+    hive_context.save_context_for_pane(
+        anchor_pane, team=team_name, workspace=str(ws_path), agent="worker"
+    )
+    _remember_context(team=team_name, workspace=str(ws_path), agent="worker")
+
+    try:
+        reloaded = Team.load(team_name, prefer_pane=anchor_pane)
+        reloaded.set_peer("worker", "validator")
+    except (FileNotFoundError, KeyError, ValueError):
+        pass
+
+    from . import layout as layout_mod
+
+    layout_mod.apply_adaptive(window)
+    # The anchor pane is plumbing, not a view: fill the window with the
+    # validator (prefix-z restores the split) and land attaches on the duo
+    # window instead of the session's empty window 0.
+    tmux.zoom_pane(validator_agent.pane_id)
+    tmux.select_window(window)
+    _ensure_team_sidecar(t, ws_path)
+
+    return {
+        "team": team_name,
+        "window": window,
+        "group": "duo",
+        "worker": {"pane": anchor_pane, "name": "worker", "cli": "claude", "remote": "channel"},
+        "validator": {
+            "pane": validator_agent.pane_id,
+            "name": "validator",
+            "cli": v_cli,
+            "mode": "spawned",
+        },
+        "dispatched": ["validator"],
+        "watch": f"tmux attach -t {_CCD_SESSION_NAME}",
+        "inbox": _REMOTE_INBOX_NOTE,
+        "next": "hive skills get duo-worker",
+    }
+
+
 @duo_cmd.command("init")
 @click.option(
     "--validator-cli",
@@ -2819,7 +3210,13 @@ def _create_standalone_duo(
     default=None,
     help="CLI for validator (default: anti-family of current pane's CLI)",
 )
-def duo_init_cmd(validator_cli: str | None):
+@click.option(
+    "--channel",
+    default="",
+    help="Outside tmux only: this Claude session's channel socket "
+    "(hive-client-<pid>.sock, printed in the hive-channel MCP instructions)",
+)
+def duo_init_cmd(validator_cli: str | None, channel: str):
     """Set up a duo from the current pane: worker (=this pane) + anti-family validator.
 
     Equivalent to `hive init`. The current pane must be running
@@ -2833,8 +3230,13 @@ def duo_init_cmd(validator_cli: str | None):
 
     The validator runs the anti-family CLI (claude↔codex) so review stays
     independent.
+
+    Outside tmux (Claude Code desktop) this forms a desktop-led duo instead:
+    this session becomes the worker over its channel socket, the validator
+    spawns in a background detached tmux session (`tmux attach -t hive-ccd`
+    to watch).
     """
-    _run_duo_init(validator_cli)
+    _run_duo_init(validator_cli, channel=channel)
 
 
 # Replaces the bare index token in a window-status format with a conditional
@@ -3263,6 +3665,17 @@ def _resume_members_into_live_team(
         for m in _resume_member_order(missing):
             count += 1
             name = str(m["name"])
+            if m.get("remote"):
+                # The member is an external session (a desktop-led worker)
+                # reached through its anchor pane's channel link. There is no
+                # CLI here to restore, and spawning one would put a look-alike
+                # agent on the team's routing key. That session reconnects
+                # itself with `hive duo init --channel <its socket>`.
+                _resume_progress(
+                    f"skipping {name} — it runs outside tmux; reconnect it with "
+                    "`hive duo init --channel <socket>` from that session"
+                )
+                continue
             fresh = name in fresh_members
             if fresh:
                 _resume_progress(
diff --git a/src/hive/context.py b/src/hive/context.py
index 5116ed6..e3392fa 100644
--- a/src/hive/context.py
+++ b/src/hive/context.py
@@ -44,7 +44,9 @@ def load_current_context() -> dict[str, str]:
     return {str(k): str(v) for k, v in dict(data).items() if v}
 
 
-def save_current_context(*, team: str = "", workspace: str = "", agent: str = "") -> Path:
+def save_current_context(
+    *, team: str = "", workspace: str = "", agent: str = "", session: str = ""
+) -> Path:
     path = _context_file()
     path.parent.mkdir(parents=True, exist_ok=True)
     payload = {
@@ -52,6 +54,11 @@ def save_current_context(*, team: str = "", workspace: str = "", agent: str = ""
         "workspace": workspace,
         "agent": agent,
     }
+    if session:
+        # Which host session owns this identity. Only the desktop path sets it,
+        # to keep sibling sessions running the same inbox hook from draining a
+        # member's inbox that isn't theirs.
+        payload["session"] = session
     path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n")
     return path
 
diff --git a/src/hive/resume.py b/src/hive/resume.py
index de0c1e6..d6868d2 100644
--- a/src/hive/resume.py
+++ b/src/hive/resume.py
@@ -25,7 +25,7 @@
 _HANDLE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
 _PREV_SUFFIX = ".prev"
 
-_MEMBER_FIELDS = ("name", "cli", "model", "sessionId", "cwd")
+_MEMBER_FIELDS = ("name", "cli", "model", "sessionId", "cwd", "remote")
 
 
 def repo_label(cwd: str) -> str:
@@ -232,7 +232,7 @@ def merge_members(
             continue
         entry = by_name.setdefault(name, {field: "" for field in _MEMBER_FIELDS})
         entry["name"] = name
-        for field in ("cli", "model", "sessionId", "cwd"):
+        for field in ("cli", "model", "sessionId", "cwd", "remote"):
             value = str(obs.get(field, "") or "")
             if value:
                 entry[field] = value
diff --git a/src/hive/sidecar.py b/src/hive/sidecar.py
index 324f084..e00f5f0 100644
--- a/src/hive/sidecar.py
+++ b/src/hive/sidecar.py
@@ -1903,6 +1903,10 @@ def _write_resume_snapshot(workspace: str, team: str) -> None:
             "model": model or agent.model,
             "sessionId": session_id,
             "cwd": agent.cwd,
+            # An anchored member's process lives outside tmux, so its pane
+            # carries no session to resume; recording the marker keeps resume
+            # from spawning a look-alike CLI in its place.
+            "remote": tmux.get_pane_option(agent.pane_id, "hive-remote") or "",
         })
 
     existing = resume_store.load_snapshot(t.name)
diff --git a/src/hive/team.py b/src/hive/team.py
index 4407bee..4f11c84 100644
--- a/src/hive/team.py
+++ b/src/hive/team.py
@@ -66,6 +66,7 @@ def create_for_window(
         cwd: str = "",
         workspace: str = "",
         tag_lead: bool = True,
+        allow_outside_tmux: bool = False,
     ) -> Team:
         """Create a team bound to *window_target* (not necessarily the focused
         window).
@@ -76,9 +77,17 @@ def create_for_window(
         window explicitly so callers can break out first, then bind the team
         where the pane actually landed — team identity must follow the final
         window (Bug A).
+
+        *allow_outside_tmux* lets a caller outside tmux (a desktop-led duo
+        binding a detached session's window) create the team; the gate then
+        demands positive evidence that *window_target* exists on the server
+        instead of the caller's own location.
         """
         if not tmux.is_inside_tmux():
-            raise ValueError(_TMUX_REQUIRED_MESSAGE)
+            if not allow_outside_tmux:
+                raise ValueError(_TMUX_REQUIRED_MESSAGE)
+            if not window_target or not tmux.get_window_id(window_target):
+                raise ValueError(f"tmux window '{window_target}' not found")
         from .resume import is_archive_handle
 
         if is_archive_handle(name):
@@ -308,11 +317,18 @@ def status(self) -> dict:
                     row["peer"] = peer_name
             members.append(row)
         for name in sorted(self.agents):
+            pane_id = self.agents[name].pane_id
             row = {
                 "name": name,
                 "role": "agent",
-                "pane": self.agents[name].pane_id,
+                "pane": pane_id,
             }
+            # An anchored member hosts no CLI on its pane by design, so its
+            # `cliAlive: false` is normal rather than a dead agent — say so in
+            # the payload instead of leaving readers to guess.
+            remote = tmux.get_pane_option(pane_id, "hive-remote") or ""
+            if remote:
+                row["remote"] = remote
             group = self.member_groups.get(name, "")
             if group:
                 row["group"] = group
diff --git a/src/hive/tmux.py b/src/hive/tmux.py
index bad48cc..86050a8 100644
--- a/src/hive/tmux.py
+++ b/src/hive/tmux.py
@@ -574,6 +574,14 @@ def resize_pane(pane_id: str, width: str | None = None, height: str | None = Non
     _run(args, check=False)
 
 
+def zoom_pane(pane_id: str) -> None:
+    """Zoom *pane_id* to fill its window (idempotent: only toggles when not
+    already zoomed). Unzoom is the human's `prefix z`."""
+    r = _run(["display", "-t", pane_id, "-p", "#{window_zoomed_flag}"], check=False)
+    if r.stdout.strip() != "1":
+        _run(["resize-pane", "-Z", "-t", pane_id], check=False)
+
+
 def list_panes(target: str) -> list[str]:
     """List all pane ids in a window/session."""
     r = _run(["list-panes", "-t", target, "-F", "#{pane_id}"], check=False)
diff --git a/tests/cli/test_ccd_duo_init.py b/tests/cli/test_ccd_duo_init.py
new file mode 100644
index 0000000..13dbbaa
--- /dev/null
+++ b/tests/cli/test_ccd_duo_init.py
@@ -0,0 +1,245 @@
+"""Desktop-led duo formation: `hive duo init --channel` outside tmux.
+
+The external Claude session (Claude Code desktop) becomes the worker; its tmux
+presence is an anchor pane whose channel socket/marker are symlinks to the
+session's ``hive-client-<pid>.sock``. These tests are hermetic: tmux is the
+conftest fake, the validator spawn is stubbed, and the client socket is a
+plain file (``link_client_socket`` only checks existence + marker version —
+real listening-socket delivery is covered in tests/unit/test_claude_channel.py).
+"""
+import json
+import os
+import socket
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from hive.cli import cli
+from hive import context as hive_context
+from hive.adapters import claude_channel
+
+
+def _client_socket_files(hive_home: Path, name: str = "hive-client-777") -> Path:
+    channel = hive_home / "channel"
+    channel.mkdir(parents=True, exist_ok=True)
+    sock = channel / f"{name}.sock"
+    sock.touch()
+    (channel / f"{name}.ready").write_text("2")
+    return sock
+
+
+@pytest.fixture
+def ccd_tmux(monkeypatch):
+    """Stub the desktop-led path's tmux mutations.
+
+    Returned as a callable to invoke AFTER ``configure_hive_home(...)``:
+    ``hive.cli.tmux`` and ``hive.team.tmux`` are the same module object, so
+    these patches must land last to win over the conftest defaults
+    (``has_session`` → True in particular).
+    """
+
+    def _apply() -> dict[str, object]:
+        calls = _patch(monkeypatch)
+        return calls
+
+    return _apply
+
+
+def _patch(monkeypatch) -> dict[str, object]:
+    calls: dict[str, object] = {"pane_options": [], "killed": []}
+    monkeypatch.setattr("hive.cli.tmux.has_session", lambda _name: False)
+    monkeypatch.setattr(
+        "hive.cli.tmux.new_session", lambda name: calls.__setitem__("new_session", name) or "%49"
+    )
+    monkeypatch.setattr(
+        "hive.cli.tmux.new_window",
+        lambda session, **kw: calls.__setitem__("new_window", (session, kw)) or ("hive-ccd:1", "%50"),
+    )
+    monkeypatch.setattr("hive.cli.tmux.configure_hive_window", lambda _t: None)
+    monkeypatch.setattr("hive.cli.tmux.set_pane_title", lambda *_a: None)
+    monkeypatch.setattr(
+        "hive.cli.tmux.set_pane_option",
+        lambda pane, key, value: calls["pane_options"].append((pane, key, value)),
+    )
+    monkeypatch.setattr("hive.cli.tmux.kill_window", lambda target: calls["killed"].append(target))
+    monkeypatch.setattr("hive.cli.tmux.zoom_pane", lambda pane: calls.__setitem__("zoomed", pane))
+    monkeypatch.setattr("hive.cli.tmux.select_window", lambda w: calls.__setitem__("selected", w))
+    monkeypatch.setattr("hive.layout.apply_adaptive", lambda _w: None)
+    monkeypatch.setattr("hive.sidecar.stop_sidecar", lambda _ws: None)
+    monkeypatch.setattr("hive.cli.bus.reset_workspace", lambda _ws: None)
+    monkeypatch.setattr(
+        "hive.cli._spawn_duo_validator",
+        lambda t, **kw: calls.__setitem__("validator_kw", kw) or SimpleNamespace(pane_id="%51"),
+    )
+    return calls
+
+
+def test_duo_init_outside_tmux_forms_desktop_led_duo(
+    runner, configure_hive_home, ccd_tmux, monkeypatch
+):
+    hive_home = configure_hive_home(tmux_inside=False)
+    calls = ccd_tmux()
+    sock = _client_socket_files(hive_home)
+
+    result = runner.invoke(cli, ["duo", "init", "--channel", str(sock)])
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["worker"] == {
+        "pane": "%50",
+        "name": "worker",
+        "cli": "claude",
+        "remote": "channel",
+    }
+    assert payload["validator"]["pane"] == "%51"
+    assert payload["watch"] == "tmux attach -t hive-ccd"
+    assert calls["new_session"] == "hive-ccd"
+    assert calls["killed"] == []
+
+    # Anchor pane carries member tags plus the remote marker.
+    opts = {(k, v) for _p, k, v in calls["pane_options"]}
+    assert ("hive-agent", "worker") in opts
+    assert ("hive-cli", "claude") in opts
+    assert ("hive-remote", "channel") in opts
+
+    # Channel socket + marker are symlinks to the client's files.
+    assert os.readlink(claude_channel.channel_socket_path("%50")) == str(sock)
+    assert claude_channel.marker_version("%50") == "2"
+
+    # The default context is the external session's outbound identity.
+    ctx = hive_context.load_current_context()
+    assert ctx["team"] == payload["team"]
+    assert ctx["agent"] == "worker"
+
+    # Validator spawns beside the anchor with the outside-tmux opt-in, and the
+    # anti-family choice is derived from the anthropic *family*, not the CLI
+    # name (a claude-led desktop duo gets a codex validator).
+    assert calls["validator_kw"]["worker_pane"] == "%50"
+    assert calls["validator_kw"]["allow_outside_tmux"] is True
+    assert calls["validator_kw"]["cli"] == "codex"
+
+    # Viewport: the validator fills the window (anchor is plumbing, not a
+    # view) and an attach lands on the duo window, not the empty window 0.
+    assert calls["zoomed"] == "%51"
+    assert calls["selected"] == "hive-ccd:1"
+
+
+def test_duo_init_outside_tmux_dead_channel_socket_undoes_window(
+    runner, configure_hive_home, ccd_tmux
+):
+    hive_home = configure_hive_home(tmux_inside=False)
+    calls = ccd_tmux()
+    missing = hive_home / "channel" / "hive-client-404.sock"
+
+    result = runner.invoke(cli, ["duo", "init", "--channel", str(missing)])
+
+    assert result.exit_code != 0
+    assert "does not exist" in result.output
+    assert calls["killed"] == ["hive-ccd:1"]  # no half-built window left
+    # And no dangling identity: a failed form must not poison the saved
+    # context (it would break `hive collect` and the next init's idempotency).
+    assert hive_context.load_current_context().get("team", "") == ""
+
+
+def test_send_outside_tmux_resolves_saved_context(runner, configure_hive_home, monkeypatch):
+    """The root gate + team resolution fall back to the saved default context,
+    so a desktop-led worker can `hive send` without being inside tmux."""
+    configure_hive_home(tmux_inside=False)
+    hive_context.save_current_context(team="hive-ccd-w1", workspace="/tmp/ws", agent="worker")
+    seen: dict[str, object] = {}
+
+    def _fake_resolve(team, required=True):
+        seen["team"] = team
+        raise SystemExit(3)  # stop before any tmux resolution
+
+    monkeypatch.setattr("hive.cli._resolve_scoped_team", _fake_resolve)
+
+    result = runner.invoke(cli, ["send", "validator", "hello"])
+
+    # The command got past the root tmux gate and asked for the context team.
+    assert result.exit_code == 3
+    assert seen["team"] is None or seen["team"] == "hive-ccd-w1"
+
+
+def test_duo_init_outside_tmux_discovers_single_live_socket(
+    runner, configure_hive_home, ccd_tmux, monkeypatch
+):
+    """With no --channel, a single live (connect-probed) client socket is used;
+    corpse socket files without a listener are ignored."""
+    configure_hive_home(tmux_inside=False)
+    ccd_tmux()
+    short_home = Path(tempfile.mkdtemp(prefix="hh", dir="/tmp"))
+    monkeypatch.setenv("HIVE_HOME", str(short_home))
+    channel = short_home / "channel"
+    channel.mkdir(parents=True)
+    (channel / "hive-client-1.sock").touch()  # corpse: file, no listener
+    live = channel / "hive-client-2.sock"
+    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    srv.bind(str(live))
+    srv.listen(1)
+    (channel / "hive-client-2.ready").write_text("2")
+    try:
+        result = runner.invoke(cli, ["duo", "init"])
+        assert result.exit_code == 0, result.output
+        assert os.readlink(claude_channel.channel_socket_path("%50")) == str(live)
+    finally:
+        srv.close()
+        import shutil
+
+        shutil.rmtree(short_home, ignore_errors=True)
+
+
+def test_existing_ccd_duo_relinks_stale_socket(configure_hive_home, monkeypatch, tmp_path):
+    """A desktop session restart changes its pid-keyed socket; re-running init
+    heals the anchor's symlink and keeps the team instead of forming a new one."""
+    hive_home = configure_hive_home(tmux_inside=False)
+    hive_context.save_current_context(team="hive-ccd-w1", workspace="/tmp/ws", agent="worker")
+    worker = SimpleNamespace(pane_id="%50", cli="claude")
+    team = SimpleNamespace(
+        name="hive-ccd-w1", tmux_window="hive-ccd:1", agents={"worker": worker}
+    )
+    monkeypatch.setattr("hive.cli.Team", SimpleNamespace(load=lambda name, **kw: team))
+    monkeypatch.setattr(
+        "hive.cli.tmux.get_pane_option",
+        lambda _p, k: "channel" if k == "hive-remote" else None,
+    )
+    channel = hive_home / "channel"
+    channel.mkdir(parents=True, exist_ok=True)
+    stale = channel / "hive-client-1.sock"
+    stale.touch()
+    fresh = channel / "hive-client-2.sock"
+    fresh.touch()
+    (channel / "hive-client-2.ready").write_text("2")
+    link = claude_channel.channel_socket_path("%50")
+    os.symlink(stale, link)
+
+    from hive.cli import _existing_ccd_duo
+
+    res = _existing_ccd_duo(str(fresh))
+
+    assert res is not None and res.get("relinked") is True
+    assert os.readlink(link) == str(fresh)
+    assert claude_channel.marker_version("%50") == "2"
+
+
+def test_team_status_marks_the_anchored_member(configure_hive_home, monkeypatch):
+    """`cliAlive: false` on an anchor pane is by design, so the payload says
+    so — a reader must not have to guess whether the member is dead."""
+    configure_hive_home()
+    from hive.agent import Agent
+    from hive.team import Team
+
+    monkeypatch.setattr(
+        "hive.team.tmux.get_pane_option",
+        lambda pane, key: "channel" if (pane == "%50" and key == "hive-remote") else None,
+    )
+    team = Team(name="hive-ccd-w1", tmux_session="hive-ccd", tmux_window="hive-ccd:1")
+    team.agents["worker"] = Agent(name="worker", team_name=team.name, pane_id="%50", cli="claude")
+    team.agents["validator"] = Agent(name="validator", team_name=team.name, pane_id="%51", cli="codex")
+
+    members = {m["name"]: m for m in team.status()["members"]}
+
+    assert members["worker"]["remote"] == "channel"
+    assert "remote" not in members["validator"]
diff --git a/tests/cli/test_collect_command.py b/tests/cli/test_collect_command.py
new file mode 100644
index 0000000..b2ed120
--- /dev/null
+++ b/tests/cli/test_collect_command.py
@@ -0,0 +1,201 @@
+"""`hive collect` — the blocking inbox for members without channel push."""
+import json
+from pathlib import Path
+
+from hive import bus
+from hive import context as hive_context
+from hive.cli import cli
+
+
+def _fake_clock():
+    """Monotonic clock that jumps past any deadline after two reads."""
+    ticks = iter([0.0, 1.0] + [10_000.0] * 50)
+    return lambda: next(ticks)
+
+
+def _bind_desktop_worker(configure_hive_home, monkeypatch, tmp_path):
+    configure_hive_home(tmux_inside=False)
+    ws = tmp_path / "ws"
+    bus.init_workspace(ws)
+    hive_context.save_current_context(team="hive-ccd-w1", workspace=str(ws), agent="worker")
+    from types import SimpleNamespace
+
+    team = SimpleNamespace(name="hive-ccd-w1", workspace=str(ws))
+    monkeypatch.setattr(
+        "hive.cli._resolve_scoped_team", lambda _t, required=True: ("hive-ccd-w1", team)
+    )
+    return ws
+
+
+def test_collect_drains_unread_and_advances_cursor(
+    runner, configure_hive_home, monkeypatch, tmp_path
+):
+    ws = _bind_desktop_worker(configure_hive_home, monkeypatch, tmp_path)
+    bus.write_send_event(ws, from_agent="validator", to_agent="worker", body="verdict one")
+    bus.write_send_event(ws, from_agent="worker", to_agent="validator", body="not mine")
+    bus.write_send_event(ws, from_agent="validator", to_agent="worker", body="verdict two")
+
+    result = runner.invoke(cli, ["collect"])
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["count"] == 2
+    assert [m["body"] for m in payload["messages"]] == ["verdict one", "verdict two"]
+    assert all(m["from"] == "validator" for m in payload["messages"])
+
+    # Second drain: cursor advanced, nothing new, timedOut marks a bounded wait.
+    result2 = runner.invoke(cli, ["collect", "--wait", "1"])
+    payload2 = json.loads(result2.output)
+    assert payload2["count"] == 0
+    assert payload2["timedOut"] is True
+
+    # A new message after the cursor is picked up.
+    bus.write_send_event(ws, from_agent="validator", to_agent="worker", body="verdict three")
+    payload3 = json.loads(runner.invoke(cli, ["collect"]).output)
+    assert [m["body"] for m in payload3["messages"]] == ["verdict three"]
+
+
+def test_collect_immediate_empty_has_no_timeout_flag(
+    runner, configure_hive_home, monkeypatch, tmp_path
+):
+    _bind_desktop_worker(configure_hive_home, monkeypatch, tmp_path)
+
+    payload = json.loads(runner.invoke(cli, ["collect"]).output)
+
+    assert payload == {"agent": "worker", "team": "hive-ccd-w1", "messages": [], "count": 0}
+
+
+def test_collect_if_awaiting_arms_only_while_a_reply_is_owed(
+    runner, configure_hive_home, monkeypatch, tmp_path
+):
+    """--if-awaiting blocks only while the member's latest move is an outbound
+    still owed a reply; an idle session returns at once."""
+    ws = _bind_desktop_worker(configure_hive_home, monkeypatch, tmp_path)
+    slept: list[float] = []
+    monkeypatch.setattr("hive.cli.time.sleep", lambda s: slept.append(s))
+
+    # Nothing sent yet → nothing owed → returns at once despite --wait.
+    payload = json.loads(runner.invoke(cli, ["collect", "--wait", "30", "--if-awaiting"]).output)
+    assert payload["count"] == 0
+    assert slept == []
+
+    # Latest move is worker's own unanswered send → the wait is armed.
+    bus.write_send_event(ws, from_agent="worker", to_agent="validator", body="please review")
+    monkeypatch.setattr("hive.cli.time.monotonic", _fake_clock())
+    payload = json.loads(runner.invoke(cli, ["collect", "--wait", "30", "--if-awaiting"]).output)
+    assert payload["timedOut"] is True
+    assert slept  # it actually blocked
+
+
+def _sent_at(ws) -> float:
+    """Wall-clock 'now' matching the timestamp bus just wrote (second precision)."""
+    import datetime as dt
+
+    return dt.datetime.now(dt.timezone.utc).timestamp()
+
+
+def test_is_awaiting_reply_structure_and_recency(tmp_path):
+    """Truth table for the gate: structure (latest move is my unanswered send)
+    AND recency (that send is newer than the window). Both required."""
+    ws = tmp_path / "ws"
+    bus.init_workspace(ws)
+    t0 = _sent_at(ws)
+
+    # No history → not awaiting.
+    assert bus.is_awaiting_reply(ws, sender="worker", within_seconds=120, now=t0) is False
+
+    # Own send, unanswered, fresh → awaiting.
+    a = bus.write_send_event(ws, from_agent="worker", to_agent="validator", body="review this")
+    assert bus.is_awaiting_reply(ws, sender="worker", within_seconds=120, now=t0) is True
+
+    # Same send, but now stale (queried far in the future) → NOT awaiting.
+    # This is the bug fix: a trailing send never re-arms later unrelated turns.
+    assert bus.is_awaiting_reply(ws, sender="worker", within_seconds=120, now=t0 + 600) is False
+
+    # Peer replies → latest move is inbound → worker no longer awaits; validator does.
+    bus.write_send_event(ws, from_agent="validator", to_agent="worker", body="ok", reply_to=a.msg_id)
+    assert bus.is_awaiting_reply(ws, sender="worker", within_seconds=120, now=t0) is False
+    assert bus.is_awaiting_reply(ws, sender="validator", within_seconds=120, now=t0) is True
+
+
+def test_is_awaiting_reply_ignores_a_finished_exchanges_trailing_signoff(tmp_path):
+    """A completed exchange leaves worker's own sign-off as the last unanswered
+    send. Fresh, it may arm once (harmless); stale, it must never re-arm — the
+    original bug was an all-history test that latched True forever."""
+    ws = tmp_path / "ws"
+    bus.init_workspace(ws)
+    t0 = _sent_at(ws)
+    a = bus.write_send_event(ws, from_agent="worker", to_agent="validator", body="review this")
+    b = bus.write_send_event(
+        ws, from_agent="validator", to_agent="worker", body="VAL passed", reply_to=a.msg_id
+    )
+    bus.write_send_event(
+        ws, from_agent="worker", to_agent="validator", body="thanks, shipping", reply_to=b.msg_id
+    )
+
+    # Turns later, the stale sign-off does not drag every future turn into a wait.
+    assert bus.is_awaiting_reply(ws, sender="worker", within_seconds=120, now=t0 + 600) is False
+
+
+def test_collect_session_claim_keeps_siblings_out(runner, configure_hive_home, monkeypatch, tmp_path):
+    """Every desktop session runs the same inbox hook, so the first to claim the
+    member identity owns it; a sibling draining would deliver the member's mail
+    into the wrong conversation."""
+    ws = _bind_desktop_worker(configure_hive_home, monkeypatch, tmp_path)
+    bus.write_send_event(ws, from_agent="validator", to_agent="worker", body="verdict")
+
+    # A sibling session with no claim on file takes it and drains.
+    payload = json.loads(runner.invoke(cli, ["collect", "--session", "sess-A"]).output)
+    assert payload["count"] == 1
+    assert hive_context.load_current_context()["session"] == "sess-A"
+
+    # A different session is refused outright — no drain, no cursor movement.
+    bus.write_send_event(ws, from_agent="validator", to_agent="worker", body="second")
+    other = json.loads(runner.invoke(cli, ["collect", "--session", "sess-B"]).output)
+    assert other == {"messages": [], "count": 0, "notMine": True}
+
+    # The owner still gets it.
+    mine = json.loads(runner.invoke(cli, ["collect", "--session", "sess-A"]).output)
+    assert [m["body"] for m in mine["messages"]] == ["second"]
+
+
+def test_collect_cursor_lock_prevents_duplicate_drain(tmp_path, monkeypatch):
+    """Two collects racing on the same cursor must not both return the same
+    message. The claim (read cursor → read rows → advance cursor) is locked, so
+    the loser sees the advanced cursor and drains nothing."""
+    import threading
+
+    ws = tmp_path / "ws"
+    bus.init_workspace(ws)
+    (ws / "state").mkdir(parents=True, exist_ok=True)
+    for i in range(5):
+        bus.write_send_event(ws, from_agent="validator", to_agent="worker", body=f"m{i}")
+
+    from hive import cli as cli_mod
+
+    cursor = Path(ws) / "state" / "collect-cursor-worker"
+    drained: list[list[int]] = []
+    barrier = threading.Barrier(2)
+
+    def claim():
+        barrier.wait()  # maximize overlap
+        with cli_mod._cursor_claim_lock(cursor):
+            try:
+                c = int(cursor.read_text().strip())
+            except (OSError, ValueError):
+                c = 0
+            rows = bus.read_inbound_after(ws, recipient="worker", after_seq=c)
+            if rows:
+                cursor.write_text(str(rows[-1][0]))
+            drained.append([s for s, _ in rows])
+
+    ts = [threading.Thread(target=claim) for _ in range(2)]
+    for t in ts:
+        t.start()
+    for t in ts:
+        t.join()
+
+    all_seqs = [s for batch in drained for s in batch]
+    assert sorted(all_seqs) == sorted(set(all_seqs))  # no seq delivered twice
+    winner = [b for b in drained if b]
+    assert len(winner) == 1 and len(winner[0]) == 5  # one drains all, other empty
diff --git a/tests/cli/test_current_init_use.py b/tests/cli/test_current_init_use.py
index 4808c80..bca2dd4 100644
--- a/tests/cli/test_current_init_use.py
+++ b/tests/cli/test_current_init_use.py
@@ -620,12 +620,15 @@ def test_init_idempotent_rerun_from_bound_worker_pane(runner, configure_hive_hom
     assert breaks == []                   # no break-out
 
 
-def test_init_fails_outside_tmux(runner, configure_hive_home, monkeypatch):
+def test_init_outside_tmux_without_channel_socket_fails(runner, configure_hive_home, monkeypatch):
+    # Outside tmux, init now targets the desktop-led duo path; with no live
+    # client channel socket it must fail with guidance instead of forming one.
     configure_hive_home(tmux_inside=False)
     monkeypatch.setattr("hive.cli.tmux.is_inside_tmux", lambda: False)
 
     result = runner.invoke(cli, ["init"])
     assert result.exit_code != 0
+    assert "channel socket" in result.output
     assert "tmux" in result.output.lower()
 
 
diff --git a/tests/cli/test_message_commands.py b/tests/cli/test_message_commands.py
index ece41d8..239fcaf 100644
--- a/tests/cli/test_message_commands.py
+++ b/tests/cli/test_message_commands.py
@@ -405,8 +405,10 @@ def test_reply_rejects_legacy_msg_option_with_positional_hint(runner, configure_
     assert called == []
 
 
-def test_send_requires_tmux(runner, monkeypatch):
-    monkeypatch.setattr("hive.cli.tmux.is_inside_tmux", lambda: False)
+def test_send_requires_tmux(runner, configure_hive_home, monkeypatch):
+    # Hermetic HIVE_HOME: the root gate now consults the saved default context
+    # outside tmux (desktop-led duo), so the real ~/.hive must stay unread.
+    configure_hive_home(tmux_inside=False)
 
     result = runner.invoke(cli, ["send", "gpt", "hello from current context"])
 
@@ -954,8 +956,10 @@ def test_notify_uses_current_pane_by_default(runner, monkeypatch):
     }
 
 
-def test_notify_fails_outside_tmux(runner, monkeypatch):
-    monkeypatch.setattr("hive.cli.tmux.is_inside_tmux", lambda: False)
+def test_notify_fails_outside_tmux(runner, configure_hive_home, monkeypatch):
+    # Hermetic HIVE_HOME: the root gate now consults the saved default context
+    # outside tmux (desktop-led duo), so the real ~/.hive must stay unread.
+    configure_hive_home(tmux_inside=False)
     monkeypatch.setattr("hive.cli.tmux.get_current_pane_id", lambda: "")
 
     result = runner.invoke(cli, ["notify", "需要确认"])
diff --git a/tests/cli/test_resume_command.py b/tests/cli/test_resume_command.py
index 1eb4d81..2d7b052 100644
--- a/tests/cli/test_resume_command.py
+++ b/tests/cli/test_resume_command.py
@@ -1056,3 +1056,37 @@ def boom(_target):
     assert rec.killed_windows == []    # live window survives
     # snapshot is kept for retry
     assert resume_store.load_snapshot("0-w2") is not None
+
+
+def test_resume_skips_an_anchored_member(monkeypatch, configure_hive_home, tmp_path):
+    """A member whose agent lives outside tmux has no session to restore;
+    spawning a look-alike CLI would put a fake agent on its routing key."""
+    configure_hive_home()
+    from types import SimpleNamespace
+
+    from hive import cli as cli_mod
+
+    spawned: list[str] = []
+    monkeypatch.setattr(
+        cli_mod.Agent, "spawn",
+        lambda **kw: spawned.append(kw["name"]) or SimpleNamespace(pane_id="%99"),
+    )
+    progress: list[str] = []
+    monkeypatch.setattr(cli_mod, "_resume_progress", lambda m: progress.append(m))
+    monkeypatch.setattr(cli_mod.tmux, "select_window", lambda _w: None)
+    monkeypatch.setattr(cli_mod.tmux, "set_pane_option", lambda *_a: None)
+    monkeypatch.setattr("hive.layout.apply_adaptive", lambda _w: None)
+    peer_team = SimpleNamespace(peer_map={}, agents={}, set_peer=lambda a, b: None, save=lambda: None)
+    monkeypatch.setattr(cli_mod.Team, "load", classmethod(lambda cls, name, **kw: peer_team))
+
+    missing = [{"name": "worker", "cli": "claude", "cwd": "/tmp", "sessionId": "", "remote": "channel"}]
+    result = cli_mod._resume_members_into_live_team(
+        {"team": "hive-ccd-w1", "members": missing},
+        {"window": "hive-ccd:1", "workspace": str(tmp_path / "ws")},
+        {"validator": SimpleNamespace(pane_id="%51")},
+        missing,
+    )
+
+    assert spawned == []  # never respawned as a CLI
+    assert result["members"] == []
+    assert any("outside tmux" in m for m in progress)
diff --git a/tests/unit/test_claude_channel.py b/tests/unit/test_claude_channel.py
index 8533a65..81be92b 100644
--- a/tests/unit/test_claude_channel.py
+++ b/tests/unit/test_claude_channel.py
@@ -536,7 +536,10 @@ def test_server_bind_failure_writes_no_marker(_hive_home):
         proc.wait(timeout=5)
 
 
-def test_server_without_tmux_pane_skips_socket(_hive_home):
+def test_server_without_tmux_pane_binds_client_socket(_hive_home):
+    """Outside tmux (desktop session) the server binds a pid-keyed client
+    socket and self-reports its path in the MCP instructions — the agent is
+    the bridge that hands it to `hive duo init --channel`."""
     env = {**os.environ, "HIVE_HOME": str(_hive_home),
            "PYTHONPATH": str(Path(__file__).resolve().parents[2] / "src")}
     env.pop("TMUX_PANE", None)
@@ -545,13 +548,20 @@ def test_server_without_tmux_pane_skips_socket(_hive_home):
         stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
         text=True, bufsize=1, env=env,
     )
+    sock = _hive_home / "channel" / f"hive-client-{proc.pid}.sock"
+    marker = _hive_home / "channel" / f"hive-client-{proc.pid}.ready"
     try:
         proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize",
                                      "params": {}}) + "\n")
         proc.stdin.flush()
         resp = json.loads(proc.stdout.readline())
-        assert resp["id"] == 1  # handshake still works
-        assert not (_hive_home / "channel").exists()  # no socket dir created
+        assert resp["id"] == 1
+        instructions = resp["result"]["instructions"]
+        assert str(sock) in instructions
+        assert "--channel" in instructions
+        assert _wait_path(sock)
+        assert _wait_path(marker)
+        assert marker.read_text().strip() == "2"
     finally:
         proc.terminate()
         proc.wait(timeout=5)
@@ -677,3 +687,100 @@ def _flaky_replace(src, dst):
 
     srv._maybe_publish_marker(sock_path)  # retry succeeds
     assert marker.read_text() == srv.MARKER_RECEIPT_CAPABLE
+
+
+# --- desktop-led anchor pane linkage ----------------------------------------
+
+
+def _client_server_proc(hive_home: Path) -> subprocess.Popen:
+    """A channel server with no TMUX_PANE: binds hive-client-<pid>.sock."""
+    env = {**os.environ, "HIVE_HOME": str(hive_home),
+           "PYTHONPATH": str(Path(__file__).resolve().parents[2] / "src")}
+    env.pop("TMUX_PANE", None)
+    return subprocess.Popen(
+        [sys.executable, "-m", "hive.adapters.claude_channel_server"],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        text=True, bufsize=1, env=env,
+    )
+
+
+def _initialize(proc: subprocess.Popen) -> dict:
+    proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize",
+                                 "params": {}}) + "\n")
+    proc.stdin.flush()
+    return json.loads(proc.stdout.readline())
+
+
+def test_link_client_socket_refuses_missing_socket(_hive_home):
+    err = cc.link_client_socket("%42", _hive_home / "channel" / "hive-client-1.sock")
+    assert err is not None and "does not exist" in err
+
+
+def test_link_client_socket_refuses_unknown_marker(_hive_home):
+    channel = _hive_home / "channel"
+    channel.mkdir(parents=True)
+    sock = channel / "hive-client-1.sock"
+    sock.touch()
+    (channel / "hive-client-1.ready").write_text("9")
+    err = cc.link_client_socket("%42", sock)
+    assert err is not None and "unknown version" in err
+
+
+def test_link_client_socket_links_sock_and_marker(_hive_home):
+    channel = _hive_home / "channel"
+    channel.mkdir(parents=True)
+    sock = channel / "hive-client-1.sock"
+    sock.touch()
+    (channel / "hive-client-1.ready").write_text("2")
+
+    assert cc.link_client_socket("%42", sock) is None
+
+    assert os.readlink(cc.channel_socket_path("%42")) == str(sock)
+    assert cc.marker_version("%42") == "2"
+    # Relinking (e.g. a fresh desktop session) replaces, never fails.
+    assert cc.link_client_socket("%42", sock) is None
+
+
+def test_send_to_pane_delivers_through_client_symlink(_hive_home):
+    """The desktop-led duo's inbound path end to end at the transport level:
+    a client-mode server + anchor-pane symlink + pane-addressed send lands as
+    a channel notification with a local MCP-write receipt."""
+    proc = _client_server_proc(_hive_home)
+    try:
+        assert _initialize(proc)["id"] == 1
+        sock = _hive_home / "channel" / f"hive-client-{proc.pid}.sock"
+        marker = _hive_home / "channel" / f"hive-client-{proc.pid}.ready"
+        assert _wait_path(sock) and _wait_path(marker)
+        assert cc.link_client_socket("%42", sock) is None
+
+        accepted = cc.send_to_pane("%42", "<HIVE msgId=m1>hello</HIVE>")
+
+        assert accepted == cc.ACCEPTED_MCP_WRITE
+        note = json.loads(proc.stdout.readline())
+        assert note["method"] == "notifications/claude/channel"
+        assert note["params"]["content"] == "<HIVE msgId=m1>hello</HIVE>"
+        assert note["params"]["meta"] == {"msg_id": "m1"}
+    finally:
+        proc.terminate()
+        proc.wait(timeout=5)
+
+
+def test_send_to_pane_fails_closed_when_client_exits(_hive_home):
+    """Server exit unlinks its real socket+marker; the anchor's symlinks
+    dangle and pane-addressed sends fail closed like any dead pane server."""
+    proc = _client_server_proc(_hive_home)
+    try:
+        assert _initialize(proc)["id"] == 1
+        sock = _hive_home / "channel" / f"hive-client-{proc.pid}.sock"
+        marker = _hive_home / "channel" / f"hive-client-{proc.pid}.ready"
+        assert _wait_path(sock) and _wait_path(marker)
+        assert cc.link_client_socket("%42", sock) is None
+    finally:
+        proc.terminate()
+        proc.wait(timeout=5)
+
+    deadline = time.time() + 5
+    while time.time() < deadline and marker.exists():
+        time.sleep(0.05)
+    assert not marker.exists()
+    assert cc.send_to_pane("%42", "anyone home?") is None
diff --git a/tests/unit/test_inbox_hook.py b/tests/unit/test_inbox_hook.py
new file mode 100644
index 0000000..c96c7ba
--- /dev/null
+++ b/tests/unit/test_inbox_hook.py
@@ -0,0 +1,131 @@
+"""The plugin Stop hook that delivers hive messages into a push-less session."""
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+_HOOK = (
+    Path(__file__).resolve().parents[2]
+    / "plugins" / "hive" / "scripts" / "inbox_hook.py"
+)
+
+
+def _load():
+    spec = importlib.util.spec_from_file_location("inbox_hook", _HOOK)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _run(monkeypatch, capsys, *, event, collect_results, tmux_pane=""):
+    hook = _load()
+    calls: list[list[str]] = []
+
+    def _fake_collect(args):
+        calls.append(args)
+        return collect_results.pop(0) if collect_results else {}
+
+    monkeypatch.setattr(hook, "_collect", _fake_collect)
+    monkeypatch.setenv("TMUX_PANE", tmux_pane)
+    monkeypatch.setattr("sys.stdin", __import__("io").StringIO(json.dumps(event)))
+    hook.main()
+    return calls, capsys.readouterr().out
+
+
+def test_hook_blocks_the_stop_with_the_inbound_envelope(monkeypatch, capsys):
+    message = {
+        "from": "validator", "to": "worker", "msgId": "18kd",
+        "inReplyTo": "0Aea", "body": "VAL passed",
+    }
+    calls, out = _run(
+        monkeypatch, capsys,
+        event={"stop_hook_active": False},
+        collect_results=[{"messages": [message], "count": 1}],
+    )
+
+    assert calls == [[]]  # drained, no wait needed — a message was already there
+    payload = json.loads(out)
+    assert payload["decision"] == "block"
+    assert "<HIVE from=validator to=worker msgId=18kd reply-to=0Aea>" in payload["reason"]
+    assert "VAL passed" in payload["reason"]
+
+
+def test_hook_waits_only_while_a_reply_is_owed(monkeypatch, capsys):
+    calls, out = _run(
+        monkeypatch, capsys,
+        event={"stop_hook_active": False},
+        collect_results=[{"messages": []}, {"messages": [], "count": 0}],
+    )
+
+    assert calls == [[], ["--wait", "120", "--if-awaiting"]]
+    assert out == ""  # nothing arrived: the turn ends normally
+
+
+def test_hook_never_waits_again_while_already_blocking(monkeypatch, capsys):
+    """stop_hook_active means a previous Stop already blocked — draining stays
+    allowed (new messages still land) but the blocking wait must not repeat."""
+    calls, out = _run(
+        monkeypatch, capsys,
+        event={"stop_hook_active": True},
+        collect_results=[{"messages": []}],
+    )
+
+    assert calls == [[]]
+    assert out == ""
+
+
+def test_hook_is_a_no_op_inside_tmux(monkeypatch, capsys):
+    calls, out = _run(
+        monkeypatch, capsys,
+        event={"stop_hook_active": False},
+        collect_results=[{"messages": [{"body": "x"}]}],
+        tmux_pane="%9",
+    )
+
+    assert calls == []  # native transports own delivery there
+    assert out == ""
+
+
+def test_post_tool_use_injects_mid_turn_without_blocking(monkeypatch, capsys):
+    """The mid-turn path: a message that lands while the member works arrives
+    with the next tool result, as additionalContext — never a block, and never
+    a wait (a turn in progress must not stall on an empty inbox)."""
+    message = {"from": "validator", "to": "worker", "msgId": "18kd", "body": "VAL failed: retry"}
+    calls, out = _run(
+        monkeypatch, capsys,
+        event={"hook_event_name": "PostToolUse", "session_id": "sess-1"},
+        collect_results=[{"messages": [message], "count": 1}],
+    )
+
+    assert calls == [["--session", "sess-1"]]  # scoped drain, no wait args
+    payload = json.loads(out)
+    assert payload["hookSpecificOutput"]["hookEventName"] == "PostToolUse"
+    assert "VAL failed: retry" in payload["hookSpecificOutput"]["additionalContext"]
+    assert "decision" not in payload  # must not block a turn in progress
+
+
+def test_post_tool_use_is_silent_when_inbox_is_empty(monkeypatch, capsys):
+    calls, out = _run(
+        monkeypatch, capsys,
+        event={"hook_event_name": "PostToolUse", "session_id": "sess-1"},
+        collect_results=[{"messages": []}],
+    )
+
+    assert calls == [["--session", "sess-1"]]  # one drain, no blocking wait
+    assert out == ""
+
+
+def test_stop_scopes_its_drain_to_the_session(monkeypatch, capsys):
+    calls, _out = _run(
+        monkeypatch, capsys,
+        event={"hook_event_name": "Stop", "stop_hook_active": False, "session_id": "sess-1"},
+        collect_results=[{"messages": []}, {"messages": []}],
+    )
+
+    assert calls == [
+        ["--session", "sess-1"],
+        ["--session", "sess-1", "--wait", "120", "--if-awaiting"],
+    ]
diff --git a/tests/unit/test_plugin_marketplace_manifests.py b/tests/unit/test_plugin_marketplace_manifests.py
index 874f4e8..e512505 100644
--- a/tests/unit/test_plugin_marketplace_manifests.py
+++ b/tests/unit/test_plugin_marketplace_manifests.py
@@ -128,20 +128,28 @@ def test_hooks_declarations_are_asymmetric_by_design():
 
 
 def test_codex_hook_is_the_claude_hook_without_async():
-    # lockstep contract: strip Claude's `async` and Codex's stdout redirect;
-    # everything else must stay deep-equal so a command/timeout edit on one
-    # side cannot ship without the other
-    claude = _load(HOOKS_FILE)
-    codex = _load(CODEX_HOOKS_FILE)
-    for group in claude["hooks"]["SessionStart"]:
+    # lockstep contract for the shared bootstrap: strip Claude's `async` and
+    # Codex's stdout redirect; everything else must stay deep-equal so a
+    # command/timeout edit on one side cannot ship without the other
+    claude = _load(HOOKS_FILE)["hooks"]["SessionStart"]
+    codex = _load(CODEX_HOOKS_FILE)["hooks"]["SessionStart"]
+    for group in claude:
         for hook in group["hooks"]:
             hook.pop("async", None)
-    for group in codex["hooks"]["SessionStart"]:
+    for group in codex:
         for hook in group["hooks"]:
             hook["command"] = hook["command"].removesuffix(" >/dev/null")
     assert claude == codex
 
 
+def test_inbox_stop_hook_is_claude_only():
+    # The Stop hook exists for sessions channels cannot reach (the desktop
+    # app). Codex members always run in a tmux pane, where the app-server
+    # daemon delivers natively — a Stop hook there would be dead weight.
+    assert "Stop" in _load(HOOKS_FILE)["hooks"]
+    assert "Stop" not in _load(CODEX_HOOKS_FILE)["hooks"]
+
+
 def test_codex_hook_redirects_stdout_only():
     # Codex feeds SessionStart stdout into developer context, so the success
     # summaries must be dropped; stderr stays for the single-line remediation
diff --git a/tests/unit/test_retained_shell_liveness.py b/tests/unit/test_retained_shell_liveness.py
index 0b75e31..0fbf17a 100644
--- a/tests/unit/test_retained_shell_liveness.py
+++ b/tests/unit/test_retained_shell_liveness.py
@@ -338,3 +338,55 @@ def test_resume_hint_colors_command_on_terminals_only(monkeypatch, tmp_path):
     assert plain.exit_code == 0
     assert "\x1b[" not in plain.output
     assert "claude --resume sid-1" in plain.output
+
+
+# --- desktop-led anchor pane: @hive-remote=channel bypasses the process gate --
+
+
+def test_send_to_anchored_member_is_bus_only_not_a_push(monkeypatch):
+    """An anchored member has no push transport: its own inbox hook drains the
+    bus. Delivery must say exactly that — claiming an MCP write here would
+    record a push that never happens (the desktop app strips --channels, so
+    the channel server's receipt proves nothing about the session hearing it)."""
+    monkeypatch.setattr("hive.agent_cli.detect_cli_process_for_pane", lambda _p: None)
+    monkeypatch.setattr(
+        "hive.agent.tmux.get_pane_option",
+        lambda _p, key: "channel" if key == "hive-remote" else None,
+    )
+    monkeypatch.setattr("hive.adapters.claude_channel.remote_member_alive", lambda _p: True)
+    _forbid(monkeypatch, "hive.adapters.claude_channel.send_to_pane",
+            "an anchored member is never pushed to over the channel socket")
+    agent = Agent(name="w", team_name="team-x", pane_id="%9", cli="claude")
+
+    from hive.agent import ACCEPTED_BUS_INBOX
+
+    assert agent.send("hi") == ACCEPTED_BUS_INBOX
+
+
+def test_send_to_anchored_member_fails_closed_when_its_session_is_gone(monkeypatch):
+    """The external session unlinks its socket on exit, dangling the anchor's
+    symlinks — that is the liveness evidence, and it must fail closed."""
+    monkeypatch.setattr("hive.agent_cli.detect_cli_process_for_pane", lambda _p: None)
+    monkeypatch.setattr(
+        "hive.agent.tmux.get_pane_option",
+        lambda _p, key: "channel" if key == "hive-remote" else None,
+    )
+    monkeypatch.setattr("hive.adapters.claude_channel.remote_member_alive", lambda _p: False)
+    agent = Agent(name="w", team_name="team-x", pane_id="%9", cli="claude")
+
+    from hive.agent import DeliveryError
+
+    with pytest.raises(DeliveryError, match="stays on the bus"):
+        agent.send("hi")
+
+
+def test_send_to_plain_retained_shell_still_refuses(monkeypatch):
+    """No remote tag → the original fail-closed contract is untouched."""
+    monkeypatch.setattr("hive.agent_cli.detect_cli_process_for_pane", lambda _p: None)
+    monkeypatch.setattr("hive.agent.tmux.get_pane_option", lambda _p, _k: None)
+    agent = Agent(name="v", team_name="team-x", pane_id="%9", cli="claude")
+
+    from hive.agent import DeliveryError
+
+    with pytest.raises(DeliveryError, match="cli_exited"):
+        agent.send("hi")