"""Inbound email → vault (webhook-push).

A hosted brain has no scheduler and no IMAP creds, so it can't poll a mailbox.
Instead the shared email gateway receives all mail for the brain's address and
POSTs a parsed message to ``POST /inbound-email`` (wired in :mod:`.server`).

**This is a public address**, which is the whole security problem. Anyone can put
anything in a ``From`` header, so the gates run in order and every one of them
fails closed:

1. **Secret** — the gateway's shared webhook secret, constant-time.
2. **Addressed to me** — the message was actually sent to this brain's address.
3. **Not a machine** — autoresponders, bulk mail and bounces never get an answer,
   because a brain that replies to an autoresponder talks to it forever.
4. **Authentic** — aligned SPF+DKIM. A ``dkim=pass`` signed by somebody else is
   not authentication; alignment against the ``From`` domain is what makes it one.
   Verdicts the gateway couldn't read arrive as ``"unknown"`` and are refused.
5. **Known sender** — owner / team / an allowlisted domain. Everyone else is
   ignored with a 202, never an error: a policy decision must not make the
   provider retry.
6. **Not already handled** — a Message-ID claim taken *before* the model call, so
   a redelivery can't file the same mail twice.

Only then does tier decide what happens. The **owner** is the principal, so their
mail drives the model routing pass and fans out into PEAK. Everyone else is
capture-only (P3) — untrusted text must never steer a model that holds write
scope, which is the ingest-≠-send guardrail applied to a public inbox.

``handle_inbound`` never raises. Anything unexpected degrades to a refusal with a
200-class status, because the alternative is a 500, and the provider retries a
500 forever.

Kept free of any Starlette / ``mcp`` import so this core is unit-testable on the
dev venv (the thin HTTP route lives in :mod:`.server`).
"""
import hmac
import os
from email.utils import parseaddr

WEBHOOK_SECRET_ENV = "EMAIL_WEBHOOK_SECRET"
REQUIRE_AUTH_ENV = "EMAIL_INGEST_REQUIRE_AUTH"
INGEST_TEXT_MAX_CHARS = 100_000

# A verdict must be one of these to count as a pass. "bestguesspass" is the
# softened SPF result some receivers stamp for a sender they recognise but can't
# strictly verify; engine/team_inbox._auth_ok already accepts it, so this matches.
_SPF_OK = {"pass", "bestguesspass"}


def _addr(header_value) -> str:
    """The bare, lower-cased address from a possibly-``Name <addr>`` header."""
    return (parseaddr(str(header_value or ""))[1] or "").strip().lower()


def _domain(header_value) -> str:
    a = _addr(header_value)
    return a.rsplit("@", 1)[1] if "@" in a else ""


def _ingest_cfg(svc) -> dict:
    return (svc.config.get("email") or {}).get("ingest") or {}


def verify_secret(provided) -> bool:
    """Constant-time check of the gateway's shared webhook secret. Fails closed
    when the env secret is unset — an unconfigured webhook never accepts."""
    expected = os.environ.get(WEBHOOK_SECRET_ENV) or ""
    return bool(expected) and hmac.compare_digest(str(provided or ""), expected)


def classify_sender(svc, frm):
    """The sender's trust tier: ``"owner"``, ``"team"``, ``"domain"`` or ``None``.

    Owner and team are *identities* already in Config. A domain is a much weaker
    claim — it says only "somebody at this company" — which is why the domain tier
    never earns more than capture."""
    cfg = svc.config
    who = _addr(frm)
    if not who:
        return None
    if who in {str(a).strip().lower() for a in (cfg.get("owner") or {}).get("emails") or []}:
        return "owner"
    if who in {str((t or {}).get("email") or "").strip().lower() for t in cfg.get("team") or []}:
        return "team"
    domains = {str(d).strip().lower().lstrip("@")
               for d in _ingest_cfg(svc).get("allowed_domains") or []}
    if domains and _domain(frm) in domains:
        return "domain"
    return None


def is_allowed_sender(svc, frm) -> bool:
    """Back-compat shim: the original owner-only gate, now one tier of several."""
    return classify_sender(svc, frm) == "owner"


