Reference

class AsyncBroker:
    """Façade over the LLM pool: route completions, inspect state, edit the catalog.

    Every constructor argument is documented in ``docs/`` — "Model pool and calls"
    and "Direct model calls".
    """

    def __init__(  # noqa: PLR0913
        self,
        registry: RegistryProtocol | str | Path | None = None,
        *,
        secrets: SecretsProtocol | None = None,
        store: StoreProtocol | None = None,
        optimize: bool | Optimizer = True,
        sync: str | None | _SyncDefault = _SYNC_DEFAULT,
        sync_interval: float | None = _DEFAULT_SYNC_INTERVAL,
        home: str | Path | None = None,
        direct: Sequence[str | LLMConfig] = (),
    ) -> None:
        _check_sync_interval(sync_interval)
        source = _resolve_sync(sync, registry)
        self._home = home_dir(home)
        source_secrets: SecretsProtocol | None = None
        source_store: StoreProtocol | None = None
        source_label: str | None = None
        if registry is None:
            registry, source_secrets, source_store = zero_config_ports(self._home)
        elif isinstance(registry, (str, Path)):
            source_label = str(registry)
            registry, source_secrets, source_store = resolve_source(registry)

        secrets = (
            as_secrets(secrets) if secrets is not None else (source_secrets or default_secrets())
        )
        store = store if store is not None else (source_store or default_store())

        if isinstance(optimize, Optimizer):
            self._optimizer: Optimizer | None = optimize
        elif optimize:
            self._optimizer = Optimizer()
        else:
            self._optimizer = None

        self._registry = registry
        self._secrets = secrets
        self._store = store
        self._presets = PresetSource(self._home)
        self._shared_ring = KeyRing(secrets)
        self._rings: dict[str, KeyRing] = {}
        self._known: frozenset[str] | None = None

        self._declared = tuple(direct)
        self._autofetch = sync_interval is not None
        self._last_declared: DeclaredModels | None = None
        pool = LLMPool(optimizer=self._optimizer)
        self._pool = pool
        self._catalog = Catalog(
            registry,
            secrets,
            pool,
            self._shared_ring,
            store,
            overlay=self._resolve_declared if self._declared else None,
            autofill=self._autofetch,
            relearn=self._relearn,
            followed_key_info=(
                partial(_followed_key_info, self._presets, source) if source is not None else None
            ),
        )

        self._learner: Learner | None = None
        if self._optimizer is not None:
            self._learner = Learner(self._optimizer, store, pool)

        self._router = Router(
            pool,
            store,
            optimizer=self._optimizer,
            learner=self._learner,
        )
        self._pool_view = PoolView(
            pool,
            self._metrics_map,
            health=lambda: self._catalog.health,
            direct_missing_keys=lambda: self._catalog.direct_missing_keys,
            direct_unresolved=lambda: self._catalog.direct_unresolved,
            payable=lambda: self._catalog.payable,
        )

        self._refresher = ModelListRefresher(
            registry,
            self._catalog,
            store,
            self._presets,
            source=source,
            interval=sync_interval,
            home=self._home,
            declared=self._declared,
            target_label=source_label,
            rebuild=self._rebuild_after_sync,
        )

        self._provisioned = False
        self._provision_lock = asyncio.Lock()
        self._last_underprov_alert: float = float("-inf")
        self._underprov_alert_interval: float = 60.0
        self._next_exhaustion_rebuild: float = float("-inf")
        self._streams: set[StreamHandle] = set()
        self._closing: asyncio.Task[None] | None = None
        self.llms = self._caller(self._shared_ring)

    def _caller(self, ring: KeyRing) -> AsyncLLMs:
        return AsyncLLMs(
            ring,
            router=self._router,
            catalog=self._catalog,
            pool_view=self._pool_view,
            store=self._store,
            learner=self._learner,
            ensure_pool=self.ensure_pool,
            tick=self._tick,
            on_exhausted=self._on_exhausted,
            own_stream=self._own_stream,
            release_stream=self._streams.discard,
        )

    def _own_stream(self, stream: StreamHandle) -> None:
        if self._closing is not None:
            raise RuntimeError("the broker is closed")
        self._streams.add(stream)

    def for_scope(self, scope: str) -> AsyncLLMs:
        """A caller that pays with ``scope``\'s own keys, falling back to the shared
        ones, and writes ``scope`` on every row it journals. Costs no I/O."""
        if not scope:
            raise ValueError("scope must not be empty string; use broker.llms for unscoped")
        ring = self._rings.get(scope)
        if ring is None:
            if len(self._rings) >= _MAX_CALLERS:
                self._rings.pop(next(iter(self._rings)))
            ring = KeyRing(
                self._secrets,
                scope=scope,
                shared=self._shared_ring,
                known=self._known,
            )
            self._rings[scope] = ring
        return self._caller(ring)

    async def rebuild(self) -> None:
        """Rebuild the pool: every caller's keys, the registry, pool membership, the
        disabled map and quality, wholesale. Fires on exactly four triggers — start,
        the refresh clock, an explicit ``sync()``, and pool exhaustion."""
        known = await known_refs(self._secrets)
        self._known = known
        # Over a copy: a request may ask for a caller while this is awaiting.
        for ring in (self._shared_ring, *list(self._rings.values())):
            await ring.refresh(known)
        await self._catalog.rebuild(known)

    async def _rebuild_after_sync(self) -> None:
        """Rebuild a running pool once a provisioning in progress is done, since it may
        have read the registry before this write; with no pool there is nothing to rebuild."""
        async with self._provision_lock:
            live = self._provisioned
        if live:
            await self.rebuild()

    async def _relearn(self) -> None:
        if self._learner is not None:
            await self._learner.relearn()

    async def _rebuild_safely(self, reason: str) -> None:
        """A rebuild reached from a caller's own call must never fail it: an
        unreadable port leaves the pool exactly as it is, and says so."""
        try:
            await self.rebuild()
        except Exception:  # noqa: BLE001 - a background re-read may not break a request
            logger.exception("pool rebuild on %s failed, continuing on the current pool", reason)

    async def _metrics_map(self) -> dict[str, LLMMetrics]:
        """Per-LLM metrics from whatever is available: the learner's cache, a
        queryable store's tail, or nothing."""
        if self._learner is not None:
            return self._learner.metrics
        if isinstance(self._store, QueryableStoreProtocol):
            return metrics_from_calls(await self._store.calls(limit=TAIL_READ_LIMIT))
        return {}

    async def _resolve_declared(self) -> DeclaredModels:
        """Re-resolve ``direct=``: a catalog that cannot be read keeps the whole
        resolution in use, an alias it dropped keeps its own entry, and a handle that
        never resolved is carried unresolved — see ``rules/direct-by-name.md``."""
        previous = self._last_declared
        try:
            resolved, moved = await resolve_declared(
                self._declared,
                self._presets,
                previous=previous,
                fetch=self._autofetch,
            )
        except (ValueError, OSError) as exc:
            if previous is None:
                raise
            logger.warning(
                "direct= could not be re-resolved (%s) — declared models stay on the"
                " resolution already in use",
                exc,
            )
            return previous
        for fact, line in zip(moved, alias_lines(moved), strict=True):
            level = logging.WARNING if fact.change is AliasChange.DROPPED else logging.INFO
            logger.log(level, "direct=: %s", line)
        self._last_declared = resolved
        return resolved

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    async def ensure_pool(self) -> None:
        """Provision lazily and schedule a due model list refresh.
        Raises if the broker is closed or its registry is empty and unfilled."""
        if self._closing is not None:
            raise RuntimeError("the broker is closed")
        if not self._provisioned:
            async with self._provision_lock:
                if not self._provisioned:
                    await self._refresher.before_provision()
                    await self.rebuild()
                    self._catalog.check_not_empty()
                    self._provisioned = True
        # Outside the lock: a refresh calls sync(), and inside it the catalog is
        # mid-provision.
        self._refresher.schedule()

    def _tick(self) -> None:
        """The refresh clock for a call that provisions nothing — ``direct()``, whose
        declared alias has no other clock. Refuses a closed broker, as provisioning does."""
        if self._closing is not None:
            raise RuntimeError("the broker is closed")
        self._refresher.tick()

    @property
    def last_sync_report(self) -> SyncReport | None:
        """What the last sync — explicit or refreshed — did, or ``None`` if none has
        run. A host forwards it to its own admin channel."""
        return self._refresher.last_report

    async def sync(self, source: str | None = None) -> SyncReport | None:
        """Merge the curated preset named by ``source`` into the registry and return
        what it did; with no argument, whatever this installation follows — the paid
        catalog alone has no report. See ``rules/model-list.md``."""
        return await self._refresher.sync(source)

    async def aclose(self) -> None:
        """Settle owned streams before closing the installation's shared resources."""
        if self._closing is None:
            self._closing = asyncio.create_task(self._close())
        await asyncio.shield(self._closing)

    async def _close(self) -> None:
        results = await asyncio.gather(
            *(stream.aclose() for stream in tuple(self._streams)),
            return_exceptions=True,
        )
        # Before the ports: a refresh in flight would otherwise write through a
        # registry whose driver is closing.
        await self._refresher.aclose()
        await self._router.aclose()
        for port in (self._registry, self._secrets, self._store):
            if isinstance(port, AsyncResourceProtocol):
                await port.aclose()
        for result in results:
            if isinstance(result, BaseException):
                raise result

    async def __aenter__(self) -> "AsyncBroker":
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.aclose()

    # ------------------------------------------------------------------
    # Routing — delegated to the unscoped caller
    # ------------------------------------------------------------------

    async def ask(  # noqa: PLR0913 - the call knobs, one keyword each
        self,
        prompt: str,
        *,
        operation: str | None = None,
        trace_id: str | None = None,
        wait: float | None = None,
        fastest_of: int | None = None,
        parallel_recovery: bool = True,
        response_format: dict | None = None,
    ) -> AsyncResult:
        return await self.llms.ask(
            prompt,
            operation=operation,
            trace_id=trace_id,
            wait=wait,
            fastest_of=fastest_of,
            parallel_recovery=parallel_recovery,
            response_format=response_format,
        )

    async def chat(  # noqa: PLR0913 - what to send, and the call knobs
        self,
        messages: list[dict],
        *,
        tools: list[dict] | None = None,
        operation: str | None = None,
        trace_id: str | None = None,
        wait: float | None = None,
        fastest_of: int | None = None,
        parallel_recovery: bool = True,
        response_format: dict | None = None,
    ) -> AsyncResult:
        return await self.llms.chat(
            messages,
            tools=tools,
            operation=operation,
            trace_id=trace_id,
            wait=wait,
            fastest_of=fastest_of,
            parallel_recovery=parallel_recovery,
            response_format=response_format,
        )

    def stream(  # noqa: PLR0913 - the call knobs, one keyword each
        self,
        prompt: str,
        *,
        operation: str | None = None,
        trace_id: str | None = None,
        wait: float | None = None,
        fastest_of: int | None = None,
        parallel_recovery: bool = True,
        response_format: dict | None = None,
        stream_selection_window: float = 1.0,
    ) -> StreamHandle:
        """Return an owned stream that can supply another complete pool answer.
        The broker closes it; use ``async with`` for earlier release."""
        return self.llms.stream(
            prompt,
            operation=operation,
            trace_id=trace_id,
            wait=wait,
            fastest_of=fastest_of,
            parallel_recovery=parallel_recovery,
            response_format=response_format,
            stream_selection_window=stream_selection_window,
        )

    async def direct(
        self,
        alias: str | None = None,
        *,
        name: str | None = None,
    ) -> AsyncDirectClient:
        """A client for exactly one model of your own — no pool, no failover.

        Takes exactly one of ``alias`` or ``name=``; raises ``PoolModelError``,
        ``UnknownModelError`` or ``MissingKeyError``. See ``docs/`` "Direct model calls".
        """
        return await self.llms.direct(alias, name=name)

    async def record_quality(
        self,
        score: float,
        *,
        call_id: str | None = None,
        trace_id: str | None = None,
    ) -> None:
        """Rate a past call — the delayed counterpart of ``result.record_quality``.

        Takes exactly one key and rates the newest answered call it names within the
        rating window; raises ``UnknownCallError`` when nothing there answered.
        """
        await self.llms.record_quality(score, call_id=call_id, trace_id=trace_id)

    # ------------------------------------------------------------------
    # Inspection
    # ------------------------------------------------------------------

    async def get(self, name: str) -> AsyncLLM:
        return await self.llms.get(name)

    async def count(self) -> int:
        return await self.llms.count()

    async def snapshot(self) -> PoolSnapshot:
        await self.ensure_pool()
        return await self._pool_view.snapshot()

    # ------------------------------------------------------------------
    # Manual disable — the one verdict that actually excludes
    # ------------------------------------------------------------------

    async def disable_llm(self, name: str) -> None:
        """Set the manual latch: withdraws the slot, survives preset rolls, covers
        every operation including future ones. Only ``enable_llm`` clears it."""
        await self.ensure_pool()
        self._pool.set_disabled(name)
        if isinstance(self._store, DisabledMapProtocol):
            await self._store.set_disabled(name, True)

    async def enable_llm(self, name: str) -> None:
        """Clear the manual latch — a re-enabled model rehabilitates through new
        ratings, no quality reset exists."""
        await self.ensure_pool()
        await self._pool.clear_disabled(name)
        if isinstance(self._store, DisabledMapProtocol):
            await self._store.set_disabled(name, False)

    # ------------------------------------------------------------------
    # Call journal
    # ------------------------------------------------------------------

    async def calls(
        self,
        *,
        limit: int,
        since: datetime | None = None,
        operation: str | None = None,
        trace_id: str | None = None,
        call_id: str | None = None,
    ) -> list[Call]:
        """Newest-first journal tail for the whole installation — one caller's own rows
        are ``for_scope(...).calls(...)``. One row per call attempt, each carrying the
        newest score it was rated with. Never provisions the pool."""
        return await self.llms.calls(
            limit=limit,
            since=since,
            operation=operation,
            trace_id=trace_id,
            call_id=call_id,
        )

    async def stats(
        self,
        *,
        since: datetime | None = None,
        limit: int = _DEFAULT_STATS_LIMIT,
        operation: str | None = None,
    ) -> Mapping[str, LLMStats]:
        """Per-model counts of call records over a window, keyed by model name.
        ``limit`` caps rows read, not the window: totals summing to it mean the
        window may be truncated. Never provisions the pool."""
        return await self.llms.stats(since=since, limit=limit, operation=operation)

    async def _on_exhausted(self, exc: NoLLMAvailableError, ring: KeyRing) -> bool:
        """The reactive trigger: a pool that could not answer is re-read, debounced and
        skipped for a caller already holding every key. Answers whether the re-read
        actually ran, which is what makes the caller's second pass worth taking."""
        self._maybe_alert_underprov(exc)
        now = time.monotonic()
        if now < self._next_exhaustion_rebuild:
            return False
        if await self._fully_keyed(ring):
            return False
        self._next_exhaustion_rebuild = now + _EXHAUSTION_DEBOUNCE_SEC
        await self._rebuild_safely("pool exhaustion")
        return True

    async def _fully_keyed(self, ring: KeyRing) -> bool:
        """Whether this caller can already pay for every ref the pool names. An empty
        pool is not a full set: there the registry itself is what needs re-reading."""
        refs = {cfg.api_key_ref for cfg in self._pool.configs.values() if cfg.api_key_ref}
        return bool(refs) and refs <= await ring.payable(refs)

    def _maybe_alert_underprov(self, exc: NoLLMAvailableError) -> None:
        """Fire when zero *keyed* configs are routable — the genuine alarm.

        Keyless configs are excluded because they are never cooled, so one of them
        would mask "every keyed model is COOLING"; other reasons log their own line.
        """
        if exc.reason != "timeout":
            return
        if self._optimizer is None:
            return
        if not self._pool.configs:
            return
        now = time.monotonic()
        if now - self._last_underprov_alert < self._underprov_alert_interval:
            return
        payable = self._catalog.payable
        keyed_names = [
            name for name, cfg in self._pool.configs.items() if cfg.api_key_ref in payable
        ]
        all_offline = all(
            self._pool.state(name).phase is not LifecyclePhase.AVAILABLE for name in keyed_names
        )
        if all_offline:
            self._last_underprov_alert = now
            logger.warning(
                "pool under-provisioned: all LLMs are COOLING — add more LLMs to the registry",
            )

