Add Matrix NL-query Q&A surface (W2 step 5)

Read-only natural-language query over the curated nl_query endpoint, answered
in-thread. Two entry points (room-per-purpose model): a dedicated Q&A room
(MATRIX_QUERY_ROOM) where every top-level message is a question, plus the
?/@bot trigger in the intake room as a cross-room convenience. Both routes hit
the same handle_query -> crm_client.nl_query -> POST /api/query/nl; translation
runs on the box's local model, nothing leaves the box, and there is no write
path so no approval gate applies.

Pure logic (trigger parsing, answer rendering) in query.py with offline tests;
async room wiring in bot.py (live-smoke only, per the bot's convention).

Bot-side only, ships on the Spark via git pull + restart. Depends on the
box-side /api/query/nl endpoint, which lands with the v93 s9pk (reminders + W2):
until v93 is installed the Q&A surface 404s, so the bot deploy is staged to
follow that install.

backend/matrix_intake/bot.py

@@ -18,6 +18,7 @@ import email_proposals
 import matrix_io
 import parse
 import proposals
+import query
 import settings
 
 UNCLEAR_HELP = (
@@ -42,6 +43,7 @@ async def main():
     if roster:
         print(f"matrix-intake: team roster loaded ({len(roster)} names)", flush=True)
     review_room = settings.email_review_room()  # CRM-drafted email proposals (empty → feature off)
+    query_room = settings.query_room()  # dedicated read-only Q&A room (empty → use the intake trigger)
     email_threads = {}  # Matrix thread-root event_id -> {id, investor_name, note} for an email proposal
 
     async def handle_intake(room_id, root, text):
@@ -97,6 +99,19 @@ async def main():
         # easy to miss inside a thread (the full card + yes/edit/no stay in the thread).
         await nudge(room_id, proposals.summary_line(proposal), root)
 
+    async def handle_query(room_id, root, question):
+        """A read-only NL question ('@bot …' / '?…') — translate + run it on the BOX (local Qwen,
+        nothing leaves the box) and post the answer in a thread. No write path, no approval gate:
+        it only reads curated, parameterized queries. The endpoint returns its structured result
+        even on a soft no-match / model-down, so we render that; a transport/auth failure raises
+        and we show a brief error."""
+        try:
+            result = await asyncio.to_thread(crm_client.nl_query, question)
+        except Exception as exc:
+            await say(room_id, f"⚠️ couldn't run that query: {str(exc)[:200]}", root)
+            return
+        await say(room_id, query.render_answer(result), root)
+
     async def handle_reply(room_id, root, text):
         # Claim the proposal synchronously — BEFORE any await — so a second reply that
         # arrives while a commit is in flight can't double-process it. asyncio is
@@ -299,6 +314,12 @@ async def main():
             if root and root in email_threads:
                 await handle_email_reply(room.room_id, root, text)
             return
+        # Dedicated Q&A room: every top-level message IS a question — no trigger needed. Threaded
+        # messages (the answers we post, or follow-ups) aren't acted on in v1.
+        if query_room and room.room_id == query_room:
+            if not root:
+                await handle_query(room.room_id, event.event_id, text)
+            return
         if room.room_id != intake_room:
             return
         if root and store.has(root):
@@ -306,7 +327,15 @@ async def main():
         elif root:
             return  # threaded message not tied to a live proposal — ignore
         else:
-            await handle_intake(room.room_id, event.event_id, text)
+            # A top-level message is either an NL question (explicitly addressed with '?'/'@bot')
+            # or an intake note. The trigger is required, so plain notes still flow to intake.
+            q = query.parse_trigger(text)
+            if q is None:
+                await handle_intake(room.room_id, event.event_id, text)
+            elif not q:
+                await say(room.room_id, query.HELP, event.event_id)
+            else:
+                await handle_query(room.room_id, event.event_id, q)
 
     # Prime the sync token past history, THEN register the callback — only react to messages
     # arriving after startup (no backlog replay). (matrix-bridge pattern.)
@@ -325,6 +354,14 @@ async def main():
             print(f"matrix-intake: could not join review room {review_room}: {exc}", flush=True)
         tasks.append(asyncio.create_task(poll_email_proposals()))
         print(f"matrix-intake: reviewing email proposals in room {review_room} (every {EMAIL_POLL_SEC}s)", flush=True)
+    if query_room:
+        # Read-only Q&A room — just join and listen (no poll task; questions are interactive).
+        # "Invited" isn't "joined": the bot must join before it can post answers (idempotent).
+        try:
+            await client.join(query_room)
+        except Exception as exc:
+            print(f"matrix-intake: could not join Q&A room {query_room}: {exc}", flush=True)
+        print(f"matrix-intake: answering questions in room {query_room}", flush=True)
     try:
         await asyncio.gather(*tasks)
     finally: