lane_picker.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/lane_picker.py
 
"""
The drill-down picker — lane, class, family, model, and back out of any of them.
 
Purpose
-------
Plan 217 T2/T4. :class:`Navigator` walks the tree
:mod:`oc_chat.core.lane_tree` describes — lane, free/paid, family, model,
confirm — and :func:`run_lane_picker` drives it against a presenter and a
prompt driver.
 
The operator's requirement is reversibility, and it is met STRUCTURALLY
rather than carefully (P4). Every state is a frozen value, every
transition returns a NEW value, and the navigator keeps the values it
came from on a stack. "Back" is therefore not a re-derivation that might
disagree with what was on screen: it is the identical earlier object,
restored. Nothing outside the navigator is touched at any point — the
picker only ever RETURNS a :class:`LaneSelection`, and applying one is
:mod:`oc_chat.core.lane_switch`'s job. An abandoned traversal is a no-op
because there was never anything to undo.
 
Responsibilities
----------------
- :class:`PickerState` / :class:`Screen` / :class:`LaneSelection` — the values.
- :class:`Navigator` — choose / act / back / abandon over a state stack.
- :func:`render_screen` — ASCII rendering of one screen.
- :func:`run_lane_picker` — the interactive driver.
- :func:`lane_options_from_registry` / :func:`live_cache_from_overlay` —
  join the live registry with the cache the 215/216 machinery wrote.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (picker entered, selection made, abandoned).
    L3 — semantic details (level transitions, action outcomes).
    L4 — deep tracing.
 
Contracts
---------
- **Nothing mutates until the final confirmation.** The confirm level is
  never collapsed away, so a complete selection is always an explicit
  answer. Even then this module only returns a value.
- **Level one and the confirm level never collapse; the three middle
  levels do.** Level one IS the picker's purpose, so a single-lane
  install still sees which lane it has; confirm is P4's point of no
  return. A middle level with exactly one selectable answer answers
  itself, and ``back`` steps over the whole collapsed run as one unit,
  because a level the operator never saw is not a level they can return
  to.
- **Refresh is in-memory and the only network call.** It is reached only
  by the explicit action, and the fetched list lives in the picker state
  — it is NOT written back to the overlay. Writing it would be a disk
  mutation inside a traversal the operator may still abandon, which is
  exactly what P4 forbids. The persisted cache stays the wizard's to
  write.
- **A cache that does not exist is reported, never silently filled.** The
  "show all" escape is offered as unselectable with its reason when the
  lane has no cached list, alongside the refresh that would create one.
- The single-letter answers ``b`` / ``m`` / ``a`` / ``r`` are reserved at
  every prompt, so a lane literally named ``b`` is unreachable through the
  picker (it stays selectable by ``/lane b``). Same trade the plan-216
  wizard documents for ``q``.
- ASCII only: every rendered line survives a stock cp1252 console (F4).
- **A Ctrl+C is not a "back" (plan 220 P3).** With *on_interrupt* wired —
  the pre-session callers only — an interrupt is handed to that callback
  and, if it returns, the same screen is re-shown unchanged. Without it
  the in-session ``/lane`` behaviour is byte-identical to pre-220: a
  cancel abandons. One reader, two honest meanings, chosen by the caller
  that knows whether a session exists yet.
- **One presenter call per screen (plan 220 P2).** Screens are emitted as
  a single block, because every presenter opens ``show_info`` with a
  blank line and a screen emitted row by row is a double-spaced screen.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.lane_tree (the pure tree model this walks)
- oc_chat.core.lane_overlay (lazy; read-only: the cached live model lists)
- oc_chat.core.defaults (menu column widths from the shipped defaults file)
- oc_chat.core.client_interface (lazy; ClientInterface for the live model list fetch)
- oc_chat.core.provider_plugin (lazy; provider registry for the lane rows and the live fetch)
"""
from __future__ import annotations
 
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.core import lane_tree
from oc_chat.core.defaults import default
from oc_chat.core.lane_tree import Choice, LaneOption
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
#: The five levels, in walk order.
LEVEL_LANE = "lane"
LEVEL_CLASS = "class"
LEVEL_FAMILY = "family"
LEVEL_MODEL = "model"
LEVEL_CONFIRM = "confirm"
 