class AsyncDirectClient:
    """Async direct client for one named model — ``stream()``, ``ask()`` and ``chat()``.
    Pass an ``httpx.AsyncClient`` to share a connection pool, or let it open and close
    its own; ``tool_params`` ride every ``chat`` that carries tools, under ``params``."""

    def __init__(  # noqa: PLR0913 - where the model is, and how to reach it
        self,
        *,
        base_url: str,
        model: str,
        api_key: str,
        timeout: float = _DEFAULT_TIMEOUT,
        client: httpx.AsyncClient | None = None,
        tool_params: Mapping[str, object] | None = None,
    ) -> None:
        self._model = model
        self._timeout = timeout
        self._http = client
        self._owns_http = client is None
        self._request = partial(
            build_chat_request,
            base_url,
            model,
            api_key,
            tool_params=tool_params,
        )

    def _ensure_http(self) -> httpx.AsyncClient:
        if self._http is None:
            self._http = make_client(self._timeout)
        return self._http

    async def ask(
        self,
        prompt: str | None = None,
        *,
        messages: list[dict] | None = None,
        timeout: float | None = None,
        params: Mapping[str, object] | None = None,
    ) -> DirectResult:
        return await self.chat(_messages(prompt, messages), timeout=timeout, params=params)

    async def chat(
        self,
        messages: list[dict],
        *,
        tools: list[dict] | None = None,
        timeout: float | None = None,
        params: Mapping[str, object] | None = None,
    ) -> DirectResult:
        """One complete reply, which may be tool calls and no text."""
        url, headers, body = self._request(messages, tools, params=params)
        try:
            resp = await self._ensure_http().post(
                url,
                headers=headers,
                json=body,
                timeout=timeout or self._timeout,
            )
        except httpx.TimeoutException as exc:
            raise LLMTimeoutError("direct call timed out") from exc
        return _result(resp, self._model)

    async def stream(
        self,
        prompt: str | None = None,
        *,
        messages: list[dict] | None = None,
        timeout: float | None = None,
        params: Mapping[str, object] | None = None,
    ) -> AsyncIterator[str]:
        url, headers, body = self._request(
            _messages(prompt, messages),
            stream=True,
            params=params,
        )
        try:
            async with self._ensure_http().stream(
                "POST",
                url,
                headers=headers,
                json=body,
                timeout=timeout or self._timeout,
            ) as resp:
                if resp.status_code >= ERROR_FLOOR:
                    detail = (await resp.aread()).decode(errors="replace")[:DETAIL_SNIPPET]
                    raise provider_error(resp.status_code, detail, resp.headers)
                produced = False
                async for chunk in aiter_chat_chunks(resp, self._model):
                    delta, _ = parse_stream_chunk(chunk, self._model)
                    if delta:
                        produced = True
                        yield delta
                if not produced:
                    raise empty_answer_error(self._model, NO_DELTA)
        except httpx.TimeoutException as exc:
            raise LLMTimeoutError("direct stream timed out") from exc

    async def aclose(self) -> None:
        if self._owns_http and self._http is not None:
            await self._http.aclose()
            self._http = None

    async def __aenter__(self) -> "AsyncDirectClient":
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.aclose()

