lane_overlay.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/lane_overlay.py
 
"""
The untracked lane overlay — operator-added lanes, shipped rows untouched.
 
Purpose
-------
Plan 216 P2. The onboarding wizard never opens the shipped catalogue for
writing. It writes ``.oc_chat/lanes_overlay.json`` instead — the
``agent_pool.json`` override precedent — and this module merges that file
over ``providers.openai_compat.catalogue.LANES`` at load time.
 
An overlay row is a superset of a :class:`Lane`: the Lane fields the
shipped catalogue already validates, plus the four things plan 216 adds
that a Lane has no home for — the cross-pool capability tier, the
good-at / bad-at notes (P3 screen two), the recorded budget (P4), and the
cached model list with the timestamp of the fetch that produced it.
 
Responsibilities
----------------
- :class:`OverlayRow` — the row shape, Lane projection plus P3/P4 fields.
- :func:`load_overlay` / :func:`write_overlay_row` — strict read, atomic write.
- :func:`merge_lanes` — join by ``name`` with explicit override semantics.
- :func:`register_overlay_lanes` — the startup hook.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (overlay loaded, row written, lanes registered)
    L3 — semantic details (row names, counts)
    L4 — deep tracing
 
Contracts
---------
- **The shipped catalogue is never opened for writing.** Nothing in this
  module holds a writable handle on ``providers/openai_compat/**``; W3
  pins that by hashing the file across every operation here.
- Overlay rows satisfy the SAME validator the shipped rows do
  (:func:`validate_catalogue`), so an overlay lane cannot express
  something a checked-in lane could not.
- Loudness (inherited from plan 215 P5): a malformed row, an unknown
  field, a duplicate name, or a shipped-name collision without
  ``"override": true`` aborts the WHOLE merge, naming the row. There is
  no partially-merged state and no silently-dropped lane.
- This deliberately differs from ``agent_pool.json``, which degrades to
  an empty dict on a parse error. A silently-ignored lane overlay would
  make a lane the operator just added vanish with no explanation, which
  is the failure mode the wizard exists to remove.
- The overlay carries env-var NAMES only. No key value, and no field
  that could hold one, is part of the schema.
- Importing the provider bundle is OPTIONAL and lazy: without it there is
  no catalogue to overlay, and every entry point here degrades to a no-op
  rather than raising. That is what keeps a deepseek/ollama-only install
  byte-identical (plan 215's additive property).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- providers.openai_compat.catalogue (lazy; read-only: Lane / LaneModel / validator)
- option_c_agent_contracts (lazy; ModelCaps)
- providers.openai_compat.provider (lazy; OpenAICompatProvider registration of the overlay lanes)
"""
from __future__ import annotations
 
import json
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence, TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
if TYPE_CHECKING:  # pragma: no cover — typing only
    from providers.openai_compat.catalogue import Lane
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
#: The untracked overlay file (P2). Override with ``OC_CHAT_LANES_OVERLAY``.
DEFAULT_OVERLAY_PATH: Path = Path(
    os.environ.get("OC_CHAT_LANES_OVERLAY", ".oc_chat/lanes_overlay.json")
)
 
#: Schema version of the overlay document. Bumped only by a migration.
OVERLAY_VERSION = 1
 
#: P3 screen two — the cross-pool capability tier. Operator judgment, and
#: deliberately NOT the same vocabulary as the per-provider role tiers.
CAPABILITY_TIERS: tuple[str, ...] = ("frontier", "strong", "fast")
 
#: Every field a row may carry. Anything else is a typo worth reporting.
_REQUIRED_FIELDS = frozenset({
    "name", "display_label", "default_base_url", "base_url_env_var",
    "api_key_env_var", "models", "tiers", "free", "verification", "notes",
    "capability_tier",
})
_OPTIONAL_FIELDS = frozenset({
    "good_at", "bad_at", "budget", "cached_models", "models_fetched_at",
    "override", "extra_headers",
})
KNOWN_FIELDS = _REQUIRED_FIELDS | _OPTIONAL_FIELDS
 
 
class OverlayError(ValueError):
    """An overlay row or document is invalid — the merge aborts, naming it.
 
    VALIDATION_ERROR (S6 L3): the overlay is operator-authored data, so
    the repair is to fix the row, never to drop it and carry on.
    """
 
 
