Fix Google auth callback loop with pending-flow recovery

Add capability.auth.list + ash-sb auth list, strengthen google skill/auth UX rules, and inject callback-aware recovery context for capability skills to prefer auth complete over auth begin loops.

Co-Authored-By: GPT-5 Codex <codex@openai.com>

Author: dcramer

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

@@ -177,6 +177,51 @@ def auth_begin(
     typer.echo(f"  Expires: {result.get('expires_at', '')}")
 
 
+@auth_app.command("list")
+def auth_list(
+    capability: Annotated[
+        str | None,
+        typer.Option("--capability", "-c", help="Optional namespaced capability id"),
+    ] = None,
+    account_hint: Annotated[
+        str | None,
+        typer.Option("--account", help="Optional account reference hint"),
+    ] = None,
+) -> None:
+    """List pending capability auth flows for the current caller."""
+    params: dict[str, Any] = {}
+    if capability:
+        params["capability"] = capability
+    if account_hint:
+        params["account_hint"] = account_hint
+    result = _call("capability.auth.list", params)
+    flows = result.get("flows") or []
+    if not isinstance(flows, list) or not flows:
+        typer.echo("No pending capability auth flows.")
+        return
+
+    typer.echo("Pending capability auth flows:")
+    for flow in flows:
+        if not isinstance(flow, dict):
+            continue
+        flow_id = str(flow.get("flow_id", "")).strip() or "?"
+        capability_id = str(flow.get("capability", "")).strip()
+        account = str(flow.get("account_hint", "")).strip()
+        if capability_id and account:
+            typer.echo(f"- {flow_id} ({capability_id}, account={account})")
+        elif capability_id:
+            typer.echo(f"- {flow_id} ({capability_id})")
+        else:
+            typer.echo(f"- {flow_id}")
+        typer.echo(f"  Auth URL: {flow.get('auth_url', '')}")
+        typer.echo(f"  Flow type: {flow.get('flow_type', 'authorization_code')}")
+        if flow.get("user_code"):
+            typer.echo(f"  User code: {flow['user_code']}")
+        if flow.get("poll_interval_seconds") is not None:
+            typer.echo(f"  Poll interval: {flow['poll_interval_seconds']}s")
+        typer.echo(f"  Expires: {flow.get('expires_at', '')}")
+
+
 @auth_app.command("complete")
 def auth_complete(
     flow_id: Annotated[

specs/capabilities.md

@@ -379,6 +379,40 @@ Response:
 }
 ```
 
+#### `capability.auth.list`
+
+Lists pending auth flows for the verified caller so follow-up callback/code messages can
+complete an existing flow without restarting auth.
+
+Request params:
+
+```json
+{
+  "capability": "gog.calendar",
+  "account_hint": "work",
+  "context_token": "<signed-token>"
+}
+```
+
+`capability` and `account_hint` are optional filters.
+
+Response:
+
+```json
+{
+  "flows": [
+    {
+      "flow_id": "caf_01...",
+      "capability": "gog.calendar",
+      "account_hint": "work",
+      "auth_url": "https://...",
+      "expires_at": "2026-02-24T20:10:00Z",
+      "flow_type": "authorization_code"
+    }
+  ]
+}
+```
+
 #### `capability.auth.poll`
 
 Polls a pending device code auth flow. See `specs/capability-auth.md` for full contract.
@@ -388,6 +422,7 @@ Polls a pending device code auth flow. See `specs/capability-auth.md` for full c
 - `ash-sb capability list`
 - `ash-sb capability invoke --capability <id> --operation <name> --input-json <json>`
 - `ash-sb capability auth begin --capability <id> [--account <hint>]`
+- `ash-sb capability auth list [--capability <id>] [--account <hint>]`
 - `ash-sb capability auth complete --flow-id <id> (--callback-url <url> | --code <code>)`
 - `ash-sb capability auth poll --flow-id <id> [--timeout <secs>] [--interval <secs>]`
   (See `specs/capability-auth.md`.)

src/ash/capabilities/manager.py

@@ -305,6 +305,44 @@ async def auth_begin(
 
         return _auth_begin_response(flow)
 
+    async def list_auth_flows(
+        self,
+        *,
+        user_id: str,
+        capability_id: str | None = None,
+        account_hint: str | None = None,
+    ) -> list[dict[str, Any]]:
+        """List pending auth flows for the caller."""
+        normalized_user_id = _required_text(
+            value=user_id,
+            code="capability_invalid_input",
+            message="user_id is required",
+        )
+        normalized_capability_id = (
+            _required_capability_id(capability_id)
+            if capability_id is not None
+            else None
+        )
+        normalized_account_hint = _optional_text(account_hint)
+
+        async with self._lock:
+            self._prune_expired_flows_locked()
+            matches = [
+                flow
+                for flow in self._auth_flows.values()
+                if flow.user_id == normalized_user_id
+                and (
+                    normalized_capability_id is None
+                    or flow.capability_id == normalized_capability_id
+                )
+                and (
+                    normalized_account_hint is None
+                    or flow.account_hint == normalized_account_hint
+                )
+            ]
+            matches.sort(key=lambda flow: flow.expires_at, reverse=True)
+            return [_auth_begin_response(flow) for flow in matches]
+
     async def auth_complete(
         self,
         *,
@@ -940,6 +978,8 @@ def _linked_accounts_locked(
 def _auth_begin_response(flow: CapabilityAuthFlow) -> dict[str, Any]:
     result: dict[str, Any] = {
         "flow_id": flow.flow_id,
+        "capability": flow.capability_id,
+        "account_hint": flow.account_hint,
         "auth_url": flow.auth_url,
         "expires_at": flow.expires_at.isoformat().replace("+00:00", "Z"),
         "flow_type": flow.flow_type,

src/ash/integrations/skills/capabilities/google/SKILL.md

@@ -68,17 +68,18 @@ Then:
 
 When presenting auth instructions, always include:
 
+- The exact `flow_id` returned by `auth begin`.
 - The exact `auth_url` returned by `auth begin` (never paraphrase or omit it).
 - The exact `user_code` when flow type is `device_code`.
 - A single clear instruction: complete consent, then paste callback URL or code.
 
 Use one of these response templates exactly:
 
 Authorization code flow:
-`To continue, open this Google auth URL: <auth_url>\nAfter approval, paste the full callback URL (or just the code) here.`
+`To continue, open this Google auth URL: <auth_url>\nFlow ID: <flow_id>\nAfter approval, paste the full callback URL (or just the code) here.`
 
 Device code flow:
-`To continue, open: <auth_url>\nEnter this code: <user_code>\nAfter approval, tell me when done and I will continue.`
+`To continue, open: <auth_url>\nFlow ID: <flow_id>\nEnter this code: <user_code>\nAfter approval, tell me when done and I will continue.`
 
 Use these commands:
 
@@ -90,6 +91,10 @@ ash-sb capability auth complete --flow-id <id> --code '<CODE>'
 
 If user intent is setup-only, stop after successful auth confirmation.
 
+If the user provides a callback URL or auth code in a follow-up message, complete that existing flow immediately with `auth complete`.
+Do not start a new `auth begin` while a valid callback/code is present unless completion fails with invalid/expired flow.
+If `flow_id` is not already known, run `ash-sb capability auth list -c <capability> --account <alias>` and use the most recent pending flow.
+
 ### 2b. Proactive re-auth when scopes change or auth expires
 
 If a user asks for an operation and capability invoke/list returns auth-required or similar auth errors:
@@ -124,6 +129,17 @@ If the user asks a broad question and does not provide scope, use these defaults
 - Day-at-a-glance: `list_events` with `{"calendar":"primary","window":"1d"}` plus unread/recent email query
 - Message deep read: run `get_message` for each item you summarize
 
+### Account and calendar defaults
+
+Interpret account/calendar phrasing with these defaults unless user explicitly says otherwise:
+
+- "work calendar", "my work calendar", or "calendar at work" means account alias `work` and calendar `primary`.
+- "personal calendar", "my calendar", or unspecified calendar means account alias `default` and calendar `primary`.
+- "add/connect/link my <alias> calendar" means start auth for `gog.calendar` using that alias.
+
+Do not ask taxonomy prompts like "do you mean second account vs shared calendar vs subscribed calendar" before starting auth.
+If account alias is implied, proceed with that alias and only ask a single follow-up if a concrete operation later needs a non-primary calendar id.
+
 ## Behavior Playbooks
 
 ### Summarize Emails

src/ash/integrations/skills/capabilities/google/references/auth-and-failures.md

@@ -16,6 +16,8 @@ If user intent is setup-only, stop after auth succeeds.
 
 If user provides a callback URL (`http://localhost/?code=...`), pass it with `--callback-url`.
 If user provides only a code value, pass it with `--code`.
+If `flow_id` is missing, use `ash-sb capability auth list` (filtered by capability/account when possible) and complete the most recent pending flow.
+Do not start a new auth flow while a valid callback URL/code is already provided.
 
 Never ask for OAuth secrets.
 

src/ash/rpc/methods/capability.py

@@ -90,6 +90,20 @@ async def capability_auth_begin(params: dict[str, Any]) -> dict[str, Any]:
         except CapabilityError as e:
             raise ValueError(f"{e.code}: {e}") from e
 
+    async def capability_auth_list(params: dict[str, Any]) -> dict[str, Any]:
+        user_id = _required_text(params, "user_id")
+        capability_id = _optional_text(params, "capability")
+        account_hint = _optional_text(params, "account_hint")
+        try:
+            flows = await manager.list_auth_flows(
+                user_id=user_id,
+                capability_id=capability_id,
+                account_hint=account_hint,
+            )
+        except CapabilityError as e:
+            raise ValueError(f"{e.code}: {e}") from e
+        return {"flows": flows}
+
     async def capability_auth_complete(params: dict[str, Any]) -> dict[str, Any]:
         flow_id = _required_text(params, "flow_id")
         user_id = _required_text(params, "user_id")
@@ -134,6 +148,7 @@ async def capability_auth_poll(params: dict[str, Any]) -> dict[str, Any]:
     server.register("capability.list", capability_list)
     server.register("capability.invoke", capability_invoke)
     server.register("capability.auth.begin", capability_auth_begin)
+    server.register("capability.auth.list", capability_auth_list)
     server.register("capability.auth.complete", capability_auth_complete)
     server.register("capability.auth.poll", capability_auth_poll)
 

src/ash/tools/builtin/skills.py

@@ -19,6 +19,8 @@
 _SECRET_ENV_NAME_PATTERNS = (
     r"(?i)(?:^|_)(?:api[_-]?key|token|secret|password|passwd|auth)(?:$|_)",
 )
+_OAUTH_CALLBACK_URL_PATTERN = re.compile(r"https?://localhost[^\s]*[?&]code=")
+_OAUTH_CODE_ONLY_PATTERN = re.compile(r"^\s*4/[^\s]+\s*$")
 
 # Built-in skills that are handled specially (not loaded from SKILL.md files)
 BUILTIN_SKILLS: dict[str, str] = {}
@@ -75,9 +77,13 @@
 This skill declares host capabilities. When capability auth is required or re-authorization is needed:
 
 1. Start auth immediately with `ash-sb capability auth begin -c <capability>` (do not only say "need auth").
-2. Include the exact `auth_url` returned by the command.
-3. Include the exact `user_code` for device-code flows.
-4. Ask for one clear next step: paste callback URL or code (or confirm completion for device flow polling).
+2. Include the exact `flow_id` returned by the command.
+3. Include the exact `auth_url` returned by the command.
+4. Include the exact `user_code` for device-code flows.
+5. Ask for one clear next step: paste callback URL or code (or confirm completion for device flow polling).
+
+If the user already pasted a callback URL or auth code, complete the existing flow first.
+Do not run another `auth begin` until completion fails as invalid/expired.
 
 Never replace auth instructions with generic hints or slash-command suggestions.
 Preserve auth URLs/codes verbatim in your `complete` output.
@@ -460,6 +466,37 @@ async def _check_capability_availability(
             f"Skill '{skill.name}' has unavailable capabilities: {', '.join(missing)}"
         )
 
+    def _looks_like_oauth_callback_or_code(self, message: str) -> bool:
+        """Detect user-provided OAuth callback URLs or code-only replies."""
+        return bool(
+            _OAUTH_CALLBACK_URL_PATTERN.search(message)
+            or _OAUTH_CODE_ONLY_PATTERN.match(message)
+        )
+
+    def _build_capability_auth_recovery_context(
+        self,
+        *,
+        skill: SkillDefinition,
+        message: str,
+        user_context: str,
+    ) -> str:
+        """Inject deterministic auth-completion guidance for callback follow-ups."""
+        if not skill.capabilities:
+            return user_context
+        if not self._looks_like_oauth_callback_or_code(message):
+            return user_context
+
+        hint = (
+            "OAuth callback/code detected in the latest user message.\n"
+            "Do this before any new auth begin:\n"
+            "1. Run `ash-sb capability auth list` (add `-c <capability>` / `--account <alias>` if known).\n"
+            "2. Complete the newest matching pending flow with `ash-sb capability auth complete --flow-id <flow_id> --callback-url '<user_callback_url>'` or `--code '<user_code>'`.\n"
+            "3. Only run `auth begin` if completion fails with invalid/expired flow."
+        )
+        if not user_context:
+            return hint
+        return f"{user_context}\n\n{hint}"
+
     async def execute(
         self,
         input_data: dict[str, Any],
@@ -475,6 +512,9 @@ async def execute(
         if not message:
             return ToolResult.error("Missing required field: message")
 
+        message = str(message)
+        user_context = str(user_context)
+
         if not self._registry.has(skill_name):
             self._registry.reload_all(self._config.workspace)
             if not self._registry.has(skill_name):
@@ -486,6 +526,11 @@ async def execute(
                 )
 
         skill = self._registry.get(skill_name)
+        user_context = self._build_capability_auth_recovery_context(
+            skill=skill,
+            message=message,
+            user_context=user_context,
+        )
         skill_config = self._config.skills.get(skill_name)
 
         if skill_config and not skill_config.enabled:

tests/test_bundled_gog_skill.py

@@ -57,3 +57,12 @@ def test_google_skill_includes_archive_and_label_mutation_guidance() -> None:
     assert "archive_messages" in text
     assert "update_labels" in text
     assert "always confirm key details" in text
+
+
+def test_google_skill_avoids_auth_loop_and_includes_flow_recovery_guidance() -> None:
+    text = _load_google_skill_text().lower()
+    assert "flow id: <flow_id>" in text
+    assert (
+        "do not start a new `auth begin` while a valid callback/code is present" in text
+    )
+    assert "ash-sb capability auth list" in text

tests/test_capabilities.py

@@ -562,6 +562,44 @@ async def test_auth_begin_reuses_pending_flow_for_same_scope() -> None:
     assert len(provider.begin_calls) == 1
 
 
+@pytest.mark.asyncio
+async def test_list_auth_flows_filters_and_scopes_by_user() -> None:
+    manager = CapabilityManager(auth_flow_ttl_seconds=300)
+    provider = _RecordingProvider(namespace="gog")
+    await manager.register_provider(provider)
+
+    user_one_work = await manager.auth_begin(
+        capability_id="gog.email",
+        user_id="user-1",
+        chat_type="private",
+        account_hint="work",
+    )
+    await manager.auth_begin(
+        capability_id="gog.email",
+        user_id="user-1",
+        chat_type="private",
+        account_hint="personal",
+    )
+    await manager.auth_begin(
+        capability_id="gog.email",
+        user_id="user-2",
+        chat_type="private",
+        account_hint="work",
+    )
+
+    user_one_flows = await manager.list_auth_flows(user_id="user-1")
+    assert len(user_one_flows) == 2
+    assert {flow["account_hint"] for flow in user_one_flows} == {"work", "personal"}
+    assert all(flow["capability"] == "gog.email" for flow in user_one_flows)
+
+    filtered = await manager.list_auth_flows(
+        user_id="user-1",
+        capability_id="gog.email",
+        account_hint="work",
+    )
+    assert [flow["flow_id"] for flow in filtered] == [user_one_work["flow_id"]]
+
+
 @pytest.mark.asyncio
 async def test_provider_registration_enforces_namespace_prefix() -> None:
     manager = CapabilityManager()