LEVEL_ORDER: tuple[str, ...] = (
    LEVEL_LANE, LEVEL_CLASS, LEVEL_FAMILY, LEVEL_MODEL, LEVEL_CONFIRM,
)
 
#: Only the three middle levels answer themselves — see the module contract.
COLLAPSIBLE_LEVELS: frozenset[str] = frozenset({LEVEL_CLASS, LEVEL_FAMILY, LEVEL_MODEL})
 
#: The confirm level's single answer key.
CONFIRM_KEY = "yes"
 
#: The two model-pool actions.
ACTION_SHOW_ALL = "show_all"
ACTION_REFRESH = "refresh"
 
#: Reserved single-letter answers, at every prompt.
BACK_ANSWER = "b"
ABANDON_ANSWERS: frozenset[str] = frozenset({"m", "q"})
SHOW_ALL_ANSWER = "a"
REFRESH_ANSWER = "r"
 
#: Menu column widths — values in the shipped defaults file (CONFIG-01).
_NAME_W = default("lane_picker_name_width")
_LABEL_W = default("lane_picker_label_width")
 
 
@dataclass(frozen=True)
class LaneSelection:
    """A complete, confirmed answer: which lane, which model."""
 
    lane_name: str
    model_id: str
 
 
@dataclass(frozen=True)
class Answer:
    """One answered level. ``auto`` marks a level the collapse rule filled."""
 
    level: str
    key: str
    auto: bool = False
 
 
@dataclass(frozen=True)
class PickerState:
    """The whole traversal as one immutable value.
 
    ``live`` holds refreshed model lists as a tuple of pairs rather than a
    dict so the state stays hashable and obviously copy-free; a traversal
    refreshes at most a handful of lanes.
    """
 
    lanes: tuple[LaneOption, ...]
    answers: tuple[Answer, ...] = ()
    show_all: bool = False
    live: tuple[tuple[str, tuple[str, ...]], ...] = ()
 
    def answer_for(self, level: str) -> str:
        """The key answered at *level*, or ``""``."""
        for answer in self.answers:
            if answer.level == level:
                return answer.key
        return ""
 
    def live_models(self, lane_name: str) -> tuple[str, ...]:
        """The list a refresh fetched for *lane_name* this traversal."""
        for name, models in self.live:
            if name == lane_name:
                return models
        return ()
 
    def lane_option(self) -> LaneOption | None:
        """The chosen lane's row, or ``None`` before level one is answered."""
        name = self.answer_for(LEVEL_LANE)
        if not name:
            return None
        return next((lane for lane in self.lanes if lane.name == name), None)
 
 
@dataclass(frozen=True)
class Screen:
    """One rendered level: what to show and what may be answered."""
 
    level: str
    title: str
    choices: tuple[Choice, ...]
    actions: tuple[Choice, ...] = ()
    highlight: str = ""
    can_back: bool = True
    can_abandon: bool = True
 
 
# ── Pure tree walking ───────────────────────────────────────────────────────
 
 
def current_level(state: PickerState) -> str | None:
    """The level awaiting an answer, or ``None`` when the walk is complete."""
    for level in LEVEL_ORDER:
        if not state.answer_for(level):
            return level
    return None
 
 
def full_pool(state: PickerState, option: LaneOption) -> tuple[str, ...]:
    """The lane's live-discovered list: this traversal's refresh, else the cache."""
    return state.live_models(option.name) or option.cached_models
 
 
def _base_pool(state: PickerState, option: LaneOption) -> tuple[str, ...]:
    """Curated by default; the full live list once "show all" is engaged (P2)."""
    if state.show_all:
        return full_pool(state, option)
    return option.models
 
 
def pool_for(state: PickerState, level: str) -> tuple[str, ...]:
    """The model ids still in play when *level* is being asked."""
    option = state.lane_option()
    if option is None:
        return ()
    ids = _base_pool(state, option)
    if level in (LEVEL_LANE, LEVEL_CLASS):
        return ids
    chosen_class = state.answer_for(LEVEL_CLASS)
    if chosen_class:
        ids = lane_tree.models_in_class(
            ids, lane_free=option.free, model_class=chosen_class
        )
    if level == LEVEL_FAMILY:
        return ids
    chosen_family = state.answer_for(LEVEL_FAMILY)
    if chosen_family:
        ids = lane_tree.models_in_family(ids, chosen_family)
    return ids
 
 
