Implement device code auth flow (RFC 8628) for Google capabilities

Add device code grant support across the capability stack so Google auth
actually works instead of returning a placeholder URL. When GOOGLE_CLIENT_ID
and GOOGLE_CLIENT_SECRET are configured, the bridge requests a device code
from Google, returns a verification URL + user code, and polls for completion.
Falls back to the legacy authorization_code placeholder when credentials
are not configured.

- Add flow_type, user_code, poll_interval_seconds to CapabilityAuthBeginResult
- Add CapabilityAuthPollResult type and auth_poll to CapabilityProvider protocol
- Add CapabilityManager.auth_poll with flow validation and account storage
- Add SubprocessCapabilityProvider.auth_poll bridge dispatch
- Register capability.auth.poll RPC method
- Add ash-sb capability auth poll CLI command with blocking --timeout mode
- Implement real device code flow in gogcli_bridge with scope mapping,
  token polling, and token refresh before operations
- Update Google SKILL.md for device code UX workflow

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

Author: dcramer

packages/ash-sandbox-cli/src/ash_sandbox_cli/commands/capability.py

@@ -145,6 +145,12 @@ def auth_begin(
     typer.echo(f"Started capability auth flow (flow_id={result.get('flow_id', '?')})")
     typer.echo(f"  Capability: {capability}")
     typer.echo(f"  Auth URL: {result.get('auth_url', '')}")
+    flow_type = result.get("flow_type", "authorization_code")
+    typer.echo(f"  Flow type: {flow_type}")
+    if result.get("user_code"):
+        typer.echo(f"  User code: {result['user_code']}")
+    if result.get("poll_interval_seconds") is not None:
+        typer.echo(f"  Poll interval: {result['poll_interval_seconds']}s")
     typer.echo(f"  Expires: {result.get('expires_at', '')}")
 
 
@@ -184,4 +190,61 @@ def auth_complete(
     )
 
 
+@auth_app.command("poll")
+def auth_poll(
+    flow_id: Annotated[
+        str,
+        typer.Option("--flow-id", help="Auth flow id from auth-begin"),
+    ],
+    timeout: Annotated[
+        int | None,
+        typer.Option(
+            "--timeout", help="Block and poll until complete or timeout (seconds)"
+        ),
+    ] = None,
+    interval: Annotated[
+        int | None,
+        typer.Option("--interval", help="Override poll interval (seconds)"),
+    ] = None,
+) -> None:
+    """Poll a device code auth flow for completion."""
+    import time
+
+    params: dict[str, Any] = {"flow_id": flow_id}
+    result = _call("capability.auth.poll", params)
+
+    if timeout is None:
+        # Single poll
+        if result.get("ok"):
+            typer.echo(
+                f"Capability auth completed "
+                f"(flow_id={flow_id}, account_ref={result.get('account_ref', '')})"
+            )
+        else:
+            retry = result.get("retry_after_seconds", 5)
+            typer.echo(f"Auth flow pending (flow_id={flow_id}, retry_after={retry}s)")
+        return
+
+    # Blocking poll loop
+    deadline = time.monotonic() + max(1, timeout)
+    while True:
+        if result.get("ok"):
+            typer.echo(
+                f"Capability auth completed "
+                f"(flow_id={flow_id}, account_ref={result.get('account_ref', '')})"
+            )
+            return
+
+        retry_after = result.get("retry_after_seconds") or 5
+        sleep_seconds = interval if interval is not None else retry_after
+        sleep_seconds = max(1, sleep_seconds)
+
+        if time.monotonic() + sleep_seconds > deadline:
+            typer.echo(f"Auth flow timed out (flow_id={flow_id})", err=True)
+            raise typer.Exit(1)
+
+        time.sleep(sleep_seconds)
+        result = _call("capability.auth.poll", params)
+
+
 app.add_typer(auth_app, name="auth")

src/ash/capabilities/__init__.py

@@ -22,6 +22,7 @@
 from ash.capabilities.providers import (
     CapabilityAuthBeginResult,
     CapabilityAuthCompleteResult,
+    CapabilityAuthPollResult,
     CapabilityCallContext,
     CapabilityProvider,
     SubprocessCapabilityProvider,
@@ -41,6 +42,7 @@
     "CapabilityCallContext",
     "CapabilityAuthBeginResult",
     "CapabilityAuthCompleteResult",
+    "CapabilityAuthPollResult",
     "CapabilityAccount",
     "CapabilityAuthFlow",
     "CapabilityDefinition",

src/ash/capabilities/manager.py

@@ -14,6 +14,7 @@
 from ash.capabilities.providers import (
     CapabilityAuthBeginResult,
     CapabilityAuthCompleteResult,
+    CapabilityAuthPollResult,
     CapabilityCallContext,
     CapabilityProvider,
 )
@@ -271,6 +272,7 @@ async def auth_begin(
         expires_at = begin_result.expires_at or (
             datetime.now(UTC) + timedelta(seconds=self._auth_flow_ttl_seconds)
         )
+        flow_type = begin_result.flow_type or "authorization_code"
         async with self._lock:
             self._auth_flows[flow_id] = CapabilityAuthFlow(
                 flow_id=flow_id,
@@ -279,13 +281,20 @@ async def auth_begin(
                 account_hint=normalized_account_hint,
                 expires_at=expires_at,
                 flow_state=dict(begin_result.flow_state),
+                flow_type=flow_type,
             )
 
-        return {
+        result: dict[str, Any] = {
             "flow_id": flow_id,
             "auth_url": begin_result.auth_url,
             "expires_at": expires_at.isoformat().replace("+00:00", "Z"),
+            "flow_type": flow_type,
         }
+        if begin_result.user_code is not None:
+            result["user_code"] = begin_result.user_code
+        if begin_result.poll_interval_seconds is not None:
+            result["poll_interval_seconds"] = begin_result.poll_interval_seconds
+        return result
 
     async def auth_complete(
         self,
@@ -386,6 +395,96 @@ async def auth_complete(
 
         return {"ok": True, "account_ref": account_ref}
 
+    async def auth_poll(
+        self,
+        *,
+        flow_id: str,
+        user_id: str,
+        chat_id: str | None = None,
+        chat_type: str | None = None,
+        provider: str | None = None,
+        thread_id: str | None = None,
+        session_key: str | None = None,
+        source_username: str | None = None,
+        source_display_name: str | None = None,
+    ) -> dict[str, Any]:
+        """Poll a pending device code auth flow."""
+        normalized_user_id = _required_text(
+            value=user_id,
+            code="capability_invalid_input",
+            message="user_id is required",
+        )
+        normalized_flow_id = _required_text(
+            value=flow_id,
+            code="capability_invalid_input",
+            message="flow_id is required",
+        )
+
+        async with self._lock:
+            self._prune_expired_flows_locked()
+            flow = self._auth_flows.get(normalized_flow_id)
+            if flow is None:
+                raise CapabilityError(
+                    "capability_auth_flow_invalid",
+                    f"auth flow is invalid or expired: {normalized_flow_id}",
+                )
+            if flow.user_id != normalized_user_id:
+                raise CapabilityError(
+                    "capability_auth_flow_invalid",
+                    "auth flow does not belong to caller",
+                )
+            if flow.flow_type != "device_code":
+                raise CapabilityError(
+                    "capability_invalid_input",
+                    "auth_poll is only supported for device_code flows",
+                )
+            _, provider_impl = self._get_definition_and_provider_locked(
+                flow.capability_id
+            )
+
+        call_context = CapabilityCallContext(
+            user_id=normalized_user_id,
+            chat_id=_optional_text(chat_id),
+            chat_type=_optional_text(chat_type),
+            provider=_optional_text(provider),
+            thread_id=_optional_text(thread_id),
+            session_key=_optional_text(session_key),
+            source_username=_optional_text(source_username),
+            source_display_name=_optional_text(source_display_name),
+        )
+        poll_result = await self._provider_auth_poll(
+            provider_impl,
+            capability_id=flow.capability_id,
+            flow_state=dict(flow.flow_state),
+            context=call_context,
+        )
+
+        if poll_result.status == "complete":
+            account_ref = _required_text(
+                value=poll_result.account_ref,
+                code="capability_invalid_output",
+                message="auth poll completion must return account_ref",
+            )
+            now = datetime.now(UTC)
+            async with self._lock:
+                self._accounts[(flow.user_id, flow.capability_id, account_ref)] = (
+                    CapabilityAccount(
+                        capability_id=flow.capability_id,
+                        user_id=flow.user_id,
+                        account_ref=account_ref,
+                        created_at=now,
+                        credential_material=dict(poll_result.credential_material),
+                        metadata=dict(poll_result.metadata),
+                    )
+                )
+                del self._auth_flows[normalized_flow_id]
+            return {"ok": True, "account_ref": account_ref}
+
+        return {
+            "status": "pending",
+            "retry_after_seconds": poll_result.retry_after_seconds,
+        }
+
     async def invoke(
         self,
         *,
@@ -543,6 +642,28 @@ async def _provider_auth_begin(
             auth_url=auth_url,
             expires_at=result.expires_at,
             flow_state=dict(result.flow_state),
+            flow_type=result.flow_type or "authorization_code",
+            user_code=result.user_code,
+            poll_interval_seconds=result.poll_interval_seconds,
+        )
+
+    async def _provider_auth_poll(
+        self,
+        provider_impl: CapabilityProvider | None,
+        *,
+        capability_id: str,
+        flow_state: dict[str, Any],
+        context: CapabilityCallContext,
+    ) -> CapabilityAuthPollResult:
+        if provider_impl is None:
+            raise CapabilityError(
+                "capability_invalid_input",
+                "auth_poll requires a provider-backed capability",
+            )
+        return await provider_impl.auth_poll(
+            capability_id=capability_id,
+            flow_state=flow_state,
+            context=context,
         )
 
     async def _provider_auth_complete(

src/ash/capabilities/providers/__init__.py

@@ -3,6 +3,7 @@
 from ash.capabilities.providers.base import (
     CapabilityAuthBeginResult,
     CapabilityAuthCompleteResult,
+    CapabilityAuthPollResult,
     CapabilityCallContext,
     CapabilityProvider,
 )
@@ -11,6 +12,7 @@
 __all__ = [
     "CapabilityAuthBeginResult",
     "CapabilityAuthCompleteResult",
+    "CapabilityAuthPollResult",
     "CapabilityCallContext",
     "CapabilityProvider",
     "SubprocessCapabilityProvider",

src/ash/capabilities/providers/base.py

@@ -33,6 +33,9 @@ class CapabilityAuthBeginResult:
     auth_url: str
     expires_at: datetime | None = None
     flow_state: dict[str, Any] = field(default_factory=dict)
+    flow_type: str = "authorization_code"  # or "device_code"
+    user_code: str | None = None
+    poll_interval_seconds: int | None = None
 
 
 @dataclass(slots=True)
@@ -44,6 +47,17 @@ class CapabilityAuthCompleteResult:
     metadata: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(slots=True)
+class CapabilityAuthPollResult:
+    """Provider response for device code auth polling."""
+
+    status: str  # "pending" | "complete"
+    retry_after_seconds: int | None = None
+    account_ref: str | None = None
+    credential_material: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
 class CapabilityProvider(Protocol):
     """Interface for capability provider backends."""
 
@@ -85,3 +99,12 @@ async def auth_complete(
         context: CapabilityCallContext,
     ) -> CapabilityAuthCompleteResult:
         """Complete auth flow and return linked account result."""
+
+    async def auth_poll(
+        self,
+        *,
+        capability_id: str,
+        flow_state: dict[str, Any],
+        context: CapabilityCallContext,
+    ) -> CapabilityAuthPollResult:
+        """Poll a device code auth flow for completion."""

src/ash/capabilities/providers/subprocess.py

@@ -20,6 +20,7 @@
 from ash.capabilities.providers.base import (
     CapabilityAuthBeginResult,
     CapabilityAuthCompleteResult,
+    CapabilityAuthPollResult,
     CapabilityCallContext,
     CapabilityProvider,
 )
@@ -90,10 +91,63 @@ async def auth_begin(
             code="capability_invalid_output",
             message="bridge auth_begin must return auth_url",
         )
+        flow_type = str(result.get("flow_type") or "authorization_code").strip()
+        raw_user_code = result.get("user_code")
+        user_code = str(raw_user_code).strip() if raw_user_code is not None else None
+        raw_poll_interval = result.get("poll_interval_seconds")
+        poll_interval: int | None = None
+        if raw_poll_interval is not None:
+            try:
+                poll_interval = int(raw_poll_interval)
+            except (TypeError, ValueError):
+                pass
         return CapabilityAuthBeginResult(
             auth_url=auth_url,
             expires_at=_parse_optional_datetime(result.get("expires_at")),
             flow_state=_as_object(result.get("flow_state"), default={}),
+            flow_type=flow_type,
+            user_code=user_code,
+            poll_interval_seconds=poll_interval,
+        )
+
+    async def auth_poll(
+        self,
+        *,
+        capability_id: str,
+        flow_state: dict[str, Any],
+        context: CapabilityCallContext,
+    ) -> CapabilityAuthPollResult:
+        result = await self._call_bridge(
+            "auth_poll",
+            {
+                "capability_id": capability_id,
+                "flow_state": dict(flow_state),
+                "context_token": self._issue_context_token(context),
+            },
+        )
+        status = _required_text(
+            value=result.get("status"),
+            code="capability_invalid_output",
+            message="bridge auth_poll must return status",
+        )
+        raw_retry = result.get("retry_after_seconds")
+        retry_after: int | None = None
+        if raw_retry is not None:
+            try:
+                retry_after = int(raw_retry)
+            except (TypeError, ValueError):
+                pass
+        account_ref = result.get("account_ref")
+        if account_ref is not None:
+            account_ref = str(account_ref).strip() or None
+        return CapabilityAuthPollResult(
+            status=status,
+            retry_after_seconds=retry_after,
+            account_ref=account_ref,
+            credential_material=_as_object(
+                result.get("credential_material"), default={}
+            ),
+            metadata=_as_object(result.get("metadata"), default={}),
         )
 
     async def auth_complete(

src/ash/capabilities/types.py

@@ -43,6 +43,7 @@ class CapabilityAuthFlow:
     account_hint: str | None
     expires_at: datetime
     flow_state: dict[str, Any] = field(default_factory=dict)
+    flow_type: str = "authorization_code"
 
 
 @dataclass(slots=True)