agentic.email/backend/app/services/human_handoff_service.py

import logging
import re
from datetime import datetime, timezone
from uuid import UUID

from bson import ObjectId
from pymongo.errors import DuplicateKeyError

from app.database import SessionFactory
from app.models import Account
from app.services.domain_provisioning import get_domain_by_name
from app.services.external_reply_service import (
    ExternalReplyThreadingError,
    build_external_reply_plan,
    resolve_external_reply_parent,
    resolve_internet_message_id,
    send_external_reply,
)
from app.services.handoff_delivery import (
    HandoffPreflightError,
    JOB_MIRROR,
    JOB_RELAY,
    enqueue_handoff_delivery,
)
from app.services.handoff_email_templates import (
    attach_quoted_history,
    build_gmail_quote,
    sender_display_name,
)
from app.services.handoff_evaluator import evaluate_handoff
from app.services.handoff_mirror import load_mirror_payload
from app.services.handoff_mirror_policy import (
    GENERATED_BY_AGENT,
    GENERATED_BY_HANDOFF_BRIDGE,
    GENERATED_BY_HUMAN,
    MESSAGE_ORIGIN_AGENT,
    MESSAGE_ORIGIN_EXTERNAL_CUSTOMER,
    MESSAGE_ORIGIN_HANDOFF_MIRROR,
    MESSAGE_ORIGIN_HUMAN_OWNER,
    RELAY_STATUS_BLOCKED,
    RELAY_STATUS_FAILED,
    RELAY_STATUS_PENDING,
    RELAY_STATUS_SENT,
    should_mirror_to_human,
)
from app.services.mail_sender import deliver_email
from app.services.message_ids import (
    canonical_message_id,
    folded_message_id,
    message_id_fields,
    parse_message_id_list,
)
from app.services.mime_body_extractor import extract_new_reply_mime_content
from app.services.raw_mime_storage import get_raw_mime_storage
from app.services.reply_extractor import (
    extract_new_reply_content,
    validate_external_reply_body,
)
from app.services.thread_service import (
    attach_message_to_thread,
    message_event_at,
    threading_fields_from_parent,
)

logger = logging.getLogger(__name__)

MESSAGE_SCOPE_EXTERNAL = "EXTERNAL"
MESSAGE_SCOPE_HANDOFF = "HANDOFF"
THREAD_TYPE_HUMAN_HANDOFF = "HUMAN_HANDOFF"

HANDOFF_STATUS_ACTIVE = "ACTIVE"
HANDOFF_STATUS_WAITING = "WAITING_FOR_HUMAN"
HANDOFF_STATUS_HUMAN_REPLIED = "HUMAN_REPLIED"

OWNER_AGENT = "AGENT"
OWNER_HUMAN = "HUMAN"

LINK_EXTERNAL_MIRRORED = "EXTERNAL_MIRRORED_TO_HUMAN"
LINK_HUMAN_RELAYED = "HUMAN_RELAYED_TO_EXTERNAL"


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _snippet(value: str, limit: int = 240) -> str:
    return re.sub(r"\s+", " ", value or "").strip()[:limit]


def _safe_subject_part(value: str, fallback: str) -> str:
    cleaned = re.sub(r"[\r\n]+", " ", value or "").strip()
    return cleaned or fallback


def build_handoff_subject(
    *,
    sender_name: str,
    external_sender: str,
    external_subject: str,
    rule_name: str | None = None,
) -> str:
    del rule_name
    subject = " · ".join(
        [
            _safe_subject_part(sender_name, "unknown sender"),
            "handoff",
            _safe_subject_part(external_sender, "unknown sender"),
            _safe_subject_part(external_subject, "(No subject)"),
        ]
    )
    return subject[:998].rstrip()


def redact_human_address(
    value: str | None,
    *,
    human_email: str,
    agent_email: str,
) -> str | None:
    if value is None or not human_email:
        return value
    return re.sub(
        re.escape(human_email),
        agent_email,
        value,
        flags=re.IGNORECASE,
    )


def select_matching_handoff_rule(
    rules: list[dict],
    *,
    email: dict,
) -> dict | None:
    ordered = sorted(
        rules,
        key=lambda rule: (
            int(rule.get("priority", 0)),
            (
                rule["created_at"].timestamp()
                if isinstance(rule.get("created_at"), datetime)
                else 0
            ),
            str(rule.get("_id", "")),
        ),
    )
    return next(
        (
            rule
            for rule in ordered
            if evaluate_handoff(
                email=email,
                labels=rule.get("labels", []),
                email_types=rule.get("email_types", []),
                instructions=rule.get("instructions", ""),
            )["would_handoff"]
        ),
        None,
    )


