Email-proposal review over Matrix + a bot role (v0.1.0:89)

The email-capture "proposed grid notes" gain two review surfaces:

1. Inline source email — each proposed-note card on the Email Capture page
   gets a "View email" toggle that lazily fetches the existing
   GET /api/email/detail and shows from/to/cc/date/subject + scrollable body,
   so a reviewer can judge the note against the email it was drafted from.

2. CRM->Matrix review bridge — the CRM (box, stdlib, no matrix-nio) can't post
   to Matrix, so the intake bot (Spark) PULLS: GET /api/intake/email-proposals
   returns to_post/open/to_close work-lists; the bot posts a review card
   (metadata + snippet + draft note) to a dedicated review room
   (MATRIX_EMAIL_REVIEW_ROOM) and relays in-thread yes / no / NL-edit
   (POST .../{id}/decide, note revised via local Qwen). Decisions sync both
   ways: web decide -> bot announces + closes the thread; Matrix decide -> the
   web panel's ~25s poll clears the card. State lives CRM-side in the new
   email_proposal_matrix side row (email-integration migration 0003, additive
   + idempotent CREATE TABLE IF NOT EXISTS), so it survives a bot restart.

Adds a 'bot' role (authenticated, never admin; require_bot_or_admin) to gate
the email-proposal endpoints rather than handing the bot full admin — the
principled base for the coming agentic capabilities. Role controls reach;
the draft->approve gate still controls autonomy (a human approves every write).

Deploy split: endpoints + migration + role + frontend ship in the s9pk; the
bot poll loop + review-room handling ship on the Spark. The bot's CRM user
must be flipped member->bot and joined to the review room (one-time).

Tests: backend/test_email_proposal_matrix.py + matrix_intake/test_email_proposals.py
(30/30 suite green, render-smoke green, migration verified twice on a DB copy).

backend/matrix_intake/bot.py

@@ -14,6 +14,7 @@ import asyncio
 from nio import AsyncClient, MatrixRoom, RoomMessageText
 
 import crm_client
+import email_proposals
 import matrix_io
 import parse
 import proposals
@@ -25,6 +26,8 @@ UNCLEAR_HELP = (
     "or a note like `Note for Acme Capital: wants the Q3 deck, follow up next week`."
 )
 
+EMAIL_POLL_SEC = 20  # how often the bot polls the CRM for new/decided email-activity proposals
+
 
 async def main():
     mx = settings.matrix_settings()
@@ -37,6 +40,8 @@ async def main():
     roster = settings.team_roster()  # frames the parse: teammates do outreach, aren't prospects
     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)
+    email_threads = {}  # Matrix thread-root event_id -> {id, investor_name, note} for an email proposal
 
     async def handle_intake(room_id, root, text):
         # A bare yes/no/approve typed in the MAIN timeline (not inside a proposal's thread) is
@@ -157,15 +162,103 @@ async def main():
             store.put(root, proposal)
             await say(room_id, "I didn't catch that.\n\n" + proposals.render_disambiguation(proposal), root)
 
