fix(*): stop the connect button promising what it cannot do

i18n/messages.json

@@ -751,6 +751,10 @@
       "en": "Connect",
       "zh": "连接"
     },
+    "gui.agent.connecting": {
+      "en": "Connecting...",
+      "zh": "连接中..."
+    },
     "gui.agent.disconnect": {
       "en": "Disconnect",
       "zh": "断开"
@@ -795,6 +799,14 @@
       "en": "openai",
       "zh": "openai"
     },
+    "gui.agent.not_installed": {
+      "en": "Not Installed",
+      "zh": "未安装"
+    },
+    "gui.agent.unauthorized": {
+      "en": "Unauthorized",
+      "zh": "未授权"
+    },
     "gui.agent.none": {
       "en": "Nothing connected yet.",
       "zh": "还没有连接任何智能体。"

raven/acp_client/acp_agent.py

@@ -691,6 +691,7 @@ def __init__(
         mcps: list[str] | None = None,
         allow_mcp_secrets: bool = False,
         session_mcp: bool = True,
+        pool: Any = None,
     ) -> None:
         self.name = name
         self.command = command
@@ -708,6 +709,15 @@ def __init__(
         self._model_baseline: dict[str, str] = {}
         self._model_pushed: dict[str, str] = {}
         self._registry = registry or get_registry()
+        # Which pool this backend's turns are served from. Almost always the
+        # process-wide one, and held unresolved until it is used so that
+        # ``close_pool`` still means what it says. A caller passes its own when
+        # its turns must not disturb the roster's: the pool keys a connection on
+        # its launch arguments, ``cwd`` among them, and a caller whose workspace
+        # is a throwaway directory therefore never matches a held connection --
+        # so on the shared pool it would retire this agent's live connections
+        # before opening its own. ``ping_agent`` is that caller.
+        self._pool = pool
         self.mcps = mcps
         self.allow_mcp_secrets = allow_mcp_secrets
         self.session_mcp = session_mcp
@@ -720,6 +730,17 @@ def __init__(
         self._caps_listener: Any = None
         self._unprompted_announce: Any = None
 
+    @property
+    def pool(self) -> Any:
+        """The connection pool this backend's turns are served from.
+
+        Resolved on each read rather than in the constructor: ``close_pool``
+        replaces the process-wide pool with a fresh one, and a backend that had
+        captured the old object would go on acquiring from a closed pool for the
+        rest of its life.
+        """
+        return self._pool if self._pool is not None else get_pool()
+
     def _resident_sinks(self) -> tuple[Any, Any]:
         """The two callables a resident recorder routes through, bound to this backend.
 
@@ -764,7 +785,7 @@ def repoint_pooled_resident(self) -> None:
         never reaches here, so N stays connected.
         """
         try:
-            connections = get_pool().live(self.name)
+            connections = self.pool.live(self.name)
         except Exception:  # noqa: BLE001 - a swap must never fail on a transport that cannot be listed
             return
         if not connections:
@@ -1169,7 +1190,7 @@ async def run(
                 parent_protocol = str(getattr(provider, "api_protocol", "") or "")
                 if parent_protocol:
                     binding["RAVEN_PARENT_PROTOCOL"] = parent_protocol
-                connection = await get_pool().acquire(
+                connection = await self.pool.acquire(
                     name=self.name,
                     command=self.command,
                     cwd=launch_cwd,
@@ -1491,9 +1512,7 @@ async def request_or_busy(method: str, params: dict[str, Any]) -> Any:
                 # fork, three session opens in a row died on one wedged
                 # process because nothing ever gave up on it. Dropping it here
                 # is what turns the judge's retry into a fresh launch.
-                from raven.acp_client.pool import get_pool
-
-                await get_pool().drop(self.name)
+                await self.pool.drop(self.name)
                 raise AcpTimeoutError(
                     f"acp agent {self.name!r}: {method} timed out after {budget:.0f}s with no turn "
                     f"in flight -- the agent answered nothing the whole wait. Its connection was "

raven/acp_client/capabilities.py

@@ -148,6 +148,14 @@ class CapabilitySnapshot:
     picks from is the one the agent actually serves: a row that named its own
     would drift the first time the agent gained or dropped a mode."""
     auth_methods: tuple[str, ...] = ()
+    needs_auth: bool = False
+    """The agent answered, and then refused to open a session without a credential.
+
+    Measured, not inferred. ``auth_methods`` alone cannot stand in for it: an
+    agent that works advertises those too, so "not ready and has auth methods"
+    is a guess. Deliberately outside ``usable``, which stays "ready and not
+    stale" -- an agent waiting to be signed in is neither usable nor broken, and
+    the surface that tells a reader which is which needs the distinction."""
     elapsed_ms: int = 0
     stale: bool = False
     """Measured against a launch config this agent no longer has.
@@ -186,6 +194,7 @@ def to_wire(self) -> dict[str, Any]:
                 {"id": m.id, "name": m.name, "description": m.description} for m in self.available_modes
             ],
             "authMethods": list(self.auth_methods),
+            "needsAuth": self.needs_auth,
             "elapsedMs": self.elapsed_ms,
         }
 
@@ -255,6 +264,7 @@ def _modes(key: str) -> tuple[AcpMode, ...]:
             model_choices=_choices("modelChoices"),
             available_modes=_modes("availableModes"),
             auth_methods=_strs("authMethods"),
+            needs_auth=bool(row.get("needsAuth")),
             elapsed_ms=int(row.get("elapsedMs") or 0),
         )
 
@@ -625,7 +635,9 @@ async def verify_agent(cfg: Any) -> CapabilitySnapshot:
     fingerprint = snapshot_fingerprint(cfg)
     budget = max(1.0, (getattr(cfg, "ready_timeout_ms", None) or 30000) / 1000)
 
-    def done(status: SnapshotStatus, detail: str, hs: _Handshake | None = None) -> CapabilitySnapshot:
+    def done(
+        status: SnapshotStatus, detail: str, hs: _Handshake | None = None, *, needs_auth: bool = False
+    ) -> CapabilitySnapshot:
         hs = hs or _Handshake()
         full = "; ".join([detail, *hs.warnings]) if hs.warnings else detail
         return CapabilitySnapshot(
@@ -649,6 +661,7 @@ def done(status: SnapshotStatus, detail: str, hs: _Handshake | None = None) -> C
             model_choices=hs.model_choices,
             available_modes=hs.available_modes,
             auth_methods=hs.auth_methods,
+            needs_auth=needs_auth,
             elapsed_ms=int((time.monotonic() - started) * 1000),
         )
 
@@ -685,10 +698,25 @@ def done(status: SnapshotStatus, detail: str, hs: _Handshake | None = None) -> C
             try:
                 session = await client.request("session/new", {"cwd": tmp, "mcpServers": []}, timeout=budget)
             except AcpRemoteError as exc:
-                needs_auth = bool(handshake.auth_methods) or _looks_like_auth(exc.message)
-                status: SnapshotStatus = "attention" if needs_auth else "unknown"
+                # Two readings of one refusal, and they do not take the same
+                # evidence. The advertisement is enough for the coarse status --
+                # it only decides whether a reader should look at this row -- and
+                # it is not enough for `needs_auth`, which is persisted, rendered
+                # as a disabled control and read as "go and sign in". Every agent
+                # that works advertises auth methods too, so taking the
+                # advertisement there would label any unrelated session failure,
+                # a transient one included, as a credential story with no way out.
+                advertised = bool(handshake.auth_methods)
+                refused_over_a_credential = _looks_like_auth(exc.message)
+                needs_auth = refused_over_a_credential
+                status: SnapshotStatus = "attention" if advertised or refused_over_a_credential else "unknown"
                 hint = f" (auth methods: {', '.join(handshake.auth_methods)})" if handshake.auth_methods else ""
-                return done(status, f"connected, but no session could be opened: {exc.message}{hint}", handshake)
+                return done(
+                    status,
+                    f"connected, but no session could be opened: {exc.message}{hint}",
+                    handshake,
+                    needs_auth=needs_auth,
+                )
             except AcpError as exc:
                 tail = client.stderr_tail(400)
                 suffix = f"; stderr: {tail}" if tail else ""

raven/agent/subagent/backends/__init__.py

@@ -316,7 +316,12 @@ def enabled_agents(configs: Sequence[Any]) -> list[Any]:
 
 
 def build_third_party_backend(
-    cfg: Any, *, registry: Any = None, timeout: int | None = None, ready_timeout_ms: int | None = None
+    cfg: Any,
+    *,
+    registry: Any = None,
+    timeout: int | None = None,
+    ready_timeout_ms: int | None = None,
+    pool: Any = None,
 ) -> SubagentBackend:
     """Build a third-party backend from a config object (duck-typed on ``kind``).
 
@@ -333,6 +338,10 @@ def build_third_party_backend(
     would drift the moment a field is added, and the test would then silently
     exercise a different command than a real spawn. ``registry`` is ignored for
     kind ``openai``, which has no session store.
+
+    ``pool`` is the acp transport's equivalent of ``registry``: a caller whose
+    turns must not disturb the roster's connections hands in one of its own.
+    Ignored for every other kind, none of which pools anything.
     """
     kind = getattr(cfg, "kind", None)
     if kind == "cli":
@@ -370,6 +379,7 @@ def build_third_party_backend(
             mcps=cfg.mcps,
             allow_mcp_secrets=cfg.allow_mcp_secrets,
             session_mcp=session_mcp_for(cfg),
+            pool=pool,
         )
     if kind == "openai":
         return OpenAIApiBackend(