[sarah-chitty-1]

What

Give verifiers the run's conversation, including the user sim's turns, and the trigger firings, in a general way.

Today env_outcome_verifier gets only the MCP URL. A task with triggers or a user sim can't grade "noticed the change", "asked before acting" or "didn't leak the answer" without pulling those records itself.

• run_context.load_run_context builds a typed, read-only Run from what the run recorded. Run, Turn, Message, Firing and RunRecordUnavailable are exported from agent_env.task_step.task_steps.verifiers and are on the plugin surface.
  • Turns: each Turn has the user-side message that opened it (the prompt, then each user-sim or human message) and the agent's reply. A user message carries the agent triggers that wrote it and the user sim's extra fields.
  • Firings: one Firing per fire or failure. The gateway's detected → fired/failed sequence and the user agent's entries are folded in one adapter. Each firing carries its source (env:<id> or user_agent), the turn it fell in (0 = before the first message), its time, its virtual time, and the env call that provoked it.
  • Missing records raise: a query that needs a record that couldn't be read raises RunRecordUnavailable(source, reason), so a missing log never reads as "never fired". run.unavailable names each one.
• run.as_of(k) cuts it the way feat(rubrics_verifier): grade a multi-turn run as of turn k #222 grades "as of turn k": turns 1..k, and what fired before turn k+1 began.
• rubrics_verifier: a new run_context_for_judge: ["conversation", "trigger_log"] adds Conversation and Trigger firings sections to the judge prompt, cut at grade_up_to_turn. They render from the same Run, so the judge reads "confirm fired during turn 1". Size is capped: 4000 characters per message, the latest 200 firings.
• env_outcome_verifier: a script whose verify names a run parameter is called as verify(mcp_url, run=...). Every other script, one with an unrelated second parameter included, is called exactly as before. prompt_id picks which prompt, defaulting to the run's last one.

from agent_env.task_step.task_steps.verifiers import Run

async def verify(mcp_url: str, run: Run | None = None):
    moved = run.firing("reschedule", env="calendar")  # None: it never fired. Raises: its log is missing.
    reply = run.turn(moved.turn).reply if moved else None
    return [{"name": "told_user", "passed": bool(reply and "12:30" in reply.text)}]

Worth a look

1. Opt-in only: the run reaches a verify that names run, and the judge only when run_context_for_judge is set. Nothing changes for existing tasks or scripts.
2. Multi-prompt runs: the env firing log comes from the capture prompt_agent stores at the end of a prompt, so the latest capture is the env's whole log. An earlier prompt's env firings are cut where the next prompt's first message was sent; if that can't be read, or the next prompt shares this one's conversation (one context_id), they're withheld and marked unavailable. Prompts sharing one conversation each get only their own messages, from their first message to the next sharing prompt's; if either can't be found exactly once, the conversation is withheld.
3. Not covered: agent_prompt_response_verifier doesn't get the run. Should it get criteria like "trigger X fired"?
4. feat(rubrics_verifier): grade a multi-turn run as of turn k #222's rule: feat(rubrics_verifier): grade a multi-turn run as of turn k #222 refuses grade_up_to_turn with use_trajectory=False. With the conversation shown to the judge, that could be allowed; the rule is left unchanged here.
5. When the user agent's log counts as missing: prompt_agent only logs a warning when it can't read the user agent's firing log back, so the loader infers that a log was expected. It was if the step recorded a trigger firing or the prompt's user agent had registered triggers. If the conversation can't be read, any registered agent trigger counts.
6. Without the conversation: firings can't be placed in turns, so every non-empty firing log is withheld and marked unavailable. A grade never sees a firing it can't place.
7. Follow-up: once "who changed what" (feat(gateway): name the role and call behind each change to an env's state #240) lands, add an "env_changes" part.

Tests

• Unit: tst/unit/task_step/verifiers/test_run_context.py, 36 tests against the real local conversation and object stores: the loader and the turn placement, each message's env_clock, the gateway fold (queued detects, a time trigger's several fires, failures), the turn cut and the next-prompt cut (and a shared conversation, cut per prompt or withheld), missing records raising (the conversation, an env log, an unknown env, an unparseable log, a user-agent log that wasn't read back, an env or user-agent log that lost events to its cap), the judge prompt sections, size caps and option validation, and which scripts get run (a run parameter does; one, two-with-another-name don't).
• Integration: test_verifier_run_context_local.py runs a real two-turn conversation, with the echo agent as both solver and user sim, and reads it back as a Run.
• Results: unit suite 8835 passed, 25 skipped; ruff, plugin-api (no break against the merge base) and uv lock --check pass; the local integration test passes.