@dataclass(frozen=True)
class OverlayRow:
    """One operator-added lane: a :class:`Lane` plus the P3/P4 fields."""
 
    lane: "Lane"
    capability_tier: str
    good_at: str = ""
    bad_at: str = ""
    budget: str | None = None
    cached_models: tuple[str, ...] = ()
    models_fetched_at: str | None = None
    override: bool = False
 
    @property
    def name(self) -> str:
        """The lane name this row joins on."""
        return self.lane.name
 
 
def _catalogue():
    """Import the shipped catalogue module, or ``None`` when not installed."""
    try:
        import providers.openai_compat.catalogue as module
    except ImportError:
        return None
    return module
 
 
def _require(condition: bool, row: str, message: str) -> None:
    """Raise :exc:`OverlayError` naming *row* when *condition* is false."""
    if not condition:
        raise OverlayError(f"overlay row {row!r}: {message}")
 
 
def _model_caps():
    """The honest capability record for an operator-added model id.
 
    The same as-shipped position the shipped catalogue takes for an
    unmeasured model: tools attemptable, nothing claimed about
    thinking/reasoning, and the contract's conservative context default
    rather than a figure nobody measured. The wizard never asks the
    operator to guess a context window, so it never records a guess.
    """
    from option_c_agent_contracts import ModelCaps
 
    return ModelCaps(
        supports_thinking=False,
        supports_reasoning=False,
        supports_tools=True,
        max_context_tokens=65_536,
        tool_schema_dialect="openai_chat",
    )
 
 
def _str_list(value: Any, row: str, field: str) -> tuple[str, ...]:
    """Coerce *value* to a tuple of non-empty strings, or raise."""
    _require(isinstance(value, list), row, f"{field} must be a list of strings")
    out: list[str] = []
    for item in value:
        _require(
            isinstance(item, str) and item.strip(),
            row,
            f"{field} must hold non-empty strings, found {item!r}",
        )
        out.append(item.strip())
    return tuple(out)
 
 
def row_from_json(data: Mapping[str, Any]) -> OverlayRow:
    """Build an :class:`OverlayRow` from one overlay JSON row.
 
    Structural validation only — shape, types, and vocabulary. The lane's
    own semantic rules (tier map completeness, https base URL, env-var
    name spelling) are left to :func:`validate_catalogue` during the
    merge, so overlay rows and shipped rows are judged by one validator.
    """
    catalogue = _catalogue()
    if catalogue is None:
        raise OverlayError(
            "the openai_compat provider bundle is not installed, so there is "
            "no lane catalogue for the overlay to extend"
        )
 
    _require(isinstance(data, Mapping), "<unnamed>", "must be a JSON object")
    name = data.get("name")
    row = name if isinstance(name, str) and name.strip() else "<unnamed>"
 
    unknown = sorted(set(data) - KNOWN_FIELDS)
    _require(not unknown, row, f"unknown field(s): {', '.join(unknown)}")
    missing = sorted(_REQUIRED_FIELDS - set(data))
    _require(not missing, row, f"missing required field(s): {', '.join(missing)}")
 
    for field in ("name", "display_label", "default_base_url",
                  "base_url_env_var", "api_key_env_var", "notes", "verification"):
        _require(
            isinstance(data[field], str),
            row,
            f"{field} must be a string, found {type(data[field]).__name__}",
        )
    _require(
        isinstance(data["free"], bool),
        row,
        f"free must be a bool, found {type(data['free']).__name__}",
    )
    _require(isinstance(data["tiers"], Mapping), row, "tiers must be a JSON object")
    for tier_name, model_id in data["tiers"].items():
        _require(
            isinstance(tier_name, str) and isinstance(model_id, str),
            row,
            "tiers must map tier names to model ids",
        )
    _require(
        data["capability_tier"] in CAPABILITY_TIERS,
        row,
        f"capability_tier must be one of {list(CAPABILITY_TIERS)}, "
        f"found {data['capability_tier']!r}",
    )
 
    models = _str_list(data["models"], row, "models")
    cached = _str_list(data.get("cached_models", []), row, "cached_models")
 
    budget = data.get("budget")
    _require(
        budget is None or isinstance(budget, str),
        row,
        "budget must be a string or null",
    )
    fetched = data.get("models_fetched_at")
    _require(
        fetched is None or isinstance(fetched, str),
        row,
        "models_fetched_at must be a string or null",
    )
    override = data.get("override", False)
    _require(isinstance(override, bool), row, "override must be a bool")
 
    headers = data.get("extra_headers", {})
    _require(isinstance(headers, Mapping), row, "extra_headers must be a JSON object")
 
    caps = _model_caps()
    lane = catalogue.Lane(
        name=data["name"],
        display_label=data["display_label"],
        default_base_url=data["default_base_url"],
        base_url_env_var=data["base_url_env_var"],
        api_key_env_var=data["api_key_env_var"],
        models=tuple(catalogue.LaneModel(model_id, caps) for model_id in models),
        tiers=dict(data["tiers"]),
        free=data["free"],
        verification=data["verification"],  # validated by validate_catalogue
        notes=data["notes"],
        extra_headers={str(k): str(v) for k, v in headers.items()},
    )
    return OverlayRow(
        lane=lane,
        capability_tier=data["capability_tier"],
        good_at=str(data.get("good_at", "")),
        bad_at=str(data.get("bad_at", "")),
        budget=budget,
        cached_models=cached,
        models_fetched_at=fetched,
        override=override,
    )
 
 
