feat: gmail api integration

app/api/server.py

@@ -121,6 +121,8 @@ async def lifespan(app: FastAPI):
     await close_maps_service()
     from app.api.v1.gcs import close_gcs_service
     await close_gcs_service()
+    from app.api.v1.gmail import close_gmail_service
+    await close_gmail_service()
     from app.services.media_storage_service import close_storage_service
     await close_storage_service()
     from app.utils.http_utils import close_shared_aiohttp_sessions

app/api/v1/gmail.py

@@ -0,0 +1,1181 @@
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, Header, HTTPException
+
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.models.schemas import (
+    GmailAttachmentContent,
+    GmailBatchDeleteRequest,
+    GmailBatchModifyRequest,
+    GmailComposeMessageRequest,
+    GmailDraftCreateRequest,
+    GmailDraftSendRequest,
+    GmailDraftUpdateRequest,
+    GmailFilterCreateRequest,
+    GmailGenericResponse,
+    GmailInsertMessageRequest,
+    GmailLabelRequest,
+    GmailMessageModifyRequest,
+    GmailRawMessageRequest,
+    GmailRefreshResponse,
+    GmailSendResponse,
+    GmailSettingsUpdateRequest,
+)
+from app.services.gmail_service import (
+    GmailAPIError,
+    GmailCredentials,
+    GmailService,
+)
+
+router = APIRouter(prefix="/google/gmail", tags=["Gmail"])
+_logger = get_logger(__name__)
+_settings = get_settings()
+
+_gmail_service = GmailService()
+
+
+def get_gmail_service() -> GmailService:
+    return _gmail_service
+
+
+async def close_gmail_service() -> None:
+    await _gmail_service.close()
+
+
+def _credentials(
+    x_access_token: Optional[str] = Header(None, alias="X-Access-Token"),
+    x_refresh_token: Optional[str] = Header(None, alias="X-Refresh-Token"),
+    x_client_id: Optional[str] = Header(None, alias="X-Client-Id"),
+    x_client_secret: Optional[str] = Header(None, alias="X-Client-Secret"),
+    x_token_expires_at: Optional[float] = Header(None, alias="X-Token-Expires-At"),
+) -> GmailCredentials:
+    if not x_access_token:
+        raise HTTPException(
+            status_code=401,
+            detail="X-Access-Token header is required.",
+        )
+    return GmailCredentials(
+        access_token=x_access_token,
+        refresh_token=x_refresh_token,
+        client_id=x_client_id,
+        client_secret=x_client_secret,
+        expires_at=x_token_expires_at,
+    )
+
+
+def _http_error(exc: GmailAPIError) -> HTTPException:
+    return HTTPException(status_code=exc.status_code, detail=exc.message)
+
+
+def _elapsed_ms(start: float) -> float:
+    return round((time.perf_counter() - start) * 1000, 2)
+
+
+def _ok(
+    start: float,
+    creds: GmailCredentials,
+    data: Any = None,
+    *,
+    error: Optional[str] = None,
+) -> GmailGenericResponse:
+    return GmailGenericResponse(
+        success=error is None,
+        time_ms=_elapsed_ms(start),
+        data=data,
+        refreshed_access_token=creds.refreshed_access_token,
+        error=error,
+    )
+
+
+def _user_id(delegate: str) -> str:
+    return delegate or "me"
+
+
+def _attachment_content(
+    message_id: str,
+    attachment_id: str,
+    raw: Dict[str, Any],
+) -> GmailAttachmentContent:
+    return GmailAttachmentContent(
+        message_id=message_id,
+        attachment_id=attachment_id,
+        filename=raw.get("filename", ""),
+        mime_type=raw.get("mimeType", ""),
+        data_base64=raw.get("data", ""),
+        size_bytes=int(raw.get("size", 0) or 0),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Service metadata & token lifecycle
+# ---------------------------------------------------------------------------
+
+@router.get("/scopes", response_model=GmailGenericResponse,
+            summary="List all supported Gmail OAuth scopes")
+async def list_scopes(
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    scopes = await service.list_available_scopes()
+    _logger.info("Listed %d Gmail scopes (%.2fms)", len(scopes), _elapsed_ms(start))
+    return GmailGenericResponse(success=True, time_ms=_elapsed_ms(start), data=scopes)
+
+
+@router.post("/token/refresh", response_model=GmailRefreshResponse,
+             summary="Refresh an access token (stateless, returned to the client)")
+async def token_refresh(
+    body: Dict[str, str],
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    client_id = body.get("client_id", "")
+    client_secret = body.get("client_secret", "")
+    refresh_token = body.get("refresh_token", "")
+    if not (client_id and client_secret and refresh_token):
+        raise HTTPException(
+            status_code=400,
+            detail="client_id, client_secret and refresh_token are required.",
+        )
+    try:
+        tokens = await service.refresh_access_token(
+            client_id=client_id,
+            client_secret=client_secret,
+            refresh_token=refresh_token,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail token refreshed (%.2fms)", _elapsed_ms(start))
+    return GmailRefreshResponse(
+        success=True,
+        access_token=tokens.get("access_token"),
+        expires_in=int(tokens.get("expires_in", 0)),
+        token_type=tokens.get("token_type", "Bearer"),
+        scope=tokens.get("scope"),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Profile
+# ---------------------------------------------------------------------------
+
+@router.get("/profile", response_model=GmailGenericResponse,
+            summary="Get the authenticated user's Gmail profile")
+async def get_profile(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_profile(creds)
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail profile fetched (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+# ---------------------------------------------------------------------------
+# Messages
+# ---------------------------------------------------------------------------
+
+@router.get("/messages", response_model=GmailGenericResponse,
+            summary="List messages (supports Gmail search query and pagination)")
+async def list_messages(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    q: Optional[str] = None,
+    label_ids: Optional[str] = None,
+    max_results: Optional[int] = None,
+    page_token: Optional[str] = None,
+    include_spam_trash: bool = False,
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_messages(
+            creds,
+            user_id=user_id or "me",
+            q=q,
+            label_ids=label_ids.split(",") if label_ids else None,
+            max_results=max_results,
+            page_token=page_token,
+            include_spam_trash=include_spam_trash,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail messages listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/messages/{message_id}", response_model=GmailGenericResponse,
+            summary="Get a single message (format: minimal, metadata, full, raw)")
+async def get_message(
+    message_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    format: str = "full",
+    metadata_headers: Optional[str] = None,
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_message(
+            creds,
+            message_id,
+            user_id=user_id or "me",
+            format=format,
+            metadata_headers=metadata_headers.split(",") if metadata_headers else None,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message %s fetched (%.2fms)", message_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/messages/{message_id}/parsed", response_model=GmailGenericResponse,
+            summary="Get a message parsed into headers, bodies and attachment metadata")
+async def get_parsed_message(
+    message_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        parsed = await service.get_parsed_message(creds, message_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message %s parsed (%.2fms)", message_id, _elapsed_ms(start))
+    return _ok(start, creds, parsed.model_dump())
+
+
+@router.get("/messages/{message_id}/attachments/{attachment_id}",
+            response_model=GmailGenericResponse,
+            summary="Get a message attachment's content (base64url)")
+async def get_attachment(
+    message_id: str,
+    attachment_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        raw = await service.get_attachment(
+            creds, message_id, attachment_id, user_id=user_id or "me"
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    content = _attachment_content(message_id, attachment_id, raw)
+    _logger.info(
+        "Gmail attachment %s fetched for message %s (%.2fms)",
+        attachment_id,
+        message_id,
+        _elapsed_ms(start),
+    )
+    return _ok(start, creds, content.model_dump())
+
+
+@router.post("/messages/send", response_model=GmailSendResponse,
+             summary="Send an email built from a structured compose request")
+async def send_composed(
+    body: GmailComposeMessageRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.send_composed(creds, body, user_id=_user_id(body.delegate))
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail email sent (%.2fms)", _elapsed_ms(start))
+    return GmailSendResponse(
+        success=True,
+        time_ms=_elapsed_ms(start),
+        message_id=data.get("id"),
+        thread_id=data.get("threadId"),
+        label_ids=data.get("labelIds"),
+        refreshed_access_token=creds.refreshed_access_token,
+        data=data,
+    )
+
+
+@router.post("/messages/send/raw", response_model=GmailSendResponse,
+             summary="Send a pre-encoded RFC 2822 message")
+async def send_raw(
+    body: GmailRawMessageRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.send_raw(
+            creds,
+            body.raw,
+            user_id=_user_id(body.delegate),
+            thread_id=body.thread_id,
+            labels=body.labels,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail raw message sent (%.2fms)", _elapsed_ms(start))
+    return GmailSendResponse(
+        success=True,
+        time_ms=_elapsed_ms(start),
+        message_id=data.get("id"),
+        thread_id=data.get("threadId"),
+        label_ids=data.get("labelIds"),
+        refreshed_access_token=creds.refreshed_access_token,
+        data=data,
+    )
+
+
+@router.post("/messages", response_model=GmailSendResponse,
+             summary="Insert a message directly into the mailbox (does not send)")
+async def insert_message(
+    body: GmailInsertMessageRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.insert_message(
+            creds,
+            body.raw,
+            user_id=_user_id(body.delegate),
+            label_ids=body.label_ids,
+            internal_date_source=body.internal_date_source,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message inserted (%.2fms)", _elapsed_ms(start))
+    return GmailSendResponse(
+        success=True,
+        time_ms=_elapsed_ms(start),
+        message_id=data.get("id"),
+        thread_id=data.get("threadId"),
+        refreshed_access_token=creds.refreshed_access_token,
+        data=data,
+    )
+
+
+@router.post("/messages/batchModify", response_model=GmailGenericResponse,
+             summary="Add/remove labels across many messages")
+async def batch_modify(
+    body: GmailBatchModifyRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        await service.batch_modify(
+            creds,
+            body.ids,
+            user_id="me",
+            add_label_ids=body.add_label_ids,
+            remove_label_ids=body.remove_label_ids,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail batch modify on %d messages (%.2fms)", len(body.ids), _elapsed_ms(start))
+    return _ok(start, creds, {"modified": len(body.ids)})
+
+
+@router.post("/messages/batchDelete", response_model=GmailGenericResponse,
+             summary="Permanently delete many messages")
+async def batch_delete(
+    body: GmailBatchDeleteRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        await service.batch_delete(creds, body.ids, user_id="me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail batch delete on %d messages (%.2fms)", len(body.ids), _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": len(body.ids)})
+
+
+@router.post("/messages/{message_id}/modify", response_model=GmailGenericResponse,
+             summary="Add/remove labels on a single message (e.g. mark read/unread)")
+async def modify_message(
+    message_id: str,
+    body: GmailMessageModifyRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.modify_message(
+            creds,
+            message_id,
+            user_id=user_id or "me",
+            add_label_ids=body.add_label_ids,
+            remove_label_ids=body.remove_label_ids,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message %s modified (%.2fms)", message_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/messages/{message_id}/trash", response_model=GmailGenericResponse,
+             summary="Move a message to trash")
+async def trash_message(
+    message_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.trash_message(creds, message_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message %s trashed (%.2fms)", message_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/messages/{message_id}/untrash", response_model=GmailGenericResponse,
+             summary="Restore a message from trash")
+async def untrash_message(
+    message_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.untrash_message(creds, message_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message %s untrashed (%.2fms)", message_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.delete("/messages/{message_id}", response_model=GmailGenericResponse,
+               summary="Permanently delete a message")
+async def delete_message(
+    message_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_message(creds, message_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message %s deleted (%.2fms)", message_id, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": message_id})
+
+
+# ---------------------------------------------------------------------------
+# Drafts
+# ---------------------------------------------------------------------------
+
+@router.get("/drafts", response_model=GmailGenericResponse,
+            summary="List drafts")
+async def list_drafts(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    max_results: Optional[int] = None,
+    page_token: Optional[str] = None,
+    q: Optional[str] = None,
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_drafts(
+            creds,
+            user_id=user_id or "me",
+            max_results=max_results,
+            page_token=page_token,
+            q=q,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail drafts listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/drafts/{draft_id}", response_model=GmailGenericResponse,
+            summary="Get a single draft")
+async def get_draft(
+    draft_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    format: str = "full",
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_draft(creds, draft_id, user_id=user_id or "me", format=format)
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail draft %s fetched (%.2fms)", draft_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/drafts", response_model=GmailGenericResponse,
+             summary="Create a draft from a pre-encoded RFC 2822 message")
+async def create_draft(
+    body: GmailDraftCreateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.create_draft(
+            creds,
+            body.raw,
+            user_id=_user_id(body.delegate),
+            thread_id=body.thread_id,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail draft created (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/drafts/{draft_id}", response_model=GmailGenericResponse,
+            summary="Replace an existing draft")
+async def update_draft(
+    draft_id: str,
+    body: GmailDraftUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_draft(
+            creds,
+            draft_id,
+            body.raw,
+            user_id=_user_id(body.delegate),
+            thread_id=body.thread_id,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail draft %s updated (%.2fms)", draft_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/drafts/send", response_model=GmailSendResponse,
+             summary="Send an existing draft")
+async def send_draft(
+    body: GmailDraftSendRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.send_draft(creds, body.draft_id, user_id=_user_id(body.delegate))
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail draft %s sent (%.2fms)", body.draft_id, _elapsed_ms(start))
+    return GmailSendResponse(
+        success=True,
+        time_ms=_elapsed_ms(start),
+        message_id=data.get("id"),
+        thread_id=data.get("threadId"),
+        refreshed_access_token=creds.refreshed_access_token,
+        data=data,
+    )
+
+
+@router.delete("/drafts/{draft_id}", response_model=GmailGenericResponse,
+               summary="Delete a draft")
+async def delete_draft(
+    draft_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_draft(creds, draft_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail draft %s deleted (%.2fms)", draft_id, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": draft_id})
+
+
+# ---------------------------------------------------------------------------
+# Threads
+# ---------------------------------------------------------------------------
+
+@router.get("/threads", response_model=GmailGenericResponse,
+            summary="List threads")
+async def list_threads(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    q: Optional[str] = None,
+    label_ids: Optional[str] = None,
+    max_results: Optional[int] = None,
+    page_token: Optional[str] = None,
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_threads(
+            creds,
+            user_id=user_id or "me",
+            q=q,
+            label_ids=label_ids.split(",") if label_ids else None,
+            max_results=max_results,
+            page_token=page_token,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail threads listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/threads/{thread_id}", response_model=GmailGenericResponse,
+            summary="Get a single thread with all its messages")
+async def get_thread(
+    thread_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    format: str = "full",
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_thread(creds, thread_id, user_id=user_id or "me", format=format)
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail thread %s fetched (%.2fms)", thread_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/threads/{thread_id}/modify", response_model=GmailGenericResponse,
+             summary="Add/remove labels on all messages in a thread")
+async def modify_thread(
+    thread_id: str,
+    body: GmailMessageModifyRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.modify_thread(
+            creds,
+            thread_id,
+            user_id=user_id or "me",
+            add_label_ids=body.add_label_ids,
+            remove_label_ids=body.remove_label_ids,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail thread %s modified (%.2fms)", thread_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/threads/{thread_id}/trash", response_model=GmailGenericResponse,
+             summary="Move all messages in a thread to trash")
+async def trash_thread(
+    thread_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.trash_thread(creds, thread_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail thread %s trashed (%.2fms)", thread_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/threads/{thread_id}/untrash", response_model=GmailGenericResponse,
+             summary="Restore all messages in a thread from trash")
+async def untrash_thread(
+    thread_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.untrash_thread(creds, thread_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail thread %s untrashed (%.2fms)", thread_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.delete("/threads/{thread_id}", response_model=GmailGenericResponse,
+               summary="Permanently delete a thread")
+async def delete_thread(
+    thread_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_thread(creds, thread_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail thread %s deleted (%.2fms)", thread_id, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": thread_id})
+
+
+# ---------------------------------------------------------------------------
+# Labels
+# ---------------------------------------------------------------------------
+
+@router.get("/labels", response_model=GmailGenericResponse,
+            summary="List all labels")
+async def list_labels(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_labels(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail labels listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/labels/{label_id}", response_model=GmailGenericResponse,
+            summary="Get a single label")
+async def get_label(
+    label_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_label(creds, label_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail label %s fetched (%.2fms)", label_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/labels", response_model=GmailGenericResponse,
+             summary="Create a user label")
+async def create_label(
+    body: GmailLabelRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.create_label(creds, body, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail label '%s' created (%.2fms)", body.name, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/labels/{label_id}", response_model=GmailGenericResponse,
+            summary="Update a user label")
+async def update_label(
+    label_id: str,
+    body: GmailLabelRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_label(creds, label_id, body, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail label %s updated (%.2fms)", label_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.delete("/labels/{label_id}", response_model=GmailGenericResponse,
+               summary="Delete a user label")
+async def delete_label(
+    label_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_label(creds, label_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail label %s deleted (%.2fms)", label_id, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": label_id})
+
+
+# ---------------------------------------------------------------------------
+# Settings
+# ---------------------------------------------------------------------------
+
+@router.get("/settings/auto-forwarding", response_model=GmailGenericResponse,
+            summary="Get auto-forwarding settings")
+async def get_auto_forwarding(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_auto_forwarding(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail auto-forwarding fetched (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/settings/auto-forwarding", response_model=GmailGenericResponse,
+            summary="Update auto-forwarding settings")
+async def update_auto_forwarding(
+    body: GmailSettingsUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_auto_forwarding(creds, body.payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail auto-forwarding updated (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/vacation", response_model=GmailGenericResponse,
+            summary="Get vacation responder settings")
+async def get_vacation(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_vacation(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail vacation settings fetched (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/settings/vacation", response_model=GmailGenericResponse,
+            summary="Update vacation responder settings")
+async def update_vacation(
+    body: GmailSettingsUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_vacation(creds, body.payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail vacation settings updated (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/filters", response_model=GmailGenericResponse,
+            summary="List all filters")
+async def list_filters(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_filters(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail filters listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/filters/{filter_id}", response_model=GmailGenericResponse,
+            summary="Get a single filter")
+async def get_filter(
+    filter_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_filter(creds, filter_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail filter %s fetched (%.2fms)", filter_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/settings/filters", response_model=GmailGenericResponse,
+             summary="Create a filter")
+async def create_filter(
+    body: GmailFilterCreateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.create_filter(creds, body, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail filter created (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.delete("/settings/filters/{filter_id}", response_model=GmailGenericResponse,
+               summary="Delete a filter")
+async def delete_filter(
+    filter_id: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_filter(creds, filter_id, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail filter %s deleted (%.2fms)", filter_id, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": filter_id})
+
+
+@router.get("/settings/forwarding-addresses", response_model=GmailGenericResponse,
+            summary="List forwarding addresses")
+async def list_forwarding_addresses(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_forwarding_addresses(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail forwarding addresses listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/forwarding-addresses/{forwarding_email}", response_model=GmailGenericResponse,
+            summary="Get a forwarding address")
+async def get_forwarding_address(
+    forwarding_email: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_forwarding_address(creds, forwarding_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail forwarding address %s fetched (%.2fms)", forwarding_email, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/settings/forwarding-addresses", response_model=GmailGenericResponse,
+             summary="Create a forwarding address")
+async def create_forwarding_address(
+    body: Dict[str, str],
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    forwarding_email = body.get("forwarding_email", "")
+    if not forwarding_email:
+        raise HTTPException(status_code=400, detail="forwarding_email is required.")
+    try:
+        data = await service.create_forwarding_address(creds, forwarding_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail forwarding address %s created (%.2fms)", forwarding_email, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.delete("/settings/forwarding-addresses/{forwarding_email}", response_model=GmailGenericResponse,
+               summary="Delete a forwarding address")
+async def delete_forwarding_address(
+    forwarding_email: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_forwarding_address(creds, forwarding_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail forwarding address %s deleted (%.2fms)", forwarding_email, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": forwarding_email})
+
+
+@router.get("/settings/send-as", response_model=GmailGenericResponse,
+            summary="List send-as aliases")
+async def list_send_as(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_send_as(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail send-as aliases listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/send-as/{send_as_email}", response_model=GmailGenericResponse,
+            summary="Get a send-as alias")
+async def get_send_as(
+    send_as_email: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_send_as(creds, send_as_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail send-as %s fetched (%.2fms)", send_as_email, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/settings/send-as", response_model=GmailGenericResponse,
+             summary="Create a send-as alias")
+async def create_send_as(
+    body: GmailSettingsUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.create_send_as(creds, body.payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail send-as created (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.patch("/settings/send-as/{send_as_email}", response_model=GmailGenericResponse,
+              summary="Update a send-as alias")
+async def update_send_as(
+    send_as_email: str,
+    body: GmailSettingsUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_send_as(creds, send_as_email, body.payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail send-as %s updated (%.2fms)", send_as_email, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.delete("/settings/send-as/{send_as_email}", response_model=GmailGenericResponse,
+               summary="Delete a send-as alias")
+async def delete_send_as(
+    send_as_email: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_send_as(creds, send_as_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail send-as %s deleted (%.2fms)", send_as_email, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": send_as_email})
+
+
+@router.get("/settings/delegates", response_model=GmailGenericResponse,
+            summary="List delegates")
+async def list_delegates(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_delegates(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail delegates listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/delegates/{delegate_email}", response_model=GmailGenericResponse,
+            summary="Get a delegate")
+async def get_delegate(
+    delegate_email: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_delegate(creds, delegate_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail delegate %s fetched (%.2fms)", delegate_email, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/settings/delegates", response_model=GmailGenericResponse,
+             summary="Create a delegate")
+async def create_delegate(
+    body: Dict[str, str],
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    delegate_email = body.get("delegate_email", "")
+    if not delegate_email:
+        raise HTTPException(status_code=400, detail="delegate_email is required.")
+    try:
+        data = await service.create_delegate(creds, delegate_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail delegate %s created (%.2fms)", delegate_email, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.delete("/settings/delegates/{delegate_email}", response_model=GmailGenericResponse,
+               summary="Delete a delegate")
+async def delete_delegate(
+    delegate_email: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.delete_delegate(creds, delegate_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail delegate %s deleted (%.2fms)", delegate_email, _elapsed_ms(start))
+    return _ok(start, creds, {"deleted": delegate_email})

app/api/v1/router.py

@@ -12,6 +12,7 @@ from app.api.v1 import (
     database,
     embeddings,
     gcs,
+    gmail,
     google_maps,
     google_oauth,
     json_extract,
@@ -21,6 +22,7 @@ from app.api.v1 import (
     qr_generator,
     reconcile,
     scheduler,
+    scopes,
     scraper,
     semantic_router,
     sql_validator,
@@ -57,7 +59,9 @@ api_v1_router.include_router(webhook_socket.router, tags=["Webhook / Socket"])
 api_v1_router.include_router(csv_analysis.router, tags=["CSV Analysis"])
 api_v1_router.include_router(google_maps.router, tags=["Google Maps"])
 api_v1_router.include_router(google_oauth.router, tags=["Google OAuth"])
+api_v1_router.include_router(scopes.router, tags=["Google Scopes"])
 api_v1_router.include_router(gcs.router, tags=["Google Cloud Storage"])
+api_v1_router.include_router(gmail.router, tags=["Gmail"])
 api_v1_router.include_router(media_convert.router, tags=["Media-to-Media Conversion"])
 api_v1_router.include_router(json_extract.router, tags=["JSON Extractor"])
 api_v1_router.include_router(keys_extract.router, prefix="/json", tags=["Keys Extractor"])

app/api/v1/scopes/__init__.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from fastapi import APIRouter
+
+from . import detail, list, names, set
+
+router = APIRouter(tags=["Google Scopes"])
+_PREFIX = "/google/scopes"
+router.include_router(list.router, prefix=_PREFIX)
+router.include_router(names.router, prefix=_PREFIX)
+router.include_router(detail.router, prefix=_PREFIX)
+router.include_router(set.router, prefix=_PREFIX)

app/api/v1/scopes/deps.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from app.services.google_scope_service import GoogleScopeService
+
+SOURCE = "google_scopes_data.GOOGLE_APIS (in-memory Python object structure)"
+REFERENCE = "https://developers.google.com/identity/protocols/oauth2/scopes"
+GENERATED = "2026-08-02"
+
+_service = GoogleScopeService()
+
+
+def get_scope_service() -> GoogleScopeService:
+    return _service

app/api/v1/scopes/detail.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import time
+from typing import List, Optional, Union
+
+from fastapi import APIRouter, HTTPException, Query
+
+from app.models.schemas import (
+    GoogleApiScopesResponse,
+    GoogleScopeNamesResponse,
+)
+from app.services.google_scope_service import GoogleScopeError
+
+from .deps import GENERATED, REFERENCE, SOURCE, get_scope_service
+
+router = APIRouter()
+
+
+@router.get("/{api_name}", response_model=Union[GoogleApiScopesResponse, GoogleScopeNamesResponse])
+async def get_google_api_scopes(
+    api_name: str,
+    scope_name: Optional[str] = Query(None, description="Keep scopes whose name or URI contains this value"),
+    permission_level: Optional[List[str]] = Query(None, description="Filter scopes by permission level (repeatable)"),
+    required: Optional[bool] = Query(None, description="Filter scopes by required flag"),
+    recommended: Optional[bool] = Query(None, description="Filter scopes by recommended flag"),
+    scope_name_only: bool = Query(False, description="Return only scope names instead of full scope objects"),
+) -> Union[GoogleApiScopesResponse, GoogleScopeNamesResponse]:
+    """Return a single Google integration's scopes, with optional scope filters."""
+    from .list import _validate_permission_levels
+
+    _validate_permission_levels(permission_level)
+    started = time.perf_counter()
+    service = get_scope_service()
+    try:
+        plans = await service.load_all_plans()
+        state = await service.load_state()
+        data = service.build_data(plans, state)
+    except GoogleScopeError as exc:
+        return GoogleApiScopesResponse(
+            success=False,
+            time_ms=round((time.perf_counter() - started) * 1000, 3),
+            error=exc.message,
+        )
+
+    filtered = service.filter_data(
+        data,
+        api_name=api_name,
+        scope_name=scope_name,
+        permission_level=permission_level,
+        required=required,
+        recommended=recommended,
+        scope_name_only=scope_name_only,
+    )
+    if not filtered:
+        raise HTTPException(status_code=404, detail=f"Unknown API id '{api_name}'")
+
+    entry = filtered[0]
+    if scope_name_only:
+        return GoogleScopeNamesResponse(api_name=entry["id"], scopes=entry["scopes"])
+
+    return GoogleApiScopesResponse(
+        success=True,
+        time_ms=round((time.perf_counter() - started) * 1000, 3),
+        count=1,
+        total=1,
+        page=1,
+        page_size=1,
+        total_pages=1,
+        generated=GENERATED,
+        reference=REFERENCE,
+        source=SOURCE,
+        filters={
+            "api_name": api_name,
+            "scope_name": scope_name,
+            "permission_level": permission_level,
+            "required": required,
+            "recommended": recommended,
+            "scope_name_only": scope_name_only,
+        },
+        data=[entry],
+    )

app/api/v1/scopes/list.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import time
+from typing import List, Optional, Union
+
+from fastapi import APIRouter, HTTPException, Query
+
+from app.models.schemas import (
+    GoogleApiIdsResponse,
+    GoogleApiScopesResponse,
+    GoogleScopeNamesEntry,
+    GoogleScopeNamesListResponse,
+    GoogleScopeNamesResponse,
+)
+from app.services.google_scope_service import GoogleScopeError
+
+from .deps import GENERATED, REFERENCE, SOURCE, get_scope_service
+
+router = APIRouter()
+
+_VALID_PERMISSION_LEVELS = frozenset({"non-sensitive", "sensitive", "restricted"})
+
+
+def _filters_echo(
+    api_name: Optional[str],
+    oauth_only: bool,
+    no_scope: bool,
+    search: Optional[str],
+    scope_name: Optional[str],
+    permission_level: Optional[List[str]],
+    required: Optional[bool],
+    recommended: Optional[bool],
+    scope_name_only: bool,
+    ids_only: bool,
+) -> dict:
+    return {
+        "api_name": api_name,
+        "oauth_only": oauth_only,
+        "no_scope": no_scope,
+        "search": search,
+        "scope_name": scope_name,
+        "permission_level": permission_level,
+        "required": required,
+        "recommended": recommended,
+        "scope_name_only": scope_name_only,
+        "ids_only": ids_only,
+    }
+
+
+def _validate_permission_levels(permission_level: Optional[List[str]]) -> None:
+    if permission_level:
+        bad = [p for p in permission_level if p.strip().lower() not in _VALID_PERMISSION_LEVELS]
+        if bad:
+            raise HTTPException(
+                status_code=422,
+                detail=f"Invalid permission_level values: {', '.join(bad)}. "
+                "Allowed: non-sensitive, sensitive, restricted.",
+            )
+
+
+@router.get("", response_model=Union[GoogleApiScopesResponse, GoogleScopeNamesResponse, GoogleScopeNamesListResponse, GoogleApiIdsResponse])
+async def list_google_scopes(
+    api_name: Optional[str] = Query(None, description="Filter to a single API id (e.g. gmail, youtube, drive)"),
+    oauth_only: bool = Query(False, description="Only APIs that require OAuth scopes"),
+    no_scope: bool = Query(False, description="Only APIs that do not require OAuth scopes"),
+    search: Optional[str] = Query(None, description="Free-text search across API id/name and scope fields"),
+    scope_name: Optional[str] = Query(None, description="Keep scopes whose name or URI contains this value"),
+    permission_level: Optional[List[str]] = Query(None, description="Filter scopes by permission level (repeatable)"),
+    required: Optional[bool] = Query(None, description="Filter scopes by required flag"),
+    recommended: Optional[bool] = Query(None, description="Filter scopes by recommended flag"),
+    scope_name_only: bool = Query(False, description="Return only scope names instead of full scope objects"),
+    ids_only: bool = Query(False, description="Return only the matching API ids"),
+    page: int = Query(1, ge=1, description="Page number"),
+    page_size: int = Query(50, ge=1, le=200, description="Items per page"),
+) -> Union[GoogleApiScopesResponse, GoogleScopeNamesResponse, GoogleScopeNamesListResponse, GoogleApiIdsResponse]:
+    """List all Google integrations with their OAuth scopes and current selection.
+
+    Supports combining every filter: ``api_name``, ``oauth_only``/``no_scope``,
+    ``search``, ``scope_name``, ``permission_level``, ``required``,
+    ``recommended``, ``scope_name_only``, ``ids_only`` and pagination.
+    """
+    _validate_permission_levels(permission_level)
+    started = time.perf_counter()
+    service = get_scope_service()
+    try:
+        plans = await service.load_all_plans()
+        state = await service.load_state()
+        data = service.build_data(plans, state)
+    except GoogleScopeError as exc:
+        return GoogleApiScopesResponse(
+            success=False,
+            time_ms=round((time.perf_counter() - started) * 1000, 3),
+            error=exc.message,
+        )
+
+    filtered = service.filter_data(
+        data,
+        api_name=api_name,
+        oauth_only=oauth_only,
+        no_scope=no_scope,
+        search=search,
+        scope_name=scope_name,
+        permission_level=permission_level,
+        required=required,
+        recommended=recommended,
+        scope_name_only=scope_name_only,
+    )
+
+    if ids_only:
+        return GoogleApiIdsResponse(
+            success=True,
+            time_ms=round((time.perf_counter() - started) * 1000, 3),
+            count=len(filtered),
+            ids=[e["id"] for e in filtered],
+        )
+
+    # Compact single-API shape for api_name + scope_name_only (see example).
+    if scope_name_only and api_name:
+        entry = next((e for e in filtered if e["id"] == api_name), None)
+        if entry is None:
+            raise HTTPException(status_code=404, detail=f"Unknown API id '{api_name}'")
+        return GoogleScopeNamesResponse(api_name=entry["id"], scopes=entry["scopes"])
+
+    if scope_name_only:
+        start = (page - 1) * page_size
+        chunk = filtered[start : start + page_size]
+        return GoogleScopeNamesListResponse(
+            success=True,
+            time_ms=round((time.perf_counter() - started) * 1000, 3),
+            count=len(chunk),
+            data=[GoogleScopeNamesEntry(api_name=e["id"], scopes=e["scopes"]) for e in chunk],
+        )
+
+    total = len(filtered)
+    total_pages = max(1, (total + page_size - 1) // page_size)
+    start = (page - 1) * page_size
+    chunk = filtered[start : start + page_size]
+    return GoogleApiScopesResponse(
+        success=True,
+        time_ms=round((time.perf_counter() - started) * 1000, 3),
+        count=len(chunk),
+        total=total,
+        page=page,
+        page_size=page_size,
+        total_pages=total_pages,
+        generated=GENERATED,
+        reference=REFERENCE,
+        source=SOURCE,
+        filters=_filters_echo(
+            api_name=api_name,
+            oauth_only=oauth_only,
+            no_scope=no_scope,
+            search=search,
+            scope_name=scope_name,
+            permission_level=permission_level,
+            required=required,
+            recommended=recommended,
+            scope_name_only=scope_name_only,
+            ids_only=ids_only,
+        ),
+        data=chunk,
+    )

app/api/v1/scopes/names.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import time
+from typing import List, Optional, Union
+
+from fastapi import APIRouter, HTTPException, Query
+
+from app.models.schemas import (
+    GoogleScopeNamesEntry,
+    GoogleScopeNamesListResponse,
+    GoogleScopeNamesResponse,
+)
+from app.services.google_scope_service import GoogleScopeError
+
+from .deps import get_scope_service
+
+router = APIRouter()
+
+
+@router.get("/names", response_model=Union[GoogleScopeNamesResponse, GoogleScopeNamesListResponse])
+async def list_scope_names(
+    api_name: Optional[str] = Query(None, description="Filter to a single API id"),
+    oauth_only: bool = Query(False, description="Only APIs that require OAuth scopes"),
+    no_scope: bool = Query(False, description="Only APIs that do not require OAuth scopes"),
+    search: Optional[str] = Query(None, description="Free-text search across API id/name and scope fields"),
+    scope_name: Optional[str] = Query(None, description="Keep scopes whose name or URI contains this value"),
+    permission_level: Optional[List[str]] = Query(None, description="Filter scopes by permission level (repeatable)"),
+    required: Optional[bool] = Query(None, description="Filter scopes by required flag"),
+    recommended: Optional[bool] = Query(None, description="Filter scopes by recommended flag"),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200),
+) -> Union[GoogleScopeNamesResponse, GoogleScopeNamesListResponse]:
+    """Return only the scope names for the matching APIs."""
+    from .list import _validate_permission_levels
+
+    _validate_permission_levels(permission_level)
+    started = time.perf_counter()
+    service = get_scope_service()
+    try:
+        plans = await service.load_all_plans()
+        state = await service.load_state()
+        data = service.build_data(plans, state)
+    except GoogleScopeError as exc:
+        return GoogleScopeNamesListResponse(
+            success=False,
+            time_ms=round((time.perf_counter() - started) * 1000, 3),
+            error=exc.message,
+        )
+
+    filtered = service.filter_data(
+        data,
+        api_name=api_name,
+        oauth_only=oauth_only,
+        no_scope=no_scope,
+        search=search,
+        scope_name=scope_name,
+        permission_level=permission_level,
+        required=required,
+        recommended=recommended,
+        scope_name_only=True,
+    )
+
+    if api_name:
+        entry = next((e for e in filtered if e["id"] == api_name), None)
+        if entry is None:
+            raise HTTPException(status_code=404, detail=f"Unknown API id '{api_name}'")
+        return GoogleScopeNamesResponse(api_name=entry["id"], scopes=entry["scopes"])
+
+    start = (page - 1) * page_size
+    chunk = filtered[start : start + page_size]
+    return GoogleScopeNamesListResponse(
+        success=True,
+        time_ms=round((time.perf_counter() - started) * 1000, 3),
+        count=len(chunk),
+        data=[GoogleScopeNamesEntry(api_name=e["id"], scopes=e["scopes"]) for e in chunk],
+    )

app/api/v1/scopes/set.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import time
+
+from fastapi import APIRouter, HTTPException
+
+from app.models.schemas import (
+    GoogleApiScopesResponse,
+    GoogleSetScopesRequest,
+)
+from app.services.google_scope_service import GoogleScopeError
+
+from .deps import GENERATED, REFERENCE, SOURCE, get_scope_service
+
+router = APIRouter()
+
+
+@router.put("", response_model=GoogleApiScopesResponse)
+async def set_google_scopes(body: GoogleSetScopesRequest) -> GoogleApiScopesResponse:
+    """Set the current OAuth scopes for one or more Google integrations.
+
+    The selection is validated against the catalog and stored in memory as a
+    Python object structure (resets on restart). Pass ``reset: true`` to restore
+    recommended defaults.
+    """
+    started = time.perf_counter()
+    service = get_scope_service()
+    try:
+        state = await service.set_current_scopes(
+            updates=body.apis or {}, reset=body.reset
+        )
+        plans = await service.load_all_plans()
+        data = service.build_data(plans, state)
+        return GoogleApiScopesResponse(
+            success=True,
+            time_ms=round((time.perf_counter() - started) * 1000, 3),
+            count=len(data),
+            total=len(data),
+            generated=GENERATED,
+            reference=REFERENCE,
+            source=SOURCE,
+            data=data,
+        )
+    except GoogleScopeError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc

app/config.py

@@ -124,6 +124,20 @@ class Settings(BaseSettings):
     google_oauth_state_ttl_minutes: int = 10
     google_oauth_jwks_ttl_seconds: int = 3600
 
+    # Gmail API settings
+    gmail_api_base_url: str = "https://gmail.googleapis.com/gmail/v1"
+    gmail_batch_url: str = "https://gmail.googleapis.com/batch/gmail/v1"
+    gmail_timeout: float = 30.0
+    gmail_max_retries: int = 3
+    gmail_base_backoff_seconds: float = 1.0
+    gmail_max_backoff_seconds: float = 32.0
+    gmail_retryable_statuses: str = "429,500,502,503,504"
+    gmail_max_batch_size: int = 50
+    gmail_max_attachment_bytes: int = 25 * 1024 * 1024
+    gmail_max_payload_bytes: int = 35 * 1024 * 1024
+    gmail_default_scope: str = "https://www.googleapis.com/auth/gmail.modify"
+    gmail_refresh_buffer_seconds: int = 60
+
     # Media-to-Media conversion settings
     media_output_dir: str = "./data/media-convert"
     media_max_workers: int = 4

app/models/schemas.py

@@ -959,6 +959,100 @@ class GoogleOAuthVerifyResponse(BaseModel):
     error: Optional[str] = None
 
 
+# ---------------------------------------------------------------------------
+# Google per-API scope catalog (single endpoint to view/set current scopes)
+# ---------------------------------------------------------------------------
+
+class GoogleScopeDetail(BaseModel):
+    """Metadata for a single OAuth scope (from the in-memory scope catalog)."""
+
+    name: Optional[str] = Field(None, description="Short scope name")
+    uri: Optional[str] = Field(None, description="Full OAuth scope URI")
+    description: Optional[str] = Field(None, description="What the scope allows")
+    permission_level: Optional[str] = Field(
+        None, description="non-sensitive | sensitive | restricted"
+    )
+    required: Optional[bool] = Field(None, description="Whether the scope is required")
+    recommended: Optional[bool] = Field(None, description="Whether the scope is the recommended default")
+    reason: Optional[str] = Field(None, description="Why the scope may be needed")
+    least_privilege_recommendation: Optional[str] = Field(
+        None, description="Least-privilege guidance for this scope"
+    )
+
+
+class GoogleApiScopeEntry(BaseModel):
+    """Scope catalog entry for a single Google integration."""
+
+    id: Optional[str] = Field(None, description="Stable API identifier (file name)")
+    api: Optional[str] = Field(None, description="Display name of the Google API")
+    version: Optional[str] = Field(None, description="API version (e.g. v3)")
+    base_url: Optional[str] = Field(None, description="Google API base URL")
+    batch_endpoint: Optional[str] = Field(None, description="Batch endpoint, if any")
+    auth_model: Optional[str] = Field(None, description="Authentication model (OAuth / API key / service account)")
+    scope_required: Optional[bool] = Field(None, description="Whether the API needs OAuth scopes")
+    docs_url: Optional[str] = Field(None, description="Official Google docs reference")
+    info: Optional[str] = Field(None, description="Extra note for APIs without scopes")
+    current_scopes: List[str] = Field(default_factory=list, description="Currently selected scopes")
+    default_scopes: List[str] = Field(default_factory=list, description="Recommended default scopes")
+    scopes: List[GoogleScopeDetail] = Field(default_factory=list, description="All documented scopes")
+    recommended_combinations: Dict[str, List[str]] = Field(
+        default_factory=dict, description="Commonly used scope combinations"
+    )
+
+
+class GoogleApiScopesResponse(BaseModel):
+    success: bool
+    time_ms: float
+    count: int = 0
+    total: int = 0
+    page: int = 1
+    page_size: int = 50
+    total_pages: int = 1
+    generated: Optional[str] = None
+    reference: Optional[str] = None
+    source: Optional[str] = None
+    filters: Optional[Dict[str, Any]] = None
+    data: List[GoogleApiScopeEntry] = Field(default_factory=list)
+    error: Optional[str] = None
+
+
+class GoogleScopeNamesResponse(BaseModel):
+    """Compact response for ``scope_name_only=true`` with a single API."""
+
+    api_name: str
+    scopes: List[str] = Field(default_factory=list)
+
+
+class GoogleScopeNamesEntry(BaseModel):
+    api_name: str
+    scopes: List[str] = Field(default_factory=list)
+
+
+class GoogleScopeNamesListResponse(BaseModel):
+    success: bool
+    time_ms: float
+    count: int = 0
+    data: List[GoogleScopeNamesEntry] = Field(default_factory=list)
+    error: Optional[str] = None
+
+
+class GoogleApiIdsResponse(BaseModel):
+    success: bool
+    time_ms: float
+    count: int = 0
+    ids: List[str] = Field(default_factory=list)
+    error: Optional[str] = None
+
+
+class GoogleSetScopesRequest(BaseModel):
+    """Set the current scopes for one or more Google integrations."""
+
+    apis: Optional[Dict[str, List[str]]] = Field(
+        None, description="Map of API id to the list of scopes to select"
+    )
+    reset: bool = Field(False, description="Restore all APIs to their recommended defaults")
+
+
 # ---------------------------------------------------------------------------
 # Google Maps Static API
 # ---------------------------------------------------------------------------
@@ -1639,3 +1733,226 @@ class MediaImageUrlRequest(BaseModel):
         if not v.startswith(("http://", "https://")):
             raise ValueError("Only http/https URLs are supported.")
         return v
+
+
+# ---------------------------------------------------------------------------
+# Gmail API (stateless, client-supplied OAuth credentials)
+# ---------------------------------------------------------------------------
+
+class GmailScope(Enum):
+    """All documented Gmail OAuth scopes (see developers.google.com/gmail/api/auth/scopes)."""
+
+    READONLY = "https://www.googleapis.com/auth/gmail.readonly"
+    MODIFY = "https://www.googleapis.com/auth/gmail.modify"
+    SEND = "https://www.googleapis.com/auth/gmail.send"
+    COMPOSE = "https://www.googleapis.com/auth/gmail.compose"
+    METADATA = "https://www.googleapis.com/auth/gmail.metadata"
+    INSERT = "https://www.googleapis.com/auth/gmail.insert"
+    LABELS = "https://www.googleapis.com/auth/gmail.labels"
+    SETTINGS_BASIC = "https://www.googleapis.com/auth/gmail.settings.basic"
+    SETTINGS_SHARING = "https://www.googleapis.com/auth/gmail.settings.sharing"
+    MAIL_GOOGLE = "https://mail.google.com/"
+    ADDONS_CURRENT_ACTION = "https://www.googleapis.com/auth/gmail.addons.current.action.compose"
+    ADDONS_CURRENT_MESSAGE_METADATA = "https://www.googleapis.com/auth/gmail.addons.current.message.metadata"
+    ADDONS_CURRENT_MESSAGE_READONLY = "https://www.googleapis.com/auth/gmail.addons.current.message.readonly"
+    ADDONS_CURRENT_SAFE_DOWNLOAD = "https://www.googleapis.com/auth/gmail.addons.current.message.safe_download"
+
+    @property
+    def permission_level(self) -> str:
+        if self in (GmailScope.MAIL_GOOGLE, GmailScope.SETTINGS_SHARING):
+            return "restricted"
+        if self in (
+            GmailScope.COMPOSE,
+            GmailScope.METADATA,
+            GmailScope.READONLY,
+            GmailScope.MODIFY,
+            GmailScope.INSERT,
+            GmailScope.SETTINGS_BASIC,
+        ):
+            return "sensitive"
+        return "non-sensitive"
+
+
+class GmailComposeAttachment(BaseModel):
+    """An email attachment encoded in base64 (max ~25 MB per file)."""
+
+    filename: str = Field(..., min_length=1, max_length=255, description="Attachment file name")
+    content_base64: str = Field(..., min_length=1, description="Base64 (standard) encoded file content")
+    mime_type: Optional[str] = Field(None, description="Explicit MIME type; auto-detected from extension if omitted")
+
+
+class GmailComposeMessageRequest(BaseModel):
+    """Structured email used to build an RFC 822 MIME message before sending."""
+
+    subject: str = Field(..., min_length=1, max_length=998, description="Email subject")
+    body: str = Field("", description="Plain-text body")
+    to: List[str] = Field(..., min_length=1, description="Primary recipients")
+    cc: Optional[List[str]] = Field(None, description="Carbon-copy recipients")
+    bcc: Optional[List[str]] = Field(None, description="Blind carbon-copy recipients")
+    reply_to: Optional[List[str]] = Field(None, description="Reply-To addresses")
+    from_name: Optional[str] = Field(None, max_length=255, description="Display name for the mailbox sender")
+    html_body: Optional[str] = Field(None, description="HTML body; used when provided, otherwise a plain body is sent")
+    attachments: Optional[List[GmailComposeAttachment]] = Field(None, max_length=20, description="Attachments (max 20)")
+    thread_id: Optional[str] = Field(None, description="Gmail thread ID to continue")
+    labels: Optional[List[str]] = Field(None, description="Label IDs to apply after sending")
+    headers: Optional[Dict[str, str]] = Field(None, description="Custom MIME headers")
+    in_reply_to: Optional[str] = Field(None, description="Message-ID being replied to")
+    references: Optional[List[str]] = Field(None, description="References header values for threading")
+    delegate: str = Field("me", description="Mailbox to act on (default 'me' = authenticated user)")
+
+    @field_validator("attachments")
+    @classmethod
+    def _limit_attachments(cls, v: Optional[List[GmailComposeAttachment]]) -> Optional[List[GmailComposeAttachment]]:
+        if v is not None and len(v) > 20:
+            raise ValueError("Maximum 20 attachments per email.")
+        return v
+
+
+class GmailRawMessageRequest(BaseModel):
+    """Send an already-encoded RFC 2822 message."""
+
+    raw: str = Field(..., min_length=1, description="Base64url-encoded RFC 2822 message")
+    thread_id: Optional[str] = Field(None, description="Thread ID to associate the message with")
+    labels: Optional[List[str]] = Field(None, description="Label IDs to apply after sending")
+    delegate: str = Field("me", description="Mailbox to act on (default 'me')")
+
+
+class GmailMessageModifyRequest(BaseModel):
+    add_label_ids: Optional[List[str]] = Field(None, description="Label IDs to add")
+    remove_label_ids: Optional[List[str]] = Field(None, description="Label IDs to remove")
+
+
+class GmailBatchModifyRequest(BaseModel):
+    ids: List[str] = Field(..., min_length=1, max_length=1000, description="Message IDs to modify")
+    add_label_ids: Optional[List[str]] = Field(None, description="Label IDs to add to each message")
+    remove_label_ids: Optional[List[str]] = Field(None, description="Label IDs to remove from each message")
+
+
+class GmailBatchDeleteRequest(BaseModel):
+    ids: List[str] = Field(..., min_length=1, max_length=1000, description="Message IDs to permanently delete")
+
+
+class GmailInsertMessageRequest(BaseModel):
+    raw: str = Field(..., min_length=1, description="Base64url-encoded RFC 2822 message")
+    label_ids: Optional[List[str]] = Field(None, description="Label IDs to apply on insert")
+    internal_date_source: Optional[str] = Field(None, pattern="^(dateHeader|receivedTime)$", description="Source for internal date")
+    delegate: str = Field("me", description="Mailbox to act on (default 'me')")
+
+
+class GmailDraftCreateRequest(BaseModel):
+    raw: str = Field(..., min_length=1, description="Base64url-encoded RFC 2822 message")
+    thread_id: Optional[str] = Field(None, description="Thread ID to associate the draft with")
+    delegate: str = Field("me", description="Mailbox to act on (default 'me')")
+
+
+class GmailDraftUpdateRequest(BaseModel):
+    raw: str = Field(..., min_length=1, description="Base64url-encoded RFC 2822 message")
+    thread_id: Optional[str] = Field(None, description="Thread ID to associate the draft with")
+    delegate: str = Field("me", description="Mailbox to act on (default 'me')")
+
+
+class GmailDraftSendRequest(BaseModel):
+    draft_id: str = Field(..., min_length=1, description="ID of the draft to send")
+    delegate: str = Field("me", description="Mailbox to act on (default 'me')")
+
+
+class GmailLabelRequest(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255, description="Label name")
+    label_list_visibility: Optional[str] = Field(None, pattern="^(labelShow|labelHide|labelShowIfUnread)$", description="Label list visibility")
+    message_list_visibility: Optional[str] = Field(None, pattern="^(show|hide)$", description="Message list visibility")
+    color: Optional[Dict[str, str]] = Field(None, description="Label color (background/text color hex values)")
+
+
+class GmailMessageAttachmentMeta(BaseModel):
+    attachment_id: str = Field(..., description="Gmail attachment ID")
+    filename: str = Field("", description="Attachment file name")
+    mime_type: str = Field("", description="Attachment MIME type")
+    size_bytes: int = Field(0, description="Decoded size in bytes")
+
+
+class GmailParsedMessage(BaseModel):
+    """Convenience view of a message with headers, bodies and attachment metadata."""
+
+    id: str = Field("", description="Message ID")
+    thread_id: str = Field("", description="Thread ID")
+    label_ids: List[str] = Field(default_factory=list, description="Applied label IDs")
+    snippet: str = Field("", description="Snippet of the message body")
+    history_id: Optional[str] = Field(None, description="History ID for sync")
+    headers: Dict[str, str] = Field(default_factory=dict, description="All message headers")
+    sender: Optional[str] = Field(None, description="From header value")
+    to: str = Field("", description="To header value")
+    cc: str = Field("", description="Cc header value")
+    bcc: str = Field("", description="Bcc header value")
+    subject: str = Field("", description="Subject header value")
+    date: Optional[str] = Field(None, description="Date header value")
+    plain_body: Optional[str] = Field(None, description="Decoded plain-text body")
+    html_body: Optional[str] = Field(None, description="Decoded HTML body")
+    attachments: List[GmailMessageAttachmentMeta] = Field(default_factory=list, description="Attachment metadata")
+
+
+class GmailAttachmentContent(BaseModel):
+    message_id: str = Field(..., description="Parent message ID")
+    attachment_id: str = Field(..., description="Attachment ID")
+    filename: str = Field("", description="Attachment file name")
+    mime_type: str = Field("", description="Attachment MIME type")
+    data_base64: str = Field("", description="Base64url-encoded attachment bytes")
+    size_bytes: int = Field(0, description="Decoded size in bytes")
+
+
+class GmailSettingsUpdateRequest(BaseModel):
+    """Generic passthrough for heterogeneous settings endpoints."""
+
+    payload: Dict[str, Any] = Field(..., description="Full settings resource body (see Gmail API reference)")
+
+    @field_validator("payload")
+    @classmethod
+    def _non_empty(cls, v: Dict[str, Any]) -> Dict[str, Any]:
+        if not v:
+            raise ValueError("payload must not be empty.")
+        return v
+
+
+class GmailFilterCreateRequest(BaseModel):
+    criteria: Dict[str, Any] = Field(default_factory=dict, description="Filter criteria (from, to, subject, query, etc.)")
+    action: Dict[str, Any] = Field(default_factory=dict, description="Filter actions (addLabelIds, forward, etc.)")
+
+
+class GmailSendAsRequest(BaseModel):
+    display_name: Optional[str] = Field(None, description="Display name shown to recipients")
+    is_default: Optional[bool] = Field(None, description="Whether this is the default send-as alias")
+    reply_to_address: Optional[str] = Field(None, description="Reply-To address")
+    signature: Optional[str] = Field(None, description="Email signature (HTML)")
+
+
+class GmailGenericResponse(BaseModel):
+    """Standard Gmail operation envelope."""
+
+    success: bool
+    time_ms: float = 0.0
+    data: Optional[Any] = Field(None, description="Raw Gmail API payload")
+    refreshed_access_token: Optional[str] = Field(None, description="New access token when an automatic refresh occurred; persist it client-side")
+    error: Optional[str] = Field(None, description="Human-readable error when success is false")
+
+
+class GmailSendResponse(BaseModel):
+    """Result of sending/inserting a message."""
+
+    success: bool
+    time_ms: float = 0.0
+    message_id: Optional[str] = Field(None, description="Created message ID")
+    thread_id: Optional[str] = Field(None, description="Created thread ID")
+    label_ids: Optional[List[str]] = Field(None, description="Labels applied after send")
+    refreshed_access_token: Optional[str] = Field(None, description="New access token when an automatic refresh occurred")
+    data: Optional[Any] = Field(None, description="Raw Gmail API payload")
+    error: Optional[str] = Field(None, description="Human-readable error when success is false")
+
+
+class GmailRefreshResponse(BaseModel):
+    """Stateless token refresh result returned to the client for its own storage."""
+
+    success: bool
+    access_token: Optional[str] = None
+    expires_in: Optional[int] = None
+    token_type: Optional[str] = None
+    scope: Optional[str] = None
+    error: Optional[str] = None

app/services/gmail_service.py

@@ -0,0 +1,905 @@
+from __future__ import annotations
+
+import asyncio
+import random
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.models.schemas import (
+    GmailComposeMessageRequest,
+    GmailFilterCreateRequest,
+    GmailLabelRequest,
+    GmailParsedMessage,
+)
+from app.utils.gmail_utils import (
+    GmailMessageValidationError,
+    build_mime_message,
+    parse_message,
+)
+from app.utils.http_utils import SharedAsyncClient
+
+_logger = get_logger(__name__)
+_settings = get_settings()
+
+_RETRYABLE_STATUS = frozenset({
+    int(code)
+    for code in _settings.gmail_retryable_statuses.split(",")
+    if code.strip().isdigit()
+})
+
+
+class GmailAPIError(Exception):
+    """Raised for upstream Gmail API failures that map to a client-facing error."""
+
+    def __init__(self, message: str, status_code: int = 400, *, reason: str = "") -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+        self.reason = reason
+
+
+@dataclass
+class GmailCredentials:
+    """Client-supplied OAuth credentials (never persisted by this backend).
+
+    ``refreshed_access_token`` is populated whenever the service transparently
+    refreshes the access token, so the stateless client can persist it itself.
+    """
+
+    access_token: str
+    refresh_token: Optional[str] = None
+    client_id: Optional[str] = None
+    client_secret: Optional[str] = None
+    expires_at: Optional[float] = None
+    refreshed_access_token: Optional[str] = field(default=None, repr=False)
+
+    @property
+    def can_refresh(self) -> bool:
+        return bool(self.refresh_token and self.client_id and self.client_secret)
+
+    @property
+    def needs_proactive_refresh(self) -> bool:
+        if self.expires_at is None:
+            return False
+        return time.time() >= self.expires_at - _settings.gmail_refresh_buffer_seconds
+
+
+class GmailService:
+    """Stateless, fully-async abstraction over the Gmail REST API.
+
+    Credentials are supplied by the caller on every invocation. Access tokens
+    are transparently refreshed against Google's token endpoint when a
+    ``refresh_token`` + ``client_id`` + ``client_secret`` are provided, and the
+    resulting token is surfaced via ``GmailCredentials.refreshed_access_token``.
+    """
+
+    def __init__(self) -> None:
+        self._http = SharedAsyncClient(
+            timeout=httpx.Timeout(_settings.gmail_timeout),
+            follow_redirects=False,
+        )
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        return await self._http.get()
+
+    async def close(self) -> None:
+        await self._http.close()
+
+    # ------------------------------------------------------------------
+    # Core request pipeline: auth, retry, backoff, error mapping
+    # ------------------------------------------------------------------
+
+    async def _request(
+        self,
+        creds: GmailCredentials,
+        method: str,
+        path: str,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+        json_body: Optional[Any] = None,
+    ) -> Any:
+        """Send a Gmail API request with proactive + on-401 refresh and retries."""
+        if creds.needs_proactive_refresh:
+            await self._refresh(creds, context=f"{method} {path}")
+
+        refresh_done = False
+        attempt = 0
+        while True:
+            response = await self._send(
+                creds, method, path, params=params, json_body=json_body
+            )
+            if 200 <= response.status_code < 300:
+                return self._decode(response)
+
+            if response.status_code == 401 and creds.can_refresh and not refresh_done:
+                _logger.warning(
+                    "Gmail 401 for %s %s; refreshing access token once",
+                    method,
+                    path,
+                )
+                await self._refresh(creds, context=f"{method} {path}")
+                refresh_done = True
+                continue
+
+            if response.status_code in _RETRYABLE_STATUS and attempt < _settings.gmail_max_retries:
+                await self._backoff(response, attempt)
+                attempt += 1
+                continue
+
+            raise self._map_error(response, creds)
+
+    async def _send(
+        self,
+        creds: GmailCredentials,
+        method: str,
+        path: str,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+        json_body: Optional[Any] = None,
+    ) -> httpx.Response:
+        client = await self._get_client()
+        url = f"{_settings.gmail_api_base_url}{path}"
+        headers = {
+            "Authorization": f"Bearer {creds.access_token}",
+            "Accept": "application/json",
+        }
+        try:
+            return await client.request(
+                method,
+                url,
+                params=params,
+                json=json_body,
+                headers=headers,
+            )
+        except httpx.TimeoutException as exc:
+            raise GmailAPIError(
+                f"Gmail API request timed out for {method} {path}.", status_code=504
+            ) from exc
+        except httpx.RequestError as exc:
+            raise GmailAPIError(
+                f"Failed to reach Gmail API: {exc}", status_code=502
+            ) from exc
+
+    @staticmethod
+    def _decode(response: httpx.Response) -> Any:
+        if not response.content:
+            return None
+        try:
+            return response.json()
+        except ValueError:
+            return response.text
+
+    async def _backoff(self, response: httpx.Response, attempt: int) -> None:
+        retry_after = response.headers.get("Retry-After")
+        delay: Optional[float] = None
+        if retry_after:
+            try:
+                delay = float(retry_after)
+            except (TypeError, ValueError):
+                delay = None
+        if delay is None:
+            base = _settings.gmail_base_backoff_seconds * (2 ** attempt)
+            delay = min(base, _settings.gmail_max_backoff_seconds) * (0.5 + random.random())
+        delay = min(delay, _settings.gmail_max_backoff_seconds + 5.0)
+        _logger.warning(
+            "Gmail transient HTTP %s; retrying in %.2fs (attempt %d/%d)",
+            response.status_code,
+            delay,
+            attempt + 1,
+            _settings.gmail_max_retries,
+        )
+        await asyncio.sleep(delay)
+
+    def _map_error(self, response: httpx.Response, creds: GmailCredentials) -> GmailAPIError:
+        status = response.status_code
+        reason = ""
+        detail = ""
+        try:
+            body = response.json()
+            error = body.get("error") or {}
+            if isinstance(error, dict):
+                detail = error.get("message", "")
+                errors = error.get("errors") or []
+                if errors and isinstance(errors[0], dict):
+                    reason = errors[0].get("reason", "")
+        except Exception:
+            body = {}
+            detail = response.text[:500]
+
+        _logger.error(
+            "Gmail API error: HTTP %s reason=%s detail=%s", status, reason, detail
+        )
+
+        if status == 401:
+            if creds.can_refresh:
+                return GmailAPIError(
+                    "Access token is invalid or expired and could not be refreshed. "
+                    "Re-authorize via POST /api/v1/google/oauth/auth-url.",
+                    status_code=401,
+                    reason=reason,
+                )
+            return GmailAPIError(
+                "Access token is invalid or expired. Refresh it via "
+                "POST /api/v1/google/oauth/refresh (or /api/v1/gmail/token/refresh) "
+                "and retry, or supply a fresh access token.",
+                status_code=401,
+                reason=reason,
+            )
+        if status == 403 and reason == "PERMISSION_DENIED":
+            return GmailAPIError(
+                "Permission denied. The access token may be missing the required "
+                "Gmail scope (see /api/v1/google/gmail/scopes).",
+                status_code=403,
+                reason=reason,
+            )
+        if status == 429:
+            return GmailAPIError(
+                "Gmail API rate limit exceeded. Please retry after a short delay.",
+                status_code=429,
+                reason=reason,
+            )
+        if status in _RETRYABLE_STATUS:
+            return GmailAPIError(
+                detail or f"Gmail API service error (HTTP {status}).",
+                status_code=status,
+                reason=reason,
+            )
+        return GmailAPIError(
+            detail or f"Gmail API error (HTTP {status}).",
+            status_code=status,
+            reason=reason,
+        )
+
+    # ------------------------------------------------------------------
+    # Token lifecycle
+    # ------------------------------------------------------------------
+
+    async def refresh_access_token(
+        self,
+        *,
+        client_id: str,
+        client_secret: str,
+        refresh_token: str,
+    ) -> Dict[str, Any]:
+        """Refresh an access token without touching the mailbox (standalone)."""
+        tokens = await self._refresh(
+            creds=None,
+            client_id=client_id,
+            client_secret=client_secret,
+            refresh_token=refresh_token,
+            context="standalone refresh",
+        )
+        return tokens
+
+    async def _refresh(
+        self,
+        creds: Optional[GmailCredentials],
+        *,
+        context: str,
+        client_id: Optional[str] = None,
+        client_secret: Optional[str] = None,
+        refresh_token: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        if creds is not None:
+            client_id = creds.client_id
+            client_secret = creds.client_secret
+            refresh_token = creds.refresh_token
+
+        if not (client_id and client_secret and refresh_token):
+            raise GmailAPIError(
+                "Token refresh requires refresh_token, client_id and client_secret.",
+                status_code=400,
+            )
+
+        client = await self._get_client()
+        data = {
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "refresh_token": refresh_token,
+            "grant_type": "refresh_token",
+        }
+        try:
+            response = await client.post(_settings.google_oauth_token_url, data=data)
+        except httpx.TimeoutException as exc:
+            raise GmailAPIError(
+                "Google token refresh timed out.", status_code=504
+            ) from exc
+        except httpx.RequestError as exc:
+            raise GmailAPIError(
+                f"Failed to reach Google token endpoint: {exc}", status_code=502
+            ) from exc
+
+        if response.status_code == 200:
+            tokens = response.json()
+            new_token = tokens.get("access_token")
+            if not new_token:
+                raise GmailAPIError(
+                    "Google did not return an access token.", status_code=502
+                )
+            expires_in = int(tokens.get("expires_in", 3600))
+            if creds is not None:
+                creds.access_token = new_token
+                creds.expires_at = time.time() + expires_in
+                creds.refreshed_access_token = new_token
+            _logger.info("Gmail access token refreshed (%s)", context)
+            return tokens
+
+        try:
+            body = response.json()
+            error = body.get("error", "")
+            description = body.get("error_description", "")
+        except Exception:
+            error = ""
+            description = ""
+            body = {}
+
+        if error == "invalid_grant":
+            raise GmailAPIError(
+                description or "Refresh token is invalid, expired, or revoked. "
+                "Re-authorize via POST /api/v1/google/oauth/auth-url.",
+                status_code=401,
+                reason=error,
+            )
+        if error == "invalid_client":
+            raise GmailAPIError(
+                description or "Invalid client_id or client_secret.",
+                status_code=401,
+                reason=error,
+            )
+        raise GmailAPIError(
+            description or f"Google token refresh failed (HTTP {response.status_code}).",
+            status_code=response.status_code if response.status_code >= 400 else 502,
+            reason=error,
+        )
+
+    # ------------------------------------------------------------------
+    # Users profile & scopes
+    # ------------------------------------------------------------------
+
+    async def get_profile(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/profile")
+
+    async def list_available_scopes(self) -> List[Dict[str, Any]]:
+        from app.models.schemas import GmailScope
+
+        return [
+            {
+                "name": scope.name,
+                "uri": scope.value,
+                "permission_level": scope.permission_level,
+            }
+            for scope in GmailScope
+        ]
+
+    # ------------------------------------------------------------------
+    # Messages
+    # ------------------------------------------------------------------
+
+    async def list_messages(
+        self,
+        creds: GmailCredentials,
+        user_id: str = "me",
+        *,
+        q: Optional[str] = None,
+        label_ids: Optional[List[str]] = None,
+        max_results: Optional[int] = None,
+        page_token: Optional[str] = None,
+        include_spam_trash: bool = False,
+        batch_size: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {}
+        if q is not None:
+            params["q"] = q
+        if label_ids:
+            params["labelIds"] = label_ids
+        if max_results is not None:
+            params["maxResults"] = max_results
+        if page_token:
+            params["pageToken"] = page_token
+        if include_spam_trash:
+            params["includeSpamTrash"] = "true"
+        if batch_size is not None:
+            params["batchSize"] = batch_size
+        return await self._request(
+            creds, "GET", f"/users/{user_id}/messages", params=params or None
+        )
+
+    async def get_message(
+        self,
+        creds: GmailCredentials,
+        message_id: str,
+        user_id: str = "me",
+        *,
+        format: str = "full",
+        metadata_headers: Optional[List[str]] = None,
+        headers: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {"format": format}
+        if metadata_headers:
+            params["metadataHeaders"] = metadata_headers
+        if headers:
+            params["headers"] = headers
+        return await self._request(
+            creds, "GET", f"/users/{user_id}/messages/{message_id}", params=params
+        )
+
+    async def get_parsed_message(
+        self,
+        creds: GmailCredentials,
+        message_id: str,
+        user_id: str = "me",
+        *,
+        format: str = "full",
+    ) -> GmailParsedMessage:
+        data = await self.get_message(creds, message_id, user_id=user_id, format=format)
+        return parse_message(data)
+
+    async def send_raw(
+        self,
+        creds: GmailCredentials,
+        raw: str,
+        user_id: str = "me",
+        *,
+        thread_id: Optional[str] = None,
+        labels: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"raw": raw}
+        if thread_id:
+            body["threadId"] = thread_id
+        if labels:
+            body["labelIds"] = labels
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/messages/send", json_body=body
+        )
+
+    async def send_composed(
+        self,
+        creds: GmailCredentials,
+        request: GmailComposeMessageRequest,
+        user_id: str = "me",
+    ) -> Dict[str, Any]:
+        try:
+            raw = build_mime_message(request)
+        except GmailMessageValidationError as exc:
+            raise GmailAPIError(str(exc), status_code=400) from exc
+        return await self.send_raw(
+            creds,
+            raw,
+            user_id=user_id,
+            thread_id=request.thread_id,
+            labels=request.labels,
+        )
+
+    async def insert_message(
+        self,
+        creds: GmailCredentials,
+        raw: str,
+        user_id: str = "me",
+        *,
+        label_ids: Optional[List[str]] = None,
+        internal_date_source: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"raw": raw}
+        if label_ids:
+            body["labelIds"] = label_ids
+        if internal_date_source:
+            body["internalDateSource"] = internal_date_source
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/messages", json_body=body
+        )
+
+    async def modify_message(
+        self,
+        creds: GmailCredentials,
+        message_id: str,
+        user_id: str = "me",
+        *,
+        add_label_ids: Optional[List[str]] = None,
+        remove_label_ids: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {}
+        if add_label_ids:
+            body["addLabelIds"] = add_label_ids
+        if remove_label_ids:
+            body["removeLabelIds"] = remove_label_ids
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/messages/{message_id}/modify", json_body=body
+        )
+
+    async def batch_modify(
+        self,
+        creds: GmailCredentials,
+        ids: List[str],
+        user_id: str = "me",
+        *,
+        add_label_ids: Optional[List[str]] = None,
+        remove_label_ids: Optional[List[str]] = None,
+    ) -> None:
+        body: Dict[str, Any] = {"ids": ids}
+        if add_label_ids:
+            body["addLabelIds"] = add_label_ids
+        if remove_label_ids:
+            body["removeLabelIds"] = remove_label_ids
+        await self._request(creds, "POST", f"/users/{user_id}/messages/batchModify", json_body=body)
+
+    async def batch_delete(
+        self, creds: GmailCredentials, ids: List[str], user_id: str = "me"
+    ) -> None:
+        await self._request(
+            creds, "POST", f"/users/{user_id}/messages/batchDelete", json_body={"ids": ids}
+        )
+
+    async def trash_message(
+        self, creds: GmailCredentials, message_id: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/messages/{message_id}/trash"
+        )
+
+    async def untrash_message(
+        self, creds: GmailCredentials, message_id: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/messages/{message_id}/untrash"
+        )
+
+    async def delete_message(
+        self, creds: GmailCredentials, message_id: str, user_id: str = "me"
+    ) -> None:
+        await self._request(creds, "DELETE", f"/users/{user_id}/messages/{message_id}")
+
+    async def get_attachment(
+        self,
+        creds: GmailCredentials,
+        message_id: str,
+        attachment_id: str,
+        user_id: str = "me",
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds,
+            "GET",
+            f"/users/{user_id}/messages/{message_id}/attachments/{attachment_id}",
+        )
+
+    # ------------------------------------------------------------------
+    # Drafts
+    # ------------------------------------------------------------------
+
+    async def list_drafts(
+        self,
+        creds: GmailCredentials,
+        user_id: str = "me",
+        *,
+        max_results: Optional[int] = None,
+        page_token: Optional[str] = None,
+        q: Optional[str] = None,
+        include_spam_trash: bool = False,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {}
+        if max_results is not None:
+            params["maxResults"] = max_results
+        if page_token:
+            params["pageToken"] = page_token
+        if q:
+            params["q"] = q
+        if include_spam_trash:
+            params["includeSpamTrash"] = "true"
+        return await self._request(creds, "GET", f"/users/{user_id}/drafts", params=params or None)
+
+    async def get_draft(
+        self,
+        creds: GmailCredentials,
+        draft_id: str,
+        user_id: str = "me",
+        *,
+        format: str = "full",
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "GET", f"/users/{user_id}/drafts/{draft_id}", params={"format": format}
+        )
+
+    async def create_draft(
+        self,
+        creds: GmailCredentials,
+        raw: str,
+        user_id: str = "me",
+        *,
+        thread_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"message": {"raw": raw}}
+        if thread_id:
+            body["message"]["threadId"] = thread_id
+        return await self._request(creds, "POST", f"/users/{user_id}/drafts", json_body=body)
+
+    async def update_draft(
+        self,
+        creds: GmailCredentials,
+        draft_id: str,
+        raw: str,
+        user_id: str = "me",
+        *,
+        thread_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"message": {"raw": raw}}
+        if thread_id:
+            body["message"]["threadId"] = thread_id
+        return await self._request(
+            creds, "PUT", f"/users/{user_id}/drafts/{draft_id}", json_body=body
+        )
+
+    async def send_draft(
+        self, creds: GmailCredentials, draft_id: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/drafts/send", json_body={"id": draft_id}
+        )
+
+    async def delete_draft(
+        self, creds: GmailCredentials, draft_id: str, user_id: str = "me"
+    ) -> None:
+        await self._request(creds, "DELETE", f"/users/{user_id}/drafts/{draft_id}")
+
+    # ------------------------------------------------------------------
+    # Threads
+    # ------------------------------------------------------------------
+
+    async def list_threads(
+        self,
+        creds: GmailCredentials,
+        user_id: str = "me",
+        *,
+        q: Optional[str] = None,
+        label_ids: Optional[List[str]] = None,
+        max_results: Optional[int] = None,
+        page_token: Optional[str] = None,
+        include_spam_trash: bool = False,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {}
+        if q:
+            params["q"] = q
+        if label_ids:
+            params["labelIds"] = label_ids
+        if max_results is not None:
+            params["maxResults"] = max_results
+        if page_token:
+            params["pageToken"] = page_token
+        if include_spam_trash:
+            params["includeSpamTrash"] = "true"
+        return await self._request(creds, "GET", f"/users/{user_id}/threads", params=params or None)
+
+    async def get_thread(
+        self,
+        creds: GmailCredentials,
+        thread_id: str,
+        user_id: str = "me",
+        *,
+        format: str = "full",
+        metadata_headers: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {"format": format}
+        if metadata_headers:
+            params["metadataHeaders"] = metadata_headers
+        return await self._request(
+            creds, "GET", f"/users/{user_id}/threads/{thread_id}", params=params
+        )
+
+    async def modify_thread(
+        self,
+        creds: GmailCredentials,
+        thread_id: str,
+        user_id: str = "me",
+        *,
+        add_label_ids: Optional[List[str]] = None,
+        remove_label_ids: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {}
+        if add_label_ids:
+            body["addLabelIds"] = add_label_ids
+        if remove_label_ids:
+            body["removeLabelIds"] = remove_label_ids
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/threads/{thread_id}/modify", json_body=body
+        )
+
+    async def trash_thread(
+        self, creds: GmailCredentials, thread_id: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/threads/{thread_id}/trash"
+        )
+
+    async def untrash_thread(
+        self, creds: GmailCredentials, thread_id: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/threads/{thread_id}/untrash"
+        )
+
+    async def delete_thread(
+        self, creds: GmailCredentials, thread_id: str, user_id: str = "me"
+    ) -> None:
+        await self._request(creds, "DELETE", f"/users/{user_id}/threads/{thread_id}")
+
+    # ------------------------------------------------------------------
+    # Labels
+    # ------------------------------------------------------------------
+
+    async def list_labels(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/labels")
+
+    async def get_label(
+        self, creds: GmailCredentials, label_id: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/labels/{label_id}")
+
+    async def create_label(
+        self, creds: GmailCredentials, label: GmailLabelRequest, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"name": label.name}
+        if label.label_list_visibility:
+            body["labelListVisibility"] = label.label_list_visibility
+        if label.message_list_visibility:
+            body["messageListVisibility"] = label.message_list_visibility
+        if label.color:
+            body["color"] = label.color
+        return await self._request(creds, "POST", f"/users/{user_id}/labels", json_body=body)
+
+    async def update_label(
+        self,
+        creds: GmailCredentials,
+        label_id: str,
+        label: GmailLabelRequest,
+        user_id: str = "me",
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"name": label.name}
+        if label.label_list_visibility:
+            body["labelListVisibility"] = label.label_list_visibility
+        if label.message_list_visibility:
+            body["messageListVisibility"] = label.message_list_visibility
+        if label.color:
+            body["color"] = label.color
+        return await self._request(
+            creds, "PUT", f"/users/{user_id}/labels/{label_id}", json_body=body
+        )
+
+    async def delete_label(
+        self, creds: GmailCredentials, label_id: str, user_id: str = "me"
+    ) -> None:
+        await self._request(creds, "DELETE", f"/users/{user_id}/labels/{label_id}")
+
+    # ------------------------------------------------------------------
+    # Settings: auto-forwarding / vacation / filters / forwarding / send-as / delegates
+    # ------------------------------------------------------------------
+
+    async def get_auto_forwarding(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/autoForwarding")
+
+    async def update_auto_forwarding(
+        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "PUT", f"/users/{user_id}/settings/autoForwarding", json_body=payload
+        )
+
+    async def get_vacation(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/vacation")
+
+    async def update_vacation(
+        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "PUT", f"/users/{user_id}/settings/vacation", json_body=payload
+        )
+
+    async def list_filters(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/filters")
+
+    async def get_filter(
+        self, creds: GmailCredentials, filter_id: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/filters/{filter_id}")
+
+    async def create_filter(
+        self, creds: GmailCredentials, req: GmailFilterCreateRequest, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"criteria": req.criteria or {}, "action": req.action or {}}
+        return await self._request(creds, "POST", f"/users/{user_id}/settings/filters", json_body=body)
+
+    async def delete_filter(
+        self, creds: GmailCredentials, filter_id: str, user_id: str = "me"
+    ) -> None:
+        await self._request(creds, "DELETE", f"/users/{user_id}/settings/filters/{filter_id}")
+
+    async def list_forwarding_addresses(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/forwardingAddresses")
+
+    async def get_forwarding_address(
+        self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "GET", f"/users/{user_id}/settings/forwardingAddresses/{forwarding_email}"
+        )
+
+    async def create_forwarding_address(
+        self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds,
+            "POST",
+            f"/users/{user_id}/settings/forwardingAddresses",
+            json_body={"forwardingEmail": forwarding_email},
+        )
+
+    async def delete_forwarding_address(
+        self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me"
+    ) -> None:
+        await self._request(
+            creds, "DELETE", f"/users/{user_id}/settings/forwardingAddresses/{forwarding_email}"
+        )
+
+    async def list_send_as(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/sendAs")
+
+    async def get_send_as(
+        self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "GET", f"/users/{user_id}/settings/sendAs/{send_as_email}"
+        )
+
+    async def create_send_as(
+        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/settings/sendAs", json_body=payload
+        )
+
+    async def update_send_as(
+        self,
+        creds: GmailCredentials,
+        send_as_email: str,
+        payload: Dict[str, Any],
+        user_id: str = "me",
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "PATCH", f"/users/{user_id}/settings/sendAs/{send_as_email}", json_body=payload
+        )
+
+    async def delete_send_as(
+        self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
+    ) -> None:
+        await self._request(
+            creds, "DELETE", f"/users/{user_id}/settings/sendAs/{send_as_email}"
+        )
+
+    async def list_delegates(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/delegates")
+
+    async def get_delegate(
+        self, creds: GmailCredentials, delegate_email: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "GET", f"/users/{user_id}/settings/delegates/{delegate_email}"
+        )
+
+    async def create_delegate(
+        self, creds: GmailCredentials, delegate_email: str, user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds,
+            "POST",
+            f"/users/{user_id}/settings/delegates",
+            json_body={"delegateEmail": delegate_email},
+        )
+
+    async def delete_delegate(
+        self, creds: GmailCredentials, delegate_email: str, user_id: str = "me"
+    ) -> None:
+        await self._request(
+            creds, "DELETE", f"/users/{user_id}/settings/delegates/{delegate_email}"
+        )

app/services/google_scope_service.py

@@ -0,0 +1,245 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from app.services.google_scopes_data import GOOGLE_APIS
+
+# OIDC short names that are conventionally sent without a full URI. They are
+# already present as scope names for the oauth_signin plan, but accepting them
+# explicitly keeps validation predictable for all APIs.
+_OIDC_SHORT_NAMES = frozenset({"openid", "email", "profile"})
+
+
+class GoogleScopeError(Exception):
+    """Raised for invalid scope configuration requests."""
+
+    def __init__(self, message: str, status_code: int = 400) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
+
+def _default_scopes(plan: Dict[str, Any]) -> List[str]:
+    """Derive the recommended default scopes for a plan."""
+    pn = plan.get("production_notes") or {}
+    dflt = pn.get("default_scope_request")
+    if dflt:
+        return [s.strip() for s in dflt.split() if s.strip()]
+    recs = plan.get("recommended_combinations") or {}
+    if recs:
+        first = next(iter(recs.values()))
+        return list(first)
+    return []
+
+
+def _allowed_scopes(plan: Dict[str, Any]) -> frozenset:
+    """Set of scope identifiers (URI or short name) accepted for a plan."""
+    allowed = set(_OIDC_SHORT_NAMES)
+    for scope in plan.get("scopes") or []:
+        if scope.get("uri"):
+            allowed.add(scope["uri"])
+        if scope.get("name"):
+            allowed.add(scope["name"])
+    return frozenset(allowed)
+
+
+class GoogleScopeService:
+    """Serves the Google per-API scope catalog entirely from memory.
+
+    The catalog is a Python object structure (``google_scopes_data.GOOGLE_APIS``)
+    and the currently selected scopes are held as an in-memory dict of objects.
+    No JSON files are read or written.
+    """
+
+    def __init__(self) -> None:
+        self._plans: Dict[str, Dict[str, Any]] = dict(GOOGLE_APIS)
+        self._current_scopes: Optional[Dict[str, List[str]]] = None
+
+    @property
+    def catalog(self) -> Dict[str, Dict[str, Any]]:
+        return self._plans
+
+    @property
+    def current_scopes(self) -> Dict[str, List[str]]:
+        return self._state()
+
+    def _state(self) -> Dict[str, List[str]]:
+        if self._current_scopes is None:
+            self._current_scopes = {
+                pid: _default_scopes(plan) for pid, plan in self._plans.items()
+            }
+        return self._current_scopes
+
+    async def load_all_plans(self) -> Dict[str, Dict[str, Any]]:
+        return self._plans
+
+    async def load_default_state(self) -> Dict[str, List[str]]:
+        return {pid: _default_scopes(plan) for pid, plan in self._plans.items()}
+
+    async def load_state(self) -> Dict[str, List[str]]:
+        return self._state()
+
+    async def set_current_scopes(
+        self, updates: Dict[str, List[str]], reset: bool = False
+    ) -> Dict[str, List[str]]:
+        """Validate and store the requested current scopes in memory.
+
+        Raises GoogleScopeError with a 400 status for unknown API ids or scopes
+        that are not defined in the catalog.
+        """
+        if reset:
+            self._current_scopes = {
+                pid: _default_scopes(plan) for pid, plan in self._plans.items()
+            }
+            return self._current_scopes
+
+        current = self._state()
+        errors: List[str] = []
+        for pid, scopes in (updates or {}).items():
+            if pid not in self._plans:
+                errors.append(f"Unknown API id '{pid}'. Valid ids: {', '.join(sorted(self._plans))}")
+                continue
+            allowed = _allowed_scopes(self._plans[pid])
+            cleaned = [str(s).strip() for s in scopes if str(s).strip()]
+            bad = [s for s in cleaned if s not in allowed]
+            if bad:
+                errors.append(f"Invalid scopes for '{pid}': {', '.join(bad)}")
+                continue
+            current[pid] = cleaned
+
+        if errors:
+            raise GoogleScopeError("; ".join(errors), status_code=400)
+
+        self._current_scopes = current
+        return current
+
+    @staticmethod
+    def _scope_text(scope: Dict[str, Any]) -> str:
+        return " ".join(
+            str(scope.get(k) or "") for k in ("name", "uri", "description")
+        ).lower()
+
+    @staticmethod
+    def _scope_matches(
+        scope: Dict[str, Any],
+        *,
+        scope_name: Optional[str] = None,
+        permission_level: Optional[List[str]] = None,
+        required: Optional[bool] = None,
+        recommended: Optional[bool] = None,
+    ) -> bool:
+        if scope_name:
+            needle = scope_name.lower()
+            if needle not in (str(scope.get("name") or "").lower()) and needle not in (
+                str(scope.get("uri") or "").lower()
+            ):
+                return False
+        if permission_level:
+            level = str(scope.get("permission_level") or "").lower()
+            if level not in {p.lower() for p in permission_level}:
+                return False
+        if required is not None and bool(scope.get("required")) is not required:
+            return False
+        if recommended is not None and bool(scope.get("recommended")) is not recommended:
+            return False
+        return True
+
+    @classmethod
+    def filter_data(
+        cls,
+        data: List[Dict[str, Any]],
+        *,
+        api_name: Optional[str] = None,
+        oauth_only: bool = False,
+        no_scope: bool = False,
+        search: Optional[str] = None,
+        scope_name: Optional[str] = None,
+        permission_level: Optional[List[str]] = None,
+        required: Optional[bool] = None,
+        recommended: Optional[bool] = None,
+        scope_name_only: bool = False,
+    ) -> List[Dict[str, Any]]:
+        """Apply discoverability filters to the built catalog entries.
+
+        - ``api_name`` keeps a single API (case-insensitive id match).
+        - ``oauth_only`` / ``no_scope`` filter by whether scopes are required.
+        - ``search`` free-text matches the API id/name and scope fields.
+        - ``scope_name``, ``permission_level``, ``required``, ``recommended``
+          filter the ``scopes`` array of each entry; entries left with zero
+          matching scopes are dropped unless ``api_name`` was requested.
+        - ``scope_name_only`` replaces each ``scopes`` array with the list of
+          scope names (falling back to the URI).
+        """
+        api_name_l = api_name.strip().lower() if api_name else None
+        scope_filters_active = any(
+            [scope_name, permission_level, required is not None, recommended is not None]
+        )
+        needle = search.strip().lower() if search else None
+        levels = [p.strip().lower() for p in (permission_level or []) if p.strip()]
+
+        filtered: List[Dict[str, Any]] = []
+        for entry in data:
+            if api_name_l and str(entry.get("id") or "").lower() != api_name_l:
+                continue
+            if oauth_only and not entry.get("scope_required"):
+                continue
+            if no_scope and entry.get("scope_required"):
+                continue
+
+            haystack = (
+                f"{entry.get('id') or ''} {entry.get('api') or ''} "
+                + " ".join(cls._scope_text(s) for s in entry.get("scopes") or [])
+            ).lower()
+            if needle and needle not in haystack:
+                continue
+
+            scopes = entry.get("scopes") or []
+            if scope_filters_active:
+                scopes = [
+                    s for s in scopes
+                    if cls._scope_matches(
+                        s,
+                        scope_name=scope_name,
+                        permission_level=levels or None,
+                        required=required,
+                        recommended=recommended,
+                    )
+                ]
+                if not scopes and not api_name_l:
+                    continue
+
+            item = dict(entry)
+            if scope_name_only:
+                item["scopes"] = [str(s.get("name") or s.get("uri")) for s in scopes]
+            else:
+                item["scopes"] = scopes
+            filtered.append(item)
+        return filtered
+
+    @staticmethod
+    def build_data(
+        plans: Dict[str, Dict[str, Any]],
+        state: Dict[str, List[str]],
+    ) -> List[Dict[str, Any]]:
+        """Merge catalog plans with current scopes into the API response payload."""
+        data: List[Dict[str, Any]] = []
+        for pid in sorted(plans):
+            plan = plans[pid]
+            data.append(
+                {
+                    "id": pid,
+                    "api": plan.get("api"),
+                    "version": plan.get("version"),
+                    "base_url": plan.get("base_url"),
+                    "batch_endpoint": plan.get("batch_endpoint"),
+                    "auth_model": plan.get("auth_model"),
+                    "scope_required": plan.get("scope_required", False),
+                    "docs_url": plan.get("docs_url"),
+                    "info": plan.get("info"),
+                    "current_scopes": state.get(pid, []),
+                    "default_scopes": _default_scopes(plan),
+                    "scopes": plan.get("scopes", []),
+                    "recommended_combinations": plan.get("recommended_combinations", {}),
+                }
+            )
+        return data

app/utils/gmail_utils.py

@@ -0,0 +1,293 @@
+from __future__ import annotations
+
+import base64
+import binascii
+import mimetypes
+import os
+from email import message_from_bytes
+from email.mime.application import MIMEApplication
+from email.mime.audio import MIMEAudio
+from email.mime.base import MIMEBase
+from email.mime.image import MIMEImage
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from typing import Any, Dict, List, Optional
+
+from app.config import get_settings
+from app.models.schemas import (
+    GmailComposeMessageRequest,
+    GmailMessageAttachmentMeta,
+    GmailParsedMessage,
+)
+
+_settings = get_settings()
+
+
+class GmailMessageValidationError(ValueError):
+    """Raised when a composed message fails size/content validation."""
+
+
+class GmailParseError(ValueError):
+    """Raised when a Gmail API message payload cannot be parsed."""
+
+
+_EXTENSION_TO_MIME: Dict[str, str] = {
+    ".pdf": "application/pdf",
+    ".png": "image/png",
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".gif": "image/gif",
+    ".bmp": "image/bmp",
+    ".webp": "image/webp",
+    ".txt": "text/plain",
+    ".csv": "text/csv",
+    ".json": "application/json",
+    ".xml": "application/xml",
+    ".zip": "application/zip",
+    ".doc": "application/msword",
+    ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+    ".xls": "application/vnd.ms-excel",
+    ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    ".ppt": "application/vnd.ms-powerpoint",
+    ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
+    ".mp3": "audio/mpeg",
+    ".wav": "audio/wav",
+    ".mp4": "video/mp4",
+    ".mov": "video/quicktime",
+}
+
+_IMMUTABLE_HEADERS = frozenset({
+    "from", "to", "subject", "cc", "bcc", "reply-to",
+    "in-reply-to", "references", "content-type", "mime-version",
+    "content-transfer-encoding",
+})
+
+
+def guess_mime_type(filename: str) -> str:
+    """Best-effort MIME type from a filename; falls back to octet-stream."""
+    mime_type, _ = mimetypes.guess_type(filename)
+    if mime_type is None:
+        ext = os.path.splitext(filename)[1].lower()
+        mime_type = _EXTENSION_TO_MIME.get(ext, "application/octet-stream")
+    return mime_type
+
+
+def _create_attachment_part(filename: str, content_bytes: bytes, mime_type: str) -> MIMEBase:
+    main_type, _, sub_type = mime_type.partition("/")
+    sub_type = sub_type or "octet-stream"
+
+    if main_type == "image":
+        part: MIMEBase = MIMEImage(content_bytes, _subtype=sub_type)
+    elif main_type == "audio":
+        part = MIMEAudio(content_bytes, _subtype=sub_type)
+    elif main_type == "application":
+        part = MIMEApplication(content_bytes, _subtype=sub_type, Name=filename)
+    else:
+        part = MIMEBase(main_type or "application", sub_type)
+        part.set_payload(content_bytes)
+
+    part.add_header("Content-Disposition", f'attachment; filename="{filename}"')
+    part.add_header("Content-ID", f"<{filename}>")
+    return part
+
+
+def build_mime_message(
+    request: GmailComposeMessageRequest,
+) -> str:
+    """Build an RFC 2822 message from a structured compose request.
+
+    Returns the base64url-encoded message string accepted by the Gmail API.
+    The ``From`` header is intentionally omitted so Gmail fills it with the
+    authenticated mailbox address (client-supplied, stateless design).
+    """
+    message = MIMEMultipart("mixed")
+
+    to = request.to
+    if not to:
+        raise GmailMessageValidationError("At least one 'to' recipient is required.")
+
+    message["To"] = ", ".join(to)
+    message["Subject"] = request.subject
+
+    if request.cc:
+        message["Cc"] = ", ".join(request.cc)
+    if request.bcc:
+        message["Bcc"] = ", ".join(request.bcc)
+    if request.reply_to:
+        message["Reply-To"] = ", ".join(request.reply_to)
+    if request.in_reply_to:
+        message["In-Reply-To"] = request.in_reply_to
+    if request.references:
+        message["References"] = " ".join(request.references)
+
+    if request.headers:
+        for key, value in request.headers.items():
+            if key.lower() not in _IMMUTABLE_HEADERS:
+                message[key] = value
+
+    alternative = MIMEMultipart("alternative")
+    if request.body:
+        alternative.attach(MIMEText(request.body, "plain", "utf-8"))
+    if request.html_body:
+        alternative.attach(MIMEText(request.html_body, "html", "utf-8"))
+    elif request.body:
+        alternative.attach(MIMEText(request.body, "html", "utf-8"))
+
+    if alternative.get_payload():
+        message.attach(alternative)
+
+    if request.attachments:
+        total_bytes = sum(
+            len(_b64_decode(s.content_base64)) for s in request.attachments
+        )
+        if total_bytes > _settings.gmail_max_payload_bytes:
+            raise GmailMessageValidationError(
+                f"Total attachment payload of {total_bytes} bytes exceeds the "
+                f"limit of {_settings.gmail_max_payload_bytes} bytes."
+            )
+
+        for spec in request.attachments:
+            content = _b64_decode(spec.content_base64)
+            if len(content) > _settings.gmail_max_attachment_bytes:
+                raise GmailMessageValidationError(
+                    f"Attachment '{spec.filename}' is {len(content)} bytes, exceeding "
+                    f"the {_settings.gmail_max_attachment_bytes} byte per-file limit."
+                )
+            mime_type = spec.mime_type or guess_mime_type(spec.filename)
+            message.attach(_create_attachment_part(spec.filename, content, mime_type))
+
+    return encode_mime_to_base64(message)
+
+
+def encode_mime_to_base64(mime_message: Any) -> str:
+    """Base64url-encode a MIME message for the Gmail API."""
+    return base64.urlsafe_b64encode(mime_message.as_bytes()).decode("ascii")
+
+
+def _b64_decode(data: str) -> bytes:
+    """Decode base64url (or standard base64) into bytes."""
+    try:
+        return base64.urlsafe_b64decode(data.encode("ascii"))
+    except (binascii.Error, ValueError) as exc:
+        raise GmailMessageValidationError(
+            "Attachment content must be valid base64."
+        ) from exc
+
+
+def _base64_decode_text(data: str) -> str:
+    if not data:
+        return ""
+    try:
+        return _b64_decode(data).decode("utf-8", errors="replace")
+    except GmailMessageValidationError:
+        return ""
+
+
+def parse_message(message: Dict[str, Any]) -> GmailParsedMessage:
+    """Parse a Gmail ``format=full`` or ``format=metadata`` message resource.
+
+    Handles the ``raw`` format transparently when present. Recursively extracts
+    plain/HTML bodies and top-level attachment metadata.
+    """
+    try:
+        message_id = message.get("id", "")
+        thread_id = message.get("threadId", "")
+        label_ids = message.get("labelIds") or []
+        snippet = message.get("snippet", "")
+        history_id = message.get("historyId")
+
+        headers: Dict[str, str] = {}
+        plain_body: Optional[str] = None
+        html_body: Optional[str] = None
+        attachments: List[GmailMessageAttachmentMeta] = []
+
+        payload = message.get("payload") or {}
+        if message.get("raw"):
+            decoded = _b64_decode(message["raw"])
+            parsed = message_from_bytes(decoded)
+            headers = {
+                h: v for h, v in parsed.items()
+            }
+            if parsed.get_content_type() == "multipart/mixed":
+                for part in parsed.walk():
+                    ctype = part.get_content_type()
+                    if ctype == "text/plain":
+                        plain_body = part.get_payload(decode=True).decode("utf-8", errors="replace")
+                    elif ctype == "text/html":
+                        html_body = part.get_payload(decode=True).decode("utf-8", errors="replace")
+                    else:
+                        filename = part.get_filename()
+                        if filename:
+                            attachments.append(GmailMessageAttachmentMeta(
+                                attachment_id="",
+                                filename=filename,
+                                mime_type=part.get_content_type(),
+                                size_bytes=len(part.get_payload(decode=True) or b""),
+                            ))
+            elif parsed.get_content_type() == "text/plain":
+                plain_body = parsed.get_payload(decode=True).decode("utf-8", errors="replace")
+        else:
+            for header in payload.get("headers") or []:
+                name = header.get("name", "")
+                if name:
+                    headers[name] = header.get("value", "")
+            plain_body, html_body, attachments = _walk_payload(payload)
+
+        subject = headers.get("Subject", headers.get("subject", ""))
+        if subject and headers.get("Subject") is None and "subject" in headers:
+            subject = headers["subject"]
+
+        return GmailParsedMessage(
+            id=message_id,
+            thread_id=thread_id,
+            label_ids=label_ids,
+            snippet=snippet,
+            history_id=history_id,
+            headers=headers,
+            sender=headers.get("From", ""),
+            to=headers.get("To", ""),
+            cc=headers.get("Cc", ""),
+            bcc=headers.get("Bcc", ""),
+            subject=subject,
+            date=headers.get("Date"),
+            plain_body=plain_body,
+            html_body=html_body,
+            attachments=attachments,
+        )
+    except GmailMessageValidationError as exc:
+        raise GmailParseError(str(exc)) from exc
+
+
+def _walk_payload(
+    payload: Dict[str, Any],
+) -> tuple[Optional[str], Optional[str], List[GmailMessageAttachmentMeta]]:
+    """Recursively extract text bodies and attachment metadata from a payload tree."""
+    plain_body: Optional[str] = None
+    html_body: Optional[str] = None
+    attachments: List[GmailMessageAttachmentMeta] = []
+
+    def visit(node: Dict[str, Any]) -> None:
+        nonlocal plain_body, html_body
+        mime_type = node.get("mimeType", "")
+        filename = node.get("filename", "")
+        body = node.get("body") or {}
+        parts = node.get("parts") or []
+
+        if filename:
+            attachments.append(GmailMessageAttachmentMeta(
+                attachment_id=body.get("attachmentId", ""),
+                filename=filename,
+                mime_type=mime_type,
+                size_bytes=int(body.get("size", 0) or 0),
+            ))
+
+        if mime_type == "text/plain" and plain_body is None:
+            plain_body = _base64_decode_text(body.get("data", ""))
+        elif mime_type == "text/html" and html_body is None:
+            html_body = _base64_decode_text(body.get("data", ""))
+
+        for child in parts:
+            visit(child)
+
+    visit(payload)
+    return plain_body, html_body, attachments