🤖 Generated with Claude Code

Confidence Score: 5/5

The PR appears safe to merge; both numbered findings are addressed and no new blocking issue was found.

Summary

Adds opt-in conversation and trigger evidence for verifiers through a typed Run.

• Verifiers can inspect a typed run of turns and trigger firings.
• Outcome scripts can ask for the run they grade.
• Rubric judges can see selected parts of a run.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  C[Stored conversation] --> L[load_run_context]
  E[Environment firing logs] --> L
  U[User-agent firing log] --> L
  L --> R[Run]
  R --> S[Opt-in verifier script]
  R --> A[as_of selected turn]
  A --> J[Requested judge sections]

Loading

_{Reviews (7) · Last reviewed commit: "review: a firing log that lost events to..." · Reviewed by Greptile}

[earakely-scale]

🤖 Drafted by Claude Code, reviewed by Edgar Arakelyan before posting.

Suggestion: cut an earlier prompt's env firing logs where the next prompt began.

env_trigger_state keeps one entry per env, and each prompt_agent overwrites it, so what this reads is the env's whole log as of the last prompt. Your point 2 notes this, but it goes further than "cut by turn, not by prompt": the cut only happens when grade_up_to_turn is set. Without it, a verifier grading prompt 1 of a two-prompt task sees prompt 2's firings as if they were prompt 1's.

Example: a task books lunch (prompt 1), then dinner (prompt 2). A verifier with prompt_id set to the lunch prompt and run_context_for_judge: ["trigger_log"] would show the judge dinner_confirm firing during the lunch prompt. The test in the diff below reproduces it: on this branch, grading the first prompt returns ["lunch_confirm", "dinner_confirm"].

Proposal. When a later prompt exists, cut the env logs at its first stored message. prompt_agent stores that message when it's sent (record_turn), so it marks when the next prompt began, just as as_of_turn uses the next turn's message. If that start can't be read, withhold the logs and mark them unavailable, the same rule as_of_turn already applies.

The last prompt, which is the default, is unchanged. The user agent's log is already kept per step, so it isn't touched.

Diff (applied to edc94b4 and run locally: test_run_context.py 27 passed, tst/unit/task_step/verifiers 373 passed)

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 f19cfab..ebaa427 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 cbd3374..95a227b 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"]

[sarah-chitty-1]

Applied in ef4565a, with your test (fails before, passes after). 14b9342 keeps it, on the typed Run.

[earakely-scale]

🤖 Drafted by Claude Code, reviewed by Edgar Arakelyan before posting.

Suggestion: hand verifiers a typed, read-only run record instead of raw dicts.

RunContext passes the records as the stores and logs keep them: message dicts, plus each firing log's raw events. Take a verify script that checks "did the agent tell the user about the reschedule the world made mid-run?" It has to know:

• two log formats: the gateway's detected → actions → fired/failed sequence, and the user agent's single fired entries;
• the unavailable key naming;
• how to place an ISO-string timestamp in a turn.

async def verify(mcp_url, run=None):
    key = "env_trigger_events.calendar"
    if key in run.unavailable:                       # forget this and a missing log reads as "never fired"
        raise RuntimeError(run.unavailable[key])
    fired = next((e for e in run.env_trigger_events["calendar"]
                  if e["kind"] == "fired" and e.get("trigger_id") == "reschedule"), None)
    starts = [m["ts"] for m in run.conversation if m["role"] == "user"]
    turn = sum(s <= fired["ts"] for s in starts) if fired else None   # string-compare stamps to find the turn
    reply = next((m for m in run.conversation if m["role"] == "agent" and m["turn"] == turn), None)
    return [{"name": "told_user", "passed": bool(fired and reply and "12:30" in reply["text"])}]

Every line of that ties a verifier script to an internal format, so the gateway's log format becomes a public contract by accident.

Proposal. Keep the loader as it is, and have it build a small typed model of what graders reason about (messages, turns and fires) rather than log events:

