diff --git a/backend/app/routers/agent_handoff.py b/backend/app/routers/agent_handoff.py index 5bcba20..ac669b5 100644 --- a/backend/app/routers/agent_handoff.py +++ b/backend/app/routers/agent_handoff.py @@ -8,18 +8,12 @@ from app.database import get_mongo_database from app.models import Account from app.routers.auth import get_current_account from app.schemas import ( - HandoffConversationRead, - HandoffConversationSummary, - HandoffLinkedExternal, - HandoffMessageRead, HandoffRuleCreate, HandoffRuleRead, HandoffRuleUpdate, HandoffTestRequest, HandoffTestResult, ) -from app.services.handoff_mirror_policy import is_relay_acknowledgment_message -from app.services.human_handoff_service import MESSAGE_SCOPE_HANDOFF from app.services.handoff_evaluator import EMAIL_TYPE_META, HANDOFF_LABELS, evaluate_handoff router = APIRouter(prefix="/agent/handoff", tags=["agent-handoff"]) @@ -34,20 +28,6 @@ def _owner_name_from_email(email: str) -> str: return re.sub(r"[._-]+", " ", local).strip().title() or email -def _status_label(status: str, current_owner: str) -> str: - if status == "WAITING_FOR_HUMAN": - return "Waiting for Human" - if status == "HUMAN_REPLIED": - return "Needs Agent Reply" - if status == "AGENT_REPLIED": - return "Replied" - if status == "CLOSED": - return "Closed" - if current_owner == "HUMAN": - return "Waiting for Human" - return "Active" - - def _serialize(document: dict) -> HandoffRuleRead: return HandoffRuleRead( id=str(document["_id"]), @@ -171,150 +151,15 @@ async def test_handoff( return HandoffTestResult( owner_email=owner_email, **evaluation, - result_path="Agent → Human Owner" if evaluation["would_handoff"] else "Agent handles email", + result_path=( + "Would match handoff rule (configuration only)" + if evaluation["would_handoff"] + else "Would not match handoff rule" + ), reason=evaluation["instruction_reason"], ) -@router.get("/conversations", response_model=list[HandoffConversationSummary]) -async def list_handoff_conversations( - database=Depends(get_mongo_database), - account: Account = Depends(get_current_account), -): - account_id = str(account.id) - cursor = database.human_handoffs.find({"account_id": account_id}).sort("updated_at", -1) - items: list[HandoffConversationSummary] = [] - async for handoff in cursor: - handoff_thread_id = handoff["handoff_thread_id"] - count = await database.mailbox_entries.count_documents( - { - "account_id": account_id, - "thread_id": handoff_thread_id, - "message_scope": MESSAGE_SCOPE_HANDOFF, - } - ) - latest = await database.mailbox_entries.find_one( - { - "account_id": account_id, - "thread_id": handoff_thread_id, - "message_scope": MESSAGE_SCOPE_HANDOFF, - }, - sort=[("event_at", -1), ("_id", -1)], - ) - snippet = latest.get("snippet", "") if latest else "" - items.append( - HandoffConversationSummary( - id=str(handoff["_id"]), - 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=_status_label( - handoff.get("status", "ACTIVE"), - handoff.get("current_owner", "AGENT"), - ), - current_owner=handoff.get("current_owner", "AGENT"), - snippet=snippet, - message_count=count, - updated_at=handoff.get("updated_at") or handoff.get("created_at"), - ) - ) - return items - - -@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.") - - pipeline = [ - { - "$match": { - "account_id": account_id, - "thread_id": handoff["handoff_thread_id"], - "message_scope": MESSAGE_SCOPE_HANDOFF, - } - }, - {"$sort": {"event_at": 1, "_id": 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] - human_email = handoff.get("human_email", "").lower() - messages: list[HandoffMessageRead] = [] - for row in rows: - message = row["message"] - if is_relay_acknowledgment_message(message): - continue - sender_address = (message.get("sender") or {}).get("address", "").lower() - if message.get("mirrored_from_external_message_id"): - role = "customer" - elif sender_address == human_email: - role = "human" - else: - role = "agent" - event_at = row.get("event_at") or message.get("received_at") or message.get("sent_at") or message["created_at"] - raw_body = message.get("raw_body") - new_reply_content = message.get("new_reply_content") - display_body = new_reply_content or message.get("body_text", "") - messages.append( - HandoffMessageRead( - id=str(message["_id"]), - role=role, - subject=message.get("subject", ""), - snippet=message.get("snippet", "") or display_body[:240], - body_text=display_body, - new_reply_content=new_reply_content, - raw_body=raw_body, - has_quoted_text=bool( - raw_body - and new_reply_content - and raw_body.strip() != new_reply_content.strip() - ), - relay_status=message.get("relay_status") if role == "human" else None, - event_at=event_at, - ) - ) - - return HandoffConversationRead( - id=str(handoff["_id"]), - external_subject=handoff.get("external_subject", ""), - external_sender=handoff.get("external_sender", ""), - human_email=handoff.get("human_email", ""), - handoff_reason=handoff.get("handoff_reason", ""), - status=handoff.get("status", "ACTIVE"), - status_label=_status_label( - handoff.get("status", "ACTIVE"), - handoff.get("current_owner", "AGENT"), - ), - current_owner=handoff.get("current_owner", "AGENT"), - linked_external=HandoffLinkedExternal( - thread_id=str(handoff["external_thread_id"]), - subject=handoff.get("external_subject", ""), - sender=handoff.get("external_sender", ""), - ), - messages=messages, - ) - - @router.get("/meta") async def handoff_meta(): return { diff --git a/backend/app/services/human_handoff_service.py b/backend/app/services/human_handoff_service.py index 1595f59..1b5d8db 100644 --- a/backend/app/services/human_handoff_service.py +++ b/backend/app/services/human_handoff_service.py @@ -1,867 +1,8 @@ -import logging -import re -from datetime import datetime, timezone +"""Human Handoff runtime is disabled — configuration-only via agent_handoff router. -from bson import ObjectId - -from app.services.external_reply_service import ( - ExternalReplyThreadingError, - build_external_reply_plan, - internet_message_id_of, - log_original_external_message, - resolve_external_reply_parent, - resolve_internet_message_id, - send_external_reply, -) -from app.services.handoff_email_templates import ( - external_reply_subject, - handoff_email_subject, -) -from app.services.handoff_mirror_policy import ( - GENERATED_BY_HANDOFF_BRIDGE, - GENERATED_BY_HUMAN, - GENERATED_BY_SYSTEM, - MESSAGE_ORIGIN_HANDOFF_MIRROR, - MESSAGE_ORIGIN_HUMAN_OWNER, - RELAY_STATUS_FAILED, - RELAY_STATUS_SENT, - should_mirror_to_human, -) -from app.services.handoff_mirror import load_mirror_payload, load_mirror_update_payload -from app.services.handoff_evaluator import evaluate_handoff -from app.services.mail_sender import deliver_email -from app.services.message_ids import ( - canonical_message_id, - folded_message_id, - message_id_fields, - parse_message_id_list, -) -from app.services.reply_extractor import extract_new_reply_content, validate_external_reply_body -from app.services.thread_service import ( - attach_message_to_thread, - message_event_at, - threading_fields_from_parent, -) - -logger = logging.getLogger(__name__) +This module retains scope constants used by normal mail filtering. All email +sending, receiving, mirroring, forwarding, and relay behavior has been removed. +""" MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" MESSAGE_SCOPE_HANDOFF = "HANDOFF" -THREAD_TYPE_EXTERNAL = "EXTERNAL" -THREAD_TYPE_HUMAN_HANDOFF = "HUMAN_HANDOFF" - -HANDOFF_STATUS_ACTIVE = "ACTIVE" -HANDOFF_STATUS_WAITING = "WAITING_FOR_HUMAN" -HANDOFF_STATUS_HUMAN_REPLIED = "HUMAN_REPLIED" -HANDOFF_STATUS_AGENT_REPLIED = "AGENT_REPLIED" - -OWNER_AGENT = "AGENT" -OWNER_HUMAN = "HUMAN" - -LINK_EXTERNAL_TRIGGERED = "EXTERNAL_TRIGGERED_HANDOFF" -LINK_EXTERNAL_RELAYED = "EXTERNAL_RELAYED_TO_HUMAN" -LINK_HUMAN_TRIGGERED_EXTERNAL = "HUMAN_TRIGGERED_EXTERNAL_REPLY" -LINK_EXTERNAL_UPDATE_RELAYED = "EXTERNAL_UPDATE_RELAYED_TO_HUMAN" - - -def _now() -> datetime: - return datetime.now(timezone.utc) - - -def _snippet(text: str, limit: int = 240) -> str: - return re.sub(r"\s+", " ", (text or "")).strip()[:limit] - - -def _handoff_reply_subject(external_subject: str) -> str: - return f"Re: {handoff_email_subject(external_subject)}" - - -async def create_empty_handoff_thread( - database, - *, - account_id: str, - subject: str, -) -> ObjectId: - now = _now() - document = { - "account_id": account_id, - "thread_type": THREAD_TYPE_HUMAN_HANDOFF, - "normalized_subject": subject, - "created_at": now, - "updated_at": now, - "last_message_at": now, - "latest_message_id": None, - "message_count": 0, - } - result = await database.email_threads.insert_one(document) - return result.inserted_id - - -async def _record_message_link( - database, - *, - handoff_id: ObjectId, - external_message_id: ObjectId | None, - handoff_message_id: ObjectId, - relationship: str, -) -> None: - await database.handoff_message_links.insert_one( - { - "handoff_id": handoff_id, - "external_message_id": external_message_id, - "handoff_message_id": handoff_message_id, - "relationship": relationship, - "created_at": _now(), - } - ) - - -async def _latest_external_customer_message( - database, - account_id: str, - external_thread_id: ObjectId, - customer_address: str, -) -> dict | None: - customer = customer_address.lower() - pipeline = [ - { - "$match": { - "account_id": account_id, - "thread_id": external_thread_id, - "message_scope": MESSAGE_SCOPE_EXTERNAL, - "deletion_state": {"$in": [None, ""]}, - } - }, - {"$sort": {"event_at": -1, "_id": -1}}, - { - "$lookup": { - "from": "messages", - "localField": "message_id", - "foreignField": "_id", - "as": "message", - } - }, - {"$unwind": "$message"}, - { - "$match": { - "$or": [ - {"message.direction": "inbound"}, - {"message.sender.address": customer}, - ] - } - }, - {"$limit": 1}, - ] - cursor = await database.mailbox_entries.aggregate(pipeline) - rows = [row async for row in cursor] - return rows[0]["message"] if rows else None - - -async def _latest_external_message(database, account_id: str, external_thread_id: ObjectId) -> dict | None: - pipeline = [ - { - "$match": { - "account_id": account_id, - "thread_id": external_thread_id, - "message_scope": MESSAGE_SCOPE_EXTERNAL, - "deletion_state": {"$in": [None, ""]}, - } - }, - {"$sort": {"event_at": -1, "_id": -1}}, - {"$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 _latest_handoff_message(database, account_id: str, handoff_thread_id: ObjectId) -> dict | None: - pipeline = [ - { - "$match": { - "account_id": account_id, - "thread_id": handoff_thread_id, - "message_scope": MESSAGE_SCOPE_HANDOFF, - } - }, - {"$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_active_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": {"$in": [HANDOFF_STATUS_ACTIVE, HANDOFF_STATUS_WAITING, HANDOFF_STATUS_HUMAN_REPLIED, HANDOFF_STATUS_AGENT_REPLIED]}, - } - ) - - -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.lower() - rules = await database.human_handoff_rules.find( - {"account_id": account_id, "owner_email": sender, "active": True} - ).to_list(length=10) - if not rules: - return None - - search_ids = [*([in_reply_to] if in_reply_to else []), *references] - if not search_ids: - return None - - for internet_id in reversed(parse_message_id_list(" ".join(search_ids))): - folded = folded_message_id(internet_id) - message = await database.messages.find_one( - { - "$or": [ - {"message_id_folded": folded}, - {"message_id": internet_id}, - ], - "message_scope": MESSAGE_SCOPE_HANDOFF, - } - ) - if message is None: - continue - entry = await database.mailbox_entries.find_one( - { - "account_id": account_id, - "message_id": message["_id"], - "message_scope": MESSAGE_SCOPE_HANDOFF, - } - ) - 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, - "status": {"$ne": "CLOSED"}, - } - ) - if handoff: - return handoff - return None - - -async def _handoff_mirror_exists(database, external_message_id: ObjectId) -> bool: - existing = await database.messages.find_one( - { - "mirrored_from_external_message_id": external_message_id, - "message_scope": MESSAGE_SCOPE_HANDOFF, - }, - {"_id": 1}, - ) - return existing is not None - - -async def _bridged_external_exists(database, handoff_message_id: ObjectId) -> dict | None: - return await database.messages.find_one( - { - "generated_from_handoff_message_id": handoff_message_id, - "message_scope": MESSAGE_SCOPE_EXTERNAL, - } - ) - - -async def store_handoff_message( - database, - *, - account_id: str, - agent_email: str, - handoff_thread_id: ObjectId, - direction: str, - sender: dict, - recipients: list[dict], - subject: str, - body_text: str, - internet_message_id: str | None = None, - in_reply_to: str | None = None, - references: list[str] | None = None, - handoff_id: ObjectId | None = None, - link_relationship: str | None = None, - external_message_id: ObjectId | None = None, - event_at: datetime | None = None, - raw_body: str | None = None, - new_reply_content: str | None = None, - body_html: str | None = None, - message_origin: str | None = None, - generated_by: str | None = None, - mirrored_from_external_message_id: ObjectId | None = None, - relay_status: str | None = None, - generated_external_message_id: ObjectId | None = None, -) -> dict: - now = event_at or _now() - display_body = new_reply_content if new_reply_content is not None else body_text - document = { - "direction": direction, - "message_scope": MESSAGE_SCOPE_HANDOFF, - "account_id": account_id, - "sender": sender, - "to": recipients, - "subject": subject, - "snippet": _snippet(display_body), - "body_text": display_body, - "raw_body": raw_body, - "new_reply_content": new_reply_content, - "body_html": body_html, - "attachments": [], - "created_at": now, - "sent_at": now if direction == "outbound" else None, - "received_at": now if direction == "inbound" else None, - "status": "received" if direction == "inbound" else "accepted", - **message_id_fields(internet_message_id), - "in_reply_to": canonical_message_id(in_reply_to), - "references": references or [], - } - if message_origin: - document["message_origin"] = message_origin - if generated_by: - document["generated_by"] = generated_by - if mirrored_from_external_message_id: - document["mirrored_from_external_message_id"] = mirrored_from_external_message_id - if relay_status: - document["relay_status"] = relay_status - if generated_external_message_id: - document["generated_external_message_id"] = generated_external_message_id - result = await database.messages.insert_one(document) - document["_id"] = result.inserted_id - await attach_message_to_thread( - database, - account_id=account_id, - thread_id=handoff_thread_id, - message_id=result.inserted_id, - subject=subject, - event_at=now, - ) - await database.mailbox_entries.insert_one( - { - "account_id": account_id, - "message_id": result.inserted_id, - "thread_id": handoff_thread_id, - "mailbox": "handoff", - "message_scope": MESSAGE_SCOPE_HANDOFF, - "direction": direction, - "sender": sender, - "recipients": recipients, - "subject": subject, - "snippet": document["snippet"], - "attachments": [], - "status": document["status"], - "is_read": direction == "outbound", - "is_starred": False, - "event_at": now, - "created_at": now, - "trashed_at": None, - "deletion_state": None, - } - ) - if handoff_id and link_relationship: - await _record_message_link( - database, - handoff_id=handoff_id, - external_message_id=external_message_id, - handoff_message_id=result.inserted_id, - relationship=link_relationship, - ) - return document - - -async def send_handoff_to_human( - database, - *, - account_id: str, - agent_email: str, - human_email: str, - external_message: dict, - external_thread_id: ObjectId, - handoff_reason: str, - domain, - existing_handoff: dict | None = None, -) -> dict: - if await _handoff_mirror_exists(database, external_message["_id"]): - logger.info( - "handoff_mirror_skipped_duplicate external_message=%s", - external_message["_id"], - ) - if existing_handoff: - return existing_handoff - existing = await database.human_handoffs.find_one( - { - "account_id": account_id, - "external_thread_id": external_thread_id, - } - ) - if existing: - return existing - raise RuntimeError("Handoff mirror duplicate without handoff record") - - external_subject = external_message.get("subject", "") - stable_subject = handoff_email_subject(external_subject) - mirror_source = {**external_message, "_agent_email": agent_email} - - if existing_handoff: - handoff_id = existing_handoff["_id"] - handoff_thread_id = existing_handoff["handoff_thread_id"] - parent = await _latest_handoff_message(database, account_id, handoff_thread_id) - threading = threading_fields_from_parent(parent) if parent else {} - subject = _handoff_reply_subject(external_subject) - mirror = await load_mirror_update_payload(mirror_source, agent_email=agent_email) - relationship = LINK_EXTERNAL_UPDATE_RELAYED - else: - handoff_thread_id = await create_empty_handoff_thread( - database, - account_id=account_id, - subject=stable_subject, - ) - threading = {} - subject = stable_subject - mirror = await load_mirror_payload(mirror_source) - external_sender_text = (external_message.get("sender") or {}).get("address") or "unknown sender" - handoff_doc = { - "account_id": account_id, - "agent_email": agent_email.lower(), - "human_email": human_email.lower(), - "external_thread_id": external_thread_id, - "handoff_thread_id": handoff_thread_id, - "external_subject": external_subject, - "external_sender": external_sender_text, - "handoff_reason": handoff_reason, - "status": HANDOFF_STATUS_WAITING, - "current_owner": OWNER_HUMAN, - "created_at": _now(), - "updated_at": _now(), - "last_external_message_at": message_event_at(external_message), - "last_human_message_at": None, - "trigger_external_message_id": external_message["_id"], - "trigger_internet_message_id": internet_message_id_of(external_message), - } - insert = await database.human_handoffs.insert_one(handoff_doc) - handoff_id = insert.inserted_id - relationship = LINK_EXTERNAL_TRIGGERED - - internet_message_id = await deliver_email( - sender=agent_email, - recipients=[human_email], - cc=[], - bcc=[], - subject=subject, - body_text=mirror.body_text, - body_html=mirror.body_html, - domain=domain, - in_reply_to=threading.get("in_reply_to"), - references=threading.get("references"), - inline_parts=mirror.inline_parts, - file_attachments=mirror.file_attachments, - ) - - handoff_message = await store_handoff_message( - database, - account_id=account_id, - agent_email=agent_email, - handoff_thread_id=handoff_thread_id, - direction="outbound", - sender={"name": "Gentic Agent", "address": agent_email}, - recipients=[{"name": human_email.split("@")[0], "address": human_email}], - subject=subject, - body_text=mirror.body_text, - body_html=mirror.body_html, - internet_message_id=internet_message_id, - in_reply_to=threading.get("in_reply_to"), - references=threading.get("references"), - handoff_id=handoff_id, - link_relationship=relationship, - external_message_id=external_message["_id"], - message_origin=MESSAGE_ORIGIN_HANDOFF_MIRROR, - generated_by=GENERATED_BY_SYSTEM, - mirrored_from_external_message_id=external_message["_id"], - ) - - await database.human_handoffs.update_one( - {"_id": handoff_id}, - { - "$set": { - "status": HANDOFF_STATUS_WAITING, - "current_owner": OWNER_HUMAN, - "updated_at": _now(), - "last_external_message_at": message_event_at(external_message), - "latest_handoff_message_id": handoff_message["_id"], - } - }, - ) - logger.info( - "handoff_sent handoff_id=%s external_thread=%s handoff_thread=%s", - handoff_id, - external_thread_id, - handoff_thread_id, - ) - return await database.human_handoffs.find_one({"_id": handoff_id}) - - -async def _handoff_rule_for_human(database, account_id: str, human_email: str) -> dict | None: - return await database.human_handoff_rules.find_one( - { - "account_id": account_id, - "owner_email": human_email.lower(), - "active": True, - } - ) - - -async def process_human_reply( - database, - *, - account_id: str, - agent_email: str, - handoff: dict, - inbound_parsed: dict, - inbound_body: str, - domain, -) -> None: - handoff_thread_id = handoff["handoff_thread_id"] - external_thread_id = handoff["external_thread_id"] - human_email = handoff["human_email"] - rule = await _handoff_rule_for_human(database, account_id, human_email) - allow_replies = rule.get("allow_replies", True) if rule else True - - extracted = extract_new_reply_content( - body_text=inbound_body, - body_html=inbound_parsed.get("body_html"), - ) - new_reply_content = extracted["new_reply_content"].strip() - raw_body = extracted["raw_body"] - - human_message = await store_handoff_message( - database, - account_id=account_id, - agent_email=agent_email, - handoff_thread_id=handoff_thread_id, - direction="inbound", - sender={"name": human_email.split("@")[0], "address": human_email}, - recipients=[{"name": "Gentic Agent", "address": agent_email}], - subject=inbound_parsed.get("subject", ""), - body_text=raw_body, - raw_body=raw_body, - new_reply_content=new_reply_content, - body_html=inbound_parsed.get("body_html"), - internet_message_id=inbound_parsed.get("message_id"), - in_reply_to=inbound_parsed.get("in_reply_to"), - references=inbound_parsed.get("references", []), - handoff_id=handoff["_id"], - message_origin=MESSAGE_ORIGIN_HUMAN_OWNER, - generated_by=GENERATED_BY_HUMAN, - ) - - now = _now() - if not allow_replies: - await database.human_handoffs.update_one( - {"_id": handoff["_id"]}, - { - "$set": { - "status": HANDOFF_STATUS_HUMAN_REPLIED, - "current_owner": OWNER_AGENT, - "updated_at": now, - "last_human_message_at": now, - } - }, - ) - return - - if not new_reply_content: - logger.warning( - "handoff_reply_empty_extract handoff_id=%s account=%s", - handoff["_id"], - account_id, - ) - await database.human_handoffs.update_one( - {"_id": handoff["_id"]}, - { - "$set": { - "status": HANDOFF_STATUS_HUMAN_REPLIED, - "current_owner": OWNER_AGENT, - "updated_at": now, - "last_human_message_at": now, - } - }, - ) - return - - if not validate_external_reply_body(new_reply_content): - logger.error( - "handoff_reply_leakage_guard handoff_id=%s account=%s", - handoff["_id"], - account_id, - ) - return - - external_recipient = handoff.get("external_sender") or "" - external_parent = await resolve_external_reply_parent( - database, - account_id=account_id, - external_thread_id=external_thread_id, - customer_address=external_recipient, - handoff=handoff, - ) - if external_parent is None: - logger.warning( - "handoff_reply_missing_external_parent handoff_id=%s external_thread=%s " - "trigger_internet_message_id=%s", - handoff["_id"], - external_thread_id, - handoff.get("trigger_internet_message_id"), - ) - return - - parent_sender = (external_parent.get("sender") or {}).get("address") - log_original_external_message( - internal_id=external_parent["_id"], - thread_id=external_thread_id, - internet_message_id=internet_message_id_of(external_parent), - sender=parent_sender, - recipient=agent_email, - subject=external_parent.get("subject"), - in_reply_to=external_parent.get("in_reply_to"), - references=external_parent.get("references"), - raw_message_available=bool(external_parent.get("raw_mime")), - ) - - external_recipient = external_recipient or parent_sender - if not external_recipient: - return - - parent_internet_id = await resolve_internet_message_id(database, external_parent) - if not parent_internet_id: - logger.error( - "handoff_reply_missing_parent_message_id handoff_id=%s parent_internal_id=%s", - handoff["_id"], - external_parent["_id"], - ) - return - - logger.info( - "handoff_external_reply_threading handoff_id=%s parent_internal_id=%s " - "parent_internet_message_id=%s human_in_reply_to=%s human_references=%s", - handoff["_id"], - external_parent["_id"], - parent_internet_id, - human_message.get("in_reply_to"), - human_message.get("references"), - ) - - existing_bridge = await _bridged_external_exists(database, human_message["_id"]) - if existing_bridge: - logger.info( - "handoff_bridge_already_processed handoff_message=%s external=%s", - human_message["_id"], - existing_bridge["_id"], - ) - return - - customer_body = new_reply_content - reply_plan = build_external_reply_plan( - parent_external_message=external_parent, - body_text=customer_body, - subject_override=external_reply_subject( - handoff.get("external_subject") or external_parent.get("subject", "") - ), - recipients=[external_recipient], - parent_internet_id_override=parent_internet_id, - ) - logger.info( - "handoff_external_reply_headers handoff_id=%s b_in_reply_to=%s b_references=%s", - handoff["_id"], - reply_plan.threading.in_reply_to, - reply_plan.threading.references, - ) - try: - external_doc = await send_external_reply( - database, - account_id=account_id, - sender=agent_email, - sender_name="Gentic Agent", - external_thread_id=external_thread_id, - plan=reply_plan, - domain=domain, - resolution_method="human_handoff_bridge", - generated_by=GENERATED_BY_HANDOFF_BRIDGE, - generated_from_handoff_message_id=human_message["_id"], - ) - except ExternalReplyThreadingError: - await database.messages.update_one( - {"_id": human_message["_id"]}, - {"$set": {"relay_status": RELAY_STATUS_FAILED, "relay_failed_at": now}}, - ) - logger.error( - "handoff_bridge_threading_rejected handoff_id=%s handoff_message=%s " - "parent_internet_message_id=%s", - handoff["_id"], - human_message["_id"], - parent_internet_id, - ) - return - except Exception: - await database.messages.update_one( - {"_id": human_message["_id"]}, - {"$set": {"relay_status": RELAY_STATUS_FAILED, "relay_failed_at": now}}, - ) - logger.exception( - "handoff_bridge_send_failed handoff_id=%s handoff_message=%s", - handoff["_id"], - human_message["_id"], - ) - raise - - await database.messages.update_one( - {"_id": human_message["_id"]}, - { - "$set": { - "relay_status": RELAY_STATUS_SENT, - "generated_external_message_id": external_doc["_id"], - "relay_delivered_at": now, - } - }, - ) - await _record_message_link( - database, - handoff_id=handoff["_id"], - external_message_id=external_doc["_id"], - handoff_message_id=human_message["_id"], - relationship=LINK_HUMAN_TRIGGERED_EXTERNAL, - ) - - await database.human_handoffs.update_one( - {"_id": handoff["_id"]}, - { - "$set": { - "status": HANDOFF_STATUS_WAITING, - "current_owner": OWNER_HUMAN, - "updated_at": now, - "last_human_message_at": now, - } - }, - ) - logger.info( - "handoff_bridge_complete handoff_id=%s handoff_message=%s external=%s", - handoff["_id"], - human_message["_id"], - external_doc["_id"], - ) - - -async def maybe_trigger_handoff( - database, - *, - account_id: str, - agent_email: str, - external_message: dict, - external_thread_id: ObjectId, - domain, -) -> None: - if not should_mirror_to_human(external_message): - logger.debug( - "handoff_mirror_skipped message_id=%s generated_by=%s origin=%s direction=%s", - external_message.get("_id"), - external_message.get("generated_by"), - external_message.get("message_origin"), - external_message.get("direction"), - ) - return - - existing = await find_active_handoff_for_external_thread( - database, account_id, external_thread_id - ) - if existing: - if await _handoff_mirror_exists(database, external_message["_id"]): - return - rule = await _handoff_rule_for_human( - database, account_id, existing.get("human_email", "") - ) - if rule and not rule.get("allow_replies", True): - return - await send_handoff_to_human( - database, - account_id=account_id, - agent_email=agent_email, - human_email=existing["human_email"], - external_message=external_message, - external_thread_id=external_thread_id, - handoff_reason=existing.get("handoff_reason", ""), - domain=domain, - existing_handoff=existing, - ) - return - - rules = await database.human_handoff_rules.find( - {"account_id": account_id, "active": True} - ).to_list(length=20) - if not rules: - return - - body_text = external_message.get("body_text") or external_message.get("snippet") or "" - email_payload = { - "subject": external_message.get("subject", ""), - "snippet": external_message.get("snippet", ""), - "body_text": body_text, - "is_starred": False, - } - - for rule in rules: - evaluation = evaluate_handoff( - email=email_payload, - labels=rule.get("labels", []), - email_types=rule.get("email_types", []), - instructions=rule.get("instructions", ""), - ) - if not evaluation["would_handoff"]: - continue - if not rule.get("allow_replies", True): - continue - - reason = (rule.get("instructions") or "").strip() or "This conversation requires human attention." - await send_handoff_to_human( - database, - account_id=account_id, - agent_email=agent_email, - human_email=rule["owner_email"], - external_message=external_message, - external_thread_id=external_thread_id, - handoff_reason=reason, - domain=domain, - existing_handoff=None, - ) - return diff --git a/backend/app/services/inbound_smtp.py b/backend/app/services/inbound_smtp.py index 1e8e3ce..a68c6ad 100644 --- a/backend/app/services/inbound_smtp.py +++ b/backend/app/services/inbound_smtp.py @@ -9,19 +9,8 @@ from sqlalchemy import func, select from app.config import get_settings from app.database import SessionFactory, mongo_db from app.models import Account -from app.services.domain_provisioning import get_domain_by_name -from app.services.handoff_mirror_policy import ( - GENERATED_BY_CUSTOMER, - MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, -) -from app.services.human_handoff_service import ( - MESSAGE_SCOPE_EXTERNAL, - MESSAGE_SCOPE_HANDOFF, - process_human_reply, - resolve_handoff_from_inbound, - maybe_trigger_handoff, -) from app.services.external_reply_service import log_original_external_message +from app.services.human_handoff_service import MESSAGE_SCOPE_EXTERNAL from app.services.message_ids import canonical_message_id, message_id_fields from app.services.mime_parser import parse_mime from app.services.raw_mime_storage import delivery_digest, get_raw_mime_storage @@ -136,62 +125,6 @@ class InboundHandler: internet_message_id = canonical_message_id(parsed.get("message_id")) id_fields = message_id_fields(internet_message_id) - for recipient in recipients: - account_id = envelope.account_ids[recipient] - agent_email = recipient - - handoff = await resolve_handoff_from_inbound( - mongo_db, - account_id=account_id, - agent_email=agent_email, - sender_address=sender_address, - in_reply_to=parsed.get("in_reply_to"), - references=parsed.get("references", []), - ) - if handoff: - async with SessionFactory() as database_session: - domain = await get_domain_by_name( - database_session, - agent_email.split("@", 1)[1], - ) - if domain is None: - logger.warning("handoff_reply_no_domain account=%s", account_id) - continue - message = { - **parsed, - "body_text": body_text, - "body_html": body_html, - "direction": "inbound", - "message_scope": MESSAGE_SCOPE_HANDOFF, - "delivery_key": delivery_key, - "received_at": now, - "created_at": now, - **id_fields, - } - await process_human_reply( - mongo_db, - account_id=account_id, - agent_email=agent_email, - handoff=handoff, - inbound_parsed=message, - inbound_body=body_text, - domain=domain, - ) - log_smtp_message_received( - message_id=internet_message_id, - in_reply_to=parsed.get("in_reply_to"), - references=parsed.get("references", []), - sender=sender_address, - recipient=recipient, - subject=parsed.get("subject", ""), - resolved_scope=MESSAGE_SCOPE_HANDOFF, - resolved_thread_id=handoff.get("handoff_thread_id"), - resolution_method="handoff_in_reply_to", - stored_message_id=None, - mailbox_id="handoff", - ) - return f"250 2.0.0 Handoff reply accepted as {delivery_key}" - message = { **parsed, "direction": "inbound", @@ -232,7 +165,6 @@ class InboundHandler: participants = participants_from_parsed(parsed) for recipient in recipients: account_id = envelope.account_ids[recipient] - agent_email = recipient resolution = await resolve_thread( mongo_db, account_id=account_id, @@ -263,8 +195,6 @@ class InboundHandler: "parent_message_id": resolution.parent_message_id, "thread_relation": resolution.thread_relation, "thread_resolution_method": resolution.thread_resolution_method, - "message_origin": MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, - "generated_by": GENERATED_BY_CUSTOMER, } }, ) @@ -298,28 +228,6 @@ class InboundHandler: }, upsert=True, ) - async with SessionFactory() as database_session: - domain = await get_domain_by_name( - database_session, - agent_email.split("@", 1)[1], - ) - persisted.update( - { - "message_scope": MESSAGE_SCOPE_EXTERNAL, - "direction": "inbound", - "message_origin": MESSAGE_ORIGIN_EXTERNAL_CUSTOMER, - "generated_by": GENERATED_BY_CUSTOMER, - } - ) - if domain is not None: - await maybe_trigger_handoff( - mongo_db, - account_id=account_id, - agent_email=agent_email, - external_message=persisted, - external_thread_id=thread_id, - domain=domain, - ) log_original_external_message( internal_id=persisted["_id"], thread_id=thread_id, diff --git a/backend/tests/test_human_handoff_service.py b/backend/tests/test_human_handoff_service.py index 99872c3..5119427 100644 --- a/backend/tests/test_human_handoff_service.py +++ b/backend/tests/test_human_handoff_service.py @@ -1,125 +1,13 @@ -import pytest -from bson import ObjectId - -from app.services.human_handoff_service import ( - MESSAGE_SCOPE_EXTERNAL, - MESSAGE_SCOPE_HANDOFF, - resolve_handoff_from_inbound, -) +from app.services.human_handoff_service import MESSAGE_SCOPE_EXTERNAL, MESSAGE_SCOPE_HANDOFF -class FakeCollection: - def __init__(self, documents=None): - self.documents = list(documents or []) - - async def find_one(self, query): - for document in self.documents: - if all(document.get(key) == value for key, value in query.items()): - return document - if "$ne" in query.get("status", {}): - if document.get("status") == query["status"]["$ne"]: - continue - for document in self.documents: - matches = True - for key, value in query.items(): - if key == "status" and isinstance(value, dict) and "$ne" in value: - if document.get("status") == value["$ne"]: - matches = False - break - elif document.get(key) != value: - matches = False - break - if matches: - return document - return None - - def find(self, query): - class Cursor: - def __init__(self, items): - self.items = items - - async def to_list(self, length): - return self.items[:length] - - filtered = [ - document - for document in self.documents - if all(document.get(key) == value for key, value in query.items()) - ] - return Cursor(filtered) - - -class FakeDatabase: - def __init__(self): - self.human_handoff_rules = FakeCollection() - self.messages = FakeCollection() - self.mailbox_entries = FakeCollection() - self.human_handoffs = FakeCollection() - - -@pytest.mark.asyncio -async def test_resolve_handoff_requires_owner_and_thread_context(): - database = FakeDatabase() - handoff_thread_id = ObjectId() - message_id = ObjectId() - database.human_handoff_rules.documents = [ - { - "account_id": "acct-1", - "owner_email": "owner@gmail.com", - "active": True, - } - ] - database.messages.documents = [ - { - "_id": message_id, - "message_id": "", - "message_id_folded": "", - "message_scope": MESSAGE_SCOPE_HANDOFF, - } - ] - database.mailbox_entries.documents = [ - { - "account_id": "acct-1", - "message_id": message_id, - "thread_id": handoff_thread_id, - "message_scope": MESSAGE_SCOPE_HANDOFF, - } - ] - database.human_handoffs.documents = [ - { - "_id": ObjectId(), - "account_id": "acct-1", - "handoff_thread_id": handoff_thread_id, - "human_email": "owner@gmail.com", - "status": "WAITING_FOR_HUMAN", - } - ] - - handoff = await resolve_handoff_from_inbound( - database, - account_id="acct-1", - agent_email="support@gentic.email", - sender_address="owner@gmail.com", - in_reply_to="", - references=[], - ) - assert handoff is not None - - -@pytest.mark.asyncio -async def test_resolve_handoff_rejects_unknown_sender(): - database = FakeDatabase() - handoff = await resolve_handoff_from_inbound( - database, - account_id="acct-1", - agent_email="support@gentic.email", - sender_address="stranger@example.com", - in_reply_to="", - references=[], - ) - assert handoff is None - - -@pytest.mark.asyncio -async def test_external_scope_constant_isolated_from_handoff(): +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 + + assert not hasattr(module, "maybe_trigger_handoff") + assert not hasattr(module, "process_human_reply") + assert not hasattr(module, "resolve_handoff_from_inbound") diff --git a/frontend/src/api.js b/frontend/src/api.js index 112f872..07604c3 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -145,10 +145,6 @@ 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 listHandoffConversations = (token) => - request("/agent/handoff/conversations", { token }); -export const getHandoffConversation = (handoffId, token) => - request(`/agent/handoff/conversations/${encodeURIComponent(handoffId)}`, { token }); export const listStickyNotes = (token) => request("/sticky-notes", { token }); diff --git a/frontend/src/components/HandoffSettings.jsx b/frontend/src/components/HandoffSettings.jsx index 8bf6352..2ee02cf 100644 --- a/frontend/src/components/HandoffSettings.jsx +++ b/frontend/src/components/HandoffSettings.jsx @@ -130,23 +130,12 @@ export default function HandoffSettings() {

- Your Agent handles email. You step in when needed. + Configure Human Handoff rules for your Agent.

- Human Handoff sends selected conversations to the Agent's Human Owner. The owner can - reply from their normal inbox, and Gentic carries that response back to the original sender - without exposing or mixing the private Agent ↔ Human conversation. + Saved settings are stored for future use. They do not send, forward, mirror, or relay + email today — all mail continues through the normal Gentic inbox and outbox.

- - - The external conversation and Human Handoff remain separate and independently threaded. -
@@ -167,8 +156,8 @@ export default function HandoffSettings() {

No handoff rules yet

- Configure when your Agent should hand conversations to the Human Owner. - Handoff transport messages stay out of your normal Inbox and Sent folders. + Configure when your Agent should involve a Human Owner. Settings are saved only; + no emails are sent automatically.

- -
- - {tab === "handoffs" && ( - selectedHandoffId ? ( - setSelectedHandoffId(null)} - /> - ) : ( - - ) - )} - {tab === "settings" && } + ); } diff --git a/frontend/src/components/HumanHandoffDrawer.jsx b/frontend/src/components/HumanHandoffDrawer.jsx index 0a3cf74..77d3564 100644 --- a/frontend/src/components/HumanHandoffDrawer.jsx +++ b/frontend/src/components/HumanHandoffDrawer.jsx @@ -244,14 +244,8 @@ export default function HumanHandoffDrawer({ onChange={setAllowReplies} />

- When you reply to a handed-off email from your regular inbox, AGMail will recognize your response and continue the original conversation through the agent's email address. + Saved for future use. Human reply handling is not active yet — all email uses the normal Gentic inbox and outbox.

- {allowReplies && ( -
- How it works -

Email arrives → Agent handles it. If human attention is required → Human Handoff → Your inbox. You reply normally → AGMail → Original conversation.

-
- )} {allowReplies && (
Replies will be sent as diff --git a/frontend/src/utils/handoffUtils.js b/frontend/src/utils/handoffUtils.js index 70935ca..e67aec3 100644 --- a/frontend/src/utils/handoffUtils.js +++ b/frontend/src/utils/handoffUtils.js @@ -76,7 +76,7 @@ export function buildHandoffSummary({ ? "when they require your personal decision or attention" : "when your handoff instructions apply"; - return `Your agent will handle email normally. ${labelText} that are ${typeText} will be handed to ${ownerEmail || "the human owner"} ${instructionText}.`; + return `Saved rule: ${labelText} that are ${typeText} would involve ${ownerEmail || "the human owner"} ${instructionText}. Email routing is not active yet.`; } export function buildRuleCardSummary(rule) {