What Is Agent Simulation Testing? A 3-Actor Harness

"""
harness.py

A reusable, provider-agnostic three-actor simulation harness for testing
conversational AI agents:

    - UserSimulatorAgent: an LLM playing a persona with a goal, generating
      the next user turn (or signaling the conversation is done).
    - AgentAdapter: a thin wrapper around ANY agent implementation, so the
      harness works regardless of the framework (LangGraph, CrewAI, a raw
      OpenAI Assistants thread, or hand-rolled code) sitting behind it.
    - JudgeAgent: an independent LLM call that scores the full transcript
      against an explicit, checkable rubric, defaulting to FAIL whenever
      the verdict is ambiguous.

`SimulationRunner` alternates user and agent turns until the simulator
signals it is done (or `max_turns` is reached), then always hands the full
transcript to the judge, even if the conversation ended early.

`LLMClient` is the only class in this module that touches the OpenAI SDK.
Swapping model providers means changing `LLMClient`, not the pattern.

Requires the OPENAI_API_KEY environment variable to be set. This module
has no __main__ block; it is imported by `order_cancellation_scenario.py`
and the pytest suite in `tests/test_non_determinism.py`.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any, Callable, Optional

from openai import OpenAI


@dataclass
class Message:
    """A single turn in a conversation."""

    role: str  # "user" | "assistant" | "system" | "tool"
    content: str


@dataclass
class JudgeVerdict:
    """The outcome of a JudgeAgent evaluating a transcript."""

    passed: bool
    reasoning: str


@dataclass
class SimulationResult:
    """The full output of a SimulationRunner.run() call."""

    transcript: list[Message]
    verdict: JudgeVerdict


class LLMClient:
    """
    Thin wrapper around the OpenAI chat completions API.

    This is the ONLY class in this module that touches the OpenAI SDK.
    Swap it for a different provider's client to port the whole harness to
    another model provider without touching any other class in this file.
    """

    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.7) -> None:
        api_key = os.environ.get("OPENAI_API_KEY")
        if not api_key:
            raise RuntimeError(
                "OPENAI_API_KEY is not set. Export it before running any "
                "code in this repository: export OPENAI_API_KEY=sk-..."
            )
        self._client = OpenAI(api_key=api_key)
        self.model = model
        self.temperature = temperature

    def complete(self, system_prompt: str, messages: list[Message]) -> str:
        """
        Send a system prompt plus a list of Message turns to the model and
        return the assistant's reply text.
        """
        api_messages: list[dict[str, str]] = [
            {"role": "system", "content": system_prompt}
        ]
        for msg in messages:
            api_messages.append({"role": msg.role, "content": msg.content})

        response = self._client.chat.completions.create(
            model=self.model,
            temperature=self.temperature,
            messages=api_messages,  # type: ignore[arg-type]
        )
        content = response.choices[0].message.content
        return content.strip() if content else ""


class UserSimulatorAgent:
    """
    An LLM playing a persona with a goal. Generates the next user message
    given the conversation so far, and decides on its own when the goal
    has been reached (or when to give up) by returning None instead of a
    string.
    """

    DONE_TOKEN = "<<SIMULATION_DONE>>"

    def __init__(self, persona: str, goal: str, llm: LLMClient) -> None:
        self.persona = persona
        self.goal = goal
        self.llm = llm

    def _system_prompt(self) -> str:
        return (
            "You are role-playing as a customer talking to a support agent. "
            f"Persona: {self.persona}\n"
            f"Goal: {self.goal}\n\n"
            "Speak ONLY as the customer, in first person, one message at a "
            "time. Do not narrate, and do not describe the agent's behavior "
            "from the outside, and do not break character.\n"
            f"If your goal has been fully reached, or you have decided to "
            f"give up, respond with exactly the token {self.DONE_TOKEN} and "
            "nothing else."
        )

    def next_message(self, history: list[Message]) -> Optional[str]:
        """
        Return the next user message given the conversation so far, or
        None if the goal has been reached (or the simulator has given up).
        """
        reply = self.llm.complete(self._system_prompt(), history)
        if self.DONE_TOKEN in reply:
            return None
        return reply


AgentFn = Callable[[list[Message]], "tuple[str, list[dict[str, Any]]]"]


class AgentAdapter:
    """
    Wraps an arbitrary agent implementation behind one interface: take the
    conversation history, return a reply and whatever tool calls were made.

    `agent_fn` can be backed by LangGraph, CrewAI, an OpenAI Assistants
    thread, or hand-written code. As long as it matches the signature
    `(history: list[Message]) -> (reply_text: str, tool_calls: list[dict])`
    it plugs into this harness unchanged.
    """

    def __init__(self, agent_fn: AgentFn) -> None:
        self.agent_fn = agent_fn

    def respond(self, history: list[Message]) -> "tuple[str, list[dict[str, Any]]]":
        return self.agent_fn(history)


class JudgeAgent:
    """
    An independent LLM call that scores a full transcript against an
    explicit, checkable rubric. Defaults to FAIL whenever the model's
    verdict is ambiguous or unparseable, rather than optimistically
    passing.
    """

    def __init__(self, criteria: str, llm: LLMClient) -> None:
        self.criteria = criteria
        self.llm = llm

    def _system_prompt(self) -> str:
        return (
            "You are a strict QA judge reviewing a conversation transcript "
            "between a simulated customer and a support agent.\n\n"
            f"Pass/fail criteria:\n{self.criteria}\n\n"
            "Respond with exactly two lines, in this exact format:\n"
            "VERDICT: PASS\n"
            "REASONING: <one or two sentences citing specific parts of the transcript>\n"
            "(or VERDICT: FAIL with the same REASONING line format)\n\n"
            "If the transcript is ambiguous with respect to the criteria, "
            "or evidence is missing, default to VERDICT: FAIL. Never guess "
            "in the agent's favor."
        )

    def evaluate(self, transcript: list[Message]) -> JudgeVerdict:
        rendered = "\n".join(f"{m.role.upper()}: {m.content}" for m in transcript)
        judge_input = [Message(role="user", content=f"Transcript:\n{rendered}")]
        raw = self.llm.complete(self._system_prompt(), judge_input)

        passed = False
        reasoning = raw.strip() or "Judge returned an empty response; defaulting to FAIL."

        for line in raw.splitlines():
            stripped = line.strip()
            if stripped.upper().startswith("VERDICT:"):
                verdict_text = stripped.split(":", 1)[1].strip().upper()
                passed = verdict_text.startswith("PASS")
            elif stripped.upper().startswith("REASONING:"):
                reasoning = stripped.split(":", 1)[1].strip()

        if "VERDICT:" not in raw.upper():
            # Model didn't follow the format at all; ambiguous -> FAIL.
            passed = False
            reasoning = (
                "Judge response did not include a parseable VERDICT line; "
                f"defaulting to FAIL. Raw response: {raw!r}"
            )

        return JudgeVerdict(passed=passed, reasoning=reasoning)


class SimulationRunner:
    """
    Ties the three actors together: alternates user and agent turns up to
    `max_turns`, stops early if the simulator signals the goal is reached
    (or given up on), and always hands the full transcript to the judge at
    the end, even if the conversation ended early.
    """

    def __init__(
        self,
        user_simulator: UserSimulatorAgent,
        agent_adapter: AgentAdapter,
        judge: JudgeAgent,
        max_turns: int = 8,
    ) -> None:
        self.user_simulator = user_simulator
        self.agent_adapter = agent_adapter
        self.judge = judge
        self.max_turns = max_turns

    def run(self) -> SimulationResult:
        transcript: list[Message] = []

        for _ in range(self.max_turns):
            user_text = self.user_simulator.next_message(transcript)
            if user_text is None:
                break
            transcript.append(Message(role="user", content=user_text))

            reply_text, _tool_calls = self.agent_adapter.respond(transcript)
            transcript.append(Message(role="assistant", content=reply_text))

        # Always judge whatever transcript exists, even a short/early one.
        verdict = self.judge.evaluate(transcript)
        return SimulationResult(transcript=transcript, verdict=verdict)

"""
order_cancellation_scenario.py

