eval/run_eval.py

"""
EDD harness — Day 3

Runs each of the three Day-3 agents' JSON eval cases and checks:
1. The output validates against its Pydantic schema (this alone catches
   most silent multi-agent failures, per the design doc's Section 1).
2. A handful of soft field assertions per case (contains / not_null /
   equals / min / max / length checks).

Each agent is run IN ISOLATION (not through the full pipeline) with
session state seeded directly. This matters for url_reputation cases
specifically: running the full pipeline for every case would burn IPQS's
1,000/month and VirusTotal's 4/min budget on cases that don't even test
reputation logic. Only the url_reputation suite touches those APIs.

This is intentionally lightweight — full trajectory-mode scoring (exact
tool-call sequence via ADK's eval module) is a Day 6 evaluation-layer
task, not a Day 3 blocker. This harness exists to catch schema mismatches
and obviously wrong extractions before they compound into Day 4's Risk
Synthesis Agent.

Usage (from repo root):
    python eval/run_eval.py intake
    python eval/run_eval.py linguistic
    python eval/run_eval.py url_reputation   # touches IPQS/VirusTotal
    python eval/run_eval.py all
"""

import asyncio
import json
import sys
import uuid
from pathlib import Path

# Same fix as scripts/smoke_test_day3.py: running `python eval/run_eval.py`
# puts eval/ on sys.path, not the repo root, so `agents` isn't importable
# without this.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

# Same fix as scripts/smoke_test_day3.py — must run before any agent
# touches the Gemini client.
from dotenv import load_dotenv

load_dotenv()

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

from agents.intake_agent import intake_agent
from agents.linguistic_agent import linguistic_agent
from agents.schemas import IntakeResult, LinguisticSignal, ReputationSignalList
from agents.url_reputation_agent import url_reputation_agent

APP_NAME = "scamsentry-eval"
CASES_DIR = Path(__file__).resolve().parent / "cases"

AGENTS = {
    "intake": intake_agent,
    "url_reputation": url_reputation_agent,
    "linguistic": linguistic_agent,
}
SCHEMAS = {
    "intake": IntakeResult,
    "url_reputation": ReputationSignalList,
    "linguistic": LinguisticSignal,
}
STATE_KEYS = {
    "intake": "intake_result",
    "url_reputation": "reputation_signals",
    "linguistic": "linguistic_signal",
}


def check_soft_assertions(output: dict, expected: dict) -> list[str]:
    """Returns a list of human-readable failure messages (empty = pass).

    NOTE ON ORDERING (fixed 2026-07-03): suffixes are checked longest/most
    specific first. "_len_min" and "_len_equals" MUST be checked before
    "_min"/"_max"/"_equals" respectively, because e.g. "signals_len_min"
    also ends with "_min" — str.endswith() doesn't know one suffix is
    "more specific" than another, so the shorter/more general check will
    silently steal the match if it comes first in the chain, computing
    the wrong field name (e.g. "signals_len" instead of "signals") and
    producing a false failure against a field that never existed. This
    caused every url_reputation case to fail with "expected 'signals_len'
    ... got None" even when the underlying agent output was fully valid
    and had already passed schema validation above this check.
    """
    failures = []
    for key, expectation in expected.items():
        if key.endswith("_contains"):
            field = key[: -len("_contains")]
            actual_list = output.get(field) or []
            for item in expectation:
                if not any(str(item).lower() in str(v).lower() for v in actual_list):
                    failures.append(
                        f"expected {field!r} to contain {item!r}, got {actual_list!r}"
                    )
        elif key.endswith("_not_null"):
            field = key[: -len("_not_null")]
            if expectation and output.get(field) is None:
                failures.append(f"expected {field!r} to be non-null, got null")
        elif key.endswith("_len_min"):
            field = key[: -len("_len_min")]
            if len(output.get(field) or []) < expectation:
                failures.append(
                    f"expected len({field!r}) >= {expectation!r}, "
                    f"got {len(output.get(field) or [])}"
                )
        elif key.endswith("_len_equals"):
            field = key[: -len("_len_equals")]
            if len(output.get(field) or []) != expectation:
                failures.append(
                    f"expected len({field!r}) == {expectation!r}, "
                    f"got {len(output.get(field) or [])}"
                )
        elif key.endswith("_equals"):
            field = key[: -len("_equals")]
            if output.get(field) != expectation:
                failures.append(
                    f"expected {field!r} == {expectation!r}, got {output.get(field)!r}"
                )
        elif key.endswith("_min"):
            field = key[: -len("_min")]
            if (output.get(field) or 0) < expectation:
                failures.append(
                    f"expected {field!r} >= {expectation!r}, got {output.get(field)!r}"
                )
        elif key.endswith("_max"):
            field = key[: -len("_max")]
            if (output.get(field) or 0) > expectation:
                failures.append(
                    f"expected {field!r} <= {expectation!r}, got {output.get(field)!r}"
                )
    return failures


async def run_agent_isolated(agent_key: str, seed_state: dict, user_text: str) -> dict:
    """Runs a single agent with pre-seeded session state (bypasses the
    rest of the pipeline)."""
    agent = AGENTS[agent_key]
    session_service = InMemorySessionService()
    user_id, session_id = "eval_user", str(uuid.uuid4())
    await session_service.create_session(
        app_name=APP_NAME, user_id=user_id, session_id=session_id, state=seed_state
    )
    runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
    content = types.Content(role="user", parts=[types.Part(text=user_text)])
    async for _ in runner.run_async(
        user_id=user_id, session_id=session_id, new_message=content
    ):
        pass
    final_session = await session_service.get_session(
        app_name=APP_NAME, user_id=user_id, session_id=session_id
    )
    raw = final_session.state.get(STATE_KEYS[agent_key])
    if raw is None:
        return {"_missing_output": True}
    try:
        return json.loads(raw) if isinstance(raw, str) else raw
    except json.JSONDecodeError:
        return {"_parse_error": True, "_raw": str(raw)}


async def run_case(agent_key: str, case: dict) -> tuple[bool, str]:
    inp = case["input"]
    if agent_key == "intake":
        seed_state, user_text = {}, inp["text"]
    else:
        # url_reputation and linguistic both read {intake_result} from
        # state — seed it directly rather than re-running Intake.
        seed_state = {"intake_result": json.dumps(inp["intake_result"])}
        user_text = "continue"

    output = await run_agent_isolated(agent_key, seed_state, user_text)
    if output.get("_missing_output"):
        return False, "agent produced no output for this state key"
    if output.get("_parse_error"):
        return False, f"output was not valid JSON: {output['_raw'][:200]}"

    try:
        SCHEMAS[agent_key].model_validate(output)
    except Exception as e:
        return False, f"schema validation failed: {e}"

    failures = check_soft_assertions(output, case.get("expected_output", {}))
    if failures:
        return False, "; ".join(failures)
    return True, "ok"


async def run_suite(agent_key: str) -> None:
    case_file = CASES_DIR / f"{agent_key}_cases.json"
    if not case_file.exists():
        print(f"No case file found at {case_file}")
        return
    cases = json.loads(case_file.read_text())
    passed = 0
    for case in cases:
        ok, message = await run_case(agent_key, case)
        status = "PASS" if ok else "FAIL"
        print(f"[{status}] {case['case_id']}: {message}")
        if ok:
            passed += 1
    print(f"\n{agent_key}: {passed}/{len(cases)} passed")


async def main():
    target = sys.argv[1] if len(sys.argv) > 1 else "all"
    keys = list(AGENTS.keys()) if target == "all" else [target]
    for key in keys:
        print(f"\n=== {key} ===")
        await run_suite(key)


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