def row_to_json(row: OverlayRow) -> dict[str, Any]:
    """Serialise one :class:`OverlayRow` back to its JSON shape."""
    lane = row.lane
    data: dict[str, Any] = {
        "name": lane.name,
        "display_label": lane.display_label,
        "default_base_url": lane.default_base_url,
        "base_url_env_var": lane.base_url_env_var,
        "api_key_env_var": lane.api_key_env_var,
        "models": [model.model_id for model in lane.models],
        "tiers": dict(lane.tiers),
        "free": lane.free,
        "verification": lane.verification,
        "notes": lane.notes,
        "capability_tier": row.capability_tier,
        "good_at": row.good_at,
        "bad_at": row.bad_at,
        "budget": row.budget,
        "cached_models": list(row.cached_models),
        "models_fetched_at": row.models_fetched_at,
        "override": row.override,
    }
    if lane.extra_headers:
        data["extra_headers"] = dict(lane.extra_headers)
    return data
 
 
def overlay_rows_to_json(rows: Iterable[OverlayRow]) -> dict[str, Any]:
    """Serialise *rows* as a whole overlay document."""
    return {"version": OVERLAY_VERSION, "lanes": [row_to_json(row) for row in rows]}
 
 
def load_overlay(path: str | Path = DEFAULT_OVERLAY_PATH) -> tuple[OverlayRow, ...]:
    """Read the overlay document at *path*; ``()`` when the file is absent.
 
    Raises :exc:`OverlayError` on a corrupt or wrongly-shaped document —
    never silently treated as "no overlay" (see the module contract).
    """
    func = "load_overlay"
    source = Path(path)
    if not source.is_file():
        return ()
    try:
        data = json.loads(source.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise OverlayError(
            f"could not parse the lane overlay at {source}: "
            f"{type(exc).__name__}: {exc}"
        ) from exc
    if not isinstance(data, Mapping):
        raise OverlayError(f"lane overlay at {source} must contain a JSON object")
    if "lanes" not in data:
        raise OverlayError(f"lane overlay at {source} has no 'lanes' list")
    lanes = data["lanes"]
    if not isinstance(lanes, list):
        raise OverlayError(
            f"lane overlay at {source}: 'lanes' must be a list, "
            f"found {type(lanes).__name__}"
        )
    rows = tuple(row_from_json(entry) for entry in lanes)
    _dbg("OC_CHAT.MAIN", func,
         f"loaded {len(rows)} overlay row(s) from {source}: "
         f"{', '.join(row.name for row in rows) or '(none)'}", 3)
    return rows
 
 
def write_overlay_row(
    row: OverlayRow,
    *,
    path: str | Path = DEFAULT_OVERLAY_PATH,
) -> Path:
    """Add or replace *row* in the overlay at *path*; atomic, never partial.
 
    Re-running the wizard for an existing lane UPDATES that row rather
    than stacking a duplicate the merge would then reject.
    """
    func = "write_overlay_row"
    target = Path(path)
    existing = list(load_overlay(target)) if target.is_file() else []
    merged = [item for item in existing if item.name != row.name]
    merged.append(row)
 
    target.parent.mkdir(parents=True, exist_ok=True)
    text = json.dumps(overlay_rows_to_json(merged), indent=2, ensure_ascii=False) + "\n"
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        newline="\n",
        dir=target.parent,
        prefix=".lanes_overlay_",
        suffix=".tmp",
        delete=False,
    ) as tmp:
        tmp.write(text)
        tmp_path = Path(tmp.name)
    os.replace(tmp_path, target)
    _dbg("OC_CHAT.MAIN", func,
         f"wrote overlay row {row.name!r} to {target} "
         f"({len(merged)} row(s) held)", 2)
    return target
 
 