def _lane_choices(state: PickerState) -> tuple[Choice, ...]:
    out: list[Choice] = []
    for option in state.lanes:
        detail = (
            f"{option.badge:<5} "
            f"verified: {(option.verification or '-'):<7}"
            f"{len(option.models)} curated"
        )
        reason = (
            ""
            if option.selectable
            else (
                f"no key ({option.api_key_env_var}) - set it, or onboard the lane "
                f"with [5] Add provider"
            )
        )
        out.append(Choice(
            key=option.name,
            label=option.display_label if option.display_label != option.name else "",
            detail=detail,
            selectable=option.selectable,
            reason=reason,
        ))
    return tuple(out)
 
 
def _class_choices(state: PickerState) -> tuple[Choice, ...]:
    option = state.lane_option()
    if option is None:
        return ()
    ids = pool_for(state, LEVEL_CLASS)
    out: list[Choice] = []
    for name in lane_tree.model_classes(ids, lane_free=option.free):
        members = lane_tree.models_in_class(
            ids, lane_free=option.free, model_class=name
        )
        out.append(Choice(
            key=name,
            label=lane_tree.CLASS_LABELS.get(name, name),
            detail=f"{len(members)} model(s)",
        ))
    return tuple(out)
 
 
def _family_choices(state: PickerState) -> tuple[Choice, ...]:
    ids = pool_for(state, LEVEL_FAMILY)
    return tuple(
        Choice(key=family, label="", detail=f"{len(members)} model(s)")
        for family, members in lane_tree.group_by_family(ids)
    )
 
 
def _model_choices(state: PickerState) -> tuple[Choice, ...]:
    # Plan 220 P4: a routed lane's id names its VENDOR, and on a lane like
    # OpenRouter that is the only thing on the row that says who actually
    # built the model. An id without the convention gets no vendor line
    # rather than a guessed one (lane_tree.vendor_detail returns "").
    return tuple(
        Choice(key=model_id, label="", detail=lane_tree.vendor_detail(model_id))
        for model_id in pool_for(state, LEVEL_MODEL)
    )
 
 
def _confirm_choices(state: PickerState) -> tuple[Choice, ...]:
    lane = state.answer_for(LEVEL_LANE)
    model = state.answer_for(LEVEL_MODEL)
    # Plan 220 P4: the point of no return names the lane AND the vendor, so
    # the last thing on screen before the switch is the whole attribution.
    return (Choice(
        key=CONFIRM_KEY,
        label=f"switch to {lane} / {model}",
        detail=lane_tree.vendor_detail(model),
    ),)
 
 
_CHOICE_BUILDERS: Mapping[str, Callable[[PickerState], tuple[Choice, ...]]] = {
    LEVEL_LANE: _lane_choices,
    LEVEL_CLASS: _class_choices,
    LEVEL_FAMILY: _family_choices,
    LEVEL_MODEL: _model_choices,
    LEVEL_CONFIRM: _confirm_choices,
}
 
 
def choices_for(state: PickerState) -> tuple[Choice, ...]:
    """The choices at the level awaiting an answer."""
    level = current_level(state)
    if level is None:
        return ()
    return _CHOICE_BUILDERS[level](state)
 
 
def collapse(state: PickerState) -> PickerState:
    """Answer every collapsible level that has exactly one selectable answer."""
    func = "collapse"
    while True:
        level = current_level(state)
        if level is None or level not in COLLAPSIBLE_LEVELS:
            return state
        only = lane_tree.collapse_choices(_CHOICE_BUILDERS[level](state))
        if only is None:
            return state
        state = replace(
            state, answers=state.answers + (Answer(level, only.key, auto=True),)
        )
        _dbg("OC_CHAT.MAIN", func, f"level {level!r} collapsed to {only.key!r}", 4)
 
 
def advance(state: PickerState, key: str) -> PickerState | None:
    """Answer the current level with *key*, or ``None`` when it is not an answer."""
    func = "advance"
    level = current_level(state)
    if level is None:
        return None
    match = next(
        (c for c in _CHOICE_BUILDERS[level](state) if c.key == key and c.selectable),
        None,
    )
    if match is None:
        return None
    _dbg("OC_CHAT.MAIN", func, f"level {level!r} answered {key!r}", 3)
    return collapse(
        replace(state, answers=state.answers + (Answer(level, key),))
    )
 
 
