provider_wizard.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/provider_wizard.py
 
"""
The "Add provider" wizard — onboard a lane from the opening menu, safely.
 
Purpose
-------
Plan 216 T3/T4 and the flow pinned in P5: the operator picks "Add
provider" in the opening menu, optionally bases the new lane on a known
one, enters a key ONCE with masked input, settles the model list (cached,
fetched live, or typed), answers the two tier questions SEPARATELY (P3),
records good-at / bad-at notes and an optional budget (P4), confirms, and
the lane appears in the menu immediately.
 
Two properties matter more than the flow itself:
 
- **Cancel safety.** Every prompt can be cancelled, and a cancelled run
  writes NOTHING. This is structural rather than careful: the run
  accumulates answers in memory and commits both files only after the
  final confirmation, so there is no interior state in which a partial
  sidecar or a partial overlay can exist (W5).
- **No key material anywhere but the sidecar.** The captured value is
  passed to :func:`write_secret` and to the live fetch, and nowhere else.
  Third-party error text is REDACTED against it before display, because
  an upstream 401 body is entirely capable of quoting the key back.
 
Responsibilities
----------------
- :func:`run_add_provider_wizard` — the whole flow; returns the written
  :class:`OverlayRow`, or ``None`` when cancelled or refused.
- :func:`fetch_models_for_lane` — T4: cache-first, live only on the
  explicit action.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (run started, cancelled, committed)
    L3 — semantic details (step names, lane name, model counts)
    L4 — deep tracing
 
Contracts
---------
- **Validation happens ONCE, at the end**, against the same validator the
  shipped catalogue uses. The wizard never re-prompts on a bad answer, so
  a rejected run is indistinguishable from a cancelled one where it
  matters: nothing is written either way. The rejection names the row and
  the reason.
- Typing ``q`` at any prompt cancels, as do Ctrl+C and EOF. A lane
  literally named ``q`` is therefore unreachable through the wizard; it
  remains reachable by hand-editing the overlay.
- The ONLY network call this module can make is
  :func:`fetch_models_for_lane` with ``force=True``, which happens only
  on the operator's explicit "fetch" answer (P6). Nothing else here opens
  a socket, and the test suite never sets ``force=True`` against a real
  client.
- ``verification`` is always written as ``"chat"``. The wizard has no way
  to evidence tool support, and 215 P4 reserves promotion for a human who
  ran the conformance runner against a live key.
- Commit order is sidecar THEN overlay: the sidecar is the write that can
  legitimately refuse (the W1 canary), so it goes first. A failure after
  it leaves an unused environment variable, which is inert; the reverse
  order would leave a visible lane with no key behind it.
- Env-var names default to a derivation from the NEW lane name even when
  a template lane is copied, so two lanes never silently share one key
  variable.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.provider_secrets (masked capture + the gitignored sidecar)
- oc_chat.core.lane_overlay (the untracked overlay row + merge)
- providers.openai_compat.catalogue (lazy; the shipped Lane rows and the Lane type)
- providers.openai_compat.client (lazy; OpenAICompatClient for the live model fetch)
"""
from __future__ import annotations
 
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Sequence, TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
from oc_chat.core import provider_secrets
from oc_chat.core.lane_overlay import (
    CAPABILITY_TIERS,
    DEFAULT_OVERLAY_PATH,
    OverlayError,
    OverlayRow,
    load_overlay,
    merge_lanes,
    row_from_json,
    write_overlay_row,
)
 
if TYPE_CHECKING:  # pragma: no cover — typing only
    from providers.openai_compat.catalogue import Lane
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
#: Typing this at any prompt cancels the run (see the module contract).
CANCEL_ANSWER = "q"
 
#: The four role tiers, in the order the wizard asks for them (P3 screen one).
ROLE_TIERS: tuple[str, ...] = ("default", "cheap", "reasoning", "summary")
 
_NON_NAME_RE = re.compile(r"[^A-Z0-9]+")
 
 
def _now_iso() -> str:
    """Current UTC time as an ISO-8601 string."""
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
 
 
def _redact(text: str, secret: str | None) -> str:
    """Replace *secret* wherever it appears in *text*.
 
    Upstream error bodies quote request material more often than anyone
    would like. Since the key is known here, it can be scrubbed rather
    than hoped about.
    """
    if not secret or not text:
        return text
    return text.replace(secret, "***")
 
 
