ten31-database/backend/test_email_proposal_matrix.py

#!/usr/bin/env python3
"""Test the Matrix review-bot bridge for email-activity proposals (Features 2/3):
the bot work-lists (to_post / open / to_close), the Matrix side-row mark helpers, and an
in-thread (source='matrix') decision that closes the thread — plus the bot-or-admin role gate.
Synthetic data only (guardrail #9). The local model is stubbed.
Run: cd backend && python3 test_email_proposal_matrix.py
"""
import json
import os
import sqlite3
import sys
import tempfile

os.environ["CRM_DB_PATH"] = os.path.join(tempfile.mkdtemp(), "crm.db")
os.environ.setdefault("CRM_DATA_DIR", os.path.dirname(os.environ["CRM_DB_PATH"]))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import server  # noqa: E402

server._summarize_email_gist = lambda subject, body: "fundraising update; proposed a call"

FAILS = []


def check(cond, msg):
    print(("  PASS " if cond else "  FAIL ") + msg)
    if not cond:
        FAILS.append(msg)


def setup():
    conn = sqlite3.connect(os.environ["CRM_DB_PATH"])
    conn.row_factory = sqlite3.Row
    conn.executescript("""
        CREATE TABLE app_settings (key TEXT PRIMARY KEY, value_json TEXT, updated_at TEXT);
        CREATE TABLE email_accounts (id TEXT, email_address TEXT, sync_enabled INT DEFAULT 1, sync_status TEXT, backfill_complete INT);
        CREATE TABLE emails (id TEXT PRIMARY KEY, subject TEXT, body_text TEXT, snippet TEXT, from_name TEXT, from_email TEXT, sent_at TEXT, is_matched INT, match_status TEXT);
        CREATE TABLE email_investor_links (id TEXT, email_id TEXT, fundraising_investor_id TEXT, organization_id TEXT, contact_id TEXT, match_confidence REAL);
        CREATE TABLE email_activity_proposals (id TEXT PRIMARY KEY, email_id TEXT UNIQUE, investor_id TEXT, investor_name TEXT,
            direction TEXT, summary TEXT, proposed_note TEXT, email_subject TEXT, email_date TEXT, status TEXT DEFAULT 'pending',
            decided_by TEXT, decided_at TEXT, final_note TEXT, created_at TEXT);
        CREATE TABLE email_proposal_matrix (proposal_id TEXT PRIMARY KEY, event_id TEXT, posted_at TEXT, closed_at TEXT, created_at TEXT);
        CREATE TABLE users (id TEXT PRIMARY KEY, username TEXT);
        CREATE TABLE fundraising_investors (id TEXT PRIMARY KEY, investor_name TEXT, notes TEXT);
        CREATE TABLE fundraising_state (id TEXT PRIMARY KEY, grid_json TEXT, views_json TEXT, version INT,
            updated_by TEXT REFERENCES users(id), updated_at TEXT);
        CREATE TABLE interaction_log (id TEXT PRIMARY KEY, ts TEXT, actor_type TEXT, actor_id TEXT, action TEXT, target_type TEXT, target_id TEXT, payload TEXT, source TEXT, created_at TEXT);
    """)
    conn.execute("INSERT INTO users (id,username) VALUES ('user-1','grant')")
    conn.execute("INSERT INTO app_settings VALUES ('email_activity_since', ?, ?)", (json.dumps("2026-01-01T00:00:00"), "x"))
    conn.execute("INSERT INTO email_accounts (id,email_address) VALUES ('a','grant@ten31.xyz')")
    conn.execute("INSERT INTO fundraising_investors (id,investor_name,notes) VALUES ('inv1','Harbor & Vine','existing note')")
    grid = {"columns": [], "rows": [{"id": "inv1", "investor_name": "Harbor & Vine", "notes": "existing note"}]}
    conn.execute("INSERT INTO fundraising_state (id,grid_json,views_json,version) VALUES ('main',?,?,1)", (json.dumps(grid), "[]"))
    conn.executemany("INSERT INTO emails (id,subject,body_text,snippet,from_name,from_email,sent_at,is_matched,match_status) VALUES (?,?,?,?,?,?,?,1,'matched')", [
        ("e1", "Fund III", "Here is the update", "the quarterly update is attached", "Grant", "grant@ten31.xyz", "2026-06-01T10:00:00"),
        ("e2", "Re: Fund III", "Thanks, a question", "thanks — one question on terms", "LP Contact", "lp@harborvine.example", "2026-06-02T10:00:00"),
    ])
    conn.executemany("INSERT INTO email_investor_links (id,email_id,fundraising_investor_id,match_confidence) VALUES (?,?, 'inv1', 1.0)",
                     [("l1", "e1"), ("l2", "e2")])
    conn.commit()
    conn.close()


