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 984f35d0..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 @@ -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,8 @@ class EnvOutcomeVerifierTaskStep(TaskStep): + """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 = ( EntityRef.env("env_id"), @@ -37,11 +41,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 +62,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 +95,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 +119,11 @@ 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 "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) score = aggregate_score(results, self.score_aggregator) if "verifications" not in context.metadata: @@ -125,3 +138,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 f0eda557..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,6 +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 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, @@ -173,6 +174,18 @@ 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. +RUN_CONTEXT_PARTS = ("conversation", "trigger_log") + + +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)) + + # 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 = ( @@ -310,6 +323,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_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" @@ -347,10 +362,13 @@ 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) + # 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. self.grade_up_to_turn = _validated_turn(grade_up_to_turn) @@ -468,6 +486,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 @@ -497,6 +517,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]: @@ -532,6 +553,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(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): @@ -704,6 +730,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)) @@ -724,6 +751,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( @@ -1302,6 +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[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``.""" @@ -1333,6 +1362,10 @@ 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, + 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 new file mode 100644 index 00000000..d41fe30f --- /dev/null +++ b/src/agent_env/task_step/task_steps/verifiers/run_context.py @@ -0,0 +1,324 @@ +"""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, 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 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 + + _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) -> 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(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 + 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[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) + 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)[:Run.ERROR_MAX_CHARS]}" + + +def _time(ts: Optional[str]) -> Optional[datetime]: + return datetime.fromisoformat(ts) if ts else None + + +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.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["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[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: + 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"] = _error(e) + 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[Message] = [] + turn = 0 + 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, + 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 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]]: + """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:{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:{env_id}"] = _error(e) + continue + events[env_id] = list((envelope.get("state") or {}).get("events") or []) + return events + + +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 env_events: + return env_events + 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)) + 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]: + 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 + 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..68c12421 --- /dev/null +++ b/tst/integration/task_step/test_verifier_run_context_local.py @@ -0,0 +1,36 @@ +"""A real two-turn conversation (echo agent as solver and user-sim) reads back whole as a ``Run``.""" + +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 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 new file mode 100644 index 00000000..fa7c6cd8 --- /dev/null +++ b/tst/unit/task_step/verifiers/test_run_context.py @@ -0,0 +1,490 @@ +"""The run a verifier grades: the stored conversation, the firing logs, and the cut as of a turn.""" +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 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."), + ("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 _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]}" + 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", + "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}, + ] + 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_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") + 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_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 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"}, + ] + + 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_run() + run = await load_run_context(context, prompt_response) + + first, second = run.as_of(1), run.as_of(2) + + 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 +async def test_a_prompt_with_no_stored_conversation_is_one_turn_of_its_prompt_and_reply(): + 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) + + [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_raises_rather_than_reading_as_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.unavailable["conversation"] == "conversation 'conv-gone' is not in the store" + 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 +@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) + + 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 + + +@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.firing("reschedule") is None and run.unavailable == {} + + +@pytest.mark.asyncio +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) + + 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_has_the_env_firings_alone(): + context, _, _ = recorded_run() + + run = await load_run_context(context, None) + + assert run.turns == () and [(f.trigger_id, f.turn) for f in run.firings()] == [("confirm", 0), ("late", 0)] + + +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 + +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, _ = _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: 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 +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 "confirm fired during turn 1" in prompt and "late fired" 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 " fired during " not in prompt + assert "(not available: env:calendar: withheld" in prompt and "(not available: user_agent: 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"]), + (["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(): + max_chars = RubricsVerifierTaskStep.RUN_CONTEXT_MESSAGE_MAX_CHARS + long = "x" * (max_chars + 50) + 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, ["conversation", "trigger_log"]) + + assert "x" * max_chars + "… (+1 non-text part)" in text and long not in text + 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"]]) +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""" +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]}] +""" + + +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_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")) + + 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_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() + 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() + + +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 [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")