A worked simulation-testing scenario built on top of harness.py: a customer
wants order A1092 cancelled because the event it was for moved to a date
they can no longer attend. The `cancel_order` tool times out on its first
call and succeeds on the second (retry), and the toy agent under test has
to notice the failure, tell the user, retry, and confirm success naming
the order id and the word "cancelled".

Run standalone:

    python src/order_cancellation_scenario.py

Requires the OPENAI_API_KEY environment variable to be set.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

# Allow running this file directly (`python src/order_cancellation_scenario.py`)
# in addition to being imported from tests/test_non_determinism.py.
sys.path.insert(0, str(Path(__file__).resolve().parent))

from harness import (  # noqa: E402
    AgentAdapter,
    JudgeAgent,
    LLMClient,
    Message,
    SimulationResult,
    SimulationRunner,
    UserSimulatorAgent,
)


class ToolTimeoutError(Exception):
    """Raised by cancel_order() on its first call for a given order."""


def build_scenario() -> "tuple[dict[str, dict[str, str]], Any]":
    """
    Build a fresh, isolated instance of the order-cancellation scenario: an
    in-memory ORDERS table and a bound `cancel_order` tool whose first call
    for any order raises ToolTimeoutError and whose second call succeeds.

    Returns (orders, order_cancellation_agent). Each call to this function
    creates independent state (a fresh ORDERS dict and a fresh call
    counter), so callers that need to run the scenario multiple times in
    the same process (like the non-determinism pytest suite) get a clean
    slate every time instead of module-level globals that would only
    behave correctly once per process.
    """
    orders: dict[str, dict[str, str]] = {"A1092": {"status": "confirmed"}}
    call_counts: dict[str, int] = {}

    def cancel_order(order_id: str) -> str:
        call_counts[order_id] = call_counts.get(order_id, 0) + 1
        if call_counts[order_id] == 1:
            raise ToolTimeoutError(f"Timed out cancelling order {order_id}")
        orders[order_id]["status"] = "cancelled"
        return f"Order {order_id} cancelled."

    def order_cancellation_agent(
        history: list[Message],
    ) -> "tuple[str, list[dict[str, Any]]]":
        """
        Toy agent under test. Attempts to cancel order A1092: calls
        cancel_order, catches the timeout, tells the user the attempt
        failed and that it is retrying, retries, then confirms success
        naming the order id and the word "cancelled".
        """
        tool_calls: list[dict[str, Any]] = []
        order_id = "A1092"

        try:
            cancel_order(order_id)
            tool_calls.append(
                {"name": "cancel_order", "args": {"order_id": order_id}, "result": "ok"}
            )
            reply = f"Order {order_id} has been cancelled."
        except ToolTimeoutError as exc:
            tool_calls.append(
                {"name": "cancel_order", "args": {"order_id": order_id}, "error": str(exc)}
            )
            # Acknowledge the failure to the user before retrying.
            acknowledgment = (
                f"I tried to cancel order {order_id} but the request timed "
                "out. Retrying now."
            )
            try:
                cancel_order(order_id)
                tool_calls.append(
                    {"name": "cancel_order", "args": {"order_id": order_id}, "result": "ok"}
                )
                reply = (
                    f"{acknowledgment} Good news: order {order_id} has now "
                    "been cancelled."
                )
            except ToolTimeoutError as exc2:
                tool_calls.append(
                    {"name": "cancel_order", "args": {"order_id": order_id}, "error": str(exc2)}
                )
                reply = (
                    f"{acknowledgment} Unfortunately the retry also failed. "
                    "I've escalated this to a human agent."
                )

        return reply, tool_calls

    return orders, order_cancellation_agent


