feat(scouts): debug scripts + deliver_pending diagnostic logs

- scripts/trigger_gmail_scout.py: manually fire ScoutEngine.trigger_scout
- scripts/inspect_gmail_scout_token.py: decrypt + show stored OAuth scopes
- scripts/show_gmail_scout_state.py: print scout config + queue/log counts
- scripts/reset_triage_queue_to_queued.py: revert delivered → queued for re-delivery
- engine.py: info logs around deliver_pending (rows found, send_json roundtrip)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

alembic/versions/009_cloud_scout_gmail_address.py

@@ -0,0 +1,25 @@
+"""Add gmail_address to cloud_scout_configs.
+
+Revision ID: 009
+Revises: 008
+Create Date: 2026-05-16
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision: str = "009"
+down_revision: Union[str, None] = "008"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("cloud_scout_configs", sa.Column("gmail_address", sa.String(320), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("cloud_scout_configs", "gmail_address")

app/api/routes/scouts.py

@@ -26,7 +26,7 @@ from datetime import datetime, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.responses import RedirectResponse
-from sqlalchemy import func, select
+from sqlalchemy import delete as sa_delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from pydantic import BaseModel
@@ -39,9 +39,13 @@ from app.core.scout_runner import is_agent_running, run_local_agent
 from app.core.device_manager import device_manager
 from app.core.note_summarizer import generate_note_summary
 from app.db import get_session
-from app.integrations import encrypt_token
+from app.integrations import decrypt_token, encrypt_token
 from app.models import CloudScoutConfig, ScoutRunLog, LocalScoutConfig
+from app.scouts.connectors.registry import get_connector
 from app.schemas import (
+    CloudScoutCreateRequest,
+    CloudScoutResponse,
+    CloudScoutUpdateRequest,
     ScoutCatalogItem,
     ScoutCreationCheckRequest,
     ScoutCreationCheckResponse,
@@ -269,6 +273,154 @@ async def summarize_note(
     return NoteSummarizeResponse(summary=summary)
 
 
+# ── Cloud scout CRUD ──────────────────────────────────────────────────────────
+
+_DEFAULT_CLOUD_SCHEDULE = "0 */6 * * *"
+
+
+def _to_cloud_response(scout: CloudScoutConfig) -> dict:
+    return {
+        "id": scout.id,
+        "user_id": scout.user_id,
+        "provider": scout.provider,
+        "name": scout.name,
+        "data_types": scout.data_types or [],
+        "prompt_template": scout.prompt_template or "",
+        "schedule_cron": scout.schedule_cron,
+        "filter_config": scout.filter_config,
+        "auto_trash_spam": scout.auto_trash_spam,
+        "enabled": scout.enabled,
+        "last_run_at": _dt_ms_opt(scout.last_run_at),
+        "gmail_address": scout.gmail_address,
+        "oauth_connected": scout.oauth_token_encrypted is not None,
+        "created_at": _dt_ms(scout.created_at),
+        "updated_at": _dt_ms(scout.updated_at),
+    }
+
+
+@router.get("/cloud", response_model=list[CloudScoutResponse])
+async def list_cloud_scouts(
+    db: AsyncSession = Depends(get_session),
+    current_user: UserProfile = Depends(get_current_user),
+):
+    rows = (await db.execute(
+        select(CloudScoutConfig).where(CloudScoutConfig.user_id == current_user.id)
+    )).scalars().all()
+    return [_to_cloud_response(s) for s in rows]
+
+
+@router.post("/cloud", response_model=CloudScoutResponse, status_code=status.HTTP_201_CREATED)
+async def create_cloud_scout(
+    body: CloudScoutCreateRequest,
+    db: AsyncSession = Depends(get_session),
+    current_user: UserProfile = Depends(get_current_user),
+):
+    scout = CloudScoutConfig(
+        id=str(uuid.uuid4()),
+        user_id=current_user.id,
+        provider=body.provider,
+        name=body.name,
+        data_types=body.data_types,
+        prompt_template=body.prompt_template,
+        filter_config=body.filter_config,
+        schedule_cron=body.schedule_cron or _DEFAULT_CLOUD_SCHEDULE,
+        auto_trash_spam=body.auto_trash_spam,
+        enabled=True,
+    )
+    db.add(scout)
+    await db.commit()
+    await db.refresh(scout)
+    return _to_cloud_response(scout)
+
+
+@router.put("/cloud/{scout_id}", response_model=CloudScoutResponse)
+async def update_cloud_scout(
+    scout_id: str,
+    body: CloudScoutUpdateRequest,
+    db: AsyncSession = Depends(get_session),
+    current_user: UserProfile = Depends(get_current_user),
+):
+    scout = await db.get(CloudScoutConfig, scout_id)
+    if scout is None or scout.user_id != current_user.id:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Scout not found")
+    if body.name is not None:
+        scout.name = body.name
+    if body.data_types is not None:
+        scout.data_types = body.data_types
+    if body.prompt_template is not None:
+        scout.prompt_template = body.prompt_template
+    if body.schedule_cron is not None:
+        scout.schedule_cron = body.schedule_cron
+    if body.filter_config is not None:
+        scout.filter_config = body.filter_config
+    if body.auto_trash_spam is not None:
+        scout.auto_trash_spam = body.auto_trash_spam
+    if body.enabled is not None:
+        scout.enabled = body.enabled
+    await db.commit()
+    await db.refresh(scout)
+    return _to_cloud_response(scout)
+
+
+@router.delete("/cloud/{scout_id}")
+async def delete_cloud_scout(
+    scout_id: str,
+    db: AsyncSession = Depends(get_session),
+    current_user: UserProfile = Depends(get_current_user),
+):
+    scout = await db.get(CloudScoutConfig, scout_id)
+    if scout is None or scout.user_id != current_user.id:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Scout not found")
+    # Core deletes bypass the polymorphic ScoutRunLog relationship whose
+    # varchar scout_id vs uuid id join is not directly comparable in Postgres.
+    # scout_run_logs.scout_id is a plain string (matches the str scout_id);
+    # scout_triage_queue rows cascade automatically via their FK ondelete.
+    await db.execute(sa_delete(ScoutRunLog).where(ScoutRunLog.scout_id == scout_id))
+    await db.execute(sa_delete(CloudScoutConfig).where(CloudScoutConfig.id == scout_id))
+    await db.commit()
+    return {"ok": True}
+
+
+@router.get("/cloud/{scout_id}/gmail-labels")
+async def list_gmail_labels(
+    scout_id: str,
+    db: AsyncSession = Depends(get_session),
+    current_user: UserProfile = Depends(get_current_user),
+):
+    scout = await db.get(CloudScoutConfig, scout_id)
+    if scout is None or scout.user_id != current_user.id:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Scout not found")
+    try:
+        connector = get_connector("gmail")
+    except KeyError:
+        return []
+    return await connector.list_labels(scout)
+
+
+@router.post("/cloud/{scout_id}/gmail-disconnect", response_model=CloudScoutResponse)
+async def disconnect_gmail(
+    scout_id: str,
+    db: AsyncSession = Depends(get_session),
+    current_user: UserProfile = Depends(get_current_user),
+):
+    scout = await db.get(CloudScoutConfig, scout_id)
+    if scout is None or scout.user_id != current_user.id:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Scout not found")
+    try:
+        connector = get_connector("gmail")
+        await connector.stop_watch(scout)
+    except KeyError:
+        pass
+    scout.oauth_token_encrypted = None
+    scout.gmail_history_id = None
+    scout.gmail_watch_expires_at = None
+    scout.gmail_address = None
+    scout.enabled = False
+    await db.commit()
+    await db.refresh(scout)
+    return _to_cloud_response(scout)
+
+
 # ── Gmail OAuth setup (scout-specific) ───────────────────────────────────────
 
 # Scopes required for Gmail scout connectivity.
@@ -283,11 +435,35 @@ _GMAIL_SCOUT_SCOPES = [
 _GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
 _GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
 
-# In-memory pending OAuth states for scout Gmail consent:
+# In-memory pending OAuth states for scout Gmail consent.
-# state → (code_verifier, scout_id, user_id, expires_at_epoch_s)
+#
-# Production note: replace with Redis for multi-process deployments.
+# state → {
-_pending_scout_oauth_states: dict[str, tuple[str, str, str, float]] = {}
+#   "code_verifier": str,
-_SCOUT_OAUTH_TTL_SECONDS = 600  # 10 minutes
+#   "user_id": str,
+#   "expires_at": float (epoch seconds),
+#   "mode": "reconnect" | "create",
+#   "scout_id": str | None,            # set for reconnect mode
+#   "draft": {name, prompt_template, auto_trash_spam} | None,  # set for create mode
+#   "token_encrypted": str | None,     # populated after a successful create-mode callback
+#   "gmail_address": str | None,
+# }
+#
+# Zero-trust: in create mode the encrypted Gmail token lives ONLY here, in
+# process memory, for at most _SCOUT_OAUTH_TTL_SECONDS. It is persisted to the
+# DB only when the user finalizes the scout (POST /scouts/cloud/finalize).
+# An abandoned/errored flow leaves no scout row and no stored token.
+#
+# Production note: this in-memory store is single-process only — replace with
+# Redis (keyed by state, TTL'd) for multi-worker deployments.
+_pending_scout_oauth_states: dict[str, dict] = {}
+_SCOUT_OAUTH_TTL_SECONDS = 900  # 15 minutes
+
+
+def _purge_expired_oauth_states() -> None:
+    now = time.time()
+    expired = [s for s, e in _pending_scout_oauth_states.items() if e.get("expires_at", 0) < now]
+    for s in expired:
+        del _pending_scout_oauth_states[s]
 
 
 def _scout_gmail_redirect_uri() -> str:
@@ -311,6 +487,34 @@ class _ScoutGmailCallbackBody(BaseModel):
     state: str
 
 
+class _ScoutGmailAuthorizeDraftBody(BaseModel):
+    name: str
+    prompt_template: str = ""
+    auto_trash_spam: bool = False
+
+
+class _ScoutGmailFinalizeBody(BaseModel):
+    session: str
+    filter_config: dict | None = None
+
+
+def _build_gmail_authorize_url(state: str, code_challenge: str) -> str:
+    """Build the Google consent URL for the scout Gmail flow (shared by both modes)."""
+    redirect_uri = _scout_gmail_redirect_uri()
+    params = {
+        "client_id": settings.GOOGLE_AUTH_CLIENT_ID,
+        "redirect_uri": redirect_uri,
+        "response_type": "code",
+        "scope": " ".join(_GMAIL_SCOUT_SCOPES),
+        "state": state,
+        "code_challenge": code_challenge,
+        "code_challenge_method": "S256",
+        "access_type": "offline",
+        "prompt": "consent",
+    }
+    return f"{_GOOGLE_AUTH_URL}?{urllib.parse.urlencode(params)}"
+
+
 @router.get("/oauth/gmail/authorize", response_model=_ScoutGmailAuthorizeResponse)
 async def scout_gmail_oauth_authorize(
     scout_id: str,
@@ -331,28 +535,63 @@ async def scout_gmail_oauth_authorize(
     code_verifier, code_challenge = generate_pkce_pair()
     state = secrets.token_urlsafe(32)
 
-    # Purge expired states to prevent unbounded growth.
+    _purge_expired_oauth_states()
-    now = time.time()
-    expired = [s for s, (_, _, _, exp) in _pending_scout_oauth_states.items() if exp < now]
-    for s in expired:
-        del _pending_scout_oauth_states[s]
 
-    _pending_scout_oauth_states[state] = (code_verifier, scout_id, current_user.id, now + _SCOUT_OAUTH_TTL_SECONDS)
+    _pending_scout_oauth_states[state] = {
-
+        "code_verifier": code_verifier,
-    redirect_uri = _scout_gmail_redirect_uri()
+        "user_id": current_user.id,
-    params = {
+        "expires_at": time.time() + _SCOUT_OAUTH_TTL_SECONDS,
-        "client_id": settings.GOOGLE_AUTH_CLIENT_ID,
+        "mode": "reconnect",
-        "redirect_uri": redirect_uri,
+        "scout_id": scout_id,
-        "response_type": "code",
+        "draft": None,
-        "scope": " ".join(_GMAIL_SCOUT_SCOPES),
+        "token_encrypted": None,
-        "state": state,
+        "gmail_address": None,
-        "code_challenge": code_challenge,
-        "code_challenge_method": "S256",
-        "access_type": "offline",
-        "prompt": "consent",
     }
-    authorize_url = f"{_GOOGLE_AUTH_URL}?{urllib.parse.urlencode(params)}"
+
-    return _ScoutGmailAuthorizeResponse(authorize_url=authorize_url)
+    return _ScoutGmailAuthorizeResponse(
+        authorize_url=_build_gmail_authorize_url(state, code_challenge)
+    )
+
+
+@router.post("/oauth/gmail/authorize-draft", response_model=_ScoutGmailAuthorizeResponse)
+async def scout_gmail_oauth_authorize_draft(
+    body: _ScoutGmailAuthorizeDraftBody,
+    current_user: UserProfile = Depends(get_current_user),
+) -> _ScoutGmailAuthorizeResponse:
+    """Start the Gmail OAuth flow in *creation* mode — no scout row exists yet.
+
+    The draft scout fields are held in the pending OAuth session; the scout is
+    only created once the user finalizes (POST /scouts/cloud/finalize).
+    """
+    if not settings.GOOGLE_AUTH_CLIENT_ID or not settings.GOOGLE_AUTH_CLIENT_SECRET:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "Google OAuth is not configured on this server",
+        )
+
+    code_verifier, code_challenge = generate_pkce_pair()
+    state = secrets.token_urlsafe(32)
+
+    _purge_expired_oauth_states()
+
+    _pending_scout_oauth_states[state] = {
+        "code_verifier": code_verifier,
+        "user_id": current_user.id,
+        "expires_at": time.time() + _SCOUT_OAUTH_TTL_SECONDS,
+        "mode": "create",
+        "scout_id": None,
+        "draft": {
+            "name": body.name,
+            "prompt_template": body.prompt_template,
+            "auto_trash_spam": body.auto_trash_spam,
+        },
+        "token_encrypted": None,
+        "gmail_address": None,
+    }
+
+    return _ScoutGmailAuthorizeResponse(
+        authorize_url=_build_gmail_authorize_url(state, code_challenge)
+    )
 
 
 @router.get("/oauth/gmail/web-callback", include_in_schema=False)
@@ -379,10 +618,16 @@ async def scout_gmail_oauth_callback(
     the ``code`` and ``state`` params.
     """
     entry = _pending_scout_oauth_states.pop(body.state, None)