def auth_ok(svc, payload):
    """``(ok, reason)`` for the SPF/DKIM gate.

    Fails closed on a missing ``auth`` block, which is exactly what a pre-v2
    gateway sends — so an un-upgraded gateway degrades to refusing mail rather
    than to trusting it. ``EMAIL_INGEST_REQUIRE_AUTH=0`` (or
    ``email.ingest.require_auth: false``) is the escape hatch for an operator
    whose mail path genuinely cannot stamp the headers; it mirrors
    PRINCIPAL_INBOX_REQUIRE_AUTH on the IMAP side."""
    env = os.environ.get(REQUIRE_AUTH_ENV)
    required = (env not in ("0", "false", "no")) if env is not None \
        else bool(_ingest_cfg(svc).get("require_auth", True))
    auth = (payload or {}).get("auth") or {}
    if not required:
        return True, "auth not required (unverified sender identity)"
    if not auth:
        return False, ("no authentication results on this message — the gateway is "
                       "pre-v2 or the mail path stamped none")
    if str(auth.get("dkim") or "").lower() != "pass":
        return False, f"dkim={auth.get('dkim') or 'unknown'}"
    if str(auth.get("spf") or "").lower() not in _SPF_OK:
        return False, f"spf={auth.get('spf') or 'unknown'}"
    if not auth.get("aligned"):
        # The spoof a naive pass-check waves through: a valid signature from a
        # domain that has nothing to do with the From address.
        return False, (f"signature domain {auth.get('signing_domain') or '?'} is not aligned "
                       f"with sender domain {auth.get('from_domain') or '?'}")
    return True, "aligned spf+dkim"


def is_auto(svc, payload) -> bool:
    """Whether this is machine mail we must never answer.

    Trusts the gateway's header analysis and adds the one check it cannot make:
    mail from the brain's *own* address, which is the self-loop a misconfigured
    forward creates."""
    if ((payload or {}).get("auto") or {}).get("is_auto"):
        return True
    agent = svc.config.get("agent") or {}
    mine = {_addr(agent.get("email")), _addr(agent.get("ingress_email")),
            _addr(os.environ.get("AGENT_EMAIL")), _addr(os.environ.get("AGENT_INGRESS_EMAIL"))}
    mine.discard("")
    return _addr((payload or {}).get("from")) in mine


def my_address(svc) -> str:
    """This brain's own inbound address. Env first: a re-provisioned or
    bundle-restored tenant carries it in the environment, while the rendered
    config may predate the address being assigned."""
    return (_addr(os.environ.get("AGENT_INGRESS_EMAIL"))
            or _addr((svc.config.get("agent") or {}).get("ingress_email"))
            or _addr(os.environ.get("AGENT_EMAIL"))
            or _addr((svc.config.get("agent") or {}).get("email")))


def addressed_to_me(svc, payload):
    """``(ok, reason)`` — was this actually sent to this brain?

    Without the check, anything the gateway misroutes is ingested as though it had
    been addressed here. No configured address → accept with a warning, preserving
    the behavior of brains provisioned before addresses existed."""
    mine = my_address(svc)
    if not mine:
        return True, "this brain has no configured address — cannot verify the recipient"
    seen = {_addr(a) for a in (payload or {}).get("recipients") or []}
    seen.add(_addr((payload or {}).get("to")))
    return (True, "") if mine in seen else (False, f"not addressed to {mine}")


def _clamp(text):
    s = str(text or "")
    return s if len(s) <= INGEST_TEXT_MAX_CHARS else s[:INGEST_TEXT_MAX_CHARS]


def ingest_email(svc, frm, subject, text) -> dict:
    """Auto-ingest one email: model routing pass → apply (one commit) → owner
    summary. Imports the tool helpers lazily (heavy module; no import cycle)."""
    from alexandria_vault.mcp import tools
    from alexandria_vault.mcp.seats import READ_WRITE, Seat

    content = f"# Email — {subject}\n\nFrom: {frm}\n\n{_clamp(text)}".strip()
    manifest = tools._ingest_pass(svc, content, instructions=f"Filed from an email from {frm}.")
    applied = tools._apply_ingest(svc, Seat(id="inbox", scope=READ_WRITE),
                                  {"manifest_id": manifest["manifest_id"]})
    written = applied.get("written") or []
    owner = ((svc.config.get("owner") or {}).get("emails") or [None])[0]
    if owner:
        body = (f"Ingested your email “{subject}”.\n\n"
                f"{manifest.get('tldr') or ''}\n\nFiled: {', '.join(written) or '(nothing filed)'}")
        try:
            tools._email_deliver(svc, owner, f"Filed: {subject}", body)
        except Exception:  # noqa: BLE001 — a failed summary must not fail the ingest
            pass
    return {"filed": written, "tldr": manifest.get("tldr"), "commit": applied.get("committed")}


