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>

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,7 +39,7 @@ 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 (
@@ -371,7 +371,12 @@ async def delete_cloud_scout(
     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")
-    await db.delete(scout)
+    # 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}
 
@@ -430,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:
-# state → (code_verifier, scout_id, user_id, expires_at_epoch_s)
-# Production note: replace with Redis for multi-process deployments.
-_pending_scout_oauth_states: dict[str, tuple[str, str, str, float]] = {}
-_SCOUT_OAUTH_TTL_SECONDS = 600  # 10 minutes
+# In-memory pending OAuth states for scout Gmail consent.
+#
+# state → {
+#   "code_verifier": str,
+#   "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:
@@ -458,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,
@@ -478,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.
-    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]
+    _purge_expired_oauth_states()
 
-    _pending_scout_oauth_states[state] = (code_verifier, scout_id, current_user.id, now + _SCOUT_OAUTH_TTL_SECONDS)
-
-    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",
+    _pending_scout_oauth_states[state] = {
+        "code_verifier": code_verifier,
+        "user_id": current_user.id,
+        "expires_at": time.time() + _SCOUT_OAUTH_TTL_SECONDS,
+        "mode": "reconnect",
+        "scout_id": scout_id,
+        "draft": None,
+        "token_encrypted": None,
+        "gmail_address": None,
     }
-    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)
@@ -526,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()
 
@@ -567,12 +665,8 @@ async def scout_gmail_oauth_callback(
     }
     encrypted = encrypt_token(creds_dict)
 
-    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
-
     # Fetch the connected Gmail address for display.
+    gmail_address: str | None = None
     try:
         from googleapiclient.discovery import build
         from google.oauth2.credentials import Credentials
@@ -590,9 +684,25 @@ async def scout_gmail_oauth_callback(
             profile = service.users().getProfile(userId="me").execute()
             return profile.get("emailAddress")
 
-        scout.gmail_address = await asyncio.to_thread(_fetch_email)
+        gmail_address = await asyncio.to_thread(_fetch_email)
     except Exception:
-        logger.exception("failed to fetch gmail address for scout %s", scout_id)
+        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()
 
@@ -606,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/scouts/connectors/gmail.py

@@ -45,12 +45,15 @@ def _extract_plain_text_body(payload: dict) -> str:
     return ""
 
 
-def _get_gmail_service(scout):
-    """Return a synchronous Google API client for low-level metadata/history calls."""
+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"),
@@ -62,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"
 

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

@@ -2,16 +2,19 @@
 
 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
+from tests.conftest import _TestSessionLocal, make_jwt, TEST_USER_IDS
 
 
 def _auth_headers(tier: str = "power") -> dict:
@@ -147,3 +150,80 @@ async def test_gmail_disconnect_clears_token():
     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