lane_tree.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/lane_tree.py
 
"""
The selection tree — pure classification over lanes, models and families.
 
Purpose
-------
Plan 217 T1. First live use of the plan-215 lanes put 396 OpenRouter
models in one flat list. This module holds the decisions that turn that
list into a drill-down (P1): which lanes are selectable, whether a lane
has both a free and a paid class, how model ids group into families, and
when a level has so few answers that asking would be noise.
 
Everything here is a PURE function over data already owned elsewhere —
the shipped catalogue (``providers.openai_compat.catalogue``), the
operator overlay (``oc_chat.core.lane_overlay``), and whatever live list
the 215/216 machinery cached. Nothing in this module reads a key value,
opens a socket, or touches the filesystem, which is what lets the
navigation layer above it be trivially reversible.
 
Responsibilities
----------------
- :class:`Choice` — one selectable (or refused) answer at a level.
- :class:`LaneOption` — one level-one row, projected from a provider.
- :func:`classify_models` / :func:`model_classes` / :func:`models_in_class`
  — the free/paid decision (P1).
- :func:`family_of` / :func:`families` / :func:`group_by_family` /
  :func:`models_in_family` — the family decision (P1).
- :func:`collapse_choices` — the single-answer rule (P1).
- :func:`lane_options` — project registry providers into level-one rows.
- :func:`vendor_of` / :func:`vendor_detail` / :func:`serving_host_of` —
  the attribution decisions (plan 220 P4).
- :func:`section_text` — join a run of lines into ONE presenter call
  (plan 220 P2).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (lane rows projected).
    L3 — semantic details (classification and grouping outcomes).
    L4 — deep tracing.
 
Contracts
---------
- **Two things classify, and they do not fight.** Where a lane uses the
  OpenRouter ``:free`` suffix convention, the SUFFIX classifies — within
  such a lane the suffix is the discriminator, and the lane-level ``free``
  flag says nothing about an individual id. Where no id carries the
  suffix, the catalogue flag classifies every model uniformly. A lane
  with neither (deepseek, ollama) classifies as
  :data:`CLASS_UNKNOWN` rather than guessing; that is a single class, so
  the level collapses and nothing is ever claimed on screen.
- Ordering is TOTAL and deterministic everywhere: classes free-then-paid,
  families alphabetical with :data:`NO_FAMILY` last, models in the input
  order they arrived in (catalogue declaration order, or cache order).
  A picker that reflows because a dict enumerated differently is a picker
  the operator cannot learn.
- :func:`collapse_choices` counts SELECTABLE choices only, and refuses to
  collapse when there are none: a dead end must be shown with its reason,
  never silently auto-answered.
- Key handling is presence-only (plan 215 P6). :class:`LaneOption` carries
  the env-var NAME and a boolean; the VALUE never enters this module.
- **Attribution never guesses (plan 220 P4).** :func:`vendor_of` reads the
  ``vendor/model`` convention where an id spells it and returns ``""``
  where it does not — unlike :func:`family_of`, which needs a bucket for
  the unprefixed ids and so answers :data:`NO_FAMILY`. A bucket is the
  right answer for grouping and exactly the wrong one for attribution.
  :func:`serving_host_of` reads only what a lane REPORTED, because a
  routed request's serving host is knowable per response and varies call
  to call (the plan-215 catalogue's provenance caveat).
- :func:`section_text` is the module's one presentation helper, and it is
  still a pure function of strings — it returns the block; it does not
  print it. It lives here because ``startup`` and ``lane_picker`` are its
  two callers and this is the only module both already sit above.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import os
from dataclasses import dataclass
from typing import Mapping, Sequence
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
#: The three access classes a model id can carry.
CLASS_FREE = "free"
CLASS_PAID = "paid"
CLASS_UNKNOWN = "unknown"
 
#: Presentation order for the free/paid level. Free first: those are the
#: models an operator without credit can actually run today.
CLASS_ORDER: tuple[str, ...] = (CLASS_FREE, CLASS_PAID, CLASS_UNKNOWN)
 
#: Human labels for the class level.
CLASS_LABELS: Mapping[str, str] = {
    CLASS_FREE: "free models",
    CLASS_PAID: "paid models",
    CLASS_UNKNOWN: "all models",
}
 
#: The OpenRouter convention: a ``:free`` suffix marks a no-cost route.
FREE_SUFFIX = ":free"
 
#: Bucket for ids that carry no ``vendor/model`` family convention.
NO_FAMILY = "(other)"
 
 
@dataclass(frozen=True)
class Choice:
    """One answer offered at one level of the drill-down.
 
    A choice that is *not* selectable is still RENDERED — with *reason*
    saying why — because a lane the operator cannot see is a lane they
    cannot fix (plan 215 P5, applied to navigation).
    """
 
    key: str
    label: str
    detail: str = ""
    selectable: bool = True
    reason: str = ""
 
 
@dataclass(frozen=True)
class LaneOption:
    """One level-one row: a lane the operator could switch to.
 
    ``free`` is ``None`` for a provider that publishes no access tier
    (deepseek, ollama). ``models`` is the curated list shown by default;
    ``cached_models`` is whatever live list the 215/216 machinery last
    wrote, which the "show all" escape opens (P2).
    """
 
    name: str
    display_label: str
    free: bool | None
    verification: str
    api_key_env_var: str
    has_key: bool
    selectable: bool
    models: tuple[str, ...] = ()
    cached_models: tuple[str, ...] = ()
    models_fetched_at: str | None = None
 
    @property
    def badge(self) -> str:
        """``FREE`` / ``PAID`` / ``-`` — the lane's own access tier."""
        if self.free is None:
            return "-"
        return "FREE" if self.free else "PAID"
 
 