+    async def handle_email_reply(room_id, root, text):
+        """An in-thread reply to a CRM-drafted email-proposal card: yes commits, no dismisses, and
+        anything else is a natural-language revision of the note (re-drafted by local Qwen; the
+        human still approves the revised note, so the draft→approve gate holds)."""
+        item = email_threads.get(root)
+        if item is None:
+            return  # a threaded reply we don't own (or already resolved)
+        decision = email_proposals.interpret(text)
+        if decision == "approve":
+            # Claim before the await (double-approve guard, like the intake commit path).
+            email_threads.pop(root, None)
+            try:
+                await asyncio.to_thread(crm_client.decide_email_proposal, item["id"], "approve", item.get("note"))
+            except Exception as exc:
+                email_threads[root] = item  # restore for retry
+                await say(room_id, f"⚠️ couldn't add it ({str(exc)[:200]}). Reply **yes** to retry, **no** to dismiss.", root)
+                return
+            await say(room_id, f"✅ Added to the grid for **{item.get('investor_name') or 'the investor'}**.", root)
+        elif decision == "reject":
+            email_threads.pop(root, None)
+            try:
+                await asyncio.to_thread(crm_client.decide_email_proposal, item["id"], "dismiss")
+            except Exception as exc:
+                email_threads[root] = item
+                await say(room_id, f"⚠️ couldn't dismiss it ({str(exc)[:200]}). Try again.", root)
+                return
+            await say(room_id, "🗑️ Dismissed — nothing added to the grid.", root)
+        else:
+            try:
+                new_note = await asyncio.to_thread(email_proposals.revise_note, item.get("note") or "", text)
+            except Exception as exc:
+                await say(room_id, f"⚠️ couldn't revise that ({str(exc)[:200]}). Reply **yes** to add as-is, "
+                                   "**no** to dismiss, or rephrase.", root)
+                return
+            if not new_note:
+                await say(room_id, "I didn't catch a change. Reply **yes** to add the note as-is, **no** to "
+                                   "dismiss, or tell me how to change it.", root)
+                return
+            item["note"] = new_note
+            email_threads[root] = item
+            await say(room_id, f"✏️ Updated draft note:\n\n{new_note}\n\nReply **yes** to add it, **no** to "
+                               "dismiss, or refine again.", root)
+
+    async def poll_email_proposals():
+        """Poll the CRM for email-activity proposals: post a review card for each new one, rebuild
+        the reply-routing map from already-posted threads (so replies still route after a restart),
+        and announce+close any decided on the web. One failing cycle logs and retries next tick."""
+        while True:
+            try:
+                lists = await asyncio.to_thread(crm_client.list_email_proposals)
+                for it in lists["open"]:  # rebuild routing for threads posted before (e.g. a restart)
+                    ev = it.get("event_id")
+                    if ev and ev not in email_threads:
+                        email_threads[ev] = {"id": it["id"], "investor_name": it.get("investor_name"),
+                                             "note": it.get("proposed_note") or ""}
+                for it in lists["to_post"]:
+                    try:
+                        resp = await client.room_send(
+                            review_room, "m.room.message",
+                            matrix_io.thread_content(email_proposals.render_card(it), None))
+                        ev = getattr(resp, "event_id", None)
+                        if not ev:
+                            print(f"matrix-intake: card send returned no event_id for {it['id']}", flush=True)
+                            continue
+                        await asyncio.to_thread(crm_client.mark_email_proposal_posted, it["id"], ev)
+                        email_threads[ev] = {"id": it["id"], "investor_name": it.get("investor_name"),
+                                             "note": it.get("proposed_note") or ""}
+                    except Exception as exc:
+                        print(f"matrix-intake: failed to post email proposal {it.get('id')}: {exc}", flush=True)
+                for it in lists["to_close"]:  # decided on the web → announce in-thread, then close
+                    ev = it.get("event_id")
+                    if not ev:
+                        continue
+                    try:
+                        await say(review_room, email_proposals.closure_line(it.get("status")), ev)
+                        await asyncio.to_thread(crm_client.mark_email_proposal_closed, it["id"])
+                        email_threads.pop(ev, None)
+                    except Exception as exc:
+                        print(f"matrix-intake: failed to close email proposal {it.get('id')}: {exc}", flush=True)
+            except Exception as exc:
+                print(f"matrix-intake: email-proposal poll error: {exc}", flush=True)
+            await asyncio.sleep(EMAIL_POLL_SEC)
+
     async def on_message(room: MatrixRoom, event: RoomMessageText):
         if event.sender == mx["user_id"]:
             return  # never react to our own messages (we post in-thread — this prevents loops)
-        if room.room_id != intake_room:
-            return
         text = (event.body or "").strip()
         if not text:
             return
         root = matrix_io.thread_root_of(event)
+        # Email-proposal review room: only a threaded reply to a card we posted is actionable.
+        if review_room and room.room_id == review_room:
+            if root and root in email_threads:
+                await handle_email_reply(room.room_id, root, text)
+            return
+        if room.room_id != intake_room:
+            return
         if root and store.has(root):
             await handle_reply(room.room_id, root, text)
         elif root:
@@ -180,8 +273,12 @@ async def main():
     client.add_event_callback(on_message, RoomMessageText)
     who = await client.whoami()
     print(f"matrix-intake: listening as {who.user_id} in room {intake_room}", flush=True)
+    tasks = [asyncio.create_task(client.sync_forever(timeout=30000))]
+    if review_room:
+        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)
     try:
-        await client.sync_forever(timeout=30000)
+        await asyncio.gather(*tasks)
     finally:
         await client.close()