fix(bot-mode): DMs to a Desktop-owned Bot Chat land in the live session instead of being dropped (#100523)

tests/tools/test_bot_mode_dm.py

@@ -422,6 +422,29 @@ def test_delivery_runner_preserves_child_failure_and_unlinks(tmp_path):
     assert not dm_file.exists()
 
 
+def test_delivery_runner_surfaces_live_owner_refusal(tmp_path, capsys):
+    """#100523: the CLI's single-owner lease refusal is a delivery FAILURE the
+    sender can read, not a raw exit-1 with the payload silently gone."""
+    dm_file = tmp_path / "message.txt"
+    dm_file.write_text("hi", encoding="utf-8")
+    child = tmp_path / "owned.py"
+    child.write_text(
+        "import sys\n"
+        "print('Session abc already has a live owner (desktop, pid 1).', file=sys.stderr)\n"
+        "raise SystemExit(1)\n",
+        encoding="utf-8",
+    )
+
+    returncode = bot_mode_dm._run_delivery(
+        [sys.executable, str(child), "-p", "ops"], str(dm_file), stdin_file=False
+    )
+
+    assert returncode == 1
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["reason"] == "target_busy"
+    assert "NOT delivered" in payload["error"]
+
+
 def test_query_file_delivery_closes_stdin_for_initial_attempt_and_retry(
     tmp_path, monkeypatch
 ):

tests/tui_gateway/test_bot_relay_methods.py

@@ -106,6 +106,45 @@ def test_deliver_requires_params(home):
     assert "error" in err
 
 
+def test_deliver_lands_in_live_bot_chat_instead_of_subprocess(home, monkeypatch):
+    """#100523: a Desktop-owned Bot Chat receives the DM as a normal user turn.
+
+    With the target's Bot Chat live in this gateway, the subprocess transport
+    would be fenced out by the single-owner lease and drop the payload. The
+    handler must route through prompt.submit (the composer's choke point) and
+    never spawn the CLI.
+    """
+    spawned = []
+    submitted = []
+    monkeypatch.setattr("subprocess.run", lambda *a, **k: spawned.append(a) or None)
+    monkeypatch.setitem(
+        srv._methods, "prompt.submit", lambda rid, p: submitted.append(p) or srv._ok(rid, {"status": "streaming"})
+    )
+    monkeypatch.setattr(srv, "_profile_home", lambda name: home / "profiles" / name)
+    monkeypatch.setitem(
+        srv._sessions,
+        "live-ops",
+        {"profile_home": str(home / "profiles" / "ops"), "pending_title": "Bot Chat", "history": []},
+    )
+    out = _result(srv._methods["bot_relay.deliver"](1, {"profile": "ops", "message": "ping"}))
+    # queued=True is the invariant: a DM never interrupts a turn in flight.
+    assert submitted == [{"session_id": "live-ops", "text": "ping", "queued": True}]
+    assert not spawned
+    assert "reply" in out
+
+    # A live session titled anything else for the same profile does not qualify:
+    # the subprocess path runs exactly as before.
+    srv._sessions["live-ops"]["pending_title"] = "Scratch"
+    submitted.clear()
+
+    class _Proc:
+        returncode, stdout, stderr = 0, "pong", ""
+
+    monkeypatch.setattr("subprocess.run", lambda *a, **k: spawned.append(a) or _Proc())
+    out = _result(srv._methods["bot_relay.deliver"](2, {"profile": "ops", "message": "ping"}))
+    assert out["reply"] == "pong" and spawned and not submitted
+
+
 def test_reply_roundtrip_and_id_validation(home):
     envelope_id = "c" * 32
     _result(srv._methods["bot_relay.reply"](1, {"id": envelope_id, "reply": "hi"}))

tools/bot_mode_dm.py

@@ -609,6 +609,17 @@ def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int:
                     )
             # Re-emit the transport's streams: stdout is the reply text the
             # completion notification carries back to the sending agent.
+            if proc.returncode != 0 and "already has a live owner" in (proc.stderr or ""):
+                # #100523: the target's Bot Chat is held live by another
+                # surface (Desktop). The turn never ran, so tell the sender
+                # plainly instead of leaking a raw lease error + exit code.
+                who = argv[argv.index("-p") + 1] if "-p" in argv[:-1] else "the teammate"
+                print(json.dumps({
+                    "error": f"Delivery failed: @{who}'s Bot Chat is open on another "
+                             "surface right now, so your message was NOT delivered. Try again later.",
+                    "reason": "target_busy",
+                }))
+                return 1
             if proc.stdout:
                 sys.stdout.write(proc.stdout)
                 sys.stdout.flush()

tui_gateway/methods_bot_relay.py

@@ -109,6 +109,41 @@ def _(rid, params: dict) -> dict:
         if resolved not in known:
             return _err(rid, 4092, f"no profile '{profile}' on this gateway")
 
+        # #100523: when THIS gateway already hosts the target's Bot Chat live
+        # (the Desktop has it open), the subprocess transport is fenced out by
+        # the single-owner lease ("already has a live owner") and the payload
+        # is dropped. Land the DM in the live session as a normal user turn
+        # via prompt.submit instead — same choke point the composer uses, so
+        # role alternation, persistence and streaming all behave as a typed
+        # message would. (Nested per method_ctx rebinding.)
+        def _live_bot_chat_sid(profile_name: str) -> str:
+            from tools.bot_mode_probe import BOT_CHAT_TITLE
+
+            live_home = _profile_home(profile_name)
+            want_home = str(live_home) if live_home is not None else None
+            for live_sid, record in list(_sessions.items()):
+                if not isinstance(record, dict):
+                    continue
+                if (record.get("profile_home") or None) != want_home:
+                    continue
+                key = _session_lookup_key(record, fallback=live_sid)
+                if _session_live_title(record, key) == BOT_CHAT_TITLE:
+                    return live_sid
+            return ""
+
+        live_sid = _live_bot_chat_sid(resolved)
+        if live_sid:
+            # queued=True: a teammate's DM runs as the NEXT turn. It must never
+            # interrupt or steer a turn already in flight (the default busy
+            # mode does); hundreds of arrivals simply queue in arrival order.
+            submitted = _methods["prompt.submit"](rid, {"session_id": live_sid, "text": message, "queued": True})
+            if "error" in submitted:
+                return submitted
+            return _ok(
+                rid,
+                {"reply": f"Delivered into @{resolved}'s open Bot Chat; the reply will appear there."},
+            )
+
         fd, tmp = tempfile.mkstemp(prefix="hermes-relay-dm-", suffix=".txt", text=True)
         try:
             with os.fdopen(fd, "w", encoding="utf-8") as f: