mail-negotiator: classify merchant replies, draft gated, audited sends

- IMAP poll (UNSEEN, stdlib imaplib) -> parse (bare address via email.utils)
- threading: In-Reply-To -> outbound sent-draft message-id -> proposal
- classify: deterministic rules; optional LLM behind provider interface;
  below-threshold confidence -> needs_review (never guessed)
- unthreadable approvals flagged needs_review (suspicious by definition)
- merchant_status recorded on proposal detail; proposal lifecycle untouched
- reply drafts generated from templates (drafted), NEVER auto-sent
- SMTP sends only draft_status=approved (human approve_draft, audited);
  sent drafts store their Message-ID for future threading
- idempotent on Message-ID; DaemonRun wrapper for monitor

tests: 9 new (classification rules, threshold->review, threading, idempotence,
blocked send without approval, approve->send->audit chain, double approve)
— 29 total passing

cayceops/components/maild/__init__.py

@@ -0,0 +1 @@
+"""maild package."""

cayceops/components/maild/classify.py

@@ -0,0 +1,68 @@
+"""maild classification: rules first, LLM optional, conservative threshold.
+
+The LLM never decides alone: any result below ``CAYCE_OPS_LLM_REVIEW_THRESHOLD``
+confidence becomes ``needs_review``. With the ``null`` provider (default) only
+deterministic rules run.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+
+from cayceops.config import Config
+from cayceops.models import MailClassification
+
+_RULES: list[tuple[MailClassification, re.Pattern[str], float]] = [
+    # (class, pattern, confidence when matched)
+    (MailClassification.REJECTION, re.compile(r"\b(declin|reject|not able to|unable to|unfortunately|we regret|won.t be able)\w*", re.I), 0.9),
+    (MailClassification.APPROVAL, re.compile(r"\b(approv|accept|welcome to the|you.re (in|approved)|live|activated|glad to)\w*", re.I), 0.85),
+    (MailClassification.QUESTION, re.compile(r"\b(could you|can you|please (provide|clarify|confirm)|what (is|are)|how (do|does|can)|need more (info|details))\b", re.I), 0.7),
+]
+
+_LLM_PROMPT = """Classify this merchant email into exactly one of:
+approval, rejection, question, other.
+Reply with strict JSON: {"classification": "...", "confidence": 0.0-1.0}
+
+Subject and body:
+{body}"""
+
+
+def _rules(text: str) -> tuple[MailClassification, float]:
+    for kind, pattern, confidence in _RULES:
+        if pattern.search(text):
+            return kind, confidence
+    return MailClassification.OTHER, 0.3
+
+
+def _llm_classify(text: str, cfg: Config) -> tuple[MailClassification, float] | None:
+    """Ask the configured LLM. Returns None on any failure (rules take over)."""
+    from cayceops.llm import get_provider
+
+    try:
+        provider = get_provider(cfg)
+        raw = provider.complete(_LLM_PROMPT.format(body=text[:4000]), system="You classify merchant emails. Reply with JSON only.")
+        data = json.loads(re.search(r"\{.*\}", raw, re.S).group(0))
+        kind = MailClassification(data["classification"].lower())
+        confidence = float(data["confidence"])
+    except Exception:  # noqa: BLE001 — LLM is best-effort by design
+        return None
+    if kind not in MailClassification or kind is MailClassification.NEEDS_REVIEW:
+        return None
+    return kind, min(max(confidence, 0.0), 1.0)
+
+
+def classify(text: str, cfg: Config) -> tuple[MailClassification, float]:
+    """Classify merchant text. Degrades to rules when no LLM is configured."""
+    if cfg.llm_provider != "null":
+        llm_result = _llm_classify(text, cfg)
+        if llm_result is not None:
+            kind, confidence = llm_result
+        else:  # LLM failed — fall through to rules
+            kind, confidence = _rules(text)
+    else:
+        kind, confidence = _rules(text)
+
+    if confidence < cfg.llm_review_threshold:
+        return MailClassification.NEEDS_REVIEW, confidence
+    return kind, confidence

cayceops/components/maild/maild.py

@@ -0,0 +1,364 @@
+"""maild — classify merchant replies, draft responses, gated SMTP.
+
+Flow
+----
+1. ``poll``: fetch unread inbound mail (IMAP), parse, thread to the proposal
+   via ``In-Reply-To`` of a previously-sent draft, classify (rules + optional
+   LLM), store, and create a REPLY DRAFT (never sent automatically).
+2. ``approve_draft``: a human approves a specific draft (audited).
+3. ``send_approved``: SMTP sends *only* drafts in state ``approved``.
+
+Classification is conservative: anything below the confidence threshold becomes
+``needs_review`` and no draft is generated.
+
+Public signatures match the CLI: ``poll()``, ``approve_draft(id, actor)``,
+``send_approved()``. The ``_*`` variants take an explicit session/fetcher for
+tests and composition.
+"""
+
+from __future__ import annotations
+
+import email
+import email.policy
+import imaplib
+import smtplib
+from datetime import UTC, datetime
+from email.message import EmailMessage
+
+from cayceops.components.maild.classify import classify as _classify_text
+from cayceops.config import Config
+from cayceops.db import session_scope
+from cayceops.models import (
+    DaemonRun,
+    DraftStatus,
+    MailClassification,
+    MailDirection,
+    MailMessage,
+    Proposal,
+    audit,
+)
+
+# --- IMAP ---------------------------------------------------------------------
+
+
+def connect_imap(cfg: Config) -> imaplib.IMAP4_SSL:
+    client = imaplib.IMAP4_SSL(cfg.imap_host, cfg.imap_port)
+    client.login(cfg.imap_user, cfg.imap_password)
+    client.select(cfg.imap_mailbox)
+    return client
+
+
+def _fetch_unread(cfg: Config) -> list[bytes]:
+    """Fetch raw unread messages. Returns list of raw RFC822 bytes."""
+    client = connect_imap(cfg)
+    try:
+        _, data = client.search(None, "UNSEEN")
+        raw_messages: list[bytes] = []
+        for num in data[0].split():
+            _, msg_data = client.fetch(num, "(RFC822)")
+            raw_messages.append(msg_data[0][1])
+            client.store(num, "+FLAGS", "\\Seen")
+        return raw_messages
+    finally:
+        client.logout()
+
+
+# --- parsing ------------------------------------------------------------------
+
+
+def parse_raw(raw: bytes) -> dict:
+    """Parse raw RFC822 bytes into our flat dict shape."""
+    msg = email.message_from_bytes(raw, policy=email.policy.default)
+    body = ""
+    if msg.is_multipart():
+        for part in msg.walk():
+            if part.get_content_type() == "text/plain" and not part.get_filename():
+                body = part.get_content()
+                break
+    else:
+        body = msg.get_content()
+    reply_to = msg.get("In-Reply-To") or msg.get("References") or ""
+    message_id = (msg.get("Message-ID") or "").strip()
+    if not message_id:
+        message_id = f"<generated-{datetime.now(UTC).timestamp()}@localhost>"
+    # store the bare address (envelope-safe); display names stay out of the DB
+    from_addr = email.utils.parseaddr(msg.get("From") or "")[1].strip()[:320]
+    return {
+        "message_id": message_id[:998],
+        "in_reply_to": reply_to.split()[-1].strip() if reply_to.strip() else None,
+        "from_addr": from_addr,
+        "subject": (msg.get("Subject") or "").strip()[:998],
+        "body": body[:20000],
+    }
+
+
+# --- proposal threading -------------------------------------------------------
+
+
+def _match_proposal(parsed: dict, sess) -> Proposal | None:
+    """Thread an inbound mail to its proposal via In-Reply-To of sent drafts."""
+    irt = parsed.get("in_reply_to")
+    if not irt:
+        return None
+    sent = (
+        sess.query(MailMessage)
+        .filter(
+            MailMessage.direction == MailDirection.OUTBOUND,
+            MailMessage.message_id == irt,
+        )
+        .one_or_none()
+    )
+    if sent is None or sent.proposal_id is None:
+        return None
+    return sess.get(Proposal, sent.proposal_id)
+
+
+# --- drafts -------------------------------------------------------------------
+
+_DRAFT_TEMPLATES = {
+    MailClassification.APPROVAL: (
+        "Hello,\n\nThank you for the approval regarding \"{title}\". "
+        "We will proceed on our side and follow up with the next steps.\n\nBest regards,\nCayce"
+    ),
+    MailClassification.REJECTION: (
+        "Hello,\n\nThank you for considering \"{title}\". "
+        "We appreciate the reply and would welcome the chance to revisit in the future.\n\n"
+        "Best regards,\nCayce"
+    ),
+    MailClassification.QUESTION: (
+        "Hello,\n\nRegarding \"{title}\": {question}\n\n"
+        "Could you help us with the details so we can move forward?\n\nBest regards,\nCayce"
+    ),
+}
+
+
+def _build_draft(parsed: dict, kind: MailClassification, proposal: Proposal | None) -> str:
+    title = proposal.title if proposal else (parsed.get("subject") or "your message")
+    template = _DRAFT_TEMPLATES.get(kind)
+    if template is None:
+        return ""
+    question = (parsed.get("body") or "")[:200].strip() or "we had a question"
+    return template.format(title=title[:120], question=question)
+
+
+# --- inbound pass ---------------------------------------------------------------
+
+
+def _poll(cfg: Config, sess, fetcher=None) -> dict:
+    """One inbound pass: parse, classify, thread, draft. Returns stats."""
+    stats = {"processed": 0, "drafted": 0, "needs_review": 0, "skipped": 0}
+    for raw in (fetcher or _fetch_unread)(cfg):
+        parsed = parse_raw(raw)
+        if (
+            sess.query(MailMessage)
+            .filter(MailMessage.message_id == parsed["message_id"])
+            .one_or_none()
+        ):
+            stats["skipped"] += 1
+            continue
+
+        proposal = _match_proposal(parsed, sess)
+        kind, confidence = _classify_text(f"{parsed['subject']}\n{parsed['body']}", cfg)
+        if proposal is None and kind == MailClassification.APPROVAL:
+            # an approval we cannot thread is suspicious — human must look
+            kind = MailClassification.NEEDS_REVIEW
+            confidence = min(confidence, 0.4)
+
+        record = MailMessage(
+            direction=MailDirection.INBOUND,
+            message_id=parsed["message_id"],
+            in_reply_to=parsed["in_reply_to"],
+            from_addr=parsed["from_addr"] or None,
+            subject=parsed["subject"] or None,
+            body=parsed["body"],
+            classification=kind,
+            confidence=confidence,
+            proposal_id=proposal.id if proposal else None,
+            processed_at=datetime.now(UTC),
+        )
+        sess.add(record)
+        sess.flush()
+        audit(
+            sess,
+            actor="daemon:maild",
+            action="classify",
+            proposal_id=proposal.id if proposal else None,
+            detail={"classification": kind.value, "confidence": confidence},
+        )
+        stats["processed"] += 1
+
+        if kind == MailClassification.NEEDS_REVIEW:
+            stats["needs_review"] += 1
+            continue
+
+        if proposal is not None:
+            proposal.detail = {
+                **(proposal.detail or {}),
+                "merchant_status": kind.value,
+                "merchant_reply_at": datetime.now(UTC).isoformat(),
+            }
+
+        draft = _build_draft(parsed, kind, proposal)
+        if draft:
+            record.draft_reply = draft
+            record.draft_status = DraftStatus.DRAFTED
+            audit(
+                sess,
+                actor="daemon:maild",
+                action="draft_reply",
+                proposal_id=proposal.id if proposal else None,
+                detail={"mail_id": record.id},
+            )
+            stats["drafted"] += 1
+    return stats
+
+
+# --- human gate: approve draft -------------------------------------------------
+
+
+def _approve_draft(mail_id: int, actor: str, sess) -> int:
+    """Human approves a drafted reply. Only then may SMTP send it."""
+    record = sess.get(MailMessage, mail_id)
+    if record is None or record.direction != MailDirection.INBOUND:
+        return 1
+    if record.draft_status != DraftStatus.DRAFTED:
+        audit(
+            sess,
+            actor=actor,
+            action="approve_draft",
+            proposal_id=record.proposal_id,
+            result="error",
+            detail={"reason": f"invalid draft state: {record.draft_status.value}"},
+        )
+        return 1
+    record.draft_status = DraftStatus.APPROVED
+    audit(
+        sess,
+        actor=actor,
+        action="approve_draft",
+        proposal_id=record.proposal_id,
+        detail={"mail_id": record.id},
+    )
+    return 0
+
+
+# --- SMTP: only approved drafts ------------------------------------------------
+
+
+def _smtp_send(cfg: Config, to_addr: str, subject: str, body: str, message_id: str) -> None:
+    msg = EmailMessage()
+    msg["From"] = cfg.mail_from
+    msg["To"] = to_addr
+    msg["Subject"] = subject
+    msg["Message-ID"] = message_id
+    msg.set_content(body)
+    with smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) as server:
+        server.starttls()
+        server.login(cfg.smtp_user, cfg.smtp_password)
+        server.send_message(msg)
+
+
+def _approved_by(sess, mail_id: int) -> str:
+    from cayceops.models import AuditEntry
+
+    for entry in sess.query(AuditEntry).all():
+        if entry.action == "approve_draft" and entry.detail.get("mail_id") == mail_id:
+            return entry.actor
+    return "unknown"
+
+
+def _send_approved(cfg: Config, sess, sender=None) -> dict:
+    """Send every APPROVED draft. Anything else stays untouched."""
+    stats = {"sent": 0, "blocked": 0}
+    pending = (
+        sess.query(MailMessage)
+        .filter(
+            MailMessage.direction == MailDirection.INBOUND,
+            MailMessage.draft_status == DraftStatus.APPROVED,
+        )
+        .all()
+    )
+    for record in pending:
+        to_addr = record.from_addr or ""
+        if not to_addr:
+            record.draft_status = DraftStatus.SKIPPED
+            stats["blocked"] += 1
+            continue
+        message_id = (
+            f"<cayce-draft-{record.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>"
+        )
+        (sender or _smtp_send)(
+            cfg,
+            to_addr,
+            f"Re: {record.subject or 'your message'}",
+            record.draft_reply or "",
+            message_id,
+        )
+        approved_by = _approved_by(sess, record.id)
+        record.draft_status = DraftStatus.SENT
+        # store the Message-ID so future replies thread to this draft
+        record.message_id = message_id
+        audit(
+            sess,
+            actor="daemon:maild",
+            action="send_draft",
+            proposal_id=record.proposal_id,
+            detail={
+                "mail_id": record.id,
+                "message_id": message_id,
+                "approved_by": approved_by,
+            },
+        )
+        stats["sent"] += 1
+    return stats
+
+
+# --- CLI wrappers (session scope lives here) ------------------------------------
+
+
+def poll(cfg: Config | None = None) -> int:
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            run_row = DaemonRun(daemon="maild")
+            sess.add(run_row)
+            sess.flush()
+            try:
+                stats = _poll(cfg, sess)
+            except Exception as exc:  # noqa: BLE001 — daemon boundary
+                run_row.status = "error"
+                run_row.detail = {"error": str(exc)[:300]}
+                run_row.finished_at = datetime.now(UTC)
+                return 1
+            run_row.status = "ok"
+            run_row.detail = stats
+            run_row.finished_at = datetime.now(UTC)
+        return 0
+    finally:
+        engine().dispose()
+
+
+def approve_draft(mail_id: int, actor: str) -> int:
+    from cayceops.db import engine
+
+    try:
+        with session_scope() as sess:
+            return _approve_draft(mail_id, actor, sess)
+    finally:
+        engine().dispose()
+
+
+def send_approved(cfg: Config | None = None) -> int:
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            _send_approved(cfg, sess)
+        return 0
+    finally:
+        engine().dispose()

tests/test_maild.py

@@ -0,0 +1,158 @@
+"""maild tests — parsing, classification, threading, gated send. Invented data only."""
+
+from __future__ import annotations
+
+from cayceops.components.maild import classify as clf
+from cayceops.components.maild import maild
+from cayceops.models import (
+    AuditEntry,
+    DraftStatus,
+    MailClassification,
+    MailDirection,
+    MailMessage,
+    Proposal,
+    ProposalStatus,
+)
+
+OUTBOUND_MSGID = "<cayce-draft-1-1000@cayce-co.xyz>"
+
+
+def _eml(subject: str, body: str, in_reply_to: str | None = None, msgid: str | None = None) -> bytes:
+    lines = [
+        "From: Partner Desk <partners@acme.example.org>",
+        "To: agent@cayce-co.xyz",
+        f"Subject: {subject}",
+        f"Message-ID: {msgid or '<acme-reply-1@acme.example.org>'}",
+    ]
+    if in_reply_to:
+        lines.append(f"In-Reply-To: {in_reply_to}")
+    lines += ["MIME-Version: 1.0", "Content-Type: text/plain; charset=utf-8", "", body]
+    return "\r\n".join(lines).encode()
+
+
+def _proposal_with_sent_draft(sess) -> Proposal:
+    p = Proposal(kind="partnership", title="Acme hotel affiliate program", status=ProposalStatus.APPROVED)
+    sess.add(p)
+    sess.flush()
+    sess.add(
+        MailMessage(
+            direction=MailDirection.OUTBOUND,
+            message_id=OUTBOUND_MSGID,
+            to_addr="partners@acme.example.org",
+            subject="[cayce] partnership inquiry",
+            proposal_id=p.id,
+            draft_status=DraftStatus.SENT,
+        )
+    )
+    sess.flush()
+    return p
+
+
+# --- classifier ---------------------------------------------------------------
+
+
+def test_rules_classification(cfg):
+    config = cfg()
+    assert clf.classify("Your application was approved, welcome to the program", config)[0] is MailClassification.APPROVAL
+    assert clf.classify("Unfortunately we cannot accept new partners", config)[0] is MailClassification.REJECTION
+    assert clf.classify("Could you provide your traffic numbers?", config)[0] is MailClassification.QUESTION
+    kind, confidence = clf.classify("grocery list", config)
+    assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW
+
+
+def test_low_confidence_becomes_needs_review(cfg):
+    config = cfg(CAYCE_OPS_LLM_REVIEW_THRESHOLD="0.95")
+    kind, confidence = clf.classify("Your application was approved, welcome", config)
+    assert kind is MailClassification.NEEDS_REVIEW
+    assert confidence < 0.95
+
+
+# --- inbound pass ---------------------------------------------------------------
+
+
+def test_poll_threads_approval_and_drafts_reply(cfg, sess):
+    p = _proposal_with_sent_draft(sess)
+    raw = _eml("Re: [cayce] partnership inquiry", "Approved! Welcome to the Acme partner program.", in_reply_to=OUTBOUND_MSGID)
+
+    stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    assert stats["processed"] == 1 and stats["drafted"] == 1
+
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert inbound.classification is MailClassification.APPROVAL
+    assert inbound.proposal_id == p.id
+    assert inbound.draft_status is DraftStatus.DRAFTED
+    assert inbound.draft_reply and "approval" in inbound.draft_reply.lower()
+
+    assert p.detail["merchant_status"] == "approval"
+    assert p.status is ProposalStatus.APPROVED  # lifecycle untouched by maild
+
+
+def test_poll_idempotent_on_message_id(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: hi", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    fetcher = lambda cfg: [raw]
+    assert maild._poll(cfg(), sess, fetcher=fetcher)["processed"] == 1
+    assert maild._poll(cfg(), sess, fetcher=fetcher)["skipped"] == 1
+
+
+def test_unthreaded_approval_needs_review(cfg, sess):
+    raw = _eml("Great news", "Approved! Welcome aboard.")  # no In-Reply-To
+    stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert inbound.classification is MailClassification.NEEDS_REVIEW
+    assert stats["needs_review"] == 1
+    assert inbound.draft_status is DraftStatus.NONE  # no draft for review items
+
+
+def test_question_gets_clarification_draft(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Could you provide your monthly traffic?", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert inbound.classification is MailClassification.QUESTION
+    assert inbound.draft_status is DraftStatus.DRAFTED
+    assert "traffic" in inbound.draft_reply
+
+
+# --- human gate -------------------------------------------------------------------
+
+
+def test_send_blocked_without_approval(cfg, sess, monkeypatch):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+
+    sent = []
+    config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
+    stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
+    assert stats["sent"] == 0 and sent == []
+    assert sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one().draft_status is DraftStatus.DRAFTED
+
+
+def test_send_requires_explicit_human_approval(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+
+    sent = []
+    config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
+    stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
+    assert stats["sent"] == 1
+    assert inbound.draft_status is DraftStatus.SENT
+    assert sent[0][1] == "partners@acme.example.org"
+
+    row = sess.query(AuditEntry).filter_by(action="send_draft").one()
+    assert "gustavo" in str(row.detail)
+
+
+def test_double_approve_is_rejected(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 1