Spaces:

SooryaKi
/

ScamSentry

Running

SooryaKiranKakkarayil

Day 3 complete: fix url_reputation_agent output_schema+tools bug, fix eval harness suffix bug, switch to gemini-3.1-flash-lite, all 9 EDD cases + full smoke test pass

e2329f0 4 months ago

Raw History Blame Contribute Delete

8.57 kB

"""
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())

	"""
	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())