def _derive_env_var(lane_name: str, suffix: str) -> str:
    """Derive an UPPER_SNAKE env-var name from *lane_name*."""
    stem = _NON_NAME_RE.sub("_", lane_name.upper()).strip("_")
    return f"{stem}_{suffix}" if stem else suffix
 
 
def _ask(
    driver: Any,
    presenter: Any,
    prompt_text: str,
    *,
    default: str = "",
) -> str | None:
    """Ask one question. Return the answer, or ``None`` when cancelled."""
    try:
        if driver is not None:
            raw = driver.read_user_input(None, prompt_text=prompt_text)
        else:
            raw = input(prompt_text)
    except (EOFError, KeyboardInterrupt):
        return None
    if raw is None:
        return None
    answer = raw.strip()
    if answer.lower() == CANCEL_ANSWER:
        return None
    return answer or default
 
 
def fetch_models_for_lane(
    lane: "Lane",
    api_key: str,
    *,
    cached: Sequence[str] = (),
    force: bool = False,
    client_factory: Callable[[Any, str], Any] | None = None,
) -> tuple[str, ...]:
    """Return the model ids for *lane* — cache first, live only on demand.
 
    Without ``force`` this returns *cached* verbatim and opens no
    connection, including when the cache is EMPTY: an empty cache is not
    an implicit licence to reach the network (P6).
 
    The live path calls ``client.list_models()``, which is exactly what
    ``OpenAICompatProvider.fetch_live_models`` delegates to. The provider
    is not constructed here because it requires a validated tier map, and
    during onboarding the tier map is the thing this call informs.
    """
    func = "fetch_models_for_lane"
    if not force:
        _dbg("OC_CHAT.MAIN", func,
             f"lane={lane.name} served {len(cached)} model(s) from cache", 3)
        return tuple(cached)
 
    factory = client_factory if client_factory is not None else _default_client_factory
    client = factory(lane, api_key)
    models = tuple(str(model) for model in client.list_models())
    _dbg("OC_CHAT.MAIN", func,
         f"lane={lane.name} fetched {len(models)} model(s) live", 2)
    return models
 
 
def _default_client_factory(lane: "Lane", api_key: str) -> Any:
    """Build the real OpenAI-compatible client for *lane*."""
    from providers.openai_compat.client import OpenAICompatClient
 
    return OpenAICompatClient(
        api_key=api_key,
        base_url=lane.default_base_url,
        extra_headers=dict(lane.extra_headers),
        lane_name=lane.name,
    )
 
 
def _known_lanes(shipped: Sequence["Lane"], overlay_path: Path) -> list["Lane"]:
    """Shipped rows followed by overlay rows — the "base this on" list."""
    try:
        overlay = load_overlay(overlay_path)
    except OverlayError:
        overlay = ()
    return list(shipped) + [row.lane for row in overlay]
 
 
def _provisional_lane(
    catalogue: Any,
    *,
    name: str,
    label: str,
    base_url: str,
    key_var: str,
    url_var: str,
    headers: dict[str, str],
) -> "Lane":
    """A Lane carrying only what the fetch step needs (no models yet).
 
    :class:`Lane` is a plain frozen dataclass: construction does not
    validate, which is what makes this legal. The row is never returned to
    a caller and never registered — it exists to carry the base URL and
    headers into the client factory.
    """
    return catalogue.Lane(
        name=name,
        display_label=label,
        default_base_url=base_url,
        base_url_env_var=url_var,
        api_key_env_var=key_var,
        models=(),
        tiers={},
        free=False,
        verification="chat",
        notes="(provisional - not registered)",
        extra_headers=headers,
    )
 
 
def _compose_notes(
    *,
    capability_tier: str,
    good_at: str,
    bad_at: str,
    model_source: str,
    stamp: str,
) -> str:
    """Build the row's provenance ``notes``, which must never be empty.
 
    The shipped catalogue requires provenance on every row, and the
    wizard is a provenance event: it records WHEN the lane was added, HOW
    the model list was obtained, and the operator's own judgment, then
    states plainly that nothing here is verified beyond chat.
    """
    return (
        f"Added by the OC Chat provider wizard on {stamp}. "
        f"Capability tier (operator judgment): {capability_tier}. "
        f"Good at: {good_at or 'not recorded'}. "
        f"Bad at: {bad_at or 'not recorded'}. "
        f"Model list: {model_source}. "
        f"Context windows are the contract default (unmeasured). "
        f"Verified: chat only - run the conformance runner against a live key "
        f"before claiming tool support."
    )
 
 
