cayce-ops/cayceops/components/maild/maild.py

"""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,
    _content_hash,
    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[tuple[bytes, bytes]]:
    """Fetch raw unread messages as [(imap_num, raw_bytes)].

    Messages are NOT marked \\Seen here — marking happens only after the
    processed rows are committed (audit finding: a crash mid-batch used to
    silently lose the rest of the batch).
    """
    client = connect_imap(cfg)
    try:
        _, data = client.search(None, "UNSEEN")
        out: list[tuple[bytes, bytes]] = []
        for num in data[0].split():
            _, msg_data = client.fetch(num, "(RFC822)")
            out.append((num, msg_data[0][1]))
        return out
    finally:
        client.logout()


def _mark_seen(cfg: Config, nums: list[bytes]) -> None:
    """Mark processed messages \\Seen (called after commit)."""
    if not nums:
        return
    client = connect_imap(cfg)
    try:
        for num in nums:
            client.store(num, "+FLAGS", "\\Seen")
    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.

    1. In-Reply-To matches a SENT OUTBOUND message (our replies carry their own
       Message-ID and an outbound row, created at send time).
    2. Fallback: it matches ANY stored message (e.g. a second reply threaded to
       our first inbound) that already carries a proposal_id.
    """
    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 not None and sent.proposal_id is not None:
        return sess.get(Proposal, sent.proposal_id)
    chain = (
        sess.query(MailMessage)
        .filter(MailMessage.message_id == irt, MailMessage.proposal_id.isnot(None))
        .one_or_none()
    )
    if chain is not None:
        return sess.get(Proposal, chain.proposal_id)
    return None


# --- 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, marker=None) -> tuple[dict, list[bytes]]:
    """One inbound pass: parse, classify, thread, draft.

    Returns (stats, processed_imap_nums). The caller marks \\Seen only after
    its transaction commits.
    """
    stats = {"processed": 0, "drafted": 0, "needs_review": 0, "skipped": 0}
    processed_nums: list[bytes] = []
    for num, 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
        processed_nums.append(num)  # stored -> safe to mark seen after commit

        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, processed_nums


# --- human gate: approve draft -------------------------------------------------


def _approve_draft(mail_id: int, actor: str, sess) -> int:
    """Human approves a drafted reply (inbound-threaded or first-contact).

    The approval is BOUND to the draft text via sha256 — a later edit breaks
    the binding and the send refuses (audit finding: approved text was not
    tied to its content).
    """
    record = sess.get(MailMessage, mail_id)
    if record is None:
        return 1
    if record.draft_status != DraftStatus.DRAFTED or not record.draft_reply:
        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
    record.approved_hash = _content_hash(record.draft_reply)
    record.approved_by = actor[:128]
    audit(
        sess,
        actor=actor,
        action="approve_draft",
        proposal_id=record.proposal_id,
        detail={"mail_id": record.id, "hash": record.approved_hash},
    )
    return 0


# --- SMTP: only approved drafts ------------------------------------------------


def _smtp_send(
    cfg: Config,
    to_addr: str,
    subject: str,
    body: str,
    message_id: str,
    in_reply_to: str | None = None,
) -> None:
    msg = EmailMessage()
    msg["From"] = cfg.mail_from
    msg["To"] = to_addr
    msg["Subject"] = subject
    msg["Message-ID"] = message_id
    if in_reply_to:
        msg["In-Reply-To"] = in_reply_to  # keep merchant threads intact
    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_one(cfg: Config, sess, mail_id: int, sender=None) -> str:
    """Send ONE approved draft. Returns: sent | hash_mismatch | missing | blocked.

    Creates the OUTBOUND row (its own Message-ID + In-Reply-To pointing at the
    inbound message) so future replies thread correctly — the inbound row's
    message_id is left untouched (audit finding: it used to be overwritten).
    The caller owns the transaction: one message per commit, so an SMTP failure
    on message N never un-sends messages 1..N-1 (audit finding: duplicates).
    """
    from cayceops.models import _content_hash

    record = sess.get(MailMessage, mail_id)
    if record is None:
        return "missing"
    if record.draft_status != DraftStatus.APPROVED:
        return "blocked"

    if record.direction == MailDirection.INBOUND:
        # reply: goes back to the inbound sender, threads to the inbound mail,
        # and gets its own OUTBOUND row
        to_addr = record.from_addr or ""
        subject = f"Re: {record.subject or 'your message'}"
        in_reply_to = record.message_id
        outbound_msgid = (
            f"<cayce-out-{record.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>"
        )
        needs_row = True
    else:
        # first-contact draft: already an OUTBOUND row; send as-is
        to_addr = record.to_addr or ""
        subject = record.subject or ""
        in_reply_to = None
        outbound_msgid = record.message_id
        needs_row = False
    if not to_addr:
        record.draft_status = DraftStatus.SKIPPED
        return "blocked"

    # binding: the text must be exactly what the human approved
    if record.approved_hash and _content_hash(record.draft_reply or "") != record.approved_hash:
        record.draft_status = DraftStatus.DRAFTED  # needs fresh approval
        audit(
            sess,
            actor="daemon:maild",
            action="send_refused_hash_mismatch",
            proposal_id=record.proposal_id,
            result="error",
            detail={"mail_id": record.id},
        )
        return "hash_mismatch"

    (sender or _smtp_send)(
        cfg,
        to_addr,
        subject,
        record.draft_reply or "",
        outbound_msgid,
        in_reply_to=in_reply_to,
    )
    approved_by = _approved_by(sess, record.id)
    if needs_row:
        sess.add(
            MailMessage(
                direction=MailDirection.OUTBOUND,
                message_id=outbound_msgid,
                in_reply_to=record.message_id,
                from_addr=cfg.mail_from,
                to_addr=to_addr,
                subject=subject,
                body=record.draft_reply,
                proposal_id=record.proposal_id,
                draft_status=DraftStatus.SENT,
                processed_at=datetime.now(UTC),
            )
        )
    record.draft_status = DraftStatus.SENT
    sess.flush()
    audit(
        sess,
        actor="daemon:maild",
        action="send_draft",
        proposal_id=record.proposal_id,
        detail={
            "mail_id": record.id,
            "message_id": outbound_msgid,
            "approved_by": approved_by,
        },
    )
    return "sent"