def main():
    setup()

    # role gate: bot passes the agent gate but is NOT an admin; member passes neither.
    check(server.require_bot_or_admin({"role": "bot"}), "bot passes require_bot_or_admin")
    check(server.require_bot_or_admin({"role": "admin"}), "admin passes require_bot_or_admin")
    check(not server.require_bot_or_admin({"role": "member"}), "member does NOT pass require_bot_or_admin")
    check(not server.require_admin({"role": "bot"}), "bot is NOT an admin (no user-mgmt/settings reach)")

    check(server.propose_email_activity_notes().get("proposed") == 2, "drafts 2 proposals")
    conn = server.get_db()
    props = server.list_email_activity_proposals(conn, status="pending")
    by_email = {p["email_id"]: p for p in props}
    p_a, p_b = by_email["e1"], by_email["e2"]

    # Both are pending + un-posted → both in to_post; card carries from/snippet/note context.
    lists = server.list_bot_email_proposals(conn)
    check(len(lists["to_post"]) == 2 and not lists["open"] and not lists["to_close"], "both proposals queued to_post")
    card = next(it for it in lists["to_post"] if it["id"] == p_a["id"])
    check(card.get("from_name") == "Grant" and "quarterly update" in (card.get("snippet") or ""), "card carries from_name + snippet")
    check("✉" in (card.get("proposed_note") or ""), "card carries the drafted note")

    # Post p_a to Matrix → it leaves to_post and becomes an open thread (event id recorded).
    server.mark_proposal_matrix_posted(conn, p_a["id"], "evtA")
    lists = server.list_bot_email_proposals(conn)
    check(len(lists["to_post"]) == 1 and lists["to_post"][0]["id"] == p_b["id"], "posting p_a leaves only p_b to_post")
    check(len(lists["open"]) == 1 and lists["open"][0]["id"] == p_a["id"] and lists["open"][0]["event_id"] == "evtA",
          "posted p_a is an open thread carrying its event id")

    # Decide p_a IN-THREAD on Matrix (approve + close in one transaction).
    r = server.decide_email_activity_proposal(conn, p_a["id"], "approve", "user-1", source="matrix", close_matrix=True)
    check(r.get("status") == "approved" and r.get("placed_in_grid") is True, "matrix approve appends to the grid")
    lists = server.list_bot_email_proposals(conn)
    check(not any(it["id"] == p_a["id"] for it in lists["open"] + lists["to_close"]),
          "matrix-decided proposal is closed (not re-announced via to_close)")
    src = conn.execute("SELECT source FROM interaction_log WHERE action='email.activity_approved'").fetchone()["source"]
    check(src == "matrix", "matrix decision is audited source='matrix'")

    # Web-decide path: post p_b, then dismiss it on the WEB (default source, no close) → the bot
    # must see it in to_close so it can announce the web decision in-thread, then close.
    server.mark_proposal_matrix_posted(conn, p_b["id"], "evtB")
    server.decide_email_activity_proposal(conn, p_b["id"], "dismiss", "user-1")  # web path: source crm_ui, no close
    lists = server.list_bot_email_proposals(conn)
    check(len(lists["to_close"]) == 1 and lists["to_close"][0]["id"] == p_b["id"] and lists["to_close"][0]["status"] == "dismissed",
          "web-decided open thread surfaces in to_close")
    src2 = conn.execute("SELECT source FROM interaction_log WHERE action='email.activity_dismissed'").fetchone()["source"]
    check(src2 == "crm_ui", "web decision is audited source='crm_ui'")

    server.mark_proposal_matrix_closed(conn, p_b["id"])
    lists = server.list_bot_email_proposals(conn)
    check(not lists["to_close"] and not lists["open"], "closing the thread clears the work-lists")

    # Marking a non-existent proposal is a clean not_found, not a crash.
    check(server.mark_proposal_matrix_posted(conn, "nope", "evtX").get("error") == "not_found", "mark posted on unknown id -> not_found")
    conn.close()

    if FAILS:
        print(f"\nFAILED ({len(FAILS)})")
        for f in FAILS:
            print("  - " + f)
        sys.exit(1)
    print("\nALL PASS (email-proposal Matrix bridge)")


if __name__ == "__main__":
    main()