def _drop_levels(state: PickerState, levels: frozenset[str]) -> PickerState:
    """Forget the answers at *levels* (used when the model pool changes)."""
    return replace(
        state, answers=tuple(a for a in state.answers if a.level not in levels)
    )
 
 
#: The answers a pool change invalidates. The lane and the class survive it
#: where they still make sense — an operator who asked for free models still
#: wants free models after opening the full list.
_POOL_DEPENDENT: frozenset[str] = frozenset({LEVEL_FAMILY, LEVEL_MODEL, LEVEL_CONFIRM})
 
 
def apply_show_all(state: PickerState) -> PickerState | None:
    """Open the full live list, keeping the lane (and the class where it fits).
 
    ``None`` when there is nothing to open — the lane has no cached list and
    none has been fetched this traversal (P2: an empty cache is not an
    implicit licence to reach the network).
    """
    func = "apply_show_all"
    option = state.lane_option()
    if option is None or not full_pool(state, option):
        return None
    opened = replace(_drop_levels(state, _POOL_DEPENDENT), show_all=True)
    # A class answer that survives into an empty pool would strand the walk;
    # drop it and let the class level be asked again over the new list.
    if state.answer_for(LEVEL_CLASS) and not pool_for(opened, LEVEL_FAMILY):
        opened = _drop_levels(opened, frozenset({LEVEL_CLASS}))
    _dbg("OC_CHAT.MAIN", func,
         f"lane={option.name} opened {len(full_pool(state, option))} live id(s)", 3)
    return collapse(opened)
 
 
def apply_refresh(
    state: PickerState,
    fetcher: Callable[[LaneOption], Sequence[str]],
) -> PickerState:
    """Fetch the lane's live list into the STATE — never onto disk.
 
    Raises whatever *fetcher* raises; the navigator turns that into a
    reported failure with the state untouched.
    """
    func = "apply_refresh"
    option = state.lane_option()
    if option is None:
        raise ValueError("no lane chosen yet")
    fetched = tuple(str(model_id) for model_id in fetcher(option))
    live = tuple(pair for pair in state.live if pair[0] != option.name)
    _dbg("OC_CHAT.MAIN", func,
         f"lane={option.name} refreshed to {len(fetched)} id(s)", 2)
    return replace(state, live=live + ((option.name, fetched),))
 
 