-    if entry is None or entry[3] < time.time() or entry[2] != current_user.id:
+    if (
+        entry is None
+        or entry["expires_at"] < time.time()
+        or entry["user_id"] != current_user.id
+    ):
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid or expired OAuth state")
 
-    code_verifier, scout_id, _, _ = entry
+    code_verifier = entry["code_verifier"]
+    mode = entry["mode"]
+    scout_id = entry.get("scout_id")
 
     redirect_uri = _scout_gmail_redirect_uri()
 
@@ -420,14 +665,48 @@ async def scout_gmail_oauth_callback(
     }
     encrypted = encrypt_token(creds_dict)
 
+    # Fetch the connected Gmail address for display.
+    gmail_address: str | None = None
+    try:
+        from googleapiclient.discovery import build
+        from google.oauth2.credentials import Credentials
+
+        def _fetch_email() -> str | None:
+            creds = Credentials(
+                token=creds_dict["token"],
+                refresh_token=creds_dict.get("refresh_token"),
+                token_uri=creds_dict["token_uri"],
+                client_id=creds_dict["client_id"],
+                client_secret=creds_dict["client_secret"],
+                scopes=creds_dict["scopes"],
+            )
+            service = build("gmail", "v1", credentials=creds, cache_discovery=False)
+            profile = service.users().getProfile(userId="me").execute()
+            return profile.get("emailAddress")
+
+        gmail_address = await asyncio.to_thread(_fetch_email)
+    except Exception:
+        logger.exception("failed to fetch gmail address (mode=%s)", mode)
+
+    if mode == "create":
+        # Do NOT create a scout yet. Hold the encrypted token + address in the
+        # transient in-memory session; the scout is created at finalize.
+        entry["token_encrypted"] = encrypted
+        entry["gmail_address"] = gmail_address
+        entry["expires_at"] = time.time() + _SCOUT_OAUTH_TTL_SECONDS
+        _pending_scout_oauth_states[body.state] = entry
+        return {"ok": True, "session_id": body.state, "gmail_address": gmail_address}
+
+    # mode == "reconnect": update the existing scout in place.
     scout = await db.get(CloudScoutConfig, scout_id)
     if scout is None or scout.user_id != current_user.id:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "Scout not found")
     scout.oauth_token_encrypted = encrypted