class AsyncLLM:
    """Handle returned by ``AsyncBroker.get(name)`` — live view into broker internals."""

    def __init__(
        self,
        name: str,
        config: LLMConfig,
        pool: LLMPool,
        metrics_source: MetricsSource,
    ) -> None:
        self._name = name
        self._config = config
        self._pool = pool
        self._metrics_source = metrics_source

    @property
    def config(self) -> LLMConfig:
        return self._config

    @property
    def disabled(self) -> bool:
        return self._pool.is_disabled(self._name)

    async def state(self) -> LLMState:
        return self._pool.state(self._name)

    async def metrics(self) -> LLMMetrics:
        all_metrics = await self._metrics_source()
        return all_metrics.get(self._name, LLMMetrics(0, None, None))

class AsyncLLMs:
    """Route and rate calls over the broker's one shared pool, paying with this
    caller's keys and writing its scope on every row it journals."""

    def __init__(  # noqa: PLR0913 - a caller is its ring over the installation's parts
        self,
        ring: KeyRing,
        *,
        router: Router,
        catalog: Catalog,
        pool_view: PoolView,
        store: StoreProtocol,
        learner: Learner | None,
        ensure_pool: Callable[[], Awaitable[None]],
        tick: Callable[[], None],
        on_exhausted: Callable[[NoLLMAvailableError, KeyRing], Awaitable[bool]],
        own_stream: Callable[[StreamHandle], None],
        release_stream: Callable[[StreamHandle], None],
    ) -> None:
        self._ring = ring
        self._router = router
        self._catalog = catalog
        self._pool_view = pool_view
        self._store = store
        self._learner = learner
        self._ensure_pool = ensure_pool
        self._tick = tick
        self._on_exhausted = on_exhausted
        self._own_stream = own_stream
        self._release_stream = release_stream

    @property
    def scope(self) -> str | None:
        """Whose calls these are — the attribution on this caller's journal rows."""
        return self._ring.scope

    # ------------------------------------------------------------------
    # Routing
    # ------------------------------------------------------------------

    async def ask(  # noqa: PLR0913 - the call knobs, one keyword each
        self,
        prompt: str,
        *,
        operation: str | None = None,
        trace_id: str | None = None,
        wait: float | None = None,
        fastest_of: int | None = None,
        parallel_recovery: bool = True,
        response_format: dict | None = None,
    ) -> AsyncResult:
        return await self.chat(
            [{"role": "user", "content": prompt}],
            operation=operation,
            trace_id=trace_id,
            wait=wait,
            fastest_of=fastest_of,
            parallel_recovery=parallel_recovery,
            response_format=response_format,
        )

    async def chat(  # noqa: PLR0913 - what to send, and the call knobs
        self,
        messages: list[dict],
        *,
        tools: list[dict] | None = None,
        operation: str | None = None,
        trace_id: str | None = None,
        wait: float | None = None,
        fastest_of: int | None = None,
        parallel_recovery: bool = True,
        response_format: dict | None = None,
    ) -> AsyncResult:
        await self._ensure_pool()
        try:
            return await self._router.chat(
                self._ring,
                messages,
                tools=tools,
                operation=operation,
                trace_id=trace_id,
                wait=wait,
                fastest_of=fastest_of,
                parallel_recovery=parallel_recovery,
                response_format=response_format,
            )
        except NoLLMAvailableError as exc:
            if not await self._on_exhausted(exc, self._ring):
                raise
        # The ports were re-read inside this request, so it may as well have the
        # answer. ``wait=0``: a second pass may not spend the budget twice.
        return await self._router.chat(
            self._ring,
            messages,
            tools=tools,
            operation=operation,
            trace_id=trace_id,
            wait=0,
            fastest_of=fastest_of,
            parallel_recovery=parallel_recovery,
            response_format=response_format,
        )

    def stream(  # noqa: PLR0913 - the call knobs, one keyword each
        self,
        prompt: str,
        *,
        operation: str | None = None,
        trace_id: str | None = None,
        wait: float | None = None,
        fastest_of: int | None = None,
        parallel_recovery: bool = True,
        response_format: dict | None = None,
        stream_selection_window: float = 1.0,
    ) -> StreamHandle:
        """Return an owned stream that can supply another complete pool answer.
        The broker closes it; use ``async with`` for earlier release."""
        receipt = CallReceipt()
        handle = StreamHandle(
            partial(
                self._start_stream,
                receipt,
                prompt,
                operation=operation,
                trace_id=trace_id,
                wait=wait,
                fastest_of=fastest_of,
                parallel_recovery=parallel_recovery,
                response_format=response_format,
                stream_selection_window=stream_selection_window,
            ),
            receipt,
            operation=operation,
            store=self._store,
            scope=self.scope,
            observe_quality=(
                self._learner.record_quality_observed if self._learner is not None else None
            ),
            on_close=self._release_stream,
        )
        self._own_stream(handle)
        return handle

    async def _start_stream(  # noqa: PLR0913 - the call knobs, one keyword each
        self,
        receipt: CallReceipt,
        prompt: str,
        *,
        operation: str | None,
        trace_id: str | None,
        wait: float | None,
        fastest_of: int | None,
        parallel_recovery: bool,
        response_format: dict | None,
        stream_selection_window: float,
    ) -> _RoutedStream:
        await self._ensure_pool()
        messages = [{"role": "user", "content": prompt}]
        return self._router.stream(
            self._ring,
            messages,
            receipt,
            operation=operation,
            trace_id=trace_id,
            wait=wait,
            fastest_of=fastest_of,
            parallel_recovery=parallel_recovery,
            response_format=response_format,
            stream_selection_window=stream_selection_window,
            _on_exhausted=lambda exc: self._on_exhausted(exc, self._ring),
        )

    # ------------------------------------------------------------------
    # Direct single-model access (no pool, no failover)
    # ------------------------------------------------------------------

    async def direct(
        self,
        alias: str | None = None,
        *,
        name: str | None = None,
    ) -> AsyncDirectClient:
        """A client for exactly one model of your own — no pool, no failover.

        Takes exactly one of ``alias`` or ``name=``; raises ``PoolModelError``,
        ``UnknownModelError`` or ``MissingKeyError``. See ``docs/`` "Direct model calls".
        """
        cfg, key = await self.resolve_direct(alias, name=name)
        return AsyncDirectClient(
            base_url=cfg.base_url,
            model=cfg.model,
            api_key=key,
            client=self._http(),
            tool_params=cfg.tool_params,
        )

    async def resolve_direct(
        self,
        alias: str | None = None,
        *,
        name: str | None = None,
    ) -> tuple[LLMConfig, str]:
        """Look the entry up in its keyspace and resolve it against this caller's ring
        (shared by the synchronous façade)."""
        if (alias is None) == (name is None):
            raise ValueError(
                "direct() takes exactly one of alias (positional) or name= —"
                " they are separate keyspaces",
            )
        self._tick()
        stored, declared = await self._catalog.entries()
        cfg = find_declared(stored, declared, self._catalog.direct_unresolved, alias, name)
        ref = alias if alias is not None else name
        key = await self._ring.resolve(cfg.api_key_ref)
        if key is None:
            hint = self._catalog.key_help(cfg.api_key_ref)
            raise MissingKeyError(
                f"api_key_ref {cfg.api_key_ref!r} for model {ref!r} could not be resolved"
                " — set the env var or configure a secrets backend" + (f". {hint}" if hint else ""),
            )
        return cfg, key

    def _http(self) -> httpx.AsyncClient:
        """The one client of the installation — a caller opens no connection pool of
        its own."""
        return self._router.http

    # ------------------------------------------------------------------
    # Inspection and rating
    # ------------------------------------------------------------------

    async def get(self, name: str) -> AsyncLLM:
        await self._ensure_pool()
        return self._pool_view.get(name)

    async def count(self) -> int:
        await self._ensure_pool()
        return self._pool_view.count()

    async def record_quality(
        self,
        score: float,
        *,
        call_id: str | None = None,
        trace_id: str | None = None,
    ) -> None:
        """Rate a past call, by exactly one key — one rating, on the newest answered
        call the key names within the rating window. Raises ``UnknownCallError`` when
        nothing there answered. See ``docs/`` "Quality rating"."""
        check_score(score)
        if (call_id is None) == (trace_id is None):
            raise ValueError(
                "record_quality() takes exactly one of call_id= or trace_id= —"
                " a rating names the call it rates",
            )
        row = await self._resolve_rated(call_id=call_id, trace_id=trace_id)
        await self._store.record_quality(row.id, score, scope=self.scope)
        if self._learner is not None:
            self._learner.record_quality_observed(row.llm_name, row.operation, row.id, score)

    async def _resolve_rated(self, *, call_id: str | None, trace_id: str | None) -> Call:
        """The one call a rating key names: the newest that answered, inside the window.
        A trace naming more rows than one call's attempts is the host reusing it, which
        is reported — the rating still lands on exactly one call."""
        key = f"call_id={call_id!r}" if call_id is not None else f"trace_id={trace_id!r}"
        # A call id names one row; a trace names one call's attempts.
        limit = 1 if call_id is not None else _RATING_PAGE
        rows = await self.calls(
            limit=limit,
            since=datetime.now(UTC) - _RATING_WINDOW,
            call_id=call_id,
            trace_id=trace_id,
        )
        if trace_id is not None and len(rows) == _RATING_PAGE:
            logger.warning(
                "record_quality: %s matched the %d-row read bound — a trace names one"
                " call, so the rating lands on the newest that answered under it",
                key,
                _RATING_PAGE,
            )
        answered = next((row for row in rows if row.status is CallStatus.OK), None)
        if answered is None:
            raise UnknownCallError(
                f"no answered call found for {key} within the last"
                f" {_RATING_WINDOW.days} days — it may be older than that, purged by"
                " retention, or every attempt under it failed",
            )
        return answered

    # ------------------------------------------------------------------
    # Call journal — the rows this caller's scope is on
    # ------------------------------------------------------------------

    async def calls(
        self,
        *,
        limit: int,
        since: datetime | None = None,
        operation: str | None = None,
        trace_id: str | None = None,
        call_id: str | None = None,
    ) -> list[Call]:
        """Newest-first journal tail for this caller: one row per call attempt, each
        carrying the newest score it was rated with. The unscoped caller sees all.
        Never provisions the pool."""
        check_limit(limit)
        return await self._require_queryable().calls(
            limit=limit,
            scope=self.scope,
            since=to_utc(since, "since") if since is not None else None,
            operation=operation,
            trace_id=trace_id,
            call_id=call_id,
        )

    async def stats(
        self,
        *,
        since: datetime | None = None,
        limit: int = _DEFAULT_STATS_LIMIT,
        operation: str | None = None,
    ) -> Mapping[str, LLMStats]:
        """Per-model counts of this caller's call records over a window, keyed by model
        name. ``limit`` caps rows read, not the window: totals summing to it mean the
        window may be truncated. Never provisions the pool."""
        rows = await self.calls(limit=limit, since=since, operation=operation)
        return stats_from_calls(rows)

    def _require_queryable(self) -> QueryableStoreProtocol:
        if not isinstance(self._store, QueryableStoreProtocol):
            raise TypeError(
                "this store backend is not queryable — use a queryable backend"
                " (e.g. llmbroker.sqlite.Store) for calls()",
            )
        return self._store