def _posture(svc, tier) -> str:
    """What this tier's mail does to the vault: ``file`` (model routing pass),
    ``capture`` (verbatim staging) or ``ignore``.

    The defaults are the security posture, and they are asymmetric on purpose. The
    owner is the principal, so their mail files. Everyone else captures: untrusted
    text must never steer a model that holds write scope."""
    default = {"owner": "file", "team": "capture", "domain": "capture"}.get(tier, "ignore")
    value = str(_ingest_cfg(svc).get(f"{tier}_posture") or default).lower()
    return value if value in ("file", "capture", "ignore") else default


def capture_email(svc, frm, subject, text, payload=None) -> dict:
    """Stage a non-owner message verbatim. Deterministic — no model, no routing."""
    from alexandria_vault.mcp import mail_capture

    payload = payload or {}
    out = mail_capture.capture_email(
        svc, frm, subject, text,
        attachments=payload.get("attachments"),
        cfg=_ingest_cfg(svc),
        tier=classify_sender(svc, frm) or "",
        message_id=payload.get("message_id") or "",
        thread_root=(payload.get("references") or [payload.get("message_id") or ""])[0])
    if not out:
        return {"filed": [], "tldr": "nothing to capture (no text, no attachments)"}
    return {"filed": [out["note"]], "tldr": f"staged “{out['title']}” for filing",
            "attachments": out.get("attachments") or []}


def _ignored(reason, **extra):
    return {"status": 202, "status_text": f"ignored: {reason}", **extra}


def handle_inbound(svc, secret, payload) -> dict:
    """Run every gate, then act on tier. Returns a dict carrying an HTTP ``status``
    for the route. Never raises: an exception here becomes a 500, and a 500 is the
    one answer that makes the provider send the same message again forever."""
    from alexandria_vault.mcp import mail_state

    payload = payload or {}
    if not verify_secret(secret):
        return {"status": 401, "error": "unauthorized: bad or missing webhook secret"}

    ok, why = addressed_to_me(svc, payload)
    if not ok:
        return _ignored(why)
    if is_auto(svc, payload):
        return _ignored("automated mail (autoresponder, bulk, or bounce)")

    frm = payload.get("from") or ""
    tier = classify_sender(svc, frm)
    if tier is None:
        return _ignored(f"{_addr(frm) or 'sender'} is not a known sender for this brain")

    ok, why = auth_ok(svc, payload)
    if not ok:
        return _ignored(f"unverified sender ({why})", tier=tier)

    subject = (payload.get("subject") or "(no subject)").strip()
    # The claim is taken BEFORE the model call, so a redelivery arriving mid-ingest
    # is answered rather than allowed to start a second pass and a second commit.
    key = mail_state.key_for(os.environ.get("VAULT_TENANT"),
                             payload.get("message_id"), payload.get("provider_message_id"))
    claimed = mail_state.claim(key, subject)
    if claimed == "in_flight":
        return {"status": 200, "status_text": "already being processed", "duplicate": True}
    if isinstance(claimed, dict):
        return {"status": 200, "status_text": "already handled", "duplicate": True,
                **(claimed.get("done") or {})}

    text = payload.get("text") or payload.get("html") or ""
    posture = _posture(svc, tier)
    if posture == "ignore":
        mail_state.release(key)
        return _ignored(f"{tier} senders are not ingested on this brain", tier=tier)

    try:
        if posture == "file":
            result = {"status": 200, "tier": tier, "posture": posture,
                      **ingest_email(svc, frm, subject, text)}
        else:
            result = {"status": 200, "tier": tier, "posture": "capture",
                      **capture_email(svc, frm, subject, text, payload)}
    except Exception as e:  # noqa: BLE001 — see the docstring: never a 500
        mail_state.release(key)
        return {"status": 200, "status_text": f"could not file this message: {e}",
                "filed": [], "error": str(e)}
    mail_state.complete(key, {k: v for k, v in result.items() if k != "status"})
    return result