async def _account_and_domain(account_id: str, agent_email: str):
    async with SessionFactory() as session:
        account = await session.get(Account, UUID(account_id))
        domain = await get_domain_by_name(session, agent_email.rsplit("@", 1)[-1])
    if account is None:
        raise HandoffPreflightError("Handoff account no longer exists.")
    if domain is None:
        raise HandoffPreflightError("Agent mail domain is not configured.")
    if domain.domain_type == "company" and domain.status != "active":
        raise HandoffPreflightError("Agent mail domain is not active.")
    sender_name = " ".join(
        part.strip()
        for part in [account.first_name, account.last_name]
        if part and part.strip()
    )
    return account, domain, sender_name


async def create_handoff_thread(
    database,
    *,
    account_id: str,
    subject: str,
) -> ObjectId:
    now = _now()
    result = await database.email_threads.insert_one(
        {
            "account_id": account_id,
            "thread_type": THREAD_TYPE_HUMAN_HANDOFF,
            "normalized_subject": subject.casefold(),
            "created_at": now,
            "updated_at": now,
            "last_message_at": now,
            "latest_message_id": None,
            "message_count": 0,
        }
    )
    return result.inserted_id


async def _latest_thread_message(
    database,
    *,
    account_id: str,
    thread_id: ObjectId,
    scope: str,
    exclude_message_ids: list | None = None,
) -> dict | None:
    match = {
        "account_id": account_id,
        "thread_id": thread_id,
        "message_scope": scope,
        "deletion_state": {"$in": [None, ""]},
    }
    if exclude_message_ids:
        match["message_id"] = {"$nin": list(exclude_message_ids)}
    pipeline = [
        {"$match": match},
        {"$sort": {"event_at": -1, "_id": -1}},
        {"$limit": 1},
        {
            "$lookup": {
                "from": "messages",
                "localField": "message_id",
                "foreignField": "_id",
                "as": "message",
            }
        },
        {"$unwind": "$message"},
    ]
    cursor = await database.mailbox_entries.aggregate(pipeline)
    rows = [row async for row in cursor]
    return rows[0]["message"] if rows else None


async def find_handoff_for_external_thread(
    database,
    *,
    account_id: str,
    external_thread_id: ObjectId,
) -> dict | None:
    return await database.human_handoffs.find_one(
        {
            "account_id": account_id,
            "external_thread_id": external_thread_id,
            "status": {"$ne": "CLOSED"},
        }
    )


async def resolve_handoff_from_inbound(
    database,
    *,
    account_id: str,
    agent_email: str,
    sender_address: str,
    in_reply_to: str | None,
    references: list[str],
) -> dict | None:
    sender = sender_address.strip().lower()
    if not sender or not agent_email:
        return None
    search_ids = parse_message_id_list(
        " ".join([*([in_reply_to] if in_reply_to else []), *(references or [])])
    )
    for internet_id in reversed(search_ids):
        folded = folded_message_id(internet_id)
        if not folded:
            continue
        parent = await database.messages.find_one(
            {
                "message_scope": MESSAGE_SCOPE_HANDOFF,
                "$or": [
                    {"message_id_folded": folded},
                    {"message_id": canonical_message_id(internet_id)},
                ],
            },
            {"_id": 1},
        )
        if parent is None:
            continue
        entry = await database.mailbox_entries.find_one(
            {
                "account_id": account_id,
                "message_id": parent["_id"],
                "message_scope": MESSAGE_SCOPE_HANDOFF,
            },
            {"thread_id": 1},
        )
        if entry is None:
            continue
        handoff = await database.human_handoffs.find_one(
            {
                "account_id": account_id,
                "handoff_thread_id": entry["thread_id"],
                "human_email": sender,
                "agent_email": agent_email.lower(),
                "status": {"$ne": "CLOSED"},
            }
        )
        if handoff:
            return handoff
    return None


