lane_switch.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/lane_switch.py
 
"""
In-session lane switching — rebuild the client and model manager, keep the talk.
 
Purpose
-------
Plan 217 T3 / P3. Choosing a lane is a real switch, not a hint for the
next launch: the active :class:`~oc_chat.core.client_interface.ClientInterface`
and :class:`~oc_chat.core.models.ModelManager` are rebuilt for the new
provider, the turn after the switch runs on the new lane, and the
conversation history and session identity survive untouched.
``OC_CHAT_PROVIDER`` keeps its one remaining job: the startup default.
 
The shape is plan 152's PREPARE / COMMIT split, for the reason that plan
found: the old ``/endpoint use`` committed five things in sequence under
an unguarded dispatch, and a Ctrl+C between any two of them split the
tuple. Here every fallible step — registry lookup, key presence, client
construction, model-registry build, endpoint re-derivation — happens
BEFORE anything is installed, and the installs themselves are pure
assignments inside
:func:`~oc_chat.core.session_transition.uninterruptible_commit`.
 
Responsibilities
----------------
- :exc:`LaneKeyMissingError` — the typed keyless refusal (W5).
- :class:`PreparedLaneSwitch` — everything built, nothing installed.
- :func:`prepare_lane_switch` — the fallible half.
- :func:`commit_lane_switch` — the no-fail half.
- :func:`switch_lane` — prepare-then-commit for callers with nothing to
  do in between.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (switch prepared, switch committed, refusal).
    L3 — semantic details (lane and model names).
    L4 — deep tracing.
 
Contracts
---------
- **The target is duck-typed on purpose.** It needs ``client`` and
  ``model_manager``; it MAY carry ``session``, ``agent``, ``config`` and
  ``endpoint_registry``. That covers both callers without either of them
  knowing about the other: the slash-command ``Context`` in a live
  session, and the bare :class:`~oc_chat.core.agent.Agent` at the opening
  menu, where no session exists yet.
- **The agent's own references move with the context's.** ``Agent.client``
  is the object ``run_turn`` passes to the provider, so updating only
  ``ctx.client`` would leave the NEXT TURN on the old lane while every
  command reported the new one. When the target exposes an ``agent``, both
  are written in the same commit block.
- **A keyless lane is refused, never fallen back from.** The refusal names
  the env var and points at the wizard. It is raised in PREPARE, so a
  refused switch has touched nothing at all.
- **The endpoint registry does not survive a provider swap.** It is
  provider-scoped, and ``ClientInterface.rebind_to`` enforces only the
  forbidden-host rule — so a stale registry would let ``/endpoint use``
  rebind the NEW lane's client to an OLD provider's base URL. It is
  rebuilt in PREPARE and replaced at COMMIT.
- Key handling is presence-only in this module's own logic (plan 215 P6):
  the value is read from the environment mapping solely to hand to
  ``ClientInterface``, and never logged, returned, or stored on the
  prepared value's own fields.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.client_interface (lazy; the rebuilt provider facade)
- oc_chat.core.models (lazy; the rebuilt model registry)
- oc_chat.core.session_transition (lazy; the shared no-fail commit block)
- oc_chat.core.endpoint_registry (lazy; the rebuilt provider-scoped endpoint registry)
- oc_chat.core.provider_oauth (lazy; API key resolution for the chosen lane)
- oc_chat.core.provider_plugin (lazy; provider registry lookup of the chosen lane)
"""
from __future__ import annotations
 
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
class LaneKeyMissingError(PermissionError):
    """The chosen lane has no API key set — refused at selection (P3).
 
    SECURITY_ERROR-adjacent (S6 L1 class of "refuse, never route around"),
    but the operator-facing condition is mundane and fixable, so the
    message names the variable and says where to set it. Raised in
    PREPARE: nothing has been rebuilt when it surfaces.
    """
 
    def __init__(self, lane_name: str, env_var: str) -> None:
        self.lane_name = lane_name
        self.env_var = env_var
        super().__init__(
            f"lane {lane_name!r} has no API key: set {env_var}, or onboard the "
            f"lane with [5] Add provider in the opening menu"
        )
 
 
@dataclass(frozen=True)
class PreparedLaneSwitch:
    """A switch fully built and not yet installed.
 
    Every field here was produced by a call that could have failed. Once
    this value exists, committing it is assignment only.
    """
 
    lane_name: str
    model_id: str
    previous_lane: str
    provider: Any
    client: Any
    model_manager: Any
    endpoint_registry: Any = field(default=None)
 
 
def _environ(env: Mapping[str, str] | None) -> Mapping[str, str]:
    return os.environ if env is None else env
 
 
