From 89ddebdb60b6064eb17e38c31806e44632f450fd Mon Sep 17 00:00:00 2001 From: debu-2 Date: Tue, 6 Oct 2026 02:44:54 +0530 Subject: [PATCH] handoff fixes --- backend/app/routers/emails.py | 46 +++- .../app/services/external_reply_service.py | 210 +++++++++++++++ .../app/services/handoff_email_templates.py | 241 +++++++----------- backend/app/services/handoff_mirror.py | 108 ++++++++ backend/app/services/html_sanitizer.py | 27 ++ backend/app/services/human_handoff_service.py | 132 +++------- backend/app/services/inbound_smtp.py | 95 ++++++- backend/app/services/mail_sender.py | 67 ++++- backend/app/services/mime_body_extractor.py | 106 ++++++++ backend/app/services/thread_resolver.py | 2 +- backend/app/services/thread_service.py | 4 +- .../tests/test_email_threading_roundtrip.py | 76 ++++++ backend/tests/test_external_reply_service.py | 57 +++++ backend/tests/test_handoff_mirror.py | 85 ++++++ backend/tests/test_reply_extractor.py | 6 +- 15 files changed, 991 insertions(+), 271 deletions(-) create mode 100644 backend/app/services/external_reply_service.py create mode 100644 backend/app/services/handoff_mirror.py create mode 100644 backend/app/services/html_sanitizer.py create mode 100644 backend/app/services/mime_body_extractor.py create mode 100644 backend/tests/test_email_threading_roundtrip.py create mode 100644 backend/tests/test_external_reply_service.py create mode 100644 backend/tests/test_handoff_mirror.py diff --git a/backend/app/routers/emails.py b/backend/app/routers/emails.py index 8acaecc..528736f 100644 --- a/backend/app/routers/emails.py +++ b/backend/app/routers/emails.py @@ -32,8 +32,8 @@ from app.services.thread_service import ( create_thread, ensure_thread_for_message, message_event_at, - threading_fields_from_parent, ) +from app.services.external_reply_service import build_external_reply_plan, log_smtp_message_sending from app.services.mailbox_scope import apply_external_scope from app.services.thread_resolver import ThreadResolution @@ -150,7 +150,6 @@ async def create_email( detail="Reply target not found.", ) thread_id = parent_entry.get("thread_id") - threading_fields = threading_fields_from_parent(parent_message) to_addresses, cc_addresses = reply_recipients( message=parent_message, viewer_email=sender, @@ -187,9 +186,20 @@ async def create_email( created_at = utc_now() subject = payload.subject - if parent_message and not subject.strip(): - parent_subject = parent_message.get("subject", "") - subject = parent_subject if parent_subject.lower().startswith("re:") else f"Re: {parent_subject}" + if parent_message: + reply_plan = build_external_reply_plan( + parent_external_message=parent_message, + body_text=payload.body_text, + body_html=payload.body_html, + recipients=recipients, + ) + threading_fields = { + "in_reply_to": reply_plan.threading.in_reply_to, + "references": reply_plan.threading.references, + "parent_message_id": reply_plan.threading.parent_internal_message_id, + } + if not subject.strip(): + subject = reply_plan.subject document = { "direction": "outbound", "message_scope": MESSAGE_SCOPE_EXTERNAL, @@ -215,6 +225,17 @@ async def create_email( } try: + log_smtp_message_sending( + generated_message_id=None, + in_reply_to=threading_fields.get("in_reply_to"), + references=threading_fields.get("references"), + sender=sender, + recipients=recipients, + subject=document["subject"], + external_thread_id=thread_id, + replying_to_internal_message_id=threading_fields.get("parent_message_id"), + replying_to_internet_message_id=threading_fields.get("in_reply_to"), + ) internet_message_id = await deliver_email( sender=sender, recipients=recipients, @@ -228,7 +249,20 @@ async def create_email( references=threading_fields.get("references"), ) document["status"] = "accepted" - document["message_id"] = normalize_message_id(internet_message_id) + normalized_id = normalize_message_id(internet_message_id) + document["message_id"] = normalized_id + document["internet_message_id"] = normalized_id + log_smtp_message_sending( + generated_message_id=normalized_id, + in_reply_to=threading_fields.get("in_reply_to"), + references=threading_fields.get("references"), + sender=sender, + recipients=recipients, + subject=document["subject"], + external_thread_id=thread_id, + replying_to_internal_message_id=threading_fields.get("parent_message_id"), + replying_to_internet_message_id=threading_fields.get("in_reply_to"), + ) except Exception as error: document["status"] = "failed" document["error"] = str(error) diff --git a/backend/app/services/external_reply_service.py b/backend/app/services/external_reply_service.py new file mode 100644 index 0000000..51361d8 --- /dev/null +++ b/backend/app/services/external_reply_service.py @@ -0,0 +1,210 @@ +import logging +import re +from dataclasses import dataclass +from datetime import datetime, timezone + +from bson import ObjectId + +from app.models import MailDomain +from app.services.handoff_email_templates import external_reply_subject +from app.services.mail_sender import deliver_email +from app.services.message_ids import normalize_message_id +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" + + +@dataclass(frozen=True) +class ExternalReplyThreading: + in_reply_to: str | None + references: list[str] + parent_internal_message_id: ObjectId + replying_to_internet_message_id: str | None + + +@dataclass(frozen=True) +class ExternalReplyPlan: + subject: str + body_text: str + body_html: str | None + threading: ExternalReplyThreading + recipients: list[str] + + +def internet_message_id_of(message: dict) -> str | None: + return normalize_message_id( + message.get("internet_message_id") or message.get("message_id") + ) + + +def build_external_reply_plan( + *, + parent_external_message: dict, + body_text: str, + body_html: str | None = None, + subject_override: str | None = None, + recipients: list[str], +) -> ExternalReplyPlan: + threading_fields = threading_fields_from_parent(parent_external_message) + parent_internet_id = internet_message_id_of(parent_external_message) + if subject_override: + subject = subject_override + else: + parent_subject = parent_external_message.get("subject", "") + subject = ( + parent_subject + if re.match(r"^re:\s*", parent_subject, flags=re.IGNORECASE) + else external_reply_subject(parent_subject) + ) + return ExternalReplyPlan( + subject=subject, + body_text=body_text, + body_html=body_html, + threading=ExternalReplyThreading( + in_reply_to=threading_fields.get("in_reply_to"), + references=threading_fields.get("references", []), + parent_internal_message_id=parent_external_message["_id"], + replying_to_internet_message_id=parent_internet_id, + ), + recipients=recipients, + ) + + +def log_smtp_message_sending( + *, + generated_message_id: str | None, + in_reply_to: str | None, + references: list[str] | None, + sender: str, + recipients: list[str], + subject: str, + external_thread_id: ObjectId | str | None = None, + replying_to_internal_message_id: ObjectId | str | None = None, + replying_to_internet_message_id: str | None = None, +) -> None: + logger.info( + "smtp_message_sending generated_message_id=%s in_reply_to=%s references=%s " + "from=%s to=%s subject=%s external_thread_id=%s " + "replying_to_internal_message_id=%s replying_to_internet_message_id=%s", + generated_message_id, + in_reply_to, + " ".join(references or []), + sender, + ",".join(recipients), + subject, + external_thread_id, + replying_to_internal_message_id, + replying_to_internet_message_id, + ) + + +async def send_external_reply( + database, + *, + account_id: str, + sender: str, + sender_name: str, + external_thread_id: ObjectId, + plan: ExternalReplyPlan, + domain: MailDomain, + resolution_method: str = "in_reply_to", +) -> dict: + log_smtp_message_sending( + generated_message_id=None, + in_reply_to=plan.threading.in_reply_to, + references=plan.threading.references, + sender=sender, + recipients=plan.recipients, + subject=plan.subject, + external_thread_id=external_thread_id, + replying_to_internal_message_id=plan.threading.parent_internal_message_id, + replying_to_internet_message_id=plan.threading.replying_to_internet_message_id, + ) + + internet_message_id = await deliver_email( + sender=sender, + recipients=plan.recipients, + cc=[], + bcc=[], + subject=plan.subject, + body_text=plan.body_text, + body_html=plan.body_html, + domain=domain, + in_reply_to=plan.threading.in_reply_to, + references=plan.threading.references, + ) + normalized_id = normalize_message_id(internet_message_id) + log_smtp_message_sending( + generated_message_id=normalized_id, + in_reply_to=plan.threading.in_reply_to, + references=plan.threading.references, + sender=sender, + recipients=plan.recipients, + subject=plan.subject, + external_thread_id=external_thread_id, + replying_to_internal_message_id=plan.threading.parent_internal_message_id, + replying_to_internet_message_id=plan.threading.replying_to_internet_message_id, + ) + + now = datetime.now(timezone.utc) + document = { + "direction": "outbound", + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "account_id": account_id, + "sender": {"name": sender_name, "address": sender}, + "to": [{"name": value.split("@")[0], "address": value} for value in plan.recipients], + "subject": plan.subject, + "snippet": re.sub(r"\s+", " ", plan.body_text).strip()[:240], + "body_text": plan.body_text, + "body_html": plan.body_html, + "attachments": [], + "created_at": now, + "sent_at": now, + "status": "accepted", + "message_id": normalized_id, + "internet_message_id": normalized_id, + "in_reply_to": plan.threading.in_reply_to, + "references": plan.threading.references, + "parent_message_id": plan.threading.parent_internal_message_id, + "thread_relation": "reply", + "thread_resolution_method": resolution_method, + } + result = await database.messages.insert_one(document) + document["_id"] = result.inserted_id + await attach_message_to_thread( + database, + account_id=account_id, + thread_id=external_thread_id, + message_id=result.inserted_id, + subject=plan.subject, + event_at=now, + ) + await database.mailbox_entries.insert_one( + { + "account_id": account_id, + "message_id": result.inserted_id, + "thread_id": external_thread_id, + "mailbox": "sent", + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "outbound", + "sender": document["sender"], + "recipients": document["to"], + "subject": plan.subject, + "snippet": document["snippet"], + "attachments": [], + "status": "accepted", + "is_read": True, + "is_starred": False, + "event_at": now, + "created_at": now, + "trashed_at": None, + "deletion_state": None, + } + ) + return document diff --git a/backend/app/services/handoff_email_templates.py b/backend/app/services/handoff_email_templates.py index 73f7b7e..f63451e 100644 --- a/backend/app/services/handoff_email_templates.py +++ b/backend/app/services/handoff_email_templates.py @@ -1,12 +1,5 @@ from html import escape -DEBUG_REASON_MARKERS = ( - "step passes", - "classifier", - "rule evaluation", - "no handoff instructions were set", -) - def _clean_external_subject(subject: str) -> str: value = (subject or "(No subject)").strip() @@ -28,171 +21,129 @@ def external_reply_subject(external_subject: str) -> str: return f"Re: {cleaned}" -def format_reason_for_human(reason: str | None, *, is_update: bool = False) -> str | None: - if is_update: - return None - if not reason or not reason.strip(): - return None - lower = reason.strip().lower() - if any(marker in lower for marker in DEBUG_REASON_MARKERS): - return "This conversation matched your Human Handoff settings." - if lower in { - "matched handoff rules.", - "new message in the linked conversation.", - }: - return "This conversation requires human attention." - return reason.strip() +def format_sender_line(sender: dict) -> str: + name = (sender.get("name") or "").strip() + address = (sender.get("address") or "").strip() + if not address: + return "unknown sender" + local = address.split("@", 1)[0].lower() + if name and name.lower() not in {local, address.lower()}: + return f"{name} <{address}>" + return address -def build_initial_handoff_text( - *, - external_sender: str, - external_subject: str, - external_body: str, - agent_email: str, - handoff_reason: str | None, -) -> str: - reason = format_reason_for_human(handoff_reason) - lines = [ - "GENTIC — HUMAN HANDOFF", - "", - "This conversation needs your attention.", - "", - f"From: {external_sender}", - f"Subject: {_clean_external_subject(external_subject)}", - "", - "LATEST MESSAGE", - "", - external_body.strip(), - "", - ] - if reason: - lines.extend(["WHY YOU'RE RECEIVING THIS", "", reason, ""]) - lines.extend( - [ - "REPLY NORMALLY", - "", - f"Reply to this email and Gentic will send only your new reply to the " - f"original sender through {agent_email}.", - "", - "This private handoff conversation and quoted history will not be shared " - "with the sender.", - "", - "—", - "Gentic Human Handoff", - "Private Agent ↔ Human conversation", - ] +def _footer_html(agent_email: str) -> str: + return f""" +
+

↩ Reply normally

+

+ Your reply will be sent to the original sender through + {escape(agent_email)}. + Only the new text you write will be sent. +

+
""" + + +def _footer_text(agent_email: str) -> str: + return ( + "↩ Reply normally\n\n" + f"Your reply will be sent to the original sender through {agent_email}.\n" + "Only the new text you write will be sent." ) - return "\n".join(lines) -def build_initial_handoff_html( +def _header_html(sender_line: str, external_subject: str) -> str: + subject = escape(_clean_external_subject(external_subject)) + return f""" +
+
✦ Human Handoff
+
+ {escape(sender_line)} + | {subject} +
+
""" + + +def _header_text(sender_line: str, external_subject: str) -> str: + return f"✦ Human Handoff\n{sender_line} | {_clean_external_subject(external_subject)}" + + +def build_mirror_handoff_html( *, - external_sender: str, + sender_line: str, external_subject: str, - external_body: str, + original_html: str, agent_email: str, - handoff_reason: str | None, ) -> str: - reason = format_reason_for_human(handoff_reason) - reason_block = "" - if reason: - reason_block = f""" - -

Why you're receiving this

-

{escape(reason)}

- """ - return f""" - - - -
- - - - - {reason_block} - - -
-

✦ Gentic

-

Human Handoff

-

This conversation needs your attention.

-
- - -
-

From

-

{escape(external_sender)}

-

Subject

-

{escape(_clean_external_subject(external_subject))}

-
-
-

Latest message

-
{escape(external_body.strip())}
-
- - -
-

↩ Reply normally

-

- Reply to this email and Gentic will send your response to the original sender as - {escape(agent_email)}. Only the new text you write will be sent. - This handoff email and quoted history stay private. -

-
-
-

Gentic Human Handoff
Private Agent ↔ Human conversation

-
-
+ +{_header_html(sender_line, external_subject)} +
+{original_html} +
+{_footer_html(agent_email)} """ -def build_update_handoff_text( +def build_mirror_handoff_text( *, - external_sender: str, - external_body: str, + sender_line: str, + external_subject: str, + original_body_text: str, agent_email: str, ) -> str: - return "\n".join( + return "\n\n".join( [ - "Customer replied", - "", - external_sender, - "", - external_body.strip(), - "", - f"Reply normally to respond through {agent_email}.", + _header_text(sender_line, external_subject), + "─" * 56, + original_body_text.strip(), + "─" * 56, + _footer_text(agent_email), ] ) -def build_update_handoff_html( +def build_mirror_update_html( *, - external_sender: str, - external_body: str, + sender_line: str, + external_subject: str, + original_html: str, agent_email: str, ) -> str: + subject = escape(_clean_external_subject(external_subject)) return f""" - - - -
- - -
-

✦ Gentic

-

Customer replied

-

{escape(external_sender)}

-
{escape(external_body.strip())}
-

- Reply normally to respond through {escape(agent_email)}. -

-
-
+ +
+
✦ Customer replied
+
+ {escape(sender_line)} + | {subject} +
+
+
+{original_html} +
+{_footer_html(agent_email)} """ +def build_mirror_update_text( + *, + sender_line: str, + external_subject: str, + original_body_text: str, + agent_email: str, +) -> str: + return "\n\n".join( + [ + f"✦ Customer replied\n{sender_line} | {_clean_external_subject(external_subject)}", + "─" * 56, + original_body_text.strip(), + "─" * 56, + _footer_text(agent_email), + ] + ) + + def build_relay_confirmation_text(customer_body: str) -> str: return f"Sent to customer:\n\n{customer_body.strip()}" diff --git a/backend/app/services/handoff_mirror.py b/backend/app/services/handoff_mirror.py new file mode 100644 index 0000000..7073dc4 --- /dev/null +++ b/backend/app/services/handoff_mirror.py @@ -0,0 +1,108 @@ +from dataclasses import dataclass, field + +from app.services.handoff_email_templates import ( + build_mirror_handoff_html, + build_mirror_handoff_text, + build_mirror_update_html, + build_mirror_update_text, + format_sender_line, +) +from app.services.html_sanitizer import plain_text_to_html, sanitize_email_html +from app.services.mime_body_extractor import ExtractedMimePart, MirroredMimeContent, extract_mirror_content +from app.services.raw_mime_storage import get_raw_mime_storage + + +@dataclass +class HandoffMirrorPayload: + body_text: str + body_html: str | None + inline_parts: list[ExtractedMimePart] = field(default_factory=list) + file_attachments: list[ExtractedMimePart] = field(default_factory=list) + + +async def load_mirror_payload(external_message: dict) -> HandoffMirrorPayload: + sender = external_message.get("sender") or {} + sender_line = format_sender_line(sender) + external_subject = external_message.get("subject", "") + agent_email = external_message.get("_agent_email", "") + + body_text = (external_message.get("body_text") or external_message.get("snippet") or "").strip() + body_html = external_message.get("body_html") + inline_parts: list[ExtractedMimePart] = [] + file_attachments: list[ExtractedMimePart] = [] + + if external_message.get("raw_mime"): + try: + raw = await get_raw_mime_storage().get(external_message["raw_mime"]) + extracted: MirroredMimeContent = extract_mirror_content(raw) + if extracted.body_text: + body_text = extracted.body_text + if extracted.body_html: + body_html = extracted.body_html + inline_parts = extracted.inline_parts + file_attachments = extracted.file_attachments + except Exception: + pass + + if body_html: + original_html = sanitize_email_html(body_html) + elif body_text: + original_html = plain_text_to_html(body_text) + else: + original_html = plain_text_to_html("(No message body)") + + if not body_text: + body_text = external_message.get("snippet") or "(No message body)" + + return HandoffMirrorPayload( + body_text=build_mirror_handoff_text( + sender_line=sender_line, + external_subject=external_subject, + original_body_text=body_text, + agent_email=agent_email, + ), + body_html=build_mirror_handoff_html( + sender_line=sender_line, + external_subject=external_subject, + original_html=original_html, + agent_email=agent_email, + ), + inline_parts=inline_parts, + file_attachments=file_attachments, + ) + + +async def load_mirror_update_payload( + external_message: dict, + *, + agent_email: str, +) -> HandoffMirrorPayload: + external_message = {**external_message, "_agent_email": agent_email} + payload = await load_mirror_payload(external_message) + sender = external_message.get("sender") or {} + sender_line = format_sender_line(sender) + external_subject = external_message.get("subject", "") + body_text = (external_message.get("body_text") or external_message.get("snippet") or "").strip() + + body_html = external_message.get("body_html") + if body_html: + original_html = sanitize_email_html(body_html) + else: + original_html = plain_text_to_html(body_text or "(No message body)") + + return HandoffMirrorPayload( + body_text=build_mirror_update_text( + sender_line=sender_line, + external_subject=external_subject, + original_body_text=body_text, + agent_email=agent_email, + ), + body_html=build_mirror_update_html( + sender_line=sender_line, + external_subject=external_subject, + original_html=original_html, + agent_email=agent_email, + ), + inline_parts=payload.inline_parts, + file_attachments=payload.file_attachments, + ) diff --git a/backend/app/services/html_sanitizer.py b/backend/app/services/html_sanitizer.py new file mode 100644 index 0000000..6e8ce7b --- /dev/null +++ b/backend/app/services/html_sanitizer.py @@ -0,0 +1,27 @@ +import re +from html import escape + +_UNSAFE_TAGS = re.compile( + r"<\s*(script|iframe|object|embed|form|base|meta|link)\b[^>]*>.*?", + re.IGNORECASE | re.DOTALL, +) +_UNSAFE_OPEN_TAGS = re.compile( + r"<\s*(script|iframe|object|embed|form|base)\b[^>]*/?>", + re.IGNORECASE, +) +_EVENT_HANDLERS = re.compile(r"\s+on[a-z]+\s*=\s*(['\"]).*?\1", re.IGNORECASE) +_JS_URLS = re.compile(r"""(href|src)\s*=\s*(['"])\s*javascript:.*?\2""", re.IGNORECASE) + + +def sanitize_email_html(html: str) -> str: + if not html: + return "" + cleaned = _UNSAFE_TAGS.sub("", html) + cleaned = _UNSAFE_OPEN_TAGS.sub("", cleaned) + cleaned = _EVENT_HANDLERS.sub("", cleaned) + cleaned = _JS_URLS.sub("", cleaned) + return cleaned.strip() + + +def plain_text_to_html(text: str) -> str: + return f'
{escape(text)}
' diff --git a/backend/app/services/human_handoff_service.py b/backend/app/services/human_handoff_service.py index 0155097..9f0f3e2 100644 --- a/backend/app/services/human_handoff_service.py +++ b/backend/app/services/human_handoff_service.py @@ -4,15 +4,16 @@ from datetime import datetime, timezone from bson import ObjectId +from app.services.external_reply_service import ( + build_external_reply_plan, + send_external_reply, +) from app.services.handoff_email_templates import ( - build_initial_handoff_html, - build_initial_handoff_text, build_relay_confirmation_text, - build_update_handoff_html, - build_update_handoff_text, external_reply_subject, handoff_email_subject, ) +from app.services.handoff_mirror import load_mirror_payload, load_mirror_update_payload from app.services.handoff_evaluator import evaluate_handoff from app.services.mail_sender import deliver_email from app.services.message_ids import normalize_message_id, parse_message_id_list @@ -297,6 +298,7 @@ async def store_handoff_message( "received_at": now if direction == "inbound" else None, "status": "received" if direction == "inbound" else "accepted", "message_id": normalize_message_id(internet_message_id), + "internet_message_id": normalize_message_id(internet_message_id), "in_reply_to": normalize_message_id(in_reply_to), "references": references or [], } @@ -355,28 +357,17 @@ async def send_handoff_to_human( domain, existing_handoff: dict | None = None, ) -> dict: - external_sender = external_message.get("sender", {}) - external_sender_text = external_sender.get("address") or "unknown sender" external_subject = external_message.get("subject", "") - external_body = external_message.get("body_text") or external_message.get("snippet") or "" - stable_subject = handoff_email_subject(external_subject) + mirror_source = {**external_message, "_agent_email": agent_email} + if existing_handoff: handoff_id = existing_handoff["_id"] handoff_thread_id = existing_handoff["handoff_thread_id"] parent = await _latest_handoff_message(database, account_id, handoff_thread_id) threading = threading_fields_from_parent(parent) if parent else {} subject = _handoff_reply_subject(external_subject) - body_text = build_update_handoff_text( - external_sender=external_sender_text, - external_body=external_body, - agent_email=agent_email, - ) - body_html = build_update_handoff_html( - external_sender=external_sender_text, - external_body=external_body, - agent_email=agent_email, - ) + mirror = await load_mirror_update_payload(mirror_source, agent_email=agent_email) relationship = LINK_EXTERNAL_UPDATE_RELAYED else: handoff_thread_id = await create_empty_handoff_thread( @@ -386,20 +377,8 @@ async def send_handoff_to_human( ) threading = {} subject = stable_subject - body_text = build_initial_handoff_text( - external_sender=external_sender_text, - external_subject=external_subject, - external_body=external_body, - handoff_reason=handoff_reason, - agent_email=agent_email, - ) - body_html = build_initial_handoff_html( - external_sender=external_sender_text, - external_subject=external_subject, - external_body=external_body, - handoff_reason=handoff_reason, - agent_email=agent_email, - ) + mirror = await load_mirror_payload(mirror_source) + external_sender_text = (external_message.get("sender") or {}).get("address") or "unknown sender" handoff_doc = { "account_id": account_id, "agent_email": agent_email.lower(), @@ -426,11 +405,13 @@ async def send_handoff_to_human( cc=[], bcc=[], subject=subject, - body_text=body_text, - body_html=body_html, + body_text=mirror.body_text, + body_html=mirror.body_html, domain=domain, in_reply_to=threading.get("in_reply_to"), references=threading.get("references"), + inline_parts=mirror.inline_parts, + file_attachments=mirror.file_attachments, ) handoff_message = await store_handoff_message( @@ -442,8 +423,8 @@ async def send_handoff_to_human( sender={"name": "Gentic Agent", "address": agent_email}, recipients=[{"name": human_email.split("@")[0], "address": human_email}], subject=subject, - body_text=body_text, - body_html=body_html, + body_text=mirror.body_text, + body_html=mirror.body_html, internet_message_id=internet_message_id, in_reply_to=threading.get("in_reply_to"), references=threading.get("references"), @@ -581,81 +562,28 @@ async def process_human_reply( logger.warning("handoff_reply_missing_external_parent handoff_id=%s", handoff["_id"]) return - threading = threading_fields_from_parent(external_parent) - external_subject = external_reply_subject( - handoff.get("external_subject") or external_parent.get("subject", "") - ) - external_recipient = external_recipient or external_parent.get("sender", {}).get("address") if not external_recipient: return customer_body = new_reply_content - internet_message_id = await deliver_email( - sender=agent_email, - recipients=[external_recipient], - cc=[], - bcc=[], - subject=external_subject, + reply_plan = build_external_reply_plan( + parent_external_message=external_parent, body_text=customer_body, - body_html=None, - domain=domain, - in_reply_to=threading.get("in_reply_to"), - references=threading.get("references"), + subject_override=external_reply_subject( + handoff.get("external_subject") or external_parent.get("subject", "") + ), + recipients=[external_recipient], ) - - external_doc = { - "direction": "outbound", - "message_scope": MESSAGE_SCOPE_EXTERNAL, - "account_id": account_id, - "sender": {"name": "Gentic Agent", "address": agent_email}, - "to": [{"name": external_recipient.split("@")[0], "address": external_recipient}], - "subject": external_subject, - "snippet": _snippet(customer_body), - "body_text": customer_body, - "body_html": None, - "attachments": [], - "created_at": now, - "sent_at": now, - "status": "accepted", - "message_id": normalize_message_id(internet_message_id), - "in_reply_to": threading.get("in_reply_to"), - "references": threading.get("references", []), - "parent_message_id": external_parent["_id"], - "thread_relation": "reply", - "thread_resolution_method": "human_handoff_bridge", - } - ext_result = await database.messages.insert_one(external_doc) - external_doc["_id"] = ext_result.inserted_id - await attach_message_to_thread( + external_doc = await send_external_reply( database, account_id=account_id, - thread_id=external_thread_id, - message_id=ext_result.inserted_id, - subject=external_subject, - event_at=now, - ) - await database.mailbox_entries.insert_one( - { - "account_id": account_id, - "message_id": ext_result.inserted_id, - "thread_id": external_thread_id, - "mailbox": "sent", - "message_scope": MESSAGE_SCOPE_EXTERNAL, - "direction": "outbound", - "sender": external_doc["sender"], - "recipients": external_doc["to"], - "subject": external_subject, - "snippet": external_doc["snippet"], - "attachments": [], - "status": "accepted", - "is_read": True, - "is_starred": False, - "event_at": now, - "created_at": now, - "trashed_at": None, - "deletion_state": None, - } + sender=agent_email, + sender_name="Gentic Agent", + external_thread_id=external_thread_id, + plan=reply_plan, + domain=domain, + resolution_method="human_handoff_bridge", ) relay_body = build_relay_confirmation_text(customer_body) diff --git a/backend/app/services/inbound_smtp.py b/backend/app/services/inbound_smtp.py index 92d38e5..de94641 100644 --- a/backend/app/services/inbound_smtp.py +++ b/backend/app/services/inbound_smtp.py @@ -17,6 +17,7 @@ from app.services.human_handoff_service import ( resolve_handoff_from_inbound, maybe_trigger_handoff, ) +from app.services.message_ids import normalize_message_id from app.services.mime_parser import parse_mime from app.services.raw_mime_storage import delivery_digest, get_raw_mime_storage from app.services.thread_resolver import participants_from_parsed, resolve_thread @@ -38,6 +39,38 @@ def peer_is_allowed(session) -> bool: return False +def log_smtp_message_received( + *, + message_id: str | None, + in_reply_to: str | None, + references: list[str], + sender: str, + recipient: str, + subject: str, + resolved_scope: str, + resolved_thread_id, + resolution_method: str, + stored_message_id, + mailbox_id: str = "inbox", +) -> None: + logger.info( + "smtp_message_received message_id=%s in_reply_to=%s references=%s " + "from=%s to=%s subject=%s resolved_scope=%s resolved_thread_id=%s " + "resolution_method=%s stored_message_id=%s mailbox_id=%s", + message_id, + in_reply_to, + " ".join(references or []), + sender, + recipient, + subject, + resolved_scope, + resolved_thread_id, + resolution_method, + stored_message_id, + mailbox_id, + ) + + class InboundHandler: async def handle_RCPT(self, server, session, envelope, address, rcpt_options): if not peer_is_allowed(session): @@ -55,6 +88,12 @@ class InboundHandler: if not hasattr(envelope, "account_ids"): envelope.account_ids = {} envelope.account_ids[normalized] = str(account.id) + logger.info( + "smtp_rcpt_accepted mail_from=%s rcpt_to=%s account_id=%s", + envelope.mail_from.lower(), + normalized, + envelope.account_ids[normalized], + ) return "250 2.1.5 Recipient OK" async def handle_DATA(self, server, session, envelope): @@ -74,12 +113,22 @@ class InboundHandler: {"_id": 1}, ): return f"250 2.0.0 Message previously accepted as {delivery_key}" + + logger.info( + "smtp_data_accepted mail_from=%s rcpt_to=%s delivery_key=%s bytes=%s", + envelope.mail_from.lower(), + ",".join(recipients), + delivery_key, + len(raw), + ) + parsed = parse_mime(raw) body_text = parsed.pop("body_text") body_html = parsed.pop("body_html") stored = await get_raw_mime_storage().put(raw, delivery_key) now = datetime.now(timezone.utc) sender_address = (parsed.get("sender") or {}).get("address", "").lower() + internet_message_id = normalize_message_id(parsed.get("message_id")) for recipient in recipients: account_id = envelope.account_ids[recipient] @@ -111,6 +160,8 @@ class InboundHandler: "delivery_key": delivery_key, "received_at": now, "created_at": now, + "message_id": internet_message_id, + "internet_message_id": internet_message_id, } await process_human_reply( mongo_db, @@ -121,10 +172,18 @@ class InboundHandler: inbound_body=body_text, domain=domain, ) - logger.info( - "handoff_human_reply handoff_id=%s account=%s", - handoff["_id"], - account_id, + log_smtp_message_received( + message_id=internet_message_id, + in_reply_to=parsed.get("in_reply_to"), + references=parsed.get("references", []), + sender=sender_address, + recipient=recipient, + subject=parsed.get("subject", ""), + resolved_scope=MESSAGE_SCOPE_HANDOFF, + resolved_thread_id=handoff.get("handoff_thread_id"), + resolution_method="handoff_in_reply_to", + stored_message_id=None, + mailbox_id="handoff", ) return f"250 2.0.0 Handoff reply accepted as {delivery_key}" @@ -135,6 +194,8 @@ class InboundHandler: "envelope_sender": envelope.mail_from.lower(), "envelope_recipients": recipients, "delivery_key": delivery_key, + "message_id": internet_message_id, + "internet_message_id": internet_message_id, "raw_mime": { "bucket": stored.bucket, "name": stored.name, @@ -156,7 +217,7 @@ class InboundHandler: if persisted is None: raise RuntimeError("Inbound message was not persisted.") - if body_text: + if body_text or body_html: await mongo_db.messages.update_one( {"_id": persisted["_id"]}, {"$set": {"body_text": body_text, "body_html": body_html}}, @@ -191,7 +252,9 @@ class InboundHandler: { "$set": { "message_scope": MESSAGE_SCOPE_EXTERNAL, - "in_reply_to": parsed.get("in_reply_to"), + "message_id": internet_message_id, + "internet_message_id": internet_message_id, + "in_reply_to": normalize_message_id(parsed.get("in_reply_to")), "references": parsed.get("references", []), "raw_headers": parsed.get("raw_headers", {}), "parent_message_id": resolution.parent_message_id, @@ -200,7 +263,7 @@ class InboundHandler: } }, ) - await mongo_db.mailbox_entries.update_one( + mailbox_result = await mongo_db.mailbox_entries.update_one( { "account_id": account_id, "message_id": persisted["_id"], @@ -244,10 +307,26 @@ class InboundHandler: external_thread_id=thread_id, domain=domain, ) + log_smtp_message_received( + message_id=internet_message_id, + in_reply_to=parsed.get("in_reply_to"), + references=parsed.get("references", []), + sender=sender_address, + recipient=recipient, + subject=parsed.get("subject", ""), + resolved_scope=MESSAGE_SCOPE_EXTERNAL, + resolved_thread_id=thread_id, + resolution_method=resolution.thread_resolution_method, + stored_message_id=persisted["_id"], + mailbox_id="inbox", + ) logger.info( - "message_received message_id=%s thread_id=%s scope=EXTERNAL", + "message_received message_id=%s internet_message_id=%s thread_id=%s " + "scope=EXTERNAL mailbox_entry_upserted=%s", persisted["_id"], + internet_message_id, thread_id, + mailbox_result.upserted_id is not None or mailbox_result.modified_count > 0, ) except Exception: logger.exception("Temporary failure accepting inbound message") diff --git a/backend/app/services/mail_sender.py b/backend/app/services/mail_sender.py index 3ace5e0..2afb986 100644 --- a/backend/app/services/mail_sender.py +++ b/backend/app/services/mail_sender.py @@ -1,16 +1,33 @@ import email.policy import email.utils -from email.message import EmailMessage +from email.message import EmailMessage, Message +from email.mime.base import MIMEBase +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText import aiosmtplib import dkim from app.config import get_settings from app.models import MailDomain +from app.services.mime_body_extractor import ExtractedMimePart SMTP_POLICY = email.policy.SMTP.clone(max_line_length=998) +def _mime_attachment(part: ExtractedMimePart) -> MIMEBase: + maintype, _, subtype = part.content_type.partition("/") + if not subtype: + maintype, subtype = "application", "octet-stream" + mime = MIMEBase(maintype, subtype, policy=SMTP_POLICY) + mime.set_payload(part.payload) + email.utils.encode_base64(mime) + mime.add_header("Content-Disposition", part.disposition, filename=part.filename) + if part.content_id: + mime.add_header("Content-ID", f"<{part.content_id}>") + return mime + + def build_message( *, sender: str, @@ -23,7 +40,47 @@ def build_message( in_reply_to: str | None = None, references: list[str] | None = None, message_id: str | None = None, -) -> EmailMessage: + inline_parts: list[ExtractedMimePart] | None = None, + file_attachments: list[ExtractedMimePart] | None = None, +) -> Message: + inline_parts = inline_parts or [] + file_attachments = file_attachments or [] + + if inline_parts or file_attachments: + root = MIMEMultipart("mixed", policy=SMTP_POLICY) + root["From"] = sender + root["To"] = ", ".join(recipients) + if cc: + root["Cc"] = ", ".join(cc) + if reply_to: + root["Reply-To"] = reply_to + root["Subject"] = subject + root["Date"] = email.utils.formatdate(localtime=True) + generated_id = message_id or email.utils.make_msgid(domain=sender.split("@")[-1]) + root["Message-ID"] = generated_id + if in_reply_to: + root["In-Reply-To"] = in_reply_to + if references: + root["References"] = " ".join(references) + + alternative = MIMEMultipart("alternative", policy=SMTP_POLICY) + alternative.attach(MIMEText(body_text, "plain", "utf-8", policy=SMTP_POLICY)) + + if body_html: + if inline_parts: + related = MIMEMultipart("related", policy=SMTP_POLICY) + related.attach(MIMEText(body_html, "html", "utf-8", policy=SMTP_POLICY)) + for part in inline_parts: + related.attach(_mime_attachment(part)) + alternative.attach(related) + else: + alternative.attach(MIMEText(body_html, "html", "utf-8", policy=SMTP_POLICY)) + + root.attach(alternative) + for attachment in file_attachments: + root.attach(_mime_attachment(attachment)) + return root + message = EmailMessage(policy=SMTP_POLICY) message["From"] = sender message["To"] = ", ".join(recipients) @@ -45,7 +102,7 @@ def build_message( return message -def sign_message(message: EmailMessage, domain: MailDomain) -> bytes: +def sign_message(message: Message, domain: MailDomain) -> bytes: raw_message = message.as_bytes(policy=SMTP_POLICY) signed_headers = [ b"from", @@ -106,6 +163,8 @@ async def deliver_email( in_reply_to: str | None = None, references: list[str] | None = None, message_id: str | None = None, + inline_parts: list[ExtractedMimePart] | None = None, + file_attachments: list[ExtractedMimePart] | None = None, ) -> str: message = build_message( sender=sender, @@ -118,6 +177,8 @@ async def deliver_email( in_reply_to=in_reply_to, references=references, message_id=message_id, + inline_parts=inline_parts, + file_attachments=file_attachments, ) signed = sign_message(message, domain) await send_via_smtp( diff --git a/backend/app/services/mime_body_extractor.py b/backend/app/services/mime_body_extractor.py new file mode 100644 index 0000000..cb89a7a --- /dev/null +++ b/backend/app/services/mime_body_extractor.py @@ -0,0 +1,106 @@ +from dataclasses import dataclass, field +from email import policy +from email.message import EmailMessage, Message +from email.parser import BytesParser + + +@dataclass +class ExtractedMimePart: + filename: str + content_type: str + payload: bytes + content_id: str | None = None + disposition: str = "attachment" + + +@dataclass +class MirroredMimeContent: + body_text: str = "" + body_html: str | None = None + inline_parts: list[ExtractedMimePart] = field(default_factory=list) + file_attachments: list[ExtractedMimePart] = field(default_factory=list) + + +def _part_payload(part: Message) -> bytes: + payload = part.get_payload(decode=True) + if isinstance(payload, bytes): + return payload + if isinstance(payload, str): + return payload.encode("utf-8", errors="replace") + return b"" + + +def _part_text(part: Message) -> str: + try: + value = part.get_content() + except (LookupError, UnicodeError): + value = _part_payload(part).decode("utf-8", errors="replace") + return value if isinstance(value, str) else "" + + +def _content_id(part: Message) -> str | None: + header = part.get("Content-ID") + if not header: + return None + return header.strip().strip("<>") + + +def _is_attachment_part(part: Message) -> bool: + disposition = (part.get_content_disposition() or "").lower() + if disposition == "attachment": + return True + if disposition == "inline": + return False + filename = part.get_filename() + content_type = part.get_content_type() + if filename and content_type not in {"text/plain", "text/html"}: + return True + return False + + +def _walk_parts(message: Message, collector: MirroredMimeContent) -> None: + if message.is_multipart(): + for part in message.get_payload(): + if isinstance(part, Message): + _walk_parts(part, collector) + return + + content_type = message.get_content_type() + cid = _content_id(message) + filename = message.get_filename() or (f"inline-{cid}" if cid else "attachment") + + if content_type == "text/plain" and not _is_attachment_part(message): + text = _part_text(message).strip() + if text and not collector.body_text: + collector.body_text = text + return + + if content_type == "text/html" and not _is_attachment_part(message): + html = _part_text(message).strip() + if html: + collector.body_html = html + return + + payload = _part_payload(message) + if not payload: + return + + extracted = ExtractedMimePart( + filename=filename, + content_type=content_type, + payload=payload, + content_id=cid, + disposition="inline" if cid or message.get_content_disposition() == "inline" else "attachment", + ) + + if cid or extracted.disposition == "inline": + collector.inline_parts.append(extracted) + elif _is_attachment_part(message): + collector.file_attachments.append(extracted) + + +def extract_mirror_content(raw: bytes) -> MirroredMimeContent: + message = BytesParser(policy=policy.default).parsebytes(raw) + collector = MirroredMimeContent() + _walk_parts(message, collector) + return collector diff --git a/backend/app/services/thread_resolver.py b/backend/app/services/thread_resolver.py index 57b0af0..78c783e 100644 --- a/backend/app/services/thread_resolver.py +++ b/backend/app/services/thread_resolver.py @@ -196,7 +196,7 @@ async def lookup_known_messages( {"$match": {"entry.0": {"$exists": True}}}, ] known: dict[str, KnownMessage] = {} - cursor = database.messages.aggregate(pipeline) + cursor = await database.messages.aggregate(pipeline) async for document in cursor: internet_id = normalize_message_id(document.get("message_id")) if not internet_id: diff --git a/backend/app/services/thread_service.py b/backend/app/services/thread_service.py index 4293c35..a81e01b 100644 --- a/backend/app/services/thread_service.py +++ b/backend/app/services/thread_service.py @@ -121,7 +121,9 @@ async def ensure_thread_for_message( def threading_fields_from_parent(parent: dict) -> dict: - parent_internet_id = normalize_message_id(parent.get("message_id")) + parent_internet_id = normalize_message_id( + parent.get("internet_message_id") or parent.get("message_id") + ) parent_references = parent.get("references") or [] if isinstance(parent_references, str): parent_references = [parent_references] diff --git a/backend/tests/test_email_threading_roundtrip.py b/backend/tests/test_email_threading_roundtrip.py new file mode 100644 index 0000000..040efa8 --- /dev/null +++ b/backend/tests/test_email_threading_roundtrip.py @@ -0,0 +1,76 @@ +import pytest +from bson import ObjectId + +from app.services.thread_resolver import KnownMessage, resolve_thread_from_known + + +def _known( + internet_id: str, + thread_id: ObjectId, + *, + references: list[str] | None = None, +) -> KnownMessage: + from datetime import datetime, timezone + + from app.services.message_ids import normalize_message_id + from app.services.subject_normalize import normalize_subject + + now = datetime.now(timezone.utc) + return KnownMessage( + id=ObjectId(), + thread_id=thread_id, + internet_message_id=normalize_message_id(internet_id) or "", + account_id="acct-1", + sent_at=now, + received_at=now, + created_at=now, + participants=frozenset({"customer@gmail.com"}), + normalized_subject=normalize_subject("lets try one more time"), + ) + + +def test_gmail_reply_to_gentic_resolves_same_thread(): + thread_id = ObjectId() + a1 = _known("", thread_id) + g1 = _known("", thread_id, references=[""]) + known = { + a1.internet_message_id: a1, + g1.internet_message_id: g1, + } + resolution = resolve_thread_from_known( + account_id="acct-1", + in_reply_to="", + references=["", ""], + subject="Re: lets try one more time", + participants={"customer@gmail.com"}, + known_by_internet_id=known, + ) + assert resolution.thread_id == thread_id + assert resolution.thread_resolution_method == "in_reply_to" + assert resolution.parent_message_id == g1.id + + +@pytest.mark.asyncio +async def test_lookup_known_messages_awaits_aggregate(): + from app.services.thread_resolver import lookup_known_messages + + class FakeCursor: + def __aiter__(self): + return self + + async def __anext__(self): + raise StopAsyncIteration + + class FakeMessages: + def __init__(self): + self.aggregate_awaited = False + + async def aggregate(self, pipeline): + self.aggregate_awaited = True + return FakeCursor() + + messages = FakeMessages() + database = type("DB", (), {"messages": messages})() + result = await lookup_known_messages(database, "acct-1", [""]) + assert result == {} + assert messages.aggregate_awaited is True diff --git a/backend/tests/test_external_reply_service.py b/backend/tests/test_external_reply_service.py new file mode 100644 index 0000000..091660f --- /dev/null +++ b/backend/tests/test_external_reply_service.py @@ -0,0 +1,57 @@ +from bson import ObjectId + +from app.services.external_reply_service import build_external_reply_plan +from app.services.thread_service import threading_fields_from_parent + + +def test_build_external_reply_references_gmail_parent(): + parent = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "references": [], + "subject": "lets try one more time", + } + plan = build_external_reply_plan( + parent_external_message=parent, + body_text="I hear you.", + recipients=["customer@gmail.com"], + ) + assert plan.threading.in_reply_to == "" + assert plan.threading.references == [""] + assert plan.subject == "Re: lets try one more time" + assert plan.threading.replying_to_internet_message_id == "" + + +def test_build_external_reply_chains_references_for_multi_round(): + parent = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "references": ["", ""], + "subject": "Re: lets try one more time", + } + threading = threading_fields_from_parent(parent) + assert threading["in_reply_to"] == "" + assert threading["references"] == [ + "", + "", + "", + ] + + +def test_external_reply_never_uses_internal_object_id_as_in_reply_to(): + parent = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "references": [], + "subject": "hello", + } + plan = build_external_reply_plan( + parent_external_message=parent, + body_text="reply", + recipients=["customer@gmail.com"], + ) + assert str(parent["_id"]) not in (plan.threading.in_reply_to or "") + assert all(str(parent["_id"]) not in ref for ref in plan.threading.references) diff --git a/backend/tests/test_handoff_mirror.py b/backend/tests/test_handoff_mirror.py new file mode 100644 index 0000000..6746b95 --- /dev/null +++ b/backend/tests/test_handoff_mirror.py @@ -0,0 +1,85 @@ +from app.services.handoff_email_templates import ( + build_mirror_handoff_html, + build_mirror_handoff_text, + external_reply_subject, + format_sender_line, + handoff_email_subject, +) +from app.services.html_sanitizer import sanitize_email_html +from app.services.mime_body_extractor import extract_mirror_content +from app.services.thread_service import threading_fields_from_parent + + +def test_mirror_template_is_minimal(): + html = build_mirror_handoff_html( + sender_line="091rabbytes@gmail.com", + external_subject="lets try one more time", + original_html="

one last attempt. i am tired

", + agent_email="suyash@gentic.email", + ) + assert "LATEST MESSAGE" not in html + assert "Why you" not in html + assert "gentic-handoff-original" in html + assert "one last attempt. i am tired" in html + assert "✦ Human Handoff" in html + assert "091rabbytes@gmail.com" in html + assert "lets try one more time" in html + + +def test_mirror_plain_text_template(): + text = build_mirror_handoff_text( + sender_line="091rabbytes@gmail.com", + external_subject="lets try one more time", + original_body_text="one last attempt. i am tired", + agent_email="suyash@gentic.email", + ) + assert "one last attempt. i am tired" in text + assert "LATEST MESSAGE" not in text + assert "Reply normally" in text + + +def test_format_sender_with_name(): + assert ( + format_sender_line({"name": "Rahul Sharma", "address": "rahul@example.com"}) + == "Rahul Sharma " + ) + + +def test_external_threading_headers_from_customer_message(): + parent = { + "_id": "mongo-id", + "message_id": "", + "references": [], + } + threading = threading_fields_from_parent(parent) + assert threading["in_reply_to"] == "" + assert threading["references"] == [""] + + +def test_external_reply_subject_uses_customer_subject(): + assert external_reply_subject("lets try one more time") == "Re: lets try one more time" + assert handoff_email_subject("lets try one more time") == "Human Handoff · lets try one more time" + + +def test_sanitize_removes_script(): + dirty = '

Hello

' + clean = sanitize_email_html(dirty) + assert "Invoice
Amount
\r\n" + b"--b--\r\n" + ) + extracted = extract_mirror_content(raw) + assert "Invoice" in (extracted.body_html or "") + assert "" in (extracted.body_html or "") diff --git a/backend/tests/test_reply_extractor.py b/backend/tests/test_reply_extractor.py index fba95eb..87d450b 100644 --- a/backend/tests/test_reply_extractor.py +++ b/backend/tests/test_reply_extractor.py @@ -1,4 +1,4 @@ -from app.services.handoff_email_templates import external_reply_subject, handoff_email_subject +from app.services.handoff_email_templates import external_reply_subject from app.services.reply_extractor import extract_new_reply_content, validate_external_reply_body SCREENSHOT_REPLY = """ahan, did you receive a response? @@ -49,10 +49,6 @@ def test_external_reply_subject_uses_customer_subject(): assert external_reply_subject("Human Handoff · are you ready?") == "Re: are you ready?" -def test_handoff_subject_format(): - assert handoff_email_subject("are you ready?") == "Human Handoff · are you ready?" - - def test_outlook_original_message_split(): body = """Thanks for the update.