+    scout.gmail_address = gmail_address
+
     await db.commit()
 
     # Attempt to set up Gmail push watch so we start receiving Pub/Sub notifications.
-    from app.scouts.connectors.registry import get_connector
     try:
         connector = get_connector("gmail")
         await connector.setup_watch(scout)
@@ -437,4 +716,92 @@ async def scout_gmail_oauth_callback(
     except Exception:
         logger.exception("setup_watch failed for scout %s", scout_id)
 
-    return {"ok": True}
+    return {"ok": True, "session_id": None, "gmail_address": gmail_address}
+
+
+@router.get("/oauth/gmail/session-labels")
+async def scout_gmail_session_labels(
+    session: str,
+    current_user: UserProfile = Depends(get_current_user),
+) -> list[dict]:
+    """List Gmail labels for a pending create-mode OAuth session (no scout row yet).
+
+    Builds a Gmail service from the session's transient decrypted token.
+    Returns [] on any error.
+    """
+    entry = _pending_scout_oauth_states.get(session)
+    if (
+        entry is None
+        or entry["expires_at"] < time.time()
+        or entry["user_id"] != current_user.id
+        or entry.get("token_encrypted") is None
+    ):
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Session not found or expired")
+
+    try:
+        from app.scouts.connectors.gmail import _gmail_service_from_token
+
+        creds = decrypt_token(entry["token_encrypted"])
+
+        def _sync() -> list[dict]:
+            service = _gmail_service_from_token(creds)
+            resp = service.users().labels().list(userId="me").execute()
+            return [{"id": lbl["id"], "name": lbl["name"]} for lbl in resp.get("labels", [])]
+
+        return await asyncio.to_thread(_sync)
+    except Exception:
+        logger.exception("session-labels failed for session %s", session)
+        return []
+
+
+@router.post("/cloud/finalize", response_model=CloudScoutResponse, status_code=status.HTTP_201_CREATED)
+async def finalize_cloud_scout(
+    body: _ScoutGmailFinalizeBody,
+    db: AsyncSession = Depends(get_session),
+    current_user: UserProfile = Depends(get_current_user),
+):
+    """Create the cloud scout from a completed create-mode OAuth session.
+
+    This is the only path that persists the Gmail token for a newly-created
+    scout. Abandoned flows never reach here, so they leave no orphan rows.
+    """
+    entry = _pending_scout_oauth_states.pop(body.session, None)
+    if (
+        entry is None
+        or entry["expires_at"] < time.time()
+        or entry["user_id"] != current_user.id
+        or entry.get("mode") != "create"
+        or entry.get("token_encrypted") is None
+    ):
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid or expired OAuth session")
+
+    draft = entry["draft"] or {}
+    scout = CloudScoutConfig(
+        id=str(uuid.uuid4()),
+        user_id=current_user.id,
+        provider="gmail",
+        name=draft.get("name", ""),
+        data_types=[],
+        prompt_template=draft.get("prompt_template", ""),
+        filter_config=body.filter_config,
+        schedule_cron=_DEFAULT_CLOUD_SCHEDULE,
+        auto_trash_spam=draft.get("auto_trash_spam", False),
+        enabled=True,
+        oauth_token_encrypted=entry["token_encrypted"],
+        gmail_address=entry.get("gmail_address"),
+    )
+    db.add(scout)
+    await db.commit()
+    await db.refresh(scout)
+
+    # Best-effort Gmail push watch — failure must not block scout creation.
+    try:
+        connector = get_connector("gmail")
+        await connector.setup_watch(scout)
+        await db.commit()
+    except KeyError:
+        logger.warning("gmail connector not registered — skipping setup_watch for scout %s", scout.id)
+    except Exception:
+        logger.exception("setup_watch failed for scout %s", scout.id)
+
+    return _to_cloud_response(scout)

app/models.py

@@ -223,6 +223,7 @@ class CloudScoutConfig(Base):
     gmail_history_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
     gmail_watch_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     device_inactivity_pause_days: Mapped[int] = mapped_column(Integer, nullable=False, default=14, server_default="14")
+    gmail_address: Mapped[str | None] = mapped_column(String(320), nullable=True)
 
     run_logs: Mapped[list["ScoutRunLog"]] = relationship(
         back_populates="cloud_scout",

app/schemas/__init__.py

@@ -276,6 +276,46 @@ class ScoutRunLogResponse(BaseModel):
     completed_at: int | None
 
 
+# ── Cloud Scout CRUD ──────────────────────────────────────────────────
+
+class CloudScoutCreateRequest(BaseModel):
+    name: str
+    provider: Literal["gmail", "teams", "outlook"]
+    data_types: list[str] = Field(default_factory=list)
+    prompt_template: str = ""
+    schedule_cron: str | None = None        # None → server default
+    filter_config: dict | None = None
+    auto_trash_spam: bool = False
+
+
+class CloudScoutUpdateRequest(BaseModel):
+    name: str | None = None
+    data_types: list[str] | None = None
+    prompt_template: str | None = None
+    schedule_cron: str | None = None
+    filter_config: dict | None = None
+    auto_trash_spam: bool | None = None
+    enabled: bool | None = None
+
+
+class CloudScoutResponse(BaseModel):
+    id: str
+    user_id: str
+    provider: str
+    name: str
+    data_types: list[str]
+    prompt_template: str
+    schedule_cron: str
+    filter_config: dict | None
+    auto_trash_spam: bool
+    enabled: bool
+    last_run_at: int | None
+    gmail_address: str | None
+    oauth_connected: bool
+    created_at: int
+    updated_at: int
+
+
 # ── Chatbot Journey ───────────────────────────────────────────────────
 
 

app/scouts/connectors/gmail.py

@@ -16,18 +16,44 @@ from datetime import datetime, timezone
 
 from app.config.settings import settings
 from app.integrations import decrypt_token
-from app.integrations.gmail import GmailClient
 from app.scouts.connectors.base import ItemContent, ItemMetadata, ItemRef
 
 logger = logging.getLogger(__name__)
 
 
-def _get_gmail_service(scout):
+def _extract_plain_text_body(payload: dict) -> str:
-    """Return a synchronous Google API client for low-level metadata/history calls."""
+    """Recursively walk a Gmail message payload to find text/plain content."""
+    import base64
+    mime_type = payload.get("mimeType", "")
+    if mime_type == "text/plain":
+        data = payload.get("body", {}).get("data", "")
+        if data:
+            return base64.urlsafe_b64decode(data + "==").decode("utf-8", errors="replace")
+        return ""
+    if mime_type.startswith("multipart/"):
+        for part in payload.get("parts", []):
+            text = _extract_plain_text_body(part)
+            if text:
+                return text
+    # text/html fallback: strip tags rudimentarily if no text/plain part
+    if mime_type == "text/html":
+        data = payload.get("body", {}).get("data", "")
+        if data:
+            import re
+            html = base64.urlsafe_b64decode(data + "==").decode("utf-8", errors="replace")
+            return re.sub(r"<[^>]+>", " ", html)
+    return ""
+
+
+def _gmail_service_from_token(creds_info: dict):
+    """Build a synchronous Gmail API client from a decrypted credentials dict.
+
+    Shared by ``_get_gmail_service`` (scout-backed) and the pending-session
+    OAuth flow which has a raw token but no scout row yet.
+    """
     from googleapiclient.discovery import build
     from google.oauth2.credentials import Credentials
 
-    creds_info = decrypt_token(scout.oauth_token_encrypted)
     credentials = Credentials(
         token=creds_info.get("token"),
         refresh_token=creds_info.get("refresh_token"),
@@ -39,6 +65,12 @@ def _get_gmail_service(scout):
     return build("gmail", "v1", credentials=credentials, cache_discovery=False)
 
 
+def _get_gmail_service(scout):
+    """Return a synchronous Google API client for low-level metadata/history calls."""
+    creds_info = decrypt_token(scout.oauth_token_encrypted)
+    return _gmail_service_from_token(creds_info)
+
+
 class GmailConnector:
     source_type = "gmail"
 
@@ -118,33 +150,28 @@ class GmailConnector:
     # ── fetch_content ─────────────────────────────────────────────────────
 
     async def fetch_content(self, scout, ref: ItemRef) -> ItemContent:
-        """Fetch full body text via GmailClient — transient, must not be persisted."""
+        """Fetch full body text for a single message — transient, must not be persisted."""
-        creds_info = decrypt_token(scout.oauth_token_encrypted)
-        client = GmailClient(creds_info)
-        # fetch_messages returns EmailMessage dataclasses with body_text already
-        # extracted and decoded. We pass an empty filter to avoid narrowing by
-        # date — callers should only invoke fetch_content for known-new messages.
-        messages = await client.fetch_messages(filter_config=None, since=None)
-
-        # Pick the message matching our ref (or fall back to first if only one returned).
-        email_msg = next(
-            (m for m in messages if m.id == ref.source_msg_ref),
-            messages[0] if messages else None,
-        )
-        if email_msg is None:
-            raise ValueError(f"Message {ref.source_msg_ref!r} not found via GmailClient")
 
+        def _sync() -> ItemContent:
+            service = _get_gmail_service(scout)
+            msg = service.users().messages().get(
+                userId="me", id=ref.source_msg_ref, format="full",
+            ).execute()
+            headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
+            body_text = _extract_plain_text_body(msg.get("payload", {}))
             return ItemContent(
                 metadata=ItemMetadata(
-                subject=email_msg.subject,
+                    subject=headers.get("Subject"),
-                sender=email_msg.sender,
+                    sender=headers.get("From"),
-                snippet=None,
+                    snippet=msg.get("snippet"),
-                received_at=email_msg.date,
+                    received_at=None,
                 ),
-            body_text=email_msg.body_text,
+                body_text=body_text,
-            raw_headers={},
+                raw_headers=headers,
             )
 
+        return await asyncio.to_thread(_sync)
+
     # ── archive ───────────────────────────────────────────────────────────
 
     async def archive(self, scout, ref: ItemRef) -> None:
@@ -193,3 +220,29 @@ class GmailConnector:
     async def renew_watch(self, scout) -> None:
         """Renew an existing Gmail Pub/Sub watch (same as setup_watch)."""
         await self.setup_watch(scout)
+
+    async def list_labels(self, scout) -> list[dict]:
+        """Return the account's Gmail labels as [{id, name}]. Empty if no token."""
+        if not scout.oauth_token_encrypted:
+            return []
+
+        def _sync() -> list[dict]:
+            service = _get_gmail_service(scout)
+            resp = service.users().labels().list(userId="me").execute()
+            return [{"id": lbl["id"], "name": lbl["name"]} for lbl in resp.get("labels", [])]
+
+        return await asyncio.to_thread(_sync)
+
+    async def stop_watch(self, scout) -> None:
+        """Stop Gmail push notifications. Swallows errors (watch may be gone)."""
+        if not scout.oauth_token_encrypted:
+            return
+
+        def _sync() -> None:
+            service = _get_gmail_service(scout)
+            service.users().stop(userId="me").execute()
+
+        try:
+            await asyncio.to_thread(_sync)
+        except Exception:
+            logger.exception("stop_watch failed for scout %s", scout.id)

app/scouts/engine.py

@@ -144,6 +144,7 @@ class ScoutEngine:
                     ScoutTriageQueue.status == "queued",
                 )
             )).scalars().all()
