conformance.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/openai_compat/conformance.py
 
"""
Lane conformance runner — the evidence behind a ``verification`` promotion.
 
Purpose
-------
``python -m providers.openai_compat.conformance <lane> --live`` exercises
one streamed chat round and one single-tool ReAct round against a lane's
real endpoint and prints a verdict. A lane's catalogue ``verification``
field is promoted from ``"chat"`` to ``"tools"`` only by a HUMAN who ran
this and recorded the date in that row's ``notes`` (plan 215 P4) — the
runner never edits the catalogue.
 
Live network access requires the explicit ``--live`` flag. Without it the
runner refuses to touch the network and exits non-zero, which is what
makes it safe to exercise from the test suite: the suite drives it with an
injected mock client and never opens a socket.
 
Responsibilities
----------------
- :func:`run_conformance` — the two rounds against an injected client.
- :func:`build_live_client` — construct a real client for a lane, or fail
  with the env-var NAME the operator must set.
- :func:`format_report` — render a :class:`ConformanceResult` as ASCII.
- :func:`main` — argv parsing, the ``--live`` gate, and the exit code.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Without ``--live`` the runner NEVER constructs a network client and
  exits 2. This is a refusal, not a silent skip.
- ``--live`` with an unset key env var exits 2 naming the variable; it
  never falls back to another lane's key.
- Exit codes: 0 both rounds passed; 1 a round failed; 2 refused to run
  (no ``--live``, unknown lane, or missing key).
- The runner reports ONLY what it observed. A tools round that produced
  no tool call is reported as a FAIL with the reason, never softened into
  a pass, and the runner never writes the catalogue.
- No key value is printed, logged, or included in any report line.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- providers.openai_compat.catalogue (LANES, Lane and lane_by_name to find the lane under test)
- providers.openai_compat.client (lazy; OpenAICompatClient built for a live run)
"""
from __future__ import annotations
 
import argparse
import json
import os
import sys
from dataclasses import dataclass, field
from typing import Any, Sequence
 
from oc_diagnostics import _dbg
 
from .catalogue import LANES, Lane, lane_by_name
 
#: The one tool the ReAct round offers. Deliberately trivial and pure: the
#: round tests whether the LANE emits a well-formed tool call, not whether a
#: model can solve anything.
PROBE_TOOL: dict = {
    "type": "function",
    "function": {
        "name": "get_current_temperature",
        "description": "Return the current temperature for a named city.",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "City name, e.g. 'Rotterdam'"},
            },
            "required": ["city"],
        },
    },
}
 
_CHAT_PROMPT = "Reply with exactly the word: pong"
_TOOLS_PROMPT = "What is the current temperature in Rotterdam? Use the tool."
 
 
@dataclass
class RoundResult:
    """Outcome of one conformance round."""
 
    name: str
    passed: bool
    detail: str
    chunks: list[str] = field(default_factory=list)
 
 
@dataclass
class ConformanceResult:
    """Outcome of a full lane conformance run."""
 
    lane_name: str
    rounds: list[RoundResult] = field(default_factory=list)
 
    @property
    def passed(self) -> bool:
        """True only when every round passed."""
        return bool(self.rounds) and all(r.passed for r in self.rounds)
 
    @property
    def verdict(self) -> str:
        """``"tools"`` when both rounds passed, else ``"chat"`` or ``"none"``."""
        by_name = {r.name: r.passed for r in self.rounds}
        if by_name.get("chat") and by_name.get("tools"):
            return "tools"
        if by_name.get("chat"):
            return "chat"
        return "none"
 
 
def _chat_round(client: Any, model: str) -> RoundResult:
    """One streamed chat round: assert content arrived."""
    func = "_chat_round"
    messages = [{"role": "user", "content": _CHAT_PROMPT}]
    text_parts: list[str] = []
    kinds: list[str] = []
    try:
        for kind, data in client.stream_chat(messages, model):
            kinds.append(kind)
            if kind == "content":
                text_parts.append(str(data))
    except Exception as exc:
        # EXTERNAL_ERROR — the lane's endpoint failed the round.
        _dbg("OC_CHAT.CLIENT", func,
             f"EXTERNAL_ERROR: {type(exc).__name__}: {exc}", 2)
        return RoundResult("chat", False, f"{type(exc).__name__}: {exc}", kinds)
 
    text = "".join(text_parts).strip()
    if not text:
        return RoundResult("chat", False, "stream yielded no content chunk", kinds)
    preview = text if len(text) <= 60 else text[:57] + "..."
    return RoundResult("chat", True, f"content received: {preview!r}", kinds)
 
 
def _tools_round(client: Any, model: str) -> RoundResult:
    """One single-tool ReAct round: assert a well-formed tool call arrived."""
    func = "_tools_round"
    messages = [{"role": "user", "content": _TOOLS_PROMPT}]
    calls: list[dict] = []
    kinds: list[str] = []
    try:
        for kind, data in client.stream_chat(messages, model, tools=[PROBE_TOOL]):
            kinds.append(kind)
            if kind == "tool_call" and isinstance(data, dict):
                calls.append(data)
    except Exception as exc:
        # EXTERNAL_ERROR — the lane's endpoint failed the round.
        _dbg("OC_CHAT.CLIENT", func,
             f"EXTERNAL_ERROR: {type(exc).__name__}: {exc}", 2)
        return RoundResult("tools", False, f"{type(exc).__name__}: {exc}", kinds)
 
    if not calls:
        return RoundResult(
            "tools", False,
            "no tool_call chunk (the lane answered in prose or ignored the tool)",
            kinds,
        )
    call = calls[0]
    if call.get("name") != PROBE_TOOL["function"]["name"]:
        return RoundResult(
            "tools", False,
            f"tool call named {call.get('name')!r}, expected "
            f"{PROBE_TOOL['function']['name']!r}",
            kinds,
        )
    try:
        args = json.loads(call.get("arguments") or "{}")
    except ValueError as exc:
        # VALIDATION_ERROR — arguments were not parseable JSON.
        _dbg("OC_CHAT.CLIENT", func,
             f"VALIDATION_ERROR: {type(exc).__name__}: {exc}", 3)
        return RoundResult(
            "tools", False,
            f"tool call arguments are not valid JSON: {exc}",
            kinds,
        )
    if not isinstance(args, dict) or "city" not in args:
        return RoundResult(
            "tools", False,
            f"tool call arguments missing the required 'city' key: {args!r}",
            kinds,
        )
    return RoundResult("tools", True, f"tool call well-formed: {args!r}", kinds)
 
 
