From 7210cd8de6a3a849c6863e818b371326ec6c038c Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Fri, 16 Oct 2026 14:07:15 -0700 Subject: [PATCH 1/7] feat(verifiers): give verifiers the run's conversation and trigger firing logs `run_context.load_run_context` reads what a run recorded about the prompt a verifier grades into a `RunContext`: the stored conversation, every user-side turn included (the prompt, then the user-sim's or a human's messages), each user message marked with the agent triggers that wrote it and the extra fields the user-sim returned; each env's gateway firing log, from the envelope prompt_agent captures; and the user agent's own firing log. What the run meant to record but can't be read is named in `unavailable`, so a missing log is not read as an empty one. `RunContext.as_of_turn(k)` cuts it the way `grade_up_to_turn` cuts the trajectory: the messages up to turn k's reply and the events from before turn k+1's message. rubrics_verifier: `run_context_for_judge` (["conversation", "trigger_log"]) appends a Conversation and a Trigger firings section to the judge prompt, cut at `grade_up_to_turn`; messages and logs are bounded. env_outcome_verifier: a script whose `verify` takes two parameters is called `verify(mcp_url, run)`; `prompt_id` picks the prompt, the run's last one by default. A one-parameter `verify` is called as before. Co-Authored-By: Claude Opus 5.5 --- .../verifiers/env_outcome_verifier.py | 26 +- .../task_steps/verifiers/rubrics_verifier.py | 77 ++++- .../task_steps/verifiers/run_context.py | 136 ++++++++ .../test_verifier_run_context_local.py | 38 +++ .../task_step/verifiers/test_run_context.py | 317 ++++++++++++++++++ 5 files changed, 591 insertions(+), 3 deletions(-) create mode 100644 src/agent_env/task_step/task_steps/verifiers/run_context.py create mode 100644 tst/integration/task_step/test_verifier_run_context_local.py create mode 100644 tst/unit/task_step/verifiers/test_run_context.py diff --git a/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py b/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py index e63cb64e..938b6ded 100644 --- a/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py @@ -4,15 +4,17 @@ import asyncio import importlib.util +import inspect import logging import os import tempfile import uuid from typing import ClassVar, Optional -from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.context import PromptResponse, TaskStepContext from agent_env.entity_refs import EntityRef from agent_env.task_step.task_step import TaskStep, TaskStepDependency +from agent_env.task_step.task_steps.verifiers.run_context import load_run_context from agent_env.task_step.task_steps.verifiers.scoring import ScoreAggregator, aggregate_score __all__ = ["EnvOutcomeVerifierTaskStep", "ScoreAggregator", "aggregate_score"] @@ -22,6 +24,10 @@ class EnvOutcomeVerifierTaskStep(TaskStep): + """Runs the artifact's ``async verify(mcp_url)`` against the deployed env and scores what it returns. A + ``verify(mcp_url, run)`` is also handed the run it grades (``run_context.RunContext``): the conversation of + ``prompt_id``'s prompt, the run's last prompt when unset, and the trigger firing logs.""" + type: ClassVar[str] = "env_outcome_verifier" entity_refs = ( EntityRef.env("env_id"), @@ -37,11 +43,13 @@ def __init__( file_artifact_version: Optional[int] = None, score_aggregator: Optional[ScoreAggregator] = None, verifier_id: Optional[str] = None, + prompt_id: Optional[str] = None, depends_on: Optional[list[TaskStepDependency]] = None, fail_task_on_error: bool = True, ): super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) self.verifier_id = verifier_id or uuid.uuid4().hex + self.prompt_id = prompt_id self.env_id = env_id self.file_artifact_id = file_artifact_id self.file_artifact_version = file_artifact_version @@ -56,6 +64,8 @@ def to_dict(self) -> dict: base["file_artifact_id"] = self.file_artifact_id base["file_artifact_version"] = self.file_artifact_version base["score_aggregator"] = self.score_aggregator.value + if self.prompt_id is not None: + base["prompt_id"] = self.prompt_id return base @classmethod @@ -87,6 +97,7 @@ def from_dict(cls, data: dict) -> EnvOutcomeVerifierTaskStep: file_artifact_version=data.get("file_artifact_version"), score_aggregator=ScoreAggregator(raw) if raw else None, verifier_id=data.get("verifier_id"), + prompt_id=data.get("prompt_id"), ) async def execute(self, context: TaskStepContext) -> TaskStepContext: @@ -110,7 +121,10 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: spec.loader.exec_module(module) logger.info(f"Running verify() from artifact '{self.file_artifact_id}' against {deployed.mcp_url}") - results = await module.verify(deployed.mcp_url) + if len(inspect.signature(module.verify).parameters) >= 2: + results = await module.verify(deployed.mcp_url, await load_run_context(context, self._graded_prompt(context))) + else: + results = await module.verify(deployed.mcp_url) score = aggregate_score(results, self.score_aggregator) if "verifications" not in context.metadata: @@ -125,3 +139,11 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: os.unlink(tmp_path) return context + + def _graded_prompt(self, context: TaskStepContext) -> Optional[PromptResponse]: + if self.prompt_id is None: + return context.prompt_responses[-1] if context.prompt_responses else None + prompt_response = next((pr for pr in context.prompt_responses if pr.prompt_id == self.prompt_id), None) + if prompt_response is None: + raise RuntimeError(f"PromptResponse with prompt_id='{self.prompt_id}' not found in context") + return prompt_response diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py index 9a04ac0e..bb8f23e9 100644 --- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py @@ -35,6 +35,7 @@ ) from agent_env.entity_refs import EntityRef from agent_env.task_step.task_step import TaskStep, TaskStepDependency +from agent_env.task_step.task_steps.verifiers.run_context import RunContext, load_run_context from agent_env.task_step.task_steps.verifiers.scoring import ScoreAggregator, aggregate_score from agent_env.task_step.task_steps.sandbox_utils.sandbox_utils import ( agent_error_text, @@ -169,6 +170,62 @@ def resolve_context_facts(facts: dict[str, str], metadata: dict) -> list[tuple[s return rendered +# What of the run beyond its trajectory `run_context_for_judge` may show the judge, and the bounds on how much. +RUN_CONTEXT_PARTS = ("conversation", "trigger_log") +RUN_CONTEXT_MESSAGE_MAX_CHARS = 4000 # a longer message is cut and ends in "…" +RUN_CONTEXT_EVENTS_MAX = 200 # per firing log, the most recent; the rest are counted + + +def _validated_run_context_parts(parts: Optional[list[str]]) -> Optional[list[str]]: + if parts is None: + return None + if not isinstance(parts, list) or not parts or any(p not in RUN_CONTEXT_PARTS for p in parts): + raise ValueError(f"run_context_for_judge must be a non-empty list of {list(RUN_CONTEXT_PARTS)}, got {parts!r}") + return list(dict.fromkeys(parts)) + + +def _cut(text: str, limit: int) -> str: + return text if len(text) <= limit else text[:limit] + "…" + + +def render_run_context(run: RunContext, parts: list[str]) -> str: + """The judge-prompt sections for ``parts`` of ``run``: the conversation turn by turn, and each firing log as one + compact JSON line per event, the most recent ``RUN_CONTEXT_EVENTS_MAX`` of each.""" + sections: list[str] = [] + if "conversation" in parts: + lines = ["## Conversation", "The run's messages, turn by turn: the user side (the task's prompt, then each " + "later message of a user simulator, a human or the triggers) and the agent's replies."] + if "conversation" in run.unavailable: + lines.append(f"(not available: {run.unavailable['conversation']})") + for turn in range(1, run.turns + 1): + lines.append(f"### Turn {turn}") + for m in (m for m in run.conversation if m["turn"] == turn): + who = ", ".join(filter(None, [m.get("author"), + f"written by triggers {m['fired']}" if m.get("fired") else None])) + files = sum(1 for p in m["parts"] if p.get("kind") != "text") + extra = f" (+{files} non-text part{'s' if files != 1 else ''})" if files else "" + lines.append(f"- {m['role']}{f' ({who})' if who else ''}: " + f"{_cut(m['text'], RUN_CONTEXT_MESSAGE_MAX_CHARS)}{extra}") + sections.append("\n".join(lines)) + if "trigger_log" in parts: + lines = ["## Trigger firings", "What the env's triggers and the user agent's triggers did during the run, " + "oldest first, one firing-log event per line."] + logs = [(f"env {env_id}", events) for env_id, events in run.env_trigger_events.items()] + logs += [("user agent", run.agent_trigger_events)] if run.agent_trigger_events else [] + for name, events in logs: + lines.append(f"### {name}") + if len(events) > RUN_CONTEXT_EVENTS_MAX: + lines.append(f"({len(events) - RUN_CONTEXT_EVENTS_MAX} earlier events not shown)") + lines += [f"- {json.dumps({k: v for k, v in e.items() if k != 'seq'}, default=str)}" + for e in events[-RUN_CONTEXT_EVENTS_MAX:]] + lines += [f"(not available: {what}: {why})" for what, why in run.unavailable.items() + if what.startswith("env_trigger_events.")] + if len(lines) == 2: + lines.append("(no trigger firings were recorded)") + sections.append("\n".join(lines)) + return "\n\n".join(sections) + + # Only fall back from structured output when the error clearly says the response # format itself is unsupported. Image/provider/transient failures should retry as-is. _SCHEMA_TERMS = ( @@ -340,10 +397,14 @@ def __init__( grading_policy_prompt: Optional[str] = None, context_facts_for_judge: Optional[dict[str, str]] = None, grade_up_to_turn: Optional[int] = None, + run_context_for_judge: Optional[list[str]] = None, depends_on: Optional[list[TaskStepDependency]] = None, fail_task_on_error: bool = True, ): super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) + # What of the run beyond its trajectory the judge also sees (RUN_CONTEXT_PARTS): the conversation with every + # user-side turn, and the trigger firing logs; cut at grade_up_to_turn like the trajectory. + self.run_context_for_judge = _validated_run_context_parts(run_context_for_judge) # Grade a multi-turn run as it stood after this (1-based) turn: the judge sees turns 1..k only, # so a per-turn env snapshot can be scored on what the agent had done by then. None grades all. self.grade_up_to_turn = _validated_turn(grade_up_to_turn) @@ -461,6 +522,8 @@ def to_dict(self) -> dict: base["context_facts_for_judge"] = self.context_facts_for_judge if self.grade_up_to_turn is not None: base["grade_up_to_turn"] = self.grade_up_to_turn + if self.run_context_for_judge is not None: + base["run_context_for_judge"] = self.run_context_for_judge return base @classmethod @@ -490,6 +553,7 @@ def from_dict(cls, data: dict) -> RubricsVerifierTaskStep: grading_policy_prompt=data.get("grading_policy_prompt"), context_facts_for_judge=data.get("context_facts_for_judge"), grade_up_to_turn=data.get("grade_up_to_turn"), + run_context_for_judge=data.get("run_context_for_judge"), ) def _resolved_criteria(self, context: TaskStepContext) -> list[dict]: @@ -525,6 +589,11 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: self._write_skipped_entries( context, "prompt_error", f"Skipped: prompt had error_type={prompt_response.error_type}") return context + run_context = None + if self.run_context_for_judge: + run_context = await load_run_context(context, prompt_response) + if grade_up_to_turn is not None: + run_context = run_context.as_of_turn(grade_up_to_turn) per_turn_uris = [u for u in (prompt_response.target_agent_per_turn_trajectory_object_urls or []) if u] if self.use_trajectory and not (per_turn_uris or prompt_response.agent_trajectory_object_url): @@ -692,6 +761,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: trajectory_path=trajectory_path, loaded_artifacts=loaded_for_judge, context_facts=context_facts, + run_context=run_context, ) if self.use_trajectory and multiturn: eval_prompt += self._MULTITURN_DIR_NOTE.format(n=len(per_turn_uris)) @@ -712,6 +782,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: image_blocks=image_blocks, per_criterion_frames=compacted.per_criterion, context_facts=context_facts, + run_context=run_context, ) if image_blocks: logger.info( @@ -1285,9 +1356,11 @@ def _build_eval_prompt( image_blocks: Optional[list[dict]] = None, per_criterion_frames: bool = False, context_facts: Optional[list[tuple[str, str]]] = None, + run_context: Optional[RunContext] = None, ) -> str: """``context_facts`` are rendered ``(label, value)`` pairs (``resolve_context_facts``); ``image_blocks`` - are the blocks attached to the judge message, labels included when ``per_criterion_frames``.""" + are the blocks attached to the judge message, labels included when ``per_criterion_frames``; ``run_context`` + is shown as ``run_context_for_judge`` asks.""" spec = self._spec() # Only the rubric_evidence templates have this placeholder; the others ignore the argument. checks_instruction = evidence_checks_instruction(has_policy=bool(self.grading_policy_prompt)) @@ -1316,6 +1389,8 @@ def _build_eval_prompt( ) if loaded_artifacts: base += "\n\n" + self._build_loaded_artifacts_block(loaded_artifacts) + if run_context is not None and self.run_context_for_judge: + base += "\n\n" + render_run_context(run_context, self.run_context_for_judge) if context_facts: # Facts before the policy: they are inputs, the policy is how to read them. base += "\n\n## Context facts\n" + self._CONTEXT_FACTS_LEAD + "\n" + "\n".join( diff --git a/src/agent_env/task_step/task_steps/verifiers/run_context.py b/src/agent_env/task_step/task_steps/verifiers/run_context.py new file mode 100644 index 00000000..dd845239 --- /dev/null +++ b/src/agent_env/task_step/task_steps/verifiers/run_context.py @@ -0,0 +1,136 @@ +"""The run a verifier grades, beyond the env's state: the conversation, every user-side turn included, and the +trigger firing logs. + +``load_run_context`` reads it from what the run recorded: the conversation ``prompt_agent`` stored for the prompt +(``metadata["a2a_conversations"]``), the env firing logs it captured at the prompt's end +(``metadata["env_trigger_state"]``), and the user agent's own firing log (``metadata["agent_trigger_state"]``). +``RunContext.as_of_turn`` cuts it the way ``grade_up_to_turn`` cuts the trajectory. +""" +from __future__ import annotations + +import asyncio +import json +from dataclasses import dataclass, field, replace +from datetime import datetime +from typing import Any, Optional + +from agent_env.a2a_agent import conversation_store +from agent_env.config import get_config +from agent_env.task_step.context import PromptResponse, TaskStepContext + + +@dataclass(frozen=True) +class RunContext: + """``conversation``: the prompt's messages, oldest first, each ``{"turn", "role", "text", "parts", "ts"}`` with + ``author`` (who sent it, when recorded), and on a user message the agent triggers that wrote it (``fired``) and + the extra fields the user-sim returned with it (``fields``). ``env_trigger_events``: each env's gateway firing + log, oldest first. ``agent_trigger_events``: the user agent's firing log. ``unavailable`` names what the run + meant to record but could not be read, and why, so a verifier can tell a missing log from an empty one.""" + + conversation: list[dict] = field(default_factory=list) + env_trigger_events: dict[str, list[dict]] = field(default_factory=dict) + agent_trigger_events: list[dict] = field(default_factory=list) + unavailable: dict[str, str] = field(default_factory=dict) + + @property + def turns(self) -> int: + return max((m["turn"] for m in self.conversation), default=0) + + def as_of_turn(self, turn: int) -> RunContext: + """The run as it stood after ``turn``: its messages up to that turn's reply, and the firing-log events from + before the next turn's user message (all of them when no turn follows).""" + if turn >= self.turns: + return self + cutoff = next(m["ts"] for m in self.conversation if m["turn"] == turn + 1) + + def before(events: list[dict]) -> list[dict]: + return [e for e in events if _before(e.get("ts"), cutoff)] + + return replace( + self, + conversation=[m for m in self.conversation if m["turn"] <= turn], + env_trigger_events={env_id: before(events) for env_id, events in self.env_trigger_events.items()}, + agent_trigger_events=before(self.agent_trigger_events), + ) + + def to_dict(self) -> dict[str, Any]: + return {"conversation": self.conversation, "env_trigger_events": self.env_trigger_events, + "agent_trigger_events": self.agent_trigger_events, "unavailable": self.unavailable} + + +async def load_run_context(context: TaskStepContext, prompt_response: Optional[PromptResponse]) -> RunContext: + """The run context of ``prompt_response``'s prompt (the env firing logs alone without one). Never raises for a + record it cannot read; that lands in ``unavailable``.""" + unavailable: dict[str, str] = {} + conversation: list[dict] = [] + agent_events: list[dict] = [] + if prompt_response is not None: + conversation = await asyncio.to_thread(_conversation, context, prompt_response, unavailable) + agent_events = list((context.metadata.get("agent_trigger_state") or {}).get(prompt_response.step_id) or []) + env_events = await asyncio.to_thread(_env_trigger_events, context, unavailable) + return RunContext(conversation=conversation, env_trigger_events=env_events, agent_trigger_events=agent_events, + unavailable=unavailable) + + +def _conversation(context: TaskStepContext, prompt_response: PromptResponse, unavailable: dict[str, str]) -> list[dict]: + """The stored conversation, numbered by user message; a prompt with none recorded (an episode's turn) is one turn + of its prompt and reply.""" + conversation_id = (context.metadata.get("a2a_conversations") or {}).get(prompt_response.step_id) + if conversation_id is None: + return [_message(1, "user", [{"kind": "text", "text": prompt_response.prompt_text or ""}]), + _message(1, "agent", [{"kind": "text", "text": prompt_response.response}])] + try: + doc = conversation_store.get_conversation(conversation_id) + except Exception as e: + unavailable["conversation"] = f"{type(e).__name__}: {str(e)[:200]}" + return [] + if doc is None: + unavailable["conversation"] = f"conversation {conversation_id!r} is not in the store" + return [] + step_id = prompt_response.step_id + fired = {f["turn"] + 1: f.get("fired") for f in (context.metadata.get("agent_trigger_firings") or {}).get(step_id) or [] + if f.get("fired")} + fields = {f["turn"] + 1: f.get("fields") for f in (context.metadata.get("usersim_turn_outputs") or {}).get(step_id) or []} + authors = {"user": doc.get("source_agent_name"), "agent": doc.get("target_agent_name")} + messages: list[dict] = [] + turn = 0 + for stored in doc.get("messages") or []: + role = stored.get("role") + if role == "user": + turn += 1 + message = _message(turn, role, stored.get("parts") or [], ts=stored.get("ts"), author=authors.get(role)) + if role == "user" and turn in fired: + message["fired"] = fired[turn] + if role == "user" and turn in fields: + message["fields"] = fields[turn] + messages.append(message) + return messages + + +def _message(turn: int, role: str, parts: list[dict], *, ts: Optional[str] = None, + author: Optional[str] = None) -> dict: + text = "\n".join(p.get("text", "") for p in parts if p.get("kind") == "text") + return {"turn": turn, "role": role, **({"author": author} if author else {}), "text": text, "parts": parts, + "ts": ts} + + +def _env_trigger_events(context: TaskStepContext, unavailable: dict[str, str]) -> dict[str, list[dict]]: + """Each env's firing log, from the envelope ``prompt_agent`` stored at the prompt's end.""" + events: dict[str, list[dict]] = {} + for env_id, entry in (context.metadata.get("env_trigger_state") or {}).items(): + if not isinstance(entry, dict) or not entry.get("object_url"): + unavailable[f"env_trigger_events.{env_id}"] = (entry or {}).get("error") or "no firing log was captured" + continue + try: + envelope = json.loads(get_config().get_object_store().get(entry["object_url"])) + except Exception as e: + unavailable[f"env_trigger_events.{env_id}"] = f"{type(e).__name__}: {str(e)[:200]}" + continue + events[env_id] = list((envelope.get("state") or {}).get("events") or []) + return events + + +def _before(ts: Optional[str], cutoff: Optional[str]) -> bool: + if ts is None or cutoff is None: + return True + return datetime.fromisoformat(ts) < datetime.fromisoformat(cutoff) diff --git a/tst/integration/task_step/test_verifier_run_context_local.py b/tst/integration/task_step/test_verifier_run_context_local.py new file mode 100644 index 00000000..33362800 --- /dev/null +++ b/tst/integration/task_step/test_verifier_run_context_local.py @@ -0,0 +1,38 @@ +"""A verifier is handed the run it grades: after a real two-turn conversation on the local defaults, with the echo +agent (tst/data/a2a_agent) as both solver and user-sim, the run context reads back every turn prompt_agent stored, the +user-sim's included. Needs Docker and a throwaway local registry.""" + +import uuid + +import pytest + +from agent_env.task import Task +from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep +from agent_env.task_step.task_steps.verifiers.run_context import load_run_context +from tst.integration.task_step.test_prompt_agent_file_parts_local import _deploy, local_registry as local_registry +from tst.util.a2a_test_agent import put_test_agent + +pytestmark = pytest.mark.integration + + +@pytest.mark.asyncio +async def test_the_run_context_reads_back_every_turn_of_a_real_conversation(local_registry): + suffix = uuid.uuid4().hex[:8] + agent = put_test_agent(f"run-context-agent-{suffix}") + task = Task.put(id=f"run-context-{suffix}", steps=[ + _deploy(agent, "solver"), + _deploy(agent, "user"), + PromptAgentTaskStep(id="ask", version=None, prompt_id="ask", agent_name="solver", user_agent_name="user", + max_conversation_turns=2, poll_interval_seconds=1, prompt="Plan a lunch for Friday."), + ]) + + context = await task.run() + assert not context.metadata.get("failed_steps"), context.metadata.get("failed_steps") + run = await load_run_context(context, next(pr for pr in context.prompt_responses if pr.prompt_id == "ask")) + + assert run.unavailable == {} and run.turns == 2 + assert [(m["turn"], m["role"], m.get("author")) for m in run.conversation] == [ + (1, "user", "user"), (1, "agent", "solver"), (2, "user", "user"), (2, "agent", "solver")] + assert run.conversation[0]["text"] == "Plan a lunch for Friday." + assert run.conversation[2]["text"] and run.conversation[2]["text"] != run.conversation[0]["text"] + assert run.as_of_turn(1).conversation == run.conversation[:2] diff --git a/tst/unit/task_step/verifiers/test_run_context.py b/tst/unit/task_step/verifiers/test_run_context.py new file mode 100644 index 00000000..14cbade1 --- /dev/null +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -0,0 +1,317 @@ +"""The run a verifier grades, read from what the run recorded: the stored conversation with every user-side turn, +the env firing logs prompt_agent captured, and the user agent's own firing log; cut as of a turn like the trajectory.""" +from __future__ import annotations + +import json +import time +import uuid +from datetime import datetime + +from types import SimpleNamespace + +import pytest + +from agent_env.a2a_agent import conversation_store +from agent_env.artifact import FileArtifact +from agent_env.config import get_config +from agent_env.task_step.context import PromptResponse, TaskStepContext +from agent_env.task_step.task_steps.verifiers.env_outcome_verifier import EnvOutcomeVerifierTaskStep +from agent_env.task_step.task_steps.verifiers.rubrics_verifier import ( + RUN_CONTEXT_EVENTS_MAX, + RUN_CONTEXT_MESSAGE_MAX_CHARS, + RubricsVerifierTaskStep, + render_run_context, +) +from agent_env.task_step.task_steps.verifiers.run_context import RunContext, load_run_context + +TURNS = [("Can you book lunch with Dana?", "Booked for noon."), + ("Make it 12:30 instead.", "Moved to 12:30."), + ("Thanks, that's all.", "Glad to help.")] + + +def _text(text: str) -> list[dict]: + return [{"kind": "text", "text": text}] + + +def _between(earlier: str, later: str) -> str: + a, b = datetime.fromisoformat(earlier), datetime.fromisoformat(later) + return (a + (b - a) / 2).isoformat() + + +def recorded_run(turns=TURNS) -> tuple[TaskStepContext, PromptResponse, dict]: + """A three-turn run as prompt_agent records it: the conversation in the store, which agent trigger wrote turn 2, + the user-sim's extra fields on turn 3, the env firing log in the object store and the user agent's firing log.""" + conversation_id = f"conv-{uuid.uuid4().hex[:8]}" + conversation_store.create_conversation(conversation_id, task_instance_id="run-1", source_agent_name="dana_sim", + target_agent_name="solver") + for user, agent in turns: + conversation_store.add_a2a_task(conversation_id, parts=_text(user), a2a_task_id=uuid.uuid4().hex, role="user") + time.sleep(0.005) # distinct message stamps, so events can fall between them + conversation_store.complete_a2a_task(conversation_id, parts=_text(agent), role="agent") + time.sleep(0.005) + messages = conversation_store.get_conversation(conversation_id)["messages"] + turn_starts = [m["ts"] for m in messages if m["role"] == "user"] + after_reply_1 = _between(messages[1]["ts"], turn_starts[1]) + env_events = [ + {"seq": 1, "ts": after_reply_1, "kind": "detected", "trigger_id": "confirm"}, + {"seq": 2, "ts": after_reply_1, "kind": "fired", "trigger_id": "confirm", "fire_count": 1}, + {"seq": 3, "ts": _between(turn_starts[2], messages[5]["ts"]), "kind": "fired", "trigger_id": "late", + "fire_count": 1}, + ] + envelope = {"env_id": "calendar", "state": {"triggers": [], "events": env_events}} + object_url = get_config().get_object_store().put(f"env_trigger_state/{conversation_id}.json", + json.dumps(envelope).encode(), content_type="application/json") + agent_events = [{"seq": 1, "ts": after_reply_1, "kind": "fired", "trigger_id": "reschedule"}] + context = TaskStepContext(metadata={ + "a2a_conversations": {"prompt": conversation_id}, + "agent_trigger_firings": {"prompt": [{"turn": 1, "fired": ["reschedule"]}, {"turn": 2, "fired": []}]}, + "usersim_turn_outputs": {"prompt": [{"turn": 2, "fields": {"mood": "satisfied"}}]}, + "env_trigger_state": {"calendar": {"object_url": object_url, "capture_step_id": "prompt"}}, + "agent_trigger_state": {"prompt": agent_events}, + }) + prompt_response = PromptResponse(prompt_id="p1", response="Glad to help.", prompt_text=TURNS[0][0], step_id="prompt") + context.prompt_responses.append(prompt_response) + return context, prompt_response, {"env": env_events, "agent": agent_events, "turn_starts": turn_starts} + + +@pytest.mark.asyncio +async def test_the_run_context_has_every_turn_and_both_firing_logs(): + context, prompt_response, recorded = recorded_run() + + run = await load_run_context(context, prompt_response) + + assert run.turns == 3 and run.unavailable == {} + assert [(m["turn"], m["role"], m.get("author"), m["text"]) for m in run.conversation] == [ + (1, "user", "dana_sim", TURNS[0][0]), (1, "agent", "solver", TURNS[0][1]), + (2, "user", "dana_sim", TURNS[1][0]), (2, "agent", "solver", TURNS[1][1]), + (3, "user", "dana_sim", TURNS[2][0]), (3, "agent", "solver", TURNS[2][1]), + ] + user = [m for m in run.conversation if m["role"] == "user"] + assert "fired" not in user[0] and user[1]["fired"] == ["reschedule"] and "fired" not in user[2] + assert user[2]["fields"] == {"mood": "satisfied"} and "fields" not in user[1] + assert run.env_trigger_events == {"calendar": recorded["env"]} + assert run.agent_trigger_events == recorded["agent"] + assert json.loads(json.dumps(run.to_dict()))["conversation"][0]["parts"] == _text(TURNS[0][0]) + + +@pytest.mark.asyncio +async def test_as_of_a_turn_the_run_stops_before_the_next_turn_begins(): + context, prompt_response, recorded = recorded_run() + run = await load_run_context(context, prompt_response) + + first, second = run.as_of_turn(1), run.as_of_turn(2) + + assert [m["turn"] for m in first.conversation] == [1, 1] + assert first.env_trigger_events == {"calendar": recorded["env"][:2]} + assert first.agent_trigger_events == recorded["agent"] # fired just before turn 2's message: it wrote it + assert [m["turn"] for m in second.conversation] == [1, 1, 2, 2] + assert second.env_trigger_events == {"calendar": recorded["env"][:2]} + assert run.as_of_turn(3) is run and run.as_of_turn(7) is run + + +@pytest.mark.asyncio +async def test_a_prompt_with_no_stored_conversation_is_one_turn_of_its_prompt_and_reply(): + """An episode's turn records a prompt and its reply, no conversation.""" + prompt_response = PromptResponse(prompt_id="p1", response="done", prompt_text="do it", step_id="turn") + + run = await load_run_context(TaskStepContext(prompt_responses=[prompt_response]), prompt_response) + + assert [(m["turn"], m["role"], m["text"]) for m in run.conversation] == [(1, "user", "do it"), (1, "agent", "done")] + assert run.env_trigger_events == {} and run.agent_trigger_events == [] and run.unavailable == {} + + +@pytest.mark.asyncio +async def test_what_the_run_meant_to_record_but_cannot_be_read_is_named_not_left_empty(): + context, prompt_response, _ = recorded_run() + context.metadata["a2a_conversations"]["prompt"] = "conv-gone" + context.metadata["env_trigger_state"] = { + "calendar": {"error": "TimeoutError: capture budget of 90s exceeded"}, + "mail": {"object_url": get_config().get_object_store().object_url("env_trigger_state/nothing-here.json")}, + } + + run = await load_run_context(context, prompt_response) + + assert run.conversation == [] and run.env_trigger_events == {} + assert run.unavailable["conversation"] == "conversation 'conv-gone' is not in the store" + assert run.unavailable["env_trigger_events.calendar"] == "TimeoutError: capture budget of 90s exceeded" + assert "env_trigger_events.mail" in run.unavailable + + +@pytest.mark.asyncio +async def test_without_a_prompt_the_run_context_has_the_env_firing_logs_alone(): + context, _, recorded = recorded_run() + + run = await load_run_context(context, None) + + assert run.conversation == [] and run.agent_trigger_events == [] + assert run.env_trigger_events == {"calendar": recorded["env"]} + + +def test_an_empty_run_context_has_no_turns(): + assert RunContext().turns == 0 and RunContext().as_of_turn(1) == RunContext() + + +# ---------------------------------------------------------------- the judge sees it + +TRAJECTORIES = {"s3://t/1.json": "TURN_ONE_EVENTS", "s3://t/2.json": "TURN_TWO_EVENTS", "s3://t/3.json": "TURN_THREE_EVENTS"} + + +def _judged_run() -> tuple[TaskStepContext, dict]: + context, prompt_response, recorded = recorded_run() + prompt_response.target_agent_per_turn_trajectory_object_urls = list(TRAJECTORIES) + prompt_response.agent_trajectory_object_url = "s3://t/3.json" + prompt_response.source_agent_per_turn_prompt_parts = [None, _text(TURNS[1][0]), _text(TURNS[2][0])] + return context, recorded + + +def _verifier(**kw) -> RubricsVerifierTaskStep: + return RubricsVerifierTaskStep(id="verify", version=1, criteria=[{"id": "c1", "description": "outcome"}], + prompt_id="p1", use_agent_judge=False, use_trajectory=True, verifier_id="vt", **kw) + + +def _judge_prompt(monkeypatch, verifier) -> dict: + captured: dict = {} + + async def fake_run(*, eval_prompt, **kwargs): + captured["eval_prompt"] = eval_prompt + return ([{**verifier.criteria[0], "score": 1.0, "result": True}], 0, [], None) + + monkeypatch.setattr(verifier, "_read_trajectory_text", lambda uri: TRAJECTORIES[uri]) + monkeypatch.setattr(verifier, "_run_judge_with_output_retries", fake_run) + return captured + + +@pytest.mark.asyncio +async def test_the_judge_sees_the_conversation_and_the_firing_logs_it_asks_for(monkeypatch): + v = _verifier(run_context_for_judge=["conversation", "trigger_log"]) + captured = _judge_prompt(monkeypatch, v) + context, recorded = _judged_run() + + await v.execute(context) + + prompt = captured["eval_prompt"] + conversation = prompt[prompt.index("## Conversation"):prompt.index("## Trigger firings")] + assert all(user in conversation and agent in conversation for user, agent in TURNS) + assert f"- user (dana_sim, written by triggers ['reschedule']): {TURNS[1][0]}" in conversation + assert f"- agent (solver): {TURNS[2][1]}" in conversation and "### Turn 3" in conversation + log = prompt[prompt.index("## Trigger firings"):] + assert "### env calendar" in log and "### user agent" in log + assert json.dumps({k: v for k, v in recorded["env"][2].items() if k != "seq"}) in log and '"seq"' not in log + + +@pytest.mark.asyncio +async def test_graded_as_of_a_turn_the_judge_sees_neither_later_turns_nor_later_firings(monkeypatch): + v = _verifier(run_context_for_judge=["conversation", "trigger_log"], grade_up_to_turn=2) + captured = _judge_prompt(monkeypatch, v) + context, _ = _judged_run() + + await v.execute(context) + + prompt = captured["eval_prompt"] + assert TURNS[1][1] in prompt and "### Turn 3" not in prompt and TURNS[2][0] not in prompt + assert '"trigger_id": "confirm"' in prompt and '"trigger_id": "late"' not in prompt + + +@pytest.mark.asyncio +async def test_the_judge_sees_only_the_parts_it_asks_for(monkeypatch): + for parts, shown, hidden in ((None, [], ["## Conversation", "## Trigger firings"]), + (["conversation"], ["## Conversation"], ["## Trigger firings"]), + (["trigger_log"], ["## Trigger firings"], ["## Conversation"])): + v = _verifier(run_context_for_judge=parts) + captured = _judge_prompt(monkeypatch, v) + await v.execute(_judged_run()[0]) + assert all(s in captured["eval_prompt"] for s in shown) and not any(s in captured["eval_prompt"] for s in hidden) + + +def test_a_long_message_and_a_long_firing_log_are_bounded_and_a_missing_log_is_named(): + long = "x" * (RUN_CONTEXT_MESSAGE_MAX_CHARS + 50) + run = RunContext( + conversation=[{"turn": 1, "role": "user", "text": long, "parts": _text(long) + [{"kind": "file"}], "ts": None}], + env_trigger_events={"calendar": [{"seq": i, "kind": "fired", "trigger_id": f"t{i}"} + for i in range(RUN_CONTEXT_EVENTS_MAX + 3)]}, + unavailable={"env_trigger_events.mail": "TimeoutError: capture budget of 90s exceeded"}) + + text = render_run_context(run, ["conversation", "trigger_log"]) + + assert "x" * RUN_CONTEXT_MESSAGE_MAX_CHARS + "… (+1 non-text part)" in text and long not in text + assert "(3 earlier events not shown)" in text and '"t2"' not in text and '"t3"' in text + assert "(not available: env_trigger_events.mail: TimeoutError: capture budget of 90s exceeded)" in text + assert "(no trigger firings were recorded)" in render_run_context(RunContext(), ["trigger_log"]) + + +@pytest.mark.parametrize("bad", [[], ["conversation", "screens"], "conversation", ["Conversation"]]) +def test_run_context_for_judge_takes_only_known_parts(bad): + with pytest.raises(ValueError, match="run_context_for_judge"): + _verifier(run_context_for_judge=bad) + + +def test_run_context_for_judge_round_trips_and_is_omitted_when_unset(): + v = _verifier(run_context_for_judge=["trigger_log", "conversation", "trigger_log"]) + assert v.to_dict()["run_context_for_judge"] == ["trigger_log", "conversation"] + assert RubricsVerifierTaskStep.from_dict(v.to_dict()).run_context_for_judge == ["trigger_log", "conversation"] + assert "run_context_for_judge" not in _verifier().to_dict() + + +# ---------------------------------------------------------------- a verify(mcp_url, run) script gets it + +RUN_SCRIPT = b""" +async def verify(mcp_url, run): + rescheduled = [m for m in run.conversation if m.get("fired") == ["reschedule"]] + return [{"id": "reschedule-reached-the-agent", "score": 1.0, "result": bool(rescheduled), + "turns": run.turns, "env_fired": [e["trigger_id"] for e in run.env_trigger_events["calendar"] + if e["kind"] == "fired"]}] +""" + + +def _outcome(script: bytes, monkeypatch, **kw) -> EnvOutcomeVerifierTaskStep: + monkeypatch.setattr(FileArtifact, "get", staticmethod(lambda id, version=None: SimpleNamespace(load=lambda: script))) + return EnvOutcomeVerifierTaskStep(id="outcome", version=None, env_id="calendar", file_artifact_id="script", + verifier_id="v", **kw) + + +@pytest.mark.asyncio +async def test_a_two_argument_verify_is_handed_the_run_it_grades(monkeypatch): + context, _, _ = recorded_run() + context.deployed_envs.append(SimpleNamespace(env_id="calendar", mcp_url="http://mcp")) + + await _outcome(RUN_SCRIPT, monkeypatch, prompt_id="p1").execute(context) + + [row] = context.metadata["verifications"]["v"]["results"] + assert (row["result"], row["turns"], row["env_fired"]) == (True, 3, ["confirm", "late"]) + + +@pytest.mark.asyncio +async def test_without_a_prompt_id_the_runs_last_prompt_is_graded(monkeypatch): + context, _, _ = recorded_run() + context.prompt_responses.insert(0, PromptResponse(prompt_id="warmup", response="ok", prompt_text="hi", step_id="w")) + context.deployed_envs.append(SimpleNamespace(env_id="calendar", mcp_url="http://mcp")) + + await _outcome(RUN_SCRIPT, monkeypatch).execute(context) + + assert context.metadata["verifications"]["v"]["results"][0]["turns"] == 3 + + +@pytest.mark.asyncio +async def test_a_one_argument_verify_is_called_as_before(monkeypatch): + context, _, _ = recorded_run() + context.deployed_envs.append(SimpleNamespace(env_id="calendar", mcp_url="http://mcp")) + script = b"async def verify(mcp_url):\n return [{'id': 'c', 'score': 1.0, 'url': mcp_url}]\n" + + await _outcome(script, monkeypatch, prompt_id="p1").execute(context) + + assert context.metadata["verifications"]["v"]["results"] == [{"id": "c", "score": 1.0, "url": "http://mcp"}] + + +@pytest.mark.asyncio +async def test_an_unknown_prompt_id_fails_the_step(monkeypatch): + context, _, _ = recorded_run() + context.deployed_envs.append(SimpleNamespace(env_id="calendar", mcp_url="http://mcp")) + + with pytest.raises(RuntimeError, match="prompt_id='nope' not found"): + await _outcome(RUN_SCRIPT, monkeypatch, prompt_id="nope").execute(context) + + +def test_prompt_id_round_trips_and_is_omitted_when_unset(monkeypatch): + step = _outcome(RUN_SCRIPT, monkeypatch, prompt_id="p1") + assert EnvOutcomeVerifierTaskStep.from_dict(step.to_dict()).prompt_id == "p1" + assert "prompt_id" not in _outcome(RUN_SCRIPT, monkeypatch).to_dict() From b18739483204b18c041ef56df6febe39facab367 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Mon, 19 Oct 2026 12:02:15 -0700 Subject: [PATCH 2/7] review: pass the run context only to a verify that asks for `run`; one-line comments - env_outcome_verifier calls `verify(mcp_url, run=...)` only when the script names a `run` parameter; a script with any other second parameter is called exactly as before (new test). - Every added comment and docstring is one line; pre-existing ones are as they were. Co-Authored-By: Claude Opus 5.5 --- .../verifiers/env_outcome_verifier.py | 9 ++++--- .../task_steps/verifiers/rubrics_verifier.py | 9 +++---- .../task_steps/verifiers/run_context.py | 24 ++++--------------- .../test_verifier_run_context_local.py | 4 +--- .../task_step/verifiers/test_run_context.py | 22 +++++++++++------ 5 files changed, 28 insertions(+), 40 deletions(-) diff --git a/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py b/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py index 938b6ded..81c41f3c 100644 --- a/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py @@ -24,9 +24,7 @@ class EnvOutcomeVerifierTaskStep(TaskStep): - """Runs the artifact's ``async verify(mcp_url)`` against the deployed env and scores what it returns. A - ``verify(mcp_url, run)`` is also handed the run it grades (``run_context.RunContext``): the conversation of - ``prompt_id``'s prompt, the run's last prompt when unset, and the trigger firing logs.""" + """Runs the artifact's ``verify(mcp_url)``; a ``run`` parameter also gets the ``RunContext`` of ``prompt_id``.""" type: ClassVar[str] = "env_outcome_verifier" entity_refs = ( @@ -121,8 +119,9 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: spec.loader.exec_module(module) logger.info(f"Running verify() from artifact '{self.file_artifact_id}' against {deployed.mcp_url}") - if len(inspect.signature(module.verify).parameters) >= 2: - results = await module.verify(deployed.mcp_url, await load_run_context(context, self._graded_prompt(context))) + if "run" in inspect.signature(module.verify).parameters: + run = await load_run_context(context, self._graded_prompt(context)) + results = await module.verify(deployed.mcp_url, run=run) else: results = await module.verify(deployed.mcp_url) diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py index bb8f23e9..ea04d144 100644 --- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py @@ -189,8 +189,7 @@ def _cut(text: str, limit: int) -> str: def render_run_context(run: RunContext, parts: list[str]) -> str: - """The judge-prompt sections for ``parts`` of ``run``: the conversation turn by turn, and each firing log as one - compact JSON line per event, the most recent ``RUN_CONTEXT_EVENTS_MAX`` of each.""" + """The judge-prompt sections for ``parts`` of ``run``: the conversation by turn, each firing log one event a line.""" sections: list[str] = [] if "conversation" in parts: lines = ["## Conversation", "The run's messages, turn by turn: the user side (the task's prompt, then each " @@ -402,8 +401,7 @@ def __init__( fail_task_on_error: bool = True, ): super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) - # What of the run beyond its trajectory the judge also sees (RUN_CONTEXT_PARTS): the conversation with every - # user-side turn, and the trigger firing logs; cut at grade_up_to_turn like the trajectory. + # RUN_CONTEXT_PARTS the judge also sees, cut at grade_up_to_turn like the trajectory. self.run_context_for_judge = _validated_run_context_parts(run_context_for_judge) # Grade a multi-turn run as it stood after this (1-based) turn: the judge sees turns 1..k only, # so a per-turn env snapshot can be scored on what the agent had done by then. None grades all. @@ -1359,8 +1357,7 @@ def _build_eval_prompt( run_context: Optional[RunContext] = None, ) -> str: """``context_facts`` are rendered ``(label, value)`` pairs (``resolve_context_facts``); ``image_blocks`` - are the blocks attached to the judge message, labels included when ``per_criterion_frames``; ``run_context`` - is shown as ``run_context_for_judge`` asks.""" + are the blocks attached to the judge message, labels included when ``per_criterion_frames``.""" spec = self._spec() # Only the rubric_evidence templates have this placeholder; the others ignore the argument. checks_instruction = evidence_checks_instruction(has_policy=bool(self.grading_policy_prompt)) diff --git a/src/agent_env/task_step/task_steps/verifiers/run_context.py b/src/agent_env/task_step/task_steps/verifiers/run_context.py index dd845239..c139c0a6 100644 --- a/src/agent_env/task_step/task_steps/verifiers/run_context.py +++ b/src/agent_env/task_step/task_steps/verifiers/run_context.py @@ -1,11 +1,4 @@ -"""The run a verifier grades, beyond the env's state: the conversation, every user-side turn included, and the -trigger firing logs. - -``load_run_context`` reads it from what the run recorded: the conversation ``prompt_agent`` stored for the prompt -(``metadata["a2a_conversations"]``), the env firing logs it captured at the prompt's end -(``metadata["env_trigger_state"]``), and the user agent's own firing log (``metadata["agent_trigger_state"]``). -``RunContext.as_of_turn`` cuts it the way ``grade_up_to_turn`` cuts the trajectory. -""" +"""What a verifier knows of the run it grades beyond the env: the conversation and the trigger firing logs.""" from __future__ import annotations import asyncio @@ -21,11 +14,7 @@ @dataclass(frozen=True) class RunContext: - """``conversation``: the prompt's messages, oldest first, each ``{"turn", "role", "text", "parts", "ts"}`` with - ``author`` (who sent it, when recorded), and on a user message the agent triggers that wrote it (``fired``) and - the extra fields the user-sim returned with it (``fields``). ``env_trigger_events``: each env's gateway firing - log, oldest first. ``agent_trigger_events``: the user agent's firing log. ``unavailable`` names what the run - meant to record but could not be read, and why, so a verifier can tell a missing log from an empty one.""" + """The prompt's messages and firing logs, oldest first; ``unavailable`` names what could not be read, and why.""" conversation: list[dict] = field(default_factory=list) env_trigger_events: dict[str, list[dict]] = field(default_factory=dict) @@ -37,8 +26,7 @@ def turns(self) -> int: return max((m["turn"] for m in self.conversation), default=0) def as_of_turn(self, turn: int) -> RunContext: - """The run as it stood after ``turn``: its messages up to that turn's reply, and the firing-log events from - before the next turn's user message (all of them when no turn follows).""" + """The run as it stood after ``turn``: its messages, and the events from before the next turn's message.""" if turn >= self.turns: return self cutoff = next(m["ts"] for m in self.conversation if m["turn"] == turn + 1) @@ -59,8 +47,7 @@ def to_dict(self) -> dict[str, Any]: async def load_run_context(context: TaskStepContext, prompt_response: Optional[PromptResponse]) -> RunContext: - """The run context of ``prompt_response``'s prompt (the env firing logs alone without one). Never raises for a - record it cannot read; that lands in ``unavailable``.""" + """The run context of ``prompt_response``'s prompt (env firing logs alone without one); never raises for a record.""" unavailable: dict[str, str] = {} conversation: list[dict] = [] agent_events: list[dict] = [] @@ -73,8 +60,7 @@ async def load_run_context(context: TaskStepContext, prompt_response: Optional[P def _conversation(context: TaskStepContext, prompt_response: PromptResponse, unavailable: dict[str, str]) -> list[dict]: - """The stored conversation, numbered by user message; a prompt with none recorded (an episode's turn) is one turn - of its prompt and reply.""" + """The stored conversation, numbered by user message; without one (an episode's turn), its prompt and reply.""" conversation_id = (context.metadata.get("a2a_conversations") or {}).get(prompt_response.step_id) if conversation_id is None: return [_message(1, "user", [{"kind": "text", "text": prompt_response.prompt_text or ""}]), diff --git a/tst/integration/task_step/test_verifier_run_context_local.py b/tst/integration/task_step/test_verifier_run_context_local.py index 33362800..3a0b073a 100644 --- a/tst/integration/task_step/test_verifier_run_context_local.py +++ b/tst/integration/task_step/test_verifier_run_context_local.py @@ -1,6 +1,4 @@ -"""A verifier is handed the run it grades: after a real two-turn conversation on the local defaults, with the echo -agent (tst/data/a2a_agent) as both solver and user-sim, the run context reads back every turn prompt_agent stored, the -user-sim's included. Needs Docker and a throwaway local registry.""" +"""A real two-turn conversation (echo agent as solver and user-sim) reads back whole through load_run_context.""" import uuid diff --git a/tst/unit/task_step/verifiers/test_run_context.py b/tst/unit/task_step/verifiers/test_run_context.py index 14cbade1..41a03594 100644 --- a/tst/unit/task_step/verifiers/test_run_context.py +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -1,5 +1,4 @@ -"""The run a verifier grades, read from what the run recorded: the stored conversation with every user-side turn, -the env firing logs prompt_agent captured, and the user agent's own firing log; cut as of a turn like the trajectory.""" +"""The run a verifier grades: the stored conversation, the firing logs, and the cut as of a turn.""" from __future__ import annotations import json @@ -39,8 +38,7 @@ def _between(earlier: str, later: str) -> str: def recorded_run(turns=TURNS) -> tuple[TaskStepContext, PromptResponse, dict]: - """A three-turn run as prompt_agent records it: the conversation in the store, which agent trigger wrote turn 2, - the user-sim's extra fields on turn 3, the env firing log in the object store and the user agent's firing log.""" + """A three-turn run as prompt_agent records it: conversation, trigger and user-sim marks, and both firing logs.""" conversation_id = f"conv-{uuid.uuid4().hex[:8]}" conversation_store.create_conversation(conversation_id, task_instance_id="run-1", source_agent_name="dana_sim", target_agent_name="solver") @@ -103,7 +101,7 @@ async def test_as_of_a_turn_the_run_stops_before_the_next_turn_begins(): assert [m["turn"] for m in first.conversation] == [1, 1] assert first.env_trigger_events == {"calendar": recorded["env"][:2]} - assert first.agent_trigger_events == recorded["agent"] # fired just before turn 2's message: it wrote it + assert first.agent_trigger_events == recorded["agent"] assert [m["turn"] for m in second.conversation] == [1, 1, 2, 2] assert second.env_trigger_events == {"calendar": recorded["env"][:2]} assert run.as_of_turn(3) is run and run.as_of_turn(7) is run @@ -111,7 +109,6 @@ async def test_as_of_a_turn_the_run_stops_before_the_next_turn_begins(): @pytest.mark.asyncio async def test_a_prompt_with_no_stored_conversation_is_one_turn_of_its_prompt_and_reply(): - """An episode's turn records a prompt and its reply, no conversation.""" prompt_response = PromptResponse(prompt_id="p1", response="done", prompt_text="do it", step_id="turn") run = await load_run_context(TaskStepContext(prompt_responses=[prompt_response]), prompt_response) @@ -270,7 +267,7 @@ def _outcome(script: bytes, monkeypatch, **kw) -> EnvOutcomeVerifierTaskStep: @pytest.mark.asyncio -async def test_a_two_argument_verify_is_handed_the_run_it_grades(monkeypatch): +async def test_a_verify_asking_for_run_is_handed_the_run_it_grades(monkeypatch): context, _, _ = recorded_run() context.deployed_envs.append(SimpleNamespace(env_id="calendar", mcp_url="http://mcp")) @@ -291,6 +288,17 @@ async def test_without_a_prompt_id_the_runs_last_prompt_is_graded(monkeypatch): assert context.metadata["verifications"]["v"]["results"][0]["turns"] == 3 +@pytest.mark.asyncio +async def test_a_verify_with_an_unrelated_second_parameter_is_called_as_before(monkeypatch): + context, _, _ = recorded_run() + context.deployed_envs.append(SimpleNamespace(env_id="calendar", mcp_url="http://mcp")) + script = b"async def verify(mcp_url, retries=3):\n return [{'id': 'c', 'score': 1.0, 'retries': retries}]\n" + + await _outcome(script, monkeypatch, prompt_id="p1").execute(context) + + assert context.metadata["verifications"]["v"]["results"] == [{"id": "c", "score": 1.0, "retries": 3}] + + @pytest.mark.asyncio async def test_a_one_argument_verify_is_called_as_before(monkeypatch): context, _, _ = recorded_run() From b87971571ecd0609f39fb7f2f6067bf391d49b93 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Mon, 19 Oct 2026 13:01:24 -0700 Subject: [PATCH 3/7] review: a user-agent firing log that wasn't read back is unavailable, not empty; a turn cut without the conversation withholds the logs - load_run_context marks agent_trigger_events unavailable when the prompt's user agent had triggers (a firing recorded for the step, or a registration for its user agent; any registration when the conversation can't be read) but prompt_agent stored no firing log. The judge's Trigger firings section now lists every unavailable log, not only the env ones. - as_of_turn withholds the firing logs, and marks each unavailable, when it can't find when the next turn began (no conversation was read), instead of returning them uncut. - The 200-character cap on a read error's message is RunContext.ERROR_MAX_CHARS. Co-Authored-By: Claude Opus 5.5 --- .../task_steps/verifiers/rubrics_verifier.py | 3 +- .../task_steps/verifiers/run_context.py | 44 +++++++++++--- .../task_step/verifiers/test_run_context.py | 57 +++++++++++++++++++ 3 files changed, 94 insertions(+), 10 deletions(-) diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py index 42e8e91e..3be6ccca 100644 --- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py @@ -221,8 +221,7 @@ def render_run_context(run: RunContext, parts: list[str]) -> str: lines.append(f"({len(events) - RUN_CONTEXT_EVENTS_MAX} earlier events not shown)") lines += [f"- {json.dumps({k: v for k, v in e.items() if k != 'seq'}, default=str)}" for e in events[-RUN_CONTEXT_EVENTS_MAX:]] - lines += [f"(not available: {what}: {why})" for what, why in run.unavailable.items() - if what.startswith("env_trigger_events.")] + lines += [f"(not available: {what}: {why})" for what, why in run.unavailable.items() if what != "conversation"] if len(lines) == 2: lines.append("(no trigger firings were recorded)") sections.append("\n".join(lines)) diff --git a/src/agent_env/task_step/task_steps/verifiers/run_context.py b/src/agent_env/task_step/task_steps/verifiers/run_context.py index c139c0a6..f19cfab6 100644 --- a/src/agent_env/task_step/task_steps/verifiers/run_context.py +++ b/src/agent_env/task_step/task_steps/verifiers/run_context.py @@ -5,7 +5,7 @@ import json from dataclasses import dataclass, field, replace from datetime import datetime -from typing import Any, Optional +from typing import Any, ClassVar, Optional from agent_env.a2a_agent import conversation_store from agent_env.config import get_config @@ -16,6 +16,8 @@ class RunContext: """The prompt's messages and firing logs, oldest first; ``unavailable`` names what could not be read, and why.""" + ERROR_MAX_CHARS: ClassVar[int] = 200 # of a read error's message, in an ``unavailable`` reason + conversation: list[dict] = field(default_factory=list) env_trigger_events: dict[str, list[dict]] = field(default_factory=dict) agent_trigger_events: list[dict] = field(default_factory=list) @@ -26,17 +28,24 @@ def turns(self) -> int: return max((m["turn"] for m in self.conversation), default=0) def as_of_turn(self, turn: int) -> RunContext: - """The run as it stood after ``turn``: its messages, and the events from before the next turn's message.""" - if turn >= self.turns: + """The run after ``turn``: its messages, and the events before the next turn's message (none if it's unknown).""" + if self.conversation and turn >= self.turns: return self - cutoff = next(m["ts"] for m in self.conversation if m["turn"] == turn + 1) + conversation = [m for m in self.conversation if m["turn"] <= turn] + cutoff = next((m["ts"] for m in self.conversation if m["turn"] == turn + 1), None) + if cutoff is None: + why = f"withheld: when turn {turn + 1} began is unknown, so later events can't be told apart" + withheld = [f"env_trigger_events.{env_id}" for env_id in self.env_trigger_events] + withheld += ["agent_trigger_events"] if self.agent_trigger_events else [] + return replace(self, conversation=conversation, env_trigger_events={}, agent_trigger_events=[], + unavailable={**self.unavailable, **dict.fromkeys(withheld, why)}) def before(events: list[dict]) -> list[dict]: return [e for e in events if _before(e.get("ts"), cutoff)] return replace( self, - conversation=[m for m in self.conversation if m["turn"] <= turn], + conversation=conversation, env_trigger_events={env_id: before(events) for env_id, events in self.env_trigger_events.items()}, agent_trigger_events=before(self.agent_trigger_events), ) @@ -53,12 +62,31 @@ async def load_run_context(context: TaskStepContext, prompt_response: Optional[P agent_events: list[dict] = [] if prompt_response is not None: conversation = await asyncio.to_thread(_conversation, context, prompt_response, unavailable) - agent_events = list((context.metadata.get("agent_trigger_state") or {}).get(prompt_response.step_id) or []) + agent_events = _agent_trigger_events(context, prompt_response.step_id, conversation, unavailable) env_events = await asyncio.to_thread(_env_trigger_events, context, unavailable) return RunContext(conversation=conversation, env_trigger_events=env_events, agent_trigger_events=agent_events, unavailable=unavailable) +def _error(e: Exception) -> str: + return f"{type(e).__name__}: {str(e)[:RunContext.ERROR_MAX_CHARS]}" + + +def _agent_trigger_events(context: TaskStepContext, step_id: Optional[str], conversation: list[dict], + unavailable: dict[str, str]) -> list[dict]: + """The user agent's firing log; unavailable, not empty, when it had triggers but ``prompt_agent`` read none back.""" + logs = context.metadata.get("agent_trigger_state") or {} + if step_id in logs: + return list(logs[step_id] or []) + armed = {r.get("agent_name") for r in context.metadata.get("agent_trigger_registrations") or []} + user_agent = next((m.get("author") for m in conversation if m["role"] == "user"), None) + # Without the conversation the user agent is unknown, so any agent's registered triggers may be its. + had_triggers = bool(armed) if "conversation" in unavailable else user_agent in armed + if had_triggers or step_id in (context.metadata.get("agent_trigger_firings") or {}): + unavailable["agent_trigger_events"] = "agent triggers were registered, but the user agent's log wasn't read" + return [] + + def _conversation(context: TaskStepContext, prompt_response: PromptResponse, unavailable: dict[str, str]) -> list[dict]: """The stored conversation, numbered by user message; without one (an episode's turn), its prompt and reply.""" conversation_id = (context.metadata.get("a2a_conversations") or {}).get(prompt_response.step_id) @@ -68,7 +96,7 @@ def _conversation(context: TaskStepContext, prompt_response: PromptResponse, una try: doc = conversation_store.get_conversation(conversation_id) except Exception as e: - unavailable["conversation"] = f"{type(e).__name__}: {str(e)[:200]}" + unavailable["conversation"] = _error(e) return [] if doc is None: unavailable["conversation"] = f"conversation {conversation_id!r} is not in the store" @@ -110,7 +138,7 @@ def _env_trigger_events(context: TaskStepContext, unavailable: dict[str, str]) - try: envelope = json.loads(get_config().get_object_store().get(entry["object_url"])) except Exception as e: - unavailable[f"env_trigger_events.{env_id}"] = f"{type(e).__name__}: {str(e)[:200]}" + unavailable[f"env_trigger_events.{env_id}"] = _error(e) continue events[env_id] = list((envelope.get("state") or {}).get("events") or []) return events diff --git a/tst/unit/task_step/verifiers/test_run_context.py b/tst/unit/task_step/verifiers/test_run_context.py index 41a03594..303c5de3 100644 --- a/tst/unit/task_step/verifiers/test_run_context.py +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -65,6 +65,7 @@ def recorded_run(turns=TURNS) -> tuple[TaskStepContext, PromptResponse, dict]: "agent_trigger_firings": {"prompt": [{"turn": 1, "fired": ["reschedule"]}, {"turn": 2, "fired": []}]}, "usersim_turn_outputs": {"prompt": [{"turn": 2, "fields": {"mood": "satisfied"}}]}, "env_trigger_state": {"calendar": {"object_url": object_url, "capture_step_id": "prompt"}}, + "agent_trigger_registrations": [{"step_id": "arm", "agent_name": "dana_sim", "added": ["reschedule"]}], "agent_trigger_state": {"prompt": agent_events}, }) prompt_response = PromptResponse(prompt_id="p1", response="Glad to help.", prompt_text=TURNS[0][0], step_id="prompt") @@ -134,6 +135,47 @@ async def test_what_the_run_meant_to_record_but_cannot_be_read_is_named_not_left assert "env_trigger_events.mail" in run.unavailable +@pytest.mark.asyncio +@pytest.mark.parametrize("dropped", ["agent_trigger_firings", "agent_trigger_registrations"]) +async def test_a_user_agent_firing_log_that_was_not_read_back_is_named_not_left_empty(dropped): + context, prompt_response, _ = recorded_run() + for key in ("agent_trigger_state", dropped): + del context.metadata[key] + + run = await load_run_context(context, prompt_response) + + assert run.agent_trigger_events == [] and "agent_trigger_events" in run.unavailable + text = render_run_context(run, ["trigger_log"]) + assert f"(not available: agent_trigger_events: {run.unavailable['agent_trigger_events']})" in text + assert "(no trigger firings were recorded)" not in text + + +@pytest.mark.asyncio +async def test_a_user_agent_with_no_triggers_of_its_own_has_no_firing_log_to_miss(): + context, prompt_response, _ = recorded_run() + for key in ("agent_trigger_state", "agent_trigger_firings"): + del context.metadata[key] + context.metadata["agent_trigger_registrations"] = [{"step_id": "arm", "agent_name": "other_sim", "added": ["x"]}] + + run = await load_run_context(context, prompt_response) + + assert run.agent_trigger_events == [] and run.unavailable == {} + + +@pytest.mark.asyncio +async def test_without_the_conversation_a_turn_cut_withholds_the_firing_logs_it_cannot_cut(): + context, prompt_response, recorded = recorded_run() + context.metadata["a2a_conversations"]["prompt"] = "conv-gone" + run = await load_run_context(context, prompt_response) + + first = run.as_of_turn(1) + + assert run.env_trigger_events == {"calendar": recorded["env"]} and run.agent_trigger_events == recorded["agent"] + assert first.conversation == [] and first.env_trigger_events == {} and first.agent_trigger_events == [] + assert first.unavailable["conversation"] == run.unavailable["conversation"] + assert first.unavailable["env_trigger_events.calendar"] == first.unavailable["agent_trigger_events"] + + @pytest.mark.asyncio async def test_without_a_prompt_the_run_context_has_the_env_firing_logs_alone(): context, _, recorded = recorded_run() @@ -209,6 +251,21 @@ async def test_graded_as_of_a_turn_the_judge_sees_neither_later_turns_nor_later_ assert '"trigger_id": "confirm"' in prompt and '"trigger_id": "late"' not in prompt +@pytest.mark.asyncio +async def test_graded_as_of_a_turn_without_the_conversation_the_judge_sees_no_firings(monkeypatch): + v = _verifier(run_context_for_judge=["trigger_log"], grade_up_to_turn=1) + captured = _judge_prompt(monkeypatch, v) + context, _ = _judged_run() + context.metadata["a2a_conversations"]["prompt"] = "conv-gone" + + await v.execute(context) + + prompt = captured["eval_prompt"] + assert '"trigger_id"' not in prompt + assert "(not available: env_trigger_events.calendar: withheld" in prompt + assert "(not available: agent_trigger_events: withheld" in prompt + + @pytest.mark.asyncio async def test_the_judge_sees_only_the_parts_it_asks_for(monkeypatch): for parts, shown, hidden in ((None, [], ["## Conversation", "## Trigger firings"]), From edc94b441067e15b5badcb492d95a0925d38bca9 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Mon, 19 Oct 2026 13:28:53 -0700 Subject: [PATCH 4/7] review: the judge's run-context limits are RubricsVerifierTaskStep class variables RUN_CONTEXT_MESSAGE_MAX_CHARS (4000) and RUN_CONTEXT_EVENTS_MAX (200) move from module level onto RubricsVerifierTaskStep as ClassVars, beside its other limits; render_run_context reads them there. Same values, same rendering. Co-Authored-By: Claude Opus 5.5 --- .../task_steps/verifiers/rubrics_verifier.py | 15 ++++++++------- tst/unit/task_step/verifiers/test_run_context.py | 14 +++++--------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py index 3be6ccca..7e0958ac 100644 --- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py @@ -174,10 +174,8 @@ def resolve_context_facts(facts: dict[str, str], metadata: dict) -> list[tuple[s return rendered -# What of the run beyond its trajectory `run_context_for_judge` may show the judge, and the bounds on how much. +# What of the run beyond its trajectory `run_context_for_judge` may show the judge. RUN_CONTEXT_PARTS = ("conversation", "trigger_log") -RUN_CONTEXT_MESSAGE_MAX_CHARS = 4000 # a longer message is cut and ends in "…" -RUN_CONTEXT_EVENTS_MAX = 200 # per firing log, the most recent; the rest are counted def _validated_run_context_parts(parts: Optional[list[str]]) -> Optional[list[str]]: @@ -208,19 +206,20 @@ def render_run_context(run: RunContext, parts: list[str]) -> str: files = sum(1 for p in m["parts"] if p.get("kind") != "text") extra = f" (+{files} non-text part{'s' if files != 1 else ''})" if files else "" lines.append(f"- {m['role']}{f' ({who})' if who else ''}: " - f"{_cut(m['text'], RUN_CONTEXT_MESSAGE_MAX_CHARS)}{extra}") + f"{_cut(m['text'], RubricsVerifierTaskStep.RUN_CONTEXT_MESSAGE_MAX_CHARS)}{extra}") sections.append("\n".join(lines)) if "trigger_log" in parts: lines = ["## Trigger firings", "What the env's triggers and the user agent's triggers did during the run, " "oldest first, one firing-log event per line."] logs = [(f"env {env_id}", events) for env_id, events in run.env_trigger_events.items()] logs += [("user agent", run.agent_trigger_events)] if run.agent_trigger_events else [] + events_max = RubricsVerifierTaskStep.RUN_CONTEXT_EVENTS_MAX for name, events in logs: lines.append(f"### {name}") - if len(events) > RUN_CONTEXT_EVENTS_MAX: - lines.append(f"({len(events) - RUN_CONTEXT_EVENTS_MAX} earlier events not shown)") + if len(events) > events_max: + lines.append(f"({len(events) - events_max} earlier events not shown)") lines += [f"- {json.dumps({k: v for k, v in e.items() if k != 'seq'}, default=str)}" - for e in events[-RUN_CONTEXT_EVENTS_MAX:]] + for e in events[-events_max:]] lines += [f"(not available: {what}: {why})" for what, why in run.unavailable.items() if what != "conversation"] if len(lines) == 2: lines.append("(no trigger firings were recorded)") @@ -365,6 +364,8 @@ class RubricsVerifierTaskStep(TaskStep): DEFAULT_JUDGE_TIMEOUT_SECONDS: ClassVar[int] = 1000 DEFAULT_MAX_RETRIES: ClassVar[int] = 3 DEFAULT_POLL_INTERVAL_SECONDS: ClassVar[int] = 10 + RUN_CONTEXT_MESSAGE_MAX_CHARS: ClassVar[int] = 4000 # a longer message is cut and ends in "…" + RUN_CONTEXT_EVENTS_MAX: ClassVar[int] = 200 # per firing log, the most recent; the rest are counted TRAJECTORY_CONTAINER_PATH: ClassVar[str] = "/tmp/prompt_trajectory.json" # Multi-turn agent judge: base for a per-run subdir holding turn_NN.json files. TRAJECTORY_CONTAINER_DIR: ClassVar[str] = "/tmp/prompt_trajectory_turns" diff --git a/tst/unit/task_step/verifiers/test_run_context.py b/tst/unit/task_step/verifiers/test_run_context.py index 303c5de3..cbd33746 100644 --- a/tst/unit/task_step/verifiers/test_run_context.py +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -15,12 +15,7 @@ from agent_env.config import get_config from agent_env.task_step.context import PromptResponse, TaskStepContext from agent_env.task_step.task_steps.verifiers.env_outcome_verifier import EnvOutcomeVerifierTaskStep -from agent_env.task_step.task_steps.verifiers.rubrics_verifier import ( - RUN_CONTEXT_EVENTS_MAX, - RUN_CONTEXT_MESSAGE_MAX_CHARS, - RubricsVerifierTaskStep, - render_run_context, -) +from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep, render_run_context from agent_env.task_step.task_steps.verifiers.run_context import RunContext, load_run_context TURNS = [("Can you book lunch with Dana?", "Booked for noon."), @@ -278,16 +273,17 @@ async def test_the_judge_sees_only_the_parts_it_asks_for(monkeypatch): def test_a_long_message_and_a_long_firing_log_are_bounded_and_a_missing_log_is_named(): - long = "x" * (RUN_CONTEXT_MESSAGE_MAX_CHARS + 50) + max_chars = RubricsVerifierTaskStep.RUN_CONTEXT_MESSAGE_MAX_CHARS + long = "x" * (max_chars + 50) run = RunContext( conversation=[{"turn": 1, "role": "user", "text": long, "parts": _text(long) + [{"kind": "file"}], "ts": None}], env_trigger_events={"calendar": [{"seq": i, "kind": "fired", "trigger_id": f"t{i}"} - for i in range(RUN_CONTEXT_EVENTS_MAX + 3)]}, + for i in range(RubricsVerifierTaskStep.RUN_CONTEXT_EVENTS_MAX + 3)]}, unavailable={"env_trigger_events.mail": "TimeoutError: capture budget of 90s exceeded"}) text = render_run_context(run, ["conversation", "trigger_log"]) - assert "x" * RUN_CONTEXT_MESSAGE_MAX_CHARS + "… (+1 non-text part)" in text and long not in text + assert "x" * max_chars + "… (+1 non-text part)" in text and long not in text assert "(3 earlier events not shown)" in text and '"t2"' not in text and '"t3"' in text assert "(not available: env_trigger_events.mail: TimeoutError: capture budget of 90s exceeded)" in text assert "(no trigger firings were recorded)" in render_run_context(RunContext(), ["trigger_log"]) From ef4565a0e7b39ce758ea0da34e463e8d2a0c9541 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 10:40:43 -0700 Subject: [PATCH 5/7] review: cut an earlier prompt's env firing logs where the next prompt began Co-Authored-By: Claude Opus 5.5 --- .../task_steps/verifiers/run_context.py | 30 ++++++++++++++- .../task_step/verifiers/test_run_context.py | 38 +++++++++++++++++++ 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/src/agent_env/task_step/task_steps/verifiers/run_context.py b/src/agent_env/task_step/task_steps/verifiers/run_context.py index f19cfab6..ebaa4276 100644 --- a/src/agent_env/task_step/task_steps/verifiers/run_context.py +++ b/src/agent_env/task_step/task_steps/verifiers/run_context.py @@ -64,8 +64,9 @@ async def load_run_context(context: TaskStepContext, prompt_response: Optional[P conversation = await asyncio.to_thread(_conversation, context, prompt_response, unavailable) agent_events = _agent_trigger_events(context, prompt_response.step_id, conversation, unavailable) env_events = await asyncio.to_thread(_env_trigger_events, context, unavailable) - return RunContext(conversation=conversation, env_trigger_events=env_events, agent_trigger_events=agent_events, - unavailable=unavailable) + run = RunContext(conversation=conversation, env_trigger_events=env_events, agent_trigger_events=agent_events, + unavailable=unavailable) + return await asyncio.to_thread(_until_next_prompt, run, context, prompt_response) if prompt_response else run def _error(e: Exception) -> str: @@ -144,6 +145,31 @@ def _env_trigger_events(context: TaskStepContext, unavailable: dict[str, str]) - return events +def _until_next_prompt(run: RunContext, context: TaskStepContext, prompt_response: PromptResponse) -> RunContext: + """The env firing logs are the whole run's (the last prompt's capture), so an earlier prompt's stop where the + next prompt began; withheld when that start is unknown.""" + index = next((i for i, pr in enumerate(context.prompt_responses) if pr is prompt_response), None) + if index is None or index + 1 >= len(context.prompt_responses) or not run.env_trigger_events: + return run + start = _first_message_ts(context, context.prompt_responses[index + 1]) + if start is None: + why = "withheld: when the next prompt began is unknown, so its events can't be told apart" + withheld = dict.fromkeys((f"env_trigger_events.{env_id}" for env_id in run.env_trigger_events), why) + return replace(run, env_trigger_events={}, unavailable={**run.unavailable, **withheld}) + return replace(run, env_trigger_events={env_id: [e for e in events if _before(e.get("ts"), start)] + for env_id, events in run.env_trigger_events.items()}) + + +def _first_message_ts(context: TaskStepContext, prompt_response: PromptResponse) -> Optional[str]: + conversation_id = (context.metadata.get("a2a_conversations") or {}).get(prompt_response.step_id) + try: + doc = conversation_store.get_conversation(conversation_id) if conversation_id else None + except Exception: + return None + messages = (doc or {}).get("messages") or [] + return messages[0].get("ts") if messages else None + + def _before(ts: Optional[str], cutoff: Optional[str]) -> bool: if ts is None or cutoff is None: return True diff --git a/tst/unit/task_step/verifiers/test_run_context.py b/tst/unit/task_step/verifiers/test_run_context.py index cbd33746..95a227be 100644 --- a/tst/unit/task_step/verifiers/test_run_context.py +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -376,3 +376,41 @@ def test_prompt_id_round_trips_and_is_omitted_when_unset(monkeypatch): step = _outcome(RUN_SCRIPT, monkeypatch, prompt_id="p1") assert EnvOutcomeVerifierTaskStep.from_dict(step.to_dict()).prompt_id == "p1" assert "prompt_id" not in _outcome(RUN_SCRIPT, monkeypatch).to_dict() + + +def _stored_prompt(step_id: str, user: str, agent: str) -> tuple[str, list[dict]]: + conversation_id = f"conv-{uuid.uuid4().hex[:8]}" + conversation_store.create_conversation(conversation_id, task_instance_id="run-1", source_agent_name="dana_sim", + target_agent_name="solver") + conversation_store.add_a2a_task(conversation_id, parts=_text(user), a2a_task_id=uuid.uuid4().hex, role="user") + time.sleep(0.005) + conversation_store.complete_a2a_task(conversation_id, parts=_text(agent), role="agent") + time.sleep(0.005) + return conversation_id, conversation_store.get_conversation(conversation_id)["messages"] + + +@pytest.mark.asyncio +async def test_an_earlier_prompt_sees_only_the_env_events_from_before_the_next_prompt_began(): + first_id, first = _stored_prompt("prompt-1", "Book lunch with Dana.", "Booked for noon.") + second_id, second = _stored_prompt("prompt-2", "Now book dinner.", "Booked for seven.") + env_events = [ # the env's whole log, as the second prompt's end-of-prompt capture stored it + {"seq": 1, "ts": _between(first[0]["ts"], first[1]["ts"]), "kind": "fired", "trigger_id": "lunch_confirm"}, + {"seq": 2, "ts": _between(second[0]["ts"], second[1]["ts"]), "kind": "fired", "trigger_id": "dinner_confirm"}, + ] + envelope = {"env_id": "calendar", "state": {"triggers": [], "events": env_events}} + object_url = get_config().get_object_store().put(f"env_trigger_state/{second_id}.json", + json.dumps(envelope).encode(), content_type="application/json") + context = TaskStepContext(metadata={ + "a2a_conversations": {"prompt-1": first_id, "prompt-2": second_id}, + "env_trigger_state": {"calendar": {"object_url": object_url, "capture_step_id": "prompt-2"}}, + }) + for prompt_id, step_id, text, reply in (("p1", "prompt-1", "Book lunch with Dana.", "Booked for noon."), + ("p2", "prompt-2", "Now book dinner.", "Booked for seven.")): + context.prompt_responses.append(PromptResponse(prompt_id=prompt_id, response=reply, prompt_text=text, + step_id=step_id)) + + graded_first = await load_run_context(context, context.prompt_responses[0]) + graded_last = await load_run_context(context, context.prompt_responses[1]) + + assert [e["trigger_id"] for e in graded_first.env_trigger_events["calendar"]] == ["lunch_confirm"] + assert [e["trigger_id"] for e in graded_last.env_trigger_events["calendar"]] == ["lunch_confirm", "dinner_confirm"] From 14b9342e9b56d53c0265e8eb14404fd8386fb4d9 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 10:52:41 -0700 Subject: [PATCH 6/7] review: hand verifiers a typed, read-only Run (Turn, Message, Firing), and render the judge's sections from it One Firing per fire, folded in one place from the gateway's detected -> fired/failed sequence and the user agent's entries, each placed in its turn. A query that needs a record that couldn't be read raises RunRecordUnavailable. The types are exported from agent_env.task_step.task_steps.verifiers and registered on the plugin surface. Co-Authored-By: Claude Opus 5.5 --- .github/scripts/check_plugin_api.py | 4 + .../task_steps/verifiers/__init__.py | 6 + .../verifiers/env_outcome_verifier.py | 2 +- .../task_steps/verifiers/rubrics_verifier.py | 53 +--- .../task_steps/verifiers/run_context.py | 292 +++++++++++++----- .../test_verifier_run_context_local.py | 14 +- .../task_step/verifiers/test_run_context.py | 204 +++++++----- 7 files changed, 376 insertions(+), 199 deletions(-) diff --git a/.github/scripts/check_plugin_api.py b/.github/scripts/check_plugin_api.py index 962fbd91..1b3208d2 100644 --- a/.github/scripts/check_plugin_api.py +++ b/.github/scripts/check_plugin_api.py @@ -45,6 +45,10 @@ "agent_env.runner.store.ensure_indexes", "agent_env.runner.store.insert_run", "agent_env.runner.store.get_run", "agent_env.runner.store.list_runs", "agent_env.runner.store.count_runs", "agent_env.runner.store.active_runs", "agent_env.runner.store.set_running", "agent_env.runner.store.mark_terminal", + # What a verify script is handed of the run it grades. + "agent_env.task_step.task_steps.verifiers.Run", "agent_env.task_step.task_steps.verifiers.Turn", + "agent_env.task_step.task_steps.verifiers.Message", "agent_env.task_step.task_steps.verifiers.Firing", + "agent_env.task_step.task_steps.verifiers.RunRecordUnavailable", # What an environment provider reads to deploy a built-in env. "agent_env.env.envs.mcp_server.MCPServerEnv.docker_image_artifact", "agent_env.env.envs.mcp_server.MCPServerEnv.environment_name", "agent_env.env.envs.website.WebsiteEnv.backend_docker_image_artifact", diff --git a/src/agent_env/task_step/task_steps/verifiers/__init__.py b/src/agent_env/task_step/task_steps/verifiers/__init__.py index acef73f3..73ae84f8 100644 --- a/src/agent_env/task_step/task_steps/verifiers/__init__.py +++ b/src/agent_env/task_step/task_steps/verifiers/__init__.py @@ -4,6 +4,7 @@ the LLM judge).""" from agent_env.task_step.task_steps.verifiers.scoring import ScoreAggregator, aggregate_score +from agent_env.task_step.task_steps.verifiers.run_context import Firing, Message, Run, RunRecordUnavailable, Turn from agent_env.task_step.task_steps.verifiers.env_outcome_verifier import EnvOutcomeVerifierTaskStep from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep from agent_env.task_step.task_steps.verifiers.agent_prompt_response_verifier import AgentPromptResponseVerifierTaskStep @@ -15,9 +16,14 @@ "AgentPromptResponseVerifierTaskStep", "AggregateVerifiersTaskStep", "EnvOutcomeVerifierTaskStep", + "Firing", + "Message", "RubricsVerifierTaskStep", + "Run", "RunContainerUnitTestsVerifierTaskStep", + "RunRecordUnavailable", "ScoreAggregator", + "Turn", "VerifySandboxTaskStep", "aggregate_score", ] diff --git a/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py b/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py index 750d24f1..3a0c9649 100644 --- a/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/env_outcome_verifier.py @@ -24,7 +24,7 @@ class EnvOutcomeVerifierTaskStep(TaskStep): - """Runs the artifact's ``verify(mcp_url)``; a ``run`` parameter also gets the ``RunContext`` of ``prompt_id``.""" + """Runs the artifact's ``verify(mcp_url)``; a ``run`` parameter also gets the ``Run`` of ``prompt_id``.""" type: ClassVar[str] = "env_outcome_verifier" entity_refs = ( diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py index 7e0958ac..f525b7fe 100644 --- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py @@ -39,7 +39,7 @@ from agent_env.store.base import NotFoundError from agent_env.store.ids import is_local_id from agent_env.task_step.task_step import TaskStep, TaskStepDependency -from agent_env.task_step.task_steps.verifiers.run_context import RunContext, load_run_context +from agent_env.task_step.task_steps.verifiers.run_context import Run, load_run_context, render_run_context from agent_env.task_step.task_steps.verifiers.scoring import ScoreAggregator, aggregate_score from agent_env.task_step.task_steps.sandbox_utils.sandbox_utils import ( agent_error_text, @@ -186,47 +186,6 @@ def _validated_run_context_parts(parts: Optional[list[str]]) -> Optional[list[st return list(dict.fromkeys(parts)) -def _cut(text: str, limit: int) -> str: - return text if len(text) <= limit else text[:limit] + "…" - - -def render_run_context(run: RunContext, parts: list[str]) -> str: - """The judge-prompt sections for ``parts`` of ``run``: the conversation by turn, each firing log one event a line.""" - sections: list[str] = [] - if "conversation" in parts: - lines = ["## Conversation", "The run's messages, turn by turn: the user side (the task's prompt, then each " - "later message of a user simulator, a human or the triggers) and the agent's replies."] - if "conversation" in run.unavailable: - lines.append(f"(not available: {run.unavailable['conversation']})") - for turn in range(1, run.turns + 1): - lines.append(f"### Turn {turn}") - for m in (m for m in run.conversation if m["turn"] == turn): - who = ", ".join(filter(None, [m.get("author"), - f"written by triggers {m['fired']}" if m.get("fired") else None])) - files = sum(1 for p in m["parts"] if p.get("kind") != "text") - extra = f" (+{files} non-text part{'s' if files != 1 else ''})" if files else "" - lines.append(f"- {m['role']}{f' ({who})' if who else ''}: " - f"{_cut(m['text'], RubricsVerifierTaskStep.RUN_CONTEXT_MESSAGE_MAX_CHARS)}{extra}") - sections.append("\n".join(lines)) - if "trigger_log" in parts: - lines = ["## Trigger firings", "What the env's triggers and the user agent's triggers did during the run, " - "oldest first, one firing-log event per line."] - logs = [(f"env {env_id}", events) for env_id, events in run.env_trigger_events.items()] - logs += [("user agent", run.agent_trigger_events)] if run.agent_trigger_events else [] - events_max = RubricsVerifierTaskStep.RUN_CONTEXT_EVENTS_MAX - for name, events in logs: - lines.append(f"### {name}") - if len(events) > events_max: - lines.append(f"({len(events) - events_max} earlier events not shown)") - lines += [f"- {json.dumps({k: v for k, v in e.items() if k != 'seq'}, default=str)}" - for e in events[-events_max:]] - lines += [f"(not available: {what}: {why})" for what, why in run.unavailable.items() if what != "conversation"] - if len(lines) == 2: - lines.append("(no trigger firings were recorded)") - sections.append("\n".join(lines)) - return "\n\n".join(sections) - - # Only fall back from structured output when the error clearly says the response # format itself is unsupported. Image/provider/transient failures should retry as-is. _SCHEMA_TERMS = ( @@ -365,7 +324,7 @@ class RubricsVerifierTaskStep(TaskStep): DEFAULT_MAX_RETRIES: ClassVar[int] = 3 DEFAULT_POLL_INTERVAL_SECONDS: ClassVar[int] = 10 RUN_CONTEXT_MESSAGE_MAX_CHARS: ClassVar[int] = 4000 # a longer message is cut and ends in "…" - RUN_CONTEXT_EVENTS_MAX: ClassVar[int] = 200 # per firing log, the most recent; the rest are counted + RUN_CONTEXT_FIRINGS_MAX: ClassVar[int] = 200 # the most recent trigger firings; the rest are counted TRAJECTORY_CONTAINER_PATH: ClassVar[str] = "/tmp/prompt_trajectory.json" # Multi-turn agent judge: base for a per-run subdir holding turn_NN.json files. TRAJECTORY_CONTAINER_DIR: ClassVar[str] = "/tmp/prompt_trajectory_turns" @@ -598,7 +557,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: if self.run_context_for_judge: run_context = await load_run_context(context, prompt_response) if grade_up_to_turn is not None: - run_context = run_context.as_of_turn(grade_up_to_turn) + run_context = run_context.as_of(grade_up_to_turn) per_turn_uris = [u for u in (prompt_response.target_agent_per_turn_trajectory_object_urls or []) if u] if self.use_trajectory and not (per_turn_uris or prompt_response.agent_trajectory_object_url): @@ -1371,7 +1330,7 @@ def _build_eval_prompt( image_blocks: Optional[list[dict]] = None, per_criterion_frames: bool = False, context_facts: Optional[list[tuple[str, str]]] = None, - run_context: Optional[RunContext] = None, + run_context: Optional[Run] = None, ) -> str: """``context_facts`` are rendered ``(label, value)`` pairs (``resolve_context_facts``); ``image_blocks`` are the blocks attached to the judge message, labels included when ``per_criterion_frames``.""" @@ -1404,7 +1363,9 @@ def _build_eval_prompt( if loaded_artifacts: base += "\n\n" + self._build_loaded_artifacts_block(loaded_artifacts) if run_context is not None and self.run_context_for_judge: - base += "\n\n" + render_run_context(run_context, self.run_context_for_judge) + base += "\n\n" + render_run_context(run_context, self.run_context_for_judge, + message_max_chars=self.RUN_CONTEXT_MESSAGE_MAX_CHARS, + firings_max=self.RUN_CONTEXT_FIRINGS_MAX) if context_facts: # Facts before the policy: they are inputs, the policy is how to read them. base += "\n\n## Context facts\n" + self._CONTEXT_FACTS_LEAD + "\n" + "\n".join( diff --git a/src/agent_env/task_step/task_steps/verifiers/run_context.py b/src/agent_env/task_step/task_steps/verifiers/run_context.py index ebaa4276..af969753 100644 --- a/src/agent_env/task_step/task_steps/verifiers/run_context.py +++ b/src/agent_env/task_step/task_steps/verifiers/run_context.py @@ -1,94 +1,236 @@ -"""What a verifier knows of the run it grades beyond the env: the conversation and the trigger firing logs.""" +"""What a verifier knows of the run it grades beyond the env: the conversation and the trigger firings.""" from __future__ import annotations import asyncio import json +from collections import defaultdict, deque from dataclasses import dataclass, field, replace from datetime import datetime -from typing import Any, ClassVar, Optional +from typing import Any, ClassVar, Literal, Mapping, Optional from agent_env.a2a_agent import conversation_store from agent_env.config import get_config from agent_env.task_step.context import PromptResponse, TaskStepContext +class RunRecordUnavailable(LookupError): + """A query needs a record that could not be read: ``source`` is "conversation" or a ``Firing.source``.""" + + def __init__(self, source: str, reason: str): + super().__init__(f"{source}: {reason}") + self.source, self.reason = source, reason + + +def _need(unavailable: Mapping[str, str], sources) -> None: + for source in sources: + if source in unavailable: + raise RunRecordUnavailable(source, unavailable[source]) + + @dataclass(frozen=True) -class RunContext: - """The prompt's messages and firing logs, oldest first; ``unavailable`` names what could not be read, and why.""" +class Message: + turn: int + role: Literal["user", "agent"] + text: str + sent_at: Optional[datetime] # None only for an episode's turn, which has no stored conversation + author: Optional[str] = None + written_by_triggers: tuple[str, ...] = () # the user agent's triggers that wrote this user message + user_sim: Optional[Mapping[str, Any]] = None # the user sim's extra fields for this turn + parts: tuple[Mapping[str, Any], ...] = () + + +@dataclass(frozen=True) +class Firing: + trigger_id: str + source: str # "env:" or "user_agent" + turn: int # the turn it fell in; 0 = before the first message + at: datetime + virtual_time: Optional[datetime] + ok: bool # fired, rather than failed + provoked_by: Optional[Mapping[str, Any]] # the env call that set it off, from its ``detected`` event + + +@dataclass(frozen=True) +class Turn: + number: int + message: Message # the user-side message that opened it + reply: Optional[Message] + _firings: tuple[Firing, ...] = field(default=(), repr=False) + _unavailable: Mapping[str, str] = field(default_factory=dict, repr=False, compare=False) + + @property + def firings(self) -> tuple[Firing, ...]: + """What fired during this turn; raises ``RunRecordUnavailable`` if a firing log is missing.""" + _need(self._unavailable, self._unavailable) + return self._firings + + +@dataclass(frozen=True) +class Run: + """The graded prompt's turns and every trigger firing; a query that needs a missing record raises.""" ERROR_MAX_CHARS: ClassVar[int] = 200 # of a read error's message, in an ``unavailable`` reason - conversation: list[dict] = field(default_factory=list) - env_trigger_events: dict[str, list[dict]] = field(default_factory=dict) - agent_trigger_events: list[dict] = field(default_factory=list) - unavailable: dict[str, str] = field(default_factory=dict) + _turns: tuple[Turn, ...] = () + _firings: tuple[Firing, ...] = () # oldest first + _logs: tuple[str, ...] = () # the firing logs that were read, as ``Firing.source`` names them + unavailable: Mapping[str, str] = field(default_factory=dict) # what could not be read, and why @property - def turns(self) -> int: - return max((m["turn"] for m in self.conversation), default=0) + def turns(self) -> tuple[Turn, ...]: + _need(self.unavailable, ["conversation"]) + return self._turns + + def turn(self, number: int) -> Turn: + if not 1 <= number <= len(self.turns): + raise LookupError(f"the run has no turn {number}") + return self._turns[number - 1] + + def firings(self, trigger_id: Optional[str] = None, *, env: Optional[str] = None) -> tuple[Firing, ...]: + """Oldest first, of ``trigger_id`` and ``env`` when given; raises ``RunRecordUnavailable`` for a missing log.""" + source = None if env is None else f"env:{env}" + _need(self.unavailable, [source] if source else self.unavailable) + if source and source not in self._logs: + raise RunRecordUnavailable(source, "no firing log was captured") + return tuple(f for f in self._firings + if (trigger_id is None or f.trigger_id == trigger_id) and (source is None or f.source == source)) + + def firing(self, trigger_id: str, *, env: Optional[str] = None) -> Optional[Firing]: + """The first firing of ``trigger_id``, or None if it never fired.""" + return next(iter(self.firings(trigger_id, env=env)), None) - def as_of_turn(self, turn: int) -> RunContext: - """The run after ``turn``: its messages, and the events before the next turn's message (none if it's unknown).""" - if self.conversation and turn >= self.turns: + def as_of(self, turn: int) -> Run: + """The run after ``turn``: its turns, and what fired before the next turn began.""" + if turn < 0: + raise ValueError(f"turn must be >= 0, got {turn}") + if "conversation" in self.unavailable or turn >= len(self._turns): return self - conversation = [m for m in self.conversation if m["turn"] <= turn] - cutoff = next((m["ts"] for m in self.conversation if m["turn"] == turn + 1), None) - if cutoff is None: - why = f"withheld: when turn {turn + 1} began is unknown, so later events can't be told apart" - withheld = [f"env_trigger_events.{env_id}" for env_id in self.env_trigger_events] - withheld += ["agent_trigger_events"] if self.agent_trigger_events else [] - return replace(self, conversation=conversation, env_trigger_events={}, agent_trigger_events=[], - unavailable={**self.unavailable, **dict.fromkeys(withheld, why)}) - - def before(events: list[dict]) -> list[dict]: - return [e for e in events if _before(e.get("ts"), cutoff)] - - return replace( - self, - conversation=conversation, - env_trigger_events={env_id: before(events) for env_id, events in self.env_trigger_events.items()}, - agent_trigger_events=before(self.agent_trigger_events), - ) - - def to_dict(self) -> dict[str, Any]: - return {"conversation": self.conversation, "env_trigger_events": self.env_trigger_events, - "agent_trigger_events": self.agent_trigger_events, "unavailable": self.unavailable} - - -async def load_run_context(context: TaskStepContext, prompt_response: Optional[PromptResponse]) -> RunContext: - """The run context of ``prompt_response``'s prompt (env firing logs alone without one); never raises for a record.""" + return replace(self, _turns=self._turns[:turn], _firings=tuple(f for f in self._firings if f.turn <= turn)) + + +def render_run_context(run: Run, parts: list[str], *, message_max_chars: int, firings_max: int) -> str: + """The judge-prompt sections for ``parts`` of ``run``: the conversation by turn, then one firing a line.""" + sections: list[str] = [] + if "conversation" in parts: + lines = ["## Conversation", "The run's messages, turn by turn: the user side (the task's prompt, then each " + "later message of a user simulator, a human or the triggers) and the agent's replies."] + if "conversation" in run.unavailable: + lines.append(f"(not available: {run.unavailable['conversation']})") + for turn in run._turns: + lines.append(f"### Turn {turn.number}") + lines += [_render_message(m, message_max_chars) for m in (turn.message, turn.reply) if m] + sections.append("\n".join(lines)) + if "trigger_log" in parts: + lines = ["## Trigger firings", "Each time an env's trigger or the user agent's trigger fired or failed, " + "oldest first, with the turn it fell in."] + if len(run._firings) > firings_max: + lines.append(f"({len(run._firings) - firings_max} earlier firings not shown)") + lines += [_render_firing(f) for f in run._firings[max(len(run._firings) - firings_max, 0):]] + lines += [f"(not available: {what}: {why})" for what, why in run.unavailable.items() if what != "conversation"] + if len(lines) == 2: + lines.append("(no trigger firings were recorded)") + sections.append("\n".join(lines)) + return "\n\n".join(sections) + + +def _render_message(m: Message, max_chars: int) -> str: + fired = f"written by triggers {', '.join(m.written_by_triggers)}" if m.written_by_triggers else None + who = ", ".join(filter(None, [m.author, fired])) + files = sum(1 for p in m.parts if p.get("kind") != "text") + extra = f" (+{files} non-text part{'s' if files != 1 else ''})" if files else "" + text = m.text if len(m.text) <= max_chars else m.text[:max_chars] + "…" + return f"- {m.role}{f' ({who})' if who else ''}: {text}{extra}" + + +def _render_firing(f: Firing) -> str: + when = f"during turn {f.turn}" if f.turn else "before the first message" + virtual = f", virtual time {f.virtual_time.isoformat()}" if f.virtual_time else "" + provoked = f", provoked by {json.dumps(f.provoked_by, default=str)}" if f.provoked_by else "" + return f"- {f.source}: {f.trigger_id} {'fired' if f.ok else 'failed'} {when}{virtual}{provoked}" + + +async def load_run_context(context: TaskStepContext, prompt_response: Optional[PromptResponse]) -> Run: + """The ``Run`` of ``prompt_response``'s prompt (env firings alone without one); never raises for a record.""" unavailable: dict[str, str] = {} - conversation: list[dict] = [] + conversation: list[Message] = [] agent_events: list[dict] = [] if prompt_response is not None: conversation = await asyncio.to_thread(_conversation, context, prompt_response, unavailable) agent_events = _agent_trigger_events(context, prompt_response.step_id, conversation, unavailable) env_events = await asyncio.to_thread(_env_trigger_events, context, unavailable) - run = RunContext(conversation=conversation, env_trigger_events=env_events, agent_trigger_events=agent_events, - unavailable=unavailable) - return await asyncio.to_thread(_until_next_prompt, run, context, prompt_response) if prompt_response else run + if prompt_response is not None: + env_events = await asyncio.to_thread(_until_next_prompt, env_events, context, prompt_response, unavailable) + return _run(conversation, {**{f"env:{e}": events for e, events in env_events.items()}, "user_agent": agent_events}, + unavailable) def _error(e: Exception) -> str: - return f"{type(e).__name__}: {str(e)[:RunContext.ERROR_MAX_CHARS]}" + return f"{type(e).__name__}: {str(e)[:Run.ERROR_MAX_CHARS]}" + +def _time(ts: Optional[str]) -> Optional[datetime]: + return datetime.fromisoformat(ts) if ts else None -def _agent_trigger_events(context: TaskStepContext, step_id: Optional[str], conversation: list[dict], + +def _run(conversation: list[Message], logs: dict[str, list[dict]], unavailable: dict[str, str]) -> Run: + """Places every firing in its turn; without the conversation the firing logs are withheld, as they can't be.""" + if "conversation" in unavailable: + why = "withheld: without the conversation, its firings can't be placed in turns" + unavailable.update({source: why for source, events in logs.items() if events and source not in unavailable}) + starts = [m.sent_at for m in conversation if m.role == "user"] + firings: list[Firing] = [] + for source, events in logs.items(): + if source in unavailable: + continue + try: + firings += _fold(source, events, starts) + except (KeyError, TypeError, ValueError) as e: + unavailable[source] = f"unreadable firing log: {_error(e)}" + firings.sort(key=lambda f: f.at) + opened = [m for m in conversation if m.role == "user"] + replies = {m.turn: m for m in reversed(conversation) if m.role == "agent"} + turns = tuple(Turn(number=m.turn, message=m, reply=replies.get(m.turn), _unavailable=unavailable, + _firings=tuple(f for f in firings if f.turn == m.turn)) for m in opened) + return Run(_turns=turns, _firings=tuple(firings), _logs=tuple(s for s in logs if s not in unavailable), + unavailable=unavailable) + + +def _fold(source: str, events: list[dict], starts: list[Optional[datetime]]) -> list[Firing]: + """One ``Firing`` per ``fired`` or ``failed`` event, provoked by its trigger's oldest unclaimed ``detected``.""" + detected: dict[str, deque] = defaultdict(deque) + provoked: dict[str, Any] = {} # a time trigger's one ``detected`` provokes each of its marks' fires + firings = [] + for event in events: + trigger_id, kind = event.get("trigger_id"), event.get("kind") + if trigger_id and kind == "detected": + detected[trigger_id].append(event.get("provoking")) + elif trigger_id and kind in ("fired", "failed"): + pending = detected[trigger_id] + provoked[trigger_id] = pending.popleft() if pending else provoked.get(trigger_id) + at = datetime.fromisoformat(event["ts"]) + firings.append(Firing(trigger_id=trigger_id, source=source, at=at, ok=kind == "fired", + turn=sum(1 for s in starts if s is None or s <= at), + virtual_time=_time(event.get("virtual_time")), provoked_by=provoked[trigger_id])) + return firings + + +def _agent_trigger_events(context: TaskStepContext, step_id: Optional[str], conversation: list[Message], unavailable: dict[str, str]) -> list[dict]: """The user agent's firing log; unavailable, not empty, when it had triggers but ``prompt_agent`` read none back.""" logs = context.metadata.get("agent_trigger_state") or {} if step_id in logs: return list(logs[step_id] or []) armed = {r.get("agent_name") for r in context.metadata.get("agent_trigger_registrations") or []} - user_agent = next((m.get("author") for m in conversation if m["role"] == "user"), None) + user_agent = next((m.author for m in conversation if m.role == "user"), None) # Without the conversation the user agent is unknown, so any agent's registered triggers may be its. had_triggers = bool(armed) if "conversation" in unavailable else user_agent in armed if had_triggers or step_id in (context.metadata.get("agent_trigger_firings") or {}): - unavailable["agent_trigger_events"] = "agent triggers were registered, but the user agent's log wasn't read" + unavailable["user_agent"] = "agent triggers were registered, but the user agent's log wasn't read" return [] -def _conversation(context: TaskStepContext, prompt_response: PromptResponse, unavailable: dict[str, str]) -> list[dict]: +def _conversation(context: TaskStepContext, prompt_response: PromptResponse, + unavailable: dict[str, str]) -> list[Message]: """The stored conversation, numbered by user message; without one (an episode's turn), its prompt and reply.""" conversation_id = (context.metadata.get("a2a_conversations") or {}).get(prompt_response.step_id) if conversation_id is None: @@ -107,26 +249,29 @@ def _conversation(context: TaskStepContext, prompt_response: PromptResponse, una if f.get("fired")} fields = {f["turn"] + 1: f.get("fields") for f in (context.metadata.get("usersim_turn_outputs") or {}).get(step_id) or []} authors = {"user": doc.get("source_agent_name"), "agent": doc.get("target_agent_name")} - messages: list[dict] = [] + messages: list[Message] = [] turn = 0 - for stored in doc.get("messages") or []: - role = stored.get("role") - if role == "user": - turn += 1 - message = _message(turn, role, stored.get("parts") or [], ts=stored.get("ts"), author=authors.get(role)) - if role == "user" and turn in fired: - message["fired"] = fired[turn] - if role == "user" and turn in fields: - message["fields"] = fields[turn] - messages.append(message) + try: + for stored in doc.get("messages") or []: + role = stored.get("role") + if role == "user": + turn += 1 + if role in ("user", "agent") and turn: + user = role == "user" + messages.append(_message(turn, role, stored.get("parts") or [], ts=stored.get("ts"), + author=authors.get(role), fired=fired.get(turn) if user else None, + user_sim=fields.get(turn) if user else None)) + except (TypeError, ValueError) as e: + unavailable["conversation"] = f"unreadable conversation: {_error(e)}" + return [] return messages -def _message(turn: int, role: str, parts: list[dict], *, ts: Optional[str] = None, - author: Optional[str] = None) -> dict: +def _message(turn: int, role: str, parts: list[dict], *, ts: Optional[str] = None, author: Optional[str] = None, + fired: Optional[list[str]] = None, user_sim: Optional[Mapping[str, Any]] = None) -> Message: text = "\n".join(p.get("text", "") for p in parts if p.get("kind") == "text") - return {"turn": turn, "role": role, **({"author": author} if author else {}), "text": text, "parts": parts, - "ts": ts} + return Message(turn=turn, role=role, text=text, sent_at=_time(ts), author=author or None, + written_by_triggers=tuple(fired or ()), user_sim=user_sim, parts=tuple(parts)) def _env_trigger_events(context: TaskStepContext, unavailable: dict[str, str]) -> dict[str, list[dict]]: @@ -134,30 +279,29 @@ def _env_trigger_events(context: TaskStepContext, unavailable: dict[str, str]) - events: dict[str, list[dict]] = {} for env_id, entry in (context.metadata.get("env_trigger_state") or {}).items(): if not isinstance(entry, dict) or not entry.get("object_url"): - unavailable[f"env_trigger_events.{env_id}"] = (entry or {}).get("error") or "no firing log was captured" + unavailable[f"env:{env_id}"] = (entry or {}).get("error") or "no firing log was captured" continue try: envelope = json.loads(get_config().get_object_store().get(entry["object_url"])) except Exception as e: - unavailable[f"env_trigger_events.{env_id}"] = _error(e) + unavailable[f"env:{env_id}"] = _error(e) continue events[env_id] = list((envelope.get("state") or {}).get("events") or []) return events -def _until_next_prompt(run: RunContext, context: TaskStepContext, prompt_response: PromptResponse) -> RunContext: - """The env firing logs are the whole run's (the last prompt's capture), so an earlier prompt's stop where the - next prompt began; withheld when that start is unknown.""" +def _until_next_prompt(env_events: dict[str, list[dict]], context: TaskStepContext, prompt_response: PromptResponse, + unavailable: dict[str, str]) -> dict[str, list[dict]]: + """An earlier prompt's env firing logs, cut where the next prompt began; withheld when that start is unknown.""" index = next((i for i, pr in enumerate(context.prompt_responses) if pr is prompt_response), None) - if index is None or index + 1 >= len(context.prompt_responses) or not run.env_trigger_events: - return run + if index is None or index + 1 >= len(context.prompt_responses) or not env_events: + return env_events start = _first_message_ts(context, context.prompt_responses[index + 1]) if start is None: why = "withheld: when the next prompt began is unknown, so its events can't be told apart" - withheld = dict.fromkeys((f"env_trigger_events.{env_id}" for env_id in run.env_trigger_events), why) - return replace(run, env_trigger_events={}, unavailable={**run.unavailable, **withheld}) - return replace(run, env_trigger_events={env_id: [e for e in events if _before(e.get("ts"), start)] - for env_id, events in run.env_trigger_events.items()}) + unavailable.update(dict.fromkeys((f"env:{env_id}" for env_id in env_events), why)) + return {} + return {env_id: [e for e in events if _before(e.get("ts"), start)] for env_id, events in env_events.items()} def _first_message_ts(context: TaskStepContext, prompt_response: PromptResponse) -> Optional[str]: diff --git a/tst/integration/task_step/test_verifier_run_context_local.py b/tst/integration/task_step/test_verifier_run_context_local.py index 3a0b073a..68c12421 100644 --- a/tst/integration/task_step/test_verifier_run_context_local.py +++ b/tst/integration/task_step/test_verifier_run_context_local.py @@ -1,4 +1,4 @@ -"""A real two-turn conversation (echo agent as solver and user-sim) reads back whole through load_run_context.""" +"""A real two-turn conversation (echo agent as solver and user-sim) reads back whole as a ``Run``.""" import uuid @@ -28,9 +28,9 @@ async def test_the_run_context_reads_back_every_turn_of_a_real_conversation(loca assert not context.metadata.get("failed_steps"), context.metadata.get("failed_steps") run = await load_run_context(context, next(pr for pr in context.prompt_responses if pr.prompt_id == "ask")) - assert run.unavailable == {} and run.turns == 2 - assert [(m["turn"], m["role"], m.get("author")) for m in run.conversation] == [ - (1, "user", "user"), (1, "agent", "solver"), (2, "user", "user"), (2, "agent", "solver")] - assert run.conversation[0]["text"] == "Plan a lunch for Friday." - assert run.conversation[2]["text"] and run.conversation[2]["text"] != run.conversation[0]["text"] - assert run.as_of_turn(1).conversation == run.conversation[:2] + assert run.unavailable == {} and len(run.turns) == 2 + assert [(t.number, t.message.author, t.reply.author) for t in run.turns] == [(1, "user", "solver"), (2, "user", "solver")] + assert run.turn(1).message.text == "Plan a lunch for Friday." + assert run.turn(2).message.text and run.turn(2).message.text != run.turn(1).message.text + assert run.turn(1).message.sent_at < run.turn(1).reply.sent_at < run.turn(2).message.sent_at + assert run.as_of(1).turns == run.turns[:1] and run.firings() == () diff --git a/tst/unit/task_step/verifiers/test_run_context.py b/tst/unit/task_step/verifiers/test_run_context.py index 95a227be..8541aa21 100644 --- a/tst/unit/task_step/verifiers/test_run_context.py +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -15,8 +15,10 @@ from agent_env.config import get_config from agent_env.task_step.context import PromptResponse, TaskStepContext from agent_env.task_step.task_steps.verifiers.env_outcome_verifier import EnvOutcomeVerifierTaskStep -from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep, render_run_context -from agent_env.task_step.task_steps.verifiers.run_context import RunContext, load_run_context +from agent_env.task_step.task_steps.verifiers import Run, RunRecordUnavailable +from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep +from agent_env.task_step.task_steps.verifiers.run_context import (_fold, _message, _run, load_run_context, + render_run_context) TURNS = [("Can you book lunch with Dana?", "Booked for noon."), ("Make it 12:30 instead.", "Moved to 12:30."), @@ -32,6 +34,11 @@ def _between(earlier: str, later: str) -> str: return (a + (b - a) / 2).isoformat() +def _render(run: Run, parts: list[str]) -> str: + return render_run_context(run, parts, message_max_chars=RubricsVerifierTaskStep.RUN_CONTEXT_MESSAGE_MAX_CHARS, + firings_max=RubricsVerifierTaskStep.RUN_CONTEXT_FIRINGS_MAX) + + def recorded_run(turns=TURNS) -> tuple[TaskStepContext, PromptResponse, dict]: """A three-turn run as prompt_agent records it: conversation, trigger and user-sim marks, and both firing logs.""" conversation_id = f"conv-{uuid.uuid4().hex[:8]}" @@ -46,7 +53,8 @@ def recorded_run(turns=TURNS) -> tuple[TaskStepContext, PromptResponse, dict]: turn_starts = [m["ts"] for m in messages if m["role"] == "user"] after_reply_1 = _between(messages[1]["ts"], turn_starts[1]) env_events = [ - {"seq": 1, "ts": after_reply_1, "kind": "detected", "trigger_id": "confirm"}, + {"seq": 1, "ts": after_reply_1, "kind": "detected", "trigger_id": "confirm", + "provoking": {"tool": "book", "role": "solver"}}, {"seq": 2, "ts": after_reply_1, "kind": "fired", "trigger_id": "confirm", "fire_count": 1}, {"seq": 3, "ts": _between(turn_starts[2], messages[5]["ts"]), "kind": "fired", "trigger_id": "late", "fire_count": 1}, @@ -69,38 +77,60 @@ def recorded_run(turns=TURNS) -> tuple[TaskStepContext, PromptResponse, dict]: @pytest.mark.asyncio -async def test_the_run_context_has_every_turn_and_both_firing_logs(): +async def test_the_run_has_every_turn_and_one_firing_per_fire_each_in_its_turn(): context, prompt_response, recorded = recorded_run() run = await load_run_context(context, prompt_response) - assert run.turns == 3 and run.unavailable == {} - assert [(m["turn"], m["role"], m.get("author"), m["text"]) for m in run.conversation] == [ - (1, "user", "dana_sim", TURNS[0][0]), (1, "agent", "solver", TURNS[0][1]), - (2, "user", "dana_sim", TURNS[1][0]), (2, "agent", "solver", TURNS[1][1]), - (3, "user", "dana_sim", TURNS[2][0]), (3, "agent", "solver", TURNS[2][1]), + assert len(run.turns) == 3 and run.unavailable == {} + assert [(t.number, t.message.author, t.message.text, t.reply.author, t.reply.text) for t in run.turns] == [ + (n, "dana_sim", user, "solver", agent) for n, (user, agent) in enumerate(TURNS, 1)] + assert [t.message.written_by_triggers for t in run.turns] == [(), ("reschedule",), ()] + assert [t.message.user_sim for t in run.turns] == [None, None, {"mood": "satisfied"}] + assert run.turn(2).message.sent_at == datetime.fromisoformat(recorded["turn_starts"][1]) + assert [(f.source, f.trigger_id, f.turn, f.ok) for f in run.firings()] == [ + ("env:calendar", "confirm", 1, True), ("user_agent", "reschedule", 1, True), ("env:calendar", "late", 3, True)] + confirm = run.firing("confirm", env="calendar") + assert confirm.provoked_by == {"tool": "book", "role": "solver"} and confirm.at == datetime.fromisoformat( + recorded["env"][1]["ts"]) + assert run.firing("late").provoked_by is None and run.firing("never") is None + assert [f.trigger_id for f in run.turn(1).firings] == ["confirm", "reschedule"] and run.turn(2).firings == () + assert run.firings(env="calendar") == run.firings("confirm") + run.firings("late") + + +def test_the_gateway_log_folds_into_one_firing_per_fire(): + ts = [f"2026-10-20T10:00:0{i}+00:00" for i in range(8)] + events = [ + {"seq": 1, "ts": ts[0], "kind": "detected", "trigger_id": "mirror", "provoking": {"tool": "a"}}, + {"seq": 2, "ts": ts[1], "kind": "detected", "trigger_id": "mirror", "provoking": {"tool": "b"}, "queued": 1}, + {"seq": 3, "ts": ts[2], "kind": "action_ok", "trigger_id": "mirror", "action_index": 0}, + {"seq": 4, "ts": ts[3], "kind": "fired", "trigger_id": "mirror", "fire_count": 1}, + {"seq": 5, "ts": ts[4], "kind": "failed", "trigger_id": "mirror", "action_index": 0}, + {"seq": 6, "ts": ts[5], "kind": "detected", "trigger_id": "tick", "provoking": {"tool": "c"}, "due": 2, + "virtual_time": "2026-12-03T07:00:00Z"}, + {"seq": 7, "ts": ts[6], "kind": "fired", "trigger_id": "tick", "virtual_time": "2026-12-03T07:00:00Z"}, + {"seq": 8, "ts": ts[7], "kind": "fired", "trigger_id": "tick", "virtual_time": "2026-12-03T08:00:00Z"}, ] - user = [m for m in run.conversation if m["role"] == "user"] - assert "fired" not in user[0] and user[1]["fired"] == ["reschedule"] and "fired" not in user[2] - assert user[2]["fields"] == {"mood": "satisfied"} and "fields" not in user[1] - assert run.env_trigger_events == {"calendar": recorded["env"]} - assert run.agent_trigger_events == recorded["agent"] - assert json.loads(json.dumps(run.to_dict()))["conversation"][0]["parts"] == _text(TURNS[0][0]) + + firings = _fold("env:calendar", events, []) + + assert [(f.trigger_id, f.ok, f.provoked_by, f.turn) for f in firings] == [ + ("mirror", True, {"tool": "a"}, 0), ("mirror", False, {"tool": "b"}, 0), + ("tick", True, {"tool": "c"}, 0), ("tick", True, {"tool": "c"}, 0)] + assert firings[3].virtual_time == datetime.fromisoformat("2026-12-03T08:00:00+00:00") @pytest.mark.asyncio async def test_as_of_a_turn_the_run_stops_before_the_next_turn_begins(): - context, prompt_response, recorded = recorded_run() + context, prompt_response, _ = recorded_run() run = await load_run_context(context, prompt_response) - first, second = run.as_of_turn(1), run.as_of_turn(2) + first, second = run.as_of(1), run.as_of(2) - assert [m["turn"] for m in first.conversation] == [1, 1] - assert first.env_trigger_events == {"calendar": recorded["env"][:2]} - assert first.agent_trigger_events == recorded["agent"] - assert [m["turn"] for m in second.conversation] == [1, 1, 2, 2] - assert second.env_trigger_events == {"calendar": recorded["env"][:2]} - assert run.as_of_turn(3) is run and run.as_of_turn(7) is run + assert [t.number for t in first.turns] == [1] and [t.number for t in second.turns] == [1, 2] + assert [f.trigger_id for f in first.firings()] == ["confirm", "reschedule"] == [ + f.trigger_id for f in second.firings()] + assert run.as_of(3) is run and run.as_of(7) is run @pytest.mark.asyncio @@ -109,12 +139,13 @@ async def test_a_prompt_with_no_stored_conversation_is_one_turn_of_its_prompt_an run = await load_run_context(TaskStepContext(prompt_responses=[prompt_response]), prompt_response) - assert [(m["turn"], m["role"], m["text"]) for m in run.conversation] == [(1, "user", "do it"), (1, "agent", "done")] - assert run.env_trigger_events == {} and run.agent_trigger_events == [] and run.unavailable == {} + [turn] = run.turns + assert (turn.message.role, turn.message.text, turn.reply.role, turn.reply.text) == ("user", "do it", "agent", "done") + assert run.firings() == () and run.unavailable == {} @pytest.mark.asyncio -async def test_what_the_run_meant_to_record_but_cannot_be_read_is_named_not_left_empty(): +async def test_what_the_run_meant_to_record_but_cannot_be_read_raises_rather_than_reading_as_empty(): context, prompt_response, _ = recorded_run() context.metadata["a2a_conversations"]["prompt"] = "conv-gone" context.metadata["env_trigger_state"] = { @@ -124,10 +155,38 @@ async def test_what_the_run_meant_to_record_but_cannot_be_read_is_named_not_left run = await load_run_context(context, prompt_response) - assert run.conversation == [] and run.env_trigger_events == {} assert run.unavailable["conversation"] == "conversation 'conv-gone' is not in the store" - assert run.unavailable["env_trigger_events.calendar"] == "TimeoutError: capture budget of 90s exceeded" - assert "env_trigger_events.mail" in run.unavailable + assert run.unavailable["env:calendar"] == "TimeoutError: capture budget of 90s exceeded" + assert "env:mail" in run.unavailable + with pytest.raises(RunRecordUnavailable, match="conv-gone") as raised: + run.turns + assert raised.value.source == "conversation" + with pytest.raises(RunRecordUnavailable, match="capture budget") as raised: + run.firing("confirm", env="calendar") + assert raised.value.source == "env:calendar" + with pytest.raises(RunRecordUnavailable): + run.firings() + + +@pytest.mark.asyncio +async def test_an_env_with_no_firing_log_raises_and_a_missing_log_reaches_the_turn_too(): + context, prompt_response, _ = recorded_run() + context.metadata["env_trigger_state"]["mail"] = {"error": "TimeoutError: capture budget of 90s exceeded"} + run = await load_run_context(context, prompt_response) + + with pytest.raises(RunRecordUnavailable, match="no firing log was captured"): + run.firings(env="drive") + with pytest.raises(RunRecordUnavailable, match="capture budget"): + run.turn(1).firings + assert [f.trigger_id for f in run.firings(env="calendar")] == ["confirm", "late"] + + +def test_a_firing_log_that_cannot_be_parsed_is_named_not_left_empty(): + run = _run([], {"env:calendar": [{"seq": 1, "kind": "fired", "trigger_id": "confirm"}]}, {}) + + assert run.unavailable["env:calendar"].startswith("unreadable firing log: KeyError") + with pytest.raises(RunRecordUnavailable): + run.firings(env="calendar") @pytest.mark.asyncio @@ -139,9 +198,12 @@ async def test_a_user_agent_firing_log_that_was_not_read_back_is_named_not_left_ run = await load_run_context(context, prompt_response) - assert run.agent_trigger_events == [] and "agent_trigger_events" in run.unavailable - text = render_run_context(run, ["trigger_log"]) - assert f"(not available: agent_trigger_events: {run.unavailable['agent_trigger_events']})" in text + with pytest.raises(RunRecordUnavailable, match="wasn't read") as raised: + run.firing("reschedule") + assert raised.value.source == "user_agent" + assert [f.trigger_id for f in run.firings(env="calendar")] == ["confirm", "late"] + text = _render(run, ["trigger_log"]) + assert f"(not available: user_agent: {run.unavailable['user_agent']})" in text assert "(no trigger firings were recorded)" not in text @@ -154,35 +216,35 @@ async def test_a_user_agent_with_no_triggers_of_its_own_has_no_firing_log_to_mis run = await load_run_context(context, prompt_response) - assert run.agent_trigger_events == [] and run.unavailable == {} + assert run.firing("reschedule") is None and run.unavailable == {} @pytest.mark.asyncio -async def test_without_the_conversation_a_turn_cut_withholds_the_firing_logs_it_cannot_cut(): - context, prompt_response, recorded = recorded_run() +async def test_without_the_conversation_the_firings_that_cannot_be_placed_in_turns_are_withheld(): + context, prompt_response, _ = recorded_run() context.metadata["a2a_conversations"]["prompt"] = "conv-gone" - run = await load_run_context(context, prompt_response) - first = run.as_of_turn(1) + run = await load_run_context(context, prompt_response) - assert run.env_trigger_events == {"calendar": recorded["env"]} and run.agent_trigger_events == recorded["agent"] - assert first.conversation == [] and first.env_trigger_events == {} and first.agent_trigger_events == [] - assert first.unavailable["conversation"] == run.unavailable["conversation"] - assert first.unavailable["env_trigger_events.calendar"] == first.unavailable["agent_trigger_events"] + assert run.unavailable["env:calendar"] == run.unavailable["user_agent"] + assert run.unavailable["env:calendar"].startswith("withheld:") and run.as_of(1) is run + with pytest.raises(RunRecordUnavailable, match="withheld"): + run.firings(env="calendar") @pytest.mark.asyncio -async def test_without_a_prompt_the_run_context_has_the_env_firing_logs_alone(): - context, _, recorded = recorded_run() +async def test_without_a_prompt_the_run_has_the_env_firings_alone(): + context, _, _ = recorded_run() run = await load_run_context(context, None) - assert run.conversation == [] and run.agent_trigger_events == [] - assert run.env_trigger_events == {"calendar": recorded["env"]} + assert run.turns == () and [(f.trigger_id, f.turn) for f in run.firings()] == [("confirm", 0), ("late", 0)] -def test_an_empty_run_context_has_no_turns(): - assert RunContext().turns == 0 and RunContext().as_of_turn(1) == RunContext() +def test_an_empty_run_has_no_turns(): + assert Run().turns == () and Run().as_of(1) == Run() + with pytest.raises(LookupError, match="no turn 1"): + Run().turn(1) # ---------------------------------------------------------------- the judge sees it @@ -219,18 +281,18 @@ async def fake_run(*, eval_prompt, **kwargs): async def test_the_judge_sees_the_conversation_and_the_firing_logs_it_asks_for(monkeypatch): v = _verifier(run_context_for_judge=["conversation", "trigger_log"]) captured = _judge_prompt(monkeypatch, v) - context, recorded = _judged_run() + context, _ = _judged_run() await v.execute(context) prompt = captured["eval_prompt"] conversation = prompt[prompt.index("## Conversation"):prompt.index("## Trigger firings")] assert all(user in conversation and agent in conversation for user, agent in TURNS) - assert f"- user (dana_sim, written by triggers ['reschedule']): {TURNS[1][0]}" in conversation + assert f"- user (dana_sim, written by triggers reschedule): {TURNS[1][0]}" in conversation assert f"- agent (solver): {TURNS[2][1]}" in conversation and "### Turn 3" in conversation log = prompt[prompt.index("## Trigger firings"):] - assert "### env calendar" in log and "### user agent" in log - assert json.dumps({k: v for k, v in recorded["env"][2].items() if k != "seq"}) in log and '"seq"' not in log + assert '- env:calendar: confirm fired during turn 1, provoked by {"tool": "book", "role": "solver"}' in log + assert "- user_agent: reschedule fired during turn 1\n- env:calendar: late fired during turn 3" in log @pytest.mark.asyncio @@ -243,7 +305,7 @@ async def test_graded_as_of_a_turn_the_judge_sees_neither_later_turns_nor_later_ prompt = captured["eval_prompt"] assert TURNS[1][1] in prompt and "### Turn 3" not in prompt and TURNS[2][0] not in prompt - assert '"trigger_id": "confirm"' in prompt and '"trigger_id": "late"' not in prompt + assert "confirm fired during turn 1" in prompt and "late fired" not in prompt @pytest.mark.asyncio @@ -256,9 +318,8 @@ async def test_graded_as_of_a_turn_without_the_conversation_the_judge_sees_no_fi await v.execute(context) prompt = captured["eval_prompt"] - assert '"trigger_id"' not in prompt - assert "(not available: env_trigger_events.calendar: withheld" in prompt - assert "(not available: agent_trigger_events: withheld" in prompt + assert " fired during " not in prompt + assert "(not available: env:calendar: withheld" in prompt and "(not available: user_agent: withheld" in prompt @pytest.mark.asyncio @@ -275,18 +336,17 @@ async def test_the_judge_sees_only_the_parts_it_asks_for(monkeypatch): def test_a_long_message_and_a_long_firing_log_are_bounded_and_a_missing_log_is_named(): max_chars = RubricsVerifierTaskStep.RUN_CONTEXT_MESSAGE_MAX_CHARS long = "x" * (max_chars + 50) - run = RunContext( - conversation=[{"turn": 1, "role": "user", "text": long, "parts": _text(long) + [{"kind": "file"}], "ts": None}], - env_trigger_events={"calendar": [{"seq": i, "kind": "fired", "trigger_id": f"t{i}"} - for i in range(RubricsVerifierTaskStep.RUN_CONTEXT_EVENTS_MAX + 3)]}, - unavailable={"env_trigger_events.mail": "TimeoutError: capture budget of 90s exceeded"}) + events = [{"seq": i, "ts": f"2026-10-20T10:{i // 60:02d}:{i % 60:02d}+00:00", "kind": "fired", "trigger_id": f"t{i}"} + for i in range(RubricsVerifierTaskStep.RUN_CONTEXT_FIRINGS_MAX + 3)] + run = _run([_message(1, "user", _text(long) + [{"kind": "file"}])], {"env:calendar": events}, + {"env:mail": "TimeoutError: capture budget of 90s exceeded"}) - text = render_run_context(run, ["conversation", "trigger_log"]) + text = _render(run, ["conversation", "trigger_log"]) assert "x" * max_chars + "… (+1 non-text part)" in text and long not in text - assert "(3 earlier events not shown)" in text and '"t2"' not in text and '"t3"' in text - assert "(not available: env_trigger_events.mail: TimeoutError: capture budget of 90s exceeded)" in text - assert "(no trigger firings were recorded)" in render_run_context(RunContext(), ["trigger_log"]) + assert "(3 earlier firings not shown)" in text and " t2 fired" not in text and " t3 fired during turn 1" in text + assert "(not available: env:mail: TimeoutError: capture budget of 90s exceeded)" in text + assert "(no trigger firings were recorded)" in _render(Run(), ["trigger_log"]) @pytest.mark.parametrize("bad", [[], ["conversation", "screens"], "conversation", ["Conversation"]]) @@ -305,11 +365,13 @@ def test_run_context_for_judge_round_trips_and_is_omitted_when_unset(): # ---------------------------------------------------------------- a verify(mcp_url, run) script gets it RUN_SCRIPT = b""" -async def verify(mcp_url, run): - rescheduled = [m for m in run.conversation if m.get("fired") == ["reschedule"]] - return [{"id": "reschedule-reached-the-agent", "score": 1.0, "result": bool(rescheduled), - "turns": run.turns, "env_fired": [e["trigger_id"] for e in run.env_trigger_events["calendar"] - if e["kind"] == "fired"]}] +from agent_env.task_step.task_steps.verifiers import Run + +async def verify(mcp_url: str, run: Run | None = None): + late = run.firing("late", env="calendar") + reply = run.turn(late.turn).reply if late else None + return [{"id": "answered-after-late", "score": 1.0, "result": bool(reply and "help" in reply.text), + "turns": len(run.turns), "env_fired": [f.trigger_id for f in run.firings(env="calendar") if f.ok]}] """ @@ -412,5 +474,5 @@ async def test_an_earlier_prompt_sees_only_the_env_events_from_before_the_next_p graded_first = await load_run_context(context, context.prompt_responses[0]) graded_last = await load_run_context(context, context.prompt_responses[1]) - assert [e["trigger_id"] for e in graded_first.env_trigger_events["calendar"]] == ["lunch_confirm"] - assert [e["trigger_id"] for e in graded_last.env_trigger_events["calendar"]] == ["lunch_confirm", "dinner_confirm"] + assert [f.trigger_id for f in graded_first.firings(env="calendar")] == ["lunch_confirm"] + assert [f.trigger_id for f in graded_last.firings(env="calendar")] == ["lunch_confirm", "dinner_confirm"] From 719fe2c65643411db1431ad4596b18262be6b9e8 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 11:02:12 -0700 Subject: [PATCH 7/7] review: an earlier prompt sharing the next prompt's conversation has its env firings withheld, not cut at the shared start Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/verifiers/run_context.py | 6 +++++- tst/unit/task_step/verifiers/test_run_context.py | 12 ++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/src/agent_env/task_step/task_steps/verifiers/run_context.py b/src/agent_env/task_step/task_steps/verifiers/run_context.py index af969753..d41fe30f 100644 --- a/src/agent_env/task_step/task_steps/verifiers/run_context.py +++ b/src/agent_env/task_step/task_steps/verifiers/run_context.py @@ -296,7 +296,11 @@ def _until_next_prompt(env_events: dict[str, list[dict]], context: TaskStepConte index = next((i for i, pr in enumerate(context.prompt_responses) if pr is prompt_response), None) if index is None or index + 1 >= len(context.prompt_responses) or not env_events: return env_events - start = _first_message_ts(context, context.prompt_responses[index + 1]) + following = context.prompt_responses[index + 1] + conversations = context.metadata.get("a2a_conversations") or {} + # Prompts sharing one conversation (one context_id) leave no boundary in it to cut at. + shared = conversations.get(following.step_id) == conversations.get(prompt_response.step_id) + start = None if shared else _first_message_ts(context, following) if start is None: why = "withheld: when the next prompt began is unknown, so its events can't be told apart" unavailable.update(dict.fromkeys((f"env:{env_id}" for env_id in env_events), why)) diff --git a/tst/unit/task_step/verifiers/test_run_context.py b/tst/unit/task_step/verifiers/test_run_context.py index 8541aa21..fa7c6cd8 100644 --- a/tst/unit/task_step/verifiers/test_run_context.py +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -476,3 +476,15 @@ async def test_an_earlier_prompt_sees_only_the_env_events_from_before_the_next_p assert [f.trigger_id for f in graded_first.firings(env="calendar")] == ["lunch_confirm"] assert [f.trigger_id for f in graded_last.firings(env="calendar")] == ["lunch_confirm", "dinner_confirm"] + + +@pytest.mark.asyncio +async def test_an_earlier_prompt_sharing_the_next_prompts_conversation_has_its_env_firings_withheld(): + context, prompt_response, _ = recorded_run() + context.prompt_responses.append(PromptResponse(prompt_id="p2", response="ok", prompt_text="again", step_id="again")) + context.metadata["a2a_conversations"]["again"] = context.metadata["a2a_conversations"]["prompt"] + + run = await load_run_context(context, prompt_response) + + with pytest.raises(RunRecordUnavailable, match="when the next prompt began is unknown"): + run.firings(env="calendar")