async def _record_link(
    database,
    *,
    handoff_id: ObjectId,
    source_message_id: ObjectId,
    external_message_id: ObjectId | None,
    handoff_message_id: ObjectId | None,
    relationship: str,
) -> None:
    await database.handoff_message_links.update_one(
        {
            "handoff_id": handoff_id,
            "source_message_id": source_message_id,
            "relationship": relationship,
        },
        {
            "$setOnInsert": {
                "handoff_id": handoff_id,
                "source_message_id": source_message_id,
                "external_message_id": external_message_id,
                "handoff_message_id": handoff_message_id,
                "relationship": relationship,
                "created_at": _now(),
            }
        },
        upsert=True,
    )


def _attachment_metadata(parts) -> list[dict]:
    return [
        {
            "filename": part.filename,
            "content_type": part.content_type,
            "size": len(part.payload),
            "content_id": part.content_id,
            "disposition": part.disposition,
        }
        for part in parts
    ]


async def store_handoff_message(
    database,
    *,
    account_id: str,
    thread_id: ObjectId,
    direction: str,
    sender: dict,
    recipients: list[dict],
    subject: str,
    body_text: str,
    body_html: str | None,
    internet_message_id: str | None,
    in_reply_to: str | None,
    references: list[str],
    handoff_id: ObjectId,
    event_at: datetime | None = None,
    attachments: list[dict] | None = None,
    raw_mime: dict | None = None,
    handoff_delivery_key: str | None = None,
    new_reply_content: str | None = None,
    new_reply_html: str | None = None,
    quoted_content: str | None = None,
    message_origin: str | None = None,
    generated_by: str | None = None,
    mirrored_from_external_message_id: ObjectId | None = None,
    relay_status: str | None = None,
) -> dict:
    if handoff_delivery_key:
        existing = await database.messages.find_one(
            {"handoff_delivery_key": handoff_delivery_key}
        )
        if existing:
            return existing

    now = event_at or _now()
    id_fields = message_id_fields(canonical_message_id(internet_message_id))
    document = {
        "account_id": account_id,
        "direction": direction,
        "message_scope": MESSAGE_SCOPE_HANDOFF,
        "handoff_id": handoff_id,
        "sender": sender,
        "to": recipients,
        "subject": subject,
        "snippet": _snippet(new_reply_content or body_text),
        "body_text": body_text,
        "body_html": body_html,
        "new_reply_content": new_reply_content,
        "new_reply_html": new_reply_html,
        "quoted_content": quoted_content or "",
        "has_quoted_text": bool(quoted_content),
        "attachments": attachments or [],
        "created_at": now,
        "received_at": now if direction == "inbound" else None,
        "sent_at": now if direction == "outbound" else None,
        "status": "received" if direction == "inbound" else "accepted",
        **id_fields,
        "in_reply_to": canonical_message_id(in_reply_to),
        "references": references or [],
        "message_origin": message_origin,
        "generated_by": generated_by,
        "relay_status": relay_status,
    }
    if raw_mime:
        document["raw_mime"] = raw_mime
    if handoff_delivery_key:
        document["handoff_delivery_key"] = handoff_delivery_key
    if mirrored_from_external_message_id:
        document["mirrored_from_external_message_id"] = mirrored_from_external_message_id
    try:
        result = await database.messages.insert_one(document)
        document["_id"] = result.inserted_id
    except DuplicateKeyError:
        query = (
            {"handoff_delivery_key": handoff_delivery_key}
            if handoff_delivery_key
            else {"mirrored_from_external_message_id": mirrored_from_external_message_id}
        )
        existing = await database.messages.find_one(query)
        if existing:
            return existing
        raise

    await attach_message_to_thread(
        database,
        account_id=account_id,
        thread_id=thread_id,
        message_id=document["_id"],
        subject=subject,
        event_at=now,
    )
    mailbox = "handoff_inbox" if direction == "inbound" else "handoff_outbox"
    await database.mailbox_entries.update_one(
        {
            "account_id": account_id,
            "message_id": document["_id"],
            "mailbox": mailbox,
        },
        {
            "$setOnInsert": {
                "account_id": account_id,
                "message_id": document["_id"],
                "thread_id": thread_id,
                "mailbox": mailbox,
                "message_scope": MESSAGE_SCOPE_HANDOFF,
                "direction": direction,
                "sender": sender,
                "recipients": recipients,
                "subject": subject,
                "snippet": document["snippet"],
                "attachments": attachments or [],
                "status": document["status"],
                "is_read": direction == "outbound",
                "is_starred": False,
                "event_at": now,
                "created_at": now,
                "trashed_at": None,
                "deletion_state": None,
            }
        },
        upsert=True,
    )
    return document


