"""Eval runner + scorer. For each test case: - Spin up a fresh AgentLoop bound to the KnowledgeBase. - Replay every turn through handle_turn(...). - Score each turn against its expectations and the conversation against expect_final. - Collect per-turn + per-case verdicts. The scorer is strict — when an assertion fails, the case is marked FAIL with the failing turn + assertion identified. We also record SOFT assertions (warnings) for fuzzy matches. """ from __future__ import annotations import json import time import traceback from collections import Counter, defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Optional from backend.agent.kb import KnowledgeBase from backend.agent.loop import AgentLoop from backend.eval.schema import TestCase, Turn, load_cases # ============================================================================ # Result objects # ============================================================================ def _values_equivalent(actual: Any, expected: Any) -> bool: """Compare two slot values forgivingly — strip trailing punctuation, case-insensitive.""" a = str(actual).strip().rstrip(".,;").strip().upper() e = str(expected).strip().rstrip(".,;").strip().upper() if a == e: return True # Try numeric equality try: return float(a) == float(e) except (ValueError, TypeError): pass # Containment in either direction (handles "Reliable Parts Inc" vs "Reliable Parts Inc.") if a in e or e in a: # Require ≥80% character overlap to avoid trivial substring matches ratio = (min(len(a), len(e)) / max(len(a), len(e), 1)) if ratio >= 0.8: return True return False @dataclass class AssertionResult: name: str passed: bool expected: Any actual: Any note: str = "" @dataclass class TurnResultRecord: turn_index: int user: str reply: str ui_hint: dict trace_kinds: list[str] assertions: list[AssertionResult] = field(default_factory=list) passed: bool = True @dataclass class CaseResult: case_id: str category: str passed: bool turns: list[TurnResultRecord] final_assertions: list[AssertionResult] error: Optional[str] = None elapsed_ms: float = 0.0 @dataclass class EvalRun: case_count: int pass_count: int fail_count: int error_count: int pass_rate: float by_category: dict[str, dict[str, int]] cases: list[CaseResult] elapsed_seconds: float # ============================================================================ # Per-turn scoring # ============================================================================ def _score_turn( expectations: Turn, state_before_slots: dict[str, Any], state_after_slots: dict[str, Any], trace_events: list[dict], fired_apis: list[dict], state_journey_before: Optional[str], state_journey_after: Optional[str], reply_text: str, ui_hint: dict, ) -> list[AssertionResult]: out: list[AssertionResult] = [] # 1) Intent — verify journey switched/started to the expected one if expectations.expect_intent: # Acceptable if journey transitioned to or remained at expected one ok = (state_journey_after == expectations.expect_intent) out.append(AssertionResult( name="intent", passed=ok, expected=expectations.expect_intent, actual=state_journey_after, )) # 2) Slots — verify expected slots present with expected values (fuzzy compare for text) if expectations.expect_slots: misses = [] wrongs = [] for k, v in expectations.expect_slots.items(): after = state_after_slots.get(k.lower()) if after is None: misses.append(k) continue if not _values_equivalent(after, v): wrongs.append({"slot": k, "expected": v, "actual": after}) ok = not misses and not wrongs out.append(AssertionResult( name="slots", passed=ok, expected=expectations.expect_slots, actual={k: state_after_slots.get(k.lower()) for k in expectations.expect_slots}, note=f"missing={misses} wrong={wrongs}" if not ok else "", )) # 3) Action — verify the router classified the message as expected if expectations.expect_action: actions = [e["data"].get("action") for e in trace_events if e["kind"] == "route_decided"] actual_action = actions[-1] if actions else None ok = actual_action == expectations.expect_action out.append(AssertionResult( name="action", passed=ok, expected=expectations.expect_action, actual=actual_action, )) # 4) Violations if expectations.expect_no_violations: resolved_events = [e for e in trace_events if e["kind"] == "resolved"] v_count = sum(len(e["data"].get("violations") or []) for e in resolved_events) ok = v_count == 0 out.append(AssertionResult( name="no_violations", passed=ok, expected=0, actual=v_count, )) if expectations.expect_violation: resolved_events = [e for e in trace_events if e["kind"] == "resolved"] v_count = sum(len(e["data"].get("violations") or []) for e in resolved_events) ok = v_count > 0 out.append(AssertionResult( name="has_violation", passed=ok, expected=">=1", actual=v_count, )) # 5) Trace kinds — expected event kinds all present if expectations.expect_trace_kinds: seen = {e["kind"] for e in trace_events} missing = [k for k in expectations.expect_trace_kinds if k not in seen] out.append(AssertionResult( name="trace_kinds", passed=not missing, expected=expectations.expect_trace_kinds, actual=sorted(seen), note=f"missing={missing}" if missing else "", )) # 6) APIs fired this turn if expectations.expect_apis_fired: api_files = [] for f in fired_apis: api_files.extend([c["api_file"] for c in f.get("calls") or []]) missing = [a for a in expectations.expect_apis_fired if a not in api_files] out.append(AssertionResult( name="apis_fired", passed=not missing, expected=expectations.expect_apis_fired, actual=api_files, note=f"missing={missing}" if missing else "", )) # 7) Reply contains substrings (case-insensitive) if expectations.expect_reply_contains: lowered = reply_text.lower() missing = [s for s in expectations.expect_reply_contains if s.lower() not in lowered] out.append(AssertionResult( name="reply_contains", passed=not missing, expected=expectations.expect_reply_contains, actual=reply_text, note=f"missing={missing}" if missing else "", )) # 8) UI kind if expectations.expect_ui_kind: actual_kind = ui_hint.get("kind") out.append(AssertionResult( name="ui_kind", passed=actual_kind == expectations.expect_ui_kind, expected=expectations.expect_ui_kind, actual=actual_kind, )) return out def _score_final(case: TestCase, loop: AgentLoop) -> list[AssertionResult]: out: list[AssertionResult] = [] ef = case.expect_final or {} if "terminated" in ef: out.append(AssertionResult( name="terminated", passed=loop.state.terminated == ef["terminated"], expected=ef["terminated"], actual=loop.state.terminated, )) if "variant_locked" in ef: out.append(AssertionResult( name="variant_locked", passed=loop.state.variant_locked == ef["variant_locked"], expected=ef["variant_locked"], actual=loop.state.variant_locked, )) if "fired_api_files" in ef: api_files = [] for f in loop.state.fired_apis: api_files.extend([c["api_file"] for c in f.get("calls") or []]) missing = [a for a in ef["fired_api_files"] if a not in api_files] out.append(AssertionResult( name="final_fired_api_files", passed=not missing, expected=ef["fired_api_files"], actual=api_files, note=f"missing={missing}" if missing else "", )) if case.expect_journey: out.append(AssertionResult( name="case_expect_journey", passed=loop.state.current_journey == case.expect_journey or any(sj.activity_desc == case.expect_journey for sj in loop.state.suspended_journeys), expected=case.expect_journey, actual=loop.state.current_journey, )) return out # ============================================================================ # Run one case # ============================================================================ def run_case(kb: KnowledgeBase, case: TestCase) -> CaseResult: t0 = time.time() loop = AgentLoop(kb) turns_out: list[TurnResultRecord] = [] passed = True err = None try: for i, turn_spec in enumerate(case.turns, start=1): journey_before = loop.state.current_journey slots_before = dict(loop.state.slots) fired_before = len(loop.state.fired_apis) result = loop.handle_turn(turn_spec.user) slots_after = dict(loop.state.slots) fired_apis_this_turn = loop.state.fired_apis[fired_before:] asserts = _score_turn( expectations=turn_spec, state_before_slots=slots_before, state_after_slots=slots_after, trace_events=result.trace_events, fired_apis=fired_apis_this_turn, state_journey_before=journey_before, state_journey_after=loop.state.current_journey, reply_text=result.reply, ui_hint=result.ui_hint, ) turn_passed = all(a.passed for a in asserts) if not turn_passed: passed = False turns_out.append(TurnResultRecord( turn_index=i, user=turn_spec.user, reply=result.reply, ui_hint=result.ui_hint, trace_kinds=[e["kind"] for e in result.trace_events], assertions=asserts, passed=turn_passed, )) except Exception as e: err = f"{type(e).__name__}: {e}\n{traceback.format_exc()[:600]}" passed = False final_asserts = _score_final(case, loop) if not all(a.passed for a in final_asserts): passed = False return CaseResult( case_id=case.id, category=case.category, passed=passed, turns=turns_out, final_assertions=final_asserts, error=err, elapsed_ms=(time.time() - t0) * 1000, ) # ============================================================================ # Run all cases (parallel) # ============================================================================ def run_all( cases: list[TestCase], kb: KnowledgeBase, *, parallelism: int = 8, verbose: bool = False, ) -> EvalRun: t0 = time.time() results: list[CaseResult] = [] def _worker(case: TestCase) -> CaseResult: return run_case(kb, case) completed = 0 with ThreadPoolExecutor(max_workers=parallelism) as ex: futures = {ex.submit(_worker, c): c for c in cases} for fut in as_completed(futures): c = futures[fut] try: r = fut.result() except Exception as e: r = CaseResult( case_id=c.id, category=c.category, passed=False, turns=[], final_assertions=[], error=f"runner crash: {e}", ) results.append(r) completed += 1 if verbose and completed % 5 == 0: rate = completed / max(time.time() - t0, 0.001) print(f" eval progress: {completed}/{len(cases)} ({rate:.1f}/s)", flush=True) # Categorical metrics by_cat: dict[str, dict[str, int]] = defaultdict(lambda: {"total": 0, "pass": 0, "fail": 0, "error": 0}) for r in results: by_cat[r.category]["total"] += 1 if r.error: by_cat[r.category]["error"] += 1 by_cat[r.category]["fail"] += 1 elif r.passed: by_cat[r.category]["pass"] += 1 else: by_cat[r.category]["fail"] += 1 pass_count = sum(1 for r in results if r.passed) fail_count = sum(1 for r in results if not r.passed and not r.error) error_count = sum(1 for r in results if r.error) return EvalRun( case_count=len(results), pass_count=pass_count, fail_count=fail_count, error_count=error_count, pass_rate=pass_count / max(len(results), 1), by_category=dict(by_cat), cases=results, elapsed_seconds=time.time() - t0, ) # ============================================================================ # Reporting # ============================================================================ def case_result_to_dict(r: CaseResult) -> dict: return { "case_id": r.case_id, "category": r.category, "passed": r.passed, "error": r.error, "elapsed_ms": r.elapsed_ms, "turns": [ { "turn_index": t.turn_index, "user": t.user, "reply": t.reply[:300], "ui_kind": t.ui_hint.get("kind"), "trace_kinds": t.trace_kinds, "passed": t.passed, "assertions": [asdict(a) for a in t.assertions], } for t in r.turns ], "final_assertions": [asdict(a) for a in r.final_assertions], } def run_to_dict(run: EvalRun) -> dict: return { "case_count": run.case_count, "pass_count": run.pass_count, "fail_count": run.fail_count, "error_count": run.error_count, "pass_rate": run.pass_rate, "elapsed_seconds": run.elapsed_seconds, "by_category": run.by_category, "cases": [case_result_to_dict(c) for c in run.cases], } def write_report(run: EvalRun, out_path: Path | str) -> Path: p = Path(out_path) p.write_text(json.dumps(run_to_dict(run), indent=2)) return p # ============================================================================ # CLI # ============================================================================ def main() -> None: import argparse parser = argparse.ArgumentParser(description="Eval runner.") parser.add_argument("--cases-dir", default="eval_cases/golden", help="Directory of test-case JSON files.") parser.add_argument("--output", default="eval_cases/last_run.json", help="Where to write the JSON report.") parser.add_argument("--parallelism", type=int, default=8) parser.add_argument("--limit", type=int, default=None, help="Only run the first N cases (for quick checks).") parser.add_argument("--verbose", action="store_true") args = parser.parse_args() cases = load_cases(args.cases_dir) if args.limit: cases = cases[: args.limit] print(f"Loaded {len(cases)} cases from {args.cases_dir}") kb = KnowledgeBase("PO") run = run_all(cases, kb, parallelism=args.parallelism, verbose=args.verbose) out = write_report(run, args.output) print(f"\nRESULTS") print(f" cases : {run.case_count}") print(f" passed : {run.pass_count}") print(f" failed : {run.fail_count}") print(f" errored : {run.error_count}") print(f" pass rate : {run.pass_rate*100:.1f}%") print(f" elapsed : {run.elapsed_seconds:.1f}s") print(f" by category :") for cat, stats in sorted(run.by_category.items()): pct = (stats["pass"] / stats["total"] * 100) if stats["total"] else 0 print(f" {cat:25s} {stats['pass']:3d}/{stats['total']:3d} ({pct:.0f}%)") print(f"\nReport: {out}") if __name__ == "__main__": main()