From 36a55930f99c4cd2b039ae09705728f98680d02c Mon Sep 17 00:00:00 2001 From: debu-2 Date: Tue, 6 Oct 2026 01:51:23 +0530 Subject: [PATCH] human handoffs --- backend/app/main.py | 57 ++ backend/app/routers/agent_handoff.py | 309 ++++++++ backend/app/routers/emails.py | 103 ++- backend/app/routers/threads.py | 314 ++++++++ backend/app/schemas.py | 150 +++- backend/app/services/handoff_evaluator.py | 117 +++ backend/app/services/human_handoff_service.py | 669 ++++++++++++++++++ backend/app/services/inbound_smtp.py | 133 +++- backend/app/services/mail_sender.py | 19 +- backend/app/services/mailbox_scope.py | 18 + backend/app/services/message_ids.py | 51 ++ backend/app/services/mime_parser.py | 21 +- backend/app/services/reply_recipients.py | 49 ++ backend/app/services/subject_normalize.py | 16 + backend/app/services/thread_resolver.py | 252 +++++++ backend/app/services/thread_service.py | 137 ++++ backend/tests/test_handoff_evaluator.py | 33 + backend/tests/test_human_handoff_service.py | 124 ++++ backend/tests/test_reply_recipients.py | 30 + backend/tests/test_thread_resolver.py | 129 ++++ frontend/src/api.js | 19 + frontend/src/components/AgAgentPanel.jsx | 17 +- frontend/src/components/ComposeDock.jsx | 26 +- frontend/src/components/ComposeModal.jsx | 34 +- frontend/src/components/DashboardPage.jsx | 12 +- .../src/components/HandoffConversation.jsx | 74 ++ frontend/src/components/HandoffInbox.jsx | 87 +++ frontend/src/components/HandoffSettings.jsx | 294 ++++++++ frontend/src/components/HumanHandoff.jsx | 58 ++ .../src/components/HumanHandoffDrawer.jsx | 292 ++++++++ frontend/src/components/MailDetail.jsx | 232 +++--- frontend/src/components/MailRow.jsx | 2 + frontend/src/components/MailboxView.jsx | 97 ++- frontend/src/styles.css | 132 ++++ frontend/src/utils/handoffUtils.js | 71 ++ 35 files changed, 4039 insertions(+), 139 deletions(-) create mode 100644 backend/app/routers/agent_handoff.py create mode 100644 backend/app/routers/threads.py create mode 100644 backend/app/services/handoff_evaluator.py create mode 100644 backend/app/services/human_handoff_service.py create mode 100644 backend/app/services/mailbox_scope.py create mode 100644 backend/app/services/message_ids.py create mode 100644 backend/app/services/reply_recipients.py create mode 100644 backend/app/services/subject_normalize.py create mode 100644 backend/app/services/thread_resolver.py create mode 100644 backend/app/services/thread_service.py create mode 100644 backend/tests/test_handoff_evaluator.py create mode 100644 backend/tests/test_human_handoff_service.py create mode 100644 backend/tests/test_reply_recipients.py create mode 100644 backend/tests/test_thread_resolver.py create mode 100644 frontend/src/components/HandoffConversation.jsx create mode 100644 frontend/src/components/HandoffInbox.jsx create mode 100644 frontend/src/components/HandoffSettings.jsx create mode 100644 frontend/src/components/HumanHandoff.jsx create mode 100644 frontend/src/components/HumanHandoffDrawer.jsx create mode 100644 frontend/src/utils/handoffUtils.js diff --git a/backend/app/main.py b/backend/app/main.py index 7f25612..5ab47ff 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -8,6 +8,7 @@ from app.config import get_settings from app.database import SessionFactory, engine, mongo_client, mongo_db from app.models import Base from app.routers import ( + agent_handoff, aliases, auth, dns_setup, @@ -18,7 +19,9 @@ from app.routers import ( people, sections, sticky_notes, + threads, ) +from app.services.thread_service import create_thread, message_event_at from app.services.domain_provisioning import ensure_platform_domain from app.services.inbound_smtp import InboundSMTPServer @@ -26,6 +29,25 @@ settings = get_settings() inbound_server = InboundSMTPServer() +async def backfill_missing_thread_ids(database) -> None: + cursor = database.mailbox_entries.find({"thread_id": None}) + async for entry in cursor: + message = await database.messages.find_one({"_id": entry["message_id"]}) + if message is None: + continue + thread_id = await create_thread( + database, + account_id=entry["account_id"], + subject=entry.get("subject") or message.get("subject", ""), + message_id=entry["message_id"], + event_at=entry.get("event_at") or message_event_at(message), + ) + await database.mailbox_entries.update_one( + {"_id": entry["_id"]}, + {"$set": {"thread_id": thread_id}}, + ) + + async def migrate_mail_group_columns(connection) -> None: await connection.execute( text( @@ -168,6 +190,39 @@ async def lifespan(_: FastAPI): ], name="mailbox_search", ) + await mongo_db.messages.create_index( + [("message_id", 1)], + sparse=True, + ) + await mongo_db.mailbox_entries.create_index( + [("account_id", 1), ("thread_id", 1), ("event_at", -1), ("_id", -1)] + ) + await mongo_db.email_threads.create_index( + [("account_id", 1), ("last_message_at", -1), ("_id", -1)] + ) + await mongo_db.human_handoff_rules.create_index( + [("account_id", 1), ("updated_at", -1)] + ) + 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)] + ) + await mongo_db.human_handoffs.create_index( + [("account_id", 1), ("handoff_thread_id", 1)] + ) + await mongo_db.handoff_message_links.create_index( + [("handoff_id", 1), ("created_at", -1)] + ) + await mongo_db.mailbox_entries.create_index( + [("account_id", 1), ("message_scope", 1), ("mailbox", 1), ("event_at", -1)] + ) + await mongo_db.messages.create_index( + [("message_id", 1), ("message_scope", 1)], + sparse=True, + ) + await backfill_missing_thread_ids(mongo_db) if settings.inbound_smtp_enabled: await inbound_server.start() try: @@ -192,6 +247,8 @@ app.add_middleware( allow_headers=["*"], ) app.include_router(emails.router, prefix=settings.api_prefix) +app.include_router(threads.router, prefix=settings.api_prefix) +app.include_router(agent_handoff.router, prefix=settings.api_prefix) app.include_router(sticky_notes.router, prefix=settings.api_prefix) app.include_router(auth.router, prefix=settings.api_prefix) app.include_router(domains.router, prefix=settings.api_prefix) diff --git a/backend/app/routers/agent_handoff.py b/backend/app/routers/agent_handoff.py new file mode 100644 index 0000000..9845d3b --- /dev/null +++ b/backend/app/routers/agent_handoff.py @@ -0,0 +1,309 @@ +import re +from datetime import datetime, timezone +from bson import ObjectId +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import EmailStr, Field + +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.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"]) + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _owner_name_from_email(email: str) -> str: + local = email.split("@", 1)[0] + 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"]), + owner_email=document["owner_email"], + owner_name=document.get("owner_name") or _owner_name_from_email(document["owner_email"]), + labels=document.get("labels", []), + email_types=document.get("email_types", []), + instructions=document.get("instructions", ""), + allow_replies=document.get("allow_replies", True), + reply_as=document.get("reply_as", "agent"), + active=document.get("active", True), + created_at=document["created_at"], + updated_at=document["updated_at"], + ) + + +@router.get("", response_model=list[HandoffRuleRead]) +async def list_handoff_rules( + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + cursor = database.human_handoff_rules.find( + {"account_id": str(account.id)} + ).sort("updated_at", -1) + return [_serialize(document) async for document in cursor] + + +@router.post("", response_model=HandoffRuleRead, status_code=status.HTTP_201_CREATED) +async def create_handoff_rule( + payload: HandoffRuleCreate, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + now = _now() + document = { + "_id": ObjectId(), + "account_id": str(account.id), + "owner_email": str(payload.owner_email).lower(), + "owner_name": payload.owner_name or _owner_name_from_email(str(payload.owner_email)), + "labels": payload.labels, + "email_types": payload.email_types, + "instructions": payload.instructions.strip(), + "allow_replies": payload.allow_replies, + "reply_as": payload.reply_as, + "active": payload.active, + "created_at": now, + "updated_at": now, + } + await database.human_handoff_rules.insert_one(document) + return _serialize(document) + + +@router.patch("/{rule_id}", response_model=HandoffRuleRead) +async def update_handoff_rule( + rule_id: str, + payload: HandoffRuleUpdate, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + 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 "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, + ) + if result is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Handoff rule not found.") + return _serialize(result) + + +@router.delete("/{rule_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_handoff_rule( + rule_id: str, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + if not ObjectId.is_valid(rule_id): + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid rule ID.") + result = await database.human_handoff_rules.delete_one( + {"_id": ObjectId(rule_id), "account_id": str(account.id)} + ) + if result.deleted_count == 0: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Handoff rule not found.") + + +@router.post("/test", response_model=HandoffTestResult) +async def test_handoff( + payload: HandoffTestRequest, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + labels = payload.labels + email_types = payload.email_types + instructions = payload.instructions + owner_email = payload.owner_email + + if payload.rule_id: + if not ObjectId.is_valid(payload.rule_id): + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid rule ID.") + rule = await database.human_handoff_rules.find_one( + {"_id": ObjectId(payload.rule_id), "account_id": str(account.id)} + ) + if rule is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Handoff rule not found.") + labels = rule.get("labels", []) + email_types = rule.get("email_types", []) + instructions = rule.get("instructions", "") + owner_email = rule.get("owner_email", owner_email) + + evaluation = evaluate_handoff( + email=payload.email.model_dump(), + labels=labels, + email_types=email_types, + instructions=instructions or "", + ) + return HandoffTestResult( + owner_email=owner_email, + **evaluation, + result_path="Agent → Human Owner" if evaluation["would_handoff"] else "Agent handles email", + 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"}, + ] + rows = [row async for row in database.mailbox_entries.aggregate(pipeline)] + human_email = handoff.get("human_email", "").lower() + messages: list[HandoffMessageRead] = [] + for row in rows: + message = row["message"] + sender_address = (message.get("sender") or {}).get("address", "").lower() + role = "human" if sender_address == human_email else "agent" + event_at = row.get("event_at") or message.get("received_at") or message.get("sent_at") or message["created_at"] + messages.append( + HandoffMessageRead( + id=str(message["_id"]), + role=role, + subject=message.get("subject", ""), + snippet=message.get("snippet", ""), + body_text=message.get("body_text", ""), + 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 { + "labels": HANDOFF_LABELS, + "email_types": [ + { + "id": key, + "title": meta["title"], + } + for key, meta in EMAIL_TYPE_META.items() + ], + } diff --git a/backend/app/routers/emails.py b/backend/app/routers/emails.py index 119c3a1..8acaecc 100644 --- a/backend/app/routers/emails.py +++ b/backend/app/routers/emails.py @@ -24,8 +24,20 @@ from app.schemas import ( ) from app.services.domain_provisioning import get_domain_by_name from app.services.mail_sender import deliver_email +from app.services.message_ids import normalize_message_id from app.services.mime_parser import parse_mime from app.services.raw_mime_storage import get_raw_mime_storage +from app.services.reply_recipients import reply_recipients +from app.services.thread_service import ( + create_thread, + ensure_thread_for_message, + message_event_at, + threading_fields_from_parent, +) +from app.services.mailbox_scope import apply_external_scope +from app.services.thread_resolver import ThreadResolution + +MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" router = APIRouter(prefix="/emails", tags=["emails"]) @@ -93,6 +105,7 @@ def summarize(message: dict, entry: dict) -> MessageSummary: is_starred=entry.get("is_starred", False), trashed_at=entry.get("trashed_at"), attachments=message.get("attachments", []), + thread_id=str(entry["thread_id"]) if entry.get("thread_id") else None, ) @@ -117,10 +130,44 @@ async def create_email( account: Account = Depends(get_current_account), ): sender = account.email.lower() - recipients = [str(recipient).lower() for recipient in payload.recipients] - cc = [str(recipient).lower() for recipient in payload.cc] + parent_message = None + thread_id = None + threading_fields: dict = {} + if payload.reply_to_message_id: + if not ObjectId.is_valid(payload.reply_to_message_id): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid reply target.", + ) + parent_id = ObjectId(payload.reply_to_message_id) + parent_entry = await database.mailbox_entries.find_one( + {"account_id": str(account.id), "message_id": parent_id} + ) + parent_message = await database.messages.find_one({"_id": parent_id}) + if parent_message is None or parent_entry is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Reply target not found.", + ) + thread_id = parent_entry.get("thread_id") + threading_fields = threading_fields_from_parent(parent_message) + to_addresses, cc_addresses = reply_recipients( + message=parent_message, + viewer_email=sender, + reply_all=payload.reply_all, + ) + recipients = to_addresses or [str(recipient).lower() for recipient in payload.recipients] + cc = cc_addresses or [str(recipient).lower() for recipient in payload.cc] + else: + recipients = [str(recipient).lower() for recipient in payload.recipients] + cc = [str(recipient).lower() for recipient in payload.cc] bcc = [str(recipient).lower() for recipient in payload.bcc] all_recipients = [*recipients, *cc, *bcc] + if not all_recipients: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="At least one recipient is required.", + ) if len(set(all_recipients)) != len(all_recipients): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, @@ -139,8 +186,13 @@ async def create_email( ) created_at = utc_now() + subject = payload.subject + if parent_message and not subject.strip(): + parent_subject = parent_message.get("subject", "") + subject = parent_subject if parent_subject.lower().startswith("re:") else f"Re: {parent_subject}" document = { "direction": "outbound", + "message_scope": MESSAGE_SCOPE_EXTERNAL, "account_id": str(account.id), "sender": address( sender, @@ -149,17 +201,21 @@ async def create_email( "to": [address(value) for value in recipients], "cc": [address(value) for value in cc], "bcc": [address(value) for value in bcc], - "subject": payload.subject, + "subject": subject, "snippet": re.sub(r"\s+", " ", payload.body_text).strip()[:240], "body_text": payload.body_text, "body_html": payload.body_html, "attachments": [], "created_at": created_at, + "sent_at": created_at, "status": "queued", + **threading_fields, + "thread_relation": "reply" if parent_message else "new_thread", + "thread_resolution_method": "in_reply_to" if parent_message else "new_thread", } try: - await deliver_email( + internet_message_id = await deliver_email( sender=sender, recipients=recipients, cc=cc, @@ -168,19 +224,45 @@ async def create_email( body_text=document["body_text"], body_html=document.get("body_html"), domain=domain, + in_reply_to=threading_fields.get("in_reply_to"), + references=threading_fields.get("references"), ) document["status"] = "accepted" + document["message_id"] = normalize_message_id(internet_message_id) except Exception as error: document["status"] = "failed" document["error"] = str(error) result = await database.messages.insert_one(document) document["_id"] = result.inserted_id + if parent_message and thread_id: + resolution = ThreadResolution( + thread_id=thread_id, + parent_message_id=parent_message["_id"], + thread_relation="reply", + thread_resolution_method="in_reply_to", + ) + thread_id = await ensure_thread_for_message( + database, + account_id=str(account.id), + message=document, + resolution=resolution, + ) + else: + thread_id = await create_thread( + database, + account_id=str(account.id), + subject=document["subject"], + message_id=result.inserted_id, + event_at=message_event_at(document), + ) await database.mailbox_entries.insert_one( { "account_id": str(account.id), "message_id": result.inserted_id, + "thread_id": thread_id, "mailbox": "sent", + "message_scope": MESSAGE_SCOPE_EXTERNAL, "direction": "outbound", "sender": document["sender"], "recipients": document["to"], @@ -207,7 +289,7 @@ async def create_email( recipients=recipients, cc=cc, bcc=bcc, - subject=payload.subject, + subject=subject, body_text=payload.body_text, body_html=payload.body_html, status=document["status"], @@ -236,6 +318,10 @@ async def get_email( if entry is not None else None ) + if entry is not None and ( + entry.get("message_scope") == "HANDOFF" or entry.get("mailbox") == "handoff" + ): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Email not found.") if document is None or entry is None: legacy = await legacy_email(database, message_id, viewer_email) if legacy is not None: @@ -260,6 +346,12 @@ async def get_email( body_text=body_text, body_html=body_html, message_id=document.get("message_id"), + in_reply_to=document.get("in_reply_to"), + references=document.get("references", []), + parent_message_id=str(document["parent_message_id"]) + if document.get("parent_message_id") + else None, + thread_relation=document.get("thread_relation"), ) @@ -309,6 +401,7 @@ async def list_emails( entry_match["event_at"] = event_range if q and q.strip(): entry_match["$text"] = {"$search": q.strip()} + apply_external_scope(entry_match) if cursor: created_at, entry_id = decode_cursor(cursor) cursor_clause = [ diff --git a/backend/app/routers/threads.py b/backend/app/routers/threads.py new file mode 100644 index 0000000..ac5020f --- /dev/null +++ b/backend/app/routers/threads.py @@ -0,0 +1,314 @@ +import base64 +import json +import re +from datetime import date, datetime, time, timedelta, timezone +from typing import Annotated + +from bson import ObjectId +from fastapi import APIRouter, Depends, HTTPException, Query, status + +from app.database import get_mongo_database +from app.models import Account +from app.routers.auth import get_current_account +from app.routers.emails import decode_cursor, encode_cursor, summarize +from app.schemas import MessageRead, ThreadPage, ThreadRead, ThreadSummary +from app.services.mailbox_scope import apply_external_scope +from app.services.mime_parser import parse_mime +from app.services.raw_mime_storage import get_raw_mime_storage + +router = APIRouter(prefix="/threads", tags=["threads"]) + + +def encode_thread_cursor(thread: dict) -> str: + value = json.dumps( + { + "latest_at": thread["latest_at"].isoformat(), + "id": str(thread["_id"]), + }, + separators=(",", ":"), + ) + return base64.urlsafe_b64encode(value.encode()).decode().rstrip("=") + + +def decode_thread_cursor(value: str) -> tuple[datetime, ObjectId]: + try: + padded = value + "=" * (-len(value) % 4) + payload = json.loads(base64.urlsafe_b64decode(padded).decode()) + return datetime.fromisoformat(payload["latest_at"]), ObjectId(payload["id"]) + except (ValueError, TypeError, KeyError, json.JSONDecodeError) as error: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid cursor.", + ) from error + + +def participant_key(address: dict) -> str: + return address.get("address", "").lower() + + +def unique_addresses(entries: list[dict]) -> list[dict]: + seen: set[str] = set() + values: list[dict] = [] + for entry in entries: + for key in ("sender",): + address = entry.get(key) + if isinstance(address, dict): + token = participant_key(address) + if token and token not in seen: + seen.add(token) + values.append(address) + return values + + +def thread_summary_from_group(row: dict, mailbox: str) -> ThreadSummary: + latest_entry = row["latest_entry"] + latest_message = latest_entry["message"] + senders = unique_addresses(row.get("entries", [latest_entry])) + participants = senders[:] + for entry in row.get("entries", []): + for recipient in entry.get("recipients", []): + token = participant_key(recipient) + if token and token not in {participant_key(item) for item in participants}: + participants.append(recipient) + latest_at = latest_entry.get("event_at") or latest_message.get("received_at") or latest_message["created_at"] + return ThreadSummary( + id=str(row["_id"]), + mailbox=mailbox, + subject=latest_message.get("subject", ""), + snippet=latest_message.get("snippet", ""), + participants=[{"name": item.get("name", ""), "address": item.get("address", "")} for item in participants[:6]], + message_count=row.get("message_count", 1), + unread_count=row.get("unread_count", 0), + is_read=row.get("unread_count", 0) == 0, + is_starred=bool(row.get("is_starred")), + has_attachments=bool(row.get("has_attachments")), + latest_message_id=str(latest_message["_id"]), + latest_at=latest_at, + senders=[{"name": item.get("name", ""), "address": item.get("address", "")} for item in senders[:4]], + ) + + +@router.get("", response_model=ThreadPage) +async def list_threads( + mailbox: Annotated[str, Query(pattern="^(inbox|sent|trash)$")] = "inbox", + limit: Annotated[int, Query(ge=1, le=100)] = 25, + cursor: str | None = None, + q: Annotated[str | None, Query(max_length=200)] = None, + is_read: bool | None = None, + sender: Annotated[str | None, Query(max_length=320)] = None, + date_from: date | None = None, + date_to: date | None = None, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + entry_match: dict = { + "account_id": str(account.id), + "deletion_state": {"$in": [None, ""]}, + "thread_id": {"$ne": None}, + } + if mailbox == "trash": + entry_match["trashed_at"] = {"$ne": None} + else: + entry_match.update({"mailbox": mailbox, "trashed_at": None}) + if sender: + pattern = f"^{re.escape(sender.strip())}" + entry_match["$or"] = [ + {"sender.name": {"$regex": pattern, "$options": "i"}}, + {"sender.address": {"$regex": pattern, "$options": "i"}}, + ] + if date_from or date_to: + event_range: dict = {} + if date_from: + event_range["$gte"] = datetime.combine(date_from, time.min, tzinfo=timezone.utc) + if date_to: + event_range["$lt"] = datetime.combine( + date_to + timedelta(days=1), + time.min, + tzinfo=timezone.utc, + ) + entry_match["event_at"] = event_range + apply_external_scope(entry_match) + + pipeline: list[dict] = [ + {"$match": entry_match}, + { + "$lookup": { + "from": "messages", + "localField": "message_id", + "foreignField": "_id", + "as": "message", + } + }, + {"$unwind": "$message"}, + ] + if q and q.strip(): + pipeline.append( + { + "$match": { + "$or": [ + {"subject": {"$regex": re.escape(q.strip()), "$options": "i"}}, + {"snippet": {"$regex": re.escape(q.strip()), "$options": "i"}}, + {"message.subject": {"$regex": re.escape(q.strip()), "$options": "i"}}, + {"message.snippet": {"$regex": re.escape(q.strip()), "$options": "i"}}, + ] + } + } + ) + pipeline.extend( + [ + { + "$group": { + "_id": "$thread_id", + "entries": {"$push": "$$ROOT"}, + "latest_at": {"$max": "$event_at"}, + "unread_count": { + "$sum": {"$cond": [{"$eq": ["$is_read", False]}, 1, 0]} + }, + "is_starred": {"$max": "$is_starred"}, + "has_attachments": { + "$max": { + "$gt": [ + {"$size": {"$ifNull": ["$attachments", []]}}, + 0, + ] + } + }, + "message_count": {"$sum": 1}, + } + }, + ] + ) + if is_read is not None: + pipeline.append( + { + "$match": { + "unread_count": 0 if is_read else {"$gt": 0}, + } + } + ) + pipeline.append({"$sort": {"latest_at": -1, "_id": -1}}) + if cursor: + latest_at, thread_id = decode_thread_cursor(cursor) + pipeline.append( + { + "$match": { + "$or": [ + {"latest_at": {"$lt": latest_at}}, + {"latest_at": latest_at, "_id": {"$lt": thread_id}}, + ] + } + } + ) + pipeline.append({"$limit": limit + 1}) + pipeline.append( + { + "$addFields": { + "latest_entry": { + "$first": { + "$sortArray": { + "input": "$entries", + "sortBy": {"event_at": -1, "_id": -1}, + } + } + } + } + } + ) + rows = [row async for row in database.mailbox_entries.aggregate(pipeline)] + has_more = len(rows) > limit + rows = rows[:limit] + items = [thread_summary_from_group(row, mailbox) for row in rows] + next_cursor = None + if has_more and rows: + next_cursor = encode_thread_cursor( + {"latest_at": rows[-1]["latest_at"], "_id": rows[-1]["_id"]} + ) + return ThreadPage(items=items, next_cursor=next_cursor) + + +async def message_read( + database, + message: dict, + entry: dict, + account: Account, +) -> MessageRead: + body_text = message.get("body_text", "") + body_html = message.get("body_html") + if message.get("raw_mime"): + raw = await get_raw_mime_storage().get(message["raw_mime"]) + parsed = parse_mime(raw) + body_text = parsed["body_text"] + body_html = parsed["body_html"] + summary = summarize(message, entry) + return MessageRead( + **summary.model_dump(), + cc=message.get("cc", []), + bcc=message.get("bcc", []) if message.get("account_id") == str(account.id) else [], + reply_to=message.get("reply_to", []), + body_text=body_text, + body_html=body_html, + message_id=message.get("message_id"), + in_reply_to=message.get("in_reply_to"), + references=message.get("references", []), + parent_message_id=str(message["parent_message_id"]) + if message.get("parent_message_id") + else None, + thread_relation=message.get("thread_relation"), + ) + + +@router.get("/{thread_id}", response_model=ThreadRead) +async def get_thread( + thread_id: str, + database=Depends(get_mongo_database), + account: Account = Depends(get_current_account), +): + if not ObjectId.is_valid(thread_id): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid thread ID.", + ) + thread_object_id = ObjectId(thread_id) + thread = await database.email_threads.find_one( + {"_id": thread_object_id, "account_id": str(account.id)} + ) + if thread is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Thread not found.") + if thread.get("thread_type") == "HUMAN_HANDOFF": + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Thread not found.") + + entry_match = { + "account_id": str(account.id), + "thread_id": thread_object_id, + "deletion_state": {"$in": [None, ""]}, + } + apply_external_scope(entry_match) + pipeline = [ + { + "$match": entry_match, + }, + {"$sort": {"event_at": 1, "_id": 1}}, + { + "$lookup": { + "from": "messages", + "localField": "message_id", + "foreignField": "_id", + "as": "message", + } + }, + {"$unwind": "$message"}, + ] + rows = [row async for row in database.mailbox_entries.aggregate(pipeline)] + if not rows: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Thread not found.") + + messages = [ + await message_read(database, row["message"], row, account) + for row in rows + ] + return ThreadRead( + thread_id=thread_id, + subject=messages[-1].subject if messages else "", + message_count=len(messages), + messages=messages, + ) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 76c496f..756c055 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -224,6 +224,112 @@ class OrgSecuritySettingsRead(BaseModel): model_config = ConfigDict(from_attributes=True) +class HandoffEmailSample(BaseModel): + subject: str = "" + snippet: str = "" + body_text: str = "" + is_starred: bool = False + + +class HandoffRuleCreate(BaseModel): + owner_email: EmailStr + owner_name: str | None = None + labels: list[str] = Field(default_factory=list) + email_types: list[str] = Field(default_factory=list) + instructions: str = "" + allow_replies: bool = True + reply_as: Literal["agent", "human"] = "agent" + active: bool = True + + +class HandoffRuleUpdate(BaseModel): + owner_email: EmailStr | None = None + owner_name: str | None = None + labels: list[str] | None = None + email_types: list[str] | None = None + instructions: str | None = None + allow_replies: bool | None = None + reply_as: Literal["agent", "human"] | None = None + active: bool | None = None + + +class HandoffRuleRead(BaseModel): + id: str + owner_email: EmailStr + owner_name: str + labels: list[str] + email_types: list[str] + instructions: str + allow_replies: bool + reply_as: Literal["agent", "human"] + active: bool + created_at: datetime + updated_at: datetime + + +class HandoffTestRequest(BaseModel): + rule_id: str | None = None + owner_email: EmailStr | None = None + labels: list[str] = Field(default_factory=list) + email_types: list[str] = Field(default_factory=list) + instructions: str = "" + email: HandoffEmailSample + + +class HandoffTestResult(BaseModel): + owner_email: EmailStr | None = None + label_matched: bool + matched_labels: list[str] + type_matched: bool + matched_types: list[str] + instruction_matched: bool + instruction_reason: str + would_handoff: bool + result_path: str + reason: str + + +class HandoffLinkedExternal(BaseModel): + thread_id: str + subject: str + sender: str + + +class HandoffConversationSummary(BaseModel): + id: str + external_subject: str + external_sender: str + human_email: str + status: str + status_label: str + current_owner: str + snippet: str + message_count: int + updated_at: datetime + + +class HandoffMessageRead(BaseModel): + id: str + role: Literal["agent", "human"] + subject: str + snippet: str + body_text: str + event_at: datetime + + +class HandoffConversationRead(BaseModel): + id: str + external_subject: str + external_sender: str + human_email: str + handoff_reason: str + status: str + status_label: str + current_owner: str + linked_external: HandoffLinkedExternal + messages: list[HandoffMessageRead] + + class OrgSecuritySettingsUpdate(BaseModel): require_strong_passwords: bool | None = None require_mfa_admins: bool | None = None @@ -236,12 +342,21 @@ class OrgMemberRoleUpdate(BaseModel): class EmailCreate(BaseModel): - recipients: list[EmailStr] = Field(min_length=1) + recipients: list[EmailStr] = Field(default_factory=list) cc: list[EmailStr] = Field(default_factory=list) bcc: list[EmailStr] = Field(default_factory=list) subject: str = Field(default="", max_length=998, pattern=r"^[^\r\n]*$") body_text: str = "" body_html: str | None = None + reply_to_message_id: str | None = None + reply_all: bool = False + thread_id: str | None = None + + @model_validator(mode="after") + def recipients_required_without_reply(self): + if not self.reply_to_message_id and not self.recipients: + raise ValueError("At least one recipient is required.") + return self class EmailRead(EmailCreate): @@ -278,6 +393,7 @@ class MessageSummary(BaseModel): is_starred: bool = False trashed_at: datetime | None = None attachments: list[AttachmentRead] = Field(default_factory=list) + thread_id: str | None = None class MessageRead(MessageSummary): @@ -287,6 +403,10 @@ class MessageRead(MessageSummary): body_text: str = "" body_html: str | None = None message_id: str | None = None + in_reply_to: str | None = None + references: list[str] = Field(default_factory=list) + parent_message_id: str | None = None + thread_relation: str | None = None class MessageReadState(BaseModel): @@ -322,6 +442,34 @@ class MessagePage(BaseModel): next_cursor: str | None = None +class ThreadSummary(BaseModel): + id: str + mailbox: Literal["inbox", "sent", "trash"] + subject: str + snippet: str + participants: list[MailAddress] = Field(default_factory=list) + message_count: int + unread_count: int = 0 + is_read: bool = True + is_starred: bool = False + has_attachments: bool = False + latest_message_id: str + latest_at: datetime + senders: list[MailAddress] = Field(default_factory=list) + + +class ThreadPage(BaseModel): + items: list[ThreadSummary] + next_cursor: str | None = None + + +class ThreadRead(BaseModel): + thread_id: str + subject: str + message_count: int + messages: list[MessageRead] + + class DomainCreate(BaseModel): domain_name: str = Field(min_length=3, max_length=253) account_id: UUID | None = None diff --git a/backend/app/services/handoff_evaluator.py b/backend/app/services/handoff_evaluator.py new file mode 100644 index 0000000..79e442d --- /dev/null +++ b/backend/app/services/handoff_evaluator.py @@ -0,0 +1,117 @@ +import re + +HANDOFF_LABELS = ["Important", "Finance", "Clients", "Projects", "Personal", "Support"] + +EMAIL_TYPE_META = { + "utility": { + "title": "Utility", + "patterns": r"account update|alert|notice|security|verification|password|service information|noreply|no-reply", + }, + "transactional": { + "title": "Transactional", + "patterns": r"order|payment|invoice|receipt|billing|subscription|shipped|confirmation|refund|approval", + }, + "promotional": { + "title": "Promotional", + "patterns": r"offer|deal|discount|promo|sale|limited time|save \d+%", + }, + "marketing": { + "title": "Marketing", + "patterns": r"newsletter|campaign|marketing|outreach|webinar|subscribe|unsubscribe", + }, +} + + +def _email_text(email: dict) -> str: + parts = [ + email.get("subject", ""), + email.get("snippet", ""), + email.get("body_text", ""), + ] + return " ".join(parts).lower() + + +def infer_labels(email: dict) -> list[str]: + text = _email_text(email) + matched: list[str] = [] + if re.search(r"urgent|important|asap|priority", text) or email.get("is_starred"): + matched.append("Important") + if re.search(r"invoice|bill|payment|finance|receipt", text): + matched.append("Finance") + if re.search(r"client|customer", text): + matched.append("Clients") + if re.search(r"project|proposal|contract", text): + matched.append("Projects") + if re.search(r"personal|family", text): + matched.append("Personal") + if re.search(r"support|help desk|ticket|assistance", text): + matched.append("Support") + return matched + + +def infer_email_types(email: dict) -> list[str]: + text = _email_text(email) + matched: list[str] = [] + for key, meta in EMAIL_TYPE_META.items(): + if re.search(meta["patterns"], text, flags=re.IGNORECASE): + matched.append(key) + return matched + + +def instruction_matches(instructions: str, email: dict) -> tuple[bool, str]: + text = _email_text(email) + instructions = (instructions or "").strip() + if not instructions: + return True, "No handoff instructions were set, so this step passes." + + keywords = re.findall(r"[a-z]{4,}", instructions.lower()) + topic_hits = [word for word in keywords if word in text and len(word) > 4] + if topic_hits: + return True, f"Mentions topics related to your instructions ({', '.join(topic_hits[:3])})." + + intent_patterns = [ + (r"refund|chargeback", "This email appears to involve a refund or payment issue."), + (r"approval|authorize|sign off", "This email appears to require approval."), + (r"human|speak with|talk to|call me", "The sender appears to be asking for a person."), + (r"not confident|unsure|escalat", "Your instructions mention uncertain situations."), + (r"urgent|asap|immediately", "The message appears urgent."), + ] + for pattern, reason in intent_patterns: + if re.search(pattern, instructions, flags=re.IGNORECASE) and re.search(pattern, text, flags=re.IGNORECASE): + return True, reason + + if re.search(r"refund|approval|human|speak with|person|payment", instructions, flags=re.IGNORECASE): + if re.search(r"refund|approval|human|speak|person|payment|decision", text, flags=re.IGNORECASE): + return True, "The email situation matches your handoff instructions." + + return False, "This email does not clearly match your handoff instructions." + + +def evaluate_handoff( + *, + email: dict, + labels: list[str], + email_types: list[str], + instructions: str, +) -> dict: + inferred_labels = infer_labels(email) + inferred_types = infer_email_types(email) + + label_ok = not labels or any(label in inferred_labels for label in labels) + matched_labels = [label for label in labels if label in inferred_labels] if labels else inferred_labels[:1] + + type_ok = not email_types or any(item in inferred_types for item in email_types) + matched_types = [item for item in email_types if item in inferred_types] if email_types else inferred_types[:1] + + instruction_ok, instruction_reason = instruction_matches(instructions, email) + would_handoff = label_ok and type_ok and instruction_ok + + return { + "label_matched": label_ok, + "matched_labels": matched_labels, + "type_matched": type_ok, + "matched_types": matched_types, + "instruction_matched": instruction_ok, + "instruction_reason": instruction_reason, + "would_handoff": would_handoff, + } diff --git a/backend/app/services/human_handoff_service.py b/backend/app/services/human_handoff_service.py new file mode 100644 index 0000000..f15f809 --- /dev/null +++ b/backend/app/services/human_handoff_service.py @@ -0,0 +1,669 @@ +import logging +import re +from datetime import datetime, timezone + +from bson import ObjectId + +from app.services.handoff_evaluator import evaluate_handoff +from app.services.mail_sender import deliver_email +from app.services.message_ids import build_references_header, normalize_message_id, parse_message_id_list +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_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_subject(external_subject: str) -> str: + subject = (external_subject or "(No subject)").strip() + if subject.lower().startswith("human handoff:"): + return subject + return f"Human Handoff: {subject}" + + +def _build_handoff_body( + *, + external_sender: str, + external_subject: str, + external_body: str, + handoff_reason: str, + agent_email: str, +) -> str: + return ( + "Human Handoff\n\n" + "A conversation needs your attention.\n\n" + f"From\n{external_sender}\n\n" + f"Subject\n{external_subject}\n\n" + "Latest message\n\n" + f"\"{external_body.strip()}\"\n\n" + "Why this was handed off\n\n" + f"{handoff_reason}\n\n" + "────────────────────────────────────────\n\n" + f"Reply to this email to respond to the original sender through {agent_email}.\n\n" + "Your reply will be sent to the customer.\n\n" + "────────────────────────────────────────\n\n" + "Gentic Human Handoff" + ) + + +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_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"}, + ] + rows = [row async for row in database.mailbox_entries.aggregate(pipeline)] + 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"}, + ] + rows = [row async for row in database.mailbox_entries.aggregate(pipeline)] + 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))): + message = await database.messages.find_one( + { + "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 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, +) -> dict: + now = event_at or _now() + document = { + "direction": direction, + "message_scope": MESSAGE_SCOPE_HANDOFF, + "account_id": account_id, + "sender": sender, + "to": recipients, + "subject": subject, + "snippet": _snippet(body_text), + "body_text": body_text, + "body_html": None, + "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": normalize_message_id(internet_message_id), + "in_reply_to": normalize_message_id(in_reply_to), + "references": references or [], + } + 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: + external_sender = external_message.get("sender", {}) + external_sender_text = external_sender.get("address") or "unknown sender" + external_subject = external_message.get("subject", "") + external_body = external_message.get("body_text") or external_message.get("snippet") or "" + + 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_subject(external_subject) + body = _build_handoff_body( + external_sender=external_sender_text, + external_subject=external_subject, + external_body=external_body, + handoff_reason="New message in the linked conversation.", + agent_email=agent_email, + ) + relationship = LINK_EXTERNAL_UPDATE_RELAYED + else: + handoff_thread_id = await create_empty_handoff_thread( + database, + account_id=account_id, + subject=_handoff_subject(external_subject), + ) + threading = {} + subject = _handoff_subject(external_subject) + body = _build_handoff_body( + external_sender=external_sender_text, + external_subject=external_subject, + external_body=external_body, + handoff_reason=handoff_reason, + agent_email=agent_email, + ) + 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, + } + 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=body, + body_html=None, + domain=domain, + in_reply_to=threading.get("in_reply_to"), + references=threading.get("references"), + ) + + 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=body, + 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"], + ) + + 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 + + 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=inbound_body, + 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"], + link_relationship=LINK_HUMAN_TRIGGERED_EXTERNAL, + ) + + 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 + + external_parent = await _latest_external_message(database, account_id, external_thread_id) + if external_parent is None: + logger.warning("handoff_reply_missing_external_parent handoff_id=%s", handoff["_id"]) + return + + threading = threading_fields_from_parent(external_parent) + external_subject = handoff.get("external_subject") or external_parent.get("subject", "") + if not re.match(r"^re:\s*", external_subject, flags=re.IGNORECASE): + external_subject = f"Re: {external_subject}" + + external_recipient = handoff.get("external_sender") or external_parent.get("sender", {}).get("address") + if not external_recipient: + return + + customer_body = inbound_body.strip() + internet_message_id = await deliver_email( + sender=agent_email, + recipients=[external_recipient], + cc=[], + bcc=[], + subject=external_subject, + body_text=customer_body, + body_html=None, + domain=domain, + in_reply_to=threading.get("in_reply_to"), + references=threading.get("references"), + ) + + external_doc = { + "direction": "outbound", + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "account_id": account_id, + "sender": {"name": "Gentic Agent", "address": agent_email}, + "to": [{"name": external_recipient.split("@")[0], "address": external_recipient}], + "subject": external_subject, + "snippet": _snippet(customer_body), + "body_text": customer_body, + "body_html": None, + "attachments": [], + "created_at": now, + "sent_at": now, + "status": "accepted", + "message_id": normalize_message_id(internet_message_id), + "in_reply_to": threading.get("in_reply_to"), + "references": threading.get("references", []), + "parent_message_id": external_parent["_id"], + "thread_relation": "reply", + "thread_resolution_method": "human_handoff_bridge", + } + ext_result = await database.messages.insert_one(external_doc) + external_doc["_id"] = ext_result.inserted_id + await attach_message_to_thread( + database, + account_id=account_id, + thread_id=external_thread_id, + message_id=ext_result.inserted_id, + subject=external_subject, + event_at=now, + ) + await database.mailbox_entries.insert_one( + { + "account_id": account_id, + "message_id": ext_result.inserted_id, + "thread_id": external_thread_id, + "mailbox": "sent", + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "direction": "outbound", + "sender": external_doc["sender"], + "recipients": external_doc["to"], + "subject": external_subject, + "snippet": external_doc["snippet"], + "attachments": [], + "status": "accepted", + "is_read": True, + "is_starred": False, + "event_at": now, + "created_at": now, + "trashed_at": None, + "deletion_state": None, + } + ) + + relay_body = f"Sent to customer:\n\n\"{customer_body}\"" + relay_subject = f"Re: {_handoff_subject(handoff.get('external_subject', ''))}" + relay_threading = threading_fields_from_parent(human_message) + relay_internet_id = await deliver_email( + sender=agent_email, + recipients=[human_email], + cc=[], + bcc=[], + subject=relay_subject, + body_text=relay_body, + body_html=None, + domain=domain, + in_reply_to=relay_threading.get("in_reply_to"), + references=relay_threading.get("references"), + ) + await 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=relay_subject, + body_text=relay_body, + internet_message_id=relay_internet_id, + in_reply_to=relay_threading.get("in_reply_to"), + references=relay_threading.get("references"), + handoff_id=handoff["_id"], + link_relationship=LINK_EXTERNAL_RELAYED, + external_message_id=ext_result.inserted_id, + ) + + await database.human_handoffs.update_one( + {"_id": handoff["_id"]}, + { + "$set": { + "status": HANDOFF_STATUS_AGENT_REPLIED, + "current_owner": OWNER_AGENT, + "updated_at": now, + "last_human_message_at": now, + } + }, + ) + + +async def maybe_trigger_handoff( + database, + *, + account_id: str, + agent_email: str, + external_message: dict, + external_thread_id: ObjectId, + domain, +) -> None: + 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 + + existing = await find_active_handoff_for_external_thread( + database, account_id, external_thread_id + ) + reason = evaluation.get("instruction_reason") or rule.get("instructions") or "Matched handoff rules." + 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=existing, + ) + return diff --git a/backend/app/services/inbound_smtp.py b/backend/app/services/inbound_smtp.py index 5468b3e..92d38e5 100644 --- a/backend/app/services/inbound_smtp.py +++ b/backend/app/services/inbound_smtp.py @@ -9,8 +9,18 @@ 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.human_handoff_service import ( + MESSAGE_SCOPE_EXTERNAL, + MESSAGE_SCOPE_HANDOFF, + process_human_reply, + resolve_handoff_from_inbound, + maybe_trigger_handoff, +) from app.services.mime_parser import parse_mime from app.services.raw_mime_storage import delivery_digest, get_raw_mime_storage +from app.services.thread_resolver import participants_from_parsed, resolve_thread +from app.services.thread_service import ensure_thread_for_message, message_event_at logger = logging.getLogger(__name__) @@ -69,9 +79,59 @@ class InboundHandler: body_html = parsed.pop("body_html") stored = await get_raw_mime_storage().put(raw, delivery_key) now = datetime.now(timezone.utc) + sender_address = (parsed.get("sender") or {}).get("address", "").lower() + + 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, + } + 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, + ) + logger.info( + "handoff_human_reply handoff_id=%s account=%s", + handoff["_id"], + account_id, + ) + return f"250 2.0.0 Handoff reply accepted as {delivery_key}" + message = { **parsed, "direction": "inbound", + "message_scope": MESSAGE_SCOPE_EXTERNAL, "envelope_sender": envelope.mail_from.lower(), "envelope_recipients": recipients, "delivery_key": delivery_key, @@ -92,14 +152,54 @@ class InboundHandler: {"$setOnInsert": message}, upsert=True, ) - persisted = await mongo_db.messages.find_one( - {"delivery_key": delivery_key}, - {"_id": 1}, - ) + persisted = await mongo_db.messages.find_one({"delivery_key": delivery_key}) if persisted is None: raise RuntimeError("Inbound message was not persisted.") + + if body_text: + await mongo_db.messages.update_one( + {"_id": persisted["_id"]}, + {"$set": {"body_text": body_text, "body_html": body_html}}, + ) + persisted["body_text"] = body_text + persisted["body_html"] = body_html + + 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, + in_reply_to=parsed.get("in_reply_to"), + references=parsed.get("references", []), + subject=parsed.get("subject", ""), + participants=participants, + ) + thread_id = await ensure_thread_for_message( + mongo_db, + account_id=account_id, + message=persisted, + resolution=resolution, + ) + await mongo_db.email_threads.update_one( + {"_id": thread_id}, + {"$set": {"thread_type": "EXTERNAL"}}, + ) + await mongo_db.messages.update_one( + {"_id": persisted["_id"]}, + { + "$set": { + "message_scope": MESSAGE_SCOPE_EXTERNAL, + "in_reply_to": parsed.get("in_reply_to"), + "references": parsed.get("references", []), + "raw_headers": parsed.get("raw_headers", {}), + "parent_message_id": resolution.parent_message_id, + "thread_relation": resolution.thread_relation, + "thread_resolution_method": resolution.thread_resolution_method, + } + }, + ) await mongo_db.mailbox_entries.update_one( { "account_id": account_id, @@ -111,6 +211,7 @@ class InboundHandler: "account_id": account_id, "message_id": persisted["_id"], "mailbox": "inbox", + "message_scope": MESSAGE_SCOPE_EXTERNAL, "direction": "inbound", "sender": parsed["sender"], "recipients": parsed["to"], @@ -120,14 +221,34 @@ class InboundHandler: "status": "received", "is_read": False, "is_starred": False, - "event_at": now, + "event_at": message_event_at(persisted), "created_at": now, "trashed_at": None, "deletion_state": None, - } + }, + "$set": {"thread_id": thread_id}, }, upsert=True, ) + async with SessionFactory() as database_session: + domain = await get_domain_by_name( + database_session, + agent_email.split("@", 1)[1], + ) + 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, + ) + logger.info( + "message_received message_id=%s thread_id=%s scope=EXTERNAL", + persisted["_id"], + thread_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 311fb70..3ace5e0 100644 --- a/backend/app/services/mail_sender.py +++ b/backend/app/services/mail_sender.py @@ -20,6 +20,9 @@ def build_message( body_text: str, body_html: str | None, reply_to: str | None = None, + in_reply_to: str | None = None, + references: list[str] | None = None, + message_id: str | None = None, ) -> EmailMessage: message = EmailMessage(policy=SMTP_POLICY) message["From"] = sender @@ -30,7 +33,12 @@ def build_message( message["Reply-To"] = reply_to message["Subject"] = subject message["Date"] = email.utils.formatdate(localtime=True) - message["Message-ID"] = email.utils.make_msgid(domain=sender.split("@")[-1]) + generated_id = message_id or email.utils.make_msgid(domain=sender.split("@")[-1]) + message["Message-ID"] = generated_id + if in_reply_to: + message["In-Reply-To"] = in_reply_to + if references: + message["References"] = " ".join(references) message.set_content(body_text) if body_html: message.add_alternative(body_html, subtype="html") @@ -95,7 +103,10 @@ async def deliver_email( body_html: str | None, domain: MailDomain, reply_to: str | None = None, -) -> None: + in_reply_to: str | None = None, + references: list[str] | None = None, + message_id: str | None = None, +) -> str: message = build_message( sender=sender, recipients=recipients, @@ -104,6 +115,9 @@ async def deliver_email( body_text=body_text, body_html=body_html, reply_to=reply_to, + in_reply_to=in_reply_to, + references=references, + message_id=message_id, ) signed = sign_message(message, domain) await send_via_smtp( @@ -111,3 +125,4 @@ async def deliver_email( sender=sender, recipients=[*recipients, *cc, *bcc], ) + return str(message["Message-ID"]) diff --git a/backend/app/services/mailbox_scope.py b/backend/app/services/mailbox_scope.py new file mode 100644 index 0000000..563f2ca --- /dev/null +++ b/backend/app/services/mailbox_scope.py @@ -0,0 +1,18 @@ +EXTERNAL_SCOPE_FILTER = { + "$or": [ + {"message_scope": "EXTERNAL"}, + {"message_scope": {"$exists": False}}, + ] +} + + +def apply_external_scope(entry_match: dict) -> dict: + if "$and" in entry_match: + entry_match["$and"].append(EXTERNAL_SCOPE_FILTER) + return entry_match + if "$or" in entry_match: + existing = entry_match.pop("$or") + entry_match["$and"] = [{"$or": existing}, EXTERNAL_SCOPE_FILTER] + return entry_match + entry_match.update(EXTERNAL_SCOPE_FILTER) + return entry_match diff --git a/backend/app/services/message_ids.py b/backend/app/services/message_ids.py new file mode 100644 index 0000000..132191d --- /dev/null +++ b/backend/app/services/message_ids.py @@ -0,0 +1,51 @@ +import re + +_MESSAGE_ID_PATTERN = re.compile(r"<[^>]+>") + + +def normalize_message_id(value: str | None) -> str | None: + if value is None: + return None + cleaned = value.strip() + if not cleaned: + return None + if not cleaned.startswith("<"): + cleaned = f"<{cleaned.strip('<>')}>" + return cleaned.lower() + + +def parse_message_id_list(header_value: str | None) -> list[str]: + if not header_value: + return [] + folded = re.sub(r"\s+", " ", header_value.strip()) + seen: set[str] = set() + ordered: list[str] = [] + for match in _MESSAGE_ID_PATTERN.findall(folded): + normalized = normalize_message_id(match) + if normalized and normalized not in seen: + seen.add(normalized) + ordered.append(normalized) + if not ordered: + single = normalize_message_id(folded) + if single: + ordered.append(single) + return ordered + + +def first_message_id(header_value: str | None) -> str | None: + values = parse_message_id_list(header_value) + return values[0] if values else None + + +def build_references_header(references: list[str], parent_message_id: str) -> str: + chain: list[str] = [] + seen: set[str] = set() + for value in references: + normalized = normalize_message_id(value) + if normalized and normalized not in seen: + seen.add(normalized) + chain.append(normalized) + parent = normalize_message_id(parent_message_id) + if parent and parent not in seen: + chain.append(parent) + return " ".join(chain) diff --git a/backend/app/services/mime_parser.py b/backend/app/services/mime_parser.py index d76d61a..332d950 100644 --- a/backend/app/services/mime_parser.py +++ b/backend/app/services/mime_parser.py @@ -5,6 +5,8 @@ from email.parser import BytesParser from email.utils import getaddresses, parsedate_to_datetime from html import unescape +from app.services.message_ids import first_message_id, parse_message_id_list + def _addresses(values: list[str]) -> list[dict[str, str]]: return [ @@ -75,8 +77,25 @@ def parse_mime(raw: bytes) -> dict: pass body_text = "\n".join(text_parts).strip() body_html = "\n".join(html_parts).strip() or None + internet_message_id = first_message_id(message.get("Message-ID")) + in_reply_to = first_message_id(message.get("In-Reply-To")) + references = parse_message_id_list(message.get("References")) + raw_headers = { + "message_id": message.get("Message-ID"), + "in_reply_to": message.get("In-Reply-To"), + "references": message.get("References"), + "subject": message.get("Subject"), + "date": message.get("Date"), + "from": message.get("From"), + "to": message.get("To"), + "cc": message.get("Cc"), + "reply_to": message.get("Reply-To"), + } return { - "message_id": (message.get("Message-ID") or "").strip() or None, + "message_id": internet_message_id, + "in_reply_to": in_reply_to, + "references": references, + "raw_headers": raw_headers, "sender": from_value, "to": _addresses(message.get_all("to", [])), "cc": _addresses(message.get_all("cc", [])), diff --git a/backend/app/services/reply_recipients.py b/backend/app/services/reply_recipients.py new file mode 100644 index 0000000..236ae0b --- /dev/null +++ b/backend/app/services/reply_recipients.py @@ -0,0 +1,49 @@ +from app.schemas import MailAddress + + +def _addresses(items: list[dict] | list[MailAddress] | None) -> list[str]: + if not items: + return [] + values: list[str] = [] + for item in items: + if isinstance(item, dict): + address = item.get("address", "") + else: + address = item.address + if address: + values.append(address.lower()) + return values + + +def reply_recipients( + *, + message: dict, + viewer_email: str, + reply_all: bool = False, +) -> tuple[list[str], list[str]]: + viewer = viewer_email.lower() + own_addresses = {viewer} + + reply_to = _addresses(message.get("reply_to")) + sender = _addresses([message.get("sender", {})]) + to = _addresses(message.get("to") or message.get("recipients")) + cc = _addresses(message.get("cc")) + + primary = reply_to or sender + primary = [address for address in primary if address not in own_addresses] + + if not reply_all: + return primary, [] + + recipients = [] + for address in [*primary, *to, *cc]: + if address in own_addresses or address in recipients: + continue + recipients.append(address) + + if not recipients and primary: + recipients = primary + + to_out = recipients[:1] if recipients else [] + cc_out = recipients[1:] if len(recipients) > 1 else [] + return to_out, cc_out diff --git a/backend/app/services/subject_normalize.py b/backend/app/services/subject_normalize.py new file mode 100644 index 0000000..37bb537 --- /dev/null +++ b/backend/app/services/subject_normalize.py @@ -0,0 +1,16 @@ +import re + +_PREFIX_PATTERN = re.compile( + r"^(?:(?:re|fw|fwd)\s*(?:\[\d+\])?\s*:\s*)+", + flags=re.IGNORECASE, +) + + +def normalize_subject(subject: str | None) -> str: + value = (subject or "").strip() + while True: + next_value = _PREFIX_PATTERN.sub("", value, count=1).strip() + if next_value == value: + break + value = next_value + return re.sub(r"\s+", " ", value).strip().lower() diff --git a/backend/app/services/thread_resolver.py b/backend/app/services/thread_resolver.py new file mode 100644 index 0000000..57b0af0 --- /dev/null +++ b/backend/app/services/thread_resolver.py @@ -0,0 +1,252 @@ +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone + +from bson import ObjectId + +from app.services.message_ids import ( + first_message_id, + normalize_message_id, + parse_message_id_list, +) +from app.services.subject_normalize import normalize_subject + +logger = logging.getLogger(__name__) + +FALLBACK_WINDOW = timedelta(days=7) + + +@dataclass(frozen=True) +class ThreadResolution: + thread_id: ObjectId | None + parent_message_id: ObjectId | None + thread_relation: str + thread_resolution_method: str + + +@dataclass(frozen=True) +class KnownMessage: + id: ObjectId + thread_id: ObjectId + internet_message_id: str + account_id: str + sent_at: datetime | None + received_at: datetime | None + created_at: datetime + participants: frozenset[str] + normalized_subject: str + + +def _message_timestamp(message: KnownMessage) -> datetime: + return message.sent_at or message.received_at or message.created_at + + +def resolve_thread_from_known( + *, + account_id: str, + in_reply_to: str | None, + references: list[str], + subject: str, + participants: set[str], + known_by_internet_id: dict[str, KnownMessage], + known_threads: dict[ObjectId, list[KnownMessage]] | None = None, + now: datetime | None = None, +) -> ThreadResolution: + scoped = { + key: value + for key, value in known_by_internet_id.items() + if value.account_id == account_id + } + in_reply = normalize_message_id(in_reply_to) + if in_reply and in_reply in scoped: + parent = scoped[in_reply] + return ThreadResolution( + thread_id=parent.thread_id, + parent_message_id=parent.id, + thread_relation="reply", + thread_resolution_method="in_reply_to", + ) + + reference_ids = [ + normalize_message_id(value) + for value in references + if normalize_message_id(value) + ] + for reference in reversed(reference_ids): + if reference in scoped: + parent = scoped[reference] + return ThreadResolution( + thread_id=parent.thread_id, + parent_message_id=parent.id, + thread_relation="reply", + thread_resolution_method="references", + ) + + fallback = _conservative_fallback( + account_id=account_id, + subject=subject, + participants=participants, + known_by_internet_id=scoped, + known_threads=known_threads or {}, + now=now or datetime.now(timezone.utc), + ) + if fallback is not None: + return fallback + + return ThreadResolution( + thread_id=None, + parent_message_id=None, + thread_relation="new_thread", + thread_resolution_method="new_thread", + ) + + +def _conservative_fallback( + *, + account_id: str, + subject: str, + participants: set[str], + known_by_internet_id: dict[str, KnownMessage], + known_threads: dict[ObjectId, list[KnownMessage]], + now: datetime, +) -> ThreadResolution | None: + normalized = normalize_subject(subject) + if not normalized or not participants: + return None + + candidates: dict[ObjectId, KnownMessage] = {} + for message in known_by_internet_id.values(): + if message.normalized_subject != normalized: + continue + if not participants.intersection(message.participants): + continue + timestamp = _message_timestamp(message) + if timestamp < now - FALLBACK_WINDOW: + continue + candidates[message.thread_id] = message + + if len(candidates) != 1: + return None + + parent = next(iter(candidates.values())) + return ThreadResolution( + thread_id=parent.thread_id, + parent_message_id=parent.id, + thread_relation="unknown", + thread_resolution_method="fallback", + ) + + +def participants_from_parsed(parsed: dict) -> set[str]: + addresses: set[str] = set() + for key in ("sender",): + sender = parsed.get(key) + if isinstance(sender, dict) and sender.get("address"): + addresses.add(sender["address"].lower()) + for key in ("to", "cc", "reply_to"): + for item in parsed.get(key, []) or []: + if isinstance(item, dict) and item.get("address"): + addresses.add(item["address"].lower()) + return addresses + + +async def lookup_known_messages( + database, + account_id: str, + internet_ids: list[str], +) -> dict[str, KnownMessage]: + normalized_ids = [ + value + for value in {normalize_message_id(item) for item in internet_ids} + if value + ] + if not normalized_ids: + return {} + + pipeline = [ + { + "$match": { + "message_id": {"$in": normalized_ids}, + "$or": [ + {"message_scope": "EXTERNAL"}, + {"message_scope": {"$exists": False}}, + ], + } + }, + { + "$lookup": { + "from": "mailbox_entries", + "let": {"message_id": "$_id"}, + "pipeline": [ + { + "$match": { + "$expr": { + "$and": [ + {"$eq": ["$message_id", "$$message_id"]}, + {"$eq": ["$account_id", account_id]}, + ] + } + } + }, + {"$limit": 1}, + ], + "as": "entry", + } + }, + {"$match": {"entry.0": {"$exists": True}}}, + ] + known: dict[str, KnownMessage] = {} + cursor = database.messages.aggregate(pipeline) + async for document in cursor: + internet_id = normalize_message_id(document.get("message_id")) + if not internet_id: + continue + entry = document["entry"][0] + thread_id = entry.get("thread_id") or document.get("thread_id") + if thread_id is None: + continue + participants = participants_from_parsed(document) + known[internet_id] = KnownMessage( + id=document["_id"], + thread_id=thread_id, + internet_message_id=internet_id, + account_id=account_id, + sent_at=document.get("sent_at"), + received_at=document.get("received_at"), + created_at=document.get("created_at"), + participants=frozenset(participants), + normalized_subject=normalize_subject(document.get("subject")), + ) + return known + + +async def resolve_thread( + database, + *, + account_id: str, + in_reply_to: str | None, + references: list[str], + subject: str, + participants: set[str], +) -> ThreadResolution: + search_ids = [ + *([in_reply_to] if in_reply_to else []), + *references, + ] + known = await lookup_known_messages(database, account_id, search_ids) + resolution = resolve_thread_from_known( + account_id=account_id, + in_reply_to=in_reply_to, + references=references, + subject=subject, + participants=participants, + known_by_internet_id=known, + ) + logger.info( + "thread_resolved account_id=%s method=%s relation=%s parent_found=%s", + account_id, + resolution.thread_resolution_method, + resolution.thread_relation, + resolution.parent_message_id is not None, + ) + return resolution diff --git a/backend/app/services/thread_service.py b/backend/app/services/thread_service.py new file mode 100644 index 0000000..4293c35 --- /dev/null +++ b/backend/app/services/thread_service.py @@ -0,0 +1,137 @@ +from datetime import datetime, timezone + +from bson import ObjectId + +from app.services.message_ids import build_references_header, normalize_message_id +from app.services.subject_normalize import normalize_subject +from app.services.thread_resolver import ThreadResolution + + +def message_event_at(message: dict) -> datetime: + return ( + message.get("received_at") + or message.get("sent_at") + or message.get("created_at") + or datetime.now(timezone.utc) + ) + + +async def create_thread( + database, + *, + account_id: str, + subject: str, + message_id: ObjectId, + event_at: datetime, + thread_type: str = "EXTERNAL", +) -> ObjectId: + now = datetime.now(timezone.utc) + document = { + "account_id": account_id, + "thread_type": thread_type, + "normalized_subject": normalize_subject(subject), + "created_at": now, + "updated_at": now, + "last_message_at": event_at, + "latest_message_id": message_id, + "message_count": 1, + } + result = await database.email_threads.insert_one(document) + return result.inserted_id + + +async def message_already_in_thread( + database, + *, + account_id: str, + thread_id: ObjectId, + message_id: ObjectId, +) -> bool: + existing = await database.mailbox_entries.find_one( + { + "account_id": account_id, + "thread_id": thread_id, + "message_id": message_id, + }, + {"_id": 1}, + ) + return existing is not None + + +async def attach_message_to_thread( + database, + *, + account_id: str, + thread_id: ObjectId, + message_id: ObjectId, + subject: str, + event_at: datetime, +) -> None: + if await message_already_in_thread( + database, + account_id=account_id, + thread_id=thread_id, + message_id=message_id, + ): + return + now = datetime.now(timezone.utc) + await database.email_threads.update_one( + {"_id": thread_id, "account_id": account_id}, + { + "$set": { + "updated_at": now, + "last_message_at": event_at, + "latest_message_id": message_id, + "normalized_subject": normalize_subject(subject), + }, + "$inc": {"message_count": 1}, + }, + ) + + +async def ensure_thread_for_message( + database, + *, + account_id: str, + message: dict, + resolution: ThreadResolution, +) -> ObjectId: + message_id = message["_id"] + event_at = message_event_at(message) + subject = message.get("subject", "") + + if resolution.thread_id is not None: + await attach_message_to_thread( + database, + account_id=account_id, + thread_id=resolution.thread_id, + message_id=message_id, + subject=subject, + event_at=event_at, + ) + return resolution.thread_id + + return await create_thread( + database, + account_id=account_id, + subject=subject, + message_id=message_id, + event_at=event_at, + ) + + +def threading_fields_from_parent(parent: dict) -> dict: + parent_internet_id = normalize_message_id(parent.get("message_id")) + parent_references = parent.get("references") or [] + if isinstance(parent_references, str): + parent_references = [parent_references] + references = [] + if parent_references: + references = list(parent_references) + if parent_internet_id: + references = build_references_header(references, parent_internet_id).split() + return { + "in_reply_to": parent_internet_id, + "references": references, + "parent_message_id": parent["_id"], + } diff --git a/backend/tests/test_handoff_evaluator.py b/backend/tests/test_handoff_evaluator.py new file mode 100644 index 0000000..ac6fecc --- /dev/null +++ b/backend/tests/test_handoff_evaluator.py @@ -0,0 +1,33 @@ +from app.services.handoff_evaluator import evaluate_handoff + + +def test_handoff_requires_all_selected_filters(): + email = { + "subject": "Payment approval required", + "snippet": "Please approve this invoice payment today.", + "body_text": "We need your approval for payment of $4,500.", + } + result = evaluate_handoff( + email=email, + labels=["Important", "Finance"], + email_types=["transactional"], + instructions="Forward when approval is required.", + ) + assert result["would_handoff"] is True + assert result["label_matched"] is True + assert result["type_matched"] is True + + +def test_handoff_does_not_match_unrelated_instruction(): + email = { + "subject": "Team lunch", + "snippet": "Where should we eat tomorrow?", + "body_text": "Let me know your preference.", + } + result = evaluate_handoff( + email=email, + labels=[], + email_types=[], + instructions="Forward when a refund requires approval.", + ) + assert result["would_handoff"] is False diff --git a/backend/tests/test_human_handoff_service.py b/backend/tests/test_human_handoff_service.py new file mode 100644 index 0000000..bb8bd91 --- /dev/null +++ b/backend/tests/test_human_handoff_service.py @@ -0,0 +1,124 @@ +import pytest +from bson import ObjectId + +from app.services.human_handoff_service import ( + MESSAGE_SCOPE_EXTERNAL, + MESSAGE_SCOPE_HANDOFF, + resolve_handoff_from_inbound, +) + + +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_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(): + assert MESSAGE_SCOPE_EXTERNAL != MESSAGE_SCOPE_HANDOFF diff --git a/backend/tests/test_reply_recipients.py b/backend/tests/test_reply_recipients.py new file mode 100644 index 0000000..743aa6a --- /dev/null +++ b/backend/tests/test_reply_recipients.py @@ -0,0 +1,30 @@ +from app.services.reply_recipients import reply_recipients + + +def test_reply_prefers_reply_to_header(): + message = { + "sender": {"name": "Sender", "address": "sender@example.com"}, + "reply_to": [{"name": "Support", "address": "support@example.com"}], + "to": [{"name": "Me", "address": "me@gentic.email"}], + "cc": [], + } + to, cc = reply_recipients(message=message, viewer_email="me@gentic.email", reply_all=False) + assert to == ["support@example.com"] + assert cc == [] + + +def test_reply_all_excludes_viewer_and_deduplicates(): + message = { + "sender": {"name": "Sender", "address": "sender@example.com"}, + "to": [ + {"name": "Me", "address": "me@gentic.email"}, + {"name": "Other", "address": "other@example.com"}, + ], + "cc": [ + {"name": "Sender", "address": "sender@example.com"}, + {"name": "Third", "address": "third@example.com"}, + ], + } + to, cc = reply_recipients(message=message, viewer_email="me@gentic.email", reply_all=True) + assert to == ["sender@example.com"] + assert set(cc) == {"other@example.com", "third@example.com"} diff --git a/backend/tests/test_thread_resolver.py b/backend/tests/test_thread_resolver.py new file mode 100644 index 0000000..4a77f0b --- /dev/null +++ b/backend/tests/test_thread_resolver.py @@ -0,0 +1,129 @@ +from datetime import datetime, timezone +from bson import ObjectId + +from app.services.message_ids import build_references_header, normalize_message_id, parse_message_id_list +from app.services.subject_normalize import normalize_subject +from app.services.thread_resolver import KnownMessage, resolve_thread_from_known + + +def _known( + internet_id: str, + thread_id: ObjectId, + account_id: str = "acct-1", + participants: frozenset[str] | None = None, +) -> KnownMessage: + now = datetime.now(timezone.utc) + return KnownMessage( + id=ObjectId(), + thread_id=thread_id, + internet_message_id=normalize_message_id(internet_id) or "", + account_id=account_id, + sent_at=now, + received_at=now, + created_at=now, + participants=participants or frozenset({"sender@example.com"}), + normalized_subject=normalize_subject("Project update"), + ) + + +def test_new_message_creates_new_thread(): + resolution = resolve_thread_from_known( + account_id="acct-1", + in_reply_to=None, + references=[], + subject="Project update", + participants={"sender@example.com"}, + known_by_internet_id={}, + ) + assert resolution.thread_id is None + assert resolution.thread_resolution_method == "new_thread" + assert resolution.thread_relation == "new_thread" + + +def test_direct_reply_matches_in_reply_to(): + thread_id = ObjectId() + parent = _known("", thread_id) + known = {parent.internet_message_id: parent} + resolution = resolve_thread_from_known( + account_id="acct-1", + in_reply_to="", + references=[""], + subject="Re: Project update", + participants={"rahul@example.com"}, + known_by_internet_id=known, + ) + assert resolution.thread_id == thread_id + assert resolution.thread_resolution_method == "in_reply_to" + assert resolution.parent_message_id == parent.id + + +def test_references_fallback_when_parent_missing(): + thread_id = ObjectId() + parent = _known("", thread_id) + known = {parent.internet_message_id: parent} + resolution = resolve_thread_from_known( + account_id="acct-1", + in_reply_to="", + references=["", ""], + subject="Re: Project update", + participants={"rahul@example.com"}, + known_by_internet_id=known, + ) + assert resolution.thread_id == thread_id + assert resolution.thread_resolution_method == "references" + + +def test_fake_re_subject_does_not_thread(): + thread_id = ObjectId() + parent = _known("", thread_id) + known = {parent.internet_message_id: parent} + resolution = resolve_thread_from_known( + account_id="acct-1", + in_reply_to=None, + references=[], + subject="Re: Project update", + participants={"rahul@example.com"}, + known_by_internet_id=known, + ) + assert resolution.thread_id is None + assert resolution.thread_resolution_method == "new_thread" + + +def test_changed_subject_with_valid_reply_headers(): + thread_id = ObjectId() + parent = _known("", thread_id) + known = {parent.internet_message_id: parent} + resolution = resolve_thread_from_known( + account_id="acct-1", + in_reply_to="", + references=[], + subject="Completely changed subject", + participants={"rahul@example.com"}, + known_by_internet_id=known, + ) + assert resolution.thread_id == thread_id + + +def test_cross_tenant_message_is_ignored(): + thread_id = ObjectId() + parent = _known("", thread_id, account_id="other-acct") + known = {parent.internet_message_id: parent} + resolution = resolve_thread_from_known( + account_id="acct-1", + in_reply_to="", + references=[], + subject="Re: Project update", + participants={"rahul@example.com"}, + known_by_internet_id=known, + ) + assert resolution.thread_id is None + + +def test_build_references_header_appends_parent(): + header = build_references_header([""], "") + assert header == " " + + +def test_parse_message_id_list_handles_folded_references(): + values = parse_message_id_list("\r\n ") + assert values == ["", ""] diff --git a/frontend/src/api.js b/frontend/src/api.js index 6bc5b81..112f872 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -101,6 +101,10 @@ function emailQuery(parameters = {}) { export const listEmails = (parameters, token) => request(`/emails${emailQuery(parameters)}`, { token }); +export const listThreads = (parameters, token) => + request(`/threads${emailQuery(parameters)}`, { token }); +export const getThread = (threadId, token) => + request(`/threads/${encodeURIComponent(threadId)}`, { token }); export const getEmail = (emailId, token) => request(`/emails/${encodeURIComponent(emailId)}`, { token }); export const setEmailRead = (emailId, isRead, token) => @@ -131,6 +135,21 @@ export const permanentlyDeleteEmail = (emailId, token) => token, }); +export const listHandoffRules = (token) => + request("/agent/handoff", { token }); +export const createHandoffRule = (payload, token) => + request("/agent/handoff", { method: "POST", payload, token }); +export const updateHandoffRule = (ruleId, payload, token) => + request(`/agent/handoff/${encodeURIComponent(ruleId)}`, { method: "PATCH", payload, token }); +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 }); export const createStickyNote = (payload, token) => diff --git a/frontend/src/components/AgAgentPanel.jsx b/frontend/src/components/AgAgentPanel.jsx index a10c8b9..7e3c160 100644 --- a/frontend/src/components/AgAgentPanel.jsx +++ b/frontend/src/components/AgAgentPanel.jsx @@ -111,10 +111,24 @@ function emailContextActions(email) { return actions.slice(0, 3); } +function threadContextSummary(thread) { + const messages = thread?.messages || []; + if (messages.length <= 1) return null; + const senders = [...new Set(messages.map((message) => message.sender?.name || message.sender?.address).filter(Boolean))]; + const latest = messages[messages.length - 1]; + const snippet = (latest?.body_text || latest?.snippet || "").trim(); + const preview = snippet.split(/[.!?\n]/).find((part) => part.trim().length > 8)?.trim(); + if (preview) { + return `${senders.join(" and ")} exchanged ${messages.length} messages. Latest: “${preview.length > 100 ? `${preview.slice(0, 97)}…` : preview}.”`; + } + return `${senders.join(" and ")} exchanged ${messages.length} messages in this conversation.`; +} + export default function AgAgentPanel({ account, emails, contextEmail, + contextThread, focus, insight, loading = false, @@ -132,6 +146,7 @@ export default function AgAgentPanel({ const allMetricsZero = metricTotal(stats) === 0; const hasQuery = agentQuery.trim().length > 0; const emailContextMode = Boolean(contextEmail) && !insight; + const threadContextMode = Boolean(contextThread?.messages?.length > 1) && !insight; const showOnboarding = !insight && !loading && mailboxEmpty && isBrandNewMailbox; const showQuietEmpty = !insight && !loading && mailboxEmpty && !isBrandNewMailbox; const showFullPanel = !insight && !loading && !mailboxEmpty && !emailContextMode; @@ -187,7 +202,7 @@ export default function AgAgentPanel({ {emailContextMode && (

Summary

-

{emailContextSummary(contextEmail)}

+

{threadContextMode ? threadContextSummary(contextThread) : emailContextSummary(contextEmail)}

Action items

    {emailContextActions(contextEmail).map((item) => ( diff --git a/frontend/src/components/ComposeDock.jsx b/frontend/src/components/ComposeDock.jsx index be7410e..c84c902 100644 --- a/frontend/src/components/ComposeDock.jsx +++ b/frontend/src/components/ComposeDock.jsx @@ -51,7 +51,27 @@ export default function ComposeDock({ account, onSent, agentOpen = true }) { focusRef.current = windowId; }, []); - const openCompose = useCallback(() => { + const openCompose = useCallback((context = null) => { + if (context) { + const id = `compose-${Date.now()}`; + focusRef.current = id; + const draft = { + to: context.to || [], + cc: context.cc || [], + bcc: context.bcc || [], + subject: context.subject || "", + bodyHtml: context.bodyHtml || "", + replyToMessageId: context.replyToMessageId || null, + replyAll: Boolean(context.replyAll), + updatedAt: Date.now(), + }; + localStorage.setItem(`agmail_draft_${id}`, JSON.stringify(draft)); + setWindows((current) => [ + { id, minimized: false, expanded: false, focused: true, lastActive: Date.now() }, + ...current, + ]); + return; + } setWindows((current) => { const blank = current.find((window) => !window.minimized && isDraftEmpty(window.id)); if (blank) { @@ -250,6 +270,6 @@ export default function ComposeDock({ account, onSent, agentOpen = true }) { ); } -export function triggerOpenCompose() { - window.agmailOpenCompose?.(); +export function triggerOpenCompose(context = null) { + window.agmailOpenCompose?.(context); } diff --git a/frontend/src/components/ComposeModal.jsx b/frontend/src/components/ComposeModal.jsx index 7261792..af0b9fd 100644 --- a/frontend/src/components/ComposeModal.jsx +++ b/frontend/src/components/ComposeModal.jsx @@ -196,6 +196,8 @@ export default function ComposeModal({ const [warnings, setWarnings] = useState([]); const [showEmoji, setShowEmoji] = useState(false); const [status, setStatus] = useState({ sending: false, error: "" }); + const [replyToMessageId, setReplyToMessageId] = useState(null); + const [replyAll, setReplyAll] = useState(false); const orgDomain = account.company_domain || account.email.split("@")[1]; const draftKey = draftStorageKey(windowId || account.id); @@ -216,7 +218,9 @@ export default function ComposeModal({ ], [account, fromAddress.email, orgDomain]); const saveDraft = useCallback(() => { - const payload = { to, cc, bcc, subject, bodyHtml, updatedAt: Date.now() }; + const payload = { + to, cc, bcc, subject, bodyHtml, replyToMessageId, replyAll, updatedAt: Date.now(), + }; const empty = !subject && !bodyHtml && !to.length && !cc.length && !bcc.length; if (empty) { clearDraft(windowId || account.id); @@ -243,6 +247,8 @@ export default function ComposeModal({ setBcc(draft.bcc || []); setSubject(draft.subject || ""); setBodyHtml(draft.bodyHtml || ""); + setReplyToMessageId(draft.replyToMessageId || null); + setReplyAll(Boolean(draft.replyAll)); setShowCc(Boolean(draft.cc?.length)); setShowBcc(Boolean(draft.bcc?.length)); if (editorRef.current && draft.bodyHtml) { @@ -385,24 +391,26 @@ export default function ComposeModal({ } setStatus({ sending: true, error: "" }); try { - await sendEmail( - { - recipients: recipientsToPayload(to), - cc: recipientsToPayload(cc), - bcc: recipientsToPayload(bcc), - subject, - body_text: plainTextFromHtml(bodyHtml), - body_html: bodyHtml || null, - }, - sessionStorage.getItem("agmail_access_token"), - ); + const payload = { + recipients: recipientsToPayload(to), + cc: recipientsToPayload(cc), + bcc: recipientsToPayload(bcc), + subject, + body_text: plainTextFromHtml(bodyHtml), + body_html: bodyHtml || null, + }; + if (replyToMessageId) { + payload.reply_to_message_id = replyToMessageId; + payload.reply_all = replyAll; + } + await sendEmail(payload, sessionStorage.getItem("agmail_access_token")); clearDraft(windowId || account.id); flashNotice("Message sent. Undo available for 5 seconds."); onSent?.(); } catch (error) { setStatus({ sending: false, error: error.message }); } - }, [to, cc, bcc, subject, bodyHtml, windowId, account.id, onSent]); + }, [to, cc, bcc, subject, bodyHtml, replyToMessageId, replyAll, windowId, account.id, onSent]); useEffect(() => { function onKeyDown(event) { diff --git a/frontend/src/components/DashboardPage.jsx b/frontend/src/components/DashboardPage.jsx index a2659f6..0d5cede 100644 --- a/frontend/src/components/DashboardPage.jsx +++ b/frontend/src/components/DashboardPage.jsx @@ -12,6 +12,7 @@ import { LogOut, Mail, Menu, + HandHelping, Plug, Send, Settings, @@ -38,6 +39,7 @@ import CompanyOverview from "./CompanyOverview"; import CompanyOverviewSetup, { isOrgSetupComplete } from "./CompanyOverviewSetup"; import ChangePasswordModal from "./ChangePasswordModal"; import ComposeDock, { triggerOpenCompose } from "./ComposeDock"; +import HumanHandoff from "./HumanHandoff"; import MailboxView from "./MailboxView"; const mailItems = [ @@ -63,6 +65,7 @@ const agentItems = [ [Sparkles, "Ask Agent", "ask"], [ListChecks, "Action Items", "actions"], [Sparkles, "Summaries", "summaries"], + [HandHelping, "Human Handoff", "handoff"], [Plug, "Integrations", "integrations"], ]; @@ -204,8 +207,8 @@ export default function DashboardPage() { setMenuOpen(false); } - function openCompose() { - triggerOpenCompose(); + function openCompose(context = null) { + triggerOpenCompose(context); } function emailSent() { @@ -409,7 +412,10 @@ export default function DashboardPage() { {workspaceLabel} )} - {!companyActive && active === "Inbox" && ( + {!companyActive && active === "Inbox" && agentNav === "handoff" && ( + + )} + {!companyActive && active === "Inbox" && agentNav !== "handoff" && ( { + setLoading(true); + setError(""); + try { + const data = await getHandoffConversation(handoffId, token); + setConversation(data); + } catch (err) { + setError(err.message); + } finally { + setLoading(false); + } + }, [handoffId, token]); + + useEffect(() => { + load(); + }, [load]); + + if (loading) { + return

    Loading conversation…

    ; + } + if (error) { + return

    {error}

    ; + } + if (!conversation) return null; + + return ( +
    + + +
    +

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

    +
    + Linked conversation: + {conversation.linked_external.sender} + “{conversation.linked_external.subject}” +
    +
    + +
    + {conversation.messages.map((message) => ( +
    +
    + {message.role === "human" ? "HUMAN OWNER" : "AGENT"} + +
    +
    + {message.body_text || message.snippet} +
    +
    + ))} +
    +
    + ); +} diff --git a/frontend/src/components/HandoffInbox.jsx b/frontend/src/components/HandoffInbox.jsx new file mode 100644 index 0000000..688b8b8 --- /dev/null +++ b/frontend/src/components/HandoffInbox.jsx @@ -0,0 +1,87 @@ +import { ArrowLeftRight } from "lucide-react"; +import { useCallback, useEffect, useState } from "react"; +import { listHandoffConversations } from "../api"; + +function formatRelativeTime(dateValue) { + if (!dateValue) return ""; + const date = new Date(dateValue); + const diffMs = Date.now() - date.getTime(); + const minutes = Math.floor(diffMs / 60000); + if (minutes < 1) return "Just now"; + if (minutes < 60) return `${minutes} min ago`; + const hours = Math.floor(minutes / 60); + if (hours < 24) return `${hours} hour${hours === 1 ? "" : "s"} ago`; + const days = Math.floor(hours / 24); + if (days === 1) return "Yesterday"; + if (days < 7) return `${days} days ago`; + return date.toLocaleDateString(); +} + +export default function HandoffInbox({ onSelect }) { + const token = sessionStorage.getItem("agmail_access_token"); + const [items, setItems] = useState([]); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(""); + + const load = useCallback(async () => { + setLoading(true); + setError(""); + try { + const data = await listHandoffConversations(token); + setItems(data); + } catch (err) { + setError(err.message); + } finally { + setLoading(false); + } + }, [token]); + + useEffect(() => { + load(); + }, [load]); + + return ( +
    +

    Agent ↔ Human Owner conversations

    + {error &&

    {error}

    } + {loading ? ( +

    Loading handoffs…

    + ) : items.length === 0 ? ( +
    + +

    No handoff conversations yet

    +

    + When the Agent hands off a customer conversation, it appears here as a private + Agent ↔ Human thread — separate from your normal Inbox. +

    +
    + ) : ( +
    + {items.map((item) => ( + + ))} +
    + )} +
    + ); +} diff --git a/frontend/src/components/HandoffSettings.jsx b/frontend/src/components/HandoffSettings.jsx new file mode 100644 index 0000000..8bf6352 --- /dev/null +++ b/frontend/src/components/HandoffSettings.jsx @@ -0,0 +1,294 @@ +import { MoreHorizontal, Plus, UserRound } from "lucide-react"; +import { useCallback, useEffect, useState } from "react"; +import { + createHandoffRule, + deleteHandoffRule, + listHandoffRules, + testHandoffRule, + updateHandoffRule, +} from "../api"; +import AgentSpark from "./AgentSpark"; +import HumanHandoffDrawer from "./HumanHandoffDrawer"; +import { buildRuleCardSummary, EMAIL_TYPE_OPTIONS } from "../utils/handoffUtils"; + +function typeLabel(id) { + return EMAIL_TYPE_OPTIONS.find((item) => item.id === id)?.title || id; +} + +export default function HandoffSettings() { + const token = sessionStorage.getItem("agmail_access_token"); + const [rules, setRules] = useState([]); + const [loading, setLoading] = useState(true); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(""); + const [notice, setNotice] = useState(""); + const [drawerOpen, setDrawerOpen] = useState(false); + const [editingRule, setEditingRule] = useState(null); + const [menuOpenId, setMenuOpenId] = useState(null); + const [testOpen, setTestOpen] = useState(false); + const [testDraft, setTestDraft] = useState(null); + const [testSubject, setTestSubject] = useState(""); + const [testBody, setTestBody] = useState(""); + const [testResult, setTestResult] = useState(null); + const [testing, setTesting] = useState(false); + + const loadRules = useCallback(async () => { + setLoading(true); + setError(""); + try { + const data = await listHandoffRules(token); + setRules(data); + } catch (err) { + setError(err.message); + } finally { + setLoading(false); + } + }, [token]); + + useEffect(() => { + loadRules(); + }, [loadRules]); + + function openCreate() { + setEditingRule(null); + setDrawerOpen(true); + } + + function openEdit(rule) { + setEditingRule(rule); + setDrawerOpen(true); + setMenuOpenId(null); + } + + async function handleSave(payload) { + setSaving(true); + setError(""); + try { + if (editingRule) { + await updateHandoffRule(editingRule.id, payload, token); + setNotice("Human Handoff updated."); + } else { + await createHandoffRule(payload, token); + setNotice("Human Handoff created."); + } + setDrawerOpen(false); + setEditingRule(null); + await loadRules(); + window.setTimeout(() => setNotice(""), 2500); + } catch (err) { + setError(err.message); + } finally { + setSaving(false); + } + } + + async function handleDelete(rule) { + if (!window.confirm("Delete this handoff rule?")) return; + try { + await deleteHandoffRule(rule.id, token); + setMenuOpenId(null); + await loadRules(); + } catch (err) { + setError(err.message); + } + } + + function openTest(draft) { + setTestDraft(draft); + setTestSubject(""); + setTestBody(""); + setTestResult(null); + setTestOpen(true); + } + + async function runTest() { + if (!testDraft) return; + setTesting(true); + setError(""); + try { + const result = await testHandoffRule({ + rule_id: editingRule?.id, + owner_email: testDraft.owner_email, + labels: testDraft.labels, + email_types: testDraft.email_types, + instructions: testDraft.instructions, + email: { + subject: testSubject, + snippet: testBody, + body_text: testBody, + }, + }, token); + setTestResult(result); + } catch (err) { + setError(err.message); + } finally { + setTesting(false); + } + } + + return ( +
    +
    +

    + Your Agent handles email. You step in when needed. +

    +

    + 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. +

    + + + The external conversation and Human Handoff remain separate and independently threaded. + +
    + +
    + +
    + + {error &&

    {error}

    } + {notice &&

    {notice}

    } + + {loading ? ( +

    Loading settings…

    + ) : rules.length === 0 ? ( +
    + +

    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. +

    + +
    + ) : ( +
    + {rules.map((rule) => ( +
    +
    + +
    + {rule.owner_name} + {rule.owner_email} +
    + + {rule.active ? "Active" : "Paused"} + +
    +

    {buildRuleCardSummary(rule)}

    + {rule.instructions && ( +
    “{rule.instructions}”
    + )} +
    + +
    + + {menuOpenId === rule.id && ( +
    + + +
    + )} +
    +
    +
    + ))} +
    + )} + + { setDrawerOpen(false); setEditingRule(null); }} + onSave={handleSave} + onTest={(draft) => openTest(draft)} + /> + + {testOpen && ( +
    setTestOpen(false)}> +