def _build_endpoint_registry(provider: Any) -> Any:
    """Re-derive the provider-scoped endpoint registry, best effort.
 
    Mirrors ``main.py``'s own startup init, including its posture: a
    registry that cannot be built degrades to ``None`` (``/endpoint``
    then reports that there is none) rather than aborting a switch the
    operator asked for.
    """
    func = "_build_endpoint_registry"
    try:
        from oc_chat.core.endpoint_registry import EndpointRegistry
 
        registry = EndpointRegistry(provider)
        registry.load_env(os.environ)
        registry.load_toml(Path.home() / ".oc_chat" / "endpoints.toml")
        return registry
    except Exception as exc:  # noqa: BLE001 — best effort, exactly as main() is
        # SYSTEM_ERROR (S6 L1) — surfaced through _dbg; the switch still lands.
        _dbg("OC_CHAT.MAIN", func,
             f"SYSTEM_ERROR: endpoint registry rebuild: {type(exc).__name__}: {exc}",
             1, override=True)
        return None
 
 
def prepare_lane_switch(
    target: Any,
    lane_name: str,
    model_id: str = "",
    *,
    env: Mapping[str, str] | None = None,
) -> PreparedLaneSwitch:
    """Build everything the switch needs. Raises rather than half-applying.
 
    :raises KeyError: no lane called *lane_name* is registered.
    :raises LaneKeyMissingError: the lane declares a key variable that is
        unset or blank.
    """
    func = "prepare_lane_switch"
    from oc_chat.core.client_interface import ClientInterface
    from oc_chat.core.models import ModelManager
    from oc_chat.core.provider_plugin import registry
 
    environ = _environ(env)
    provider = registry.get(lane_name)  # KeyError names the available lanes
 
    key_var = getattr(provider, "api_key_env_var", "") or ""
    from oc_chat.core.provider_oauth import resolve_api_key
    api_key = resolve_api_key(provider, environ=environ)
    if key_var and not api_key:
        _dbg("OC_CHAT.MAIN", func,
             f"refused {lane_name!r}: {key_var} is unset", 2, override=True)
        raise LaneKeyMissingError(lane_name, key_var)
 
    url_var = getattr(provider, "base_url_env_var", "") or ""
    base_url = (
        (environ.get(url_var) or "") if url_var else ""
    ) or getattr(provider, "default_base_url", "")
 
    chosen = model_id or getattr(provider, "default_model", "") or ""
    client = ClientInterface(provider, api_key=api_key, base_url=base_url)
    manager = ModelManager(provider)
    if chosen:
        manager.ensure(chosen)
 
    previous = getattr(getattr(target, "client", None), "provider", None)
    prepared = PreparedLaneSwitch(
        lane_name=lane_name,
        model_id=chosen,
        previous_lane=getattr(previous, "name", "") or "",
        provider=provider,
        client=client,
        model_manager=manager,
        endpoint_registry=_build_endpoint_registry(provider),
    )
    _dbg("OC_CHAT.MAIN", func,
         f"prepared {prepared.previous_lane or '(none)'} -> {lane_name} "
         f"model={chosen!r} (nothing installed yet)", 2)
    return prepared
 
 
def commit_lane_switch(target: Any, prepared: PreparedLaneSwitch) -> None:
    """Install *prepared* on *target*. Assignment only — this cannot fail.
 
    Order matters only for readability, because the whole block runs with
    Ctrl+C deferred: engine objects first, then the agent's mirrors, then
    the session's model and the now-stale endpoint state.
    """
    func = "commit_lane_switch"
    from oc_chat.core.session_transition import uninterruptible_commit
 
    with uninterruptible_commit():
        target.client = prepared.client
        target.model_manager = prepared.model_manager
 
        agent = getattr(target, "agent", None)
        if agent is not None:
            agent.client = prepared.client
            agent.model_manager = prepared.model_manager
 
        for holder in (agent, target):
            config = getattr(holder, "config", None)
            if config is not None and hasattr(config, "provider_name"):
                config.provider_name = prepared.lane_name
 
        session = getattr(target, "session", None)
        if session is not None:
            if prepared.model_id:
                session.set_model(prepared.model_id)
            if hasattr(session, "endpoint_name"):
                # Named an endpoint of the OUTGOING provider; meaningless now.
                session.endpoint_name = ""
            session.dirty = True
 
        if hasattr(target, "endpoint_registry"):
            target.endpoint_registry = prepared.endpoint_registry
 
    _dbg("OC_CHAT.MAIN", func,
         f"committed lane {prepared.lane_name} model={prepared.model_id!r} "
         f"(history preserved)", 2)
 
 
def switch_lane(
    target: Any,
    lane_name: str,
    model_id: str = "",
    *,
    env: Mapping[str, str] | None = None,
) -> PreparedLaneSwitch:
    """Prepare and commit in one call; returns what was installed."""
    prepared = prepare_lane_switch(target, lane_name, model_id, env=env)
    commit_lane_switch(target, prepared)
    return prepared
 
 
__all__ = [
    "LaneKeyMissingError",
    "PreparedLaneSwitch",
    "commit_lane_switch",
    "prepare_lane_switch",
    "switch_lane",
]