# ── The navigator ───────────────────────────────────────────────────────────
 
 
class Navigator:
    """Walk the tree over a stack of states, so ``back`` restores exactly.
 
    Not thread-safe and not meant to be: one navigator per picker run, on
    the main thread, between turns.
    """
 
    def __init__(
        self,
        lanes: Sequence[LaneOption],
        *,
        fetcher: Callable[[LaneOption], Sequence[str]] | None = None,
    ) -> None:
        func = "Navigator.__init__"
        self._fetcher = fetcher
        self._stack: list[tuple[PickerState, str]] = [
            (collapse(PickerState(lanes=tuple(lanes))), "")
        ]
        self._abandoned = False
        self._undone = ""
        self.last_error = ""
        _dbg("OC_CHAT.MAIN", func, f"picker opened over {len(lanes)} lane(s)", 2)
 
    # ── Reading ─────────────────────────────────────────────────────────────
 
    @property
    def state(self) -> PickerState:
        """The live state. Valid even after abandoning (it is a value)."""
        return self._stack[-1][0]
 
    @property
    def abandoned(self) -> bool:
        """Whether the traversal was given up, by ``back`` or by ``abandon``."""
        return self._abandoned
 
    def pending_model(self) -> str:
        """The model the walk has settled on, or ``""``."""
        return self.state.answer_for(LEVEL_MODEL)
 
    def selection(self) -> LaneSelection | None:
        """The confirmed selection, or ``None`` while incomplete or abandoned."""
        if self._abandoned or current_level(self.state) is not None:
            return None
        return LaneSelection(
            lane_name=self.state.answer_for(LEVEL_LANE),
            model_id=self.state.answer_for(LEVEL_MODEL),
        )
 
    def screen(self) -> Screen | None:
        """The screen to render, or ``None`` when abandoned or complete."""
        if self._abandoned:
            return None
        state = self.state
        level = current_level(state)
        if level is None:
            return None
        return Screen(
            level=level,
            title=_title_for(state, level),
            choices=_CHOICE_BUILDERS[level](state),
            actions=self._actions(state, level),
            highlight=self._undone,
        )
 
    def _actions(self, state: PickerState, level: str) -> tuple[Choice, ...]:
        """The pool actions, offered on every screen below the lane level.
 
        Offering them only at the model level would strand the operator
        whenever that level collapsed away, which is precisely the
        single-model lane where "show me everything" is most wanted.
        """
        option = state.lane_option()
        if option is None or level == LEVEL_LANE:
            return ()
        live = full_pool(state, option)
        if state.show_all:
            show_all = Choice(
                key=ACTION_SHOW_ALL,
                label=f"showing all {len(live)} live model(s)",
                selectable=False,
                reason="already showing the full list",
            )
        elif live:
            stamp = option.models_fetched_at or "this session"
            show_all = Choice(
                key=ACTION_SHOW_ALL,
                label=f"show all {len(live)} (cached {stamp})",
            )
        else:
            show_all = Choice(
                key=ACTION_SHOW_ALL,
                label="show all",
                selectable=False,
                reason=(
                    f"no cached model list for {option.name}; "
                    f"'{REFRESH_ANSWER}' fetches one"
                ),
            )
        if self._fetcher is None:
            refresh = Choice(
                key=ACTION_REFRESH,
                label="refresh the live list",
                selectable=False,
                reason="no live fetcher is wired into this picker",
            )
        else:
            refresh = Choice(
                key=ACTION_REFRESH, label="refresh the live list (one network call)"
            )
        return (show_all, refresh)
 
    # ── Transitions ─────────────────────────────────────────────────────────
 
    def choose(self, key: str) -> bool:
        """Answer the current level. ``False`` leaves the state untouched."""
        if self._abandoned:
            return False
        nxt = advance(self.state, key)
        if nxt is None:
            return False
        self._stack.append((nxt, key))
        self._undone = ""
        self.last_error = ""
        return True
 
    def act(self, action: str) -> bool:
        """Apply a pool action. ``False`` leaves the state untouched."""
        func = "Navigator.act"
        if self._abandoned:
            return False
        offered = next(
            (a for a in self._actions(self.state, current_level(self.state) or "")
             if a.key == action),
            None,
        )
        if offered is None or not offered.selectable:
            self.last_error = offered.reason if offered is not None else (
                f"unknown action {action!r}"
            )
            return False
        if action == ACTION_SHOW_ALL:
            nxt = apply_show_all(self.state)
            if nxt is None:
                self.last_error = "there is no live list to open"
                return False
        else:
            try:
                nxt = apply_refresh(self.state, self._fetcher)  # type: ignore[arg-type]
            except Exception as exc:  # noqa: BLE001 — any client failure is reported
                # EXTERNAL_ERROR (S6 L2): the lane refused; the traversal is intact.
                _dbg("OC_CHAT.MAIN", func,
                     f"EXTERNAL_ERROR: refresh failed: {type(exc).__name__}", 2,
                     override=True)
                self.last_error = f"could not refresh: {type(exc).__name__}: {exc}"
                return False
        self._stack.append((nxt, action))
        self._undone = ""
        self.last_error = ""
        return True
 
    def back(self) -> bool:
        """Step back one screen. ``False`` means the walk was abandoned.
 
        Stepping back off the first level IS abandoning — P4 asks for back
        at every level, and there is nothing above level one.
        """
        func = "Navigator.back"
        if self._abandoned:
            return False
        self.last_error = ""
        if len(self._stack) == 1:
            self._abandoned = True
            _dbg("OC_CHAT.MAIN", func, "backed out of the first level - abandoned", 2)
            return False
        _, key = self._stack.pop()
        self._undone = key
        return True
 
    def abandon(self) -> None:
        """Give up the whole traversal. Nothing was ever mutated."""
        func = "Navigator.abandon"
        self._abandoned = True
        _dbg("OC_CHAT.MAIN", func, "traversal abandoned", 2)
 
 
