diff --git a/backend/app/main.py b/backend/app/main.py index 4ce717e..1ab8841 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -23,6 +23,7 @@ from app.routers import ( ) from app.services.thread_service import create_thread, message_event_at from app.services.domain_provisioning import ensure_platform_domain +from app.services.handoff_delivery import handoff_delivery_worker from app.services.inbound_smtp import InboundSMTPServer settings = get_settings() @@ -208,6 +209,11 @@ async def lifespan(_: FastAPI): sparse=True, unique=True, ) + await mongo_db.messages.create_index( + [("handoff_delivery_key", 1)], + sparse=True, + unique=True, + ) await mongo_db.mailbox_entries.create_index( [("account_id", 1), ("thread_id", 1), ("event_at", -1), ("_id", -1)] ) @@ -217,18 +223,54 @@ async def lifespan(_: FastAPI): await mongo_db.human_handoff_rules.create_index( [("account_id", 1), ("updated_at", -1)] ) + account_ids = await mongo_db.human_handoff_rules.distinct("account_id") + for account_id in account_ids: + existing_rules = ( + await mongo_db.human_handoff_rules.find({"account_id": account_id}) + .sort([("priority", 1), ("created_at", 1), ("_id", 1)]) + .to_list(length=1000) + ) + for index, rule in enumerate(existing_rules): + if rule.get("priority") != index: + await mongo_db.human_handoff_rules.update_one( + {"_id": rule["_id"]}, + {"$set": {"priority": index}}, + ) + await mongo_db.human_handoff_rules.create_index( + [("account_id", 1), ("priority", 1)], + unique=True, + ) + await mongo_db.human_handoff_rules.create_index( + [("account_id", 1), ("name_key", 1)], + unique=True, + sparse=True, + ) await mongo_db.human_handoffs.create_index( [("account_id", 1), ("updated_at", -1)] ) await mongo_db.human_handoffs.create_index( - [("account_id", 1), ("external_thread_id", 1)] + [("account_id", 1), ("external_thread_id", 1)], + unique=True, ) await mongo_db.human_handoffs.create_index( - [("account_id", 1), ("handoff_thread_id", 1)] + [("account_id", 1), ("handoff_thread_id", 1)], + unique=True, ) await mongo_db.handoff_message_links.create_index( [("handoff_id", 1), ("created_at", -1)] ) + await mongo_db.handoff_message_links.create_index( + [("handoff_id", 1), ("source_message_id", 1), ("relationship", 1)], + unique=True, + sparse=True, + ) + await mongo_db.handoff_delivery_jobs.create_index( + [("idempotency_key", 1)], + unique=True, + ) + await mongo_db.handoff_delivery_jobs.create_index( + [("status", 1), ("next_attempt_at", 1), ("created_at", 1)] + ) await mongo_db.mailbox_entries.create_index( [("account_id", 1), ("message_scope", 1), ("mailbox", 1), ("event_at", -1)] ) @@ -239,9 +281,11 @@ async def lifespan(_: FastAPI): await backfill_missing_thread_ids(mongo_db) if settings.inbound_smtp_enabled: await inbound_server.start() + await handoff_delivery_worker.start() try: yield finally: + await handoff_delivery_worker.stop() await inbound_server.stop() await engine.dispose() await mongo_client.close() diff --git a/backend/app/routers/agent_handoff.py b/backend/app/routers/agent_handoff.py index ac669b5..b071b2d 100644 --- a/backend/app/routers/agent_handoff.py +++ b/backend/app/routers/agent_handoff.py @@ -1,23 +1,36 @@ import re from datetime import datetime, timezone from bson import ObjectId -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import EmailStr, Field +from pymongo.errors import DuplicateKeyError from app.database import get_mongo_database from app.models import Account from app.routers.auth import get_current_account from app.schemas import ( + HandoffRuleOrderUpdate, HandoffRuleCreate, HandoffRuleRead, HandoffRuleUpdate, HandoffTestRequest, HandoffTestResult, + HandoffConversationRead, + HandoffConversationSummary, + HandoffLinkedExternal, + HandoffMessageRead, ) from app.services.handoff_evaluator import EMAIL_TYPE_META, HANDOFF_LABELS, evaluate_handoff +from app.services.handoff_delivery import ( + STATUS_AMBIGUOUS, + STATUS_FAILED, + STATUS_PENDING, +) router = APIRouter(prefix="/agent/handoff", tags=["agent-handoff"]) +HANDOFF_MAILBOXES = {"handoff_inbox", "handoff_outbox"} + def _now() -> datetime: return datetime.now(timezone.utc) @@ -31,8 +44,10 @@ def _owner_name_from_email(email: str) -> str: def _serialize(document: dict) -> HandoffRuleRead: return HandoffRuleRead( id=str(document["_id"]), + name=document.get("name") or document.get("owner_name") or "Human handoff", owner_email=document["owner_email"], owner_name=document.get("owner_name") or _owner_name_from_email(document["owner_email"]), + priority=document.get("priority", 0), labels=document.get("labels", []), email_types=document.get("email_types", []), instructions=document.get("instructions", ""), @@ -51,10 +66,231 @@ async def list_handoff_rules( ): cursor = database.human_handoff_rules.find( {"account_id": str(account.id)} - ).sort("updated_at", -1) + ).sort([("priority", 1), ("created_at", 1), ("_id", 1)]) return [_serialize(document) async for document in cursor] +def _handoff_status_label(value: str) -> str: + return { + "ACTIVE": "Active", + "WAITING_FOR_HUMAN": "Waiting for human", + "HUMAN_REPLIED": "Human replied", + "CLOSED": "Closed", + }.get(value, value.replace("_", " ").title()) + + +@router.get("/conversations", response_model=list[HandoffConversationSummary]) +async def list_handoff_conversations( + mailbox: str = Query(default="handoff_inbox"), + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + if mailbox not in HANDOFF_MAILBOXES: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid handoff mailbox.", + ) + account_id = str(account.id) + cursor = database.mailbox_entries.find( + { + "account_id": account_id, + "message_scope": "HANDOFF", + "mailbox": mailbox, + "deletion_state": {"$in": [None, ""]}, + } + ).sort([("event_at", -1), ("_id", -1)]) + seen: set[ObjectId] = set() + items: list[HandoffConversationSummary] = [] + async for entry in cursor: + thread_id = entry.get("thread_id") + if not thread_id or thread_id in seen: + continue + seen.add(thread_id) + handoff = await database.human_handoffs.find_one( + { + "account_id": account_id, + "handoff_thread_id": thread_id, + } + ) + message = await database.messages.find_one({"_id": entry["message_id"]}) + thread = await database.email_threads.find_one({"_id": thread_id}) + if not handoff or not message: + continue + items.append( + HandoffConversationSummary( + id=str(handoff["_id"]), + thread_id=str(thread_id), + mailbox=mailbox, + rule_name=handoff.get("rule_name", "Human handoff"), + external_subject=handoff.get("external_subject", ""), + external_sender=handoff.get("external_sender", ""), + human_email=handoff.get("human_email", ""), + status=handoff.get("status", "ACTIVE"), + status_label=_handoff_status_label( + handoff.get("status", "ACTIVE") + ), + current_owner=handoff.get("current_owner", "HUMAN"), + snippet=message.get("new_reply_content") + or message.get("snippet", ""), + message_count=(thread or {}).get("message_count", 1), + updated_at=entry.get("event_at") + or handoff.get("updated_at") + or _now(), + ) + ) + return items + + +@router.get("/conversations/counts") +async def handoff_conversation_counts( + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + account_id = str(account.id) + result = {} + for mailbox in sorted(HANDOFF_MAILBOXES): + thread_ids = await database.mailbox_entries.distinct( + "thread_id", + { + "account_id": account_id, + "message_scope": "HANDOFF", + "mailbox": mailbox, + "deletion_state": {"$in": [None, ""]}, + }, + ) + result[mailbox] = len([value for value in thread_ids if value]) + return result + + +@router.post("/deliveries/{job_id}/retry") +async def retry_handoff_delivery( + job_id: str, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + if not ObjectId.is_valid(job_id): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid delivery job ID.", + ) + now = _now() + result = await database.handoff_delivery_jobs.find_one_and_update( + { + "_id": ObjectId(job_id), + "account_id": str(account.id), + "status": {"$in": [STATUS_FAILED, STATUS_AMBIGUOUS]}, + }, + { + "$set": { + "status": STATUS_PENDING, + "next_attempt_at": now, + "updated_at": now, + }, + "$unset": { + "error": "", + "failed_at": "", + "ambiguous_at": "", + }, + }, + return_document=True, + ) + if result is None: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="This delivery cannot be retried.", + ) + return {"id": str(result["_id"]), "status": result["status"]} + + +@router.get( + "/conversations/{handoff_id}", + response_model=HandoffConversationRead, +) +async def get_handoff_conversation( + handoff_id: str, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + if not ObjectId.is_valid(handoff_id): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid handoff ID.", + ) + account_id = str(account.id) + handoff = await database.human_handoffs.find_one( + {"_id": ObjectId(handoff_id), "account_id": account_id} + ) + if handoff is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Handoff conversation not found.", + ) + cursor = database.mailbox_entries.find( + { + "account_id": account_id, + "thread_id": handoff["handoff_thread_id"], + "message_scope": "HANDOFF", + "deletion_state": {"$in": [None, ""]}, + } + ).sort([("event_at", 1), ("_id", 1)]) + messages: list[HandoffMessageRead] = [] + async for entry in cursor: + message = await database.messages.find_one({"_id": entry["message_id"]}) + if not message: + continue + delivery = await database.handoff_delivery_jobs.find_one( + { + "$or": [ + {"source_message_id": message["_id"]}, + {"target_message_id": message["_id"]}, + ] + }, + sort=[("created_at", -1)], + ) + messages.append( + HandoffMessageRead( + id=str(message["_id"]), + role="human" if message.get("direction") == "inbound" else "agent", + subject=message.get("subject", ""), + snippet=message.get("snippet", ""), + body_text=message.get("body_text", ""), + body_html=message.get("body_html"), + new_reply_content=message.get("new_reply_content"), + raw_body=message.get("body_text") + if message.get("direction") == "inbound" + else None, + has_quoted_text=bool(message.get("has_quoted_text")), + relay_status=message.get("relay_status"), + delivery_status=(delivery or {}).get("status"), + delivery_job_id=str(delivery["_id"]) if delivery else None, + direction=message.get("direction", "outbound"), + mailbox=entry["mailbox"], + event_at=entry.get("event_at") + or message.get("received_at") + or message.get("sent_at") + or message["created_at"], + ) + ) + return HandoffConversationRead( + id=str(handoff["_id"]), + thread_id=str(handoff["handoff_thread_id"]), + rule_name=handoff.get("rule_name", "Human handoff"), + external_subject=handoff.get("external_subject", ""), + external_sender=handoff.get("external_sender", ""), + human_email=handoff.get("human_email", ""), + handoff_reason="Matched handoff rule", + status=handoff.get("status", "ACTIVE"), + status_label=_handoff_status_label(handoff.get("status", "ACTIVE")), + current_owner=handoff.get("current_owner", "HUMAN"), + linked_external=HandoffLinkedExternal( + thread_id=str(handoff["external_thread_id"]), + subject=handoff.get("external_subject", ""), + sender=handoff.get("external_sender", ""), + ), + messages=messages, + ) + + @router.post("", response_model=HandoffRuleRead, status_code=status.HTTP_201_CREATED) async def create_handoff_rule( payload: HandoffRuleCreate, @@ -62,9 +298,30 @@ async def create_handoff_rule( account: Account = Depends(get_current_account), ): now = _now() + account_id = str(account.id) + name = payload.name.strip() + duplicate = await database.human_handoff_rules.find_one( + {"account_id": account_id, "name_key": name.casefold()}, + {"_id": 1}, + ) + if duplicate: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="A handoff rule with this name already exists.", + ) + priority = payload.priority + if priority is None: + latest = await database.human_handoff_rules.find_one( + {"account_id": account_id}, + sort=[("priority", -1)], + ) + priority = int(latest.get("priority", -1)) + 1 if latest else 0 document = { "_id": ObjectId(), - "account_id": str(account.id), + "account_id": account_id, + "name": name, + "name_key": name.casefold(), + "priority": priority, "owner_email": str(payload.owner_email).lower(), "owner_name": payload.owner_name or _owner_name_from_email(str(payload.owner_email)), "labels": payload.labels, @@ -76,10 +333,60 @@ async def create_handoff_rule( "created_at": now, "updated_at": now, } - await database.human_handoff_rules.insert_one(document) + try: + await database.human_handoff_rules.insert_one(document) + except DuplicateKeyError: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="A handoff rule with this name or priority already exists.", + ) return _serialize(document) +@router.put("/reorder", response_model=list[HandoffRuleRead]) +async def reorder_handoff_rules( + payload: HandoffRuleOrderUpdate, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + if len(payload.rule_ids) != len(set(payload.rule_ids)): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Rule IDs must be unique.", + ) + if any(not ObjectId.is_valid(value) for value in payload.rule_ids): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid rule ID.", + ) + account_id = str(account.id) + object_ids = [ObjectId(value) for value in payload.rule_ids] + count = await database.human_handoff_rules.count_documents( + {"account_id": account_id, "_id": {"$in": object_ids}} + ) + if count != len(object_ids): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="One or more handoff rules were not found.", + ) + now = _now() + # Shift priorities out of the unique range first so swaps cannot collide. + for offset, rule_id in enumerate(object_ids): + await database.human_handoff_rules.update_one( + {"_id": rule_id, "account_id": account_id}, + {"$set": {"priority": offset + 10_000, "updated_at": now}}, + ) + for priority, rule_id in enumerate(object_ids): + await database.human_handoff_rules.update_one( + {"_id": rule_id, "account_id": account_id}, + {"$set": {"priority": priority, "updated_at": now}}, + ) + cursor = database.human_handoff_rules.find( + {"account_id": account_id} + ).sort([("priority", 1), ("created_at", 1), ("_id", 1)]) + return [_serialize(document) async for document in cursor] + + @router.patch("/{rule_id}", response_model=HandoffRuleRead) async def update_handoff_rule( rule_id: str, @@ -90,14 +397,36 @@ async def update_handoff_rule( if not ObjectId.is_valid(rule_id): raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid rule ID.") patch = payload.model_dump(exclude_unset=True) + if "name" in patch: + patch["name"] = patch["name"].strip() + patch["name_key"] = patch["name"].casefold() + duplicate = await database.human_handoff_rules.find_one( + { + "account_id": str(account.id), + "name_key": patch["name_key"], + "_id": {"$ne": ObjectId(rule_id)}, + }, + {"_id": 1}, + ) + if duplicate: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="A handoff rule with this name already exists.", + ) if "owner_email" in patch and patch["owner_email"]: patch["owner_email"] = str(patch["owner_email"]).lower() patch["updated_at"] = _now() - result = await database.human_handoff_rules.find_one_and_update( - {"_id": ObjectId(rule_id), "account_id": str(account.id)}, - {"$set": patch}, - return_document=True, - ) + try: + result = await database.human_handoff_rules.find_one_and_update( + {"_id": ObjectId(rule_id), "account_id": str(account.id)}, + {"$set": patch}, + return_document=True, + ) + except DuplicateKeyError: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="A handoff rule with this name or priority already exists.", + ) if result is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Handoff rule not found.") return _serialize(result) @@ -152,7 +481,7 @@ async def test_handoff( owner_email=owner_email, **evaluation, result_path=( - "Would match handoff rule (configuration only)" + "Would hand this email to the Human Owner" if evaluation["would_handoff"] else "Would not match handoff rule" ), diff --git a/backend/app/routers/emails.py b/backend/app/routers/emails.py index a055a29..bae1ba8 100644 --- a/backend/app/routers/emails.py +++ b/backend/app/routers/emails.py @@ -1,5 +1,6 @@ import base64 import json +import logging import re from datetime import date, datetime, time, timedelta, timezone from typing import Annotated @@ -35,10 +36,13 @@ from app.services.thread_service import ( message_event_at, ) from app.services.external_reply_service import build_external_reply_plan, log_smtp_message_sending +from app.services.handoff_mirror_policy import GENERATED_BY_AGENT, MESSAGE_ORIGIN_AGENT +from app.services.human_handoff_service import maybe_trigger_handoff from app.services.mailbox_scope import apply_external_scope from app.services.thread_resolver import ThreadResolution MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" +logger = logging.getLogger(__name__) router = APIRouter(prefix="/emails", tags=["emails"]) @@ -191,6 +195,11 @@ async def create_email( ) created_at = utc_now() + sender_name = " ".join( + part.strip() + for part in [account.first_name, account.last_name] + if part and part.strip() + ) subject = payload.subject if parent_message: reply_plan = build_external_reply_plan( @@ -213,10 +222,12 @@ async def create_email( document = { "direction": "outbound", "message_scope": MESSAGE_SCOPE_EXTERNAL, + "message_origin": MESSAGE_ORIGIN_AGENT, + "generated_by": GENERATED_BY_AGENT, "account_id": str(account.id), "sender": address( sender, - " ".join(part for part in [account.first_name, account.last_name] if part), + sender_name, ), "to": [address(value) for value in recipients], "cc": [address(value) for value in cc], @@ -251,6 +262,7 @@ async def create_email( ) internet_message_id = await deliver_email( sender=sender, + sender_name=sender_name, recipients=recipients, cc=cc, bcc=bcc, @@ -324,6 +336,22 @@ async def create_email( "deletion_state": None, } ) + if document["status"] == "accepted": + try: + await maybe_trigger_handoff( + database, + account_id=str(account.id), + agent_email=sender, + external_message=document, + external_thread_id=thread_id, + ) + except Exception: + logger.exception( + "handoff_trigger_failed account_id=%s thread_id=%s message_id=%s", + account.id, + thread_id, + document["_id"], + ) if document["status"] == "failed": raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, diff --git a/backend/app/schemas.py b/backend/app/schemas.py index aea9398..dd4cb5d 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -232,8 +232,10 @@ class HandoffEmailSample(BaseModel): class HandoffRuleCreate(BaseModel): + name: str = Field(min_length=1, max_length=120) owner_email: EmailStr owner_name: str | None = None + priority: int | None = Field(default=None, ge=0, le=10000) labels: list[str] = Field(default_factory=list) email_types: list[str] = Field(default_factory=list) instructions: str = "" @@ -243,8 +245,10 @@ class HandoffRuleCreate(BaseModel): class HandoffRuleUpdate(BaseModel): + name: str | None = Field(default=None, min_length=1, max_length=120) owner_email: EmailStr | None = None owner_name: str | None = None + priority: int | None = Field(default=None, ge=0, le=10000) labels: list[str] | None = None email_types: list[str] | None = None instructions: str | None = None @@ -255,8 +259,10 @@ class HandoffRuleUpdate(BaseModel): class HandoffRuleRead(BaseModel): id: str + name: str owner_email: EmailStr owner_name: str + priority: int labels: list[str] email_types: list[str] instructions: str @@ -267,6 +273,10 @@ class HandoffRuleRead(BaseModel): updated_at: datetime +class HandoffRuleOrderUpdate(BaseModel): + rule_ids: list[str] = Field(min_length=1, max_length=100) + + class HandoffTestRequest(BaseModel): rule_id: str | None = None owner_email: EmailStr | None = None @@ -297,6 +307,9 @@ class HandoffLinkedExternal(BaseModel): class HandoffConversationSummary(BaseModel): id: str + thread_id: str + mailbox: Literal["handoff_inbox", "handoff_outbox"] + rule_name: str external_subject: str external_sender: str human_email: str @@ -314,15 +327,22 @@ class HandoffMessageRead(BaseModel): subject: str snippet: str body_text: str + body_html: str | None = None new_reply_content: str | None = None raw_body: str | None = None has_quoted_text: bool = False relay_status: str | None = None + delivery_status: str | None = None + delivery_job_id: str | None = None + direction: Literal["inbound", "outbound"] + mailbox: Literal["handoff_inbox", "handoff_outbox"] event_at: datetime class HandoffConversationRead(BaseModel): id: str + thread_id: str + rule_name: str external_subject: str external_sender: str human_email: str diff --git a/backend/app/services/external_reply_service.py b/backend/app/services/external_reply_service.py index 2d5bb29..b7399e7 100644 --- a/backend/app/services/external_reply_service.py +++ b/backend/app/services/external_reply_service.py @@ -17,6 +17,7 @@ from app.services.message_ids import ( message_id_fields, ) from app.services.mime_parser import parse_mime +from app.services.mime_body_extractor import ExtractedMimePart from app.services.reply_extractor import build_reply_display from app.services.raw_mime_storage import get_raw_mime_storage from app.services.thread_service import threading_fields_from_parent @@ -351,6 +352,9 @@ async def send_external_reply( generated_by: str | None = None, generated_from_handoff_message_id: ObjectId | None = None, message_origin: str | None = None, + message_id: str | None = None, + inline_parts: list[ExtractedMimePart] | None = None, + file_attachments: list[ExtractedMimePart] | None = None, ) -> dict: if generated_from_handoff_message_id: existing = await database.messages.find_one( @@ -386,6 +390,7 @@ async def send_external_reply( internet_message_id = await deliver_email( sender=sender, + sender_name=sender_name, recipients=plan.recipients, cc=[], bcc=[], @@ -395,6 +400,9 @@ async def send_external_reply( domain=domain, in_reply_to=plan.threading.in_reply_to, references=plan.threading.references, + message_id=message_id, + inline_parts=inline_parts, + file_attachments=file_attachments, ) normalized_id = canonical_message_id(internet_message_id) id_fields = message_id_fields(normalized_id) @@ -428,7 +436,16 @@ async def send_external_reply( "has_quoted_text": reply_display["has_quoted_text"], "body_text": plan.body_text, "body_html": plan.body_html, - "attachments": [], + "attachments": [ + { + "filename": part.filename, + "content_type": part.content_type, + "size": len(part.payload), + "content_id": part.content_id, + "disposition": part.disposition, + } + for part in [*(inline_parts or []), *(file_attachments or [])] + ], "created_at": now, "sent_at": now, "status": "accepted", @@ -469,7 +486,7 @@ async def send_external_reply( "recipients": document["to"], "subject": plan.subject, "snippet": document["snippet"], - "attachments": [], + "attachments": document["attachments"], "status": "accepted", "is_read": True, "is_starred": False, diff --git a/backend/app/services/handoff_delivery.py b/backend/app/services/handoff_delivery.py new file mode 100644 index 0000000..31b6c16 --- /dev/null +++ b/backend/app/services/handoff_delivery.py @@ -0,0 +1,215 @@ +import asyncio +import email.utils +import logging +from datetime import datetime, timezone + +from bson import ObjectId + +from app.database import mongo_db + +logger = logging.getLogger(__name__) + +JOB_MIRROR = "MIRROR_TO_HUMAN" +JOB_RELAY = "RELAY_TO_EXTERNAL" + +STATUS_PENDING = "PENDING" +STATUS_SENDING = "SENDING" +STATUS_SENT = "SENT" +STATUS_FAILED = "FAILED" +STATUS_AMBIGUOUS = "AMBIGUOUS" + + +class HandoffPreflightError(RuntimeError): + """A deterministic error that happened before SMTP submission.""" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +async def enqueue_handoff_delivery( + database, + *, + kind: str, + account_id: str, + handoff_id: ObjectId, + source_message_id: ObjectId, + agent_email: str, +) -> dict: + idempotency_key = f"{kind}:{account_id}:{handoff_id}:{source_message_id}" + now = _now() + document = { + "idempotency_key": idempotency_key, + "kind": kind, + "account_id": account_id, + "handoff_id": handoff_id, + "source_message_id": source_message_id, + "agent_email": agent_email.lower(), + "generated_message_id": email.utils.make_msgid( + domain=agent_email.rsplit("@", 1)[-1].lower() + ), + "status": STATUS_PENDING, + "attempt_count": 0, + "next_attempt_at": now, + "created_at": now, + "updated_at": now, + } + await database.handoff_delivery_jobs.update_one( + {"idempotency_key": idempotency_key}, + {"$setOnInsert": document}, + upsert=True, + ) + return await database.handoff_delivery_jobs.find_one( + {"idempotency_key": idempotency_key} + ) + + +async def process_handoff_delivery_once(database=mongo_db) -> bool: + now = _now() + job = await database.handoff_delivery_jobs.find_one_and_update( + { + "status": STATUS_PENDING, + "next_attempt_at": {"$lte": now}, + }, + { + "$set": { + "status": STATUS_SENDING, + "started_at": now, + "updated_at": now, + }, + "$inc": {"attempt_count": 1}, + }, + sort=[("created_at", 1), ("_id", 1)], + return_document=True, + ) + if not job: + return False + + try: + from app.services.human_handoff_service import deliver_handoff_job + + target_message_id = await deliver_handoff_job(database, job) + except HandoffPreflightError as error: + logger.error( + "handoff_delivery_preflight_failed job=%s kind=%s error=%s", + job["_id"], + job["kind"], + error, + ) + await database.handoff_delivery_jobs.update_one( + {"_id": job["_id"], "status": STATUS_SENDING}, + { + "$set": { + "status": STATUS_FAILED, + "error": str(error), + "failed_at": _now(), + "updated_at": _now(), + } + }, + ) + if job["kind"] == JOB_RELAY: + await database.messages.update_one( + {"_id": job["source_message_id"]}, + { + "$set": { + "relay_status": STATUS_FAILED, + "relay_error": str(error), + "relay_failed_at": _now(), + } + }, + ) + except Exception as error: + # SMTP exceptions can be ambiguous: the remote MTA may have accepted the + # message before the acknowledgement was lost. Never auto-retry these. + logger.exception( + "handoff_delivery_ambiguous job=%s kind=%s", + job["_id"], + job["kind"], + ) + await database.handoff_delivery_jobs.update_one( + {"_id": job["_id"], "status": STATUS_SENDING}, + { + "$set": { + "status": STATUS_AMBIGUOUS, + "error": str(error), + "ambiguous_at": _now(), + "updated_at": _now(), + } + }, + ) + if job["kind"] == JOB_RELAY: + await database.messages.update_one( + {"_id": job["source_message_id"]}, + { + "$set": { + "relay_status": STATUS_AMBIGUOUS, + "relay_error": str(error), + "relay_failed_at": _now(), + } + }, + ) + else: + await database.handoff_delivery_jobs.update_one( + {"_id": job["_id"], "status": STATUS_SENDING}, + { + "$set": { + "status": STATUS_SENT, + "target_message_id": target_message_id, + "sent_at": _now(), + "updated_at": _now(), + } + }, + ) + return True + + +class HandoffDeliveryWorker: + def __init__(self, poll_seconds: float = 1.0): + self.poll_seconds = poll_seconds + self._task: asyncio.Task | None = None + self._stop = asyncio.Event() + + async def start(self) -> None: + if self._task and not self._task.done(): + return + # A process that died during SMTP submission cannot know whether the + # remote MTA accepted it. Quarantine instead of risking a duplicate. + await mongo_db.handoff_delivery_jobs.update_many( + {"status": STATUS_SENDING}, + { + "$set": { + "status": STATUS_AMBIGUOUS, + "error": "Worker restarted during SMTP submission.", + "ambiguous_at": _now(), + "updated_at": _now(), + } + }, + ) + self._stop.clear() + self._task = asyncio.create_task(self._run()) + + async def stop(self) -> None: + self._stop.set() + if self._task: + await self._task + self._task = None + + async def _run(self) -> None: + while not self._stop.is_set(): + processed = False + try: + processed = await process_handoff_delivery_once() + except Exception: + logger.exception("handoff_delivery_worker_iteration_failed") + if processed: + continue + try: + await asyncio.wait_for( + self._stop.wait(), + timeout=self.poll_seconds, + ) + except TimeoutError: + pass + + +handoff_delivery_worker = HandoffDeliveryWorker() diff --git a/backend/app/services/handoff_mirror.py b/backend/app/services/handoff_mirror.py index 7073dc4..53f5530 100644 --- a/backend/app/services/handoff_mirror.py +++ b/backend/app/services/handoff_mirror.py @@ -1,13 +1,5 @@ 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 @@ -21,11 +13,6 @@ class HandoffMirrorPayload: 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] = [] @@ -44,29 +31,12 @@ async def load_mirror_payload(external_message: dict) -> HandoffMirrorPayload: 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, - ), + body_text=body_text, + body_html=body_html, inline_parts=inline_parts, file_attachments=file_attachments, ) @@ -77,32 +47,4 @@ async def load_mirror_update_payload( *, 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, - ) + return await load_mirror_payload(external_message) diff --git a/backend/app/services/handoff_mirror_policy.py b/backend/app/services/handoff_mirror_policy.py index ac403a9..495e8f4 100644 --- a/backend/app/services/handoff_mirror_policy.py +++ b/backend/app/services/handoff_mirror_policy.py @@ -19,6 +19,7 @@ GENERATED_BY_RELAY_ACK = "RELAY_ACK" RELAY_STATUS_SENT = "SENT" RELAY_STATUS_FAILED = "FAILED" RELAY_STATUS_PENDING = "PENDING" +RELAY_STATUS_BLOCKED = "BLOCKED" def should_mirror_to_human(message: dict) -> bool: @@ -27,7 +28,7 @@ def should_mirror_to_human(message: dict) -> bool: if scope != MESSAGE_SCOPE_EXTERNAL: return False - if message.get("direction") != "inbound": + if message.get("direction") not in ("inbound", "outbound"): return False generated_by = message.get("generated_by") @@ -49,6 +50,11 @@ def should_mirror_to_human(message: dict) -> bool: if message.get("generated_from_handoff_message_id"): return False + if message.get("direction") == "outbound": + return ( + generated_by == GENERATED_BY_AGENT + or origin == MESSAGE_ORIGIN_AGENT + ) return True diff --git a/backend/app/services/human_handoff_service.py b/backend/app/services/human_handoff_service.py index 1b5d8db..bd9b9f1 100644 --- a/backend/app/services/human_handoff_service.py +++ b/backend/app/services/human_handoff_service.py @@ -1,8 +1,918 @@ -"""Human Handoff runtime is disabled — configuration-only via agent_handoff router. +import logging +import re +from datetime import datetime, timezone +from uuid import UUID -This module retains scope constants used by normal mail filtering. All email -sending, receiving, mirroring, forwarding, and relay behavior has been removed. -""" +from bson import ObjectId +from pymongo.errors import DuplicateKeyError + +from app.database import SessionFactory +from app.models import Account +from app.services.domain_provisioning import get_domain_by_name +from app.services.external_reply_service import ( + ExternalReplyThreadingError, + build_external_reply_plan, + resolve_external_reply_parent, + resolve_internet_message_id, + send_external_reply, +) +from app.services.handoff_delivery import ( + HandoffPreflightError, + JOB_MIRROR, + JOB_RELAY, + enqueue_handoff_delivery, +) +from app.services.handoff_evaluator import evaluate_handoff +from app.services.handoff_mirror import load_mirror_payload +from app.services.handoff_mirror_policy import ( + GENERATED_BY_AGENT, + GENERATED_BY_HANDOFF_BRIDGE, + GENERATED_BY_HUMAN, + MESSAGE_ORIGIN_AGENT, + MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, + MESSAGE_ORIGIN_HANDOFF_MIRROR, + MESSAGE_ORIGIN_HUMAN_OWNER, + RELAY_STATUS_BLOCKED, + RELAY_STATUS_FAILED, + RELAY_STATUS_PENDING, + RELAY_STATUS_SENT, + should_mirror_to_human, +) +from app.services.mail_sender import deliver_email +from app.services.message_ids import ( + canonical_message_id, + folded_message_id, + message_id_fields, + parse_message_id_list, +) +from app.services.mime_body_extractor import extract_new_reply_mime_content +from app.services.raw_mime_storage import get_raw_mime_storage +from app.services.reply_extractor import ( + extract_new_reply_content, + validate_external_reply_body, +) +from app.services.thread_service import ( + attach_message_to_thread, + message_event_at, + threading_fields_from_parent, +) + +logger = logging.getLogger(__name__) MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" MESSAGE_SCOPE_HANDOFF = "HANDOFF" +THREAD_TYPE_HUMAN_HANDOFF = "HUMAN_HANDOFF" + +HANDOFF_STATUS_ACTIVE = "ACTIVE" +HANDOFF_STATUS_WAITING = "WAITING_FOR_HUMAN" +HANDOFF_STATUS_HUMAN_REPLIED = "HUMAN_REPLIED" + +OWNER_AGENT = "AGENT" +OWNER_HUMAN = "HUMAN" + +LINK_EXTERNAL_MIRRORED = "EXTERNAL_MIRRORED_TO_HUMAN" +LINK_HUMAN_RELAYED = "HUMAN_RELAYED_TO_EXTERNAL" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _snippet(value: str, limit: int = 240) -> str: + return re.sub(r"\s+", " ", value or "").strip()[:limit] + + +def _safe_subject_part(value: str, fallback: str) -> str: + cleaned = re.sub(r"[\r\n]+", " ", value or "").strip() + return cleaned or fallback + + +def build_handoff_subject( + *, + rule_name: str, + external_sender: str, + external_subject: str, +) -> str: + subject = " · ".join( + [ + _safe_subject_part(rule_name, "Human"), + "handoff", + _safe_subject_part(external_sender, "unknown sender"), + _safe_subject_part(external_subject, "(No subject)"), + ] + ) + return subject[:998].rstrip() + + +def redact_human_address( + value: str | None, + *, + human_email: str, + agent_email: str, +) -> str | None: + if value is None or not human_email: + return value + return re.sub( + re.escape(human_email), + agent_email, + value, + flags=re.IGNORECASE, + ) + + +def select_matching_handoff_rule( + rules: list[dict], + *, + email: dict, +) -> dict | None: + ordered = sorted( + rules, + key=lambda rule: ( + int(rule.get("priority", 0)), + ( + rule["created_at"].timestamp() + if isinstance(rule.get("created_at"), datetime) + else 0 + ), + str(rule.get("_id", "")), + ), + ) + return next( + ( + rule + for rule in ordered + if evaluate_handoff( + email=email, + labels=rule.get("labels", []), + email_types=rule.get("email_types", []), + instructions=rule.get("instructions", ""), + )["would_handoff"] + ), + None, + ) + + +async def _account_and_domain(account_id: str, agent_email: str): + async with SessionFactory() as session: + account = await session.get(Account, UUID(account_id)) + domain = await get_domain_by_name(session, agent_email.rsplit("@", 1)[-1]) + if account is None: + raise HandoffPreflightError("Handoff account no longer exists.") + if domain is None: + raise HandoffPreflightError("Agent mail domain is not configured.") + if domain.domain_type == "company" and domain.status != "active": + raise HandoffPreflightError("Agent mail domain is not active.") + sender_name = " ".join( + part.strip() + for part in [account.first_name, account.last_name] + if part and part.strip() + ) + return account, domain, sender_name + + +async def create_handoff_thread( + database, + *, + account_id: str, + subject: str, +) -> ObjectId: + now = _now() + result = await database.email_threads.insert_one( + { + "account_id": account_id, + "thread_type": THREAD_TYPE_HUMAN_HANDOFF, + "normalized_subject": subject.casefold(), + "created_at": now, + "updated_at": now, + "last_message_at": now, + "latest_message_id": None, + "message_count": 0, + } + ) + return result.inserted_id + + +async def _latest_thread_message( + database, + *, + account_id: str, + thread_id: ObjectId, + scope: str, +) -> dict | None: + pipeline = [ + { + "$match": { + "account_id": account_id, + "thread_id": thread_id, + "message_scope": scope, + "deletion_state": {"$in": [None, ""]}, + } + }, + {"$sort": {"event_at": -1, "_id": -1}}, + {"$limit": 1}, + { + "$lookup": { + "from": "messages", + "localField": "message_id", + "foreignField": "_id", + "as": "message", + } + }, + {"$unwind": "$message"}, + ] + cursor = await database.mailbox_entries.aggregate(pipeline) + rows = [row async for row in cursor] + return rows[0]["message"] if rows else None + + +async def find_handoff_for_external_thread( + database, + *, + account_id: str, + external_thread_id: ObjectId, +) -> dict | None: + return await database.human_handoffs.find_one( + { + "account_id": account_id, + "external_thread_id": external_thread_id, + "status": {"$ne": "CLOSED"}, + } + ) + + +async def resolve_handoff_from_inbound( + database, + *, + account_id: str, + agent_email: str, + sender_address: str, + in_reply_to: str | None, + references: list[str], +) -> dict | None: + sender = sender_address.strip().lower() + if not sender or not agent_email: + return None + search_ids = parse_message_id_list( + " ".join([*([in_reply_to] if in_reply_to else []), *(references or [])]) + ) + for internet_id in reversed(search_ids): + folded = folded_message_id(internet_id) + if not folded: + continue + parent = await database.messages.find_one( + { + "message_scope": MESSAGE_SCOPE_HANDOFF, + "$or": [ + {"message_id_folded": folded}, + {"message_id": canonical_message_id(internet_id)}, + ], + }, + {"_id": 1}, + ) + if parent is None: + continue + entry = await database.mailbox_entries.find_one( + { + "account_id": account_id, + "message_id": parent["_id"], + "message_scope": MESSAGE_SCOPE_HANDOFF, + }, + {"thread_id": 1}, + ) + if entry is None: + continue + handoff = await database.human_handoffs.find_one( + { + "account_id": account_id, + "handoff_thread_id": entry["thread_id"], + "human_email": sender, + "agent_email": agent_email.lower(), + "status": {"$ne": "CLOSED"}, + } + ) + if handoff: + return handoff + return None + + +async def _record_link( + database, + *, + handoff_id: ObjectId, + source_message_id: ObjectId, + external_message_id: ObjectId | None, + handoff_message_id: ObjectId | None, + relationship: str, +) -> None: + await database.handoff_message_links.update_one( + { + "handoff_id": handoff_id, + "source_message_id": source_message_id, + "relationship": relationship, + }, + { + "$setOnInsert": { + "handoff_id": handoff_id, + "source_message_id": source_message_id, + "external_message_id": external_message_id, + "handoff_message_id": handoff_message_id, + "relationship": relationship, + "created_at": _now(), + } + }, + upsert=True, + ) + + +def _attachment_metadata(parts) -> list[dict]: + return [ + { + "filename": part.filename, + "content_type": part.content_type, + "size": len(part.payload), + "content_id": part.content_id, + "disposition": part.disposition, + } + for part in parts + ] + + +async def store_handoff_message( + database, + *, + account_id: str, + thread_id: ObjectId, + direction: str, + sender: dict, + recipients: list[dict], + subject: str, + body_text: str, + body_html: str | None, + internet_message_id: str | None, + in_reply_to: str | None, + references: list[str], + handoff_id: ObjectId, + event_at: datetime | None = None, + attachments: list[dict] | None = None, + raw_mime: dict | None = None, + handoff_delivery_key: str | None = None, + new_reply_content: str | None = None, + new_reply_html: str | None = None, + quoted_content: str | None = None, + message_origin: str | None = None, + generated_by: str | None = None, + mirrored_from_external_message_id: ObjectId | None = None, + relay_status: str | None = None, +) -> dict: + if handoff_delivery_key: + existing = await database.messages.find_one( + {"handoff_delivery_key": handoff_delivery_key} + ) + if existing: + return existing + + now = event_at or _now() + id_fields = message_id_fields(canonical_message_id(internet_message_id)) + document = { + "account_id": account_id, + "direction": direction, + "message_scope": MESSAGE_SCOPE_HANDOFF, + "handoff_id": handoff_id, + "sender": sender, + "to": recipients, + "subject": subject, + "snippet": _snippet(new_reply_content or body_text), + "body_text": body_text, + "body_html": body_html, + "new_reply_content": new_reply_content, + "new_reply_html": new_reply_html, + "quoted_content": quoted_content or "", + "has_quoted_text": bool(quoted_content), + "attachments": attachments or [], + "created_at": now, + "received_at": now if direction == "inbound" else None, + "sent_at": now if direction == "outbound" else None, + "status": "received" if direction == "inbound" else "accepted", + **id_fields, + "in_reply_to": canonical_message_id(in_reply_to), + "references": references or [], + "message_origin": message_origin, + "generated_by": generated_by, + "relay_status": relay_status, + } + if raw_mime: + document["raw_mime"] = raw_mime + if handoff_delivery_key: + document["handoff_delivery_key"] = handoff_delivery_key + if mirrored_from_external_message_id: + document["mirrored_from_external_message_id"] = mirrored_from_external_message_id + try: + result = await database.messages.insert_one(document) + document["_id"] = result.inserted_id + except DuplicateKeyError: + query = ( + {"handoff_delivery_key": handoff_delivery_key} + if handoff_delivery_key + else {"mirrored_from_external_message_id": mirrored_from_external_message_id} + ) + existing = await database.messages.find_one(query) + if existing: + return existing + raise + + await attach_message_to_thread( + database, + account_id=account_id, + thread_id=thread_id, + message_id=document["_id"], + subject=subject, + event_at=now, + ) + mailbox = "handoff_inbox" if direction == "inbound" else "handoff_outbox" + await database.mailbox_entries.update_one( + { + "account_id": account_id, + "message_id": document["_id"], + "mailbox": mailbox, + }, + { + "$setOnInsert": { + "account_id": account_id, + "message_id": document["_id"], + "thread_id": thread_id, + "mailbox": mailbox, + "message_scope": MESSAGE_SCOPE_HANDOFF, + "direction": direction, + "sender": sender, + "recipients": recipients, + "subject": subject, + "snippet": document["snippet"], + "attachments": attachments or [], + "status": document["status"], + "is_read": direction == "outbound", + "is_starred": False, + "event_at": now, + "created_at": now, + "trashed_at": None, + "deletion_state": None, + } + }, + upsert=True, + ) + return document + + +async def _create_handoff_mapping( + database, + *, + account_id: str, + agent_email: str, + external_message: dict, + external_thread_id: ObjectId, + rule: dict, +) -> dict: + sender = external_message.get("sender") or {} + sender_text = sender.get("address") or sender.get("name") or "unknown sender" + private_subject = build_handoff_subject( + rule_name=rule.get("name") or rule.get("owner_name") or "Human", + external_sender=sender_text, + external_subject=external_message.get("subject", ""), + ) + thread_id = await create_handoff_thread( + database, + account_id=account_id, + subject=private_subject, + ) + now = _now() + document = { + "account_id": account_id, + "external_thread_id": external_thread_id, + "handoff_thread_id": thread_id, + "rule_id": rule["_id"], + "rule_name": rule.get("name") or rule.get("owner_name") or "Human", + "rule_priority": rule.get("priority", 0), + "human_email": rule["owner_email"].lower(), + "human_name": rule.get("owner_name") or rule["owner_email"].split("@", 1)[0], + "agent_email": agent_email.lower(), + "external_sender": sender.get("address", "").lower(), + "external_subject": external_message.get("subject", ""), + "private_subject": private_subject, + "allow_replies": rule.get("allow_replies", True), + "status": HANDOFF_STATUS_WAITING, + "current_owner": OWNER_HUMAN, + "trigger_external_message_id": external_message["_id"], + "trigger_internet_message_id": external_message.get("internet_message_id") + or external_message.get("message_id"), + "created_at": now, + "updated_at": now, + } + try: + result = await database.human_handoffs.insert_one(document) + document["_id"] = result.inserted_id + return document + except DuplicateKeyError: + await database.email_threads.delete_one({"_id": thread_id, "message_count": 0}) + existing = await find_handoff_for_external_thread( + database, + account_id=account_id, + external_thread_id=external_thread_id, + ) + if existing: + return existing + raise + + +async def maybe_trigger_handoff( + database, + *, + account_id: str, + agent_email: str, + external_message: dict, + external_thread_id: ObjectId, +) -> dict | None: + if not should_mirror_to_human(external_message): + return None + sender_address = ( + (external_message.get("sender") or {}).get("address") or "" + ).lower() + if sender_address == agent_email.lower(): + return None + + handoff = await find_handoff_for_external_thread( + database, + account_id=account_id, + external_thread_id=external_thread_id, + ) + if handoff is None: + if external_message.get("direction") != "inbound": + return None + owner_sender = await database.human_handoff_rules.find_one( + { + "account_id": account_id, + "owner_email": sender_address, + "active": True, + }, + {"_id": 1}, + ) + if owner_sender: + return None + rules = await database.human_handoff_rules.find( + {"account_id": account_id, "active": True} + ).sort([("priority", 1), ("created_at", 1), ("_id", 1)]).to_list(length=100) + email_payload = { + "subject": external_message.get("subject", ""), + "snippet": external_message.get("snippet", ""), + "body_text": external_message.get("body_text", ""), + "is_starred": False, + } + rule = select_matching_handoff_rule( + rules, + email=email_payload, + ) + if rule is None: + return None + handoff = await _create_handoff_mapping( + database, + account_id=account_id, + agent_email=agent_email, + external_message=external_message, + external_thread_id=external_thread_id, + rule=rule, + ) + + await enqueue_handoff_delivery( + database, + kind=JOB_MIRROR, + account_id=account_id, + handoff_id=handoff["_id"], + source_message_id=external_message["_id"], + agent_email=agent_email, + ) + return handoff + + +async def process_human_reply( + database, + *, + account_id: str, + agent_email: str, + handoff: dict, + parsed: dict, + body_text: str, + body_html: str | None, + raw_mime: dict, + delivery_key: str, +) -> dict: + extracted = extract_new_reply_content( + body_text=body_text, + body_html=body_html, + ) + rich = None + try: + raw = await get_raw_mime_storage().get(raw_mime) + rich = extract_new_reply_mime_content(raw) + except Exception: + logger.exception("handoff_reply_rich_extract_failed delivery=%s", delivery_key) + + human_email = (parsed.get("sender") or {}).get("address", "").lower() + new_text = ( + rich.body_text if rich and rich.body_text else extracted["new_reply_content"] + ).strip() + new_html = rich.body_html if rich else None + if human_email: + new_text = redact_human_address( + new_text, + human_email=human_email, + agent_email=agent_email, + ) + new_html = redact_human_address( + new_html, + human_email=human_email, + agent_email=agent_email, + ) + auto_submitted = ( + (parsed.get("raw_headers") or {}).get("auto_submitted") or "" + ).strip().lower() + is_automatic = bool(auto_submitted and auto_submitted != "no") + private_markers_ok = validate_external_reply_body(new_text) + relay_status = ( + RELAY_STATUS_PENDING + if ( + new_text + and private_markers_ok + and not is_automatic + and handoff.get("allow_replies", True) + ) + else RELAY_STATUS_BLOCKED + ) + human_message = await store_handoff_message( + database, + account_id=account_id, + thread_id=handoff["handoff_thread_id"], + direction="inbound", + sender=parsed.get("sender") or {"name": "", "address": human_email}, + recipients=parsed.get("to", []), + subject=parsed.get("subject", ""), + body_text=body_text, + body_html=body_html, + internet_message_id=parsed.get("message_id"), + in_reply_to=parsed.get("in_reply_to"), + references=parsed.get("references", []), + handoff_id=handoff["_id"], + event_at=parsed.get("sent_at") or _now(), + attachments=parsed.get("attachments", []), + raw_mime=raw_mime, + handoff_delivery_key=f"handoff:{account_id}:{handoff['_id']}:{delivery_key}", + new_reply_content=new_text, + new_reply_html=new_html, + quoted_content=extracted.get("quoted_content", ""), + message_origin=MESSAGE_ORIGIN_HUMAN_OWNER, + generated_by=GENERATED_BY_HUMAN, + relay_status=relay_status, + ) + if relay_status == RELAY_STATUS_PENDING: + await enqueue_handoff_delivery( + database, + kind=JOB_RELAY, + account_id=account_id, + handoff_id=handoff["_id"], + source_message_id=human_message["_id"], + agent_email=agent_email, + ) + else: + await database.messages.update_one( + {"_id": human_message["_id"]}, + { + "$set": { + "relay_error": ( + "Human replies are disabled for this handoff." + if not handoff.get("allow_replies", True) + else ( + "Automatic replies are not relayed through Human Handoff." + if is_automatic + else "Reply content could not be safely separated from private history." + ) + ) + } + }, + ) + await database.human_handoffs.update_one( + {"_id": handoff["_id"]}, + { + "$set": { + "status": HANDOFF_STATUS_HUMAN_REPLIED, + "current_owner": OWNER_AGENT, + "last_human_message_at": _now(), + "updated_at": _now(), + } + }, + ) + return human_message + + +async def _deliver_mirror(database, job: dict, handoff: dict, source: dict) -> ObjectId: + existing = await database.messages.find_one( + { + "message_scope": MESSAGE_SCOPE_HANDOFF, + "mirrored_from_external_message_id": source["_id"], + } + ) + if existing: + return existing["_id"] + + _, domain, sender_name = await _account_and_domain( + job["account_id"], handoff["agent_email"] + ) + payload = await load_mirror_payload(source) + parent = await _latest_thread_message( + database, + account_id=job["account_id"], + thread_id=handoff["handoff_thread_id"], + scope=MESSAGE_SCOPE_HANDOFF, + ) + threading = threading_fields_from_parent(parent) if parent else {} + subject = handoff["private_subject"] + if parent and not re.match(r"^re:\s*", subject, re.IGNORECASE): + subject = f"Re: {subject}" + internet_message_id = await deliver_email( + sender=handoff["agent_email"], + sender_name=sender_name, + recipients=[handoff["human_email"]], + cc=[], + bcc=[], + subject=subject, + body_text=payload.body_text, + body_html=payload.body_html, + domain=domain, + in_reply_to=threading.get("in_reply_to"), + references=threading.get("references"), + message_id=job["generated_message_id"], + inline_parts=payload.inline_parts, + file_attachments=payload.file_attachments, + ) + attachments = _attachment_metadata( + [*payload.inline_parts, *payload.file_attachments] + ) + mirror = await store_handoff_message( + database, + account_id=job["account_id"], + thread_id=handoff["handoff_thread_id"], + direction="outbound", + sender={"name": sender_name, "address": handoff["agent_email"]}, + recipients=[ + { + "name": handoff.get("human_name", ""), + "address": handoff["human_email"], + } + ], + subject=subject, + body_text=payload.body_text, + body_html=payload.body_html, + internet_message_id=internet_message_id, + in_reply_to=threading.get("in_reply_to"), + references=threading.get("references", []), + handoff_id=handoff["_id"], + attachments=attachments, + message_origin=MESSAGE_ORIGIN_HANDOFF_MIRROR, + generated_by=GENERATED_BY_HANDOFF_BRIDGE, + mirrored_from_external_message_id=source["_id"], + ) + await _record_link( + database, + handoff_id=handoff["_id"], + source_message_id=source["_id"], + external_message_id=source["_id"], + handoff_message_id=mirror["_id"], + relationship=LINK_EXTERNAL_MIRRORED, + ) + await database.human_handoffs.update_one( + {"_id": handoff["_id"]}, + { + "$set": { + "status": HANDOFF_STATUS_WAITING, + "current_owner": OWNER_HUMAN, + "last_mirrored_at": _now(), + "updated_at": _now(), + } + }, + ) + return mirror["_id"] + + +async def _deliver_relay(database, job: dict, handoff: dict, source: dict) -> ObjectId: + existing = await database.messages.find_one( + { + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "generated_from_handoff_message_id": source["_id"], + } + ) + if existing: + return existing["_id"] + new_text = (source.get("new_reply_content") or "").strip() + if not new_text or not validate_external_reply_body(new_text): + raise HandoffPreflightError( + "Reply content failed the private-content leakage guard." + ) + external_parent = await resolve_external_reply_parent( + database, + account_id=job["account_id"], + external_thread_id=handoff["external_thread_id"], + customer_address=handoff["external_sender"], + handoff=handoff, + ) + if external_parent is None: + raise HandoffPreflightError("No valid external reply parent was found.") + parent_internet_id = await resolve_internet_message_id(database, external_parent) + if not parent_internet_id: + raise HandoffPreflightError( + "The external reply parent has no RFC Message-ID." + ) + recipient = handoff.get("external_sender") or ( + external_parent.get("sender") or {} + ).get("address") + if not recipient: + raise HandoffPreflightError("The external sender address is missing.") + + _, domain, sender_name = await _account_and_domain( + job["account_id"], handoff["agent_email"] + ) + inline_parts = [] + file_attachments = [] + body_html = source.get("new_reply_html") + if source.get("raw_mime"): + raw = await get_raw_mime_storage().get(source["raw_mime"]) + rich = extract_new_reply_mime_content(raw) + inline_parts = rich.inline_parts + file_attachments = rich.file_attachments + plan = build_external_reply_plan( + parent_external_message=external_parent, + body_text=new_text, + body_html=body_html, + subject_override=None, + recipients=[recipient], + parent_internet_id_override=parent_internet_id, + ) + try: + external = await send_external_reply( + database, + account_id=job["account_id"], + sender=handoff["agent_email"], + sender_name=sender_name, + external_thread_id=handoff["external_thread_id"], + plan=plan, + domain=domain, + resolution_method="human_handoff_bridge", + generated_by=GENERATED_BY_HANDOFF_BRIDGE, + generated_from_handoff_message_id=source["_id"], + message_origin=MESSAGE_ORIGIN_HUMAN_OWNER, + message_id=job["generated_message_id"], + inline_parts=inline_parts, + file_attachments=file_attachments, + ) + except ExternalReplyThreadingError as error: + raise HandoffPreflightError(str(error)) from error + await database.messages.update_one( + {"_id": source["_id"]}, + { + "$set": { + "relay_status": RELAY_STATUS_SENT, + "generated_external_message_id": external["_id"], + "relay_delivered_at": _now(), + } + }, + ) + await _record_link( + database, + handoff_id=handoff["_id"], + source_message_id=source["_id"], + external_message_id=external["_id"], + handoff_message_id=source["_id"], + relationship=LINK_HUMAN_RELAYED, + ) + await database.human_handoffs.update_one( + {"_id": handoff["_id"]}, + { + "$set": { + "status": HANDOFF_STATUS_ACTIVE, + "current_owner": OWNER_AGENT, + "updated_at": _now(), + } + }, + ) + return external["_id"] + + +async def deliver_handoff_job(database, job: dict) -> ObjectId: + handoff = await database.human_handoffs.find_one( + {"_id": job["handoff_id"], "account_id": job["account_id"]} + ) + if handoff is None: + raise HandoffPreflightError("Linked handoff no longer exists.") + source = await database.messages.find_one({"_id": job["source_message_id"]}) + if source is None: + raise HandoffPreflightError("Source message no longer exists.") + if job["kind"] == JOB_MIRROR: + return await _deliver_mirror(database, job, handoff, source) + if job["kind"] == JOB_RELAY: + return await _deliver_relay(database, job, handoff, source) + raise HandoffPreflightError(f"Unknown handoff delivery kind: {job['kind']}") diff --git a/backend/app/services/inbound_smtp.py b/backend/app/services/inbound_smtp.py index 0589b4a..9a64854 100644 --- a/backend/app/services/inbound_smtp.py +++ b/backend/app/services/inbound_smtp.py @@ -10,7 +10,16 @@ from app.config import get_settings from app.database import SessionFactory, mongo_db from app.models import Account from app.services.external_reply_service import log_original_external_message +from app.services.handoff_mirror_policy import ( + GENERATED_BY_CUSTOMER, + MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, +) from app.services.human_handoff_service import MESSAGE_SCOPE_EXTERNAL +from app.services.human_handoff_service import ( + maybe_trigger_handoff, + process_human_reply, + resolve_handoff_from_inbound, +) from app.services.message_ids import canonical_message_id, message_id_fields from app.services.mime_parser import parse_mime from app.services.reply_extractor import build_reply_display @@ -121,26 +130,65 @@ class InboundHandler: body_text = parsed.pop("body_text") body_html = parsed.pop("body_html") stored = await get_raw_mime_storage().put(raw, delivery_key) + raw_mime = { + "bucket": stored.bucket, + "name": stored.name, + "generation": stored.generation, + "size": stored.size, + "crc32c": stored.crc32c, + } now = datetime.now(timezone.utc) sender_address = (parsed.get("sender") or {}).get("address", "").lower() internet_message_id = canonical_message_id(parsed.get("message_id")) id_fields = message_id_fields(internet_message_id) + external_recipients: list[str] = [] + for recipient in recipients: + account_id = envelope.account_ids[recipient] + handoff = await resolve_handoff_from_inbound( + mongo_db, + account_id=account_id, + agent_email=recipient, + sender_address=sender_address, + in_reply_to=parsed.get("in_reply_to"), + references=parsed.get("references", []), + ) + if handoff is None: + external_recipients.append(recipient) + continue + await process_human_reply( + mongo_db, + account_id=account_id, + agent_email=recipient, + handoff=handoff, + parsed=parsed, + body_text=body_text, + body_html=body_html, + raw_mime=raw_mime, + delivery_key=delivery_key, + ) + logger.info( + "handoff_reply_received account_id=%s handoff_id=%s message_id=%s", + account_id, + handoff["_id"], + internet_message_id, + ) + + if not external_recipients: + return f"250 2.0.0 Handoff message accepted as {delivery_key}" + recipients = external_recipients + message = { **parsed, "direction": "inbound", "message_scope": MESSAGE_SCOPE_EXTERNAL, + "message_origin": MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, + "generated_by": GENERATED_BY_CUSTOMER, "envelope_sender": envelope.mail_from.lower(), "envelope_recipients": recipients, "delivery_key": delivery_key, **id_fields, - "raw_mime": { - "bucket": stored.bucket, - "name": stored.name, - "generation": stored.generation, - "size": stored.size, - "crc32c": stored.crc32c, - }, + "raw_mime": raw_mime, "received_at": now, "created_at": now, "has_text_body": bool(body_text), @@ -275,6 +323,21 @@ class InboundHandler: thread_id, mailbox_result.upserted_id is not None or mailbox_result.modified_count > 0, ) + try: + await maybe_trigger_handoff( + mongo_db, + account_id=account_id, + agent_email=recipient, + external_message=persisted, + external_thread_id=thread_id, + ) + except Exception: + logger.exception( + "handoff_trigger_failed account_id=%s thread_id=%s message_id=%s", + account_id, + thread_id, + persisted["_id"], + ) except Exception: logger.exception("Temporary failure accepting inbound message") return "451 4.3.0 Temporary server failure; please retry" diff --git a/backend/app/services/mail_sender.py b/backend/app/services/mail_sender.py index a441a4d..02dd868 100644 --- a/backend/app/services/mail_sender.py +++ b/backend/app/services/mail_sender.py @@ -16,6 +16,36 @@ from app.services.mime_body_extractor import ExtractedMimePart SMTP_POLICY = email.policy.SMTP.clone(max_line_length=998) logger = logging.getLogger(__name__) +SYSTEM_SENDER_LOCALS = { + "invite", + "invites", + "notification", + "notifications", + "no-reply", + "noreply", + "support", +} + + +def _sender_identity(sender: str, sender_name: str | None = None) -> tuple[str, str]: + """Return an RFC-compliant From header and the bare SMTP envelope address.""" + if "\r" in sender or "\n" in sender or (sender_name and ("\r" in sender_name or "\n" in sender_name)): + raise ValueError("Sender identity cannot contain line breaks.") + + existing_name, address = email.utils.parseaddr(sender) + address = address.strip().lower() + if not address or "@" not in address: + raise ValueError("A valid sender email address is required.") + + display_name = (sender_name or existing_name).strip() + if not display_name: + local_part = address.split("@", 1)[0] + if local_part in SYSTEM_SENDER_LOCALS: + display_name = "aGentic Email" + else: + display_name = local_part.replace(".", " ").replace("_", " ").replace("-", " ").title() + + return email.utils.formataddr((display_name, address), charset="utf-8"), address def log_outbound_mime_headers(message: Message) -> None: @@ -62,6 +92,7 @@ def _mime_attachment(part: ExtractedMimePart) -> MIMEBase: def build_message( *, sender: str, + sender_name: str | None = None, recipients: list[str], cc: list[str], subject: str, @@ -76,10 +107,11 @@ def build_message( ) -> Message: inline_parts = inline_parts or [] file_attachments = file_attachments or [] + sender_header, sender_address = _sender_identity(sender, sender_name) if inline_parts or file_attachments: root = MIMEMultipart("mixed", policy=SMTP_POLICY) - root["From"] = sender + root["From"] = sender_header root["To"] = ", ".join(recipients) if cc: root["Cc"] = ", ".join(cc) @@ -87,7 +119,7 @@ def build_message( 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]) + generated_id = message_id or email.utils.make_msgid(domain=sender_address.split("@")[-1]) root["Message-ID"] = generated_id if in_reply_to: root["In-Reply-To"] = in_reply_to @@ -115,7 +147,7 @@ def build_message( return root message = EmailMessage(policy=SMTP_POLICY) - message["From"] = sender + message["From"] = sender_header message["To"] = ", ".join(recipients) if cc: message["Cc"] = ", ".join(cc) @@ -123,7 +155,7 @@ def build_message( message["Reply-To"] = reply_to message["Subject"] = subject message["Date"] = email.utils.formatdate(localtime=True) - generated_id = message_id or email.utils.make_msgid(domain=sender.split("@")[-1]) + generated_id = message_id or email.utils.make_msgid(domain=sender_address.split("@")[-1]) message["Message-ID"] = generated_id if in_reply_to: message["In-Reply-To"] = in_reply_to @@ -191,6 +223,7 @@ async def send_via_smtp( async def deliver_email( *, sender: str, + sender_name: str | None = None, recipients: list[str], cc: list[str], bcc: list[str], @@ -205,8 +238,10 @@ async def deliver_email( inline_parts: list[ExtractedMimePart] | None = None, file_attachments: list[ExtractedMimePart] | None = None, ) -> str: + _, sender_address = _sender_identity(sender, sender_name) message = build_message( sender=sender, + sender_name=sender_name, recipients=recipients, cc=cc, subject=subject, @@ -224,7 +259,7 @@ async def deliver_email( log_signed_mime_headers(signed) await send_via_smtp( signed, - sender=sender, + sender=sender_address, recipients=[*recipients, *cc, *bcc], ) return str(message["Message-ID"]) diff --git a/backend/app/services/mime_body_extractor.py b/backend/app/services/mime_body_extractor.py index cb89a7a..53b9490 100644 --- a/backend/app/services/mime_body_extractor.py +++ b/backend/app/services/mime_body_extractor.py @@ -2,6 +2,10 @@ from dataclasses import dataclass, field from email import policy from email.message import EmailMessage, Message from email.parser import BytesParser +from html.parser import HTMLParser +import re + +from app.services.reply_extractor import extract_new_reply_content @dataclass @@ -104,3 +108,109 @@ def extract_mirror_content(raw: bytes) -> MirroredMimeContent: collector = MirroredMimeContent() _walk_parts(message, collector) return collector + + +class _NewReplyHTMLParser(HTMLParser): + QUOTE_CLASSES = { + "gmail_attr", + "gmail_quote", + "protonmail_quote", + "yahoo_quoted", + "moz-cite-prefix", + } + QUOTE_IDS = {"divRplyFwdMsg", "appendonsend"} + VOID_TAGS = {"br", "hr", "img", "meta", "link", "input"} + + def __init__(self) -> None: + super().__init__(convert_charrefs=False) + self.output: list[str] = [] + self.open_tags: list[str] = [] + self.stopped = False + + def _is_quote_start(self, tag: str, attrs: list[tuple[str, str | None]]) -> bool: + values = {key.lower(): (value or "") for key, value in attrs} + classes = set(values.get("class", "").split()) + return ( + tag.lower() == "blockquote" + or bool(classes.intersection(self.QUOTE_CLASSES)) + or values.get("id") in self.QUOTE_IDS + ) + + def handle_starttag(self, tag: str, attrs) -> None: + if self.stopped: + return + if self._is_quote_start(tag, attrs): + self.stopped = True + return + self.output.append(self.get_starttag_text()) + if tag.lower() not in self.VOID_TAGS: + self.open_tags.append(tag) + + def handle_startendtag(self, tag: str, attrs) -> None: + if not self.stopped: + self.output.append(self.get_starttag_text()) + + def handle_endtag(self, tag: str) -> None: + if self.stopped: + return + self.output.append(f"") + if tag in self.open_tags: + reverse_index = self.open_tags[::-1].index(tag) + del self.open_tags[len(self.open_tags) - reverse_index - 1] + + def handle_data(self, data: str) -> None: + if self.stopped: + return + if re.search(r"(?:^|\n)\s*On .+? wrote:\s*$", data, re.IGNORECASE): + self.stopped = True + return + self.output.append(data) + + def handle_entityref(self, name: str) -> None: + if not self.stopped: + self.output.append(f"&{name};") + + def handle_charref(self, name: str) -> None: + if not self.stopped: + self.output.append(f"&#{name};") + + def html(self) -> str: + value = "".join(self.output).strip() + if not value: + return "" + return value + "".join(f"" for tag in reversed(self.open_tags)) + + +def _new_reply_html(html: str | None) -> str | None: + if not html: + return None + parser = _NewReplyHTMLParser() + parser.feed(html) + parser.close() + value = parser.html().strip() + return value or None + + +def extract_new_reply_mime_content(raw: bytes) -> MirroredMimeContent: + """Extract only newly authored MIME content while retaining rich formatting.""" + full = extract_mirror_content(raw) + extracted = extract_new_reply_content( + body_text=full.body_text, + body_html=full.body_html, + ) + new_html = _new_reply_html(full.body_html) + referenced_cids = { + match.casefold() + for match in re.findall(r"cid:([^\"'\s>)]+)", new_html or "", re.IGNORECASE) + } + inline_parts = [ + part + for part in full.inline_parts + if part.content_id and part.content_id.casefold() in referenced_cids + ] + return MirroredMimeContent( + body_text=extracted["new_reply_content"].strip(), + body_html=new_html, + inline_parts=inline_parts, + file_attachments=full.file_attachments, + ) diff --git a/backend/app/services/mime_parser.py b/backend/app/services/mime_parser.py index 7f478bb..58a1490 100644 --- a/backend/app/services/mime_parser.py +++ b/backend/app/services/mime_parser.py @@ -94,6 +94,11 @@ def parse_mime(raw: bytes) -> dict: "to": _header_str(message.get("To")), "cc": _header_str(message.get("Cc")), "reply_to": _header_str(message.get("Reply-To")), + "auto_submitted": _header_str(message.get("Auto-Submitted")), + "precedence": _header_str(message.get("Precedence")), + "x_auto_response_suppress": _header_str( + message.get("X-Auto-Response-Suppress") + ), } return { "message_id": internet_message_id, diff --git a/backend/app/services/org_invitations.py b/backend/app/services/org_invitations.py index 238a99d..125abd9 100644 --- a/backend/app/services/org_invitations.py +++ b/backend/app/services/org_invitations.py @@ -44,9 +44,10 @@ async def send_employee_invitation( expires_hours=INVITE_EXPIRY_HOURS, ) - sender = f"{BRAND_NAME} <{INVITE_FROM_LOCAL}@{settings.platform_domain}>" + sender = f"{INVITE_FROM_LOCAL}@{settings.platform_domain}" await deliver_email( sender=sender, + sender_name=BRAND_NAME, recipients=[member.invitation_email], cc=[], bcc=[], diff --git a/backend/app/services/reply_extractor.py b/backend/app/services/reply_extractor.py index 4ab8b87..8a57cd2 100644 --- a/backend/app/services/reply_extractor.py +++ b/backend/app/services/reply_extractor.py @@ -16,6 +16,7 @@ TEXT_QUOTE_MARKERS = [ PRIVATE_LEAK_MARKERS = [ "human handoff", + " · handoff · ", "a conversation needs your attention", "why this was handed off", "why you're receiving this", @@ -37,11 +38,19 @@ def _normalize_newlines(text: str) -> str: class _ReplyHTMLTextParser(HTMLParser): BLOCK_TAGS = {"div", "p", "pre", "li", "ul", "ol", "table", "tr"} IGNORED_TAGS = {"script", "style", "iframe", "object"} + QUOTE_CLASSES = { + "gmail_attr", + "gmail_quote", + "protonmail_quote", + "yahoo_quoted", + "moz-cite-prefix", + } def __init__(self) -> None: super().__init__(convert_charrefs=True) self.parts: list[str] = [] self.quote_depth = 0 + self.quote_tags: list[str] = [] self.ignored_depth = 0 self.at_line_start = True @@ -70,11 +79,19 @@ class _ReplyHTMLTextParser(HTMLParser): return if self.ignored_depth: return + classes = { + value + for key, value in attrs + if key.lower() == "class" + for value in (value or "").split() + } + is_quote = tag == "blockquote" or bool(classes.intersection(self.QUOTE_CLASSES)) if tag == "br": self._newline() - elif tag == "blockquote": + elif is_quote: self._newline() self.quote_depth += 1 + self.quote_tags.append(tag) elif tag in self.BLOCK_TAGS: self._newline() @@ -85,9 +102,10 @@ class _ReplyHTMLTextParser(HTMLParser): return if self.ignored_depth: return - if tag == "blockquote": - self._newline() + if self.quote_tags and self.quote_tags[-1] == tag: + self.quote_tags.pop() self.quote_depth = max(0, self.quote_depth - 1) + self._newline() elif tag in self.BLOCK_TAGS: self._newline() diff --git a/backend/app/services/welcome_email.py b/backend/app/services/welcome_email.py index 0e581f0..0ef88fb 100644 --- a/backend/app/services/welcome_email.py +++ b/backend/app/services/welcome_email.py @@ -31,9 +31,10 @@ async def send_welcome_email(session: AsyncSession, account: Account) -> None: login_url=login_url, is_company_account=account.account_type == "company", ) - sender = f"AGMail Team <{NOTIFICATIONS_FROM_LOCAL}@{settings.platform_domain}>" + sender = f"{NOTIFICATIONS_FROM_LOCAL}@{settings.platform_domain}" await deliver_email( sender=sender, + sender_name=BRAND_NAME, recipients=[account.email], cc=[], bcc=[], diff --git a/backend/tests/test_handoff_delivery.py b/backend/tests/test_handoff_delivery.py new file mode 100644 index 0000000..1e80dd8 --- /dev/null +++ b/backend/tests/test_handoff_delivery.py @@ -0,0 +1,83 @@ +from datetime import datetime, timezone + +import pytest +from bson import ObjectId + +from app.services import human_handoff_service +from app.services.handoff_delivery import ( + STATUS_AMBIGUOUS, + STATUS_PENDING, + STATUS_SENT, + process_handoff_delivery_once, +) + + +class FakeJobs: + def __init__(self): + self.job = { + "_id": ObjectId(), + "kind": "MIRROR_TO_HUMAN", + "account_id": "account-1", + "handoff_id": ObjectId(), + "source_message_id": ObjectId(), + "status": STATUS_PENDING, + "created_at": datetime.now(timezone.utc), + "next_attempt_at": datetime.now(timezone.utc), + } + + async def find_one_and_update(self, query, update, **kwargs): + if self.job["status"] != STATUS_PENDING: + return None + self.job["status"] = "SENDING" + self.job["attempt_count"] = self.job.get("attempt_count", 0) + 1 + return dict(self.job) + + async def update_one(self, query, update): + self.job.update(update.get("$set", {})) + + +class FakeMessages: + async def update_one(self, query, update): + return None + + +class FakeDatabase: + def __init__(self): + self.handoff_delivery_jobs = FakeJobs() + self.messages = FakeMessages() + + +@pytest.mark.asyncio +async def test_delivery_job_is_claimed_and_sent_once(monkeypatch): + database = FakeDatabase() + calls = 0 + + async def deliver(database, job): + nonlocal calls + calls += 1 + return ObjectId() + + monkeypatch.setattr(human_handoff_service, "deliver_handoff_job", deliver) + + assert await process_handoff_delivery_once(database) is True + assert await process_handoff_delivery_once(database) is False + assert calls == 1 + assert database.handoff_delivery_jobs.job["status"] == STATUS_SENT + + +@pytest.mark.asyncio +async def test_ambiguous_smtp_failure_is_not_automatically_retried(monkeypatch): + database = FakeDatabase() + calls = 0 + + async def deliver(database, job): + nonlocal calls + calls += 1 + raise TimeoutError("SMTP acknowledgement lost") + + monkeypatch.setattr(human_handoff_service, "deliver_handoff_job", deliver) + + assert await process_handoff_delivery_once(database) is True + assert await process_handoff_delivery_once(database) is False + assert calls == 1 + assert database.handoff_delivery_jobs.job["status"] == STATUS_AMBIGUOUS diff --git a/backend/tests/test_handoff_mirror.py b/backend/tests/test_handoff_mirror.py index 6746b95..7bf5108 100644 --- a/backend/tests/test_handoff_mirror.py +++ b/backend/tests/test_handoff_mirror.py @@ -1,41 +1,45 @@ +import pytest + 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.handoff_mirror import load_mirror_payload from app.services.html_sanitizer import sanitize_email_html -from app.services.mime_body_extractor import extract_mirror_content +from app.services.mime_body_extractor import ( + extract_mirror_content, + extract_new_reply_mime_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", +@pytest.mark.asyncio +async def test_mirror_body_is_unchanged_without_gentic_wrapper(): + payload = await load_mirror_payload( + { + "body_text": "one last attempt. i am tired", + "body_html": "

