Add headless "ask" mode: ?-prefixed message runs claude -p, answer posted back

A message starting with `?` in a mapped room runs `claude -p` one-shot in that
repo on the Mac and posts the full answer back into the room — Matrix as a
request/response interface, not just a trigger. Non-`?` messages keep launching
interactive sessions as before.

New scripts/ask-claude.sh is a login-shell wrapper (so ~/.zprofile puts claude on
PATH) that exports CLAUDE_CODE_OAUTH_TOKEN from the Mac's .env and runs
`claude -p "$prompt" < /dev/null`, printing the answer to stdout. The bot adds a
`?`-dispatch with run_ask/ask: SSH stdout captured, 300s timeout, fail-loud, output
chunked under Matrix's event cap (no truncation).

Headless claude -p needs the long-lived token because a non-GUI SSH session can't
reach the login Keychain (reports "Not logged in") — the deliberate Approach A that
the interactive GUI-Terminal path (D11) avoided. Token is kept Mac-side only; the
Spark never runs claude. Sovereignty unchanged: claude -p uses the subscription, no
frontier API touches message payloads.

Proven live on the Spark; fresh-eyes reviewed before commit.

src/bot.py

@@ -22,6 +22,10 @@ from nio import AsyncClient, MatrixRoom, RoomMessageText
 
 REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
+# Headless "ask" mode tunables.
+ASK_TIMEOUT = 300        # seconds to wait for `claude -p` before giving up
+MAX_MSG_CHARS = 30000    # split answers into chunks well under Matrix's ~64KB event cap
+
 
 def load_env(path):
     env = {}
@@ -39,6 +43,27 @@ def load_config(path):
         return tomllib.load(f)
 
 
+def split_message(text, limit=MAX_MSG_CHARS):
+    """Split text into <=limit-char chunks on newline boundaries (no truncation)."""
+    if len(text) <= limit:
+        return [text]
+    chunks, buf = [], ""
+    for line in text.splitlines(keepends=True):
+        while len(line) > limit:            # one oversized line: hard-split it
+            if buf:
+                chunks.append(buf)
+                buf = ""
+            chunks.append(line[:limit])
+            line = line[limit:]
+        if len(buf) + len(line) > limit:
+            chunks.append(buf)
+            buf = ""
+        buf += line
+    if buf:
+        chunks.append(buf)
+    return chunks
+
+
 async def main():
     env = load_env(os.path.join(REPO_ROOT, ".env"))
     cfg = load_config(os.path.join(REPO_ROOT, "config.toml"))
@@ -52,6 +77,7 @@ async def main():
     all_projects_room = cfg.get("all_projects", {}).get("room_id")
     ssh_alias = os.environ.get("MB_SSH_ALIAS") or cfg["mac"]["ssh_alias"]
     launcher = cfg["mac"]["launcher"]
+    ask_launcher = cfg["mac"].get("ask_launcher")
 
     client = AsyncClient(homeserver, user_id)
     client.restore_login(user_id=user_id, device_id=device_id, access_token=token)
@@ -73,6 +99,28 @@ async def main():
         out, _ = await proc.communicate()
         return proc.returncode, out.decode(errors="replace").strip()
 
+    async def run_ask(repo_dir, prompt):
+        """Run ask-claude.sh on the Mac over SSH; return (rc, stdout, stderr).
+
+        Headless `claude -p`: its stdout is the answer (captured here), stderr is diagnostics.
+        This path never opens a GUI Terminal and is not surfaced to the phone.
+        """
+        remote = f"{shlex.quote(ask_launcher)} {shlex.quote(repo_dir)} {shlex.quote(prompt)}"
+        proc = await asyncio.create_subprocess_exec(
+            "ssh", ssh_alias, remote,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        try:
+            out, err = await asyncio.wait_for(proc.communicate(), timeout=ASK_TIMEOUT)
+        except asyncio.TimeoutError:
+            proc.kill()
+            await proc.wait()           # reap the killed ssh client (no zombie)
+            return None, "", f"timed out after {ASK_TIMEOUT}s"
+        return (proc.returncode,
+                out.decode(errors="replace").strip(),
+                err.decode(errors="replace").strip())
+
     async def say(room_id, text):
         await client.room_send(
             room_id, "m.room.message", {"msgtype": "m.text", "body": text}
@@ -88,6 +136,24 @@ async def main():
                                f"(rc={rc}): {out[:300] or 'no output'}")
         return False
 
+    async def ask(report_room, repo, prompt):
+        """Headless ask: run `claude -p` in the repo and post the full answer back."""
+        if not ask_launcher:
+            await say(report_room,
+                      "⚠️ matrix-bridge: ask mode not configured ([mac].ask_launcher missing).")
+            return
+        await say(report_room, f"🤔 asking claude in {repo['label']}…")
+        rc, out, err = await run_ask(repo["repo_dir"], prompt)
+        if rc == 0:                     # success — even an empty answer is not a failure
+            print(f"ask {repo['label']}: {len(out)} chars", flush=True)
+            for chunk in split_message(out or "(claude returned no output)"):
+                await say(report_room, chunk)
+            return
+        detail = err or out or "no output"
+        print(f"ASK FAILED {repo['label']}: rc={rc} {detail[:300]}", flush=True)
+        await say(report_room, f"⚠️ matrix-bridge: ask failed in {repo['label']} "
+                               f"(rc={rc}): {detail[:500]}")
+
     async def on_message(room: MatrixRoom, event: RoomMessageText):
         if event.sender == user_id:
             return  # never react to our own messages
@@ -95,7 +161,7 @@ async def main():
         if not prompt:
             return
 
-        if room.room_id == all_projects_room:
+        if room.room_id == all_projects_room:   # fan-out room always launches, never asks
             date = datetime.date.today().isoformat()
             print(f"[all-projects] fan-out to {len(rooms)} repos: {prompt!r}", flush=True)
             results = await asyncio.gather(*[
@@ -106,7 +172,11 @@ async def main():
                       f"matrix-bridge: launched {sum(results)}/{len(rooms)} sessions ({date}).")
         elif room.room_id in rooms:
             r = rooms[room.room_id]
-            if await launch_one(room.room_id, r, prompt):
+            if prompt.startswith("?"):                      # headless ask mode
+                ask_prompt = prompt[1:].strip()
+                if ask_prompt:
+                    await ask(room.room_id, r, ask_prompt)
+            elif await launch_one(room.room_id, r, prompt):
                 await say(room.room_id,
                           f"matrix-bridge: launched {r['label']} — drive it on your phone.")