Reuse pending capability auth flows and complete from pasted callback

Avoid regenerating OAuth state when a matching pending flow exists, persist auth begin response fields on flow records, and update Google skill instructions to immediately complete auth when callback URL/code is already present.

Also document pending-flow reuse in capabilities spec and add coverage for auth_begin reuse semantics.

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

Author: dcramer

specs/capabilities.md

@@ -324,6 +324,10 @@ Response:
 Starts auth for a capability/account and returns an auth flow handle.
 For device code flow extensions (`flow_type`, `user_code`, `auth_poll`), see `specs/capability-auth.md`.
 
+If an unexpired pending auth flow already exists for the same caller scope
+(`effective_user_id`, `capability`, `account_hint`), the host returns that
+existing flow instead of creating a new one.
+
 Request params:
 
 ```json

src/ash/capabilities/manager.py

@@ -255,6 +255,13 @@ async def auth_begin(
                 normalized_capability_id
             )
             _assert_chat_type_allowed(definition, normalized_chat_type)
+            existing_flow = self._find_pending_auth_flow_locked(
+                user_id=normalized_user_id,
+                capability_id=normalized_capability_id,
+                account_hint=normalized_account_hint,
+            )
+            if existing_flow is not None:
+                return _auth_begin_response(existing_flow)
 
         call_context = CapabilityCallContext(
             user_id=normalized_user_id,
@@ -278,29 +285,23 @@ async def auth_begin(
             datetime.now(UTC) + timedelta(seconds=self._auth_flow_ttl_seconds)
         )
         flow_type = begin_result.flow_type or "authorization_code"
+        flow = CapabilityAuthFlow(
+            flow_id=flow_id,
+            capability_id=normalized_capability_id,
+            user_id=normalized_user_id,
+            account_hint=normalized_account_hint,
+            expires_at=expires_at,
+            auth_url=begin_result.auth_url,
+            flow_state=dict(begin_result.flow_state),
+            flow_type=flow_type,
+            user_code=begin_result.user_code,
+            poll_interval_seconds=begin_result.poll_interval_seconds,
+            expected_callback_state=begin_result.expected_callback_state,
+        )
         async with self._lock:
-            self._auth_flows[flow_id] = CapabilityAuthFlow(
-                flow_id=flow_id,
-                capability_id=normalized_capability_id,
-                user_id=normalized_user_id,
-                account_hint=normalized_account_hint,
-                expires_at=expires_at,
-                flow_state=dict(begin_result.flow_state),
-                flow_type=flow_type,
-                expected_callback_state=begin_result.expected_callback_state,
-            )
+            self._auth_flows[flow_id] = flow
 
-        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
+        return _auth_begin_response(flow)
 
     async def auth_complete(
         self,
@@ -639,6 +640,24 @@ def _get_definition_and_provider_locked(
         provider_impl = self._providers.get(provider_name) if provider_name else None
         return definition, provider_impl
 
+    def _find_pending_auth_flow_locked(
+        self,
+        *,
+        user_id: str,
+        capability_id: str,
+        account_hint: str | None,
+    ) -> CapabilityAuthFlow | None:
+        matches = [
+            flow
+            for flow in self._auth_flows.values()
+            if flow.user_id == user_id
+            and flow.capability_id == capability_id
+            and flow.account_hint == account_hint
+        ]
+        if not matches:
+            return None
+        return max(matches, key=lambda flow: flow.expires_at)
+
     async def _provider_auth_begin(
         self,
         provider_impl: CapabilityProvider | None,
@@ -864,6 +883,20 @@ def _first_account_ref_locked(
     return sorted(refs)[0]
 
 
+def _auth_begin_response(flow: CapabilityAuthFlow) -> dict[str, Any]:
+    result: dict[str, Any] = {
+        "flow_id": flow.flow_id,
+        "auth_url": flow.auth_url,
+        "expires_at": flow.expires_at.isoformat().replace("+00:00", "Z"),
+        "flow_type": flow.flow_type,
+    }
+    if flow.user_code is not None:
+        result["user_code"] = flow.user_code
+    if flow.poll_interval_seconds is not None:
+        result["poll_interval_seconds"] = flow.poll_interval_seconds
+    return result
+
+
 def _find_sensitive_key_path(value: Any, path: str = "output") -> str | None:
     if isinstance(value, dict):
         for raw_key, nested in value.items():

src/ash/capabilities/types.py

@@ -42,8 +42,11 @@ class CapabilityAuthFlow:
     user_id: str
     account_hint: str | None
     expires_at: datetime
+    auth_url: str
     flow_state: dict[str, Any] = field(default_factory=dict)
     flow_type: str = "authorization_code"
+    user_code: str | None = None
+    poll_interval_seconds: int | None = None
     expected_callback_state: str | None = None
 
 

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

@@ -61,6 +61,14 @@ Total: 2 capability(ies)
 
 Run this step for each capability where `Authenticated: no`. If the user's request is setup-only (e.g. "set up my email"), stop after authentication is complete — do not invoke any operations.
 
+Before prompting the user again, check whether the current task already contains a pasted Google callback URL (`http://localhost/?...code=...`) or a raw auth code. If yes:
+
+1. Run `ash-sb capability auth begin -c <capability>` first.
+2. Immediately run `ash-sb capability auth complete --flow-id <id> --callback-url <URL>` (or `--code <CODE>`).
+3. Re-run `ash-sb capability list` and continue to operations if authenticated.
+
+Do not ask the user for another URL/code when one is already present in the task.
+
 **2a. Begin auth flow**
 
 Use `--account work` or `--account personal` if the user specifies an account preference:

tests/test_capabilities.py

@@ -400,6 +400,29 @@ async def test_auth_complete_rejects_callback_state_mismatch() -> None:
     assert exc_info.value.code == "capability_auth_state_mismatch"
 
 
+@pytest.mark.asyncio
+async def test_auth_begin_reuses_pending_flow_for_same_scope() -> None:
+    manager = CapabilityManager(auth_flow_ttl_seconds=300)
+    provider = _RecordingProvider(namespace="gog")
+    await manager.register_provider(provider)
+
+    first = await manager.auth_begin(
+        capability_id="gog.email",
+        user_id="user-1",
+        chat_type="private",
+        account_hint="work",
+    )
+    second = await manager.auth_begin(
+        capability_id="gog.email",
+        user_id="user-1",
+        chat_type="private",
+        account_hint="work",
+    )
+
+    assert first["flow_id"] == second["flow_id"]
+    assert len(provider.begin_calls) == 1
+
+
 @pytest.mark.asyncio
 async def test_provider_registration_enforces_namespace_prefix() -> None:
     manager = CapabilityManager()