def run_scenario() -> "tuple[SimulationResult, dict[str, dict[str, str]]]":
    """
    Wire up the three actors around a fresh scenario instance and run one
    full simulation. Returns (SimulationResult, orders) so callers can make
    both the judge-based assertion and a deterministic final-state
    assertion.
    """
    orders, order_cancellation_agent = build_scenario()

    llm = LLMClient()

    user_simulator = UserSimulatorAgent(
        persona=(
            "A customer who booked order A1092 for an event that has since "
            "moved to a new date they can no longer attend. Polite but "
            "persistent. If the agent goes quiet or seems to have given up "
            "without explanation, push back once by asking directly "
            "whether the cancellation actually went through."
        ),
        goal=(
            "Get order A1092 cancelled and receive explicit confirmation "
            "that it was cancelled."
        ),
        llm=llm,
    )

    agent_adapter = AgentAdapter(order_cancellation_agent)

    judge = JudgeAgent(
        criteria=(
            "PASS only if ALL of the following are true:\n"
            "1. By the end of the conversation, order A1092's status is "
            "'cancelled' (the agent reported the cancellation succeeded).\n"
            "2. The transcript shows the agent acknowledging that the "
            "first cancellation attempt failed BEFORE it reports retrying "
            "or succeeding.\n"
            "3. The agent never silently repeats the same failing action "
            "more than twice without telling the user about the failure.\n"
            "FAIL if any of these do not clearly hold, or if the evidence "
            "is ambiguous."
        ),
        llm=llm,
    )

    runner = SimulationRunner(
        user_simulator=user_simulator,
        agent_adapter=agent_adapter,
        judge=judge,
        max_turns=8,
    )
    result = runner.run()
    return result, orders