def _title_for(state: PickerState, level: str) -> str:
    """One ASCII line naming where in the tree this screen sits.
 
    Plan 220 P4: every screen below level one lists models from ONE lane,
    so the header names it — ``lane openrouter (free) / z-ai: model``. A
    list that mixes lanes is level one, where the lane is on every row.
    """
    lane = state.answer_for(LEVEL_LANE)
    chosen_class = state.answer_for(LEVEL_CLASS)
    family = state.answer_for(LEVEL_FAMILY)
    scope = f"lane {lane}" if lane else lane
    if chosen_class and chosen_class != lane_tree.CLASS_UNKNOWN:
        scope = f"{scope} ({chosen_class})"
    if family and family != lane_tree.NO_FAMILY:
        scope = f"{scope} / {family}"
    if level == LEVEL_LANE:
        return "choose a lane"
    if level == LEVEL_CLASS:
        return f"{scope}: free or paid models?"
    if level == LEVEL_FAMILY:
        return f"{scope}: model family"
    if level == LEVEL_MODEL:
        return f"{scope}: model"
    return f"{scope}: confirm"
 
 
# ── Rendering ───────────────────────────────────────────────────────────────
 
 
def _cell(text: str, width: int) -> str:
    """*text* padded to *width* — or given ONE separating space if it overflows.
 
    Plan 220: a routed model id is routinely longer than its column
    (``switch to openrouter / z-ai/glm-4.6``), and the bare ``:<`` form
    then butts the next column straight onto it —
    ``...glm-4.6vendor: z-ai``. Truncating would hide the id, so the
    column simply gives way: one space, and the row stays readable.
    """
    if len(text) < width:
        return f"{text:<{width}}"
    return text + " "
 
 
def render_screen(screen: Screen) -> list[str]:
    """Render *screen* as ASCII lines for a presenter to show."""
    lines = ["", f"  Lane picker - {screen.title}", "  " + "-" * 60]
    for index, choice in enumerate(screen.choices, 1):
        marker = ">" if choice.key == screen.highlight else " "
        body = _cell(choice.key, _NAME_W) + _cell(choice.label, _LABEL_W) + choice.detail
        lines.append(f"  {marker} {index:>2}. {body}".rstrip())
        if not choice.selectable and choice.reason:
            lines.append(f"          {choice.reason}")
    for action in screen.actions:
        letter = SHOW_ALL_ANSWER if action.key == ACTION_SHOW_ALL else REFRESH_ANSWER
        if action.selectable:
            lines.append(f"     [{letter}] {action.label}")
        else:
            lines.append(f"     [{letter}] {action.label} - {action.reason}")
    hint = f"  Enter number - '{BACK_ANSWER}' back - 'm' main menu"
    if screen.highlight:
        hint += f" - Enter repeats {screen.highlight!r}"
    lines.append(hint)
    return lines
 
 
def prompt_for(screen: Screen) -> str:
    """The one-line prompt shown under *screen*."""
    return f"  {screen.level.capitalize()} choice: "
 
 
# ── Registry / overlay joins ────────────────────────────────────────────────
 
 
def live_cache_from_overlay(
    path: str | Path | None = None,
) -> tuple[dict[str, tuple[str, ...]], dict[str, str | None]]:
    """The cached live lists the 215/216 machinery wrote, by lane name (P2).
 
    Degrades to empty on a missing OR malformed overlay. The startup hook
    ``_load_provider_overlays`` already reports a malformed overlay loudly
    and is the right place for that; a second fatal reporter here would
    turn a bad row into a picker that cannot open.
    """
    func = "live_cache_from_overlay"
    from oc_chat.core.lane_overlay import DEFAULT_OVERLAY_PATH, OverlayError, load_overlay
 
    target = Path(path) if path is not None else DEFAULT_OVERLAY_PATH
    try:
        rows = load_overlay(target)
    except (OverlayError, OSError) as exc:
        # VALIDATION_ERROR (S6 L3) — operator-authored data; the picker degrades
        # to "no cache" and the startup hook surfaces the row itself.
        _dbg("OC_CHAT.MAIN", func,
             f"VALIDATION_ERROR: {type(exc).__name__}: {exc}", 3)
        return {}, {}
    cached = {row.name: tuple(row.cached_models) for row in rows}
    stamps = {row.name: row.models_fetched_at for row in rows}
    _dbg("OC_CHAT.MAIN", func, f"cache for {len(cached)} overlay lane(s)", 3)
    return cached, stamps
 
 