+            logger.info("deliver_pending: user=%s found %d queued rows", user_id, len(rows))
 
             for row in rows:
                 try:
@@ -173,7 +174,9 @@ class ScoutEngine:
                         "payload": None,
                     },
                 }
+                logger.info("deliver_pending: sending proposal id=%s subject=%r", row.id, meta.subject)
                 await ws.send_json(payload)
+                logger.info("deliver_pending: send_json returned for proposal id=%s", row.id)
                 row.status = "delivered"
                 row.delivered_at = datetime.now(tz=timezone.utc)
 

scripts/inspect_gmail_scout_token.py

@@ -0,0 +1,56 @@
+"""Decrypt and inspect the Gmail scout's stored OAuth token.
+
+Shows what scopes were granted at consent time. If gmail.readonly / gmail.modify
+are missing, the consent screen didn't actually grant them.
+
+Usage:
+    python scripts/inspect_gmail_scout_token.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+
+_API_ROOT = Path(__file__).resolve().parent.parent
+if str(_API_ROOT) not in sys.path:
+    sys.path.insert(0, str(_API_ROOT))
+
+from sqlalchemy import select
+
+from app.db import async_session
+from app.integrations import decrypt_token
+from app.models import CloudScoutConfig
+
+
+async def main() -> None:
+    async with async_session() as session:
+        scouts = (
+            await session.execute(
+                select(CloudScoutConfig).where(CloudScoutConfig.provider == "gmail")
+            )
+        ).scalars().all()
+
+    if not scouts:
+        print("No Gmail scouts found.")
+        return
+
+    for scout in scouts:
+        print(f"\nScout: {scout.name} (id={scout.id})")
+        if not scout.oauth_token_encrypted:
+            print("  (no token stored)")
+            continue
+        try:
+            creds = decrypt_token(scout.oauth_token_encrypted)
+        except Exception as exc:
+            print(f"  decrypt failed: {exc}")
+            continue
+        print(f"  has refresh_token: {bool(creds.get('refresh_token'))}")
+        print(f"  stored scopes:     {creds.get('scopes')}")
+        print(f"  token_uri:         {creds.get('token_uri')}")
+        print(f"  client_id (last 8): ...{(creds.get('client_id') or '')[-8:]}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

scripts/reset_triage_queue_to_queued.py

@@ -0,0 +1,35 @@
+"""Re-queue all delivered (but not acked) triage rows so deliver_pending sends them again.
+
+Usage:
+    python scripts/reset_triage_queue_to_queued.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+
+_API_ROOT = Path(__file__).resolve().parent.parent
+if str(_API_ROOT) not in sys.path:
+    sys.path.insert(0, str(_API_ROOT))
+
+from sqlalchemy import update
+
+from app.db import async_session
+from app.models import ScoutTriageQueue
+
+
+async def main() -> None:
+    async with async_session() as session:
+        result = await session.execute(
+            update(ScoutTriageQueue)
+            .where(ScoutTriageQueue.status == "delivered")
+            .values(status="queued", delivered_at=None)
+        )
+        await session.commit()
+        print(f"Reset {result.rowcount} rows from delivered → queued")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