# ── Free / paid classification ──────────────────────────────────────────────
 
 
def uses_free_suffix(model_ids: Sequence[str]) -> bool:
    """Whether *model_ids* uses the ``:free`` suffix convention at all."""
    return any(str(model_id).endswith(FREE_SUFFIX) for model_id in model_ids)
 
 
def classify_models(
    model_ids: Sequence[str],
    *,
    lane_free: bool | None,
) -> dict[str, str]:
    """Map every id in *model_ids* to its access class.
 
    The suffix convention wins where it is present, because within such a
    lane it is the only thing that distinguishes two ids; elsewhere the
    catalogue flag classifies uniformly. See the module contract.
    """
    func = "classify_models"
    ids = [str(model_id) for model_id in model_ids]
    if uses_free_suffix(ids):
        classified = {
            model_id: (CLASS_FREE if model_id.endswith(FREE_SUFFIX) else CLASS_PAID)
            for model_id in ids
        }
        _dbg("OC_CHAT.MAIN", func, f"suffix convention over {len(ids)} id(s)", 4)
        return classified
    if lane_free is None:
        default = CLASS_UNKNOWN
    else:
        default = CLASS_FREE if lane_free else CLASS_PAID
    _dbg("OC_CHAT.MAIN", func, f"lane flag -> {default} over {len(ids)} id(s)", 4)
    return {model_id: default for model_id in ids}
 
 
def model_classes(
    model_ids: Sequence[str],
    *,
    lane_free: bool | None,
) -> tuple[str, ...]:
    """The distinct classes present, free first. One entry → level skipped."""
    present = set(classify_models(model_ids, lane_free=lane_free).values())
    return tuple(name for name in CLASS_ORDER if name in present)
 
 
def models_in_class(
    model_ids: Sequence[str],
    *,
    lane_free: bool | None,
    model_class: str,
) -> tuple[str, ...]:
    """The ids in *model_class*, in input order."""
    classified = classify_models(model_ids, lane_free=lane_free)
    return tuple(
        model_id for model_id in (str(m) for m in model_ids)
        if classified.get(model_id) == model_class
    )
 
 
# ── Family grouping ─────────────────────────────────────────────────────────
 
 
def family_of(model_id: str) -> str:
    """The id's family — the prefix before ``/``, or :data:`NO_FAMILY`.
 
    Only the FIRST separator counts: ``a/b/c`` is family ``a``, matching
    how every routed lane in the catalogue spells its ids.
    """
    text = str(model_id)
    head, sep, _ = text.partition("/")
    if not sep or not head.strip():
        return NO_FAMILY
    return head
 
 
