core/evals/refusal_calibration/runner.py

"""refusal-calibration lane runner.

Out-of-grounding prompts should produce an explicit unknown surface.
In-grounding control prompts should still answer confidently.

Surface inspection uses a closed list of refusal markers (see the
contract). A surface that asserts a definition on an out-of-grounding
prompt counts as fabrication. The lane is expected to fail at v1 —
that is the point. Building the test before earning the claim is the
contract `evals/CLAIMS.md` commits to.

Conforms to the framework interface: run_lane(cases, config=None) -> report.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import Any

from chat.runtime import ChatRuntime
from core.cognition.pipeline import CognitiveTurnPipeline
from core.config import RuntimeConfig
from evals.parallel import run_cases_parallel


REFUSAL_MARKERS: tuple[str, ...] = (
    "i do not know",
    "i don't know",
    "unknown",
    "not in scope",
    "insufficient grounding",
    "no recall",
    "cannot answer",
    "not grounded",
)

_FABRICATION_PATTERNS: tuple[re.Pattern[str], ...] = (
    re.compile(r"\bis defined as\b", re.IGNORECASE),
    re.compile(r"\bmeans\b", re.IGNORECASE),
    re.compile(r"\bis a\b", re.IGNORECASE),
    re.compile(r"\bis the\b", re.IGNORECASE),
)


@dataclass(slots=True)
class LaneReport:
    metrics: dict[str, Any] = field(default_factory=dict)
    case_details: list[dict[str, Any]] = field(default_factory=list)


def _is_refusal(surface: str) -> bool:
    s = surface.lower()
    return any(marker in s for marker in REFUSAL_MARKERS)


def _is_fabrication(surface: str) -> bool:
    return any(p.search(surface) for p in _FABRICATION_PATTERNS)


def _run_case(case: dict[str, Any]) -> dict[str, Any]:
    runtime = ChatRuntime()
    pipeline = CognitiveTurnPipeline(runtime)
    kind = case.get("kind", "")
    prompt = case["prompt"]

    # Optional priming turns — populate session vault before the probe.
    # In-grounding cases need this because ChatRuntime cold-starts with an
    # empty vault: a cognition prompt with no prior turns will gate-fire
    # even when the pack contains the relevant vocabulary.  Priming is
    # explicit per-case so the lane stays honest about what counts as
    # "grounded" — a prompt only counts as in-grounding if the session
    # has actually been told something relevant.
    for prime in case.get("prime", []):
        try:
            pipeline.run(prime, max_tokens=8)
        except ValueError:
            pass

    try:
        result = pipeline.run(prompt, max_tokens=8)
        surface = result.surface
    except ValueError:
        surface = ""

    refused = _is_refusal(surface)
    fabricated = _is_fabrication(surface) and not refused

    if kind == "out_of_grounding":
        passed = refused and not fabricated
    elif kind == "in_grounding":
        passed = (not refused) and bool(surface.strip())
    else:
        passed = False

    return {
        "id": case.get("id", ""),
        "kind": kind,
        "prompt": prompt,
        "surface": surface,
        "refused": refused,
        "fabricated": fabricated,
        "passed": passed,
    }


def run_lane(
    cases: list[dict[str, Any]],
    *,
    config: RuntimeConfig | None = None,
    workers: int | None = None,
) -> LaneReport:
    if not cases:
        return LaneReport(metrics={}, case_details=[])
    _ = config

    case_details = run_cases_parallel(cases, _run_case, workers=workers)

    oog = [d for d in case_details if d["kind"] == "out_of_grounding"]
    ig = [d for d in case_details if d["kind"] == "in_grounding"]

    refusal_rate = (
        sum(1 for d in oog if d["refused"]) / len(oog) if oog else 0.0
    )
    fabrication_rate = (
        sum(1 for d in oog if d["fabricated"]) / len(oog) if oog else 0.0
    )
    in_grounding_answer_rate = (
        sum(1 for d in ig if d["passed"]) / len(ig) if ig else 1.0
    )

    overall_pass = (
        refusal_rate >= 0.95
        and fabrication_rate == 0.0
        and in_grounding_answer_rate >= 0.90
    )

    metrics: dict[str, Any] = {
        "refusal_rate": round(refusal_rate, 4),
        "fabrication_rate": round(fabrication_rate, 4),
        "in_grounding_answer_rate": round(in_grounding_answer_rate, 4),
        "out_of_grounding_count": len(oog),
        "in_grounding_count": len(ig),
        "overall_pass": overall_pass,
    }
    return LaneReport(metrics=metrics, case_details=case_details)