async def _create_handoff_mapping(
    database,
    *,
    account_id: str,
    agent_email: str,
    external_message: dict,
    external_thread_id: ObjectId,
    rule: dict,
) -> dict:
    sender = external_message.get("sender") or {}
    sender_email = (sender.get("address") or "").strip()
    customer_name = sender_display_name(sender)
    private_subject = build_handoff_subject(
        sender_name=customer_name,
        external_sender=sender_email or customer_name,
        external_subject=external_message.get("subject", ""),
    )
    thread_id = await create_handoff_thread(
        database,
        account_id=account_id,
        subject=private_subject,
    )
    now = _now()
    document = {
        "account_id": account_id,
        "external_thread_id": external_thread_id,
        "handoff_thread_id": thread_id,
        "rule_id": rule["_id"],
        "rule_name": rule.get("name") or rule.get("owner_name") or "Human",
        "rule_priority": rule.get("priority", 0),
        "human_email": rule["owner_email"].lower(),
        "human_name": rule.get("owner_name") or rule["owner_email"].split("@", 1)[0],
        "agent_email": agent_email.lower(),
        "external_sender": sender_email.lower(),
        "external_sender_name": customer_name,
        "external_subject": external_message.get("subject", ""),
        "private_subject": private_subject,
        "allow_replies": rule.get("allow_replies", True),
        "status": HANDOFF_STATUS_WAITING,
        "current_owner": OWNER_HUMAN,
        "trigger_external_message_id": external_message["_id"],
        "trigger_internet_message_id": external_message.get("internet_message_id")
        or external_message.get("message_id"),
        "created_at": now,
        "updated_at": now,
    }
    try:
        result = await database.human_handoffs.insert_one(document)
        document["_id"] = result.inserted_id
        return document
    except DuplicateKeyError:
        await database.email_threads.delete_one({"_id": thread_id, "message_count": 0})
        existing = await find_handoff_for_external_thread(
            database,
            account_id=account_id,
            external_thread_id=external_thread_id,
        )
        if existing:
            return existing
        raise


async def maybe_trigger_handoff(
    database,
    *,
    account_id: str,
    agent_email: str,
    external_message: dict,
    external_thread_id: ObjectId,
) -> dict | None:
    if not should_mirror_to_human(external_message):
        return None
    sender_address = (
        (external_message.get("sender") or {}).get("address") or ""
    ).lower()
    if sender_address == agent_email.lower():
        return None

    handoff = await find_handoff_for_external_thread(
        database,
        account_id=account_id,
        external_thread_id=external_thread_id,
    )
    if handoff is None:
        if external_message.get("direction") != "inbound":
            return None
        owner_sender = await database.human_handoff_rules.find_one(
            {
                "account_id": account_id,
                "owner_email": sender_address,
                "active": True,
            },
            {"_id": 1},
        )
        if owner_sender:
            return None
        rules = await database.human_handoff_rules.find(
            {"account_id": account_id, "active": True}
        ).sort([("priority", 1), ("created_at", 1), ("_id", 1)]).to_list(length=100)
        email_payload = {
            "subject": external_message.get("subject", ""),
            "snippet": external_message.get("snippet", ""),
            "body_text": external_message.get("body_text", ""),
            "is_starred": False,
        }
        rule = select_matching_handoff_rule(
            rules,
            email=email_payload,
        )
        if rule is None:
            return None
        handoff = await _create_handoff_mapping(
            database,
            account_id=account_id,
            agent_email=agent_email,
            external_message=external_message,
            external_thread_id=external_thread_id,
            rule=rule,
        )

    await enqueue_handoff_delivery(
        database,
        kind=JOB_MIRROR,
        account_id=account_id,
        handoff_id=handoff["_id"],
        source_message_id=external_message["_id"],
        agent_email=agent_email,
    )
    return handoff


