eval/help/run_eval.py

"""Help-skill eval runner.

Feeds each golden case in `help_dataset.json` to the LIVE Help skill
(`src/agents/handlers/help.HelpAgent.astream`), then scores whether the streamed
reply obeys a set of RULE assertions — reply language, never suggesting a report
when `report_ready.ready=false`, suggesting it when true. Prints a per-case detail
table + aggregate summary and writes a timestamped JSON report under `results/`
(never overwritten — one file per run, diffable).

Unlike `eval/readiness` (deterministic, no LLM), this calls the model for real, so
it needs a working `.env` (Azure OpenAI) and spends tokens — run it before a deploy
that touches `help.md`, not on every commit. `tests/unit/agents/handlers/test_help.py`
already covers the deterministic Python guard with a fake chain; this is the
end-to-end "does the model actually obey the prompt" layer on top.

Two things the metric separates on purpose:
  - COMPLIANCE = % of rule assertions that hold. NOT accuracy — help replies are free
    prose with no single correct wording; we score rule-obedience, not similarity.
  - HELD-OUT vs CARRIED-OVER — carried_over cases mirror a help.md example (regression);
    held-out cases are absent from the prompt. Held-out compliance is the real
    generalization signal. If held-out drops while carried_over stays 100%, the prompt
    is overfitting to its own examples.

`orientation` cases are `manual_review` — run but excluded from the auto compliance
rate; read their `output_text` in the JSON report to judge suggestion quality.

Invoke as a module so `src` imports resolve:

    uv run python -m eval.help.run_eval
    uv run python -m eval.help.run_eval --limit 4     # smoke test
    uv run python -m eval.help.run_eval --no-table    # summary only
"""

from __future__ import annotations

import argparse
import asyncio
import json
import statistics
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.outputs import LLMResult

from src.agents.gate import stub_analysis_state
from src.agents.handlers.help import HelpAgent, ReportReadiness, _detect_reply_language
from src.agents.report.readiness import _MISSING_ANALYSIS, _MISSING_DELTA

_HERE = Path(__file__).resolve().parent
DATASET = _HERE / "help_dataset.json"
RESULTS_DIR = _HERE / "results"
GROUPS = ["language", "report_guard", "orientation"]

# Dataset short codes -> the exact `missing` strings is_report_ready emits. Imported
# from the module so the dataset stays readable and survives wording changes.
_CODE_TO_MISSING = {
    "analysis": _MISSING_ANALYSIS,
    "delta": _MISSING_DELTA,
}


class _UsageCollector(BaseCallbackHandler):
    """Sums token usage across the LLM calls made during one astream()."""

    def __init__(self) -> None:
        self.input_tokens = 0
        self.output_tokens = 0
        self.total_tokens = 0

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        before = self.total_tokens
        for generation_list in response.generations:
            for generation in generation_list:
                message = getattr(generation, "message", None)
                usage = getattr(message, "usage_metadata", None) if message else None
                if usage:
                    self.input_tokens += usage.get("input_tokens", 0)
                    self.output_tokens += usage.get("output_tokens", 0)
                    self.total_tokens += usage.get("total_tokens", 0)
        if self.total_tokens == before and response.llm_output:
            usage = response.llm_output.get("token_usage") or {}
            self.input_tokens += usage.get("prompt_tokens", 0)
            self.output_tokens += usage.get("completion_tokens", 0)
            self.total_tokens += usage.get("total_tokens", 0)

    @property
    def tokens(self) -> dict[str, int]:
        return {
            "input": self.input_tokens,
            "output": self.output_tokens,
            "total": self.total_tokens,
        }


# --- assertion checkers -----------------------------------------------------
# Each returns (passed, detail). `detail` explains a failure in the table/report.


def _check_language_match(output: str, spec: dict[str, Any]) -> tuple[bool, str]:
    got = _detect_reply_language([], message=output)
    return got == spec["expected"], f"want {spec['expected']}, got {got}"


def _check_must_not_contain_any(output: str, spec: dict[str, Any]) -> tuple[bool, str]:
    low = output.lower()
    hits = [p for p in spec["patterns"] if p.lower() in low]
    return (not hits), (f"found {hits}" if hits else "none present")


def _check_must_contain_any(output: str, spec: dict[str, Any]) -> tuple[bool, str]:
    low = output.lower()
    hits = [p for p in spec["patterns"] if p.lower() in low]
    return bool(hits), (f"found {hits}" if hits else f"none of {spec['patterns']}")


_ASSERT_CHECKS = {
    "language_match": _check_language_match,
    "must_not_contain_any": _check_must_not_contain_any,
    "must_contain_any": _check_must_contain_any,
}