def _acknowledgment_precedes_confirmation(transcript: list[Message]) -> bool:
    """
    Deterministic recovery check, independent of the judge: scan the
    transcript for an assistant message that mentions a failure/retry
    BEFORE any assistant message that confirms cancellation. This is the
    check that catches an agent silently retrying and only reporting
    success, which a judge could in principle miss.
    """
    ack_keywords = ("timed out", "failed", "retry", "retrying")
    confirm_keywords = ("cancelled", "canceled")

    ack_seen = False
    for msg in transcript:
        if msg.role != "assistant":
            continue
        lowered = msg.content.lower()
        if not ack_seen and any(kw in lowered for kw in ack_keywords):
            ack_seen = True
        if ack_seen and any(kw in lowered for kw in confirm_keywords):
            return True
    return False


def main() -> None:
    result, orders = run_scenario()

    print("=== Transcript ===")
    for msg in result.transcript:
        print(f"{msg.role.upper()}: {msg.content}")

    print("\n=== Verdict ===")
    print(f"Passed: {result.verdict.passed}")
    print(f"Reasoning: {result.verdict.reasoning}")
    print(f"\nFinal order state: {orders}")

    assert result.verdict.passed, f"Judge failed the scenario: {result.verdict.reasoning}"
    assert orders["A1092"]["status"] == "cancelled", "Order A1092 was not cancelled"
    assert _acknowledgment_precedes_confirmation(result.transcript), (
        "Transcript did not show an acknowledgment of the failed attempt "
        "before the confirmation message"
    )
    print("\nAll assertions passed.")


if __name__ == "__main__":
    main()