one last attempt. i am tired

", + } ) - 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 + assert payload.body_text == "one last attempt. i am tired" + assert payload.body_html == ( + "

one last attempt. i am tired

" + ) + assert "Human Handoff" not in payload.body_text + assert "gentic-handoff" not in (payload.body_html or "") -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", +@pytest.mark.asyncio +async def test_mirror_plain_text_fallback_is_unchanged(): + payload = await load_mirror_payload( + { + "body_text": "one last attempt. i am tired", + "body_html": None, + } ) - assert "one last attempt. i am tired" in text - assert "LATEST MESSAGE" not in text - assert "Reply normally" in text + assert payload.body_text == "one last attempt. i am tired" + assert payload.body_html is None def test_format_sender_with_name(): @@ -83,3 +87,42 @@ def test_extract_html_from_multipart(): extracted = extract_mirror_content(raw) assert "Invoice" in (extracted.body_html or "") assert "" in (extracted.body_html or "") + + +def test_rich_human_reply_keeps_new_html_and_removes_private_quote(): + raw = ( + b"From: owner@example.com\r\n" + b"To: agent@gentic.email\r\n" + b"Subject: Re: Finance handoff\r\n" + b'MIME-Version: 1.0\r\nContent-Type: multipart/alternative; boundary="b"\r\n\r\n' + b"--b\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n" + b"Approved with urgency.\r\n\r\n" + b"On Tue, Oct 6, 2026 at 6:00 AM agent@gentic.email wrote:\r\n" + b"> private mirror\r\n" + b"--b\r\nContent-Type: text/html; charset=utf-8\r\n\r\n" + b'
Approved with urgency.
' + b'
private mirror
\r\n' + b"--b--\r\n" + ) + extracted = extract_new_reply_mime_content(raw) + assert extracted.body_text == "Approved with urgency." + assert "Approved" in (extracted.body_html or "") + assert "private mirror" not in (extracted.body_html or "") + + +def test_rich_human_reply_keeps_new_attachment(): + raw = ( + b"From: owner@example.com\r\n" + b"To: agent@gentic.email\r\n" + b"MIME-Version: 1.0\r\n" + b'Content-Type: multipart/mixed; boundary="m"\r\n\r\n' + b"--m\r\nContent-Type: text/plain; charset=utf-8\r\n\r\nApproved\r\n" + b"--m\r\nContent-Type: application/pdf\r\n" + b'Content-Disposition: attachment; filename="approval.pdf"\r\n\r\nPDF\r\n' + b"--m--\r\n" + ) + extracted = extract_new_reply_mime_content(raw) + assert extracted.body_text == "Approved" + assert [part.filename for part in extracted.file_attachments] == [ + "approval.pdf" + ] diff --git a/backend/tests/test_handoff_mirror_policy.py b/backend/tests/test_handoff_mirror_policy.py index 91eecc1..906283b 100644 --- a/backend/tests/test_handoff_mirror_policy.py +++ b/backend/tests/test_handoff_mirror_policy.py @@ -1,11 +1,13 @@ from bson import ObjectId from app.services.handoff_mirror_policy import ( + GENERATED_BY_AGENT, GENERATED_BY_CUSTOMER, GENERATED_BY_HANDOFF_BRIDGE, GENERATED_BY_RELAY_ACK, MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, MESSAGE_ORIGIN_HUMAN_OWNER, + MESSAGE_ORIGIN_AGENT, is_relay_acknowledgment_message, should_mirror_to_human, ) @@ -40,6 +42,27 @@ def test_should_not_mirror_human_origin(): assert should_mirror_to_human(message) is False +def test_should_mirror_agent_authored_external_reply(): + message = { + "message_scope": "EXTERNAL", + "direction": "outbound", + "generated_by": GENERATED_BY_AGENT, + "message_origin": MESSAGE_ORIGIN_AGENT, + } + assert should_mirror_to_human(message) is True + + +def test_should_not_echo_human_relay_back_to_owner(): + message = { + "message_scope": "EXTERNAL", + "direction": "outbound", + "generated_by": GENERATED_BY_HANDOFF_BRIDGE, + "message_origin": MESSAGE_ORIGIN_HUMAN_OWNER, + "generated_from_handoff_message_id": ObjectId(), + } + assert should_mirror_to_human(message) is False + + def test_is_relay_acknowledgment_detects_legacy_body(): message = { "direction": "outbound", diff --git a/backend/tests/test_handoff_runtime.py b/backend/tests/test_handoff_runtime.py new file mode 100644 index 0000000..d11cd9c --- /dev/null +++ b/backend/tests/test_handoff_runtime.py @@ -0,0 +1,423 @@ +from datetime import datetime, timezone +from types import SimpleNamespace + +import pytest +from bson import ObjectId + +from app.services.handoff_delivery import enqueue_handoff_delivery +from app.services.handoff_mirror_policy import ( + GENERATED_BY_AGENT, + GENERATED_BY_HANDOFF_BRIDGE, + MESSAGE_ORIGIN_AGENT, + MESSAGE_ORIGIN_HUMAN_OWNER, + RELAY_STATUS_BLOCKED, + should_mirror_to_human, +) +from app.services import human_handoff_service +from app.services.human_handoff_service import ( + MESSAGE_SCOPE_EXTERNAL, + MESSAGE_SCOPE_HANDOFF, + build_handoff_subject, + maybe_trigger_handoff, + process_human_reply, + resolve_handoff_from_inbound, +) +from app.services.mailbox_scope import apply_external_scope +from app.services.external_reply_service import build_external_reply_plan + + +class FakeCursor: + def __init__(self, items): + self.items = items + + def sort(self, *_args, **_kwargs): + return self + + async def to_list(self, length=None): + return list(self.items)[:length] + + +class FakeCollection: + def __init__(self, items=None): + self.items = list(items or []) + self.inserted = [] + self.updated = [] + + async def find_one(self, query, projection=None): + for item in self.items: + if _matches(item, query): + return item + return None + + def find(self, query=None): + query = query or {} + return FakeCursor([item for item in self.items if _matches(item, query)]) + + async def insert_one(self, document): + document = dict(document) + document.setdefault("_id", ObjectId()) + self.items.append(document) + self.inserted.append(document) + return SimpleNamespace(inserted_id=document["_id"]) + + async def update_one(self, query, update, upsert=False): + self.updated.append((query, update, upsert)) + existing = await self.find_one(query) + if existing is None and upsert: + inserted = {"_id": ObjectId(), **query, **update.get("$setOnInsert", {})} + self.items.append(inserted) + return SimpleNamespace(upserted_id=inserted["_id"], modified_count=0) + if existing is not None: + existing.update(update.get("$set", {})) + return SimpleNamespace(upserted_id=None, modified_count=1) + return SimpleNamespace(upserted_id=None, modified_count=0) + + async def find_one_and_update(self, query, update, **kwargs): + existing = await self.find_one(query) + if existing is None: + return None + existing.update(update.get("$set", {})) + return existing + + +def _matches(document, query): + for key, expected in query.items(): + if key == "$or": + if not any(_matches(document, clause) for clause in expected): + return False + continue + if key == "$ne": + continue + actual = document.get(key) + if isinstance(expected, dict): + if "$ne" in expected and actual == expected["$ne"]: + return False + if "$in" in expected and actual not in expected["$in"]: + return False + continue + if actual != expected: + return False + return True + + +def test_external_mailbox_filter_excludes_handoff_scope(): + match = {"account_id": "account-1", "mailbox": "inbox"} + apply_external_scope(match) + assert match["$or"] == [ + {"message_scope": "EXTERNAL"}, + {"message_scope": {"$exists": False}}, + ] + + +def test_bridge_plan_never_references_private_ids(): + customer = { + "_id": ObjectId(), + "internet_message_id": "", + "message_id": "", + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "references": [], + "subject": "Invoice 42", + } + private_id = "" + plan = build_external_reply_plan( + parent_external_message=customer, + body_text="Approved", + recipients=["customer@gmail.com"], + ) + assert plan.threading.in_reply_to == "" + assert private_id not in plan.threading.references + assert plan.subject == "Re: Invoice 42" + assert "handoff" not in plan.subject.lower() + + +@pytest.mark.asyncio +async def test_existing_handoff_is_sticky_without_reevaluating_rules(): + external_thread_id = ObjectId() + source_id = ObjectId() + handoff = { + "_id": ObjectId(), + "account_id": "account-1", + "external_thread_id": external_thread_id, + "status": "ACTIVE", + "rule_name": "Finance approvals", + } + database = SimpleNamespace( + human_handoffs=FakeCollection([handoff]), + human_handoff_rules=FakeCollection( + [ + { + "_id": ObjectId(), + "name": "Finance approvals", + "priority": 1, + "labels": ["Finance"], + "email_types": ["transactional"], + "instructions": "only invoices", + "active": True, + "created_at": datetime.now(timezone.utc), + } + ] + ), + handoff_delivery_jobs=FakeCollection(), + ) + message = { + "_id": source_id, + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "inbound", + "sender": {"address": "customer@example.com"}, + "subject": "Hello there", + "body_text": "just checking in", + "snippet": "just checking in", + } + + result = await maybe_trigger_handoff( + database, + account_id="account-1", + agent_email="agent@gentic.email", + external_message=message, + external_thread_id=external_thread_id, + ) + + assert result["_id"] == handoff["_id"] + assert database.handoff_delivery_jobs.items + assert ( + database.handoff_delivery_jobs.items[0]["kind"] == "MIRROR_TO_HUMAN" + ) + + +@pytest.mark.asyncio +async def test_duplicate_delivery_jobs_share_an_idempotency_key(): + database = SimpleNamespace(handoff_delivery_jobs=FakeCollection()) + handoff_id = ObjectId() + source_id = ObjectId() + first = await enqueue_handoff_delivery( + database, + kind="MIRROR_TO_HUMAN", + account_id="account-1", + handoff_id=handoff_id, + source_message_id=source_id, + agent_email="agent@gentic.email", + ) + second = await enqueue_handoff_delivery( + database, + kind="MIRROR_TO_HUMAN", + account_id="account-1", + handoff_id=handoff_id, + source_message_id=source_id, + agent_email="agent@gentic.email", + ) + assert first["idempotency_key"] == second["idempotency_key"] + assert len(database.handoff_delivery_jobs.items) == 1 + + +@pytest.mark.asyncio +async def test_human_reply_resolves_only_to_private_handoff(): + parent_id = ObjectId() + thread_id = ObjectId() + handoff = { + "_id": ObjectId(), + "account_id": "account-1", + "handoff_thread_id": thread_id, + "human_email": "owner@example.com", + "agent_email": "agent@gentic.email", + "status": "ACTIVE", + } + database = SimpleNamespace( + messages=FakeCollection( + [ + { + "_id": parent_id, + "message_scope": MESSAGE_SCOPE_HANDOFF, + "message_id": "", + "message_id_folded": "", + } + ] + ), + mailbox_entries=FakeCollection( + [ + { + "account_id": "account-1", + "message_id": parent_id, + "message_scope": MESSAGE_SCOPE_HANDOFF, + "thread_id": thread_id, + } + ] + ), + human_handoffs=FakeCollection([handoff]), + ) + + resolved = await resolve_handoff_from_inbound( + database, + account_id="account-1", + agent_email="agent@gentic.email", + sender_address="owner@example.com", + in_reply_to="", + references=[""], + ) + assert resolved["_id"] == handoff["_id"] + + +def test_human_echo_and_agent_mirror_policy(): + assert should_mirror_to_human( + { + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "outbound", + "generated_by": GENERATED_BY_HANDOFF_BRIDGE, + "message_origin": MESSAGE_ORIGIN_HUMAN_OWNER, + "generated_from_handoff_message_id": ObjectId(), + } + ) is False + assert should_mirror_to_human( + { + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "outbound", + "generated_by": GENERATED_BY_AGENT, + "message_origin": MESSAGE_ORIGIN_AGENT, + } + ) is True + + +def test_private_subject_stays_off_the_external_plane(): + private = build_handoff_subject( + rule_name="Finance approvals", + external_sender="customer@example.com", + external_subject="Invoice 42", + ) + assert private.startswith("Finance approvals · handoff ·") + assert "Invoice 42" in private + + +@pytest.mark.asyncio +async def test_first_matching_rule_creates_a_new_linked_handoff(): + external_thread_id = ObjectId() + source_id = ObjectId() + rule = { + "_id": ObjectId(), + "account_id": "account-1", + "name": "Finance approvals", + "priority": 1, + "owner_email": "owner@example.com", + "owner_name": "Owner", + "labels": ["All mail"], + "email_types": ["all"], + "instructions": "", + "active": True, + "allow_replies": True, + "created_at": datetime.now(timezone.utc), + } + database = SimpleNamespace( + human_handoffs=FakeCollection(), + human_handoff_rules=FakeCollection([rule]), + handoff_delivery_jobs=FakeCollection(), + email_threads=FakeCollection(), + ) + message = { + "_id": source_id, + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "inbound", + "sender": {"address": "customer@example.com"}, + "subject": "Invoice 42", + "body_text": "please review this invoice", + "snippet": "please review this invoice", + } + + result = await maybe_trigger_handoff( + database, + account_id="account-1", + agent_email="agent@gentic.email", + external_message=message, + external_thread_id=external_thread_id, + ) + + assert result["rule_name"] == "Finance approvals" + assert result["external_thread_id"] == external_thread_id + assert result["handoff_thread_id"] + assert result["private_subject"].startswith("Finance approvals · handoff ·") + assert database.handoff_delivery_jobs.items[0]["kind"] == "MIRROR_TO_HUMAN" + + +@pytest.mark.asyncio +async def test_maybe_trigger_does_not_echo_human_origin_messages(): + database = SimpleNamespace( + human_handoffs=FakeCollection( + [ + { + "_id": ObjectId(), + "account_id": "account-1", + "external_thread_id": ObjectId(), + "status": "ACTIVE", + } + ] + ), + handoff_delivery_jobs=FakeCollection(), + ) + result = await maybe_trigger_handoff( + database, + account_id="account-1", + agent_email="agent@gentic.email", + external_message={ + "_id": ObjectId(), + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "outbound", + "generated_by": GENERATED_BY_HANDOFF_BRIDGE, + "message_origin": MESSAGE_ORIGIN_HUMAN_OWNER, + "generated_from_handoff_message_id": ObjectId(), + "sender": {"address": "agent@gentic.email"}, + }, + external_thread_id=database.human_handoffs.items[0]["external_thread_id"], + ) + assert result is None + assert database.handoff_delivery_jobs.items == [] + + +@pytest.mark.asyncio +async def test_leaked_private_subject_blocks_human_relay(monkeypatch): + stored = {} + + async def store(_database, **kwargs): + stored.update(kwargs) + return {"_id": ObjectId(), **kwargs} + + enqueued = [] + + async def enqueue(*args, **kwargs): + enqueued.append(kwargs) + return {} + + class BrokenStorage: + async def get(self, *_args, **_kwargs): + raise RuntimeError("raw mime unavailable") + + monkeypatch.setattr(human_handoff_service, "store_handoff_message", store) + monkeypatch.setattr(human_handoff_service, "enqueue_handoff_delivery", enqueue) + monkeypatch.setattr(human_handoff_service, "get_raw_mime_storage", lambda: BrokenStorage()) + + database = SimpleNamespace(messages=FakeCollection(), human_handoffs=FakeCollection()) + await process_human_reply( + database, + account_id="account-1", + agent_email="agent@gentic.email", + handoff={ + "_id": ObjectId(), + "handoff_thread_id": ObjectId(), + "allow_replies": True, + }, + parsed={ + "sender": {"name": "Owner", "address": "owner@example.com"}, + "to": [{"address": "agent@gentic.email"}], + "subject": "Re: Finance approvals · handoff · customer@example.com · Invoice 42", + "message_id": "", + "in_reply_to": "", + "references": [""], + "attachments": [], + "raw_headers": {}, + }, + body_text=( + "Finance approvals · handoff · customer@example.com · Invoice 42\n" + "Please approve" + ), + body_html=None, + raw_mime={"bucket": "mail", "name": "raw.eml"}, + delivery_key="delivery-1", + ) + assert stored["relay_status"] == RELAY_STATUS_BLOCKED + assert enqueued == [] diff --git a/backend/tests/test_human_handoff_service.py b/backend/tests/test_human_handoff_service.py index 5119427..ccec723 100644 --- a/backend/tests/test_human_handoff_service.py +++ b/backend/tests/test_human_handoff_service.py @@ -1,13 +1,74 @@ -from app.services.human_handoff_service import MESSAGE_SCOPE_EXTERNAL, MESSAGE_SCOPE_HANDOFF +from app.services.human_handoff_service import ( + MESSAGE_SCOPE_EXTERNAL, + MESSAGE_SCOPE_HANDOFF, + build_handoff_subject, + maybe_trigger_handoff, + process_human_reply, + redact_human_address, + resolve_handoff_from_inbound, + select_matching_handoff_rule, +) def test_scope_constants_are_distinct(): assert MESSAGE_SCOPE_EXTERNAL != MESSAGE_SCOPE_HANDOFF -def test_human_handoff_runtime_is_disabled(): - import app.services.human_handoff_service as module +def test_human_handoff_runtime_is_enabled(): + assert callable(maybe_trigger_handoff) + assert callable(process_human_reply) + assert callable(resolve_handoff_from_inbound) - assert not hasattr(module, "maybe_trigger_handoff") - assert not hasattr(module, "process_human_reply") - assert not hasattr(module, "resolve_handoff_from_inbound") + +def test_private_subject_contains_rule_sender_and_external_subject(): + subject = build_handoff_subject( + rule_name="Finance approvals", + external_sender="customer@example.com", + external_subject="Invoice 42", + ) + assert subject == ( + "Finance approvals · handoff · customer@example.com · Invoice 42" + ) + + +def test_private_subject_removes_header_line_breaks(): + subject = build_handoff_subject( + rule_name="Urgent\r\nBcc: attacker@example.com", + external_sender="customer@example.com", + external_subject="Help", + ) + assert "\r" not in subject + assert "\n" not in subject + + +def test_human_address_is_replaced_before_external_relay(): + value = "Regards,\nOwner " + assert redact_human_address( + value, + human_email="private.owner@example.com", + agent_email="agent@gentic.email", + ) == "Regards,\nOwner " + + +def test_lowest_priority_matching_rule_wins(): + rules = [ + { + "name": "Backup", + "priority": 20, + "labels": ["All mail"], + "email_types": ["all"], + "instructions": "", + }, + { + "name": "Primary", + "priority": 1, + "labels": ["All mail"], + "email_types": ["all"], + "instructions": "", + }, + ] + selected = select_matching_handoff_rule( + rules, + email={"subject": "Hello", "body_text": "Need help"}, + ) + assert selected["name"] == "Primary" diff --git a/backend/tests/test_inbound_smtp.py b/backend/tests/test_inbound_smtp.py index df1142a..71b2f02 100644 --- a/backend/tests/test_inbound_smtp.py +++ b/backend/tests/test_inbound_smtp.py @@ -22,6 +22,7 @@ class FakeMessages: async def update_one(self, *args, **kwargs): self.updates += 1 + return SimpleNamespace(upserted_id=ObjectId(), modified_count=0) async def find_one(self, *args, **kwargs): return {"_id": self.id} @@ -33,6 +34,12 @@ class FakeEntries: async def update_one(self, *args, **kwargs): self.updates += 1 + return SimpleNamespace(upserted_id=ObjectId(), modified_count=0) + + +class FakeThreads: + async def update_one(self, *args, **kwargs): + return None class FakeStorage: @@ -84,7 +91,7 @@ async def test_rcpt_rejects_unknown_account(monkeypatch): async def test_rcpt_accepts_account_and_tracks_owner(monkeypatch): account = SimpleNamespace(id="account-1") monkeypatch.setattr(inbound_smtp, "SessionFactory", lambda: FakeSession(account)) - target = SimpleNamespace(rcpt_tos=[], rcpt_options=[]) + target = SimpleNamespace(rcpt_tos=[], rcpt_options=[], mail_from="sender@example.com") response = await inbound_smtp.InboundHandler().handle_RCPT( None, @@ -95,8 +102,8 @@ async def test_rcpt_accepts_account_and_tracks_owner(monkeypatch): ) assert response.startswith("250 2.1.5") - assert target.rcpt_tos == ["user@gentic.email"] - assert target.account_ids == {"user@gentic.email": "account-1"} + assert target.rcpt_tos == ["user@getall.help"] + assert target.account_ids == {"user@getall.help": "account-1"} @pytest.mark.asyncio @@ -106,15 +113,98 @@ async def test_data_returns_success_only_after_idempotent_records(monkeypatch): monkeypatch.setattr( inbound_smtp, "mongo_db", - SimpleNamespace(messages=messages, mailbox_entries=entries), + SimpleNamespace( + messages=messages, + mailbox_entries=entries, + email_threads=FakeThreads(), + ), ) monkeypatch.setattr(inbound_smtp, "get_raw_mime_storage", lambda: FakeStorage()) + async def no_handoff(*args, **kwargs): + return None + + monkeypatch.setattr(inbound_smtp, "resolve_handoff_from_inbound", no_handoff) + monkeypatch.setattr(inbound_smtp, "maybe_trigger_handoff", no_handoff) + async def ensure_thread(*args, **kwargs): + return ObjectId() + + monkeypatch.setattr(inbound_smtp, "ensure_thread_for_message", ensure_thread) response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope()) assert response.startswith("250 2.0.0") - assert messages.updates == 1 - assert entries.updates == 1 + assert messages.updates >= 1 + assert entries.updates >= 1 + + +@pytest.mark.asyncio +async def test_human_reply_stays_on_the_private_plane(monkeypatch): + messages = FakeMessages() + entries = FakeEntries() + monkeypatch.setattr( + inbound_smtp, + "mongo_db", + SimpleNamespace( + messages=messages, + mailbox_entries=entries, + email_threads=FakeThreads(), + ), + ) + monkeypatch.setattr(inbound_smtp, "get_raw_mime_storage", lambda: FakeStorage()) + + async def resolve(*args, **kwargs): + return {"_id": ObjectId()} + + processed = [] + + async def process(*args, **kwargs): + processed.append(kwargs) + return {"_id": ObjectId()} + + monkeypatch.setattr(inbound_smtp, "resolve_handoff_from_inbound", resolve) + monkeypatch.setattr(inbound_smtp, "process_human_reply", process) + + response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope()) + + assert response.startswith("250 2.0.0 Handoff message accepted") + assert processed + assert messages.updates == 0 + assert entries.updates == 0 + + +@pytest.mark.asyncio +async def test_data_still_accepted_when_handoff_trigger_fails(monkeypatch): + messages = FakeMessages() + entries = FakeEntries() + monkeypatch.setattr( + inbound_smtp, + "mongo_db", + SimpleNamespace( + messages=messages, + mailbox_entries=entries, + email_threads=FakeThreads(), + ), + ) + monkeypatch.setattr(inbound_smtp, "get_raw_mime_storage", lambda: FakeStorage()) + + async def no_handoff(*args, **kwargs): + return None + + async def boom(*args, **kwargs): + raise RuntimeError("mirror queue unavailable") + + monkeypatch.setattr(inbound_smtp, "resolve_handoff_from_inbound", no_handoff) + monkeypatch.setattr(inbound_smtp, "maybe_trigger_handoff", boom) + + async def ensure_thread(*args, **kwargs): + return ObjectId() + + monkeypatch.setattr(inbound_smtp, "ensure_thread_for_message", ensure_thread) + + response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope()) + + assert response.startswith("250 2.0.0") + assert messages.updates >= 1 @pytest.mark.asyncio diff --git a/backend/tests/test_mail_sender.py b/backend/tests/test_mail_sender.py index 0c5081d..fe60a47 100644 --- a/backend/tests/test_mail_sender.py +++ b/backend/tests/test_mail_sender.py @@ -1,7 +1,49 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, patch -from app.services.mail_sender import build_message, send_via_smtp, sign_message +import pytest + +from app.services.mail_sender import build_message, deliver_email, send_via_smtp, sign_message + + +def test_build_message_formats_named_sender_header(): + message = build_message( + sender="suyash@gentic.email", + sender_name="Suyash Dixit", + recipients=["recipient@example.com"], + cc=[], + subject="Named sender", + body_text="Hello", + body_html=None, + ) + + assert message["From"] == "Suyash Dixit " + + +def test_build_message_preserves_preformatted_transactional_sender(): + message = build_message( + sender="aGentic Email ", + recipients=["recipient@example.com"], + cc=[], + subject="Welcome", + body_text="Hello", + body_html=None, + ) + + assert message["From"] == "aGentic Email " + + +def test_build_message_adds_safe_fallback_sender_name(): + message = build_message( + sender="notifications@gentic.email", + recipients=["recipient@example.com"], + cc=[], + subject="Notification", + body_text="Hello", + body_html=None, + ) + + assert message["From"] == "aGentic Email " def test_dkim_sign_uses_selector_before_domain(): @@ -29,6 +71,7 @@ def test_dkim_sign_uses_selector_before_domain(): assert signer.call_args.args[2] == b"ritoesis.com" +@pytest.mark.asyncio async def test_send_via_smtp_uses_configured_client_hostname(): with patch( "app.services.mail_sender.aiosmtplib.send", @@ -43,6 +86,38 @@ async def test_send_via_smtp_uses_configured_client_hostname(): assert send_mock.await_args.kwargs["local_hostname"] == "backend.gentic.email" +@pytest.mark.asyncio +async def test_deliver_email_uses_bare_address_for_smtp_envelope(): + domain = SimpleNamespace( + domain_name="gentic.email", + dkim_selector="agmail", + dkim_private_key="private-key", + ) + with ( + patch( + "app.services.mail_sender.sign_message", + return_value=b"signed", + ), + patch( + "app.services.mail_sender.send_via_smtp", + new_callable=AsyncMock, + ) as send_mock, + ): + await deliver_email( + sender="notifications@gentic.email", + sender_name="aGentic Email", + recipients=["user@example.com"], + cc=[], + bcc=[], + subject="Welcome", + body_text="Hello", + body_html=None, + domain=domain, + ) + + assert send_mock.await_args.kwargs["sender"] == "notifications@gentic.email" + + def test_build_message_sets_in_reply_to_and_references(): message = build_message( sender="suyash@gentic.email", diff --git a/backend/tests/test_reply_extractor.py b/backend/tests/test_reply_extractor.py index 72bd678..07a732b 100644 --- a/backend/tests/test_reply_extractor.py +++ b/backend/tests/test_reply_extractor.py @@ -99,6 +99,26 @@ def test_html_reply_preserves_nested_quoted_history(): assert "> > sender: C" in result["quoted_content"] +def test_apple_mail_attribution_is_treated_as_quote(): + body = """Approved. + +On 6 Oct 2026, at 04:48, Suyash Dixit wrote: +sender: C""" + result = extract_new_reply_content(body_text=body) + assert result["new_reply_content"] == "Approved." + assert "sender: C" in result["quoted_content"] + + +def test_private_handoff_subject_fingerprint_is_blocked(): + from app.services.reply_extractor import validate_external_reply_body + + leaked = ( + "Finance approvals · handoff · customer@example.com · Invoice 42\n" + "Please approve" + ) + assert validate_external_reply_body(leaked) is False + + def test_outlook_original_message_split(): body = """Thanks for the update. diff --git a/frontend/src/api.js b/frontend/src/api.js index 07604c3..37dbfc0 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -145,6 +145,23 @@ export const deleteHandoffRule = (ruleId, token) => request(`/agent/handoff/${encodeURIComponent(ruleId)}`, { method: "DELETE", token }); export const testHandoffRule = (payload, token) => request("/agent/handoff/test", { method: "POST", payload, token }); +export const reorderHandoffRules = (ruleIds, token) => + request("/agent/handoff/reorder", { + method: "PUT", + payload: { rule_ids: ruleIds }, + token, + }); +export const listHandoffConversations = (mailbox, token) => + request(`/agent/handoff/conversations?mailbox=${encodeURIComponent(mailbox)}`, { token }); +export const getHandoffConversation = (handoffId, token) => + request(`/agent/handoff/conversations/${encodeURIComponent(handoffId)}`, { token }); +export const getHandoffConversationCounts = (token) => + request("/agent/handoff/conversations/counts", { token }); +export const retryHandoffDelivery = (jobId, token) => + request(`/agent/handoff/deliveries/${encodeURIComponent(jobId)}/retry`, { + method: "POST", + token, + }); export const listStickyNotes = (token) => request("/sticky-notes", { token }); diff --git a/frontend/src/components/DashboardPage.jsx b/frontend/src/components/DashboardPage.jsx index cd63445..8d49a1d 100644 --- a/frontend/src/components/DashboardPage.jsx +++ b/frontend/src/components/DashboardPage.jsx @@ -115,6 +115,7 @@ export default function DashboardPage() { const [mailRefreshKey, setMailRefreshKey] = useState(0); const [lens, setLens] = useState("all"); const [agentNav, setAgentNav] = useState(null); + const [openThreadId, setOpenThreadId] = useState(null); const [labelsExpanded, setLabelsExpanded] = useState(false); const [companyActive, setCompanyActive] = useState(null); const [mustChangePassword, setMustChangePassword] = useState(false); @@ -211,6 +212,14 @@ export default function DashboardPage() { triggerOpenCompose(context); } + function openExternalThread(threadId) { + setCompanyActive(null); + setActive("Inbox"); + setLens("all"); + setAgentNav(null); + setOpenThreadId(threadId); + } + function emailSent() { setActive("Outbox"); setMailRefreshKey((key) => key + 1); @@ -413,7 +422,7 @@ export default function DashboardPage() { )} {!companyActive && active === "Inbox" && agentNav === "handoff" && ( - + )} {!companyActive && active === "Inbox" && agentNav !== "handoff" && ( selectMailItem("Settings")} onCompose={openCompose} diff --git a/frontend/src/components/HandoffConversation.jsx b/frontend/src/components/HandoffConversation.jsx index 329a586..003acf3 100644 --- a/frontend/src/components/HandoffConversation.jsx +++ b/frontend/src/components/HandoffConversation.jsx @@ -1,6 +1,6 @@ import { ArrowLeft } from "lucide-react"; import { useCallback, useEffect, useState } from "react"; -import { getHandoffConversation } from "../api"; +import { getHandoffConversation, retryHandoffDelivery } from "../api"; function HandoffMessageBody({ message }) { const [showQuoted, setShowQuoted] = useState(false); @@ -33,11 +33,12 @@ function formatTime(dateValue) { }); } -export default function HandoffConversation({ handoffId, onBack }) { +export default function HandoffConversation({ handoffId, onBack, onOpenExternal }) { const token = sessionStorage.getItem("agmail_access_token"); const [conversation, setConversation] = useState(null); const [loading, setLoading] = useState(true); const [error, setError] = useState(""); + const [retryingId, setRetryingId] = useState(null); const load = useCallback(async () => { setLoading(true); @@ -56,6 +57,19 @@ export default function HandoffConversation({ handoffId, onBack }) { load(); }, [load]); + async function retryDelivery(jobId) { + setRetryingId(jobId); + setError(""); + try { + await retryHandoffDelivery(jobId, token); + await load(); + } catch (err) { + setError(err.message); + } finally { + setRetryingId(null); + } + } + if (loading) { return

Loading conversation…

; } @@ -71,11 +85,20 @@ export default function HandoffConversation({ handoffId, onBack }) {
-

{conversation.external_subject || "(No subject)"}

+

{conversation.rule_name} · {conversation.external_subject || "(No subject)"}

Linked conversation: {conversation.linked_external.sender} “{conversation.linked_external.subject}” + {onOpenExternal && ( + + )}
@@ -86,23 +109,38 @@ export default function HandoffConversation({ handoffId, onBack }) { {message.role === "human" ? "HUMAN OWNER" - : message.role === "customer" - ? "CUSTOMER" - : "AGENT"} + : "GENTIC → HUMAN"} {message.role === "human" && message.relay_status === "SENT" && (

- ✓ Sent to customer + Sent through Gentic

)} - {message.role === "human" && message.relay_status === "FAILED" && ( + {message.role === "human" + && ["FAILED", "BLOCKED"].includes(message.relay_status) && (

- ⚠ Could not send to customer + Could not safely relay this reply

)} + {message.delivery_status === "AMBIGUOUS" && ( +

+ Delivery acknowledgement was lost — not retried to prevent duplicates +

+ )} + {["FAILED", "AMBIGUOUS"].includes(message.delivery_status) + && message.delivery_job_id && ( + + )} ))} diff --git a/frontend/src/components/HandoffInbox.jsx b/frontend/src/components/HandoffInbox.jsx index 688b8b8..3255019 100644 --- a/frontend/src/components/HandoffInbox.jsx +++ b/frontend/src/components/HandoffInbox.jsx @@ -17,7 +17,7 @@ function formatRelativeTime(dateValue) { return date.toLocaleDateString(); } -export default function HandoffInbox({ onSelect }) { +export default function HandoffInbox({ mailbox = "handoff_inbox", onSelect }) { const token = sessionStorage.getItem("agmail_access_token"); const [items, setItems] = useState([]); const [loading, setLoading] = useState(true); @@ -27,14 +27,14 @@ export default function HandoffInbox({ onSelect }) { setLoading(true); setError(""); try { - const data = await listHandoffConversations(token); + const data = await listHandoffConversations(mailbox, token); setItems(data); } catch (err) { setError(err.message); } finally { setLoading(false); } - }, [token]); + }, [mailbox, token]); useEffect(() => { load(); @@ -42,7 +42,11 @@ export default function HandoffInbox({ onSelect }) { return (
-

Agent ↔ Human Owner conversations

+

+ {mailbox === "handoff_inbox" + ? "Replies received from Human Owners" + : "Messages mirrored from Gentic to Human Owners"} +

{error &&

{error}

} {loading ? (

Loading handoffs…

@@ -51,7 +55,7 @@ export default function HandoffInbox({ onSelect }) { -

No handoff conversations yet

+

{mailbox === "handoff_inbox" ? "No Human replies yet" : "No mirrored messages yet"}

When the Agent hands off a customer conversation, it appears here as a private Agent ↔ Human thread — separate from your normal Inbox. @@ -67,8 +71,10 @@ export default function HandoffInbox({ onSelect }) { onClick={() => onSelect(item.id)} >

- {item.external_subject || "(No subject)"} - Human Owner + {item.rule_name} · {item.external_subject || "(No subject)"} + + {item.external_sender} ↔ {item.human_email} + {item.snippet &&

“{item.snippet}”

}
diff --git a/frontend/src/components/HandoffSettings.jsx b/frontend/src/components/HandoffSettings.jsx index 9d43fb7..c263967 100644 --- a/frontend/src/components/HandoffSettings.jsx +++ b/frontend/src/components/HandoffSettings.jsx @@ -1,9 +1,10 @@ -import { Info, MoreHorizontal, Plus, UserRound } from "lucide-react"; +import { ArrowDown, ArrowUp, MoreHorizontal, Plus, UserRound } from "lucide-react"; import { useCallback, useEffect, useState } from "react"; import { createHandoffRule, deleteHandoffRule, listHandoffRules, + reorderHandoffRules, testHandoffRule, updateHandoffRule, } from "../api"; @@ -93,6 +94,22 @@ export default function HandoffSettings() { } } + async function moveRule(ruleId, offset) { + const index = rules.findIndex((rule) => rule.id === ruleId); + const target = index + offset; + if (index < 0 || target < 0 || target >= rules.length) return; + const reordered = [...rules]; + [reordered[index], reordered[target]] = [reordered[target], reordered[index]]; + setRules(reordered); + try { + const updated = await reorderHandoffRules(reordered.map((rule) => rule.id), token); + setRules(updated); + } catch (err) { + setError(err.message); + await loadRules(); + } + } + function openTest(draft) { setTestDraft(draft); setTestSubject(""); @@ -128,14 +145,6 @@ export default function HandoffSettings() { return (
-
-
-
+ diff --git a/frontend/src/components/HumanHandoff.jsx b/frontend/src/components/HumanHandoff.jsx index 42a3a85..34cd401 100644 --- a/frontend/src/components/HumanHandoff.jsx +++ b/frontend/src/components/HumanHandoff.jsx @@ -1,15 +1,62 @@ +import { useState } from "react"; +import HandoffConversation from "./HandoffConversation"; +import HandoffInbox from "./HandoffInbox"; import HandoffSettings from "./HandoffSettings"; -export default function HumanHandoff() { +export default function HumanHandoff({ onOpenExternal }) { + const [tab, setTab] = useState("inbox"); + const [selectedHandoffId, setSelectedHandoffId] = useState(null); + + function selectTab(value) { + setTab(value); + setSelectedHandoffId(null); + } + return (

Human Handoff

-

Configure when your Agent should involve a Human Owner.

+

Private mirrored conversations between your Agent and Human Owners.

- + + {tab === "rules" ? ( + + ) : selectedHandoffId ? ( + setSelectedHandoffId(null)} + onOpenExternal={onOpenExternal} + /> + ) : ( + + )}
); } diff --git a/frontend/src/components/HumanHandoffDrawer.jsx b/frontend/src/components/HumanHandoffDrawer.jsx index 77d3564..9551c91 100644 --- a/frontend/src/components/HumanHandoffDrawer.jsx +++ b/frontend/src/components/HumanHandoffDrawer.jsx @@ -28,23 +28,25 @@ export default function HumanHandoffDrawer({ onSave, onTest, }) { + const [ruleName, setRuleName] = useState(""); + const [priority, setPriority] = useState(0); const [ownerEmail, setOwnerEmail] = useState(""); const [labels, setLabels] = useState([]); const [emailTypes, setEmailTypes] = useState([]); const [instructions, setInstructions] = useState(""); const [allowReplies, setAllowReplies] = useState(true); - const [replyAs, setReplyAs] = useState("agent"); const [active, setActive] = useState(true); const [labelQuery, setLabelQuery] = useState(""); useEffect(() => { if (!open) return; + setRuleName(rule?.name || ""); + setPriority(rule?.priority ?? 0); setOwnerEmail(rule?.owner_email || ""); setLabels(rule?.labels || []); setEmailTypes(rule?.email_types || []); setInstructions(rule?.instructions || ""); setAllowReplies(rule?.allow_replies ?? true); - setReplyAs(rule?.reply_as || "agent"); setActive(rule?.active ?? true); setLabelQuery(""); }, [open, rule]); @@ -124,6 +126,30 @@ export default function HumanHandoffDrawer({
+
+

Rule

+ + setRuleName(event.target.value)} + /> + + setPriority(Number(event.target.value) || 0)} + /> +

When multiple rules match, the lowest priority number wins.

+
+

Human owner

@@ -244,31 +270,9 @@ export default function HumanHandoffDrawer({ onChange={setAllowReplies} />

- Saved for future use. Human reply handling is not active yet — all email uses the normal Gentic inbox and outbox. + A reply from this Human Owner is stored in Handoff Inbox and relayed through the + original Gentic address. Their private address is never exposed.

- {allowReplies && ( -
- Replies will be sent as - - -
- )}
@@ -294,14 +298,16 @@ export default function HumanHandoffDrawer({