@dataclass
class AssertResult:
    type: str
    passed: bool
    detail: str


@dataclass
class CaseResult:
    id: str
    group: str
    carried_over: bool
    manual_review: bool
    output_text: str
    asserts: list[AssertResult]
    all_passed: bool | None  # None when manual_review (not auto-scored)
    latency_ms: float
    tokens: dict[str, int]


def load_cases(path: Path) -> list[dict[str, Any]]:
    """Read the `cases` array, skipping the leading `_*` doc keys and `schema`."""
    data = json.loads(path.read_text(encoding="utf-8"))
    return list(data["cases"])


def _build_state(spec: dict[str, Any]):
    """Build an AnalysisState from a case's `state` block (defaults from the stub)."""
    return stub_analysis_state().model_copy(
        update={
            "analysis_title": spec.get("analysis_title", "New analysis"),
            "objective": spec.get("objective", ""),
            "business_questions": list(spec.get("business_questions", [])),
            "report_id": spec.get("report_id"),
        }
    )


def _build_history(rows: list[dict[str, Any]]) -> list[BaseMessage]:
    out: list[BaseMessage] = []
    for row in rows:
        cls = HumanMessage if row["role"] == "human" else AIMessage
        out.append(cls(content=row["content"]))
    return out


def _build_readiness(spec: dict[str, Any]) -> ReportReadiness:
    return ReportReadiness(
        ready=bool(spec["ready"]),
        missing=[_CODE_TO_MISSING[c] for c in spec.get("missing", [])],
    )


async def run_case(case: dict[str, Any]) -> CaseResult:
    """Stream one Help reply and score its assertions; never throws."""
    state = _build_state(case["state"])
    history = _build_history(case.get("history", []))
    readiness = _build_readiness(case["report_ready"])
    collector = _UsageCollector()

    agent = HelpAgent()  # real Azure chain, constructed lazily on first astream
    start = time.perf_counter()
    try:
        output = "".join(
            [
                token
                async for token in agent.astream(
                    state,
                    history=history,
                    message=case.get("message"),
                    report_ready=readiness,
                    callbacks=[collector],
                )
            ]
        )
    except Exception as exc:  # noqa: BLE001 — one bad case shouldn't kill the run
        output = f"ERROR:{type(exc).__name__}: {exc}"
    latency_ms = round((time.perf_counter() - start) * 1000, 1)

    manual = bool(case.get("manual_review"))
    asserts: list[AssertResult] = []
    if not manual:
        for spec in case.get("asserts", []):
            check = _ASSERT_CHECKS[spec["type"]]
            passed, detail = check(output, spec)
            asserts.append(AssertResult(type=spec["type"], passed=passed, detail=detail))
    all_passed = None if manual else all(a.passed for a in asserts)

    return CaseResult(
        id=case["id"],
        group=case["group"],
        carried_over=bool(case.get("carried_over")),
        manual_review=manual,
        output_text=output,
        asserts=asserts,
        all_passed=all_passed,
        latency_ms=latency_ms,
        tokens=collector.tokens,
    )


def _compliance(results: list[CaseResult]) -> dict[str, Any]:
    scored = [r for r in results if r.all_passed is not None]
    passed = sum(1 for r in scored if r.all_passed)
    return {
        "n": len(scored),
        "passed": passed,
        "compliance": round(passed / len(scored), 3) if scored else 0.0,
    }


def summarize(results: list[CaseResult]) -> dict[str, Any]:
    scored = [r for r in results if r.all_passed is not None]
    latencies = [r.latency_ms for r in results]
    tok_total = sum(r.tokens["total"] for r in results)
    overall = _compliance(results)
    by_group = {
        g: _compliance([r for r in results if r.group == g])
        for g in GROUPS
        if any(r.group == g for r in results)
    }
    return {
        "total": len(results),
        "scored": len(scored),
        "manual_review": len(results) - len(scored),
        "passed": overall["passed"],
        "compliance": overall["compliance"],
        "runtime_avg_ms": round(statistics.mean(latencies), 1) if latencies else 0,
        "tokens_total": tok_total,
        "by_group": by_group,
        "held_out": _compliance([r for r in scored if not r.carried_over]),
        "carried_over": _compliance([r for r in scored if r.carried_over]),
    }


def _truncate(text: str, width: int) -> str:
    text = text.replace("\n", " ")
    return text if len(text) <= width else text[: width - 3] + "..."