"""
tests/test_non_determinism.py

Every actor in the simulation harness is an LLM call, which means a single
run is not a reliable pass/fail signal. This suite demonstrates the two
disciplines for handling that non-determinism in CI:

1. Run-N-times threshold gating (test_recovery_passes_at_threshold): run
   the order-cancellation scenario multiple times and require a pass rate
   above a threshold, rather than trusting any single run.
2. Semantic-similarity assertions (test_confirmation_message_is_semantically_close):
   for prose the model is free to phrase differently each run, assert on
   meaning (embedding cosine similarity) instead of exact string equality.
   Exact-match is reserved for hard facts (the order id, the word
   "cancelled"), which order_cancellation_scenario.py already checks
   deterministically.

These tests make real OpenAI API calls (a chat completion per actor per
turn, plus an embeddings call for the semantic-similarity test) and are
therefore slower and more expensive than typical unit tests. Requires
OPENAI_API_KEY to be set. Run with:

    pytest tests/test_non_determinism.py -v
"""

from __future__ import annotations

import math
import os
import sys
from pathlib import Path

import pytest
from openai import OpenAI

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))

from order_cancellation_scenario import run_scenario  # noqa: E402

requires_openai_key = pytest.mark.skipif(
    "OPENAI_API_KEY" not in os.environ,
    reason="Requires OPENAI_API_KEY (these tests make real, billed API calls).",
)


def run_scenario_n_times(n: int) -> list:
    """Rerun the order-cancellation scenario n independent times."""
    outcomes = []
    for _ in range(n):
        result, orders = run_scenario()
        outcomes.append((result, orders))
    return outcomes


def semantic_similarity(a: str, b: str) -> float:
    """
    Cosine similarity between the OpenAI text-embedding-3-small embeddings
    of two strings. Returns a value in roughly [-1, 1]; in practice,
    semantically related sentences score well above 0.8.
    """
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=[a, b],
    )
    vec_a = response.data[0].embedding
    vec_b = response.data[1].embedding

    dot = sum(x * y for x, y in zip(vec_a, vec_b))
    norm_a = math.sqrt(sum(x * x for x in vec_a))
    norm_b = math.sqrt(sum(y * y for y in vec_b))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot / (norm_a * norm_b)


@requires_openai_key
def test_recovery_passes_at_threshold() -> None:
    """
    Run the scenario 5 times and require at least 4/5 passes.

    Practitioner rule: if the pass rate is inconsistent run-to-run, the
    right fix is almost always to tighten the user simulator's persona and
    goal, and the judge's criteria, FIRST, not to loosen this threshold
    further. Inconsistent results usually mean the prompt is ambiguous, not
    that the agent under test is randomly broken. Loosening the threshold
    hides that signal instead of fixing it.
    """
    n = 5
    threshold = 4

    outcomes = run_scenario_n_times(n)
    pass_count = sum(1 for result, _orders in outcomes if result.verdict.passed)
    failures = [
        result.verdict.reasoning for result, _orders in outcomes if not result.verdict.passed
    ]

    assert pass_count >= threshold, (
        f"Only {pass_count}/{n} runs passed (need >= {threshold}). "
        f"Judge reasoning on failing runs: {failures}"
    )


@requires_openai_key
def test_confirmation_message_is_semantically_close() -> None:
    """
    Assert the agent's final confirmation message is semantically close to
    a reference sentence, rather than requiring an exact string match.

    A naive `final_message == "Your order A1092 has been cancelled."` would
    flake constantly: "I've gone ahead and cancelled order A1092 for you"
    and "Order A1092 has now been cancelled" are both correct confirmations
    but neither is byte-identical to the reference. Exact-match is the
    right tool for hard facts that must appear verbatim (the order id
    "A1092", the word "cancelled"), not for full sentences the model is
    free to phrase differently.
    """
    result, _orders = run_scenario()

    assistant_messages = [msg.content for msg in result.transcript if msg.role == "assistant"]
    assert assistant_messages, "Scenario produced no assistant messages"
    final_message = assistant_messages[-1]

    reference = "Your order A1092 has been cancelled."
    similarity = semantic_similarity(final_message, reference)

    assert similarity > 0.8, (
        f"Final confirmation message was not semantically close enough to "
        f"the reference (similarity={similarity:.3f}). "
        f"Message: {final_message!r}"
    )