scripts/show_gmail_scout_state.py

@@ -0,0 +1,59 @@
+"""Print Gmail scout state for debugging.
+
+Usage:
+    python scripts/show_gmail_scout_state.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+
+_API_ROOT = Path(__file__).resolve().parent.parent
+if str(_API_ROOT) not in sys.path:
+    sys.path.insert(0, str(_API_ROOT))
+
+from sqlalchemy import select, func
+
+from app.db import async_session
+from app.models import CloudScoutConfig, ScoutTriageQueue, ScoutRunLog
+
+
+async def main() -> None:
+    async with async_session() as session:
+        scouts = (
+            await session.execute(
+                select(CloudScoutConfig).where(CloudScoutConfig.provider == "gmail")
+            )
+        ).scalars().all()
+
+        for scout in scouts:
+            print(f"\nScout: {scout.name} (id={scout.id})")
+            print(f"  enabled:                  {scout.enabled}")
+            print(f"  gmail_history_id:         {scout.gmail_history_id}")
+            print(f"  gmail_watch_expires_at:   {scout.gmail_watch_expires_at}")
+            print(f"  auto_trash_spam:          {scout.auto_trash_spam}")
+            print(f"  last_run_at:              {scout.last_run_at}")
+
+            queued_count = (
+                await session.execute(
+                    select(func.count())
+                    .select_from(ScoutTriageQueue)
+                    .where(ScoutTriageQueue.scout_id == scout.id)
+                )
+            ).scalar()
+            print(f"  triage_queue rows:        {queued_count}")
+
+            run_count = (
+                await session.execute(
+                    select(func.count())
+                    .select_from(ScoutRunLog)
+                    .where(ScoutRunLog.scout_id == scout.id)
+                )
+            ).scalar()
+            print(f"  scout_run_logs:           {run_count}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

scripts/trigger_gmail_scout.py

@@ -0,0 +1,74 @@
+"""Manually trigger the user's Gmail scout for testing.
+
+Usage:
+    python scripts/trigger_gmail_scout.py [user_email]
+
+If user_email omitted, picks the first user with a Gmail scout.
+Runs ScoutEngine.trigger_scout — which calls Gmail history.list since last
+gmail_history_id, fetches each new message, runs LLM triage, inserts queue rows
+for relevant items.
+
+After running, check the queue:
+    psql -d adiuvai -c "select source_msg_ref, triage_verdict, status from scout_triage_queue order by triaged_at desc limit 10"
+
+Then restart the Electron app to trigger deliver_pending → frames → local
+scout_suggestions rows.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import uuid
+from pathlib import Path
+
+# Ensure api/ root is importable when running from scripts/ subdir
+_API_ROOT = Path(__file__).resolve().parent.parent
+if str(_API_ROOT) not in sys.path:
+    sys.path.insert(0, str(_API_ROOT))
+
+from sqlalchemy import select
+
+from app.db import async_session
+from app.models import CloudScoutConfig, User
+from app.scouts.connectors.gmail import GmailConnector
+from app.scouts.connectors.registry import register_connector
+from app.scouts.engine import ScoutEngine
+
+
+async def main() -> None:
+    register_connector(GmailConnector())
+
+    target_email = sys.argv[1] if len(sys.argv) > 1 else None
+
+    async with async_session() as session:
+        q = select(CloudScoutConfig).where(
+            CloudScoutConfig.provider == "gmail",
+            CloudScoutConfig.enabled.is_(True),
+        )
+        if target_email:
+            user = (
+                await session.execute(select(User).where(User.email == target_email))
+            ).scalar_one_or_none()
+            if user is None:
+                print(f"No user with email {target_email}")
+                return
+            q = q.where(CloudScoutConfig.user_id == user.id)
+
+        scouts = (await session.execute(q)).scalars().all()
+
+    if not scouts:
+        print("No enabled Gmail scouts found. Create one in Settings → Scouts first.")
+        return
+
+    for scout in scouts:
+        print(f"Triggering scout id={scout.id} name={scout.name!r} user={scout.user_id}")
+        try:
+            await ScoutEngine().trigger_scout(uuid.UUID(scout.id))
+            print("  → done")
+        except Exception as exc:
+            print(f"  → failed: {exc}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/test_scout_cloud_crud.py

@@ -0,0 +1,229 @@
+"""Tests for cloud scout CRUD routes."""
+
+from __future__ import annotations
+
+import time
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import select
+
+from app.db import get_session
+from app.integrations import encrypt_token
+from app.main import app
+from app.models import CloudScoutConfig
+from tests.conftest import _TestSessionLocal, make_jwt, TEST_USER_IDS
+
+
+def _auth_headers(tier: str = "power") -> dict:
+    return {"Authorization": f"Bearer {make_jwt(tier)}"}
+
+
+async def _test_get_session():
+    async with _TestSessionLocal() as session:
+        yield session
+
+
+@pytest.fixture(autouse=True)
+def _override_session():
+    # FastAPI resolves Depends() by the original function object, so patching the
+    # module-level name does not take effect — use dependency_overrides instead.
+    app.dependency_overrides[get_session] = _test_get_session
+    yield
+    app.dependency_overrides.pop(get_session, None)
+
+
+@pytest.mark.asyncio
+async def test_create_cloud_scout_defaults_schedule():
+    payload = {
+        "name": "Inbox",
+        "provider": "gmail",
+        "data_types": [],
+        "prompt_template": "client requests",
+        "auto_trash_spam": True,
+        # schedule_cron omitted → server default
+    }
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        resp = await client.post("/api/v1/scouts/cloud", json=payload, headers=_auth_headers())
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    assert body["name"] == "Inbox"
+    assert body["provider"] == "gmail"
+    assert body["auto_trash_spam"] is True
+    assert body["prompt_template"] == "client requests"
+    assert body["schedule_cron"]  # non-empty default applied
+    assert body["oauth_connected"] is False
+    assert body["gmail_address"] is None
+
+
+@pytest.mark.asyncio
+async def test_list_cloud_scouts_returns_only_own():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        await client.post(
+            "/api/v1/scouts/cloud",
+            json={"name": "A", "provider": "gmail"},
+            headers=_auth_headers(),
+        )
+        resp = await client.get("/api/v1/scouts/cloud", headers=_auth_headers())
+    assert resp.status_code == 200
+    rows = resp.json()
+    assert all(r["provider"] == "gmail" for r in rows)
+    assert any(r["name"] == "A" for r in rows)
+
+
+@pytest.mark.asyncio
+async def test_update_cloud_scout_applies_filter_and_autotrash():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        created = (await client.post(
+            "/api/v1/scouts/cloud",
+            json={"name": "B", "provider": "gmail"},
+            headers=_auth_headers(),
+        )).json()
+        sid = created["id"]
+        resp = await client.put(
+            f"/api/v1/scouts/cloud/{sid}",
+            json={"filter_config": {"labels": ["INBOX"], "senders": ["@client.co"]}, "auto_trash_spam": True, "prompt_template": "invoices"},
+            headers=_auth_headers(),
+        )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["filter_config"] == {"labels": ["INBOX"], "senders": ["@client.co"]}
+    assert body["auto_trash_spam"] is True
+    assert body["prompt_template"] == "invoices"
+
+
+@pytest.mark.asyncio
+async def test_delete_cloud_scout():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        created = (await client.post(
+            "/api/v1/scouts/cloud",
+            json={"name": "C", "provider": "gmail"},
+            headers=_auth_headers(),
+        )).json()
+        sid = created["id"]
+        resp = await client.delete(f"/api/v1/scouts/cloud/{sid}", headers=_auth_headers())
+        assert resp.status_code == 200
+        listing = (await client.get("/api/v1/scouts/cloud", headers=_auth_headers())).json()
+    assert all(r["id"] != sid for r in listing)
+
+
+@pytest.mark.asyncio
+async def test_gmail_labels_route_returns_labels():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        created = (await client.post(
+            "/api/v1/scouts/cloud",
+            json={"name": "L", "provider": "gmail"},
+            headers=_auth_headers(),
+        )).json()
+        sid = created["id"]
+
+        with patch("app.api.routes.scouts.get_connector") as mock_get:
+            mock_get.return_value.list_labels = AsyncMock(return_value=[{"id": "INBOX", "name": "INBOX"}])
+            resp = await client.get(f"/api/v1/scouts/cloud/{sid}/gmail-labels", headers=_auth_headers())
+    assert resp.status_code == 200
+    assert resp.json() == [{"id": "INBOX", "name": "INBOX"}]
+
+
+@pytest.mark.asyncio
+async def test_gmail_disconnect_clears_token():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        created = (await client.post(
+            "/api/v1/scouts/cloud",
+            json={"name": "D", "provider": "gmail"},
+            headers=_auth_headers(),
+        )).json()
+        sid = created["id"]
+        # mark it connected directly in the DB
+        async with _TestSessionLocal() as session:
+            row = await session.get(CloudScoutConfig, sid)
+            row.oauth_token_encrypted = "blob"
+            row.gmail_address = "a@b.com"
+            await session.commit()
+
+        with patch("app.api.routes.scouts.get_connector") as mock_get:
+            mock_get.return_value.stop_watch = AsyncMock()
+            resp = await client.post(f"/api/v1/scouts/cloud/{sid}/gmail-disconnect", headers=_auth_headers())
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["oauth_connected"] is False
+    assert body["gmail_address"] is None
+    assert body["enabled"] is False
+
+
+# ── Pending-session create-at-finalize flow ───────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_authorize_draft_returns_url_and_no_scout_created():
+    from app.config.settings import settings as app_settings
+
+    with patch.object(app_settings, "GOOGLE_AUTH_CLIENT_ID", "cid"), \
+         patch.object(app_settings, "GOOGLE_AUTH_CLIENT_SECRET", "secret"):
+        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+            resp = await client.post(
+                "/api/v1/scouts/oauth/gmail/authorize-draft",
+                json={"name": "Draft Inbox", "prompt_template": "invoices", "auto_trash_spam": True},
+                headers=_auth_headers(),
+            )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["authorize_url"].startswith("https://accounts.google.com/")
+
+    # No scout row should have been created by authorize-draft.
+    async with _TestSessionLocal() as session:
+        rows = (await session.execute(
+            select(CloudScoutConfig).where(
+                CloudScoutConfig.user_id == TEST_USER_IDS["power"],
+                CloudScoutConfig.name == "Draft Inbox",
+            )
+        )).scalars().all()
+    assert rows == []
+
+
+@pytest.mark.asyncio
+async def test_finalize_creates_scout_from_session():
+    from app.api.routes import scouts as scouts_mod
+
+    state = "test-session-" + uuid.uuid4().hex
+    token = encrypt_token({"token": "x", "refresh_token": "y", "client_id": "c", "client_secret": "s"})
+    scouts_mod._pending_scout_oauth_states[state] = {
+        "code_verifier": "v",
+        "user_id": TEST_USER_IDS["power"],
+        "expires_at": time.time() + 600,
+        "mode": "create",
+        "scout_id": None,
+        "draft": {"name": "Finalized", "prompt_template": "tasks", "auto_trash_spam": True},
+        "token_encrypted": token,
+        "gmail_address": "me@gmail.com",
+    }
+
+    # Patch get_connector to raise KeyError so setup_watch is skipped (best-effort).
+    with patch("app.api.routes.scouts.get_connector", side_effect=KeyError("gmail")):
+        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+            resp = await client.post(
+                "/api/v1/scouts/cloud/finalize",
+                json={"session": state, "filter_config": {"labels": ["INBOX"]}},
+                headers=_auth_headers(),
+            )
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    assert body["name"] == "Finalized"
+    assert body["auto_trash_spam"] is True
+    assert body["filter_config"] == {"labels": ["INBOX"]}
+    assert body["gmail_address"] == "me@gmail.com"
+    assert body["oauth_connected"] is True
+
+    # Session must have been popped.
+    assert state not in scouts_mod._pending_scout_oauth_states
+
+
+@pytest.mark.asyncio
+async def test_finalize_rejects_unknown_session():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        resp = await client.post(
+            "/api/v1/scouts/cloud/finalize",
+            json={"session": "does-not-exist", "filter_config": None},
+            headers=_auth_headers(),
+        )
+    assert resp.status_code == 401, resp.text

tests/test_scout_connectors_gmail.py

@@ -3,8 +3,7 @@
 from __future__ import annotations
 
 import uuid
-from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
-from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -53,16 +52,24 @@ async def test_fetch_metadata_returns_subject_and_snippet():
 
 @pytest.mark.asyncio
 async def test_fetch_content_returns_body_text():
+    import base64
     scout = _make_scout()
     conn = GmailConnector()
-    # decrypt_token is patched because the test doesn't set OAUTH_ENCRYPTION_KEY.
+    body_data = base64.urlsafe_b64encode(b"hello world").decode()
-    with patch("app.scouts.connectors.gmail.decrypt_token", return_value={}), \
+    fake_message = {
-         patch("app.scouts.connectors.gmail.GmailClient") as MockClient:
+        "id": "msg-1",
-        instance = MockClient.return_value
+        "snippet": "hello world",
-        instance.fetch_messages = AsyncMock(return_value=[
+        "payload": {
-            MagicMock(id="msg-1", subject="S", sender="a@b", body_text="hello world",
+            "mimeType": "text/plain",
-                     date=datetime.now(tz=timezone.utc), labels=[]),
+            "headers": [
-        ])
+                {"name": "Subject", "value": "S"},
+                {"name": "From", "value": "a@b"},
+            ],
+            "body": {"data": body_data},
+        },
+    }
+    with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
+        mock_svc.return_value.users().messages().get().execute.return_value = fake_message
         content = await conn.fetch_content(scout, ItemRef(source_msg_ref="msg-1"))
     assert content.body_text == "hello world"
     assert content.metadata.subject == "S"
@@ -75,3 +82,27 @@ async def test_archive_calls_trash():
     with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
         await conn.archive(scout, ItemRef(source_msg_ref="msg-1"))
         mock_svc.return_value.users().messages().trash.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_list_labels_returns_id_and_name():
+    scout = _make_scout()
+    conn = GmailConnector()
+    fake = {"labels": [
+        {"id": "INBOX", "name": "INBOX", "type": "system"},
+        {"id": "Label_1", "name": "Work", "type": "user"},
+    ]}
+    with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
+        mock_svc.return_value.users().labels().list().execute.return_value = fake
+        labels = await conn.list_labels(scout)
+    assert {"id": "INBOX", "name": "INBOX"} in labels
+    assert {"id": "Label_1", "name": "Work"} in labels
+
+
+@pytest.mark.asyncio
+async def test_stop_watch_calls_stop():
+    scout = _make_scout()
+    conn = GmailConnector()
+    with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
+        await conn.stop_watch(scout)
+        mock_svc.return_value.users().stop.assert_called()