def merge_lanes(
    shipped: Sequence["Lane"],
    rows: Sequence[OverlayRow],
) -> tuple["Lane", ...]:
    """Merge overlay *rows* over *shipped*, then validate the whole set.
 
    Join is by lane ``name``. A shipped-name collision is an ERROR unless
    the row carries ``"override": true``, in which case the shipped row is
    replaced IN PLACE so the menu does not reorder. New lanes append in
    overlay order. The merged set is then put through the shipped
    validator, so one bad row aborts everything (plan 215 P5).
    """
    func = "merge_lanes"
    if not rows:
        return tuple(shipped)
 
    catalogue = _catalogue()
    if catalogue is None:
        raise OverlayError(
            "the openai_compat provider bundle is not installed, so there is "
            "no lane catalogue for the overlay to extend"
        )
 
    seen: set[str] = set()
    for row in rows:
        _require(row.name not in seen, row.name, "appears twice in the overlay")
        seen.add(row.name)
 
    shipped_names = {lane.name for lane in shipped}
    for row in rows:
        if row.name in shipped_names and not row.override:
            raise OverlayError(
                f"overlay row {row.name!r}: collides with a shipped catalogue "
                f"lane of the same name. Rename the lane, or set "
                f'"override": true to shadow the shipped row deliberately.'
            )
 
    by_name = {row.name: row for row in rows}
    merged: list["Lane"] = []
    for lane in shipped:
        row = by_name.pop(lane.name, None)
        merged.append(row.lane if row is not None else lane)
    merged.extend(row.lane for row in rows if row.name in by_name)
 
    validated = catalogue.validate_catalogue(merged)
    _dbg("OC_CHAT.MAIN", func,
         f"merged {len(rows)} overlay row(s) over {len(shipped)} shipped "
         f"lane(s) -> {len(validated)} lane(s)", 2)
    return validated
 
 
def register_overlay_lanes(
    path: str | Path = DEFAULT_OVERLAY_PATH,
) -> tuple[str, ...]:
    """Register every overlay lane into the provider registry (startup hook).
 
    Returns the lane NAMES registered. Total and quiet by design: with no
    bundle, no overlay file, or no rows, it registers nothing and returns
    ``()``. A malformed overlay is reported through :func:`_dbg` and
    surfaced by the caller — it must not stop OC Chat from starting, but
    it must not be silently indistinguishable from "no overlay" either.
    """
    func = "register_overlay_lanes"
    catalogue = _catalogue()
    if catalogue is None:
        return ()
    try:
        rows = load_overlay(path)
    except OverlayError as exc:
        # VALIDATION_ERROR (S6 L3) — operator-authored data; report, don't crash.
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {exc}", 1, override=True)
        raise
    if not rows:
        return ()
 
    from providers.openai_compat.provider import OpenAICompatProvider
 
    merged = merge_lanes(catalogue.LANES, rows)
    overlay_names = {row.name for row in rows}
    registered: list[str] = []
    for lane in merged:
        if lane.name not in overlay_names:
            continue
        OpenAICompatProvider.register_lane(lane)
        registered.append(lane.name)
    _dbg("OC_CHAT.MAIN", func,
         f"registered {len(registered)} overlay lane(s): "
         f"{', '.join(registered) or '(none)'}", 2)
    return tuple(registered)
 
 
__all__ = [
    "CAPABILITY_TIERS",
    "DEFAULT_OVERLAY_PATH",
    "KNOWN_FIELDS",
    "OVERLAY_VERSION",
    "OverlayError",
    "OverlayRow",
    "load_overlay",
    "merge_lanes",
    "overlay_rows_to_json",
    "register_overlay_lanes",
    "row_from_json",
    "row_to_json",
    "write_overlay_row",
]