feat: add /provider test and /provider models slash commands

amplifier_app_cli/main.py

@@ -50,6 +50,10 @@
 from .dedicated_tty_input import close_dedicated_tty_input, get_dedicated_tty_input
 from .effective_config import get_effective_config_summary
 from .key_manager import KeyManager
+from .provider_diagnostics import DEFAULT_TIMEOUT_S as _PROVIDER_DIAGNOSTIC_TIMEOUT_S
+from .provider_diagnostics import format_model_line
+from .provider_diagnostics import invoke_list_models
+from .provider_diagnostics import test_provider_connectivity
 from .session_runner import SessionConfig, create_initialized_session
 from .session_store import SessionStore
 from .stdout_offload import patch_stdout_offloaded as patch_stdout
@@ -468,7 +472,8 @@ class CommandProcessor:
             "action": "handle_provider",
             "description": (
                 "(experimental) Show/pin the conversation-scope provider: "
-                "/provider (status) | /provider use <name> | /provider auto"
+                "/provider (status) | /provider use <name> | /provider auto | "
+                "/provider test <name> | /provider models <name>"
             ),
         },
     }
@@ -1502,6 +1507,153 @@ def _render_provider_status(self, pin: Any) -> str:
 
         return "\n".join(lines)
 
+    # === /provider test | /provider models: read-only diagnostics ===
+    #
+    # Unlike 'use'/'auto', these two are NEVER gated on the
+    # 'conversation.provider_pin' capability -- they answer "why can't I
+    # pin/why is nothing answering", which is exactly the question that
+    # matters most when the pin capability is absent or refusing. Gating
+    # them on that same capability would remove the diagnostic exactly
+    # when it's needed.
+    #
+    # They also intentionally query THIS SESSION'S MOUNTED providers
+    # (coordinator.get("providers")), never settings.yaml -- that's the
+    # entire reason to run them mid-conversation instead of dropping to a
+    # shell for `amplifier provider test`/`amplifier provider models`. If
+    # the mounted set and settings.yaml disagree, the session's live
+    # reality is what the user needs to see.
+
+    def _unknown_provider_message(self, name: str, mounted: dict[str, Any]) -> str:
+        """Refusal text for a name that isn't currently mounted.
+
+        Names what IS mounted rather than a bare failure -- matches the
+        wording ConversationProviderPin.pin() already uses for the same
+        situation (see the module docstring's WHY: this is the message a
+        failed pin sends the user here to investigate).
+
+        Deliberately NOT tagged (experimental) or dim -- matches the
+        capability-absent and unmounted-provider refusals from 'use': the
+        refusal is the whole message, tagging or dimming would dilute it.
+        """
+        available = ", ".join(sorted(mounted)) if mounted else "(none)"
+        return (
+            f"\u2717 provider {name!r} is not mounted in this session. "
+            f"Mounted providers: {available}"
+        )
+
+    def _resolve_diagnostic_targets(
+        self, name: str
+    ) -> tuple[dict[str, Any], str | None]:
+        """Resolve which mounted provider(s) a diagnostic subcommand
+        should target: all of them (no name given), or exactly one.
+
+        Returns ``(targets, error)``. ``error`` is set (and ``targets``
+        empty) only when a name was given but isn't mounted.
+        """
+        mounted: dict[str, Any] = self.session.coordinator.get("providers") or {}
+        if not name:
+            return mounted, None
+        if name not in mounted:
+            return {}, self._unknown_provider_message(name, mounted)
+        return {name: mounted[name]}, None
+
+    async def _handle_provider_test(self, name: str) -> str:
+        """`/provider test [name]` -- connectivity check against this
+        session's mounted providers. No name tests all of them,
+        concurrently (see class docstring above for why this is never
+        gated on the pin capability, and always live-session-scoped).
+
+        Reuses provider_diagnostics.test_provider_connectivity so "ok"
+        means exactly what `amplifier provider test` means (list_models()
+        succeeds) -- the two surfaces cannot silently disagree.
+        """
+        targets, error = self._resolve_diagnostic_targets(name)
+        if error:
+            return error
+        if not targets:
+            return "Provider test (experimental):\n  (no providers mounted in this session)"
+
+        # Network I/O in an interactive REPL: tell the user before the
+        # (possibly multi-second, possibly multi-provider) wait rather
+        # than freezing silently. Run all targets concurrently -- and each
+        # is individually timeout-bounded (see test_provider_connectivity)
+        # -- so one slow/hung provider can't stall the others or the
+        # session.
+        plural = "" if len(targets) == 1 else "s"
+        console.print(self._dim(f"Testing {len(targets)} provider{plural}..."))
+
+        results = await asyncio.gather(
+            *(test_provider_connectivity(n, p) for n, p in sorted(targets.items()))
+        )
+
+        lines = ["Provider test (experimental):"]
+        for r in results:
+            mark = "\u2713" if r.ok else "\u2717"
+            lines.append(f"  {mark} {r.name:<24} {r.elapsed_s:>5.1f}s  {r.detail}")
+        return "\n".join(lines)
+
+    async def _handle_provider_models(self, name: str) -> str:
+        """`/provider models [name]` -- list the models a mounted, LIVE
+        provider actually offers right now. No name given means "the
+        provider actually answering this conversation" (pinned if
+        pinned, else the priority winner) -- mirroring the CLI's own
+        "uses current provider" default, translated to session terms.
+
+        Never gated on the pin capability -- see class docstring above.
+        """
+        # Shares _resolve_diagnostic_targets with /provider test: a named,
+        # unmounted provider is refused (naming what IS mounted) the same
+        # way in both -- including when nothing at all is mounted, where
+        # the unknown-name refusal still fires rather than being masked by
+        # a generic "nothing mounted" message.
+        targets, error = self._resolve_diagnostic_targets(name)
+        if error:
+            return error
+        if not targets:
+            return "Provider models (experimental):\n  (no providers mounted in this session)"
+
+        if name:
+            target_name = name
+        else:
+            # No name: use whichever provider is actually answering right
+            # now -- pinned if pinned, else the priority winner -- so
+            # "/provider models" with no argument means "the provider
+            # currently in play", mirroring the CLI's own "uses current
+            # provider" default in session terms.
+            pin = self.session.coordinator.get_capability("conversation.provider_pin")
+            pinned_name: str | None = None
+            if pin is not None:
+                try:
+                    pinned_name = pin.current()
+                except Exception:
+                    pinned_name = None
+            if pinned_name and pinned_name in targets:
+                target_name = pinned_name
+            else:
+                target_name = min(
+                    targets,
+                    key=lambda n: self._provider_priority_for_display(targets[n]),
+                )
+        target_provider = targets[target_name]
+
+        console.print(self._dim(f"Fetching models for '{target_name}'..."))
+
+        try:
+            models = await asyncio.wait_for(
+                invoke_list_models(target_provider), _PROVIDER_DIAGNOSTIC_TIMEOUT_S
+            )
+        except TimeoutError:
+            return f"\u2717 {target_name}: timed out after {_PROVIDER_DIAGNOSTIC_TIMEOUT_S:.0f}s"
+        except Exception as e:
+            return f"\u2717 {target_name}: {type(e).__name__}: {e}"
+
+        if not models:
+            return f"Provider models (experimental):\n  {target_name}: (no models reported)"
+
+        lines = [f"Models for '{target_name}' (experimental):"]
+        lines.extend(f"  {format_model_line(model)}" for model in models)
+        return "\n".join(lines)
+
     async def _handle_provider(self, args: str) -> str:
         """Handle /provider: status (no args), 'use <name>' to pin, or
         'auto' to unpin. See REQUIRED BEHAVIORS in the task spec this
@@ -1526,6 +1678,20 @@ async def _handle_provider(self, args: str) -> str:
         if not subcmd:
             return self._render_provider_status(pin)
 
+        # 'test' and 'models' are read-only diagnostics over THIS SESSION'S
+        # mounted providers -- deliberately never gated on `pin` (see the
+        # block comment above _handle_provider_test/_handle_provider_models):
+        # they're the answer to "why isn't pinning working", so gating them
+        # on the same capability that might be missing/refusing would
+        # remove the diagnostic exactly when it's needed.
+        if subcmd == "test":
+            name = parts[1].strip() if len(parts) > 1 else ""
+            return await self._handle_provider_test(name)
+
+        if subcmd == "models":
+            name = parts[1].strip() if len(parts) > 1 else ""
+            return await self._handle_provider_models(name)
+
         if subcmd in ("use", "auto") and pin is None:
             return self._provider_pin_unavailable_message()
 
@@ -1572,7 +1738,8 @@ async def _handle_provider(self, args: str) -> str:
 
         return (
             f"Unknown /provider subcommand: {subcmd!r}. "
-            f"Usage: /provider | /provider use <name> | /provider auto"
+            f"Usage: /provider | /provider use <name> | /provider auto | "
+            f"/provider test <name> | /provider models <name>"
         )
 
     async def _rename_session(self, new_name: str) -> str:

amplifier_app_cli/provider_diagnostics.py

@@ -0,0 +1,114 @@
+"""Shared, live-instance provider diagnostics primitives.
+
+These are the reusable mechanics behind "does this provider answer" and
+"what models does it offer" -- the SAME questions ``amplifier provider
+test``/``amplifier provider models`` answer for disk-config providers (via
+``provider_loader.get_provider_models``, which instantiates a throwaway
+provider from settings.yaml), and that the in-session ``/provider
+test``/``/provider models`` slash commands answer for this session's
+already-mounted, LIVE provider instances.
+
+The two surfaces intentionally source providers differently:
+
+- CLI (``amplifier provider ...``): reads settings.yaml, instantiates a
+  disposable provider object for the single call, and is responsible for
+  closing it afterward.
+- In-session (``/provider ...``): reads ``coordinator.get("providers")`` --
+  the actual mounted objects still answering this conversation -- and must
+  NEVER close them; they are not disposable, they keep running after the
+  diagnostic completes.
+
+What must not diverge between the two is the actual definition of
+"connectivity is OK" (list_models() succeeds) and how a possibly-async
+``list_models()`` is invoked. That mechanic lives here, once, and both
+``provider_loader.get_provider_models`` and the in-session slash commands
+call through it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+from typing import Any
+
+if TYPE_CHECKING:
+    from amplifier_core import ModelInfo  # pyright: ignore[reportAttributeAccessIssue]
+
+
+async def invoke_list_models(provider: Any) -> list["ModelInfo"]:
+    """Call ``list_models()`` on an already-instantiated provider.
+
+    Awaits it if async, calls it directly if sync. Returns ``[]`` if the
+    provider has no ``list_models`` at all. Does NO cleanup and NO error
+    handling of its own -- exceptions from ``list_models()`` propagate
+    unchanged, and callers decide whether/how to close the instance
+    afterward (a throwaway CLI instance should be closed; a session-mounted
+    instance must not be).
+    """
+    list_models_fn = getattr(provider, "list_models", None)
+    if list_models_fn is None:
+        return []
+    if asyncio.iscoroutinefunction(list_models_fn):
+        return await list_models_fn()
+    return list_models_fn()
+
+
+@dataclass
+class ProviderTestResult:
+    """Outcome of a single provider connectivity check."""
+
+    name: str
+    ok: bool
+    elapsed_s: float
+    detail: str
+
+
+DEFAULT_TIMEOUT_S = 15.0
+
+
+async def test_provider_connectivity(
+    name: str, provider: Any, timeout_s: float = DEFAULT_TIMEOUT_S
+) -> ProviderTestResult:
+    """Connectivity check for one provider: call list_models(), time it.
+
+    Never raises -- failures (including a timeout) are captured in the
+    returned result, not thrown, so a caller can run many of these
+    concurrently (e.g. via ``asyncio.gather``) without one slow or failing
+    provider aborting the batch or hanging an interactive session forever.
+
+    This is the same definition of "connectivity is OK" that ``amplifier
+    provider test`` uses (list_models() succeeds), so the CLI and the
+    in-session diagnostic cannot silently disagree on what "ok" means.
+    """
+    start = time.monotonic()
+    try:
+        models = await asyncio.wait_for(invoke_list_models(provider), timeout_s)
+    except TimeoutError:
+        elapsed = time.monotonic() - start
+        detail = f"timed out after {timeout_s:.0f}s"
+        return ProviderTestResult(name=name, ok=False, elapsed_s=elapsed, detail=detail)
+    except Exception as e:
+        elapsed = time.monotonic() - start
+        detail = f"{type(e).__name__}: {e}"
+        return ProviderTestResult(name=name, ok=False, elapsed_s=elapsed, detail=detail)
+
+    elapsed = time.monotonic() - start
+    count = len(models)
+    detail = f"{count} model{'s' if count != 1 else ''} available"
+    return ProviderTestResult(name=name, ok=True, elapsed_s=elapsed, detail=detail)
+
+
+def format_model_line(model: "ModelInfo") -> str:
+    """Render one ``ModelInfo`` as a compact, single-line summary.
+
+    Used by the in-session ``/provider models`` slash command, which
+    returns a plain string (unlike the CLI's Rich ``Table``) -- kept here
+    so the fields shown (id, context window, max output, capabilities)
+    match what ``amplifier provider models`` already surfaces.
+    """
+    context = f"{model.context_window:,}" if model.context_window else "-"
+    max_out = f"{model.max_output_tokens:,}" if model.max_output_tokens else "-"
+    caps = ", ".join(model.capabilities) if model.capabilities else "-"
+    return f"{model.id:<28} context={context:<10} max_out={max_out:<8} caps={caps}"

