diff --git a/cayceops/components/maild/__init__.py b/cayceops/components/maild/__init__.py new file mode 100644 index 0000000..b3c267f --- /dev/null +++ b/cayceops/components/maild/__init__.py @@ -0,0 +1 @@ +"""maild package.""" diff --git a/cayceops/components/maild/classify.py b/cayceops/components/maild/classify.py new file mode 100644 index 0000000..0133330 --- /dev/null +++ b/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 diff --git a/cayceops/components/maild/maild.py b/cayceops/components/maild/maild.py new file mode 100644 index 0000000..34397e9 --- /dev/null +++ b/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"" + # 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"" + ) + (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() diff --git a/tests/test_maild.py b/tests/test_maild.py new file mode 100644 index 0000000..bb3a843 --- /dev/null +++ b/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 = "" + + +def _eml(subject: str, body: str, in_reply_to: str | None = None, msgid: str | None = None) -> bytes: + lines = [ + "From: Partner Desk ", + "To: agent@cayce-co.xyz", + f"Subject: {subject}", + f"Message-ID: {msgid or ''}", + ] + 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