Code-first: a Pytest Harness for Agents

code/harness/agent_client.py
"""Thin async wrapper over the Copilot Studio client that returns evidence, not just text.
pip install microsoft-agents-copilotstudio-client msal"""
from __future__ import annotations
import os, time, msal
from dataclasses import dataclass, field
from microsoft_agents.activity import ActivityTypes
from microsoft_agents.copilotstudio.client import ConnectionSettings, CopilotClient
 
SCOPES = ["https://api.powerplatform.com/.default"]   # check your SDK version's recommended scope helper
 
@dataclass
class Turn:
    user: str
    text: str
    activities: list = field(default_factory=list)
    citations: list[str] = field(default_factory=list)
    latency_s: float = 0.0
 
@dataclass
class Conversation:
    persona: str
    conversation_id: str
    turns: list[Turn] = field(default_factory=list)
 
def persona_token(persona: str) -> str:
    """Silent token for a pre-registered test persona. Cache files are created once per persona
    with tools/login_persona.py (interactive) and stored in Key Vault for CI."""
    cache = msal.SerializableTokenCache()
    path = os.path.join(os.environ.get("PERSONA_CACHE_DIR", ".personas"), f"{persona}.bin")
    with open(path, encoding="utf-8") as fh:
        cache.deserialize(fh.read())
    app = msal.PublicClientApplication(os.environ["HARNESS_CLIENT_ID"],
                                       authority=f"https://login.microsoftonline.com/{os.environ['TENANT_ID']}",
                                       token_cache=cache)
    accounts = app.get_accounts()
    result = app.acquire_token_silent(SCOPES, account=accounts[0]) if accounts else None
    if not result or "access_token" not in result:
        raise RuntimeError(f"persona {persona}: token refresh failed; re-run login_persona.py")
    if cache.has_state_changed:
        with open(path, "w", encoding="utf-8") as fh:
            fh.write(cache.serialize())
    return result["access_token"]
 
def _citations(activity) -> list[str]:
    """Collect citation names/urls from message entities. Entity shapes vary by channel and version,
    so this is defensive: inspect one real payload from your agent and tighten it."""
    out = []
    for e in (getattr(activity, "entities", None) or []):
        data = e if isinstance(e, dict) else getattr(e, "__dict__", {})
        for c in data.get("citation", []) or []:
            app_ = c.get("appearance", {}) if isinstance(c, dict) else {}
            out.append(app_.get("name") or app_.get("url") or str(c))
    return out
 
class AgentSession:
    def __init__(self, persona: str):
        self.persona = persona
        settings = ConnectionSettings(environment_id=os.environ["MCS_ENVIRONMENT_ID"],
                                      agent_identifier=os.environ["MCS_AGENT_SCHEMA_NAME"],
                                      cloud=None, copilot_agent_type=None, custom_power_platform_cloud=None)
        self.client = CopilotClient(settings, persona_token(persona))
        self.conv: Conversation | None = None
 
    async def start(self) -> Conversation:
        conv_id = None
        async for act in self.client.start_conversation(emit_start_conversation_event=True):
            conv_id = conv_id or getattr(getattr(act, "conversation", None), "id", None)
        self.conv = Conversation(self.persona, conv_id)
        return self.conv
 
    async def say(self, text: str) -> Turn:
        assert self.conv, "call start() first"
        t0, acts = time.perf_counter(), []
        async for act in self.client.ask_question(text, self.conv.conversation_id):
            acts.append(act)
        msgs = [a for a in acts if a.type == ActivityTypes.message and getattr(a, "text", None)]
        turn = Turn(user=text, text="\n".join(m.text for m in msgs), activities=acts,
                    citations=[c for m in msgs for c in _citations(m)],
                    latency_s=time.perf_counter() - t0)
        self.conv.turns.append(turn)
        return turn

code/harness/conftest.py
import os, pathlib, yaml, pytest
from golden_schema import GoldenCase   # from Episode 2
 
GOLDEN = pathlib.Path(os.environ.get("GOLDEN_DIR", "tests/golden"))
TRIALS = {"critical": int(os.environ.get("TRIALS_CRITICAL", 10)), "high": 5, "medium": 3, "low": 1}
 