amplifier_app_cli/provider_loader.py

@@ -12,6 +12,8 @@
 from typing import TYPE_CHECKING
 from typing import Any
 
+from .provider_diagnostics import invoke_list_models
+
 if TYPE_CHECKING:
     from amplifier_core import ModelInfo  # pyright: ignore[reportAttributeAccessIssue]
 
@@ -168,15 +170,22 @@ def get_provider_models(
         logger.debug(f"Provider '{provider_id}' does not have list_models()")
         return []
 
-    # Call list_models (may be sync or async)
+    # Call list_models (may be sync or async) via the shared invocation
+    # primitive in provider_diagnostics -- this is the same "call
+    # list_models, async-aware" mechanic the in-session /provider
+    # test|models slash commands use against already-mounted providers, so
+    # the two surfaces can't quietly diverge on how a provider is asked for
+    # its models. This function still owns instantiation (above) and
+    # cleanup (below) itself, since only it knows this provider instance is
+    # disposable -- a mounted session provider must never be closed.
     # Let exceptions propagate - auth errors, API errors, connection errors
     # should be shown to the user, not silently swallowed
     list_models_fn = provider.list_models
     if asyncio.iscoroutinefunction(list_models_fn):
 
         async def _list_and_cleanup():
             try:
-                return await list_models_fn()
+                return await invoke_list_models(provider)
             finally:
                 if hasattr(provider, "close") and callable(provider.close):
                     try: