feat(scouts): pending-session Gmail OAuth — create cloud scout at finalize

Refactor _pending_scout_oauth_states from a tuple to a dict carrying
mode (reconnect|create), draft fields, and a transient encrypted token.
Add authorize-draft, session-labels, and cloud/finalize endpoints so the
scout row is created only when the flow completes — abandoned flows leave
no orphan rows. Zero-trust: the encrypted token lives only in the in-memory
session (<=15 min) until finalize persists it.

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

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