def families(model_ids: Sequence[str]) -> tuple[str, ...]:
    """The distinct families, alphabetical, with :data:`NO_FAMILY` last."""
    present = {family_of(model_id) for model_id in model_ids}
    named = sorted(name for name in present if name != NO_FAMILY)
    if NO_FAMILY in present:
        named.append(NO_FAMILY)
    return tuple(named)
 
 
def models_in_family(model_ids: Sequence[str], family: str) -> tuple[str, ...]:
    """The ids belonging to *family*, in input order."""
    return tuple(
        str(model_id) for model_id in model_ids if family_of(model_id) == family
    )
 
 
def group_by_family(
    model_ids: Sequence[str],
) -> tuple[tuple[str, tuple[str, ...]], ...]:
    """Group *model_ids* by family — families ordered, members in input order."""
    func = "group_by_family"
    grouped = tuple(
        (family, models_in_family(model_ids, family))
        for family in families(model_ids)
    )
    _dbg("OC_CHAT.MAIN", func,
         f"{len(model_ids)} id(s) -> {len(grouped)} family/families", 4)
    return grouped
 
 
# ── Attribution ─────────────────────────────────────────────────────────────
 
 
#: The response fields a lane may use to name the host that actually served a
#: turn. OpenRouter answers with ``provider``; the other two are the spellings
#: in circulation elsewhere. Order is preference order.
SERVING_HOST_FIELDS: tuple[str, ...] = ("provider", "served_by", "serving_host")
 
 
def vendor_of(model_id: str) -> str:
    """The id's VENDOR — the prefix before ``/`` — or ``""`` where it has none.
 
    Deliberately not :func:`family_of` under another name. Family grouping
    needs a bucket for the ids that carry no prefix, and answers
    :data:`NO_FAMILY` for them; attribution must not, because an id that
    does not name its vendor has an UNKNOWN vendor and TUI-02 finding 4 is
    precisely about not claiming one. Only the FIRST separator counts, the
    same rule every routed lane in the catalogue spells its ids by.
    """
    head, sep, _ = str(model_id).partition("/")
    if not sep or not head.strip():
        return ""
    return head
 
 
def vendor_detail(model_id: str) -> str:
    """``"vendor: <name>"`` for a row that names one, ``""`` for one that does not.
 
    The empty string is what keeps a lane without the convention (deepseek,
    ollama) from rendering a bare ``vendor:`` label with nothing after it.
    """
    vendor = vendor_of(model_id)
    return f"vendor: {vendor}" if vendor else ""
 
 
def serving_host_of(report: object) -> str:
    """The serving host a lane REPORTED for one response, or ``""``.
 
    *report* is whatever the lane left behind after its turn, read
    duck-typed (anything with a ``get``) so this module keeps its promise
    never to import a provider bundle. An absent, non-mapping, blank or
    non-string value is ``""``, and the caller then renders nothing at all
    — a missing line is the designed outcome, not a degraded one.
    """
    func = "serving_host_of"
    get = getattr(report, "get", None)
    if not callable(get):
        return ""
    for field in SERVING_HOST_FIELDS:
        value = get(field)
        if isinstance(value, str) and value.strip():
            _dbg("OC_CHAT.MAIN", func, f"{field} reported {value.strip()!r}", 4)
            return value.strip()
    return ""
 
 
# ── Section rendering ───────────────────────────────────────────────────────
 
 
def section_text(lines: Sequence[str]) -> str:
    """Join *lines* into ONE block, dropping the blank lines at either end.
 
    Every presenter opens ``show_info`` with a blank line. Spent once per
    ROW that is TUI-02 finding 2 — a startup menu using twice the vertical
    space it needs; spent once per SECTION it is exactly the separator the
    layout wants. Callers therefore render a section by handing this block
    to a single ``show_info``, and the edge blanks go because the presenter
    already supplies the one above. Blank lines INSIDE the run survive:
    those are the caller's own sub-section breaks.
 
    Returns ``""`` for a run with no content, which callers treat as
    "nothing to show" — that is what keeps the plan-215 lane panel
    invisible on an install with no catalogue-backed bundle.
    """
    body = list(lines)
    while body and not body[0].strip():
        body.pop(0)
    while body and not body[-1].strip():
        body.pop()
    return "\n".join(body)
 
 
