From 0d505cff44d9fdfc02355d97b0c282115936d204 Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 19:44:15 +0800 Subject: [PATCH 01/12] feat: pid-keyed client channel socket for claude sessions outside tmux --- src/hive/adapters/claude_channel.py | 45 ++++++++ src/hive/adapters/claude_channel_server.py | 77 +++++++++----- tests/unit/test_claude_channel.py | 113 ++++++++++++++++++++- 3 files changed, 205 insertions(+), 30 deletions(-) diff --git a/src/hive/adapters/claude_channel.py b/src/hive/adapters/claude_channel.py index 51424fb..21c0e77 100644 --- a/src/hive/adapters/claude_channel.py +++ b/src/hive/adapters/claude_channel.py @@ -282,6 +282,51 @@ 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-.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" + + # --- 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-.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/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-.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", "hello") + + assert accepted == cc.ACCEPTED_MCP_WRITE + note = json.loads(proc.stdout.readline()) + assert note["method"] == "notifications/claude/channel" + assert note["params"]["content"] == "hello" + 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 From 1fe9feb0c5f097d66dc55714f3b6e264e1efc09d Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 19:44:22 +0800 Subject: [PATCH 02/12] feat: desktop-led duo with an anchor-pane worker --- src/hive/agent.py | 37 ++- src/hive/cli.py | 264 ++++++++++++++++++++- src/hive/team.py | 11 +- tests/cli/test_ccd_duo_init.py | 181 ++++++++++++++ tests/cli/test_current_init_use.py | 5 +- tests/cli/test_message_commands.py | 12 +- tests/unit/test_retained_shell_liveness.py | 50 ++++ 7 files changed, 540 insertions(+), 20 deletions(-) create mode 100644 tests/cli/test_ccd_duo_init.py diff --git a/src/hive/agent.py b/src/hive/agent.py index a3d0956..d801610 100644 --- a/src/hive/agent.py +++ b/src/hive/agent.py @@ -206,6 +206,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 +226,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 +265,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 +286,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 +435,13 @@ 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 Claude session (desktop) + # reached through the pane's symlinked channel socket. For that member + # "somebody is listening" evidence is the channel's own fail-closed + # marker + receipt, so delivery skips the process probe and lets the + # transport gate it. probe = None try: from .agent_cli import detect_cli_process_for_pane @@ -430,6 +450,17 @@ 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 + + accepted = claude_channel.send_to_pane(self.pane_id, text) + if accepted is None: + raise DeliveryError( + f"remote channel pane {self.pane_id} transport failed " + "(client socket gone, marker missing, or no receipt); " + "is the external Claude session still running?" + ) + return accepted 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/cli.py b/src/hive/cli.py index 743717a..a0e6591 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -184,11 +184,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 +765,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: @@ -1398,10 +1415,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 +1443,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-.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 +2173,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 +2555,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 +2577,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 +2846,203 @@ def _create_standalone_duo( return result +_CCD_SESSION_NAME = "hive-ccd" + + +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 " + "(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 :\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 socket still links to *this* channel socket.""" + 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 + + link = claude_channel.channel_socket_path(worker.pane_id) + try: + if os.readlink(link) != channel_socket: + return None + except OSError: + return None + validator = t.agents.get("validator") + return { + "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}", + } + + +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-.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") + 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") + + 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") + + 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) + _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}", + "next": "hive skills get duo-worker", + } + + @duo_cmd.command("init") @click.option( "--validator-cli", @@ -2819,7 +3050,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-.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 +3070,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 diff --git a/src/hive/team.py b/src/hive/team.py index 4407bee..8cf1033 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): diff --git a/tests/cli/test_ccd_duo_init.py b/tests/cli/test_ccd_duo_init.py new file mode 100644 index 0000000..73315a8 --- /dev/null +++ b/tests/cli/test_ccd_duo_init.py @@ -0,0 +1,181 @@ +"""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-.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.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" + + +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 + + +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) 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/unit/test_retained_shell_liveness.py b/tests/unit/test_retained_shell_liveness.py index 0b75e31..98b2d75 100644 --- a/tests/unit/test_retained_shell_liveness.py +++ b/tests/unit/test_retained_shell_liveness.py @@ -338,3 +338,53 @@ 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_remote_channel_anchor_uses_channel_transport(monkeypatch): + """An anchor pane hosts no CLI by design — its member lives in an external + Claude session. Delivery routes to the channel transport, whose own + fail-closed marker/receipt machinery is the listening evidence.""" + 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, + ) + sent: dict[str, str] = {} + monkeypatch.setattr( + "hive.adapters.claude_channel.send_to_pane", + lambda pane, text: sent.update(pane=pane, text=text) or "mcpWriteAccepted", + ) + agent = Agent(name="w", team_name="team-x", pane_id="%9", cli="claude") + + assert agent.send("hi") == "mcpWriteAccepted" + assert sent == {"pane": "%9", "text": "hi"} + + +def test_send_to_remote_channel_anchor_fails_closed_on_transport_failure(monkeypatch): + 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.send_to_pane", lambda _p, _t: None) + agent = Agent(name="w", team_name="team-x", pane_id="%9", cli="claude") + + from hive.agent import DeliveryError + + with pytest.raises(DeliveryError, match="external Claude session"): + 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") From d9c4ee318d282c622902088b4432d54118a0d190 Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 19:44:22 +0800 Subject: [PATCH 03/12] fix: resolve a bare codex name to an absolute path before daemon spawn --- src/hive/adapters/codex_app_server.py | 8 ++++++++ 1 file changed, 8 insertions(+) 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(): From e1b0c2eb9e1cab08834a04b8a4f13b6b554cd241 Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 20:24:49 +0800 Subject: [PATCH 04/12] feat: desktop duo viewport shows only the validator --- src/hive/cli.py | 5 +++++ src/hive/tmux.py | 8 ++++++++ tests/cli/test_ccd_duo_init.py | 7 +++++++ 3 files changed, 20 insertions(+) diff --git a/src/hive/cli.py b/src/hive/cli.py index a0e6591..afc6b14 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -3024,6 +3024,11 @@ def _create_ccd_duo(*, channel_socket: str, validator_cli: str | None) -> dict[s 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 { 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 index 73315a8..55691db 100644 --- a/tests/cli/test_ccd_duo_init.py +++ b/tests/cli/test_ccd_duo_init.py @@ -64,6 +64,8 @@ def _patch(monkeypatch) -> dict[str, object]: 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) @@ -118,6 +120,11 @@ def test_duo_init_outside_tmux_forms_desktop_led_duo( 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 From c3309c36fc98f85573d5d8a46dd6c83940e29e46 Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 21:35:58 +0800 Subject: [PATCH 05/12] feat: re-running init heals a restarted desktop session's channel link --- src/hive/cli.py | 20 +++++++++++++++----- tests/cli/test_ccd_duo_init.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 5 deletions(-) diff --git a/src/hive/cli.py b/src/hive/cli.py index afc6b14..5b7e402 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -2896,7 +2896,10 @@ def _discover_client_channel_socket() -> str: 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 socket still links to *this* channel socket.""" + 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: @@ -2910,14 +2913,18 @@ def _existing_ccd_duo(channel_socket: str) -> dict[str, object] | None: return None from .adapters import claude_channel + relinked = False link = claude_channel.channel_socket_path(worker.pane_id) try: - if os.readlink(link) != channel_socket: - return None + current_target = os.readlink(link) except OSError: - return None + 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") - return { + result: dict[str, object] = { "team": t.name, "window": t.tmux_window, "group": "duo", @@ -2929,6 +2936,9 @@ def _existing_ccd_duo(channel_socket: str) -> dict[str, object] | None: ), "watch": f"tmux attach -t {_CCD_SESSION_NAME}", } + if relinked: + result["relinked"] = True + return result def _create_ccd_duo(*, channel_socket: str, validator_cli: str | None) -> dict[str, object]: diff --git a/tests/cli/test_ccd_duo_init.py b/tests/cli/test_ccd_duo_init.py index 55691db..7b9aa54 100644 --- a/tests/cli/test_ccd_duo_init.py +++ b/tests/cli/test_ccd_duo_init.py @@ -186,3 +186,36 @@ def test_duo_init_outside_tmux_discovers_single_live_socket( 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" From 37a26d301c9821c0632f4587eb0db333570ae067 Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 21:45:53 +0800 Subject: [PATCH 06/12] feat: hive collect drains the bus inbox for push-less sessions --- src/hive/bus.py | 26 ++++++++++++++ src/hive/cli.py | 55 +++++++++++++++++++++++++++++ tests/cli/test_collect_command.py | 58 +++++++++++++++++++++++++++++++ 3 files changed, 139 insertions(+) create mode 100644 tests/cli/test_collect_command.py diff --git a/src/hive/bus.py b/src/hive/bus.py index 1f0fc3d..5ea2546 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, *, diff --git a/src/hive/cli.py b/src/hive/cli.py index 5b7e402..c24aacb 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -793,6 +793,57 @@ def _gc_dead_teams() -> None: hive_context.clear_current_context() +@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)", +) +def collect_cmd(wait_seconds: int): + """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. Returns inbound messages newer than the last collect + and advances a durable cursor. Long-poll with --wait from a background + shell: the command returning IS the push signal, so re-arm it after each + drain instead of polling. + + \b + Examples: + hive collect # drain unread now + hive collect --wait 3600 # block until a message arrives (max 1h) + """ + team_name, t = _resolve_scoped_team(None, required=True) + assert t is not None + me = _resolve_sender(None) + ws = _resolve_workspace(t, required=True) + cursor_path = Path(ws) / "state" / f"collect-cursor-{me}" + try: + cursor = int(cursor_path.read_text().strip()) + except (OSError, ValueError): + cursor = 0 + deadline = time.monotonic() + wait_seconds + while True: + rows = bus.read_inbound_after(ws, recipient=me, after_seq=cursor) + 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 rows: + cursor_path.parent.mkdir(parents=True, exist_ok=True) + cursor_path.write_text(str(rows[-1][0])) + elif wait_seconds: + payload["timedOut"] = True + click.echo(json.dumps(payload, indent=2, ensure_ascii=False)) + + _FORK_MIN_COLS = 80 _FORK_MIN_ROWS = 20 @@ -2935,6 +2986,7 @@ def _existing_ccd_duo(channel_socket: str) -> dict[str, object] | None: else None ), "watch": f"tmux attach -t {_CCD_SESSION_NAME}", + "inbox": "hive collect --wait 3600", } if relinked: result["relinked"] = True @@ -3054,6 +3106,9 @@ def _create_ccd_duo(*, channel_socket: str, validator_cli: str | None) -> dict[s }, "dispatched": ["validator"], "watch": f"tmux attach -t {_CCD_SESSION_NAME}", + # Channel push does not reach desktop sessions (research preview gap); + # a long-poll collect from a background shell is the push signal. + "inbox": "hive collect --wait 3600", "next": "hive skills get duo-worker", } diff --git a/tests/cli/test_collect_command.py b/tests/cli/test_collect_command.py new file mode 100644 index 0000000..5ade5a9 --- /dev/null +++ b/tests/cli/test_collect_command.py @@ -0,0 +1,58 @@ +"""`hive collect` — the blocking inbox for members without channel push.""" +import json + +from hive import bus +from hive import context as hive_context +from hive.cli import cli + + +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} From 8c55b16f2206206ab98698bc3eeec63c1a7b000e Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 21:48:24 +0800 Subject: [PATCH 07/12] fix: map collect into the Daily help section --- src/hive/cli.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/hive/cli.py b/src/hive/cli.py index c24aacb..2f7992b 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -34,6 +34,7 @@ "team": "Daily", "send": "Daily", "reply": "Daily", + "collect": "Daily", "notify": "Daily", "compact": "Daily", "skills": "Daily", From b0b8420296ea8da4ce0fccda2114a9210682b8f8 Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 21:52:25 +0800 Subject: [PATCH 08/12] fix: a failed desktop duo form leaves no dangling saved context --- src/hive/cli.py | 13 +++++++++---- tests/cli/test_ccd_duo_init.py | 3 +++ 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/src/hive/cli.py b/src/hive/cli.py index 2f7992b..e3adfcf 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -3052,10 +3052,6 @@ def _create_ccd_duo(*, channel_socket: str, validator_cli: str | None) -> dict[s 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") - 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") err = claude_channel.link_client_socket(anchor_pane, channel_socket) if err: @@ -3078,6 +3074,15 @@ def _create_ccd_duo(*, channel_socket: str, validator_cli: str | None) -> dict[s _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") diff --git a/tests/cli/test_ccd_duo_init.py b/tests/cli/test_ccd_duo_init.py index 7b9aa54..65e496e 100644 --- a/tests/cli/test_ccd_duo_init.py +++ b/tests/cli/test_ccd_duo_init.py @@ -138,6 +138,9 @@ def test_duo_init_outside_tmux_dead_channel_socket_undoes_window( 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): From d34aca5ccbe42bef11533eefa1a8611a7178d10d Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 22:34:41 +0800 Subject: [PATCH 09/12] feat: deliver hive messages into sessions channels cannot reach --- plugins/hive/hooks/hooks.json | 12 +++ plugins/hive/scripts/inbox_hook.py | 93 +++++++++++++++++++ src/hive/bus.py | 27 ++++++ src/hive/cli.py | 20 ++-- tests/cli/test_collect_command.py | 34 +++++++ tests/unit/test_inbox_hook.py | 89 ++++++++++++++++++ .../unit/test_plugin_marketplace_manifests.py | 22 +++-- 7 files changed, 283 insertions(+), 14 deletions(-) create mode 100644 plugins/hive/scripts/inbox_hook.py create mode 100644 tests/unit/test_inbox_hook.py diff --git a/plugins/hive/hooks/hooks.json b/plugins/hive/hooks/hooks.json index 439cfe0..9a7b097 100644 --- a/plugins/hive/hooks/hooks.json +++ b/plugins/hive/hooks/hooks.json @@ -13,6 +13,18 @@ } ] } + ], + "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..fda1aa3 --- /dev/null +++ b/plugins/hive/scripts/inbox_hook.py @@ -0,0 +1,93 @@ +#!/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). Plugin hooks *do* run there, so this Stop hook is +the delivery path for those sessions: it drains the member's bus inbox and +returns the messages as blocked-stop feedback, which Claude Code injects as +context and continues on — the same effect the channel would have had. + +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. +Only a real message ever blocks a stop, and each drain advances a durable +cursor, so the block cannot repeat on the same message. +""" +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"\n{message.get('body', '')}\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 = {} + already_blocking = bool(event.get("stop_hook_active")) + + payload = _collect([]) + 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(["--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": ( + f"{count} hive message{'s' if count > 1 else ''} arrived. Follow the hive " + f"protocol for each (reply with `hive reply `):\n\n{body}" + ), + "systemMessage": f"hive: {count} message{'s' if count > 1 else ''} delivered", + })) + + +if __name__ == "__main__": + main() diff --git a/src/hive/bus.py b/src/hive/bus.py index 5ea2546..b1fbc81 100644 --- a/src/hive/bus.py +++ b/src/hive/bus.py @@ -434,6 +434,33 @@ def latest_unanswered_inbound_send_event( return _row_to_event(row) if row is not None else None +def has_unanswered_outbound(workspace: str | Path, *, sender: str) -> bool: + """True when *sender* sent a message nobody has replied to yet. + + "Am I waiting on a peer right now" — the gate for blocking at the end of a + turn: a member with an open question should wait for the answer, an idle + one should not stall the session. + """ + with _connect(workspace) as conn: + row = conn.execute( + """ + SELECT 1 FROM messages AS sent + WHERE sent.intent = 'send' + AND sent.from_agent = ? + AND sent.msg_id != '' + AND NOT EXISTS ( + SELECT 1 FROM messages AS answer + WHERE answer.intent = 'send' + AND answer.in_reply_to = sent.msg_id + AND answer.to_agent = sent.from_agent + ) + LIMIT 1 + """, + (sender,), + ).fetchone() + return row is not None + + def has_send_reply_to( workspace: str | Path, *, diff --git a/src/hive/cli.py b/src/hive/cli.py index e3adfcf..30a3c47 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -802,24 +802,30 @@ def _gc_dead_teams() -> None: default=0, help="Block up to N seconds for new inbound messages (0 = return immediately)", ) -def collect_cmd(wait_seconds: int): +@click.option( + "--if-awaiting", + is_flag=True, + help="With --wait: only block while one of my own messages is unanswered", +) +def collect_cmd(wait_seconds: int, if_awaiting: bool): """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. Returns inbound messages newer than the last collect - and advances a durable cursor. Long-poll with --wait from a background - shell: the command returning IS the push signal, so re-arm it after each - drain instead of polling. + 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 3600 # block until a message arrives (max 1h) + hive collect # drain unread now + hive collect --wait 120 --if-awaiting # block only while a reply is owed """ 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.has_unanswered_outbound(ws, sender=me): + wait_seconds = 0 cursor_path = Path(ws) / "state" / f"collect-cursor-{me}" try: cursor = int(cursor_path.read_text().strip()) diff --git a/tests/cli/test_collect_command.py b/tests/cli/test_collect_command.py index 5ade5a9..facc473 100644 --- a/tests/cli/test_collect_command.py +++ b/tests/cli/test_collect_command.py @@ -6,6 +6,12 @@ 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" @@ -56,3 +62,31 @@ def test_collect_immediate_empty_has_no_timeout_flag( payload = json.loads(runner.invoke(cli, ["collect"]).output) assert payload == {"agent": "worker", "team": "hive-ccd-w1", "messages": [], "count": 0} + + +def test_collect_if_awaiting_skips_the_wait_when_nothing_is_owed( + runner, configure_hive_home, monkeypatch, tmp_path +): + """The Stop hook ends an idle turn immediately: --if-awaiting only blocks + while one of my own messages is still unanswered.""" + 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 == [] + + # An unanswered outbound → the wait is armed (loop runs until the deadline). + 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 + + # Once the peer answers, nothing is owed again. + reply = bus.write_send_event( + ws, from_agent="validator", to_agent="worker", body="looks good", reply_to="0Aea" + ) + assert reply.msg_id diff --git a/tests/unit/test_inbox_hook.py b/tests/unit/test_inbox_hook.py new file mode 100644 index 0000000..eb98729 --- /dev/null +++ b/tests/unit/test_inbox_hook.py @@ -0,0 +1,89 @@ +"""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 "" 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 == "" 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 From 4c3edac639e7d86a9006fe3d7f5bedabe76913f5 Mon Sep 17 00:00:00 2001 From: dp Date: Mon, 3 Aug 2026 23:02:57 +0800 Subject: [PATCH 10/12] feat: an anchored member is self-describing across status, resume, and docs --- docs/runtime-model.md | 24 ++++++++++++++++++++++ src/hive/cli.py | 24 ++++++++++++++++++---- src/hive/resume.py | 4 ++-- src/hive/sidecar.py | 4 ++++ src/hive/team.py | 9 ++++++++- tests/cli/test_ccd_duo_init.py | 21 ++++++++++++++++++++ tests/cli/test_resume_command.py | 34 ++++++++++++++++++++++++++++++++ 7 files changed, 113 insertions(+), 7 deletions(-) diff --git a/docs/runtime-model.md b/docs/runtime-model.md index 7384f0d..134cce6 100644 --- a/docs/runtime-model.md +++ b/docs/runtime-model.md @@ -133,12 +133,36 @@ 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-.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. +Reachability comes from the transport instead — a dangling symlink (the +external session exited and unlinked its socket) makes delivery fail closed +exactly like a dead pane server. + +Consumers — delivery routes an anchored member to the channel transport +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 `, which relinks the anchor). + ### `inputState` Source: diff --git a/src/hive/cli.py b/src/hive/cli.py index 30a3c47..398cb68 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -2906,6 +2906,13 @@ def _create_standalone_duo( _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).""" @@ -2993,7 +3000,7 @@ def _existing_ccd_duo(channel_socket: str) -> dict[str, object] | None: else None ), "watch": f"tmux attach -t {_CCD_SESSION_NAME}", - "inbox": "hive collect --wait 3600", + "inbox": _REMOTE_INBOX_NOTE, } if relinked: result["relinked"] = True @@ -3118,9 +3125,7 @@ def _create_ccd_duo(*, channel_socket: str, validator_cli: str | None) -> dict[s }, "dispatched": ["validator"], "watch": f"tmux attach -t {_CCD_SESSION_NAME}", - # Channel push does not reach desktop sessions (research preview gap); - # a long-poll collect from a background shell is the push signal. - "inbox": "hive collect --wait 3600", + "inbox": _REMOTE_INBOX_NOTE, "next": "hive skills get duo-worker", } @@ -3587,6 +3592,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 `. + _resume_progress( + f"skipping {name} — it runs outside tmux; reconnect it with " + "`hive duo init --channel ` from that session" + ) + continue fresh = name in fresh_members if fresh: _resume_progress( 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 8cf1033..4f11c84 100644 --- a/src/hive/team.py +++ b/src/hive/team.py @@ -317,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/tests/cli/test_ccd_duo_init.py b/tests/cli/test_ccd_duo_init.py index 65e496e..13dbbaa 100644 --- a/tests/cli/test_ccd_duo_init.py +++ b/tests/cli/test_ccd_duo_init.py @@ -222,3 +222,24 @@ def test_existing_ccd_duo_relinks_stale_socket(configure_hive_home, monkeypatch, 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_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) From 755ad2890f7c02e446f51af50f37aa88016caf04 Mon Sep 17 00:00:00 2001 From: dp Date: Tue, 4 Aug 2026 14:46:09 +0800 Subject: [PATCH 11/12] feat: anchored members get mid-turn delivery and an honest transport story --- docs/runtime-model.md | 32 ++++++++---- plugins/hive/hooks/hooks.json | 11 ++++ plugins/hive/scripts/inbox_hook.py | 59 +++++++++++++++++----- src/hive/adapters/claude_channel.py | 13 +++++ src/hive/agent.py | 29 +++++++---- src/hive/cli.py | 36 ++++++++++++- src/hive/context.py | 9 +++- tests/cli/test_collect_command.py | 22 ++++++++ tests/unit/test_inbox_hook.py | 42 +++++++++++++++ tests/unit/test_retained_shell_liveness.py | 30 ++++++----- 10 files changed, 234 insertions(+), 49 deletions(-) diff --git a/docs/runtime-model.md b/docs/runtime-model.md index 134cce6..385fd40 100644 --- a/docs/runtime-model.md +++ b/docs/runtime-model.md @@ -153,15 +153,29 @@ 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. -Reachability comes from the transport instead — a dangling symlink (the -external session exited and unlinked its socket) makes delivery fail closed -exactly like a dead pane server. - -Consumers — delivery routes an anchored member to the channel transport -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 `, which relinks the anchor). + +**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 `, 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` diff --git a/plugins/hive/hooks/hooks.json b/plugins/hive/hooks/hooks.json index 9a7b097..8453876 100644 --- a/plugins/hive/hooks/hooks.json +++ b/plugins/hive/hooks/hooks.json @@ -14,6 +14,17 @@ ] } ], + "PostToolUse": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/scripts/inbox_hook.py\"", + "timeout": 20 + } + ] + } + ], "Stop": [ { "hooks": [ diff --git a/plugins/hive/scripts/inbox_hook.py b/plugins/hive/scripts/inbox_hook.py index fda1aa3..ff28d1a 100644 --- a/plugins/hive/scripts/inbox_hook.py +++ b/plugins/hive/scripts/inbox_hook.py @@ -5,15 +5,25 @@ ``--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). Plugin hooks *do* run there, so this Stop hook is -the delivery path for those sessions: it drains the member's bus inbox and -returns the messages as blocked-stop feedback, which Claude Code injects as -context and continues on — the same effect the channel would have had. +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. -Only a real message ever blocks a stop, and each drain advances a durable -cursor, so the block cannot repeat on the same message. +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 @@ -59,6 +69,13 @@ def _envelope(message: dict) -> str: return f"{head}>\n{message.get('body', '')}\n" +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 `):\n\n" + ) + + def main() -> None: if os.environ.get("TMUX_PANE"): return # native channel / app-server transports own delivery in tmux @@ -66,13 +83,32 @@ def main() -> None: event = json.load(sys.stdin) except (ValueError, OSError): event = {} - already_blocking = bool(event.get("stop_hook_active")) + 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 - payload = _collect([]) + 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(["--wait", str(WAIT_SECONDS), "--if-awaiting"]) + payload = _collect([*claim, "--wait", str(WAIT_SECONDS), "--if-awaiting"]) messages = payload.get("messages") or [] if not messages: return @@ -81,10 +117,7 @@ def main() -> None: count = len(messages) print(json.dumps({ "decision": "block", - "reason": ( - f"{count} hive message{'s' if count > 1 else ''} arrived. Follow the hive " - f"protocol for each (reply with `hive reply `):\n\n{body}" - ), + "reason": _preamble(count) + body, "systemMessage": f"hive: {count} message{'s' if count > 1 else ''} delivered", })) diff --git a/src/hive/adapters/claude_channel.py b/src/hive/adapters/claude_channel.py index 21c0e77..888564c 100644 --- a/src/hive/adapters/claude_channel.py +++ b/src/hive/adapters/claude_channel.py @@ -327,6 +327,19 @@ def marker_path_for_socket_path(sock_path: str | Path) -> str: 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/agent.py b/src/hive/agent.py index d801610..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.""" @@ -437,11 +443,10 @@ def send(self, text: str) -> str: # 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 Claude session (desktop) - # reached through the pane's symlinked channel socket. For that member - # "somebody is listening" evidence is the channel's own fail-closed - # marker + receipt, so delivery skips the process probe and lets the - # transport gate it. + # 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 @@ -453,14 +458,16 @@ def send(self, text: str) -> str: if tmux.get_pane_option(self.pane_id, "hive-remote") == "channel": from .adapters import claude_channel - accepted = claude_channel.send_to_pane(self.pane_id, text) - if accepted is None: + if not claude_channel.remote_member_alive(self.pane_id): raise DeliveryError( - f"remote channel pane {self.pane_id} transport failed " - "(client socket gone, marker missing, or no receipt); " - "is the external Claude session still running?" + 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" ) - return accepted + # 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/cli.py b/src/hive/cli.py index 398cb68..f0d217f 100644 --- a/src/hive/cli.py +++ b/src/hive/cli.py @@ -794,6 +794,29 @@ def _gc_dead_teams() -> None: hive_context.clear_current_context() +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", @@ -807,7 +830,13 @@ def _gc_dead_teams() -> None: is_flag=True, help="With --wait: only block while one of my own messages is unanswered", ) -def collect_cmd(wait_seconds: int, if_awaiting: bool): +@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 @@ -820,6 +849,11 @@ def collect_cmd(wait_seconds: int, if_awaiting: bool): 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) 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/tests/cli/test_collect_command.py b/tests/cli/test_collect_command.py index facc473..c44030c 100644 --- a/tests/cli/test_collect_command.py +++ b/tests/cli/test_collect_command.py @@ -90,3 +90,25 @@ def test_collect_if_awaiting_skips_the_wait_when_nothing_is_owed( ws, from_agent="validator", to_agent="worker", body="looks good", reply_to="0Aea" ) assert reply.msg_id + + +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"] diff --git a/tests/unit/test_inbox_hook.py b/tests/unit/test_inbox_hook.py index eb98729..c96c7ba 100644 --- a/tests/unit/test_inbox_hook.py +++ b/tests/unit/test_inbox_hook.py @@ -87,3 +87,45 @@ def test_hook_is_a_no_op_inside_tmux(monkeypatch, capsys): 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_retained_shell_liveness.py b/tests/unit/test_retained_shell_liveness.py index 98b2d75..0fbf17a 100644 --- a/tests/unit/test_retained_shell_liveness.py +++ b/tests/unit/test_retained_shell_liveness.py @@ -343,38 +343,40 @@ def test_resume_hint_colors_command_on_terminals_only(monkeypatch, tmp_path): # --- desktop-led anchor pane: @hive-remote=channel bypasses the process gate -- -def test_send_to_remote_channel_anchor_uses_channel_transport(monkeypatch): - """An anchor pane hosts no CLI by design — its member lives in an external - Claude session. Delivery routes to the channel transport, whose own - fail-closed marker/receipt machinery is the listening evidence.""" +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, ) - sent: dict[str, str] = {} - monkeypatch.setattr( - "hive.adapters.claude_channel.send_to_pane", - lambda pane, text: sent.update(pane=pane, text=text) or "mcpWriteAccepted", - ) + 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") - assert agent.send("hi") == "mcpWriteAccepted" - assert sent == {"pane": "%9", "text": "hi"} + from hive.agent import ACCEPTED_BUS_INBOX + + assert agent.send("hi") == ACCEPTED_BUS_INBOX -def test_send_to_remote_channel_anchor_fails_closed_on_transport_failure(monkeypatch): +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.send_to_pane", lambda _p, _t: 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="external Claude session"): + with pytest.raises(DeliveryError, match="stays on the bus"): agent.send("hi") From 7a47222d2e5db0c069518167e797dd4e5e0f300d Mon Sep 17 00:00:00 2001 From: dp Date: Tue, 4 Aug 2026 15:13:28 +0800 Subject: [PATCH 12/12] fix: the awaiting-reply gate clears instead of latching every turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Stop-hook inbox wait armed on "any unanswered send in history", so the trailing unanswered message every finished exchange leaves behind latched it True forever — every later turn then blocked for the full wait window. Judge it from the single latest message instead, and require that send to be recent, so a stale sign-off never re-arms an unrelated turn. Also lock the collect cursor's read-claim-advance so parallel tool-call hooks can't double-deliver. --- src/hive/bus.py | 79 +++++++++++++++++------ src/hive/cli.py | 59 ++++++++++++++--- tests/cli/test_collect_command.py | 103 +++++++++++++++++++++++++++--- 3 files changed, 205 insertions(+), 36 deletions(-) diff --git a/src/hive/bus.py b/src/hive/bus.py index b1fbc81..3b4d1a5 100644 --- a/src/hive/bus.py +++ b/src/hive/bus.py @@ -434,31 +434,74 @@ def latest_unanswered_inbound_send_event( return _row_to_event(row) if row is not None else None -def has_unanswered_outbound(workspace: str | Path, *, sender: str) -> bool: - """True when *sender* sent a message nobody has replied to yet. - - "Am I waiting on a peer right now" — the gate for blocking at the end of a - turn: a member with an open question should wait for the answer, an idle - one should not stall the session. +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 1 FROM messages AS sent - WHERE sent.intent = 'send' - AND sent.from_agent = ? - AND sent.msg_id != '' - AND NOT EXISTS ( - SELECT 1 FROM messages AS answer - WHERE answer.intent = 'send' - AND answer.in_reply_to = sent.msg_id - AND answer.to_agent = sent.from_agent - ) + 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, sender), ).fetchone() - return row is not None + 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( diff --git a/src/hive/cli.py b/src/hive/cli.py index f0d217f..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 @@ -794,6 +795,31 @@ 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. @@ -858,16 +884,32 @@ def collect_cmd(wait_seconds: int, if_awaiting: bool, session_id: str): assert t is not None me = _resolve_sender(None) ws = _resolve_workspace(t, required=True) - if wait_seconds and if_awaiting and not bus.has_unanswered_outbound(ws, sender=me): + 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}" - try: - cursor = int(cursor_path.read_text().strip()) - except (OSError, ValueError): - cursor = 0 + cursor_path.parent.mkdir(parents=True, exist_ok=True) deadline = time.monotonic() + wait_seconds + rows: list[tuple[int, dict[str, object]]] = [] while True: - rows = bus.read_inbound_after(ws, recipient=me, after_seq=cursor) + # 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) @@ -877,10 +919,7 @@ def collect_cmd(wait_seconds: int, if_awaiting: bool, session_id: str): "messages": [event for _seq, event in rows], "count": len(rows), } - if rows: - cursor_path.parent.mkdir(parents=True, exist_ok=True) - cursor_path.write_text(str(rows[-1][0])) - elif wait_seconds: + if not rows and wait_seconds: payload["timedOut"] = True click.echo(json.dumps(payload, indent=2, ensure_ascii=False)) diff --git a/tests/cli/test_collect_command.py b/tests/cli/test_collect_command.py index c44030c..b2ed120 100644 --- a/tests/cli/test_collect_command.py +++ b/tests/cli/test_collect_command.py @@ -1,5 +1,6 @@ """`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 @@ -64,11 +65,11 @@ def test_collect_immediate_empty_has_no_timeout_flag( assert payload == {"agent": "worker", "team": "hive-ccd-w1", "messages": [], "count": 0} -def test_collect_if_awaiting_skips_the_wait_when_nothing_is_owed( +def test_collect_if_awaiting_arms_only_while_a_reply_is_owed( runner, configure_hive_home, monkeypatch, tmp_path ): - """The Stop hook ends an idle turn immediately: --if-awaiting only blocks - while one of my own messages is still unanswered.""" + """--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)) @@ -78,18 +79,62 @@ def test_collect_if_awaiting_skips_the_wait_when_nothing_is_owed( assert payload["count"] == 0 assert slept == [] - # An unanswered outbound → the wait is armed (loop runs until the deadline). + # 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 - # Once the peer answers, nothing is owed again. - reply = bus.write_send_event( - ws, from_agent="validator", to_agent="worker", body="looks good", reply_to="0Aea" + +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 ) - assert reply.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): @@ -112,3 +157,45 @@ def test_collect_session_claim_keeps_siblings_out(runner, configure_hive_home, m # 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