human handoffs

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()
+        ],
+    }

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 = [

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,
+    )