def run_add_provider_wizard(
    presenter: Any,
    *,
    prompt_driver: Any = None,
    key_reader: Callable[[str], str] | None = None,
    overlay_path: str | Path = DEFAULT_OVERLAY_PATH,
    secrets_path: str | Path | None = None,
    shipped_lanes: Sequence["Lane"] | None = None,
    client_factory: Callable[[Any, str], Any] | None = None,
    now: Callable[[], str] | None = None,
) -> OverlayRow | None:
    """Run the "Add provider" flow. Return the written row, or ``None``.
 
    ``None`` means "nothing was written" in every case — cancelled at a
    prompt, cancelled at the key, declined at the confirmation, a failed
    live fetch, or a row the validator refused.
    """
    func = "run_add_provider_wizard"
    try:
        import providers.openai_compat.catalogue as catalogue
    except ImportError:
        presenter.show_error(
            "  Add provider needs the openai_compat provider bundle, which is "
            "not installed."
        )
        return None
 
    overlay_target = Path(overlay_path)
    secrets_target = Path(
        secrets_path if secrets_path is not None
        else provider_secrets.DEFAULT_SECRETS_PATH
    )
    stamp = (now or _now_iso)()
    shipped = tuple(shipped_lanes) if shipped_lanes is not None else catalogue.LANES
 
    presenter.show_info("")
    presenter.show_info("  Add a provider lane")
    presenter.show_info("  " + "-" * 48)
    presenter.show_info(f"  Type '{CANCEL_ANSWER}' at any prompt to cancel. Nothing is")
    presenter.show_info("  written until the final confirmation.")
 
    # ── Step 1: base the new lane on a known one, or start blank ─────────
    known = _known_lanes(shipped, overlay_target)
    presenter.show_info("")
    presenter.show_info("  Base the new lane on:")
    for index, lane in enumerate(known, 1):
        presenter.show_info(f"    {index:>2}. {lane.name:<20} {lane.display_label}")
    presenter.show_info("     Enter. (blank template)")
    answer = _ask(prompt_driver, presenter, "  Choose a template [blank]: ")
    if answer is None:
        return _cancelled(presenter, "template")
    template: "Lane | None" = None
    if answer:
        if not answer.isdigit() or not (1 <= int(answer) <= len(known)):
            presenter.show_error(f"  Not a template choice: {answer!r}. Nothing written.")
            return None
        template = known[int(answer) - 1]
        presenter.show_info(f"  Basing on {template.name}.")
 
    # ── Step 2-4: identity ───────────────────────────────────────────────
    name = _ask(prompt_driver, presenter, "  Lane name (lowercase identifier): ")
    if name is None:
        return _cancelled(presenter, "lane name")
 
    label_default = template.display_label if template else name
    label = _ask(
        prompt_driver, presenter,
        f"  Display label [{label_default}]: ", default=label_default,
    )
    if label is None:
        return _cancelled(presenter, "display label")
 
    url_default = template.default_base_url if template else ""
    base_url = _ask(
        prompt_driver, presenter,
        f"  API base URL (https://...){f' [{url_default}]' if url_default else ''}: ",
        default=url_default,
    )
    if base_url is None:
        return _cancelled(presenter, "base URL")
 
    # ── Step 5-6: env-var NAMES (never values) ───────────────────────────
    key_var_default = _derive_env_var(name, "API_KEY")
    key_var = _ask(
        prompt_driver, presenter,
        f"  API key variable name [{key_var_default}]: ", default=key_var_default,
    )
    if key_var is None:
        return _cancelled(presenter, "key variable")
 
    url_var_default = _derive_env_var(name, "BASE_URL")
    url_var = _ask(
        prompt_driver, presenter,
        f"  Base URL variable name [{url_var_default}]: ", default=url_var_default,
    )
    if url_var is None:
        return _cancelled(presenter, "base URL variable")
 
    # ── The key: captured once, masked, held in memory only ──────────────
    api_key = provider_secrets.capture_key(key_var, reader=key_reader)
    if api_key is None:
        return _cancelled(presenter, "API key")
 
    # ── Step 7: settle the model list ────────────────────────────────────
    headers = dict(template.extra_headers) if template else {}
    cached = tuple(model.model_id for model in template.models) if template else ()
    presenter.show_info("")
    presenter.show_info("  Model list:")
    presenter.show_info(
        f"    [c] use the cached/template list ({len(cached)} model(s))"
        if cached else "    [c] use the cached list (empty)"
    )
    presenter.show_info("    [f] fetch live from the API now (the only network call)")
    presenter.show_info("    [m] type the ids by hand")
    action = _ask(prompt_driver, presenter, "  Choice (c/f/m) [c]: ", default="c")
    if action is None:
        return _cancelled(presenter, "model list")
 
    model_source = "copied from the template" if cached else "entered by hand"
    fetched_at: str | None = None
    candidates = cached
    if action.lower() == "f":
        provisional = _provisional_lane(
            catalogue, name=name, label=label, base_url=base_url,
            key_var=key_var, url_var=url_var, headers=headers,
        )
        try:
            candidates = fetch_models_for_lane(
                provisional, api_key, cached=cached, force=True,
                client_factory=client_factory,
            )
        except Exception as exc:  # noqa: BLE001 — any client failure aborts the run
            # EXTERNAL_ERROR (S6 L2). The text is redacted before display:
            # an upstream auth error is quite capable of quoting the key.
            detail = _redact(f"{type(exc).__name__}: {exc}", api_key)
            _dbg("OC_CHAT.MAIN", func,
                 f"EXTERNAL_ERROR: live fetch for {name} failed: "
                 f"{type(exc).__name__}", 2, override=True)
            presenter.show_error(f"  Could not fetch models: {detail}")
            presenter.show_error("  Nothing written.")
            return None
        fetched_at = stamp
        model_source = f"fetched live at {stamp}"
        presenter.show_info(f"  Fetched {len(candidates)} model(s).")
    elif action.lower() == "m":
        candidates = ()
        model_source = "entered by hand"
 
    shown = ", ".join(candidates)
    models_answer = _ask(
        prompt_driver, presenter,
        f"  Model ids, comma separated{f' [{shown}]' if shown else ''}: ",
        default=shown,
    )
    if models_answer is None:
        return _cancelled(presenter, "model ids")
    models = tuple(
        part.strip() for part in models_answer.split(",") if part.strip()
    )
    if not models:
        presenter.show_error("  No model ids given. Nothing written.")
        return None
 
    # ── Step 8-11: P3 screen one, the per-provider ROLE tiers ────────────
    presenter.show_info("")
    presenter.show_info("  Role tiers - which model answers each slash-command tier:")
    for index, model_id in enumerate(models, 1):
        presenter.show_info(f"    {index:>2}. {model_id}")
    tiers: dict[str, str] = {}
    for role in ROLE_TIERS:
        answer = _ask(
            prompt_driver, presenter,
            f"  Role tier '{role}' (number or id) [1={models[0]}]: ",
        )
        if answer is None:
            return _cancelled(presenter, f"role tier '{role}'")
        if not answer:
            tiers[role] = models[0]
        elif answer.isdigit() and 1 <= int(answer) <= len(models):
            tiers[role] = models[int(answer) - 1]
        else:
            tiers[role] = answer  # validated with the rest, at the end
 
    # ── Step 12-14: P3 screen two, the cross-pool CAPABILITY tier ────────
    presenter.show_info("")
    presenter.show_info(
        "  Capability tier - your judgment of this lane against the whole pool:"
    )
    for index, tier_name in enumerate(CAPABILITY_TIERS, 1):
        presenter.show_info(f"    {index:>2}. {tier_name}")
    answer = _ask(
        prompt_driver, presenter,
        f"  Capability tier {list(CAPABILITY_TIERS)}: ",
    )
    if answer is None:
        return _cancelled(presenter, "capability tier")
    capability_tier = answer
    if answer.isdigit() and 1 <= int(answer) <= len(CAPABILITY_TIERS):
        capability_tier = CAPABILITY_TIERS[int(answer) - 1]
 
    good_at = _ask(prompt_driver, presenter, "  Good at (free text, Enter to skip): ")
    if good_at is None:
        return _cancelled(presenter, "good-at notes")
    bad_at = _ask(prompt_driver, presenter, "  Bad at (free text, Enter to skip): ")
    if bad_at is None:
        return _cancelled(presenter, "bad-at notes")
 
    # ── Step 15-16: access tier and the recorded budget ──────────────────
    free_answer = _ask(
        prompt_driver, presenter,
        "  Is this lane FREE to use? (y/N): ", default="n",
    )
    if free_answer is None:
        return _cancelled(presenter, "free/paid")
    free = free_answer.lower().startswith("y")
 
    budget_answer = _ask(
        prompt_driver, presenter,
        "  Budget for this lane (recorded, not enforced; Enter to skip): ",
    )
    if budget_answer is None:
        return _cancelled(presenter, "budget")
    budget = budget_answer or None
 
    # ── Build the row and show it back before the point of no return ─────
    row_json = {
        "name": name,
        "display_label": label,
        "default_base_url": base_url,
        "base_url_env_var": url_var,
        "api_key_env_var": key_var,
        "models": list(models),
        "tiers": tiers,
        "free": free,
        # 215 P4: never promoted by a machine.
        "verification": "chat",
        "notes": _compose_notes(
            capability_tier=capability_tier, good_at=good_at, bad_at=bad_at,
            model_source=model_source, stamp=stamp,
        ),
        "capability_tier": capability_tier,
        "good_at": good_at,
        "bad_at": bad_at,
        "budget": budget,
        "cached_models": list(candidates if fetched_at else models),
        "models_fetched_at": fetched_at,
        "override": False,
    }
    if headers:
        row_json["extra_headers"] = headers
 
    presenter.show_info("")
    presenter.show_info(f"  Lane '{name}' ({label})")
    presenter.show_info(f"    base URL      {base_url}")
    presenter.show_info(f"    key variable  {key_var}  (value goes to the sidecar only)")
    presenter.show_info(f"    models        {', '.join(models)}")
    presenter.show_info(f"    role tiers    {tiers}")
    presenter.show_info(f"    capability    {capability_tier}")
    presenter.show_info(f"    access        {'FREE' if free else 'PAID'}")
    presenter.show_info(f"    budget        {budget or '(none recorded)'}")
 
    confirm = _ask(
        prompt_driver, presenter, "  Write this lane? (y/N): ", default="n",
    )
    if confirm is None or not confirm.lower().startswith("y"):
        return _cancelled(presenter, "confirmation")
 
    # ── Validate ONCE, against the shipped validator, before any write ───
    try:
        row = row_from_json(row_json)
        merge_lanes(shipped, list(load_overlay(overlay_target)) + [row])
    except (OverlayError, catalogue.CatalogueError) as exc:
        # VALIDATION_ERROR (S6 L3) — the answers do not make a legal lane.
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {exc}", 2, override=True)
        presenter.show_error(f"  Refused: {exc}")
        presenter.show_error("  Nothing written.")
        return None
 
    # ── Commit: sidecar first (it is the write that can refuse) ──────────
    try:
        provider_secrets.write_secret(key_var, api_key, path=secrets_target)
    except (provider_secrets.SecretsPathError, ValueError) as exc:
        # SECURITY_ERROR (S6 L1) on the canary path — never routed around.
        _dbg("OC_CHAT.MAIN", func,
             f"SECURITY_ERROR: sidecar write refused for {key_var}", 1, override=True)
        presenter.show_error(f"  Refused: {_redact(str(exc), api_key)}")
        presenter.show_error("  Nothing written.")
        return None
 
    write_overlay_row(row, path=overlay_target)
    provider_secrets.apply_to_environ(path=secrets_target)
 
    _dbg("OC_CHAT.MAIN", func,
         f"added lane {name!r} with {len(models)} model(s); key stored under "
         f"{key_var}", 2)
    presenter.show_info("")
    presenter.show_info(f"  Added lane '{name}'. Select it with OC_CHAT_PROVIDER={name}.")
    presenter.show_info(f"  Key stored in {secrets_target} (gitignored).")
    return row
 
 
def _cancelled(presenter: Any, step: str) -> None:
    """Report a cancel at *step* and return ``None``. Nothing was written."""
    func = "_cancelled"
    _dbg("OC_CHAT.MAIN", func, f"run cancelled at the {step} step", 2)
    presenter.show_info(f"  Cancelled at the {step} step. Nothing written.")
    return None
 
 
__all__ = [
    "CANCEL_ANSWER",
    "ROLE_TIERS",
    "fetch_models_for_lane",
    "run_add_provider_wizard",
]