def run_conformance(client: Any, lane: Lane, model: str | None = None) -> ConformanceResult:
    """Run both rounds for *lane* against an already-built *client*.
 
    The client is INJECTED so the test suite can drive this with a mock and
    stay off the network entirely (plan 215 W5).
    """
    func = "run_conformance"
    target = model or lane.default_model
    _dbg("OC_CHAT.CLIENT", func, f"lane={lane.name} model={target}", 2)
    result = ConformanceResult(lane_name=lane.name)
    result.rounds.append(_chat_round(client, target))
    result.rounds.append(_tools_round(client, target))
    return result
 
 
def build_live_client(lane: Lane) -> Any:
    """Construct a real client for *lane*, or raise :exc:`SystemExit(2)`.
 
    Reads the key from the env-var NAME the catalogue row advertises. The
    key value is never logged, echoed, or stored anywhere in this module.
    """
    func = "build_live_client"
    from .client import OpenAICompatClient
 
    api_key = os.environ.get(lane.api_key_env_var, "").strip()
    if not api_key:
        print(
            f"ERROR: {lane.api_key_env_var} is not set; cannot run lane "
            f"{lane.name!r} live.",
            file=sys.stderr,
        )
        raise SystemExit(2)
    base_url = os.environ.get(lane.base_url_env_var, "").strip() or lane.default_base_url
    _dbg("OC_CHAT.CLIENT", func, f"lane={lane.name} base_url={base_url}", 2)
    return OpenAICompatClient(
        api_key=api_key,
        base_url=base_url,
        extra_headers=lane.extra_headers,
        lane_name=lane.name,
    )
 
 
def format_report(result: ConformanceResult, lane: Lane, model: str) -> str:
    """Render *result* as an ASCII report block."""
    lines = [
        "=" * 66,
        f"  Lane conformance: {lane.name}  ({lane.display_label})",
        "=" * 66,
        f"  model      : {model}",
        f"  base URL   : {lane.default_base_url}",
        f"  catalogue  : free={str(lane.free).lower()} verification={lane.verification}",
        "",
    ]
    for round_result in result.rounds:
        status = "PASS" if round_result.passed else "FAIL"
        lines.append(f"  [{status}] {round_result.name:<6} {round_result.detail}")
    lines.append("")
    lines.append(f"  observed verification tier: {result.verdict}")
    if result.verdict == "tools" and lane.verification != "tools":
        lines.append(
            "  NOTE: this run supports promoting the catalogue row to "
            "verification=\"tools\"."
        )
        lines.append(
            "        A human makes that edit and dates it in the row's notes; "
            "this runner never writes the catalogue."
        )
    lines.append("=" * 66)
    return "\n".join(lines)
 
 
def build_arg_parser() -> argparse.ArgumentParser:
    """Build the runner's argument parser."""
    parser = argparse.ArgumentParser(
        prog="python -m providers.openai_compat.conformance",
        description=(
            "Exercise one streamed chat round and one single-tool ReAct round "
            "against a lane's live endpoint. Requires --live; without it the "
            "runner refuses to touch the network."
        ),
    )
    parser.add_argument("lane", help=f"lane name ({', '.join(l.name for l in LANES)})")
    parser.add_argument(
        "--live",
        action="store_true",
        help="REQUIRED to make real API calls. Costs money on a paid lane.",
    )
    parser.add_argument(
        "--model",
        default=None,
        help="model id to probe (default: the lane's 'default' tier)",
    )
    return parser
 
 
def main(argv: Sequence[str] | None = None) -> int:
    """Entry point. Returns the process exit code (0 pass, 1 fail, 2 refused)."""
    func = "main"
    args = build_arg_parser().parse_args(argv)
 
    try:
        lane = lane_by_name(args.lane)
    except KeyError as exc:
        # VALIDATION_ERROR — unknown lane name supplied on the command line.
        _dbg("OC_CHAT.CLIENT", func, f"VALIDATION_ERROR: {exc}", 3)
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
 
    if not args.live:
        print(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Operator-facing refusal message; high-entropy English, not a credential"
            "ERROR: refusing to run without --live.\n"
            "  This runner makes REAL API calls against the lane's endpoint and "
            "costs money on a paid lane.\n"
            f"  Re-run with: python -m providers.openai_compat.conformance "
            f"{lane.name} --live",
            file=sys.stderr,
        )
        return 2
 
    client = build_live_client(lane)
    model = args.model or lane.default_model
    result = run_conformance(client, lane, model)
    print(format_report(result, lane, model))
    return 0 if result.passed else 1
 
 
if __name__ == "__main__":
    raise SystemExit(main())
 
 
__all__ = [
    "ConformanceResult",
    "PROBE_TOOL",
    "RoundResult",
    "build_arg_parser",
    "build_live_client",
    "format_report",
    "main",
    "run_conformance",
]