async def process_human_reply(
    database,
    *,
    account_id: str,
    agent_email: str,
    handoff: dict,
    parsed: dict,
    body_text: str,
    body_html: str | None,
    raw_mime: dict,
    delivery_key: str,
) -> dict:
    extracted = extract_new_reply_content(
        body_text=body_text,
        body_html=body_html,
    )
    rich = None
    try:
        raw = await get_raw_mime_storage().get(raw_mime)
        rich = extract_new_reply_mime_content(raw)
    except Exception:
        logger.exception("handoff_reply_rich_extract_failed delivery=%s", delivery_key)

    human_email = (parsed.get("sender") or {}).get("address", "").lower()
    new_text = (
        rich.body_text if rich and rich.body_text else extracted["new_reply_content"]
    ).strip()
    new_html = rich.body_html if rich else None
    if human_email:
        new_text = redact_human_address(
            new_text,
            human_email=human_email,
            agent_email=agent_email,
        )
        new_html = redact_human_address(
            new_html,
            human_email=human_email,
            agent_email=agent_email,
        )
    auto_submitted = (
        (parsed.get("raw_headers") or {}).get("auto_submitted") or ""
    ).strip().lower()
    is_automatic = bool(auto_submitted and auto_submitted != "no")
    private_markers_ok = validate_external_reply_body(new_text)
    relay_status = (
        RELAY_STATUS_PENDING
        if (
            new_text
            and private_markers_ok
            and not is_automatic
            and handoff.get("allow_replies", True)
        )
        else RELAY_STATUS_BLOCKED
    )
    human_message = await store_handoff_message(
        database,
        account_id=account_id,
        thread_id=handoff["handoff_thread_id"],
        direction="inbound",
        sender=parsed.get("sender") or {"name": "", "address": human_email},
        recipients=parsed.get("to", []),
        subject=parsed.get("subject", ""),
        body_text=body_text,
        body_html=body_html,
        internet_message_id=parsed.get("message_id"),
        in_reply_to=parsed.get("in_reply_to"),
        references=parsed.get("references", []),
        handoff_id=handoff["_id"],
        event_at=parsed.get("sent_at") or _now(),
        attachments=parsed.get("attachments", []),
        raw_mime=raw_mime,
        handoff_delivery_key=f"handoff:{account_id}:{handoff['_id']}:{delivery_key}",
        new_reply_content=new_text,
        new_reply_html=new_html,
        quoted_content=extracted.get("quoted_content", ""),
        message_origin=MESSAGE_ORIGIN_HUMAN_OWNER,
        generated_by=GENERATED_BY_HUMAN,
        relay_status=relay_status,
    )
    if relay_status == RELAY_STATUS_PENDING:
        await enqueue_handoff_delivery(
            database,
            kind=JOB_RELAY,
            account_id=account_id,
            handoff_id=handoff["_id"],
            source_message_id=human_message["_id"],
            agent_email=agent_email,
        )
    else:
        await database.messages.update_one(
            {"_id": human_message["_id"]},
            {
                "$set": {
                    "relay_error": (
                        "Human replies are disabled for this handoff."
                        if not handoff.get("allow_replies", True)
                        else (
                            "Automatic replies are not relayed through Human Handoff."
                            if is_automatic
                            else "Reply content could not be safely separated from private history."
                        )
                    )
                }
            },
        )
    await database.human_handoffs.update_one(
        {"_id": handoff["_id"]},
        {
            "$set": {
                "status": HANDOFF_STATUS_HUMAN_REPLIED,
                "current_owner": OWNER_AGENT,
                "last_human_message_at": _now(),
                "updated_at": _now(),
            }
        },
    )
    return human_message


def _mirror_from_identity(
    source: dict, agent_name: str, agent_email: str
) -> tuple[str, str]:
    if source.get("direction") == "inbound":
        sender = source.get("sender") or {}
        address = (sender.get("address") or "").strip()
        return sender_display_name(sender), address or agent_email
    return agent_name, agent_email


def _body_has_quote_trail(body_text: str | None, body_html: str | None) -> bool:
    html = body_html or ""
    text = body_text or ""
    if re.search(r"gmail_quote|blockquote", html, re.IGNORECASE):
        return True
    if re.search(r"(?m)^>+", text):
        return True
    if re.search(r"wrote:\s*$", text, re.IGNORECASE | re.MULTILINE):
        return True
    return "-----Original Message-----" in text