def _cases() -> list[GoldenCase]:
    return [GoldenCase(**yaml.safe_load(p.read_text(encoding="utf-8")))
            for p in sorted(GOLDEN.rglob("GS-*.yaml")) if "candidates" not in p.parts]
 
def pytest_generate_tests(metafunc):
    if "case" in metafunc.fixturenames:
        params = []
        for c in _cases():
            for t in range(TRIALS[c.severity.value]):
                inputs = [c.input] + c.paraphrases if c.input else [None]
                params.append(pytest.param(c, inputs[t % len(inputs)], t,
                                           id=f"{c.id}-t{t}", marks=pytest.mark.severity(c.severity.value)))
        metafunc.parametrize("case,utterance,trial", params)

code/harness/test_golden.py
import pytest
from agent_client import AgentSession
from spy_client import calls_for          # Episode 5: reads the tool-spy log by correlation id
from judge import deterministic_checks, grounded_against_source
 
@pytest.mark.asyncio
async def test_golden_case(case, utterance, trial, record_property):
    s = AgentSession(case.identity["persona"])
    conv = await s.start()
 
    if case.turns:                                         # multi-turn: halt on first critical failure
        for i, turn in enumerate(case.turns, 1):
            reply = await s.say(turn.user)
            missing = [k for k in turn.expect_contains if k.lower() not in reply.text.lower()]
            if missing and turn.critical:
                pytest.fail(f"turn {i} missing {missing}: {reply.text[:200]}")
    else:
        reply = await s.say(utterance)
 
    # 1. deterministic checks first: cheap, exact, never hallucinate
    problems = deterministic_checks(case, reply)
 
    # 2. route and side effects: what the tool spy actually saw for this conversation
    calls = calls_for(conv.conversation_id)
    expected_tools = [c.tool for c in case.expected.route]
    unexpected = [c for c in calls if c["tool"] not in expected_tools]
    if unexpected:
        problems.append(f"unexpected tool calls: {[c['tool'] for c in unexpected]}")
    for exp in case.expected.route:
        made = [c for c in calls if c["tool"] == exp.tool]
        if exp.requires_confirmation and made and not all(c.get("confirmed") for c in made):
            problems.append(f"{exp.tool} executed without confirmation")
        if len(made) > exp.max_calls:
            problems.append(f"{exp.tool} called {len(made)}x (max {exp.max_calls})")
        for c in made:
            wrong = {k: v for k, v in exp.args.items() if str(c["args"].get(k)) != v}
            if wrong:
                problems.append(f"{exp.tool} wrong args {wrong}")
 
    # 3. LLM judge last, only where meaning matters and a source excerpt exists
    if case.expected.outcome == "answer" and case.expected.sources:
        g = grounded_against_source(case, reply)
        record_property("groundedness", g["groundedness"])
        if g["groundedness"] <= 2:                         # per-trial floor; the gate enforces mean >= 4
            problems.append(f"groundedness {g['groundedness']}: {g['groundedness_reason'][:160]}")
 
    record_property("latency_s", round(reply.latency_s, 2))
    record_property("case_id", case.id)
    record_property("severity", case.severity.value)
    assert not problems, "; ".join(problems)

code/harness/judge.py
import os, pathlib
from azure.ai.evaluation import GroundednessEvaluator, AzureOpenAIModelConfiguration
 
SOURCES = pathlib.Path(os.environ.get("APPROVED_SOURCES", "tests/approved_sources"))  # approved excerpts
 
_model = AzureOpenAIModelConfiguration(azure_endpoint=os.environ["JUDGE_AOAI_ENDPOINT"],
                                       azure_deployment=os.environ["JUDGE_AOAI_DEPLOYMENT"],
                                       api_version=os.environ.get("JUDGE_AOAI_API_VERSION", "2024-10-21"))
_groundedness = GroundednessEvaluator(model_config=_model, threshold=4)
 
def deterministic_checks(case, reply) -> list[str]:
    t, p = reply.text.lower(), []
    p += [f"missing '{k}'" for k in case.expected.must_mention if k.lower() not in t]
    p += [f"forbidden '{k}'" for k in case.expected.must_not_mention if k.lower() in t]
    if case.expected.outcome == "answer" and case.expected.sources and not reply.citations:
        p.append("in-scope answer without citation")
    if case.expected.outcome in ("fallback", "refuse") and any(ch.isdigit() for ch in t) and "%" in t:
        p.append("fallback/refusal case states a figure")      # crude but catches invented rates
    return p
 