def format_table(results: list[CaseResult]) -> str:
    header = (
        f"{'ID':<20} {'GROUP':<13} {'C/O':<4} {'ASSERTS':<22} {'OK':<4} {'MS':>7}"
    )
    rule = "-" * len(header)
    lines = [rule, header, rule]
    for r in results:
        co = "CO" if r.carried_over else "-"
        if r.manual_review:
            atypes, ok = "(manual)", "~"
        else:
            atypes = ",".join(a.type.replace("_", "")[:6] for a in r.asserts) or "-"
            ok = "ok" if r.all_passed else "X"
        lines.append(
            f"{r.id:<20} {r.group:<13} {co:<4} {_truncate(atypes, 22):<22} "
            f"{ok:<4} {r.latency_ms:>7}"
        )
    lines.append(rule)
    return "\n".join(lines)


def format_summary(summary: dict[str, Any], results: list[CaseResult]) -> str:
    lines = ["SUMMARY"]
    lines.append(
        f"  Compliance   {summary['passed']}/{summary['scored']} cases obey all rules"
        f"   ({summary['compliance'] * 100:.1f}%)   avg {summary['runtime_avg_ms']} ms"
    )
    lines.append(
        f"  Manual       {summary['manual_review']} case(s) excluded from the rate"
        " (read output_text)"
    )
    lines.append("")
    lines.append("  By group")
    for g, m in summary["by_group"].items():
        if m["n"]:
            lines.append(f"    {g:<14} {m['passed']}/{m['n']}  {m['compliance'] * 100:.0f}%")
        else:
            lines.append(f"    {g:<14} (manual only)")
    lines.append("")
    ho, co = summary["held_out"], summary["carried_over"]
    lines.append("  Held-out vs carried-over")
    lines.append(
        f"    held_out       {ho['passed']}/{ho['n']}  "
        f"{ho['compliance'] * 100:.0f}%   <- generalization"
    )
    lines.append(
        f"    carried_over   {co['passed']}/{co['n']}  "
        f"{co['compliance'] * 100:.0f}%   <- regression"
    )
    failures = [r for r in results if r.all_passed is False]
    lines.append("")
    lines.append(f"  FAILURES ({len(failures)})")
    for r in failures:
        bad = [f"{a.type}({a.detail})" for a in r.asserts if not a.passed]
        lines.append(f"    {r.id:<20} {r.group:<13} {'; '.join(bad)}")
    return "\n".join(lines)


def build_report(
    results: list[CaseResult], summary: dict[str, Any], meta: dict[str, Any]
) -> dict[str, Any]:
    run = {
        **meta,
        **{
            k: summary[k]
            for k in ("total", "scored", "manual_review", "passed", "compliance",
                      "runtime_avg_ms", "tokens_total")
        },
    }
    return {
        "run": run,
        "by_group": summary["by_group"],
        "held_out": summary["held_out"],
        "carried_over": summary["carried_over"],
        "cases": [asdict(r) for r in results],
    }


def _model_name() -> str:
    try:
        from src.config.settings import settings

        return str(settings.azureai_deployment_name_4o)
    except Exception:  # noqa: BLE001 — meta only; .env may be absent
        return "gpt-4o"


@dataclass
class _Args:
    dataset: Path = DATASET
    limit: int = 0
    no_table: bool = False
    extra: dict[str, Any] = field(default_factory=dict)


async def main() -> None:
    parser = argparse.ArgumentParser(description="Help-skill eval")
    parser.add_argument("--dataset", type=Path, default=DATASET)
    parser.add_argument("--limit", type=int, default=0, help="run first N cases only")
    parser.add_argument("--prompt-version", default="help.md")
    parser.add_argument("--no-table", action="store_true", help="skip the detail table")
    args = parser.parse_args()

    cases = load_cases(args.dataset)
    if args.limit:
        cases = cases[: args.limit]

    started = datetime.now()
    print(f"Help Skill Eval -- {started:%Y-%m-%d %H:%M:%S}")
    print(
        f"dataset: {args.dataset.name} ({len(cases)} cases)  model: {_model_name()}  "
        f"prompt: {args.prompt_version}  target: HelpAgent.astream (live)"
    )

    results = [await run_case(case) for case in cases]

    summary = summarize(results)
    if not args.no_table:
        print(format_table(results))
    print(format_summary(summary, results))

    meta = {
        "timestamp": started.isoformat(timespec="seconds"),
        "dataset": args.dataset.name,
        "model": _model_name(),
        "prompt_version": args.prompt_version,
        "target": "src/agents/handlers/help.HelpAgent.astream",
    }
    report = build_report(results, summary, meta)
    RESULTS_DIR.mkdir(parents=True, exist_ok=True)
    out_path = RESULTS_DIR / f"help_result_{started:%Y-%m-%d_%H%M%S}.json"
    out_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"\n-> saved: {out_path.relative_to(_HERE.parent.parent)}")


if __name__ == "__main__":
    asyncio.run(main())