From 8d5e9712a6ebde9fc0248aa4503c52f3bcb52bbd Mon Sep 17 00:00:00 2001 From: debu-2 Date: Tue, 6 Oct 2026 03:14:53 +0530 Subject: [PATCH] handoff fixes --- backend/app/main.py | 14 + backend/app/routers/agent_handoff.py | 11 +- backend/app/routers/emails.py | 7 +- backend/app/schemas.py | 3 +- .../app/services/external_reply_service.py | 276 +++++++++++++++- backend/app/services/handoff_mirror_policy.py | 62 ++++ backend/app/services/human_handoff_service.py | 298 ++++++++++++++---- backend/app/services/inbound_smtp.py | 42 ++- backend/app/services/mail_sender.py | 30 +- backend/app/services/message_ids.py | 70 +++- backend/app/services/mime_parser.py | 25 +- backend/app/services/thread_resolver.py | 61 ++-- backend/app/services/thread_service.py | 10 +- backend/tests/test_external_reply_service.py | 79 ++++- backend/tests/test_handoff_mirror_policy.py | 57 ++++ backend/tests/test_handoff_threading.py | 82 +++++ backend/tests/test_human_handoff_service.py | 1 + backend/tests/test_mail_sender.py | 47 +++ .../src/components/HandoffConversation.jsx | 18 +- frontend/src/styles.css | 4 + 20 files changed, 1056 insertions(+), 141 deletions(-) create mode 100644 backend/app/services/handoff_mirror_policy.py create mode 100644 backend/tests/test_handoff_mirror_policy.py create mode 100644 backend/tests/test_handoff_threading.py diff --git a/backend/app/main.py b/backend/app/main.py index 5ab47ff..4ce717e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -194,6 +194,20 @@ async def lifespan(_: FastAPI): [("message_id", 1)], sparse=True, ) + await mongo_db.messages.create_index( + [("message_id_folded", 1)], + sparse=True, + ) + await mongo_db.messages.create_index( + [("generated_from_handoff_message_id", 1)], + sparse=True, + unique=True, + ) + await mongo_db.messages.create_index( + [("mirrored_from_external_message_id", 1)], + sparse=True, + unique=True, + ) await mongo_db.mailbox_entries.create_index( [("account_id", 1), ("thread_id", 1), ("event_at", -1), ("_id", -1)] ) diff --git a/backend/app/routers/agent_handoff.py b/backend/app/routers/agent_handoff.py index e54b5de..5bcba20 100644 --- a/backend/app/routers/agent_handoff.py +++ b/backend/app/routers/agent_handoff.py @@ -18,6 +18,7 @@ from app.schemas import ( HandoffTestRequest, HandoffTestResult, ) +from app.services.handoff_mirror_policy import is_relay_acknowledgment_message from app.services.human_handoff_service import MESSAGE_SCOPE_HANDOFF from app.services.handoff_evaluator import EMAIL_TYPE_META, HANDOFF_LABELS, evaluate_handoff @@ -261,8 +262,15 @@ async def get_handoff_conversation( messages: list[HandoffMessageRead] = [] for row in rows: message = row["message"] + if is_relay_acknowledgment_message(message): + continue sender_address = (message.get("sender") or {}).get("address", "").lower() - role = "human" if sender_address == human_email else "agent" + if message.get("mirrored_from_external_message_id"): + role = "customer" + elif sender_address == human_email: + role = "human" + else: + role = "agent" event_at = row.get("event_at") or message.get("received_at") or message.get("sent_at") or message["created_at"] raw_body = message.get("raw_body") new_reply_content = message.get("new_reply_content") @@ -281,6 +289,7 @@ async def get_handoff_conversation( and new_reply_content and raw_body.strip() != new_reply_content.strip() ), + relay_status=message.get("relay_status") if role == "human" else None, event_at=event_at, ) ) diff --git a/backend/app/routers/emails.py b/backend/app/routers/emails.py index 528736f..7cafb75 100644 --- a/backend/app/routers/emails.py +++ b/backend/app/routers/emails.py @@ -24,7 +24,7 @@ from app.schemas import ( ) from app.services.domain_provisioning import get_domain_by_name from app.services.mail_sender import deliver_email -from app.services.message_ids import normalize_message_id +from app.services.message_ids import canonical_message_id, message_id_fields from app.services.mime_parser import parse_mime from app.services.raw_mime_storage import get_raw_mime_storage from app.services.reply_recipients import reply_recipients @@ -249,9 +249,8 @@ async def create_email( references=threading_fields.get("references"), ) document["status"] = "accepted" - normalized_id = normalize_message_id(internet_message_id) - document["message_id"] = normalized_id - document["internet_message_id"] = normalized_id + normalized_id = canonical_message_id(internet_message_id) + document.update(message_id_fields(normalized_id)) log_smtp_message_sending( generated_message_id=normalized_id, in_reply_to=threading_fields.get("in_reply_to"), diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 3b6b684..3049075 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -310,13 +310,14 @@ class HandoffConversationSummary(BaseModel): class HandoffMessageRead(BaseModel): id: str - role: Literal["agent", "human"] + role: Literal["agent", "human", "customer"] subject: str snippet: str body_text: str new_reply_content: str | None = None raw_body: str | None = None has_quoted_text: bool = False + relay_status: str | None = None event_at: datetime diff --git a/backend/app/services/external_reply_service.py b/backend/app/services/external_reply_service.py index 51361d8..1a82ea9 100644 --- a/backend/app/services/external_reply_service.py +++ b/backend/app/services/external_reply_service.py @@ -7,17 +7,27 @@ 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, +from app.services.mail_sender import deliver_email, log_outbound_mime_headers +from app.services.message_ids import ( + build_references_header, + canonical_message_id, + folded_message_id, + first_message_id, + internet_message_id_of_message, + message_id_fields, ) +from app.services.mime_parser import parse_mime +from app.services.raw_mime_storage import get_raw_mime_storage +from app.services.thread_service import threading_fields_from_parent logger = logging.getLogger(__name__) MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" +MESSAGE_SCOPE_HANDOFF = "HANDOFF" + + +class ExternalReplyThreadingError(RuntimeError): + pass @dataclass(frozen=True) @@ -38,9 +48,169 @@ class ExternalReplyPlan: def internet_message_id_of(message: dict) -> str | None: - return normalize_message_id( + stored = canonical_message_id( message.get("internet_message_id") or message.get("message_id") ) + raw_header = (message.get("raw_headers") or {}).get("message_id") + raw_canonical = canonical_message_id(raw_header) + if raw_canonical and stored: + if folded_message_id(raw_canonical) == folded_message_id(stored): + return raw_canonical + return stored or raw_canonical + + +async def resolve_external_reply_parent( + database, + *, + account_id: str, + external_thread_id: ObjectId, + customer_address: str, + handoff: dict | None = None, +) -> dict | None: + """Pick the latest external inbound customer message with a real Message-ID.""" + customer = customer_address.lower() + pipeline = [ + { + "$match": { + "account_id": account_id, + "thread_id": external_thread_id, + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "deletion_state": {"$in": [None, ""]}, + } + }, + {"$sort": {"event_at": -1, "_id": -1}}, + { + "$lookup": { + "from": "messages", + "localField": "message_id", + "foreignField": "_id", + "as": "message", + } + }, + {"$unwind": "$message"}, + { + "$match": { + "message.message_scope": {"$in": [MESSAGE_SCOPE_EXTERNAL, None]}, + "message.direction": "inbound", + "$or": [ + {"message.internet_message_id": {"$gt": ""}}, + {"message.message_id": {"$gt": ""}}, + ], + } + }, + {"$limit": 1}, + ] + cursor = await database.mailbox_entries.aggregate(pipeline) + rows = [row async for row in cursor] + if rows: + return rows[0]["message"] + + if handoff and handoff.get("trigger_external_message_id"): + trigger = await database.messages.find_one( + { + "_id": handoff["trigger_external_message_id"], + "message_scope": MESSAGE_SCOPE_EXTERNAL, + } + ) + if trigger and internet_message_id_of(trigger): + return trigger + + trigger_internet = handoff.get("trigger_internet_message_id") if handoff else None + if trigger_internet: + folded = folded_message_id(trigger_internet) + trigger = await database.messages.find_one( + { + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "message_id_folded": folded, + } + ) + if trigger: + return trigger + + return None + + +def validate_external_reply_parent(parent_external_message: dict) -> None: + parent_id = internet_message_id_of(parent_external_message) + if not parent_id: + raise ExternalReplyThreadingError( + "External reply parent is missing internet_message_id" + ) + scope = parent_external_message.get("message_scope") + if scope == MESSAGE_SCOPE_HANDOFF: + raise ExternalReplyThreadingError( + "External reply parent must not be a HANDOFF-scoped message" + ) + + +async def assert_in_reply_to_resolves_to_external_parent( + database, + in_reply_to: str | None, +) -> dict: + """Ensure In-Reply-To points at an EXTERNAL message, never HANDOFF.""" + canonical = canonical_message_id(in_reply_to) + folded = folded_message_id(canonical) + if not canonical or not folded: + raise ExternalReplyThreadingError("External reply is missing In-Reply-To") + + parent = await database.messages.find_one( + { + "$or": [ + {"message_id_folded": folded}, + {"message_id": canonical}, + {"internet_message_id": canonical}, + ], + } + ) + if parent is None: + raise ExternalReplyThreadingError( + f"In-Reply-To {canonical} does not resolve to a known message" + ) + scope = parent.get("message_scope") + if scope == MESSAGE_SCOPE_HANDOFF: + raise ExternalReplyThreadingError( + f"In-Reply-To {canonical} resolves to HANDOFF-scoped message" + ) + if scope not in (MESSAGE_SCOPE_EXTERNAL, None): + raise ExternalReplyThreadingError( + f"In-Reply-To {canonical} resolves to non-external message scope={scope}" + ) + return parent + + +async def assert_external_threading_chain( + database, + *, + in_reply_to: str | None, + references: list[str] | None, +) -> None: + """Ensure outbound headers only reference EXTERNAL messages.""" + await assert_in_reply_to_resolves_to_external_parent(database, in_reply_to) + + folded_targets = { + folded_message_id(value) + for value in [*(references or [])] + if folded_message_id(value) + } + in_reply_folded = folded_message_id(in_reply_to) + if in_reply_folded: + folded_targets.discard(in_reply_folded) + + for folded in folded_targets: + handoff_hit = await database.messages.find_one( + { + "$or": [ + {"message_id_folded": folded}, + {"message_id": folded}, + ], + "message_scope": MESSAGE_SCOPE_HANDOFF, + }, + {"_id": 1}, + ) + if handoff_hit: + raise ExternalReplyThreadingError( + "External reply threading references a HANDOFF message" + ) def build_external_reply_plan( @@ -50,9 +220,31 @@ def build_external_reply_plan( body_html: str | None = None, subject_override: str | None = None, recipients: list[str], + parent_internet_id_override: str | None = None, ) -> ExternalReplyPlan: + parent_internet_id = canonical_message_id( + parent_internet_id_override + ) or internet_message_id_of(parent_external_message) + if not parent_internet_id: + raise ExternalReplyThreadingError( + "External reply parent is missing internet_message_id" + ) + scope = parent_external_message.get("message_scope") + if scope == MESSAGE_SCOPE_HANDOFF: + raise ExternalReplyThreadingError( + "External reply parent must not be a HANDOFF-scoped message" + ) threading_fields = threading_fields_from_parent(parent_external_message) - parent_internet_id = internet_message_id_of(parent_external_message) + in_reply_to = parent_internet_id or threading_fields.get("in_reply_to") + references = threading_fields.get("references", []) + if parent_internet_id: + references = build_references_header( + [ref for ref in references if folded_message_id(ref) != folded_message_id(parent_internet_id)], + parent_internet_id, + ).split() + elif not references and in_reply_to: + references = [in_reply_to] + if subject_override: subject = subject_override else: @@ -67,8 +259,8 @@ def build_external_reply_plan( body_text=body_text, body_html=body_html, threading=ExternalReplyThreading( - in_reply_to=threading_fields.get("in_reply_to"), - references=threading_fields.get("references", []), + in_reply_to=in_reply_to, + references=references, parent_internal_message_id=parent_external_message["_id"], replying_to_internet_message_id=parent_internet_id, ), @@ -76,6 +268,33 @@ def build_external_reply_plan( ) +def log_original_external_message( + *, + internal_id: ObjectId | str, + thread_id: ObjectId | str | None, + internet_message_id: str | None, + sender: str | None, + recipient: str | None, + subject: str | None, + in_reply_to: str | None, + references: list[str] | None, + raw_message_available: bool, +) -> None: + logger.info( + "original_external_message internal_id=%s thread_id=%s internet_message_id=%s " + "from=%s to=%s subject=%s in_reply_to=%s references=%s raw_message_available=%s", + internal_id, + thread_id, + internet_message_id, + sender, + recipient, + subject, + in_reply_to, + references, + raw_message_available, + ) + + def log_smtp_message_sending( *, generated_message_id: str | None, @@ -114,7 +333,30 @@ async def send_external_reply( plan: ExternalReplyPlan, domain: MailDomain, resolution_method: str = "in_reply_to", + generated_by: str | None = None, + generated_from_handoff_message_id: ObjectId | None = None, + message_origin: str | None = None, ) -> dict: + if generated_from_handoff_message_id: + existing = await database.messages.find_one( + { + "generated_from_handoff_message_id": generated_from_handoff_message_id, + "message_scope": MESSAGE_SCOPE_EXTERNAL, + } + ) + if existing: + logger.info( + "external_reply_idempotent_skip handoff_message=%s external=%s", + generated_from_handoff_message_id, + existing["_id"], + ) + return existing + + await assert_external_threading_chain( + database, + in_reply_to=plan.threading.in_reply_to, + references=plan.threading.references, + ) log_smtp_message_sending( generated_message_id=None, in_reply_to=plan.threading.in_reply_to, @@ -139,7 +381,8 @@ async def send_external_reply( in_reply_to=plan.threading.in_reply_to, references=plan.threading.references, ) - normalized_id = normalize_message_id(internet_message_id) + normalized_id = canonical_message_id(internet_message_id) + id_fields = message_id_fields(normalized_id) log_smtp_message_sending( generated_message_id=normalized_id, in_reply_to=plan.threading.in_reply_to, @@ -167,16 +410,23 @@ async def send_external_reply( "created_at": now, "sent_at": now, "status": "accepted", - "message_id": normalized_id, - "internet_message_id": normalized_id, + **id_fields, "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, } + if generated_by: + document["generated_by"] = generated_by + if generated_from_handoff_message_id: + document["generated_from_handoff_message_id"] = generated_from_handoff_message_id + if message_origin: + document["message_origin"] = message_origin result = await database.messages.insert_one(document) document["_id"] = result.inserted_id + from app.services.thread_service import attach_message_to_thread + await attach_message_to_thread( database, account_id=account_id, diff --git a/backend/app/services/handoff_mirror_policy.py b/backend/app/services/handoff_mirror_policy.py new file mode 100644 index 0000000..ac403a9 --- /dev/null +++ b/backend/app/services/handoff_mirror_policy.py @@ -0,0 +1,62 @@ +"""Decide whether external messages should be mirrored into a Human Handoff thread.""" + +MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" + +# Message provenance — who authored / generated the content +MESSAGE_ORIGIN_EXTERNAL_CUSTOMER = "EXTERNAL_CUSTOMER" +MESSAGE_ORIGIN_HUMAN_OWNER = "HUMAN_OWNER" +MESSAGE_ORIGIN_AGENT = "AGENT" +MESSAGE_ORIGIN_SYSTEM = "SYSTEM" +MESSAGE_ORIGIN_HANDOFF_MIRROR = "HANDOFF_MIRROR" + +GENERATED_BY_CUSTOMER = "CUSTOMER" +GENERATED_BY_HUMAN = "HUMAN" +GENERATED_BY_AGENT = "AGENT" +GENERATED_BY_HANDOFF_BRIDGE = "HANDOFF_BRIDGE" +GENERATED_BY_SYSTEM = "SYSTEM" +GENERATED_BY_RELAY_ACK = "RELAY_ACK" + +RELAY_STATUS_SENT = "SENT" +RELAY_STATUS_FAILED = "FAILED" +RELAY_STATUS_PENDING = "PENDING" + + +def should_mirror_to_human(message: dict) -> bool: + """Return True only for new external customer content worth mirroring.""" + scope = message.get("message_scope", MESSAGE_SCOPE_EXTERNAL) + if scope != MESSAGE_SCOPE_EXTERNAL: + return False + + if message.get("direction") != "inbound": + return False + + generated_by = message.get("generated_by") + if generated_by in ( + GENERATED_BY_HANDOFF_BRIDGE, + GENERATED_BY_RELAY_ACK, + GENERATED_BY_SYSTEM, + ): + return False + + origin = message.get("message_origin") + if origin in ( + MESSAGE_ORIGIN_HUMAN_OWNER, + MESSAGE_ORIGIN_SYSTEM, + MESSAGE_ORIGIN_HANDOFF_MIRROR, + ): + return False + + if message.get("generated_from_handoff_message_id"): + return False + + return True + + +def is_relay_acknowledgment_message(message: dict) -> bool: + """Detect legacy or internal delivery-receipt messages in handoff threads.""" + if message.get("generated_by") == GENERATED_BY_RELAY_ACK: + return True + if message.get("message_origin") == MESSAGE_ORIGIN_SYSTEM: + return True + body = (message.get("body_text") or message.get("snippet") or "").strip() + return message.get("direction") == "outbound" and body.startswith("Sent to customer") diff --git a/backend/app/services/human_handoff_service.py b/backend/app/services/human_handoff_service.py index 9f0f3e2..1595f59 100644 --- a/backend/app/services/human_handoff_service.py +++ b/backend/app/services/human_handoff_service.py @@ -5,18 +5,37 @@ from datetime import datetime, timezone from bson import ObjectId from app.services.external_reply_service import ( + ExternalReplyThreadingError, build_external_reply_plan, + internet_message_id_of, + log_original_external_message, + resolve_external_reply_parent, + resolve_internet_message_id, send_external_reply, ) from app.services.handoff_email_templates import ( - build_relay_confirmation_text, external_reply_subject, handoff_email_subject, ) +from app.services.handoff_mirror_policy import ( + GENERATED_BY_HANDOFF_BRIDGE, + GENERATED_BY_HUMAN, + GENERATED_BY_SYSTEM, + MESSAGE_ORIGIN_HANDOFF_MIRROR, + MESSAGE_ORIGIN_HUMAN_OWNER, + RELAY_STATUS_FAILED, + RELAY_STATUS_SENT, + should_mirror_to_human, +) 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 +from app.services.message_ids import ( + canonical_message_id, + folded_message_id, + message_id_fields, + parse_message_id_list, +) from app.services.reply_extractor import extract_new_reply_content, validate_external_reply_body from app.services.thread_service import ( attach_message_to_thread, @@ -226,9 +245,13 @@ async def resolve_handoff_from_inbound( return None for internet_id in reversed(parse_message_id_list(" ".join(search_ids))): + folded = folded_message_id(internet_id) message = await database.messages.find_one( { - "message_id": internet_id, + "$or": [ + {"message_id_folded": folded}, + {"message_id": internet_id}, + ], "message_scope": MESSAGE_SCOPE_HANDOFF, } ) @@ -256,6 +279,26 @@ async def resolve_handoff_from_inbound( return None +async def _handoff_mirror_exists(database, external_message_id: ObjectId) -> bool: + existing = await database.messages.find_one( + { + "mirrored_from_external_message_id": external_message_id, + "message_scope": MESSAGE_SCOPE_HANDOFF, + }, + {"_id": 1}, + ) + return existing is not None + + +async def _bridged_external_exists(database, handoff_message_id: ObjectId) -> dict | None: + return await database.messages.find_one( + { + "generated_from_handoff_message_id": handoff_message_id, + "message_scope": MESSAGE_SCOPE_EXTERNAL, + } + ) + + async def store_handoff_message( database, *, @@ -277,6 +320,11 @@ async def store_handoff_message( raw_body: str | None = None, new_reply_content: str | None = None, body_html: 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, + generated_external_message_id: ObjectId | None = None, ) -> dict: now = event_at or _now() display_body = new_reply_content if new_reply_content is not None else body_text @@ -297,11 +345,20 @@ async def store_handoff_message( "sent_at": now if direction == "outbound" else None, "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), + **message_id_fields(internet_message_id), + "in_reply_to": canonical_message_id(in_reply_to), "references": references or [], } + if message_origin: + document["message_origin"] = message_origin + if generated_by: + document["generated_by"] = generated_by + if mirrored_from_external_message_id: + document["mirrored_from_external_message_id"] = mirrored_from_external_message_id + if relay_status: + document["relay_status"] = relay_status + if generated_external_message_id: + document["generated_external_message_id"] = generated_external_message_id result = await database.messages.insert_one(document) document["_id"] = result.inserted_id await attach_message_to_thread( @@ -357,6 +414,23 @@ async def send_handoff_to_human( domain, existing_handoff: dict | None = None, ) -> dict: + if await _handoff_mirror_exists(database, external_message["_id"]): + logger.info( + "handoff_mirror_skipped_duplicate external_message=%s", + external_message["_id"], + ) + if existing_handoff: + return existing_handoff + existing = await database.human_handoffs.find_one( + { + "account_id": account_id, + "external_thread_id": external_thread_id, + } + ) + if existing: + return existing + raise RuntimeError("Handoff mirror duplicate without handoff record") + external_subject = external_message.get("subject", "") stable_subject = handoff_email_subject(external_subject) mirror_source = {**external_message, "_agent_email": agent_email} @@ -394,6 +468,8 @@ async def send_handoff_to_human( "updated_at": _now(), "last_external_message_at": message_event_at(external_message), "last_human_message_at": None, + "trigger_external_message_id": external_message["_id"], + "trigger_internet_message_id": internet_message_id_of(external_message), } insert = await database.human_handoffs.insert_one(handoff_doc) handoff_id = insert.inserted_id @@ -431,6 +507,9 @@ async def send_handoff_to_human( handoff_id=handoff_id, link_relationship=relationship, external_message_id=external_message["_id"], + message_origin=MESSAGE_ORIGIN_HANDOFF_MIRROR, + generated_by=GENERATED_BY_SYSTEM, + mirrored_from_external_message_id=external_message["_id"], ) await database.human_handoffs.update_one( @@ -504,7 +583,8 @@ async def process_human_reply( in_reply_to=inbound_parsed.get("in_reply_to"), references=inbound_parsed.get("references", []), handoff_id=handoff["_id"], - link_relationship=LINK_HUMAN_TRIGGERED_EXTERNAL, + message_origin=MESSAGE_ORIGIN_HUMAN_OWNER, + generated_by=GENERATED_BY_HUMAN, ) now = _now() @@ -550,22 +630,68 @@ async def process_human_reply( return external_recipient = handoff.get("external_sender") or "" - external_parent = await _latest_external_customer_message( + external_parent = await resolve_external_reply_parent( database, - account_id, - external_thread_id, - external_recipient, + account_id=account_id, + external_thread_id=external_thread_id, + customer_address=external_recipient, + handoff=handoff, ) if external_parent is None: - external_parent = await _latest_external_message(database, account_id, external_thread_id) - if external_parent is None: - logger.warning("handoff_reply_missing_external_parent handoff_id=%s", handoff["_id"]) + logger.warning( + "handoff_reply_missing_external_parent handoff_id=%s external_thread=%s " + "trigger_internet_message_id=%s", + handoff["_id"], + external_thread_id, + handoff.get("trigger_internet_message_id"), + ) return - external_recipient = external_recipient or external_parent.get("sender", {}).get("address") + parent_sender = (external_parent.get("sender") or {}).get("address") + log_original_external_message( + internal_id=external_parent["_id"], + thread_id=external_thread_id, + internet_message_id=internet_message_id_of(external_parent), + sender=parent_sender, + recipient=agent_email, + subject=external_parent.get("subject"), + in_reply_to=external_parent.get("in_reply_to"), + references=external_parent.get("references"), + raw_message_available=bool(external_parent.get("raw_mime")), + ) + + external_recipient = external_recipient or parent_sender if not external_recipient: return + parent_internet_id = await resolve_internet_message_id(database, external_parent) + if not parent_internet_id: + logger.error( + "handoff_reply_missing_parent_message_id handoff_id=%s parent_internal_id=%s", + handoff["_id"], + external_parent["_id"], + ) + return + + logger.info( + "handoff_external_reply_threading handoff_id=%s parent_internal_id=%s " + "parent_internet_message_id=%s human_in_reply_to=%s human_references=%s", + handoff["_id"], + external_parent["_id"], + parent_internet_id, + human_message.get("in_reply_to"), + human_message.get("references"), + ) + + existing_bridge = await _bridged_external_exists(database, human_message["_id"]) + if existing_bridge: + logger.info( + "handoff_bridge_already_processed handoff_message=%s external=%s", + human_message["_id"], + existing_bridge["_id"], + ) + return + customer_body = new_reply_content reply_plan = build_external_reply_plan( parent_external_message=external_parent, @@ -574,62 +700,87 @@ async def process_human_reply( handoff.get("external_subject") or external_parent.get("subject", "") ), recipients=[external_recipient], + parent_internet_id_override=parent_internet_id, ) - external_doc = await send_external_reply( - database, - account_id=account_id, - sender=agent_email, - sender_name="Gentic Agent", - external_thread_id=external_thread_id, - plan=reply_plan, - domain=domain, - resolution_method="human_handoff_bridge", + logger.info( + "handoff_external_reply_headers handoff_id=%s b_in_reply_to=%s b_references=%s", + handoff["_id"], + reply_plan.threading.in_reply_to, + reply_plan.threading.references, ) + try: + external_doc = await send_external_reply( + database, + account_id=account_id, + sender=agent_email, + sender_name="Gentic Agent", + external_thread_id=external_thread_id, + plan=reply_plan, + domain=domain, + resolution_method="human_handoff_bridge", + generated_by=GENERATED_BY_HANDOFF_BRIDGE, + generated_from_handoff_message_id=human_message["_id"], + ) + except ExternalReplyThreadingError: + await database.messages.update_one( + {"_id": human_message["_id"]}, + {"$set": {"relay_status": RELAY_STATUS_FAILED, "relay_failed_at": now}}, + ) + logger.error( + "handoff_bridge_threading_rejected handoff_id=%s handoff_message=%s " + "parent_internet_message_id=%s", + handoff["_id"], + human_message["_id"], + parent_internet_id, + ) + return + except Exception: + await database.messages.update_one( + {"_id": human_message["_id"]}, + {"$set": {"relay_status": RELAY_STATUS_FAILED, "relay_failed_at": now}}, + ) + logger.exception( + "handoff_bridge_send_failed handoff_id=%s handoff_message=%s", + handoff["_id"], + human_message["_id"], + ) + raise - relay_body = build_relay_confirmation_text(customer_body) - relay_subject = _handoff_reply_subject(handoff.get("external_subject", "")) - relay_threading = threading_fields_from_parent(human_message) - relay_internet_id = await deliver_email( - sender=agent_email, - recipients=[human_email], - cc=[], - bcc=[], - subject=relay_subject, - body_text=relay_body, - body_html=None, - domain=domain, - in_reply_to=relay_threading.get("in_reply_to"), - references=relay_threading.get("references"), + await database.messages.update_one( + {"_id": human_message["_id"]}, + { + "$set": { + "relay_status": RELAY_STATUS_SENT, + "generated_external_message_id": external_doc["_id"], + "relay_delivered_at": now, + } + }, ) - await store_handoff_message( + await _record_message_link( database, - account_id=account_id, - agent_email=agent_email, - handoff_thread_id=handoff_thread_id, - direction="outbound", - sender={"name": "Gentic Agent", "address": agent_email}, - recipients=[{"name": human_email.split("@")[0], "address": human_email}], - subject=relay_subject, - body_text=relay_body, - internet_message_id=relay_internet_id, - in_reply_to=relay_threading.get("in_reply_to"), - references=relay_threading.get("references"), handoff_id=handoff["_id"], - link_relationship=LINK_EXTERNAL_RELAYED, - external_message_id=ext_result.inserted_id, + external_message_id=external_doc["_id"], + handoff_message_id=human_message["_id"], + relationship=LINK_HUMAN_TRIGGERED_EXTERNAL, ) await database.human_handoffs.update_one( {"_id": handoff["_id"]}, { "$set": { - "status": HANDOFF_STATUS_AGENT_REPLIED, - "current_owner": OWNER_AGENT, + "status": HANDOFF_STATUS_WAITING, + "current_owner": OWNER_HUMAN, "updated_at": now, "last_human_message_at": now, } }, ) + logger.info( + "handoff_bridge_complete handoff_id=%s handoff_message=%s external=%s", + handoff["_id"], + human_message["_id"], + external_doc["_id"], + ) async def maybe_trigger_handoff( @@ -641,6 +792,40 @@ async def maybe_trigger_handoff( external_thread_id: ObjectId, domain, ) -> None: + if not should_mirror_to_human(external_message): + logger.debug( + "handoff_mirror_skipped message_id=%s generated_by=%s origin=%s direction=%s", + external_message.get("_id"), + external_message.get("generated_by"), + external_message.get("message_origin"), + external_message.get("direction"), + ) + return + + existing = await find_active_handoff_for_external_thread( + database, account_id, external_thread_id + ) + if existing: + if await _handoff_mirror_exists(database, external_message["_id"]): + return + rule = await _handoff_rule_for_human( + database, account_id, existing.get("human_email", "") + ) + if rule and not rule.get("allow_replies", True): + return + await send_handoff_to_human( + database, + account_id=account_id, + agent_email=agent_email, + human_email=existing["human_email"], + external_message=external_message, + external_thread_id=external_thread_id, + handoff_reason=existing.get("handoff_reason", ""), + domain=domain, + existing_handoff=existing, + ) + return + rules = await database.human_handoff_rules.find( {"account_id": account_id, "active": True} ).to_list(length=20) @@ -667,9 +852,6 @@ async def maybe_trigger_handoff( if not rule.get("allow_replies", True): continue - existing = await find_active_handoff_for_external_thread( - database, account_id, external_thread_id - ) reason = (rule.get("instructions") or "").strip() or "This conversation requires human attention." await send_handoff_to_human( database, @@ -680,6 +862,6 @@ async def maybe_trigger_handoff( external_thread_id=external_thread_id, handoff_reason=reason, domain=domain, - existing_handoff=existing, + existing_handoff=None, ) return diff --git a/backend/app/services/inbound_smtp.py b/backend/app/services/inbound_smtp.py index de94641..1e8e3ce 100644 --- a/backend/app/services/inbound_smtp.py +++ b/backend/app/services/inbound_smtp.py @@ -10,6 +10,10 @@ from app.config import get_settings from app.database import SessionFactory, mongo_db from app.models import Account from app.services.domain_provisioning import get_domain_by_name +from app.services.handoff_mirror_policy import ( + GENERATED_BY_CUSTOMER, + MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, +) from app.services.human_handoff_service import ( MESSAGE_SCOPE_EXTERNAL, MESSAGE_SCOPE_HANDOFF, @@ -17,7 +21,8 @@ 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.external_reply_service import log_original_external_message +from app.services.message_ids import canonical_message_id, message_id_fields 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 @@ -128,7 +133,8 @@ class InboundHandler: 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")) + internet_message_id = canonical_message_id(parsed.get("message_id")) + id_fields = message_id_fields(internet_message_id) for recipient in recipients: account_id = envelope.account_ids[recipient] @@ -160,8 +166,7 @@ class InboundHandler: "delivery_key": delivery_key, "received_at": now, "created_at": now, - "message_id": internet_message_id, - "internet_message_id": internet_message_id, + **id_fields, } await process_human_reply( mongo_db, @@ -194,8 +199,7 @@ 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, + **id_fields, "raw_mime": { "bucket": stored.bucket, "name": stored.name, @@ -252,14 +256,15 @@ class InboundHandler: { "$set": { "message_scope": MESSAGE_SCOPE_EXTERNAL, - "message_id": internet_message_id, - "internet_message_id": internet_message_id, - "in_reply_to": normalize_message_id(parsed.get("in_reply_to")), + **id_fields, + "in_reply_to": canonical_message_id(parsed.get("in_reply_to")), "references": parsed.get("references", []), "raw_headers": parsed.get("raw_headers", {}), "parent_message_id": resolution.parent_message_id, "thread_relation": resolution.thread_relation, "thread_resolution_method": resolution.thread_resolution_method, + "message_origin": MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, + "generated_by": GENERATED_BY_CUSTOMER, } }, ) @@ -298,6 +303,14 @@ class InboundHandler: database_session, agent_email.split("@", 1)[1], ) + persisted.update( + { + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "inbound", + "message_origin": MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, + "generated_by": GENERATED_BY_CUSTOMER, + } + ) if domain is not None: await maybe_trigger_handoff( mongo_db, @@ -307,6 +320,17 @@ class InboundHandler: external_thread_id=thread_id, domain=domain, ) + log_original_external_message( + internal_id=persisted["_id"], + thread_id=thread_id, + internet_message_id=internet_message_id, + sender=sender_address, + recipient=recipient, + subject=parsed.get("subject"), + in_reply_to=canonical_message_id(parsed.get("in_reply_to")), + references=parsed.get("references", []), + raw_message_available=bool(persisted.get("raw_mime")), + ) log_smtp_message_received( message_id=internet_message_id, in_reply_to=parsed.get("in_reply_to"), diff --git a/backend/app/services/mail_sender.py b/backend/app/services/mail_sender.py index 2afb986..73f57e6 100644 --- a/backend/app/services/mail_sender.py +++ b/backend/app/services/mail_sender.py @@ -1,5 +1,7 @@ import email.policy import email.utils +import logging +from email.parser import BytesParser from email.message import EmailMessage, Message from email.mime.base import MIMEBase from email.mime.multipart import MIMEMultipart @@ -13,6 +15,20 @@ from app.models import MailDomain from app.services.mime_body_extractor import ExtractedMimePart SMTP_POLICY = email.policy.SMTP.clone(max_line_length=998) +logger = logging.getLogger(__name__) + + +def log_outbound_mime_headers(message: Message) -> None: + logger.info( + "smtp_outbound_mime_headers message_id=%s in_reply_to=%s references=%s " + "from=%s to=%s subject=%s", + message.get("Message-ID"), + message.get("In-Reply-To"), + message.get("References"), + message.get("From"), + message.get("To"), + message.get("Subject"), + ) def _mime_attachment(part: ExtractedMimePart) -> MIMEBase: @@ -60,7 +76,9 @@ def build_message( root["Message-ID"] = generated_id if in_reply_to: root["In-Reply-To"] = in_reply_to - if references: + refs = references or [in_reply_to] + root["References"] = " ".join(refs) + elif references: root["References"] = " ".join(references) alternative = MIMEMultipart("alternative", policy=SMTP_POLICY) @@ -94,7 +112,9 @@ def build_message( message["Message-ID"] = generated_id if in_reply_to: message["In-Reply-To"] = in_reply_to - if references: + refs = references or [in_reply_to] + message["References"] = " ".join(refs) + elif references: message["References"] = " ".join(references) message.set_content(body_text) if body_html: @@ -115,6 +135,10 @@ def sign_message(message: Message, domain: MailDomain) -> bytes: ] if message.get("Cc"): signed_headers.insert(3, b"cc") + if message.get("In-Reply-To"): + signed_headers.append(b"in-reply-to") + if message.get("References"): + signed_headers.append(b"references") signature = dkim.sign( raw_message, domain.dkim_selector.encode(), @@ -180,7 +204,9 @@ async def deliver_email( inline_parts=inline_parts, file_attachments=file_attachments, ) + log_outbound_mime_headers(message) signed = sign_message(message, domain) + log_signed_mime_headers(signed) await send_via_smtp( signed, sender=sender, diff --git a/backend/app/services/message_ids.py b/backend/app/services/message_ids.py index 132191d..fde0bfa 100644 --- a/backend/app/services/message_ids.py +++ b/backend/app/services/message_ids.py @@ -3,7 +3,8 @@ import re _MESSAGE_ID_PATTERN = re.compile(r"<[^>]+>") -def normalize_message_id(value: str | None) -> str | None: +def canonical_message_id(value: str | None) -> str | None: + """Normalize to angle-bracket form while preserving original case.""" if value is None: return None cleaned = value.strip() @@ -11,22 +12,59 @@ def normalize_message_id(value: str | None) -> str | None: return None if not cleaned.startswith("<"): cleaned = f"<{cleaned.strip('<>')}>" - return cleaned.lower() + return cleaned + + +def folded_message_id(value: str | None) -> str | None: + """Case-insensitive lookup key for Message-ID matching.""" + canonical = canonical_message_id(value) + return canonical.casefold() if canonical else None + + +def normalize_message_id(value: str | None) -> str | None: + """Preserve case — Gmail threading requires exact Message-ID tokens.""" + return canonical_message_id(value) + + +def internet_message_id_of_message(message: dict) -> str | None: + """Resolve Message-ID, preferring raw MIME header case when stored value was folded.""" + stored = canonical_message_id( + message.get("internet_message_id") or message.get("message_id") + ) + raw_header = (message.get("raw_headers") or {}).get("message_id") + raw_canonical = canonical_message_id(raw_header) + if raw_canonical and stored: + if folded_message_id(raw_canonical) == folded_message_id(stored): + return raw_canonical + return stored or raw_canonical + + +def message_id_fields(value: str | None) -> dict[str, str]: + canonical = canonical_message_id(value) + if not canonical: + return {} + folded = folded_message_id(canonical) + return { + "message_id": canonical, + "internet_message_id": canonical, + "message_id_folded": folded or canonical.casefold(), + } def parse_message_id_list(header_value: str | None) -> list[str]: if not header_value: return [] - folded = re.sub(r"\s+", " ", header_value.strip()) + folded_header = re.sub(r"\s+", " ", header_value.strip()) seen: set[str] = set() ordered: list[str] = [] - for match in _MESSAGE_ID_PATTERN.findall(folded): - normalized = normalize_message_id(match) - if normalized and normalized not in seen: - seen.add(normalized) - ordered.append(normalized) + for match in _MESSAGE_ID_PATTERN.findall(folded_header): + canonical = canonical_message_id(match) + key = folded_message_id(canonical) + if canonical and key and key not in seen: + seen.add(key) + ordered.append(canonical) if not ordered: - single = normalize_message_id(folded) + single = canonical_message_id(folded_header) if single: ordered.append(single) return ordered @@ -41,11 +79,13 @@ def build_references_header(references: list[str], parent_message_id: str) -> st chain: list[str] = [] seen: set[str] = set() for value in references: - normalized = normalize_message_id(value) - if normalized and normalized not in seen: - seen.add(normalized) - chain.append(normalized) - parent = normalize_message_id(parent_message_id) - if parent and parent not in seen: + canonical = canonical_message_id(value) + key = folded_message_id(canonical) + if canonical and key and key not in seen: + seen.add(key) + chain.append(canonical) + parent = canonical_message_id(parent_message_id) + parent_key = folded_message_id(parent) + if parent and parent_key and parent_key not in seen: chain.append(parent) return " ".join(chain) diff --git a/backend/app/services/mime_parser.py b/backend/app/services/mime_parser.py index 332d950..e95a876 100644 --- a/backend/app/services/mime_parser.py +++ b/backend/app/services/mime_parser.py @@ -28,6 +28,13 @@ def _text(part) -> str: return value if isinstance(value, str) else "" +def _header_str(value) -> str | None: + if value is None: + return None + cleaned = str(value).strip() + return cleaned or None + + def _snippet(text: str, html: str | None, limit: int = 240) -> str: source = text if not source and html: @@ -81,15 +88,15 @@ def parse_mime(raw: bytes) -> dict: in_reply_to = first_message_id(message.get("In-Reply-To")) references = parse_message_id_list(message.get("References")) raw_headers = { - "message_id": message.get("Message-ID"), - "in_reply_to": message.get("In-Reply-To"), - "references": message.get("References"), - "subject": message.get("Subject"), - "date": message.get("Date"), - "from": message.get("From"), - "to": message.get("To"), - "cc": message.get("Cc"), - "reply_to": message.get("Reply-To"), + "message_id": _header_str(message.get("Message-ID")), + "in_reply_to": _header_str(message.get("In-Reply-To")), + "references": _header_str(message.get("References")), + "subject": _header_str(message.get("Subject")), + "date": _header_str(message.get("Date")), + "from": _header_str(message.get("From")), + "to": _header_str(message.get("To")), + "cc": _header_str(message.get("Cc")), + "reply_to": _header_str(message.get("Reply-To")), } return { "message_id": internet_message_id, diff --git a/backend/app/services/thread_resolver.py b/backend/app/services/thread_resolver.py index 78c783e..0e8ef44 100644 --- a/backend/app/services/thread_resolver.py +++ b/backend/app/services/thread_resolver.py @@ -5,8 +5,8 @@ from datetime import datetime, timedelta, timezone from bson import ObjectId from app.services.message_ids import ( - first_message_id, - normalize_message_id, + canonical_message_id, + folded_message_id, parse_message_id_list, ) from app.services.subject_normalize import normalize_subject @@ -52,14 +52,17 @@ def resolve_thread_from_known( known_threads: dict[ObjectId, list[KnownMessage]] | None = None, now: datetime | None = None, ) -> ThreadResolution: - scoped = { - key: value - for key, value in known_by_internet_id.items() - if value.account_id == account_id - } - in_reply = normalize_message_id(in_reply_to) - if in_reply and in_reply in scoped: - parent = scoped[in_reply] + scoped: dict[str, KnownMessage] = {} + for key, value in known_by_internet_id.items(): + if value.account_id != account_id: + continue + fold_key = folded_message_id(key) or folded_message_id(value.internet_message_id) + if fold_key: + scoped[fold_key] = value + in_reply = canonical_message_id(in_reply_to) + in_reply_folded = folded_message_id(in_reply) + if in_reply_folded and in_reply_folded in scoped: + parent = scoped[in_reply_folded] return ThreadResolution( thread_id=parent.thread_id, parent_message_id=parent.id, @@ -68,9 +71,9 @@ def resolve_thread_from_known( ) reference_ids = [ - normalize_message_id(value) + folded_message_id(value) for value in references - if normalize_message_id(value) + if folded_message_id(value) ] for reference in reversed(reference_ids): if reference in scoped: @@ -155,22 +158,31 @@ async def lookup_known_messages( account_id: str, internet_ids: list[str], ) -> dict[str, KnownMessage]: - normalized_ids = [ + folded_ids = [ value - for value in {normalize_message_id(item) for item in internet_ids} + for value in {folded_message_id(item) for item in internet_ids} if value ] - if not normalized_ids: + if not folded_ids: return {} pipeline = [ { "$match": { - "message_id": {"$in": normalized_ids}, - "$or": [ - {"message_scope": "EXTERNAL"}, - {"message_scope": {"$exists": False}}, - ], + "$and": [ + { + "$or": [ + {"message_id_folded": {"$in": folded_ids}}, + {"message_id": {"$in": folded_ids}}, + ] + }, + { + "$or": [ + {"message_scope": "EXTERNAL"}, + {"message_scope": {"$exists": False}}, + ] + }, + ] } }, { @@ -198,15 +210,18 @@ async def lookup_known_messages( known: dict[str, KnownMessage] = {} cursor = await database.messages.aggregate(pipeline) async for document in cursor: - internet_id = normalize_message_id(document.get("message_id")) - if not internet_id: + internet_id = canonical_message_id( + document.get("internet_message_id") or document.get("message_id") + ) + internet_key = folded_message_id(internet_id) + if not internet_id or not internet_key: continue entry = document["entry"][0] thread_id = entry.get("thread_id") or document.get("thread_id") if thread_id is None: continue participants = participants_from_parsed(document) - known[internet_id] = KnownMessage( + known[internet_key] = KnownMessage( id=document["_id"], thread_id=thread_id, internet_message_id=internet_id, diff --git a/backend/app/services/thread_service.py b/backend/app/services/thread_service.py index a81e01b..320c3b1 100644 --- a/backend/app/services/thread_service.py +++ b/backend/app/services/thread_service.py @@ -2,7 +2,11 @@ from datetime import datetime, timezone from bson import ObjectId -from app.services.message_ids import build_references_header, normalize_message_id +from app.services.message_ids import ( + build_references_header, + canonical_message_id, + internet_message_id_of_message, +) from app.services.subject_normalize import normalize_subject from app.services.thread_resolver import ThreadResolution @@ -121,9 +125,7 @@ async def ensure_thread_for_message( def threading_fields_from_parent(parent: dict) -> dict: - parent_internet_id = normalize_message_id( - parent.get("internet_message_id") or parent.get("message_id") - ) + parent_internet_id = internet_message_id_of_message(parent) parent_references = parent.get("references") or [] if isinstance(parent_references, str): parent_references = [parent_references] diff --git a/backend/tests/test_external_reply_service.py b/backend/tests/test_external_reply_service.py index 091660f..315998c 100644 --- a/backend/tests/test_external_reply_service.py +++ b/backend/tests/test_external_reply_service.py @@ -1,6 +1,10 @@ from bson import ObjectId -from app.services.external_reply_service import build_external_reply_plan +from app.services.external_reply_service import ( + build_external_reply_plan, + internet_message_id_of, +) +from app.services.message_ids import canonical_message_id, folded_message_id from app.services.thread_service import threading_fields_from_parent @@ -55,3 +59,76 @@ def test_external_reply_never_uses_internal_object_id_as_in_reply_to(): ) 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) + + +def test_handoff_bridge_reply_uses_external_parent_not_human_reply(): + """Human handoff must thread B against customer A, never private H2.""" + customer_a = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "message_scope": "EXTERNAL", + "direction": "inbound", + "references": [], + "subject": "fresh test", + } + human_h2 = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "message_scope": "HANDOFF", + "direction": "inbound", + "in_reply_to": "", + "references": [""], + "subject": "Re: fresh test", + } + plan = build_external_reply_plan( + parent_external_message=customer_a, + body_text="human said B", + recipients=["091rabbytes@gmail.com"], + ) + assert plan.threading.in_reply_to == "" + assert plan.threading.references == [""] + assert human_h2["internet_message_id"] not in (plan.threading.in_reply_to or "") + assert all( + human_h2["internet_message_id"] not in ref for ref in plan.threading.references + ) + assert "" not in plan.threading.references + + +def test_raw_header_restores_gmail_message_id_case(): + parent = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "raw_headers": {"message_id": ""}, + "references": [], + "subject": "fresh test", + } + plan = build_external_reply_plan( + parent_external_message=parent, + body_text="human said B", + recipients=["customer@gmail.com"], + ) + assert plan.threading.in_reply_to == "" + assert plan.threading.references == [""] + + +def test_gmail_message_id_case_is_preserved_for_threading(): + raw = "" + assert canonical_message_id(raw) == raw + assert folded_message_id(raw) == "" + parent = { + "_id": ObjectId(), + "internet_message_id": raw, + "message_id": raw, + "references": [], + "subject": "fresh test", + } + plan = build_external_reply_plan( + parent_external_message=parent, + body_text="human said B", + recipients=["customer@gmail.com"], + ) + assert plan.threading.in_reply_to == raw + assert internet_message_id_of(parent) == raw diff --git a/backend/tests/test_handoff_mirror_policy.py b/backend/tests/test_handoff_mirror_policy.py new file mode 100644 index 0000000..91eecc1 --- /dev/null +++ b/backend/tests/test_handoff_mirror_policy.py @@ -0,0 +1,57 @@ +from bson import ObjectId + +from app.services.handoff_mirror_policy import ( + GENERATED_BY_CUSTOMER, + GENERATED_BY_HANDOFF_BRIDGE, + GENERATED_BY_RELAY_ACK, + MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, + MESSAGE_ORIGIN_HUMAN_OWNER, + is_relay_acknowledgment_message, + should_mirror_to_human, +) + + +def test_should_mirror_customer_inbound(): + message = { + "message_scope": "EXTERNAL", + "direction": "inbound", + "generated_by": GENERATED_BY_CUSTOMER, + "message_origin": MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, + } + assert should_mirror_to_human(message) is True + + +def test_should_not_mirror_handoff_bridge_outbound(): + message = { + "message_scope": "EXTERNAL", + "direction": "outbound", + "generated_by": GENERATED_BY_HANDOFF_BRIDGE, + "generated_from_handoff_message_id": ObjectId(), + } + assert should_mirror_to_human(message) is False + + +def test_should_not_mirror_human_origin(): + message = { + "message_scope": "EXTERNAL", + "direction": "inbound", + "message_origin": MESSAGE_ORIGIN_HUMAN_OWNER, + } + assert should_mirror_to_human(message) is False + + +def test_is_relay_acknowledgment_detects_legacy_body(): + message = { + "direction": "outbound", + "body_text": "Sent to customer:\n\nhuman said B", + } + assert is_relay_acknowledgment_message(message) is True + + +def test_is_relay_acknowledgment_detects_generated_by(): + message = { + "direction": "outbound", + "generated_by": GENERATED_BY_RELAY_ACK, + "body_text": "anything", + } + assert is_relay_acknowledgment_message(message) is True diff --git a/backend/tests/test_handoff_threading.py b/backend/tests/test_handoff_threading.py new file mode 100644 index 0000000..7c6d2a3 --- /dev/null +++ b/backend/tests/test_handoff_threading.py @@ -0,0 +1,82 @@ +"""Integration-style tests for Human Handoff → Customer RFC threading.""" + +from bson import ObjectId + +from app.services.external_reply_service import ( + build_external_reply_plan, + internet_message_id_of, +) +from app.services.mail_sender import build_message, sign_message +from types import SimpleNamespace + + +def test_handoff_bridge_first_reply_threads_against_customer_a(): + customer_a_id = "" + customer_a = { + "_id": ObjectId(), + "internet_message_id": customer_a_id, + "message_id": customer_a_id, + "message_scope": "EXTERNAL", + "direction": "inbound", + "references": [], + "subject": "fresh test", + } + plan = build_external_reply_plan( + parent_external_message=customer_a, + body_text="human said B", + recipients=["091rabbytes@gmail.com"], + ) + assert plan.threading.in_reply_to == customer_a_id + assert plan.threading.references == [customer_a_id] + assert plan.threading.in_reply_to != "" + assert "" not in plan.threading.references + + +def test_lowercased_stored_id_recovered_via_raw_headers(): + customer_a_id = "" + parent = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "raw_headers": {"message_id": customer_a_id}, + "references": [], + "subject": "fresh test", + } + plan = build_external_reply_plan( + parent_external_message=parent, + body_text="human said B", + recipients=["customer@gmail.com"], + parent_internet_id_override=internet_message_id_of(parent), + ) + assert plan.threading.in_reply_to == customer_a_id + assert plan.threading.references == [customer_a_id] + + +def test_signed_mime_preserves_threading_headers(): + customer_a_id = "" + message = build_message( + sender="suyash@gentic.email", + recipients=["091rabbytes@gmail.com"], + cc=[], + subject="Re: fresh test", + body_text="human said B", + body_html=None, + in_reply_to=customer_a_id, + references=[customer_a_id], + message_id="", + ) + domain = SimpleNamespace( + domain_name="gentic.email", + dkim_selector="agmail", + dkim_private_key="private-key", + ) + from unittest.mock import patch + + with patch( + "app.services.mail_sender.dkim.sign", + return_value=b"DKIM-Signature: test\r\n", + ): + signed = sign_message(message, domain) + + assert f"In-Reply-To: {customer_a_id}".encode() in signed + assert f"References: {customer_a_id}".encode() in signed diff --git a/backend/tests/test_human_handoff_service.py b/backend/tests/test_human_handoff_service.py index bb8bd91..99872c3 100644 --- a/backend/tests/test_human_handoff_service.py +++ b/backend/tests/test_human_handoff_service.py @@ -73,6 +73,7 @@ async def test_resolve_handoff_requires_owner_and_thread_context(): { "_id": message_id, "message_id": "", + "message_id_folded": "", "message_scope": MESSAGE_SCOPE_HANDOFF, } ] diff --git a/backend/tests/test_mail_sender.py b/backend/tests/test_mail_sender.py index 87b14ca..0c5081d 100644 --- a/backend/tests/test_mail_sender.py +++ b/backend/tests/test_mail_sender.py @@ -41,3 +41,50 @@ async def test_send_via_smtp_uses_configured_client_hostname(): ) assert send_mock.await_args.kwargs["local_hostname"] == "backend.gentic.email" + + +def test_build_message_sets_in_reply_to_and_references(): + message = build_message( + sender="suyash@gentic.email", + recipients=["customer@gmail.com"], + cc=[], + subject="Re: fresh test", + body_text="human said B", + body_html=None, + in_reply_to="", + references=[""], + message_id="", + ) + assert message["Message-ID"] == "" + assert message["In-Reply-To"] == "" + assert message["References"] == "" + raw = message.as_bytes() + assert b"In-Reply-To: " in raw + assert b"References: " in raw + + +def test_signed_mime_preserves_in_reply_to(): + from unittest.mock import patch + + message = build_message( + sender="suyash@gentic.email", + recipients=["customer@gmail.com"], + cc=[], + subject="Re: fresh test", + body_text="human said B", + body_html=None, + in_reply_to="", + references=[""], + message_id="", + ) + domain = SimpleNamespace( + domain_name="gentic.email", + dkim_selector="agmail", + dkim_private_key="private-key", + ) + with patch( + "app.services.mail_sender.dkim.sign", + return_value=b"DKIM-Signature: test\r\n", + ): + signed = sign_message(message, domain) + assert b"In-Reply-To: " in signed diff --git a/frontend/src/components/HandoffConversation.jsx b/frontend/src/components/HandoffConversation.jsx index 019b121..329a586 100644 --- a/frontend/src/components/HandoffConversation.jsx +++ b/frontend/src/components/HandoffConversation.jsx @@ -83,10 +83,26 @@ export default function HandoffConversation({ handoffId, onBack }) { {conversation.messages.map((message) => (
- {message.role === "human" ? "HUMAN OWNER" : "AGENT"} + + {message.role === "human" + ? "HUMAN OWNER" + : message.role === "customer" + ? "CUSTOMER" + : "AGENT"} +
+ {message.role === "human" && message.relay_status === "SENT" && ( +

+ ✓ Sent to customer +

+ )} + {message.role === "human" && message.relay_status === "FAILED" && ( +

+ ⚠ Could not send to customer +

+ )}
))} diff --git a/frontend/src/styles.css b/frontend/src/styles.css index bbcab18..01ab071 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -559,6 +559,10 @@ h1 em { color:var(--coral); font-style:normal; font-weight:500; } .handoff-message-human header strong { color:#11966a; } .handoff-message header time { color:#9a96a5; font-size:11px; } .handoff-message-body { color:#292536; font-size:13px; line-height:1.6; white-space:pre-wrap; } +.handoff-message-customer header strong { color:#2563eb; } +.handoff-relay-status { margin:10px 0 0; font-size:12px; } +.handoff-relay-status-sent { color:#11966a; } +.handoff-relay-status-failed { color:#c2410c; } .handoff-show-quoted { margin-top:8px; padding:0; border:0; background:transparent; color:#6d5bd0; font-size:11px; font-weight:600; cursor:pointer; } .handoff-quoted-body { margin:8px 0 0; padding:10px 12px; border:1px solid #ece9f2; border-radius:8px; background:#faf9fc; color:#77758a; font-size:11px; line-height:1.45; white-space:pre-wrap; overflow:auto; } .handoff-settings-intro { margin-bottom:18px; padding:18px 20px; border:1px solid #ece7ff; border-radius:14px; background:#fbf9ff; }