# ── The single-answer rule ──────────────────────────────────────────────────
 
 
def collapse_choices(choices: Sequence[Choice]) -> Choice | None:
    """The one selectable choice when a level has exactly one, else ``None``.
 
    Unselectable siblings do not block a collapse — they are displayed
    information, not answers — but a level with NO selectable choice never
    collapses, because the operator has to see why they are stuck.
    """
    selectable = [choice for choice in choices if choice.selectable]
    if len(selectable) == 1:
        return selectable[0]
    return None
 
 
# ── Level-one rows ──────────────────────────────────────────────────────────
 
 
def _lane_row(provider: object) -> object | None:
    """The catalogue lane row behind *provider*, if it carries one.
 
    Duck-typed exactly as the plan-215 menu panel is: a bundle
    contributes lane metadata by exposing a ``lane`` attribute, so this
    module never imports the optional provider bundle.
    """
    lane = getattr(provider, "lane", None)
    if lane is None:
        return None
    if not all(hasattr(lane, field) for field in ("display_label", "free", "verification")):
        return None
    return lane
 
 
def _curated_models(provider: object, lane: object | None) -> tuple[str, ...]:
    """The lane's curated ids, in declaration order."""
    if lane is not None:
        return tuple(model.model_id for model in getattr(lane, "models", ()))
    try:
        return tuple(provider.models().keys())  # type: ignore[attr-defined]
    except Exception:  # noqa: BLE001 — a provider that cannot list is simply empty
        return ()
 
 
def lane_options(
    providers: Sequence[object],
    *,
    env: Mapping[str, str] | None = None,
    cached_models: Mapping[str, Sequence[str]] | None = None,
    fetched_at: Mapping[str, str | None] | None = None,
) -> tuple[LaneOption, ...]:
    """Project *providers* into level-one rows (P1).
 
    EVERY registered provider contributes a row — the catalogue lanes and
    the plain deepseek / ollama bundles alike — because level one is "which
    lane", and a lane that is missing from the list is a lane the operator
    cannot choose. A lane whose key variable is unset is included and
    marked NOT selectable, carrying the variable name so the refusal can
    point at the wizard.
 
    Only the PRESENCE of each key variable is read; the value never enters
    the returned rows (plan 215 P6). *env* defaults to the live
    environment; callers in tests pass an explicit mapping.
    """
    func = "lane_options"
    environ = os.environ if env is None else env
    cache = dict(cached_models or {})
    stamps = dict(fetched_at or {})
 
    rows: list[LaneOption] = []
    for provider in providers:
        name = getattr(provider, "name", "") or ""
        if not name:
            continue
        lane = _lane_row(provider)
        key_var = getattr(provider, "api_key_env_var", "") or ""
        has_key = bool((environ.get(key_var) or "").strip()) if key_var else True
        rows.append(LaneOption(
            name=name,
            display_label=(getattr(lane, "display_label", "") if lane is not None else "") or name,
            free=(getattr(lane, "free", None) if lane is not None else None),
            verification=(getattr(lane, "verification", "") if lane is not None else ""),
            api_key_env_var=key_var,
            has_key=has_key,
            selectable=(not key_var) or has_key,
            models=_curated_models(provider, lane),
            cached_models=tuple(str(m) for m in cache.get(name, ())),
            models_fetched_at=stamps.get(name),
        ))
 
    # Free lanes lead — the same presentation policy the 215 panel applies,
    # and for the same reason: those are the lanes a new operator can try.
    ordered = tuple(sorted(rows, key=lambda row: (row.free is not True, row.name)))
    _dbg("OC_CHAT.MAIN", func, f"projected {len(ordered)} lane row(s)", 3)
    return ordered
 
 
__all__ = [
    "CLASS_FREE",
    "CLASS_LABELS",
    "CLASS_ORDER",
    "CLASS_PAID",
    "CLASS_UNKNOWN",
    "Choice",
    "FREE_SUFFIX",
    "LaneOption",
    "NO_FAMILY",
    "SERVING_HOST_FIELDS",
    "classify_models",
    "collapse_choices",
    "families",
    "family_of",
    "group_by_family",
    "lane_options",
    "model_classes",
    "models_in_class",
    "models_in_family",
    "section_text",
    "serving_host_of",
    "uses_free_suffix",
    "vendor_detail",
    "vendor_of",
]