def lane_options_from_registry(
    *,
    env: Mapping[str, str] | None = None,
    overlay_path: str | Path | None = None,
) -> tuple[LaneOption, ...]:
    """Level-one rows for every registered provider, cache joined in."""
    from oc_chat.core.provider_plugin import registry
 
    cached, stamps = live_cache_from_overlay(overlay_path)
    providers = [registry.get(name) for name in registry.names()]
    return lane_tree.lane_options(
        providers, env=env, cached_models=cached, fetched_at=stamps
    )
 
 
def default_fetcher(option: LaneOption) -> tuple[str, ...]:
    """Ask the lane's own API for its live model list. THE network call.
 
    Goes through the provider's ``fetch_live_models``, which is the same
    call :meth:`ModelManager.fetch_live` makes, so a lane that works in a
    session works here. Never called by the test suite.
    """
    func = "default_fetcher"
    from oc_chat.core.client_interface import ClientInterface
    from oc_chat.core.provider_plugin import registry
 
    provider = registry.get(option.name)
    client = ClientInterface(provider)
    models = tuple(str(model_id) for model_id in provider.fetch_live_models(client))
    _dbg("OC_CHAT.MAIN", func, f"lane={option.name} live list = {len(models)}", 2)
    return models
 
 
# ── The interactive driver ──────────────────────────────────────────────────
 
 
def _read(driver: Any, prompt_text: str, *, interruptible: bool = False) -> str | None:
    """Read one answer. ``None`` on cancel, EOF, or a non-interactive stdin.
 
    With *interruptible* set (plan 220 P3 — the picker reached from the
    opening menu, where no session exists yet) a Ctrl+C is RAISED instead
    of folded into the abandon path, so the caller can answer it with the
    quit confirmation and then re-show this very screen. A driver reports a
    cancelled menu prompt by RETURNING ``None`` rather than raising, so
    that is the interrupt too: coalescing it with "no answer" is exactly
    the swallow TUI-02 finding 3 describes.
 
    Without it the pre-220 behaviour is byte-identical, which is what keeps
    the in-session ``/lane`` semantics ("cancel abandons, nothing else")
    untouched.
    """
    try:
        if driver is not None:
            raw = driver.read_user_input(None, prompt_text=prompt_text)
        else:
            raw = input(prompt_text)
    except KeyboardInterrupt:
        if interruptible:
            raise
        return None
    except (EOFError, OSError):
        return None
    if raw is None:
        if interruptible:
            raise KeyboardInterrupt
        return None
    return raw.strip()
 
 
def _resolve_answer(screen: Screen, raw: str) -> str:
    """Map a typed answer to a choice key, or ``""`` when it matches none.
 
    Three forms, in order: the displayed index, the key typed in full, and
    an UNAMBIGUOUS case-insensitive prefix of one. The prefix form is what
    makes ``y`` answer the confirm screen — the obvious keystroke there —
    and it lets a long model id be picked by its distinctive head. It
    resolves only when exactly one selectable key matches, so it can never
    silently pick the wrong model.
    """
    if not raw:
        highlighted = next(
            (c for c in screen.choices if c.key == screen.highlight and c.selectable),
            None,
        )
        if highlighted is not None:
            return highlighted.key
        first = next((c for c in screen.choices if c.selectable), None)
        return first.key if first is not None else ""
    if raw.isdigit():
        index = int(raw)
        if 1 <= index <= len(screen.choices):
            return screen.choices[index - 1].key
        return ""
    exact = next((c.key for c in screen.choices if c.key == raw), "")
    if exact:
        return exact
    lowered = raw.lower()
    prefixed = [
        c.key for c in screen.choices
        if c.selectable and c.key.lower().startswith(lowered)
    ]
    return prefixed[0] if len(prefixed) == 1 else ""
 
 
