feat(verifiers): give verifiers the run's conversation and trigger firing logs

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 ``RunContext`` 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

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 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,
@@ -173,6 +174,59 @@ 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))
+
+
+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 = (
@@ -310,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"
@@ -347,10 +403,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 +527,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 +558,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 +594,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):
@@ -704,6 +771,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 +792,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 +1371,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,
     ) -> 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 +1403,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(