@dataclass(frozen=True)
class Message:
    turn: int
    role: Literal["user", "agent"]
    text: str
    sent_at: datetime
    author: str | None = None
    written_by_triggers: tuple[str, ...] = ()   # agent triggers that wrote this user message
    user_sim: Mapping[str, Any] | None = None    # the user sim's extra fields

@dataclass(frozen=True)
class Firing:
    trigger_id: str
    source: str                  # "env:<env_id>" or "user_agent"
    turn: int                    # the turn it fell in; 0 = before the first message
    at: datetime
    virtual_time: datetime | None
    ok: bool                     # fired, rather than failed
    provoked_by: Mapping[str, Any] | None

@dataclass(frozen=True)
class Turn:
    number: int
    message: Message             # the user-side message that opened it
    reply: Message | None
    firings: tuple[Firing, ...]  # what fired during it

class Run:
    turns: tuple[Turn, ...]
    unavailable: Mapping[str, str]
    def turn(self, number: int) -> Turn: ...
    def firings(self, trigger_id: str | None = None, *, env: str | None = None) -> tuple[Firing, ...]: ...
    def firing(self, trigger_id: str, *, env: str | None = None) -> Firing | None: ...  # the first one
    def as_of(self, turn: int) -> "Run": ...

The same check:

from agent_env.verifiers import Run

async def verify(mcp_url: str, run: Run | None = None):
    moved = run.firing("reschedule", env="calendar")   # None: it never fired. Raises: its log is missing.
    reply = run.turn(moved.turn).reply if moved else None
    return [{"name": "told_user", "passed": bool(reply and "12:30" in reply.text)}]

Three rules make it more than syntax:

• One record per fire, built in one place. The framework folds the gateway's event sequence and the user agent's entries into Firing. If a log format changes, one adapter changes and no script breaks.
• A missing log raises; it never reads as "false". A query that needs an unavailable log raises, for example RunRecordUnavailable(source, reason). The idea behind your unavailable map, that a missing log isn't an empty one, then holds even for a script author who forgets to check.
• Every firing carries its turn. The judge's sections can render from the same model, so the judge sees "fired during turn 2", not a timestamp it can't line up with the conversation.

Timing: nothing depends on the dict shape yet. Once verify scripts are stored as artifacts, they'll freeze whatever shape ships, so this is cheapest before merge. Registering these types on the plugin surface would let the Plugin API check guard the contract from then on.

Your loader keeps doing all the reading and joining, and this is a layer on top of it. Happy to prototype it against your tests if that helps.

[sarah-chitty-1]

Done in 14b9342: Run/Turn/Message/Firing in agent_env.task_step.task_steps.verifiers (where the verifier types are exported), also on the plugin surface. A missing log raises RunRecordUnavailable, and the judge sections render from the model ("confirm fired during turn 1").

[earakely-scale]

🤖 Automated Claude routine — Env Pod PR Context Review. This is a bot, not Edgar Arakelyan; not a human review.

Context from a sibling PR rather than this diff: #265 (open, same author) adds a new key to the very conversation records load_run_context reads here.

There, conversation_store._message() (src/agent_env/a2a_agent/conversation_store.py:112-116, head 8c1a11a) writes env_clock: {env_id: virtual_time} onto each stored message, and gives as its reason that "a grader can then line up the agent's messages with the trigger firing log and the gateway trajectory, which already carry virtual_time".

This PR's typed Run is the API such a grader would reach for, and it already carries one half of that alignment: Firing.virtual_time (run_context.py:48). The message half isn't wired — Message (run_context.py:31) has no clock field, and _message() (run_context.py:270-274) reads role/parts/ts only, so env_clock is dropped even once #265 lands. A verifier holding a Run could place a firing in virtual time but would have to go back to the raw conversation document for the message it's lining up against.

Adding the field stays backwards-compatible whichever lands first, so this is only a question of whether to decide it now — Message goes on the checked plugin surface in this PR (check_plugin_api.py). Raising it because the two halves sit in separate PRs with no overlapping files, so a per-diff review sees only one of them.

---

Generated by Claude Code

[sarah-chitty-1]

Addressed in b870df5: Message.env_clock ({env_id: datetime | None}) reads the env_clock #265 stores on each message, and is None for a message stored without one, so it's backward compatible whichever lands first.