def grounded_against_source(case, reply) -> dict:
    """Judge the reply against the *approved* excerpt in the golden set, not whatever was retrieved.
    That is the difference between 'grounded' and 'grounded in the truth'."""
    context = "\n\n".join((SOURCES / f"{s}.md").read_text(encoding="utf-8") for s in case.expected.sources)
    return _groundedness(query=reply.user, response=reply.text, context=context)

code/harness/gate.py
"""python gate.py results.xml --threshold 0.95  — reads pytest JUnit XML, gates per case and per category."""
import math, sys, collections, argparse, xml.etree.ElementTree as ET
 
def wilson_lower(passed: int, n: int, z: float = 1.96) -> float:
    if n == 0:
        return 0.0
    p = passed / n
    d = 1 + z * z / n
    c = p + z * z / (2 * n)
    a = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n))
    return (c - a) / d
 
def main():
    ap = argparse.ArgumentParser(); ap.add_argument("junit"); ap.add_argument("--threshold", type=float, default=0.95)
    a = ap.parse_args()
    per_case, per_cat, critical = collections.defaultdict(lambda: [0, 0]), collections.defaultdict(lambda: [0, 0]), set()
    scores = []
    for tc in ET.parse(a.junit).getroot().iter("testcase"):
        props = {p.get("name"): p.get("value") for p in tc.iter("property")}
        if props.get("groundedness"):
            scores.append(float(props["groundedness"]))
        cid = props.get("case_id") or tc.get("name").split("[")[-1].rsplit("-t", 1)[0]
        ok = tc.find("failure") is None and tc.find("error") is None
        per_case[cid][0] += ok; per_case[cid][1] += 1
        cat = cid.split("-")[1]                         # GS-<CAT>-nnn
        per_cat[cat][0] += ok; per_cat[cat][1] += 1
        if props.get("severity") == "critical":
            critical.add(cid)
    failed = False
    if scores:
        mean, ones = sum(scores) / len(scores), sum(1 for s in scores if s <= 1)
        print(f"groundedness mean {mean:.2f} over {len(scores)} rows, score-1 rows: {ones}")
        failed |= mean < 4 or ones > 0
    for cid in sorted(critical):
        p, n = per_case[cid]
        if p < n:
            print(f"BLOCK critical {cid}: {p}/{n}"); failed = True
    for cat, (p, n) in sorted(per_cat.items()):
        lb = wilson_lower(p, n)
        flag = "BLOCK" if lb < a.threshold else "ok"
        failed |= flag == "BLOCK"
        print(f"{flag:5} {cat:8} {p}/{n} = {p / n:.1%}  lower bound {lb:.1%}")
    sys.exit(1 if failed else 0)
 
if __name__ == "__main__":
    main()

code/harness/calibrate_judge.py
"""python calibrate_judge.py labels.csv  — columns: case_id, human (pass|fail), judge (pass|fail), severity"""
import csv, sys
from sklearn.metrics import cohen_kappa_score, confusion_matrix
 
rows = list(csv.DictReader(open(sys.argv[1], encoding="utf-8")))
h = [r["human"] for r in rows]; j = [r["judge"] for r in rows]
k = cohen_kappa_score(h, j)
tn, fp, fn, tp = confusion_matrix(h, j, labels=["fail", "pass"]).ravel()   # fp = judge passed a human fail
crit_fp = sum(1 for r in rows if r["severity"] == "critical" and r["human"] == "fail" and r["judge"] == "pass")
print(f"n={len(rows)} kappa={k:.2f} agreement={(tp + tn) / len(rows):.1%} false_passes={fp} critical_false_passes={crit_fp}")
ok_advisory, ok_blocking = k >= 0.7, k >= 0.8 and crit_fp == 0
print("advisory use:", "OK" if ok_advisory else "NOT OK", "| may block releases alone:", "OK" if ok_blocking else "NOT OK")
sys.exit(0 if ok_advisory else 1)