feat(scouts): add gmail label-list + disconnect routes

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

@@ -41,7 +41,11 @@ from app.core.note_summarizer import generate_note_summary
 from app.db import get_session
 from app.integrations import 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,149 @@ 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")
+    await db.delete(scout)
+    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.
@@ -424,10 +571,32 @@ async def scout_gmail_oauth_callback(
     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.
+    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")
+
+        scout.gmail_address = await asyncio.to_thread(_fetch_email)
+    except Exception:
+        logger.exception("failed to fetch gmail address for scout %s", scout_id)
+
     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)

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

@@ -211,3 +211,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)

tests/test_scout_cloud_crud.py

@@ -0,0 +1,149 @@
+"""Tests for cloud scout CRUD routes."""
+
+from __future__ import annotations
+
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.db import get_session
+from app.main import app
+from app.models import CloudScoutConfig
+from tests.conftest import _TestSessionLocal, make_jwt
+
+
+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

tests/test_scout_connectors_gmail.py

@@ -82,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()