def _send_approved(cfg: Config, sess, sender=None, factory=None) -> dict:
    """Send every APPROVED draft, one transaction per message.

    `factory` is a SQLAlchemy sessionmaker bound to the shared engine; each
    message is processed and committed in its own session (tests inject the
    engine's maker; the CLI passes appdb.session_factory()).
    """
    stats = {"sent": 0, "blocked": 0, "hash_mismatch": 0, "missing": 0}
    ids = [
        row.id
        for row in sess.query(MailMessage.id)
        .filter(MailMessage.draft_status == DraftStatus.APPROVED)
        .all()
    ]
    for mail_id in ids:
        # one transaction per message — an SMTP failure on message N must not
        # roll back (and later re-send) messages 1..N-1 (audit finding 3)
        try:
            if factory is not None:
                with factory() as one_sess:  # sessionmaker() -> Session
                    outcome = _send_one(cfg, one_sess, mail_id, sender)
                    one_sess.commit()
            else:
                outcome = _send_one(cfg, sess, mail_id, sender)
        except Exception as exc:  # noqa: BLE001 — isolate the failed message
            outcome = "error"
            if sess is not None:
                detail = {"mail_id": mail_id, "error": str(exc)[:200]}
                audit(sess, actor="daemon:maild", action="send_failed",
                      result="error", detail=detail)
        stats[outcome] = stats.get(outcome, 0) + 1
    return stats


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 _compose(proposal_id: int, to_addr: str, subject: str, body: str, sess) -> int:
    """Create the FIRST-CONTACT outbound draft for a proposal (state: drafted).

    This is the missing path the audit flagged: without it, threading only
    worked for replies to mails we never actually sent.
    """
    proposal = sess.get(Proposal, proposal_id)
    if proposal is None or not to_addr:
        return -1
    outbound = MailMessage(
        direction=MailDirection.OUTBOUND,
        message_id=f"<cayce-out-{proposal.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>",
        from_addr=None,  # filled from MAIL_FROM at send time
        to_addr=to_addr[:320],
        subject=subject[:998],
        body=body,
        draft_reply=body,  # canonical approved-text field (same as reply drafts)
        proposal_id=proposal.id,
        draft_status=DraftStatus.DRAFTED,
    )
    sess.add(outbound)
    sess.flush()
    audit(
        sess,
        actor="daemon:maild",
        action="compose_draft",
        proposal_id=proposal.id,
        detail={"mail_id": outbound.id},
    )
    return outbound.id


# --- 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:
        nums: list[bytes] = []
        with session_scope() as sess:
            run_row = DaemonRun(daemon="maild")
            sess.add(run_row)
            sess.flush()
            try:
                stats, nums = _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)
        # committed: now it is safe to mark the batch seen
        if nums:
            _mark_seen(cfg, nums)
        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, session_factory

    cfg = cfg or load()
    try:
        # ids first, then one transaction per message: a failure on message N
        # cannot roll back already-sent messages (audit finding: duplicates)
        with session_scope() as sess:
            ids = [
                row.id
                for row in sess.query(MailMessage.id)
                .filter(MailMessage.draft_status == DraftStatus.APPROVED)
                .all()
            ]
        factory = session_factory()
        for mail_id in ids:
            try:
                with factory() as one_sess:
                    _send_one(cfg, one_sess, mail_id)
                    one_sess.commit()
            except Exception as exc:  # noqa: BLE001 — isolate each send
                print(f"send {mail_id} failed: {str(exc)[:160]}")
        return 0
    finally:
        engine().dispose()


def compose(proposal_id: int, to_addr: str, subject: str, body: str) -> int:
    from cayceops.db import engine

    try:
        with session_scope() as sess:
            mail_id = _compose(proposal_id, to_addr, subject, body, sess)
            return 0 if mail_id >= 0 else 1
    finally:
        engine().dispose()