def run_lane_picker(
    presenter: Any,
    *,
    prompt_driver: Any = None,
    lanes: Sequence[LaneOption] | None = None,
    env: Mapping[str, str] | None = None,
    fetcher: Callable[[LaneOption], Sequence[str]] | None = None,
    overlay_path: str | Path | None = None,
    on_interrupt: Callable[[], None] | None = None,
) -> LaneSelection | None:
    """Drive the picker to a confirmed selection, or ``None``.
 
    ``None`` means "nothing was chosen" in every case — stepped back off
    the first level, escaped to the main menu, or a closed stdin. In none
    of them has anything been mutated: applying a selection is
    :func:`oc_chat.core.lane_switch.switch_lane`'s job, and it is only
    ever reached with a returned value.
 
    *on_interrupt* (plan 220 P3) is supplied ONLY by the pre-session
    callers — the opening menu and its new-session path — where a Ctrl+C
    must interrupt and confirm rather than quietly abandon. It is called
    with the traversal untouched and either raises (the operator confirmed
    the quit) or returns, in which case the SAME screen is re-shown:
    "back" already means back, so an interrupt must not silently mean it
    too. Omitted, the picker keeps its pre-220 in-session behaviour
    exactly.
    """
    func = "run_lane_picker"
    if lanes is None:
        lanes = lane_options_from_registry(env=env, overlay_path=overlay_path)
        if fetcher is None:
            fetcher = default_fetcher
    if not lanes:
        presenter.show_info("  No provider lanes are registered.")
        return None
 
    navigator = Navigator(lanes, fetcher=fetcher)
    while True:
        screen = navigator.screen()
        if screen is None:
            break
        # Plan 220 P2: ONE call per screen. Emitted line by line, every
        # presenter's leading newline turns a ten-row screen into a
        # twenty-line one — the same double spacing TUI-02 filed against
        # the menu, and the picker is now the new-session path.
        block = lane_tree.section_text(render_screen(screen))
        if block:
            presenter.show_info(block)
        try:
            raw = _read(
                prompt_driver, prompt_for(screen),
                interruptible=on_interrupt is not None,
            )
        except KeyboardInterrupt:
            # Raises when the operator confirms the quit; returns when they
            # decline, and the loop then re-renders THIS screen with the
            # traversal exactly as it was.
            on_interrupt()  # type: ignore[misc] — only reachable when set
            continue
        if raw is None:
            navigator.abandon()
            break
        answer = raw.lower()
        if answer in ABANDON_ANSWERS:
            navigator.abandon()
            break
        if answer == BACK_ANSWER:
            if not navigator.back():
                break
            continue
        if answer in (SHOW_ALL_ANSWER, REFRESH_ANSWER):
            action = ACTION_SHOW_ALL if answer == SHOW_ALL_ANSWER else ACTION_REFRESH
            if not navigator.act(action):
                presenter.show_info(f"  {navigator.last_error}")
            continue
        key = _resolve_answer(screen, raw)
        if not key or not navigator.choose(key):
            refused = next((c for c in screen.choices if c.key == key), None)
            if refused is not None and refused.reason:
                presenter.show_info(f"  {refused.reason}")
            else:
                presenter.show_info(f"  Not a choice here: {raw!r}")
 
    selection = navigator.selection()
    _dbg("OC_CHAT.MAIN", func,
         f"picker closed: {selection if selection else 'no selection'}", 2)
    return selection
 
 
__all__ = [
    "ABANDON_ANSWERS",
    "ACTION_REFRESH",
    "ACTION_SHOW_ALL",
    "Answer",
    "BACK_ANSWER",
    "COLLAPSIBLE_LEVELS",
    "CONFIRM_KEY",
    "LEVEL_CLASS",
    "LEVEL_CONFIRM",
    "LEVEL_FAMILY",
    "LEVEL_LANE",
    "LEVEL_MODEL",
    "LEVEL_ORDER",
    "LaneSelection",
    "Navigator",
    "PickerState",
    "REFRESH_ANSWER",
    "SHOW_ALL_ANSWER",
    "Screen",
    "advance",
    "apply_refresh",
    "apply_show_all",
    "choices_for",
    "collapse",
    "current_level",
    "default_fetcher",
    "full_pool",
    "lane_options_from_registry",
    "live_cache_from_overlay",
    "pool_for",
    "prompt_for",
    "render_screen",
    "run_lane_picker",
]