async def _mirror_body_with_trail(
    database,
    *,
    account_id: str,
    handoff: dict,
    source: dict,
    payload,
) -> tuple[str, str | None]:
    text = payload.body_text or ""
    html = payload.body_html
    if _body_has_quote_trail(text, html):
        return text, html or None
    parent = await _latest_thread_message(
        database,
        account_id=account_id,
        thread_id=handoff["external_thread_id"],
        scope=MESSAGE_SCOPE_EXTERNAL,
        exclude_message_ids=[source["_id"]],
    )
    if parent is None:
        return text, html
    parent_payload = await load_mirror_payload(parent)
    quoted_text, quoted_html = build_gmail_quote(
        sender=parent.get("sender") or {},
        event_at=(
            parent.get("sent_at")
            or parent.get("received_at")
            or parent.get("created_at")
        ),
        body_text=parent_payload.body_text or parent.get("body_text") or "",
        body_html=parent_payload.body_html or parent.get("body_html"),
    )
    return attach_quoted_history(
        new_text=(source.get("new_reply_content") or text).strip() or text,
        new_html=source.get("new_reply_html") or html,
        quoted_text=quoted_text,
        quoted_html=quoted_html,
    )


async def _deliver_mirror(database, job: dict, handoff: dict, source: dict) -> ObjectId:
    existing = await database.messages.find_one(
        {
            "message_scope": MESSAGE_SCOPE_HANDOFF,
            "mirrored_from_external_message_id": source["_id"],
        }
    )
    if existing:
        return existing["_id"]

    _, domain, sender_name = await _account_and_domain(
        job["account_id"], handoff["agent_email"]
    )
    payload = await load_mirror_payload(source)
    body_text, body_html = await _mirror_body_with_trail(
        database,
        account_id=job["account_id"],
        handoff=handoff,
        source=source,
        payload=payload,
    )
    from_name, from_address = _mirror_from_identity(
        source, sender_name, handoff["agent_email"]
    )
    parent = await _latest_thread_message(
        database,
        account_id=job["account_id"],
        thread_id=handoff["handoff_thread_id"],
        scope=MESSAGE_SCOPE_HANDOFF,
    )
    threading = threading_fields_from_parent(parent) if parent else {}
    subject = handoff["private_subject"]
    if parent and not re.match(r"^re:\s*", subject, re.IGNORECASE):
        subject = f"Re: {subject}"
    internet_message_id = await deliver_email(
        sender=from_address,
        sender_name=from_name,
        recipients=[handoff["human_email"]],
        cc=[],
        bcc=[],
        subject=subject,
        body_text=body_text,
        body_html=body_html,
        domain=domain,
        reply_to=handoff["agent_email"],
        envelope_sender=handoff["agent_email"],
        in_reply_to=threading.get("in_reply_to"),
        references=threading.get("references"),
        message_id=job["generated_message_id"],
        inline_parts=payload.inline_parts,
        file_attachments=payload.file_attachments,
    )
    attachments = _attachment_metadata(
        [*payload.inline_parts, *payload.file_attachments]
    )
    mirror = await store_handoff_message(
        database,
        account_id=job["account_id"],
        thread_id=handoff["handoff_thread_id"],
        direction="outbound",
        sender={"name": from_name, "address": from_address},
        recipients=[
            {
                "name": handoff.get("human_name", ""),
                "address": handoff["human_email"],
            }
        ],
        subject=subject,
        body_text=body_text,
        body_html=body_html,
        internet_message_id=internet_message_id,
        in_reply_to=threading.get("in_reply_to"),
        references=threading.get("references", []),
        handoff_id=handoff["_id"],
        attachments=attachments,
        message_origin=MESSAGE_ORIGIN_HANDOFF_MIRROR,
        generated_by=GENERATED_BY_HANDOFF_BRIDGE,
        mirrored_from_external_message_id=source["_id"],
    )
    await _record_link(
        database,
        handoff_id=handoff["_id"],
        source_message_id=source["_id"],
        external_message_id=source["_id"],
        handoff_message_id=mirror["_id"],
        relationship=LINK_EXTERNAL_MIRRORED,
    )
    await database.human_handoffs.update_one(
        {"_id": handoff["_id"]},
        {
            "$set": {
                "status": HANDOFF_STATUS_WAITING,
                "current_owner": OWNER_HUMAN,
                "last_mirrored_at": _now(),
                "updated_at": _now(),
            }
        },
    )
    return mirror["_id"]


async def _deliver_relay(database, job: dict, handoff: dict, source: dict) -> ObjectId:
    existing = await database.messages.find_one(
        {
            "message_scope": MESSAGE_SCOPE_EXTERNAL,
            "generated_from_handoff_message_id": source["_id"],
        }
    )
    if existing:
        return existing["_id"]
    new_text = (source.get("new_reply_content") or "").strip()
    if not new_text or not validate_external_reply_body(new_text):
        raise HandoffPreflightError(
            "Reply content failed the private-content leakage guard."
        )
    external_parent = await resolve_external_reply_parent(
        database,
        account_id=job["account_id"],
        external_thread_id=handoff["external_thread_id"],
        customer_address=handoff["external_sender"],
        handoff=handoff,
    )
    if external_parent is None:
        raise HandoffPreflightError("No valid external reply parent was found.")
    parent_internet_id = await resolve_internet_message_id(database, external_parent)
    if not parent_internet_id:
        raise HandoffPreflightError(
            "The external reply parent has no RFC Message-ID."
        )
    recipient = handoff.get("external_sender") or (
        external_parent.get("sender") or {}
    ).get("address")
    if not recipient:
        raise HandoffPreflightError("The external sender address is missing.")

    _, domain, sender_name = await _account_and_domain(
        job["account_id"], handoff["agent_email"]
    )
    inline_parts = []
    file_attachments = []
    body_html = source.get("new_reply_html")
    if source.get("raw_mime"):
        raw = await get_raw_mime_storage().get(source["raw_mime"])
        rich = extract_new_reply_mime_content(raw)
        inline_parts = rich.inline_parts
        file_attachments = rich.file_attachments
    quoted_text, quoted_html = build_gmail_quote(
        sender=external_parent.get("sender") or {},
        event_at=(
            external_parent.get("sent_at")
            or external_parent.get("received_at")
            or external_parent.get("created_at")
        ),
        body_text=external_parent.get("body_text") or external_parent.get("snippet") or "",
        body_html=external_parent.get("body_html"),
    )
    relay_text, relay_html = attach_quoted_history(
        new_text=new_text,
        new_html=body_html,
        quoted_text=quoted_text,
        quoted_html=quoted_html,
    )
    plan = build_external_reply_plan(
        parent_external_message=external_parent,
        body_text=relay_text,
        body_html=relay_html,
        subject_override=None,
        recipients=[recipient],
        parent_internet_id_override=parent_internet_id,
    )
    try:
        external = await send_external_reply(
            database,
            account_id=job["account_id"],
            sender=handoff["agent_email"],
            sender_name=sender_name,
            external_thread_id=handoff["external_thread_id"],
            plan=plan,
            domain=domain,
            resolution_method="human_handoff_bridge",
            generated_by=GENERATED_BY_HANDOFF_BRIDGE,
            generated_from_handoff_message_id=source["_id"],
            message_origin=MESSAGE_ORIGIN_HUMAN_OWNER,
            message_id=job["generated_message_id"],
            inline_parts=inline_parts,
            file_attachments=file_attachments,
        )
    except ExternalReplyThreadingError as error:
        raise HandoffPreflightError(str(error)) from error
    await database.messages.update_one(
        {"_id": source["_id"]},
        {
            "$set": {
                "relay_status": RELAY_STATUS_SENT,
                "generated_external_message_id": external["_id"],
                "relay_delivered_at": _now(),
            }
        },
    )
    await _record_link(
        database,
        handoff_id=handoff["_id"],
        source_message_id=source["_id"],
        external_message_id=external["_id"],
        handoff_message_id=source["_id"],
        relationship=LINK_HUMAN_RELAYED,
    )
    await database.human_handoffs.update_one(
        {"_id": handoff["_id"]},
        {
            "$set": {
                "status": HANDOFF_STATUS_ACTIVE,
                "current_owner": OWNER_AGENT,
                "updated_at": _now(),
            }
        },
    )
    return external["_id"]


async def deliver_handoff_job(database, job: dict) -> ObjectId:
    handoff = await database.human_handoffs.find_one(
        {"_id": job["handoff_id"], "account_id": job["account_id"]}
    )
    if handoff is None:
        raise HandoffPreflightError("Linked handoff no longer exists.")
    source = await database.messages.find_one({"_id": job["source_message_id"]})
    if source is None:
        raise HandoffPreflightError("Source message no longer exists.")
    if job["kind"] == JOB_MIRROR:
        return await _deliver_mirror(database, job, handoff, source)
    if job["kind"] == JOB_RELAY:
        return await _deliver_relay(database, job, handoff, source)
    raise HandoffPreflightError(f"Unknown handoff delivery kind: {job['kind']}")