From 1ccdf357168dec7d7a5e96ecd9f2fd64b8579fac Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Fri, 13 Nov 2026 17:07:44 -0800 Subject: [PATCH 1/3] feat(always_on_agent_verifier): with use_trajectory, the agent judge searches every wake staged as files always_on_staging builds the run's files from its stored timeline and each wake's trajectory, matched by trajectory_url: wakes/wake_NNN.jsonl (what woke the assistant with the user's whole message, every tool call with input and result, its text, every app change, messages, nudges, reads), events.jsonl, triggers.jsonl and index.md (actions and messages with file:line). Any line too long to read whole keeps a preview and moves its long fields to wakes/results/. index.md's write actions are calls of a tool that made one of the wake's app changes (the changes' actors), or, when the apps keep no change log, a guess from the tool name; the index says which. Missing or unreadable trajectories and turns compaction didn't shrink are recorded and logged. The verifier accepts use_trajectory with the agent judge only, writes the files concurrently, adds a prompt note on reading them, records the staging and what each judge call read (judge_reads), and gives the judge only Bash and Read of its own tools (builtin_tools only), with no peer-agents tools (peer_tools false): no helper agents, web, writes or messaging other agents. Its defaults: 2400 s and 400 turns per judge call, one timeout retry, at most 3 batches at once. Without a stored timeline it falls back to rubrics_verifier's per-turn loading. Co-Authored-By: Claude Opus 5.5 --- .../verifiers/always_on_agent_verifier.py | 228 ++++- .../task_steps/verifiers/always_on_staging.py | 438 ++++++++++ .../test_always_on_judge_config.py | 34 + .../test_always_on_agent_verifier.py | 161 +++- .../recorded/always_on_timeline.json | 816 ++++++++++++++++++ .../verifiers/test_always_on_staging.py | 273 ++++++ 6 files changed, 1922 insertions(+), 28 deletions(-) create mode 100644 src/agent_env/task_step/task_steps/verifiers/always_on_staging.py create mode 100644 tst/unit/agents/claude_code/test_always_on_judge_config.py create mode 100644 tst/unit/task_step/verifiers/recorded/always_on_timeline.json create mode 100644 tst/unit/task_step/verifiers/test_always_on_staging.py diff --git a/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py b/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py index 372163820..42f79d7d3 100644 --- a/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py @@ -1,35 +1,104 @@ """always_on_agent_verifier: the rubrics_verifier for an always_on step's prompt. -An always_on run may take 40–60 wakes, each with its own trajectory: together they may not fit a judge, and cost a lot. -This verifier grades the run from what the user saw and the run's context instead: the user's chat (the prompt's -response and conversation), the trigger log, and the run wake by wake with the app changes each made -(``render_for_judge``), never a wake's trajectory. So when the task loads it refuses what needs the trajectories: an -output format that cites per-criterion screenshots, use_trajectory, and grade_up_to_turn (the conversation's turns are -not the run's wakes); a per-run grade_up_to_turn is refused when the verifier runs. Every other rubrics_verifier -setting works as it does there. An always_on step's prompt takes this verifier, never a plain rubrics_verifier, and -this verifier only an always_on step's prompt: a task that pairs them otherwise is refused when it loads. +The judge's prompt holds what the user saw and the run's context: the user's chat (the prompt's response and +conversation), the trigger log, and the run wake by wake with the app changes each made (``render_for_judge``). With +use_trajectory, an agent judge also gets every tool call of every wake, staged in its sandbox as files it searches +(``always_on_staging``): one file per wake with one event per line, matched to its wake by the run's timeline, an +index, every event in one file and the trigger firings. A direct judge never gets the trajectories, so the task is +refused when it loads with use_trajectory and a direct judge, an output format that cites per-criterion screenshots, +or grade_up_to_turn (the conversation's turns are not the run's wakes); a run whose overrides ask for either is +refused when the verifier runs. Unlike rubrics_verifier, its judge has 2,400 s and 400 turns per call by default, a +call that runs out of time is tried once more, at most 3 batches run at once, and the judge has only Bash and Read: no +helper agents, no web, no writes, no messaging other agents. An always_on step's prompt takes this verifier, never a +plain rubrics_verifier, and this verifier only an always_on step's prompt: a task that pairs them otherwise is refused when it loads. """ from __future__ import annotations import asyncio +import logging +import uuid from typing import Any, ClassVar, Optional +from agent_env.providers.sandbox_providers.sandbox import into_container from agent_env.task_step.context import PromptResponse, TaskStepContext from agent_env.task_step.task_step import TaskStep -from agent_env.task_step.task_steps.always_on import ALWAYS_ON_KEY, AlwaysOnTaskStep, render_for_judge +from agent_env.task_step.task_steps.always_on import ALWAYS_ON_KEY, AlwaysOnTaskStep, render_for_judge, stored_timeline from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep +from agent_env.task_step.task_steps.verifiers.always_on_staging import ( + EVENTS_FILE, + INDEX_FILE, + RESULTS_DIR, + TRIGGERS_FILE, + WAKES_DIR, + StagedRun, + judge_reads, + stage_run, +) from agent_env.task_step.task_steps.verifiers.judge_utils.judge_output_format import ( JudgeOutputFormat, get_judge_output_format_spec, ) -from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RUN_CONTEXT_PARTS, RubricsVerifierTaskStep +from agent_env.task_step.task_steps.verifiers.judge_utils.turn_events import TurnEvents, turn_events +from agent_env.task_step.task_steps.verifiers.rubrics_verifier import ( + RUN_CONTEXT_PARTS, + RubricsVerifierTaskStep, + StagedTrajectory, +) +from agent_env.task_step.task_steps.verifiers.run_context import Run, load_run_context + +logger = logging.getLogger(__name__) + +# Where the run's files go in the judge's sandbox, under a directory of the run's own. +STAGING_DIR = "/tmp/always_on_run" +# Trajectories read, and files written into the judge's sandbox, at once. +STAGING_CONCURRENCY = 8 +# The only tools of its own the judge gets: it reads and searches the files (the CLI's search is Bash's grep), and its +# answer comes back through the structured output the CLI adds for the output format. No helper agents, which ran a +# second judge with its own verdict; no web; no writes; and no messaging other agents (peer_tools). +JUDGE_TOOLS = ("Bash", "Read") +_FILES_NOTE = """ + +## The run's files +The trajectory path above is a directory holding the whole run, one JSON event per line: +- {index}: start here. A block per wake: its pretend time, what woke the assistant, the user's message, each write \ +action (tool and key arguments; how actions were told apart is at its top) and message to the user with the \ +`file:line` it is at, and errors. +- {first} … {last}: one file per wake. First what woke the assistant ("woke", the user's own messages in full as \ +"user_message", the whole text it was sent as "wake_input"); then every tool call of its turn with input and result \ +("tool_call", numbered "call") and its own text ("assistant_text", which nobody saw); then the rest of the wake in time \ +order: every app change ("app_change", "by": "agent" for its own), its closing reply ("turn_end"; the user never saw \ +it), each message the user got ("message_to_user"), a nudge and its calls, and the user's reads of its messages. +- {results}/: what is too long for a line (a tool's input or result, a long message): the line names the file \ +("_file", such as "result_file" or "text_file") and shows its start ("_preview"). Email bodies are \ +decoded (body.text). +- {events}: every wake's events in one file, so "did it ever do X" is one search. {triggers}: every trigger firing, \ +fired or failed, with its pretend time and the wake it came after. + +How to read them: +- Don't load whole files. Search with grep -n, then Read with an offset and limit. +- Later actions replace earlier ones: after you find something, look for later corrections, edits, cancellations or \ +follow-ups before you decide. +- The user's messages are the "user_message" lines (and the first wake's "wake_input"); the user saw only the \ +"message_to_user" lines. +- Cite the wake (and call) number of every piece of evidence, and quote it. +- Only the assistant's tool calls and its "agent" app changes are its acts. A draft is not a sent message. Messages \ +from the user, the user simulator or contacts are never the assistant's. Order events by their pretend time ("at"), \ +not by where they sit in the files. +- You have Bash and Read only: no web, no other agents, and nothing to change.{gaps}""" class AlwaysOnAgentVerifier(RubricsVerifierTaskStep): type: ClassVar[str] = "always_on_agent_verifier" entity_refs = RubricsVerifierTaskStep.entity_refs # the judge, as rubrics_verifier's context_paths = RubricsVerifierTaskStep.context_paths + DEFAULT_JUDGE_TIMEOUT_SECONDS: ClassVar[int] = 2400 + DEFAULT_JUDGE_BATCH_CONCURRENCY: ClassVar[int] = 3 + DEFAULT_JUDGE_TIMEOUT_RETRIES: ClassVar[int] = 1 + # A judge reading a 100+ wake run spends a few calls a criterion (index, search, read) on top of finding its way: + # 30 criteria unbatched is 100-200 turns. The cap bounds a judge that loops; the time limit (2,400 s at about 5-10 + # s a turn) bounds it near here anyway. + DEFAULT_JUDGE_MAX_TURNS: ClassVar[int] = 400 def __init__( self, @@ -37,6 +106,7 @@ def __init__( version: Optional[int], *, use_trajectory: bool = False, + use_agent_judge: bool = True, grade_up_to_turn: Optional[int] = None, run_context_for_judge: Optional[list[str]] = None, output_format: JudgeOutputFormat | str = JudgeOutputFormat.RUBRIC_BINARY, @@ -48,26 +118,33 @@ def __init__( judge_sandbox_type: Optional[str] = None, judge_batch_size: Optional[int] = None, judge_batch_concurrency: Optional[int] = None, + judge_timeout_retries: Optional[int] = None, + judge_max_turns: Optional[int] = None, **fields: Any, ): """``run_context_for_judge`` defaults to every part of the run's context; the rest is rubrics_verifier's.""" output_format = JudgeOutputFormat(output_format) where = f"{self.type} '{id}'" - if use_trajectory: - raise ValueError(f"{where}: use_trajectory must be off; an always_on step's prompt is graded from the " - "user's chat and the run's context, never its wakes' trajectories") + if use_trajectory and not use_agent_judge: + raise ValueError(f"{where}: use_trajectory needs the agent judge (use_agent_judge): an always_on run's " + "wakes are staged as files for a judge that searches them, never inlined") if get_judge_output_format_spec(output_format).cites_evidence: raise ValueError(f"{where}: output_format={output_format.value} cites per-criterion screenshots, and an " - "always_on step's prompt is graded without its wakes' trajectories") + "always_on step's prompt is graded without its wakes' screenshots") if grade_up_to_turn is not None: raise ValueError(f"{where}: grade_up_to_turn can't cut an always_on step's prompt, whose conversation " "turns are not its wakes") super().__init__( - id, version, use_trajectory=False, run_context_for_judge=run_context_for_judge or list(RUN_CONTEXT_PARTS), - output_format=output_format, default_model=default_model, default_model_api_base=default_model_api_base, - effort=effort, max_thinking_tokens=max_thinking_tokens, judge_a2a_agent_id=judge_a2a_agent_id, + id, version, use_trajectory=use_trajectory, use_agent_judge=use_agent_judge, + run_context_for_judge=run_context_for_judge or list(RUN_CONTEXT_PARTS), output_format=output_format, + default_model=default_model, default_model_api_base=default_model_api_base, effort=effort, + max_thinking_tokens=max_thinking_tokens, judge_a2a_agent_id=judge_a2a_agent_id, judge_sandbox_type=judge_sandbox_type, judge_batch_size=judge_batch_size, - judge_batch_concurrency=judge_batch_concurrency, **fields, + judge_batch_concurrency=judge_batch_concurrency, + judge_timeout_retries=(self.DEFAULT_JUDGE_TIMEOUT_RETRIES if judge_timeout_retries is None + else judge_timeout_retries), + judge_max_turns=self.DEFAULT_JUDGE_MAX_TURNS if judge_max_turns is None else judge_max_turns, + **fields, ) @classmethod @@ -87,13 +164,122 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: if "grade_up_to_turn" in self.step_param_overrides(context): raise RuntimeError(f"Verifier '{self.verifier_id}': grade_up_to_turn can't cut an always_on step's " "prompt, whose conversation turns are not its wakes.") + if self.use_trajectory and (context.metadata.get("user_overrides") or {}).get("use_agent_judge") is False: + raise RuntimeError(f"Verifier '{self.verifier_id}': use_trajectory needs the agent judge; this run's " + "use_agent_judge=False would grade without the wakes it was set to read.") return await super().execute(context) - async def judge_prompt_sections(self, context: TaskStepContext, prompt_response: PromptResponse) -> list[str]: - """The run wake by wake, from the always_on step's record of it.""" + async def _grade(self, context: TaskStepContext) -> TaskStepContext: + """rubrics_verifier's grade; then, for a staged run, what each judge call read of it (``judge_reads``).""" + context = await super()._grade(context) + entry = (context.metadata.get("verifications") or {}).get(self.verifier_id) or {} + if "trajectory_staging" in entry: + entry["trajectory_staging"]["judge_reads"] = await self._judge_reads(entry) + return context + + def judge_agent_config(self) -> dict: + return {"builtin_tools": {"only": list(JUDGE_TOOLS)}, "peer_tools": False} + + def _record(self, context: TaskStepContext, prompt_response: PromptResponse) -> dict: record = (context.metadata.get(ALWAYS_ON_KEY) or {}).get(prompt_response.step_id) if record is None: raise RuntimeError(f"Verifier '{self.verifier_id}': prompt_id='{self.prompt_id}' is not an always_on " f"step's prompt (step '{prompt_response.step_id}' left no always_on record); grade it " "with rubrics_verifier.") - return [await asyncio.to_thread(render_for_judge, record)] + return record + + async def judge_prompt_sections(self, context: TaskStepContext, prompt_response: PromptResponse) -> list[str]: + """The run wake by wake, from the always_on step's record of it.""" + return [await asyncio.to_thread(render_for_judge, self._record(context, prompt_response))] + + async def stage_trajectory(self, context: TaskStepContext, prompt_response: PromptResponse, + run_context: Optional[Run], sandbox: Any, + container: Optional[str]) -> Optional[StagedTrajectory]: + """The run as files (``stage_run``) in the judge's sandbox. Without its stored timeline, which matches each + trajectory to its wake, None: the per-turn trajectories are loaded as rubrics_verifier loads them.""" + record = self._record(context, prompt_response) + if not record.get("object_url"): + logger.warning("Verifier '%s': the always_on run kept no stored timeline; its wakes' trajectories are " + "loaded by position instead", self.verifier_id) + return None + try: + timeline = await asyncio.to_thread(stored_timeline, record) + except Exception as e: + logger.warning("Verifier '%s': the always_on run's timeline could not be read (%s: %s); its wakes' " + "trajectories are loaded by position instead", self.verifier_id, type(e).__name__, e) + return None + urls = list(dict.fromkeys(e["trajectory_url"] for e in timeline + if e.get("event") in ("agent_turn_end", "nudge") and e.get("trajectory_url"))) + turns = await self._read_turns(urls) + run = run_context or await load_run_context(context, prompt_response) + staged = await asyncio.to_thread(stage_run, timeline, turns, firings=run.firings_read(), + unavailable=run.unavailable) + path = f"{STAGING_DIR}/{uuid.uuid4().hex[:12]}" + await self._write_files(sandbox, {f"{path}/{name}": text for name, text in staged.files.items()}, container) + self._log_staging(staged) + return StagedTrajectory(path=path, prompt_note=_files_note(path, staged), record=staged.record) + + async def _read_turns(self, urls: list[str]) -> dict[str, Optional[TurnEvents]]: + gate = asyncio.Semaphore(STAGING_CONCURRENCY) + + async def read(url: str) -> Optional[TurnEvents]: + async with gate: + try: + return await asyncio.to_thread(lambda: turn_events(self._read_trajectory_text(url))) + except Exception as e: + logger.warning("Verifier '%s': trajectory %s could not be read (%s: %s)", self.verifier_id, url, + type(e).__name__, e) + return None + + return dict(zip(urls, await asyncio.gather(*(read(u) for u in urls)))) + + async def _write_files(self, sandbox: Any, files: dict[str, str], container: Optional[str]) -> None: + gate = asyncio.Semaphore(STAGING_CONCURRENCY) + + async def write(path: str, text: str) -> None: + async with gate: + await sandbox.write_file_from_text(text, path, **into_container(container)) + + await asyncio.gather(*(write(path, text) for path, text in files.items())) + + def _log_staging(self, staged: StagedRun) -> None: + r = staged.record + logger.info("Verifier '%s': staged %d wakes as %d files; %d trajectories (%s) of %d bytes staged as %d", + self.verifier_id, r["wakes"], r["files"], r["trajectories"], r["formats"], r["raw_bytes"], + r["staged_bytes"]) + if r.get("not_compacted"): + logger.error("Verifier '%s': compaction removed nothing from %d trajectories (unknown format, or nothing " + "to drop): %s", self.verifier_id, len(r["not_compacted"]), r["not_compacted"]) + for key in ("missing_trajectories", "unreadable_trajectories"): + if r.get(key): + logger.warning("Verifier '%s': %s in wakes %s", self.verifier_id, key.replace("_", " "), r[key]) + + async def _judge_reads(self, entry: dict) -> list[dict]: + """``judge_reads`` of each judge call's trajectory (each batch's, and each timed-out attempt's).""" + calls = [{"batch": i, "url": url} for i, url in enumerate( + entry.get("judge_batch_trajectory_object_urls") or [entry.get("judge_trajectory_object_url")], 1)] + calls += [{"batch": d.get("batch", 1), "timed_out": True, "url": d["judge_trajectory_object_url"]} + for d in entry.get("judge_output_discrepancies") or [] + if d.get("kind") == "judge_timeout" and d.get("judge_trajectory_object_url")] + out = [] + for call in calls: + if not call["url"]: + out.append({**call, "unreadable": "no trajectory was kept"}) + continue + try: + events = await asyncio.to_thread(lambda u=call["url"]: turn_events(self._read_trajectory_text(u))) + out.append({**call, **judge_reads(events)}) + except Exception as e: + out.append({**call, "unreadable": f"{type(e).__name__}: {str(e)[:200]}"}) + return out + + +def _files_note(path: str, staged: StagedRun) -> str: + r = staged.record + wakes = sorted(name for name in staged.files if name.startswith(f"{WAKES_DIR}/wake_")) + gaps = sorted(set(r.get("missing_trajectories", [])) | set(r.get("unreadable_trajectories", []))) + gap_note = ("\n- These wakes' tool calls are not in the files (their trajectory is missing or unreadable), so " + "for them only the run's record is there: " + ", ".join(str(n) for n in gaps) + ".") if gaps else "" + return _FILES_NOTE.format(index=f"{path}/{INDEX_FILE}", first=f"{path}/{wakes[0] if wakes else WAKES_DIR}", + last=f"{path}/{wakes[-1] if wakes else WAKES_DIR}", results=f"{path}/{RESULTS_DIR}", + events=f"{path}/{EVENTS_FILE}", triggers=f"{path}/{TRIGGERS_FILE}", gaps=gap_note) diff --git a/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py b/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py new file mode 100644 index 000000000..033dd7f29 --- /dev/null +++ b/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py @@ -0,0 +1,438 @@ +"""An always_on run as files an agent judge reads with search and offset reads (``stage_run``), built from the run's +stored timeline and each wake's trajectory: one file per wake, one event per line, so no line is a whole turn. + +- ``wakes/wake_NNN.jsonl``: what woke the assistant (``woke``, the user's messages in full ``user_message``, the full + text it was sent ``wake_input``), then its turn's tool calls with their inputs and results (``tool_call``) and its + own text (``assistant_text``), then the rest of the wake in time order (``app_change`` for every change, + ``turn_end`` with its closing reply, ``message_to_user``, a ``nudge`` and its calls, the user's reads). A turn's + trajectory is matched to its wake by the timeline's ``trajectory_url``; one that's missing is a line saying so. +- ``wakes/results/``: what is too long for a line to be read whole (a tool's input or result, a long message), named + by wake and call (or line); the line keeps a preview. +- ``events.jsonl``: every line of every wake, and the run's start and end. ``triggers.jsonl``: every trigger firing. +- ``index.md``: a block per wake with its time, what woke it, the user's message, its write actions (tool and key + arguments) and messages with the ``file:line`` they're at, and its errors. +""" + +from __future__ import annotations + +import json +import re +from collections import Counter +from dataclasses import dataclass, field +from typing import Any, Mapping, Optional, Sequence + +from agent_env.task_step.task_steps.always_on import BULKY_EVENT_FIELDS, woke_by_text +from agent_env.task_step.task_steps.notification_delivery import label_time +from agent_env.task_step.task_steps.verifiers.judge_utils.turn_events import UNKNOWN, TurnEvents +from agent_env.task_step.task_steps.verifiers.run_context import Firing + +WAKES_DIR = "wakes" +RESULTS_DIR = "wakes/results" +INDEX_FILE = "index.md" +EVENTS_FILE = "events.jsonl" +TRIGGERS_FILE = "triggers.jsonl" +# A tool call's line stays under this, so a reader shows it whole: past it, its input or result (the longer first) goes +# to its own file under RESULTS_DIR and the line keeps a preview. +LINE_CHARS = 1800 +PREVIEW_CHARS = 300 +# What index.md shows of a message (the wake file has it whole), of one argument, and of one action line. +INDEX_TEXT_CHARS = 400 +ARG_CHARS = 40 +ARGS_SHOWN = 6 +ACTION_CHARS = 240 +_DROPPED = frozenset({*BULKY_EVENT_FIELDS, "real_time", "real_elapsed_s", "user_sim_trajectory_urls"}) +# A tool whose name has one of these words reads; index.md lists every other call as a write action. +_READS = re.compile(r"(?:^|_)(get|list|search|read|find|whoami|snapshot|view|query|lookup|fetch|navigate|wait|hover|" + r"resize|tabs|screenshot|console|network|time)(?:_|$)") +_NOT_ACTIONS = frozenset({"message_user", "stay_quiet"}) +# A line's fields that say what and where it is; ``fit`` moves only the others. +_IDENTITY = frozenset({"event", "wake", "at", "call", "turn", "tool", "kind", "state", "outcome", "phase", "by"}) + + +@dataclass(frozen=True) +class StagedRun: + """``files`` by path relative to the staging directory; ``record`` is what the verification entry keeps.""" + + files: dict[str, str] + record: dict + + +@dataclass +class _Wake: + n: int + at: Optional[str] = None + woke_by: list[str] = field(default_factory=list) + head: list[dict] = field(default_factory=list) + calls: list[dict] = field(default_factory=list) + body: list[dict] = field(default_factory=list) + call_count: int = 0 + + def lines(self) -> list[dict]: + return self.head + self.calls + self.body + + +def stage_run(timeline: Sequence[dict], turns: Mapping[str, Optional[TurnEvents]], *, + firings: Sequence[Firing] = (), unavailable: Optional[Mapping[str, str]] = None) -> StagedRun: + """The run's files: ``turns`` holds each trajectory the timeline names, by its url (None for one that couldn't be + read); ``firings`` and ``unavailable`` are the run context's trigger firings and the logs it couldn't read.""" + stager = _Stager(turns, max((e.get("n") or 0 for e in timeline if e.get("event") == "wake"), default=0)) + for event in timeline: + stager.add(event) + return stager.finish(firings, unavailable or {}) + + +def wake_file(n: int, width: int = 3) -> str: + return f"{WAKES_DIR}/wake_{n:0{width}d}.jsonl" + + +def _line(event: str, wake: Optional[int], **fields: Any) -> dict: + return {"event": event, **({"wake": wake} if wake is not None else {}), + **{k: v for k, v in fields.items() if v is not None and v is not False}} + + +def _dumps(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, default=str) + + +def _cut(text: Any, chars: int) -> str: + text = " ".join(str(text or "").split()) + return text if len(text) <= chars else text[:chars] + "…" + + +def _short_tool(tool: str) -> str: + """A tool's own name, without an MCP server prefix (``mcp__server__tool``).""" + return tool.rsplit("__", 1)[-1] + + +def is_write_action(tool: str) -> bool: + name = _short_tool(tool) + return name not in _NOT_ACTIONS and not _READS.search(name.lower()) + + +def _key_args(tool_input: Any) -> str: + if not isinstance(tool_input, dict): + return _cut(_dumps(tool_input), ARG_CHARS) if tool_input not in (None, "") else "" + shown = [f"{k}={_cut(v if isinstance(v, str) else _dumps(v), ARG_CHARS)}" + for k, v in list(tool_input.items())[:ARGS_SHOWN]] + more = len(tool_input) - ARGS_SHOWN + return " ".join(shown) + (f" (+{more} more)" if more > 0 else "") + + +class _Stager: + def __init__(self, turns: Mapping[str, Optional[TurnEvents]], last_wake: int): + self.turns = turns + self.width = max(3, len(str(last_wake))) + self.wakes: dict[int, _Wake] = {} + self.run_lines: list[dict] = [] + self.current = 0 + self.pending_user: list[dict] = [] + self.files: dict[str, str] = {} + self.formats: Counter = Counter() + self.raw_bytes = self.staged_bytes = 0 + self.missing: list[int] = [] + self.unreadable: list[int] = [] + self.not_compacted: list[dict] = [] + self.unrecorded_changes = 0 + self.by_changes = False # whether the run recorded app changes, so write actions are read from them + + def wake(self, n: Optional[int]) -> _Wake: + n = n or self.current + return self.wakes.setdefault(n, _Wake(n)) + + def add(self, event: dict) -> None: + kind = event.get("event") + at = event.get("virtual_time") + if kind == "wake": + self.current = event["n"] + w = self.wake(event["n"]) + w.at, w.woke_by = at, list(event.get("woke_by") or []) + w.head.append(_line("woke", w.n, at=at, woke_by=w.woke_by, notifications=event.get("notifications") or None, + last=event.get("last"))) + w.head += [_line(m.pop("event"), w.n, **m) for m in self.pending_user] + self.pending_user = [] + w.head.append(_line("wake_input", w.n, text=event.get("parts_text") or "")) + elif kind in ("user_reply_delivered", "scripted_user_line"): + self.pending_user.append(_line("user_message", None, at=at, text=event.get("text"), + kind="user_reply" if kind == "user_reply_delivered" else event.get("kind"), + done=event.get("done"))) + elif kind == "app_changes": + w = self.wake(event.get("wake")) + self.by_changes = True + for change in event.get("changes") or []: + # The changelog's own timestamp is wall-clock time, not the run's pretend time. + w.body.append(_line("app_change", w.n, **{k: v for k, v in change.items() if k not in ("wake", "at")}, + recorded_at=change.get("at"))) + if event.get("not_kept"): + self.unrecorded_changes += event["not_kept"] + w.body.append(_line("app_changes_not_recorded", w.n, count=event["not_kept"], phase=event.get("phase"))) + elif kind == "agent_turn_end": + w = self.wake(event.get("n")) + w.calls += self.turn(w, event.get("trajectory_url"), "turn") + w.body.append(_line("turn_end", w.n, at=at, state=event.get("state"), closing_reply=event.get("plain_reply"), + tool_calls=event.get("tool_calls"), messages_to_user=event.get("messages_to_user"), + stay_quiet_calls=event.get("stay_quiet_calls") or None, + virtual_seconds=event.get("virtual_seconds"))) + elif kind == "nudge": + w = self.wake(event.get("wake")) + w.body.append(_line("nudge", w.n, at=at, closing_text=event.get("closing_text"), + outcome=event.get("outcome"), state=event.get("state"), + reply_after=event.get("plain_reply_after"))) + w.body += self.turn(w, event.get("trajectory_url"), "nudge") + elif kind == "message_to_user": + self.wake(event.get("wake")).body.append(_line( + "message_to_user", event.get("wake") or self.current, at=at, text=event.get("text"), + after_nudge=event.get("after_nudge"), leftover=event.get("leftover"))) + else: + fields = {k: v for k, v in event.items() if k not in _DROPPED and k not in ("event", "virtual_time", "wake")} + if kind in ("start", "end") or not (event.get("wake") or self.current): + self.run_lines.append(_line(kind, None, at=at, **fields)) + else: + w = self.wake(event.get("wake")) + w.body.append(_line(kind, w.n, at=at, **fields)) + + def turn(self, w: _Wake, url: Optional[str], which: str) -> list[dict]: + """A turn's events as lines of wake ``w``; its long inputs and results go to their own files.""" + if not url: + self.missing.append(w.n) + return [_line("trajectory_missing", w.n, turn=which)] + events = self.turns.get(url) + if events is None: + self.unreadable.append(w.n) + return [_line("trajectory_unreadable", w.n, turn=which)] + self.formats[events.format] += 1 + lines, side = [], 0 + for e in events.events: + kind = e["type"] + if kind == "tool_call": + w.call_count += 1 + line, written = self.call(w, e, which) + lines.append(line) + side += written + elif kind == "text": + lines.append(_line("assistant_text", w.n, turn=which, text=e["text"], subagent=e.get("subagent"))) + elif kind == "note": + lines.append(_line("harness_note", w.n, turn=which, text=e["text"])) + elif kind == "raw": + lines.append(_line("trajectory_item", w.n, turn=which, item=e["event"])) + staged = sum(len(_dumps(line).encode()) for line in lines) + side + self.raw_bytes += events.raw_bytes + self.staged_bytes += staged + if events.format == UNKNOWN or staged >= events.raw_bytes: + self.not_compacted.append({"wake": w.n, "turn": which, "format": events.format, + "raw_bytes": events.raw_bytes, "staged_bytes": staged}) + return lines + + def call(self, w: _Wake, e: dict, which: str) -> tuple[dict, int]: + """The call's line; while it is longer than LINE_CHARS its longer of input and result goes to a file.""" + name = re.sub(r"[^A-Za-z0-9_.-]", "_", _short_tool(e["tool"]))[:60] + stem = f"{RESULTS_DIR}/wake_{w.n:0{self.width}d}_call_{w.call_count:03d}_{name}" + line = _line("tool_call", w.n, call=w.call_count, turn=which, tool=e["tool"], subagent=e.get("subagent"), + input=e.get("input"), error=e.get("error")) + if "result" in e: + line["result"] = e["result"] + else: + line["no_result"] = True + return self.fit(line, stem, ("input", "result")) + + def fit(self, line: dict, stem: str, movable: Sequence[str]) -> tuple[dict, int]: + """``line`` shortened to LINE_CHARS, so a reader shows it whole: its longest ``movable`` field at a time goes to + a file named from ``stem`` (``.json``, ``.txt``, or with ``_`` for a field other than a + tool call's result) and leaves a preview. Returns the bytes it wrote.""" + written = 0 + while len(_dumps(line)) > LINE_CHARS and (inline := [k for k in movable if line.get(k) not in (None, "")]): + key = max(inline, key=lambda k: len(_dumps(line[k]))) + value = line.pop(key) + path = f"{stem}{'' if key == 'result' else '_' + key}.{'txt' if isinstance(value, str) else 'json'}" + body = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=1, default=str) + self.files[path] = body + written += len(body.encode()) + line[f"{key}_file"] = path + line[f"{key}_preview"] = _cut(value if isinstance(value, str) else _dumps(value), PREVIEW_CHARS) + return line, written + + def fit_all(self, lines: list[dict], prefix: str) -> list[dict]: + """Every line but a tool call's (``call`` fits those) shortened by ``fit``, its files named by line number.""" + for i, line in enumerate(lines, 1): + if line["event"] != "tool_call": + self.fit(line, f"{RESULTS_DIR}/{prefix}_line_{i:03d}_{line['event']}", + [k for k in line if k not in _IDENTITY]) + return lines + + def finish(self, firings: Sequence[Firing], unavailable: Mapping[str, str]) -> StagedRun: + if self.pending_user: + self.run_lines += self.pending_user + self.fit_all(self.run_lines, "run") + start = [line for line in self.run_lines if line["event"] != "end"] + end = [line for line in self.run_lines if line["event"] == "end"] + everything = list(start) + index_blocks = [] + for n in sorted(self.wakes): + w = self.wakes[n] + lines = self.fit_all(w.lines(), f"wake_{n:0{self.width}d}") + path = wake_file(n, self.width) + self.files[path] = "".join(_dumps(line) + "\n" for line in lines) + everything += lines + index_blocks.append(_index_block(w, path, lines, self.by_changes)) + everything += end + self.files[EVENTS_FILE] = "".join(_dumps(line) + "\n" for line in everything) + self.files[TRIGGERS_FILE] = _triggers(firings, unavailable, self.wakes) + self.files[INDEX_FILE] = _index(self, index_blocks, start, end, firings) + record: dict = {"wakes": len(self.wakes), "files": len(self.files), + "side_files": sum(1 for p in self.files if p.startswith(RESULTS_DIR + "/")), + "trajectories": sum(self.formats.values()), "formats": dict(self.formats), + "raw_bytes": self.raw_bytes, "staged_bytes": self.staged_bytes} + for key, value in (("missing_trajectories", self.missing), ("unreadable_trajectories", self.unreadable), + ("not_compacted", self.not_compacted), + ("app_changes_not_recorded", self.unrecorded_changes)): + if value: + record[key] = value + return StagedRun(self.files, record) + + +def _text(line: dict, key: str) -> Any: + """A field of a line, or its preview when ``fit`` moved it to a file.""" + return line.get(key) if line.get(key) is not None else line.get(f"{key}_preview") + + +def _changing_tools(lines: list[dict]) -> set[str]: + """The tools whose calls made the assistant's app changes in a wake, as the changes' actors name them.""" + return {_short_tool(a["tool"]) for line in lines if line["event"] == "app_change" and line.get("by") == "agent" + for a in line.get("actors") or [] if a.get("tool")} + + +def _index_block(w: _Wake, path: str, lines: list[dict], by_changes: bool) -> str: + """The wake's block. With ``by_changes`` its write actions are the calls of a tool that made one of the wake's app + changes; else (a run whose apps keep no change log) calls whose tool name has no read word (``is_write_action``).""" + calls = sum(1 for line in lines if line["event"] == "tool_call") + changing = _changing_tools(lines) + out = [f"## Wake {w.n} · {label_time(w.at)} · woken by {woke_by_text(w.woke_by)} · {path} " + f"({len(lines)} lines, {calls} tool calls)"] + users = [(i, line) for i, line in enumerate(lines, 1) if line["event"] == "user_message"] + for i, line in users: + out.append(f"- the user ({label_time(line.get('at'))}): \"{_cut(_text(line, 'text'), INDEX_TEXT_CHARS)}\" " + f"({path}:{i})") + wake_input = next(((i, line) for i, line in enumerate(lines, 1) if line["event"] == "wake_input"), None) + if not users and wake_input: + i, line = wake_input + out.append(f"- input: \"{_cut(_text(line, 'text'), INDEX_TEXT_CHARS)}\" ({path}:{i})") + changes = Counter() + for i, line in enumerate(lines, 1): + event = line["event"] + writes = _short_tool(line.get("tool") or "") in changing if by_changes else is_write_action(line.get("tool") or "") + if event == "tool_call" and writes: + action = _cut(f"{_short_tool(line['tool'])} {_key_args(line.get('input', line.get('input_preview')))}", + ACTION_CHARS) + out.append(f"- {'FAILED ' if line.get('error') else ''}action: {action} ({path}:{i})") + elif event == "tool_call" and line.get("error"): + out.append(f"- error: {_short_tool(line['tool'])} failed ({path}:{i})") + elif event == "message_to_user": + out.append(f"- sent the user ({label_time(line.get('at'))}): \"{_cut(_text(line, 'text'), INDEX_TEXT_CHARS)}\" " + f"({path}:{i})") + elif event == "turn_end": + if line.get("stay_quiet_calls"): + out.append(f"- stayed quiet ({path}:{i})") + if line.get("state") not in (None, "completed", "TaskState.completed"): + out.append(f"- error: its turn ended {line.get('state')} ({path}:{i})") + elif event in ("trajectory_missing", "trajectory_unreadable"): + out.append(f"- error: the {line.get('turn')}'s trajectory is " + f"{'missing' if event == 'trajectory_missing' else 'unreadable'}: its tool calls aren't here " + f"({path}:{i})") + elif event == "nudge": + out.append(f"- nudged after closing text; then {line.get('outcome')} ({path}:{i})") + elif event == "app_change": + changes["by the assistant" if line.get("by") == "agent" else "by others" if line.get("by") + else "with no known actor"] += 1 + elif event == "app_changes_not_recorded": + out.append(f"- {line.get('count')} app changes weren't recorded ({path}:{i})") + if changes: + out.append("- app changes: " + ", ".join(f"{count} {who}" for who, count in changes.items()) + + f" (\"event\": \"app_change\" in {path})") + untold = sum(1 for line in lines if line["event"] == "app_change" and line.get("by") == "agent" + and not any(a.get("tool") for a in line.get("actors") or [])) + if untold: + out.append(f"- {untold} of the assistant's app changes name no tool (made through a website, say)") + return "\n".join(out) + + +def _index(stager: _Stager, blocks: list[str], start: list[dict], end: list[dict], + firings: Sequence[Firing]) -> str: + first = next((line for line in start if line["event"] == "start"), {}) + last = end[0] if end else {} + width = stager.width + lines = ["# The run, wake by wake", "", + f"{len(stager.wakes)} wakes, from {label_time(first.get('at'))} to {label_time(last.get('at'))}. " + f"End: {last.get('note') or '(not recorded)'} (end reason: {last.get('reason')}).", + "", + f"Each wake's events are in {wake_file(1, width)} … {wake_file(max(stager.wakes or [1]), width)}, one JSON " + "object a line; `file:line` below points into them. Long tool results and inputs are in " + f"{RESULTS_DIR}/ (the tool_call line names its file). {EVENTS_FILE} has every wake's lines in one file; " + f"{TRIGGERS_FILE} has every trigger firing. A line too long to read whole keeps a preview of its longest " + f"fields and names the file in {RESULTS_DIR}/ that holds them (\"_file\").", + "", + "Actions below are the assistant's tool calls of a tool that changed the apps' data in that wake (from the " + "apps' change logs)." if stager.by_changes else + "Actions below are the assistant's tool calls whose tool name has no read word (get, list, search, …): " + "this run's apps kept no change log, so this is a guess from the name."] + gaps = sorted(set(stager.missing) | set(stager.unreadable)) + if gaps: + lines += ["", "Wakes whose tool calls are not here (their trajectory is missing or unreadable): " + + ", ".join(str(n) for n in gaps) + "."] + if firings: + per: dict[str, list[Firing]] = {} + for f in firings: + per.setdefault(f"{f.trigger_id} ({f.source})", []).append(f) + lines += ["", "## Triggers"] + for name, fs in per.items(): + failed = sum(1 for f in fs if not f.ok) + times = [f.virtual_time for f in fs if f.virtual_time] + span = f", {label_time(min(times).isoformat())} to {label_time(max(times).isoformat())}" if times else "" + lines.append(f"- {name}: fired {len(fs) - failed}, failed {failed}{span}") + return "\n".join(lines + [""] + [block + "\n" for block in blocks]) + + +def _triggers(firings: Sequence[Firing], unavailable: Mapping[str, str], wakes: Mapping[int, _Wake]) -> str: + """One line per firing, oldest first, with the last wake at or before its pretend time (``after_wake``, 0 before + the first); then one per firing log that couldn't be read.""" + starts = sorted((w.at, n) for n, w in wakes.items() if w.at) + out = [] + for f in sorted(firings, key=lambda f: f.at): + vt = f.virtual_time.isoformat().replace("+00:00", "Z") if f.virtual_time else None + after = max((n for at, n in starts if vt and at <= vt), default=0) + out.append(_dumps({"trigger_id": f.trigger_id, "source": f.source, "fired": f.ok, "at": vt, + "recorded_at": f.at.isoformat(), "after_wake": after, "provoked_by": f.provoked_by})) + out += [_dumps({"unavailable": source, "why": why}) for source, why in unavailable.items() if source != "conversation"] + return "".join(line + "\n" for line in out) + + +_WAKE_NAME = re.compile(r"wake_(\d+)") +# A shell loop over wake numbers (``for w in 080 075; do … wake_$w.jsonl``): the numbers it names. +_WAKE_LOOP = re.compile(r"\bfor\s+(\w+)\s+in\s+([\d\s]+?)\s*;") +# A glob or directory walk that takes every wake file at once. +_ALL_WAKES = re.compile(r"wake_\*|wakes/\*|os\.walk|rglob|listdir") +_WAKE_FIELD = re.compile(r'"wake":\s*(\d+)') +_HELPER_TOOLS = frozenset({"Agent", "Task", "Workflow"}) + + +def judge_reads(judge: TurnEvents) -> dict: + """What a judge's trajectory shows it read of the staged run, whichever tool it read with (Read, Grep, or Bash + running grep, sed or python): the wakes its calls named (a file, a result file, a shell loop's numbers) or whose + events came back in a result (``"wake": N``; a result of reading index.md, which names every wake, doesn't count), + whether one call took every wake file at once (``read_all_wakes``), whether it read index.md or searched + events.jsonl / triggers.jsonl, and any helper agent it started.""" + calls = [e for e in judge.events if e["type"] == "tool_call"] + wakes: set[int] = set() + every = False + for call in calls: + named = _dumps(call.get("input")) + wakes |= {int(n) for n in _WAKE_NAME.findall(named)} + for var, numbers in _WAKE_LOOP.findall(named): + if f"wake_${var}" in named or f"wake_${{{var}}}" in named: + wakes |= {int(n) for n in numbers.split()} + every = every or bool(_ALL_WAKES.search(named)) + if INDEX_FILE not in named: + wakes |= {int(n) for n in _WAKE_FIELD.findall(_dumps(call.get("result")))} + text = " ".join(_dumps(c.get("input")) for c in calls) + return {"tool_calls": len(calls), "wakes_opened": sorted(wakes), "read_all_wakes": every, + "read_index": INDEX_FILE in text, "searched_events": EVENTS_FILE in text, + "read_triggers": TRIGGERS_FILE in text, + "helpers_started": sum(1 for c in calls if _short_tool(c["tool"]) in _HELPER_TOOLS)} diff --git a/tst/unit/agents/claude_code/test_always_on_judge_config.py b/tst/unit/agents/claude_code/test_always_on_judge_config.py new file mode 100644 index 000000000..a74026ac9 --- /dev/null +++ b/tst/unit/agents/claude_code/test_always_on_judge_config.py @@ -0,0 +1,34 @@ +"""The always-on process judge's settings are ones the Claude Code agent takes, and do what the judge needs: its card +accepts builtin_tools, max_turns and peer_tools; the CLI offers only Bash and Read of its own tools (and the structured +output it adds), with no MCP server but the task's.""" + +from typing import get_args + +from agent_env.task_step.task_steps.verifiers.always_on_agent_verifier import JUDGE_TOOLS, AlwaysOnAgentVerifier + +from tst.unit.agents.claude_code.agent import a2a_server, claude_bridge +from tst.unit.agents.claude_code.support import card + + +def _judge_config() -> a2a_server.ClaudeCodeConfig: + verifier = AlwaysOnAgentVerifier("judge", 1, criteria=[{"id": "c1"}], prompt_id="ask") + return a2a_server.ClaudeCodeConfig(**verifier.judge_agent_config(), max_turns=verifier.judge_max_turns) + + +def test_the_agent_takes_the_judge_s_settings(app): + [config] = [e for e in card(app)["capabilities"]["extensions"] if e["uri"] == "urn:agentenv:agent-config/v1"] + supported = config["params"]["methods"]["set"]["request"]["supported"] + assert {"builtin_tools", "max_turns", "peer_tools"} <= set(supported) + assert set(JUDGE_TOOLS) <= set(get_args(claude_bridge.ClaudeCodeTool)) + assert _judge_config().max_turns == 400 + + +def test_the_judge_s_cli_offers_bash_and_read_and_no_mcp_server_of_its_own(): + cfg = _judge_config() + command = claude_bridge._command("sess", False, model=None, allowed_tools=None, mcp_servers=None, + system_prompt=None, output_format=None, role=None, effort=None, + max_turns=cfg.max_turns, builtin_tools=cfg.builtin_tools) + assert command[command.index("--tools") + 1] == "Bash,Read" + assert "--strict-mcp-config" in command and "--mcp-config" not in command + assert command[command.index("--max-turns") + 1] == "400" + diff --git a/tst/unit/task_step/test_always_on_agent_verifier.py b/tst/unit/task_step/test_always_on_agent_verifier.py index 12b2bf751..e98052501 100644 --- a/tst/unit/task_step/test_always_on_agent_verifier.py +++ b/tst/unit/task_step/test_always_on_agent_verifier.py @@ -1,14 +1,20 @@ """always_on_agent_verifier: an always_on step's prompt is graded from the user's chat (the response and the -conversation) and the run's context (the trigger log, the run wake by wake with its app changes), never every wake's -trajectory. What needs the trajectories is refused when the task loads, as is a plain rubrics_verifier on an always_on -step's prompt; a plain rubrics_verifier grades any other prompt as before.""" +conversation) and the run's context (the trigger log, the run wake by wake with its app changes) and, with +use_trajectory, every wake's tool calls staged as files in an agent judge's sandbox. A direct judge with the +trajectories, and what cites screenshots or cuts turns, is refused when the task loads, as is a plain rubrics_verifier +on an always_on step's prompt; a plain rubrics_verifier grades any other prompt as before.""" from __future__ import annotations +import json +from pathlib import Path + import pytest +from agent_env.config import get_config +from agent_env.providers.sandbox_providers import sandbox_provider from agent_env.task.task import Task -from agent_env.task_step.context import PromptResponse, TaskStepContext +from agent_env.task_step.context import DeployedAgent, PromptResponse, TaskStepContext from agent_env.task_step.task_steps import always_on as ao from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep from agent_env.task_step.task_steps.verifiers import is_rubrics_verifier @@ -22,8 +28,8 @@ def _verifier(cls=AlwaysOnAgentVerifier, **kw) -> RubricsVerifierTaskStep: - return cls(id="judge", version=1, criteria=CRITERIA, prompt_id="ask", use_agent_judge=False, verifier_id="rubric", - **kw) + return cls(**{"id": "judge", "version": 1, "criteria": CRITERIA, "prompt_id": "ask", "use_agent_judge": False, + "verifier_id": "rubric", **kw}) def _capture(monkeypatch, verifier) -> dict: @@ -151,7 +157,7 @@ def test_the_wake_by_wake_text_says_when_the_record_is_unreadable(): ({"output_format": "rubric_evidence"}, "output_format=rubric_evidence cites per-criterion screenshots"), ({"output_format": "rubric_evidence_with_mistakes"}, "cites per-criterion screenshots"), ({"grade_up_to_turn": 2}, "grade_up_to_turn can't cut an always_on step's prompt"), - ({"use_trajectory": True}, "use_trajectory must be off"), + ({"use_trajectory": True, "use_agent_judge": False}, "use_trajectory needs the agent judge"), ]) def test_what_needs_the_wakes_trajectories_is_refused_when_the_task_loads(settings, why): with pytest.raises(ValueError, match=why): @@ -221,3 +227,144 @@ def test_is_rubrics_verifier_recognises_the_subclass_as_a_step_and_as_a_stored_d assert is_rubrics_verifier(verifier) and is_rubrics_verifier(verifier.to_dict()) assert not is_rubrics_verifier(_always_on()) and not is_rubrics_verifier(_always_on().to_dict()) assert not is_rubrics_verifier({"type": "no_such_step"}) and not is_rubrics_verifier({}) + + +# ---- use_trajectory: the run staged as files for an agent judge --------------------------------------------------- + +RECORDED = Path(__file__).resolve().parent / "verifiers" / "recorded" +JUDGE_TRAJECTORY = json.dumps([ + {"type": "system", "subtype": "init", "tools": ["Read", "Grep", "Bash"]}, + {"type": "assistant", "message": {"content": [ + {"type": "tool_use", "id": "a", "name": "Read", "input": {"file_path": "/tmp/always_on_run/x/index.md"}}, + {"type": "tool_use", "id": "b", "name": "Grep", + "input": {"pattern": "cancel", "path": "/tmp/always_on_run/x/wakes/wake_003.jsonl"}}]}}, + {"type": "result", "subtype": "success", "result": "{}"}]) + + +class _Sandbox: + def __init__(self): + self.files: dict[str, str] = {} + + async def write_file_from_text(self, content, destination_path, **kwargs): + self.files[destination_path] = content + + +def _staged_run(monkeypatch, *, object_url: bool = True) -> tuple[TaskStepContext, _Sandbox, dict[str, str]]: + """A recorded always_on run's context with a deployed judge; ``sources`` serves each stored trajectory.""" + stored = json.loads((RECORDED / "always_on_timeline.json").read_text()) + urls = [e["trajectory_url"] for e in stored["timeline"] if e["event"] == "agent_turn_end"] + record = {k: v for k, v in stored.items() if k != "timeline"} + if object_url: + record["object_url"] = get_config().get_object_store().put( + "always_on_timeline/week.json", json.dumps(stored).encode(), content_type="application/json") + ctx = TaskStepContext() + ctx.metadata[ao.ALWAYS_ON_KEY] = {"week": record} + ctx.prompt_responses.append(PromptResponse(prompt_id="ask", response="[Fri Nov 6] Dinner is booked", + prompt_text="dinner for 7", step_id="week", + target_agent_per_turn_trajectory_object_urls=urls, + agent_trajectory_object_url=urls[-1])) + ctx.deployed_agents.append(DeployedAgent(agent_name="judge", api_url="http://judge.test", + a2a_url="http://judge.test", sandbox_id="sb-judge")) + sandbox = _Sandbox() + + class Provider: + async def get_sandbox(self, sandbox_id): + assert sandbox_id == "sb-judge" + return sandbox + + monkeypatch.setattr(sandbox_provider, "get_agent_sandbox_provider", lambda: Provider()) + sources = {urls[1]: (RECORDED / "turn_claude_code.json").read_text(), + urls[3]: (RECORDED / "turn_otel_wake4.json").read_text(), + "mem://judge/1.json": JUDGE_TRAJECTORY} + return ctx, sandbox, sources + + +def _agent_judge(monkeypatch, verifier, sources) -> dict: + captured: dict = {} + + async def judge(*, eval_prompt, **kwargs): + captured["prompt"] = eval_prompt + return ([{**verifier.criteria[0], "score": 1.0, "result": True}], 0, [], "mem://judge/1.json") + + def read(url): + if url not in sources: + raise FileNotFoundError(url) + return sources[url] + + monkeypatch.setattr(verifier, "_run_judge_with_output_retries", judge) + monkeypatch.setattr(verifier, "_read_trajectory_text", read) + return captured + + +@pytest.mark.asyncio +async def test_with_use_trajectory_the_agent_judge_searches_every_wake_staged_as_files(monkeypatch): + ctx, sandbox, sources = _staged_run(monkeypatch) + verifier = _verifier(use_agent_judge=True, use_trajectory=True, agent_name="judge") + captured = _agent_judge(monkeypatch, verifier, sources) + + await verifier.execute(ctx) + + [root] = {path.split("/")[3] for path in sandbox.files} + base = f"/tmp/always_on_run/{root}" + assert {f"{base}/index.md", f"{base}/events.jsonl", f"{base}/triggers.jsonl", + *(f"{base}/wakes/wake_00{n}.jsonl" for n in range(1, 5))} <= set(sandbox.files) + assert '"tool": "porchbite_spending_by_month"' in sandbox.files[f"{base}/wakes/wake_004.jsonl"] + prompt = captured["prompt"] + assert f"The agent's trajectory is available at {base}" in prompt + assert "## The run, wake by wake" in prompt and "## The run's files" in prompt + assert f"- {base}/index.md: start here." in prompt and "Later actions replace earlier ones" in prompt + assert "You have Bash and Read only: no web, no other agents, and nothing to change." in prompt + assert "These wakes' tool calls are not in the files (their trajectory is missing or unreadable), so for them " \ + "only the run's record is there: 1, 3." in prompt + staging = ctx.metadata["verifications"]["rubric"]["trajectory_staging"] + assert staging["wakes"] == 4 and staging["unreadable_trajectories"] == [1, 3] + assert staging["raw_bytes"] > staging["staged_bytes"] > 0 + assert staging["judge_reads"] == [{"batch": 1, "url": "mem://judge/1.json", "tool_calls": 2, + "wakes_opened": [3], "read_all_wakes": False, "read_index": True, + "searched_events": False, "read_triggers": False, "helpers_started": 0}] + + +@pytest.mark.asyncio +async def test_without_the_stored_timeline_the_turns_are_loaded_as_rubrics_verifier_loads_them(monkeypatch): + ctx, sandbox, sources = _staged_run(monkeypatch, object_url=False) + verifier = _verifier(use_agent_judge=True, use_trajectory=True, agent_name="judge") + captured = _agent_judge(monkeypatch, verifier, sources) + loaded: list = [] + + async def load(sandbox, uris, container_dir, *args): + loaded.append((uris, container_dir)) + + monkeypatch.setattr(verifier, "_load_per_turn_trajectories", load) + + await verifier.execute(ctx) + + [(uris, container_dir)] = loaded + assert len(uris) == 4 and "NOTE: This is a multi-turn conversation" in captured["prompt"] + assert "trajectory_staging" not in ctx.metadata["verifications"]["rubric"] and sandbox.files == {} + + +@pytest.mark.asyncio +async def test_a_run_that_turns_the_agent_judge_off_is_refused_with_use_trajectory(monkeypatch): + ctx, _, sources = _staged_run(monkeypatch) + verifier = _verifier(use_agent_judge=True, use_trajectory=True, agent_name="judge") + _agent_judge(monkeypatch, verifier, sources) + ctx.metadata["user_overrides"] = {"use_agent_judge": False} + + with pytest.raises(RuntimeError, match="use_trajectory needs the agent judge"): + await verifier.execute(ctx) + + +def test_use_trajectory_with_the_agent_judge_loads_and_round_trips_with_its_defaults(): + stored = {"id": "judge", "type": "always_on_agent_verifier", "prompt_id": "ask", "criteria": CRITERIA, + "use_trajectory": True} + [verifier] = Task.from_dict({"id": "t", "version": 1, "steps": [stored]}).steps + assert verifier.use_trajectory and verifier.use_agent_judge + assert (verifier.judge_timeout_seconds, verifier.judge_timeout_retries, verifier.judge_max_turns) == (2400, 1, 400) + assert verifier.DEFAULT_JUDGE_BATCH_CONCURRENCY == 3 and verifier.judge_batch_size is None + assert verifier.judge_agent_config() == {"builtin_tools": {"only": ["Bash", "Read"]}, "peer_tools": False} + assert AlwaysOnAgentVerifier.from_dict(verifier.to_dict()).to_dict() == verifier.to_dict() + task_set = _verifier(judge_timeout_seconds=1000, judge_timeout_retries=0, judge_batch_size=6) + assert (task_set.judge_timeout_seconds, task_set.judge_timeout_retries, task_set.judge_batch_size) == (1000, 0, 6) + plain = RubricsVerifierTaskStep(id="j", version=1, criteria=CRITERIA, prompt_id="ask") + assert (plain.judge_timeout_seconds, plain.judge_timeout_retries, plain.judge_max_turns, + plain.judge_agent_config()) == (1000, None, None, {}) diff --git a/tst/unit/task_step/verifiers/recorded/always_on_timeline.json b/tst/unit/task_step/verifiers/recorded/always_on_timeline.json new file mode 100644 index 000000000..ee3488c64 --- /dev/null +++ b/tst/unit/task_step/verifiers/recorded/always_on_timeline.json @@ -0,0 +1,816 @@ +{ + "step_id": "msg-ts-rw-something-new-prompt-step", + "end_reason": "end_time", + "end_note": "The run reached its end time.", + "error": null, + "error_type": null, + "counts": { + "wakes": 11, + "messages_to_user": 6, + "user_sim_reads": 6, + "user_replies": 2, + "user_no_replies": 3, + "user_dones": 1, + "scripted_user_lines": 0, + "clock_jumps": 8, + "silent_wakes": 5, + "user_sim_malformed_answers": 0, + "app_changes": { + "by_agent": 25, + "by_others": 15, + "no_actor": 0 + }, + "nudges": 0, + "nudge_outcomes": {}, + "stay_quiet_wakes": 5 + }, + "user_sim": { + "name": "user-sim", + "version": "ours", + "agent_name": "user-sim", + "accepted_fields": [ + "model", + "output_format", + "system_prompt" + ], + "model": "claude-sonnet-4-6", + "model_from": "user_sim_step", + "model_set": true, + "output_format": { + "schema": { + "additionalProperties": false, + "properties": { + "message": { + "type": "string" + }, + "outcome": { + "enum": [ + "reply", + "no_reply", + "done" + ], + "type": "string" + }, + "score": { + "type": "integer" + }, + "used": { + "type": "string" + } + }, + "required": [ + "outcome", + "message" + ], + "type": "object" + }, + "type": "json_schema" + }, + "system_prompt_chars": 5366, + "user_sim_step_model": "claude-sonnet-4-6" + }, + "end_time": "2026-11-07T04:00:00Z", + "time_zone": "America/Chicago", + "multi_day": false, + "real_seconds": 426.5, + "timeline_events": 63, + "pending_at_end": { + "held_replies": { + "count": 1, + "items": [ + { + "at": "2026-11-07T04:00:00Z", + "text": "omg it was perfect, thank you so much!! 🥂❤️", + "done": true + } + ] + }, + "unread_by_user": { + "count": 0, + "items": [] + }, + "user_lines_not_delivered": { + "count": 0, + "items": [] + }, + "alerts_never_delivered": { + "count": 0, + "items": [] + }, + "scheduled_checks_not_due": { + "count": 0, + "items": [] + }, + "contact_replies_in_progress": { + "count": 0, + "items": [] + } + }, + "settings": { + "stay_quiet": true, + "nudge": true + }, + "timeline": [ + { + "event": "start", + "real_time": "2026-11-13T17:23:26.449320+00:00", + "real_elapsed_s": 6.8, + "virtual_time": "2026-11-06T20:00:00Z", + "end_time": "2026-11-07T04:00:00Z", + "agent": "default-agent", + "agent_role": "assistant", + "user_sim": "user-sim", + "version": "ours", + "user_model": "claude-sonnet-4-6", + "max_wakes": 20, + "time_zone": "America/Chicago", + "multi_day": false, + "scheduled_checks": { + "on": false, + "envs": [] + }, + "clock_rate": 0.0 + }, + { + "event": "wake", + "real_time": "2026-11-13T17:23:28.418692+00:00", + "real_elapsed_s": 8.8, + "n": 1, + "virtual_time": "2026-11-06T20:00:00Z", + "woke_by": [ + "user_request" + ], + "notifications": [], + "last": false, + "parts_text": "[Fri Nov 6, 8:00 PM UTC] Nina: tonight's our 5 year anniversary w jordan!! can you get dinner delivered here for 7? keep it under $120. and i want something we've literally never had before, a totally new kind of food. NOT italian and none of our usual places, we're in such a rut lol" + }, + { + "event": "app_changes", + "real_time": "2026-11-13T17:24:13.465864+00:00", + "real_elapsed_s": 53.9, + "wake": 1, + "phase": "turn", + "env_id": "world", + "count": 4, + "changes": [ + { + "env_id": "world", + "id": 2, + "table": "porchbite.carts", + "op": "INSERT", + "row_id": "1", + "summary": "Inserted: {\"id\": 1, \"card_id\": \"card-visa-4242\", \"cart_id\": \"cart-517c5c956bc9\", \"user_id\": 1, \"store_id\": \"pb-atx-9201\", \"tip_kind\": \"percent\", \"tip_value\": 18.000000, \"address_id\": \"addr-home\", \"pro…", + "at": "2026-11-13T17:23:50.983919+00:00", + "by": "agent", + "wake": 1, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 3, + "table": "porchbite.cart_items", + "op": "INSERT", + "row_id": "1", + "summary": "Inserted: {\"id\": 1, \"note\": \"Shellfish allergy - please keep free of shellfish\", \"cart_id\": \"cart-517c5c956bc9\", \"item_id\": \"pb-atx-9201-lomo-saltado\", \"line_id\": \"line-df273d68b612\", \"options\": \"[]\",…", + "at": "2026-11-13T17:23:50.983919+00:00", + "by": "agent", + "wake": 1, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 4, + "table": "porchbite.cart_items", + "op": "INSERT", + "row_id": "2", + "summary": "Inserted: {\"id\": 2, \"note\": \"Shellfish allergy - please keep free of shellfish\", \"cart_id\": \"cart-517c5c956bc9\", \"item_id\": \"pb-atx-9201-aji-de-gallina\", \"line_id\": \"line-d6fe7f165ce4\", \"options\": \"[]…", + "at": "2026-11-13T17:23:55.857396+00:00", + "by": "agent", + "wake": 1, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 5, + "table": "porchbite.cart_items", + "op": "INSERT", + "row_id": "3", + "summary": "Inserted: {\"id\": 3, \"note\": \"Shellfish allergy - fish only, please avoid any shellfish cross-contact\", \"cart_id\": \"cart-517c5c956bc9\", \"item_id\": \"pb-atx-9201-ceviche-clasico\", \"line_id\": \"line-dd000b…", + "at": "2026-11-13T17:23:55.915990+00:00", + "by": "agent", + "wake": 1, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + } + ] + }, + { + "event": "agent_turn_end", + "real_time": "2026-11-13T17:24:15.204876+00:00", + "real_elapsed_s": 55.6, + "n": 1, + "state": "completed", + "virtual_time": "2026-11-06T20:00:00Z", + "virtual_seconds": 0, + "real_seconds": 46.8, + "tool_calls": 19, + "plain_reply": "I ordered Peruvian from Ember & Saffron and messaged Nina about it. The order is PB-C2SHSLHX, coming to her home at 1507 Brookhaven between 6:30 and 7:00 PM. The total is $102.70, under her $120 limit, charged to her Visa ending 5517.\n\n- **Something new:** they've never ordered Peruvian, and this restaurant isn't one of their usual places. They often order Italian from Luigi's, but this is something different.\n- **What's coming:** Ceviche Clasico (sea bass, no shellfish), Lomo Saltado (seared beef), Aji de Gallina (chicken in a creamy yellow-pepper sauce) and Alfajores (dulce de leche cookies).\n- **Cost:** $79 of food, $9.48 service fee and an 18% tip ($14.22). Delivery was free with her PorchPass membership.\n- **Shellfish:** her past orders always say \"shellfish allergy,\" so I added that note to each dish. I skipped the restaurant's \"Anniversary Tasting for Two\" because it comes with shrimp and octopus and they don't allow substitutions.\n- **The surprise:** her calendar says Jordan thinks they're cooking, and the delivery is set to be left at the door. I flagged that so she can keep an eye out.\n\nRESULT: DONE: Peruvian dinner ordered, arriving 6:30–7:00 PM, $102.70.", + "messages_to_user": 1, + "trajectory_url": "mem://trajectories/61bd7b34.json", + "stay_quiet_calls": 0 + }, + { + "event": "message_to_user", + "real_time": "2026-11-13T17:24:15.205086+00:00", + "real_elapsed_s": 55.6, + "virtual_time": "2026-11-06T20:00:00Z", + "text": "Happy 5 years to you and Jordan! 🥂 Dinner is booked: Peruvian from Ember & Saffron, a new spot on East 6th rated 4.9. You've never ordered Peruvian before, and it isn't one of your usual places.\n\nIt's arriving at home (1507 Brookhaven) between 6:30 and 7:00 PM, so it'll be there by 7:\n• Ceviche Clasico (sea bass, no shellfish)\n• Lomo Saltado (seared beef tenderloin)\n• Aji de Gallina (chicken in a creamy yellow-pepper sauce)\n• Alfajores (dulce de leche cookies) for dessert\n\nTotal is $102.70 with an 18% tip, on your Visa ending 5517. PorchPass covered delivery. Order PB-C2SHSLHX.\n\nThey have an \"Anniversary Tasting for Two\" set menu, but I didn't get it: it comes with shrimp and octopus and they don't allow substitutions. Your past orders always say shellfish allergy, so I put that note on each dish.\n\nSince Jordan's expecting you to cook, heads up: the driver's note says to leave it at the door. If you want anything changed, tell me before then.\n\nRESULT: DONE: Peruvian dinner ordered, arriving 6:30–7:00 PM, $102.70.", + "wake": 1 + }, + { + "event": "user_read_scheduled", + "real_time": "2026-11-13T17:24:17.048058+00:00", + "real_elapsed_s": 57.4, + "read_at": "2026-11-06T20:00:00Z", + "mode": "now", + "unread": 1 + }, + { + "event": "user_sim_read", + "real_time": "2026-11-13T17:24:40.801748+00:00", + "real_elapsed_s": 81.2, + "virtual_time": "2026-11-06T20:00:00Z", + "version": "ours", + "read": "{\"time\": \"2026-11-06T20:00:00Z\", \"assistant_messages\": [{\"at\": \"2026-11-06T20:00:00Z\", \"text\": \"Happy 5 years to you and Jordan! \\ud83e\\udd42 Dinner is booked: Peruvian from Ember & Saffron, a new spo…", + "input": "It's Fri, Nov 6, 2:00 PM.\nYou last messaged your AI assistant at 2:00 PM.\n\nYOU SENT YOUR AI ASSISTANT:\n [2:00 PM] tonight's our 5 year anniversary w jordan!! can you get dinner delivered here for 7? …", + "raw": "{\"outcome\": \"no_reply\", \"message\": \"\", \"used\": \"none\"}", + "outcome": "no_reply", + "message": "", + "used": "none", + "score": null, + "schema_errors": [], + "mode": "now", + "real_seconds": 20.2, + "looks_like_assistant": false, + "user_sim_error": null, + "user_sim_trajectory_urls": [ + "mem://trajectories/c11357af.json" + ], + "user_sim_turns": 1, + "shown": null, + "past_turn_cap": null, + "last_read": false, + "retried": false, + "first_answer": null + }, + { + "event": "clock_jump", + "real_time": "2026-11-13T17:24:45.703720+00:00", + "real_elapsed_s": 86.1, + "from": "2026-11-06T20:00:00Z", + "to": "2026-11-06T21:00:00Z", + "reason": "world_event", + "settled": true, + "still_at_work": [], + "settle": { + "seconds": 1.0, + "contact_calls_seen": 0 + } + }, + { + "event": "app_changes", + "real_time": "2026-11-13T17:24:49.554179+00:00", + "real_elapsed_s": 89.9, + "wake": 2, + "phase": "before_turn", + "env_id": "world", + "count": 3, + "changes": [ + { + "env_id": "world", + "id": 14, + "table": "messaging.conversations", + "op": "UPDATE", + "row_id": "conv-tess", + "summary": "last_updated: '2026-11-05T18:03:00+00:00' -> '2026-11-06T21:00:00+00:00'", + "at": "2026-11-13T17:24:44.664937+00:00", + "by": "others", + "wake": 2, + "phase": "before_turn", + "actors": [ + { + "kind": "trigger_action", + "trigger_id": "noise-tess", + "tool": "messaging_world_send_message", + "role": "world" + } + ] + }, + { + "env_id": "world", + "id": 15, + "table": "messaging.messages", + "op": "INSERT", + "row_id": "c181c35f12a345f68c4066c04f62cf0d", + "summary": "Inserted: {\"content\": \"did you end up trying ember & saffron?? 👀\", \"sender_id\": \"+15125550303\", \"timestamp\": \"2026-11-06T21:00:00+00:00\", \"message_id\": \"c181c35f12a345f68c4066c04f62cf0d\", \"attachment_…", + "at": "2026-11-13T17:24:44.664937+00:00", + "by": "others", + "wake": 2, + "phase": "before_turn", + "actors": [ + { + "kind": "trigger_action", + "trigger_id": "noise-tess", + "tool": "messaging_world_send_message", + "role": "world" + } + ] + }, + { + "env_id": "world", + "id": 16, + "table": "messaging.world_events", + "op": "INSERT", + "row_id": "60b3fae23a99455d930ed4a9558c2156", + "summary": "Inserted: {\"at\": \"2026-11-06T21:00:00+00:00\", \"role\": \"world\", \"party\": \"Tess Albright\", \"action\": \"message\", \"event_id\": \"60b3fae23a99455d930ed4a9558c2156\", \"sender_id\": \"+15125550303\", \"message_id\":…", + "at": "2026-11-13T17:24:44.675128+00:00", + "by": "others", + "wake": 2, + "phase": "before_turn", + "actors": [ + { + "kind": "trigger_action", + "trigger_id": "noise-tess", + "tool": "messaging_world_send_message", + "role": "world" + } + ] + } + ] + }, + { + "event": "wake", + "real_time": "2026-11-13T17:24:49.554398+00:00", + "real_elapsed_s": 89.9, + "n": 2, + "virtual_time": "2026-11-06T21:00:00Z", + "woke_by": [ + "notification:Messages" + ], + "notifications": [ + { + "at": "2026-11-06T21:00:00Z", + "channel": "Messages", + "source": "trigger", + "source_id": "noise-tess", + "to": "assistant" + } + ], + "last": false, + "parts_text": "[Fri Nov 6, 9:00 PM UTC · Messages] New text from Tess Albright" + }, + { + "event": "app_changes", + "real_time": "2026-11-13T17:25:29.569621+00:00", + "real_elapsed_s": 130.0, + "wake": 2, + "phase": "turn", + "env_id": "world", + "count": 3, + "changes": [ + { + "env_id": "world", + "id": 17, + "table": "porchbite.universe_clock", + "op": "UPDATE", + "row_id": "1", + "summary": "wall_time: '2026-11-13T17:22:55.000000Z' -> '2026-11-13T17:25:04.000000Z', virtual_time: '2026-11-06T20:00:00.000000Z' -> '2026-11-06T21:00:00.000000Z'", + "at": "2026-11-13T17:25:03.831101+00:00", + "by": "agent", + "wake": 2, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_get_store", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 18, + "table": "porchbite.carts", + "op": "INSERT", + "row_id": "2", + "summary": "Inserted: {\"id\": 2, \"card_id\": \"card-visa-4242\", \"cart_id\": \"cart-92dd932172c9\", \"user_id\": 1, \"store_id\": \"pb-atx-9301\", \"tip_kind\": \"percent\", \"tip_value\": 18.000000, \"address_id\": \"addr-home\", \"pro…", + "at": "2026-11-13T17:25:09.557614+00:00", + "by": "agent", + "wake": 2, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 19, + "table": "porchbite.cart_items", + "op": "INSERT", + "row_id": "5", + "summary": "Inserted: {\"id\": 5, \"note\": \"Shellfish allergy - please keep free of shellfish\", \"cart_id\": \"cart-92dd932172c9\", \"item_id\": \"pb-atx-9301-veggie-feast-for-two\", \"line_id\": \"line-749285476299\", \"options…", + "at": "2026-11-13T17:25:09.557614+00:00", + "by": "agent", + "wake": 2, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + } + ] + }, + { + "event": "agent_turn_end", + "real_time": "2026-11-13T17:25:30.974977+00:00", + "real_elapsed_s": 131.4, + "n": 2, + "state": "completed", + "virtual_time": "2026-11-06T21:00:00Z", + "virtual_seconds": 0, + "real_seconds": 41.4, + "tool_calls": 10, + "plain_reply": "I asked Nina whether to switch dinner to a vegetarian place. Nothing is changed yet, and her Peruvian order (PB-C2SHSLHX) is still set for 6:30–7:00 PM.\n\n**What went wrong:** Jordan is doing no-meat November (no meat, no fish, per the Reyes family group chat) and asked Nina for \"something veggie\" on Friday. Everything I ordered is meat or fish except the cookies, and Ember & Saffron has no vegetarian mains. I should have checked her texts before ordering.\n\n**The fix I suggested:** cancel the Peruvian and order from Lalibela Ethiopian Kitchen instead. They've never ordered Ethiopian, and the Veggie Feast for Two is fully vegan. With lentil pastries and baklava it comes to about $90 with tip, under her $120 limit. The catch is that on Fridays the kitchen opens at 7:30, so it can't arrive before 7:30–8:00 PM. That misses her 7:00 time, so it's her call.\n\nI gave her two choices and recommended the first:\n1. Switch to the Ethiopian, arriving 7:30–8:00.\n2. Keep the Peruvian, arriving by 7.\n\nThe Peruvian order can be cancelled for free until it goes out around 6:30, so I asked her to answer by 5:30. The Ethiopian food is sitting in her cart but not ordered.\n\nHer friend Tess also texted asking whether she'd tried Ember & Saffron. I told Nina about it and didn't reply for her.", + "messages_to_user": 1, + "trajectory_url": "mem://trajectories/b8843a2d.json", + "stay_quiet_calls": 0 + }, + { + "event": "message_to_user", + "real_time": "2026-11-13T17:25:30.975179+00:00", + "real_elapsed_s": 131.4, + "virtual_time": "2026-11-06T21:00:00Z", + "text": "Problem with tonight's order, sorry I missed this earlier. In the Reyes group chat Jordan is doing no-meat November (no meat, no fish), and Jordan texted you \"something veggie 🌱\" for Friday. The Peruvian order is all meat and fish except dessert, and that restaurant has no vegetarian mains.\n\nThe best fix I can find is Lalibela Ethiopian Kitchen. You've never ordered Ethiopian, and their Veggie Feast for Two is fully vegan (lentils, chickpeas, greens, veg stews on injera). With lentil sambusas and honey baklava it comes to about $90 with tip.\n\nThe catch: on Fridays their kitchen doesn't open until 7:30, so the earliest delivery is 7:30–8:00 PM, not 7.\n\nWhich do you want?\n1) Cancel the Peruvian and switch to the Ethiopian, arriving 7:30–8:00 (my pick)\n2) Keep the Peruvian, arriving by 7\n\nThe Peruvian is still cancelable for free, but only until it goes out around 6:30. Can you reply by 5:30?\n\n(Also, Tess just texted asking if you tried Ember & Saffron 👀. I haven't replied.)", + "wake": 2 + }, + { + "event": "user_read_scheduled", + "real_time": "2026-11-13T17:25:32.284491+00:00", + "real_elapsed_s": 132.7, + "read_at": "2026-11-06T21:00:00Z", + "mode": "now", + "unread": 1 + }, + { + "event": "user_sim_read", + "real_time": "2026-11-13T17:25:56.499466+00:00", + "real_elapsed_s": 156.9, + "virtual_time": "2026-11-06T21:00:00Z", + "version": "ours", + "read": "{\"time\": \"2026-11-06T21:00:00Z\", \"assistant_messages\": [{\"at\": \"2026-11-06T21:00:00Z\", \"text\": \"Problem with tonight's order, sorry I missed this earlier. In the Reyes group chat Jordan is doing no-me…", + "input": "It's Fri, Nov 6, 3:00 PM.\nYou last messaged your AI assistant at 2:00 PM.\n\nYOUR NEW EMAILS, TEXTS AND NOTIFICATIONS (since you last checked):\n [3:00 PM · Messages] Tess Albright: did you end up tryin…", + "raw": "{\"outcome\": \"reply\", \"message\": \"omg yes good catch, he's doing no-meat november!! veggie for him for sure, i'm fine with anything \\ud83d\\ude05 please switch to the ethiopian \\u2014 later is totally f…", + "outcome": "reply", + "message": "omg yes good catch, he's doing no-meat november!! veggie for him for sure, i'm fine with anything 😅 please switch to the ethiopian — later is totally fine!", + "used": "K4, D2", + "score": null, + "schema_errors": [], + "mode": "now", + "real_seconds": 23.1, + "looks_like_assistant": false, + "user_sim_error": null, + "user_sim_trajectory_urls": [ + "mem://trajectories/5b825fea.json" + ], + "user_sim_turns": 1, + "shown": null, + "past_turn_cap": null, + "last_read": false, + "retried": false, + "first_answer": null, + "deliver_at": "2026-11-06T21:00:00Z", + "reply_to": { + "wake": 2, + "at": "2026-11-06T21:00:00Z", + "text": "Problem with tonight's order, sorry I missed this earlier. In the Reyes group chat Jordan is doing no-meat November (no meat, no fish), and Jordan texted you \"something veggie 🌱\" for Friday. The Peruvian order is all meat and fish except dessert, and that restaurant has no vegetarian mains.\n\nThe best fix I can find is Lalibela Ethiopian Kitchen. You've never ordered Ethiopian, and their Veggie Feast for Two is fully vegan (lentils, chickpeas, greens, veg stews on injera). With lentil sambusas and honey baklava it comes to about $90 with tip.\n\nThe catch: on Fridays their kitchen doesn't open until 7:30, so the earliest delivery is 7:30–8:00 PM, not 7.\n\nWhich do you want?\n1) Cancel the Peruvian and switch to the Ethiopian, arriving 7:30–8:00 (my pick)\n2) Keep the Peruvian, arriving by 7\n\nThe Peruvian is still cancelable for free, but only until it goes out around 6:30. Can you reply by 5:30?\n\n(Also, Tess just texted asking if you tried Ember & Saffron 👀. I haven't replied.)", + "matched_by": "only_message" + }, + "reply_to_candidates": 1 + }, + { + "event": "user_reply_delivered", + "real_time": "2026-11-13T17:25:59.783947+00:00", + "real_elapsed_s": 160.2, + "virtual_time": "2026-11-06T21:00:00Z", + "text": "omg yes good catch, he's doing no-meat november!! veggie for him for sure, i'm fine with anything 😅 please switch to the ethiopian — later is totally fine!", + "done": false + }, + { + "event": "wake", + "real_time": "2026-11-13T17:26:00.766708+00:00", + "real_elapsed_s": 161.2, + "n": 3, + "virtual_time": "2026-11-06T21:00:00Z", + "woke_by": [ + "user_reply" + ], + "notifications": [], + "last": false, + "parts_text": "[Fri Nov 6, 9:00 PM UTC] Nina: omg yes good catch, he's doing no-meat november!! veggie for him for sure, i'm fine with anything 😅 please switch to the ethiopian — later is totally fine!" + }, + { + "event": "app_changes", + "real_time": "2026-11-13T17:26:20.105473+00:00", + "real_elapsed_s": 180.5, + "wake": 3, + "phase": "turn", + "env_id": "world", + "count": 4, + "changes": [ + { + "env_id": "world", + "id": 20, + "table": "porchbite.orders", + "op": "UPDATE", + "row_id": "52", + "summary": "status: 'scheduled' -> 'cancelled', cancelled_at: NULL -> '2026-11-06T15:00:00'", + "at": "2026-11-13T17:26:05.095493+00:00", + "by": "agent", + "wake": 3, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_cancel_order", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 21, + "table": "porchbite.cart_items", + "op": "INSERT", + "row_id": "6", + "summary": "Inserted: {\"id\": 6, \"note\": \"Shellfish allergy - please keep free of shellfish\", \"cart_id\": \"cart-92dd932172c9\", \"item_id\": \"pb-atx-9301-sambusa\", \"line_id\": \"line-7daf3197ca66\", \"options\": \"[]\", \"use…", + "at": "2026-11-13T17:26:05.149953+00:00", + "by": "agent", + "wake": 3, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 22, + "table": "porchbite.cart_items", + "op": "INSERT", + "row_id": "7", + "summary": "Inserted: {\"id\": 7, \"note\": \"\", \"cart_id\": \"cart-92dd932172c9\", \"item_id\": \"pb-atx-9301-baklava\", \"line_id\": \"line-aff08ff67a30\", \"options\": \"[]\", \"user_id\": 1, \"quantity\": 1, \"created_at\": \"2026-11-0…", + "at": "2026-11-13T17:26:05.213908+00:00", + "by": "agent", + "wake": 3, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_add_to_cart", + "role": "assistant" + } + ] + }, + { + "env_id": "world", + "id": 23, + "table": "porchbite.carts", + "op": "UPDATE", + "row_id": "2", + "summary": "scheduled_for: NULL -> '2026-11-06T19:30:00'", + "at": "2026-11-13T17:26:07.516579+00:00", + "by": "agent", + "wake": 3, + "phase": "turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_update_checkout", + "role": "assistant" + } + ] + } + ] + }, + { + "event": "agent_turn_end", + "real_time": "2026-11-13T17:26:21.428630+00:00", + "real_elapsed_s": 181.8, + "n": 3, + "state": "completed", + "virtual_time": "2026-11-06T21:00:00Z", + "virtual_seconds": 0, + "real_seconds": 20.7, + "tool_calls": 6, + "plain_reply": "Dinner is now vegan Ethiopian from Lalibela Ethiopian Kitchen, arriving at Nina's home between 7:30 and 8:00 PM. That's the order Nina approved after we found out Jordan is doing no-meat November.\n\n- **Peruvian order:** cancelled, and she wasn't charged.\n- **New order (PB-B9SQ90ZE):** Veggie Feast for Two (fully vegan), 3 lentil sambusas and honey baklava. The two savory dishes carry the shellfish allergy note.\n- **Cost:** $89.70 with an 18% tip, under her $120 limit, on her Visa ending 5517. Delivery was free with her PorchPass membership.\n- **Timing:** later than her original 7:00 because the kitchen opens at 7:30 on Fridays. Nina said that's fine.\n\nI reminded Nina that the driver's note says to leave it at the door, in case she wants to keep the surprise. I didn't change her calendar or reply to Tess.\n\nRESULT: DONE: vegan Ethiopian dinner ordered, arriving 7:30–8:00 PM, $89.70.", + "messages_to_user": 1, + "trajectory_url": "mem://trajectories/b47075a9.json", + "stay_quiet_calls": 0 + }, + { + "event": "message_to_user", + "real_time": "2026-11-13T17:26:21.428796+00:00", + "real_elapsed_s": 181.8, + "virtual_time": "2026-11-06T21:00:00Z", + "text": "Done, you're switched! 🎉 I cancelled the Peruvian order and you weren't charged for it.\n\nNew order: Lalibela Ethiopian Kitchen, arriving at home between 7:30 and 8:00 PM.\n• Veggie Feast for Two (vegan: misir wot, shiro, gomen, kik alicha and atakilt on injera)\n• Sambusa, 3 lentil pastries\n• Honey Baklava\n\nEverything is meat- and fish-free for Jordan's no-meat November, and I put the shellfish allergy note on the food. Total is $89.70 with an 18% tip, on your Visa ending 5517, and PorchPass covered delivery. Order PB-B9SQ90ZE.\n\nThe driver's note still says to leave it at the door, so keep an eye out if you want to keep the surprise.\n\nRESULT: DONE: vegan Ethiopian dinner ordered, arriving 7:30–8:00 PM, $89.70.", + "wake": 3 + }, + { + "event": "user_read_scheduled", + "real_time": "2026-11-13T17:26:22.719703+00:00", + "real_elapsed_s": 183.1, + "read_at": "2026-11-06T21:00:00Z", + "mode": "now", + "unread": 1 + }, + { + "event": "user_sim_read", + "real_time": "2026-11-13T17:26:31.979930+00:00", + "real_elapsed_s": 192.4, + "virtual_time": "2026-11-06T21:00:00Z", + "version": "ours", + "read": "{\"time\": \"2026-11-06T21:00:00Z\", \"assistant_messages\": [{\"at\": \"2026-11-06T21:00:00Z\", \"text\": \"Done, you're switched! \\ud83c\\udf89 I cancelled the Peruvian order and you weren't charged for it.\\n\\nNe…", + "input": "It's Fri, Nov 6, 3:00 PM.\nYou last messaged your AI assistant at 3:00 PM.\n\nYOUR EARLIER EMAILS, TEXTS AND NOTIFICATIONS (you've already seen these):\n [3:00 PM · Messages] Tess Albright: did you end u…", + "raw": "{\"outcome\": \"no_reply\", \"message\": \"\", \"used\": \"none\"}", + "outcome": "no_reply", + "message": "", + "used": "none", + "score": null, + "schema_errors": [], + "mode": "now", + "real_seconds": 8.3, + "looks_like_assistant": false, + "user_sim_error": null, + "user_sim_trajectory_urls": [ + "mem://trajectories/145a2742.json" + ], + "user_sim_turns": 1, + "shown": null, + "past_turn_cap": null, + "last_read": false, + "retried": false, + "first_answer": null + }, + { + "event": "clock_jump", + "real_time": "2026-11-13T17:26:43.762323+00:00", + "real_elapsed_s": 204.2, + "from": "2026-11-06T21:00:00Z", + "to": "2026-11-06T21:30:00Z", + "reason": "world_event", + "settled": true, + "still_at_work": [], + "settle": { + "seconds": 7.8, + "contact_calls_seen": 1 + } + }, + { + "event": "app_changes", + "real_time": "2026-11-13T17:26:47.567548+00:00", + "real_elapsed_s": 208.0, + "wake": 4, + "phase": "before_turn", + "env_id": "world", + "count": 1, + "changes": [ + { + "env_id": "world", + "id": 29, + "table": "porchbite.universe_clock", + "op": "UPDATE", + "row_id": "1", + "summary": "wall_time: '2026-11-13T17:25:04.000000Z' -> '2026-11-13T17:26:38.000000Z', virtual_time: '2026-11-06T21:00:00.000000Z' -> '2026-11-06T21:30:00.000000Z'", + "at": "2026-11-13T17:26:38.303858+00:00", + "by": "others", + "wake": 4, + "phase": "before_turn", + "actors": [ + { + "kind": "tool_call", + "tool": "porchbite_list_orders", + "role": "world", + "caused_by": [ + { + "trigger_id": "kitchen-down", + "action_index": 0, + "fire": 1 + } + ] + } + ] + } + ] + }, + { + "event": "wake", + "real_time": "2026-11-13T17:26:47.567725+00:00", + "real_elapsed_s": 208.0, + "n": 4, + "virtual_time": "2026-11-06T21:30:00Z", + "woke_by": [ + "notification:Porchbite" + ], + "notifications": [ + { + "at": "2026-11-06T21:30:00Z", + "channel": "Porchbite", + "source": "trigger", + "source_id": "kitchen-down", + "to": "assistant" + } + ], + "last": false, + "parts_text": "[Fri Nov 6, 9:30 PM UTC · Porchbite] An update on your order from Ember & Saffron (if anything changed)" + }, + { + "event": "agent_turn_end", + "real_time": "2026-11-13T17:27:01.250720+00:00", + "real_elapsed_s": 221.6, + "n": 4, + "state": "completed", + "virtual_time": "2026-11-06T21:30:00Z", + "virtual_seconds": 0, + "real_seconds": 13.7, + "tool_calls": 4, + "plain_reply": "Nothing has changed since Nina's last update, so I didn't message her. The Porchbite notice was about the Ember & Saffron order, which is still cancelled, and her November spending shows it wasn't charged. The vegan Ethiopian order (PB-B9SQ90ZE) is still scheduled to arrive between 7:30 and 8:00 PM for $89.70, and there are no support messages waiting.\n\nRESULT: DONE: vegan Ethiopian dinner is still set for 7:30–8:00 PM, and the cancelled Peruvian order wasn't charged.", + "messages_to_user": 0, + "trajectory_url": "mem://trajectories/76788954.json", + "stay_quiet_calls": 1 + }, + { + "event": "clock_jump", + "real_time": "2026-11-13T17:27:07.468973+00:00", + "real_elapsed_s": 227.9, + "from": "2026-11-06T21:30:00Z", + "to": "2026-11-06T22:15:00Z", + "reason": "world_event", + "settled": true, + "still_at_work": [], + "settle": { + "seconds": 0.9, + "contact_calls_seen": 0 + } + }, + { + "event": "end", + "real_time": "2026-11-13T17:30:24.595080+00:00", + "real_elapsed_s": 425.0, + "reason": "end_time", + "note": "The run reached its end time.", + "virtual_time": "2026-11-07T04:00:00Z", + "wakes": 4, + "replies": 2, + "unread_at_end": 0, + "held_at_end": 0, + "pending": {} + } + ] +} diff --git a/tst/unit/task_step/verifiers/test_always_on_staging.py b/tst/unit/task_step/verifiers/test_always_on_staging.py new file mode 100644 index 000000000..36f0ee23f --- /dev/null +++ b/tst/unit/task_step/verifiers/test_always_on_staging.py @@ -0,0 +1,273 @@ +"""stage_run: an always_on run as files a judge searches, from a recorded timeline and recorded wake trajectories +(``recorded/``): one file per wake with one event per line, each turn in its wake by the timeline's trajectory_url.""" + +from __future__ import annotations + +import copy +import json +from datetime import datetime +from pathlib import Path + +import pytest + +from agent_env.task_step.task_steps.verifiers.always_on_staging import ( + EVENTS_FILE, + INDEX_FILE, + LINE_CHARS, + TRIGGERS_FILE, + judge_reads, + stage_run, +) +from agent_env.task_step.task_steps.verifiers.judge_utils.turn_events import turn_events +from agent_env.task_step.task_steps.verifiers.run_context import Firing + +RECORDED = Path(__file__).resolve().parent / "recorded" + + +def _timeline() -> list[dict]: + return copy.deepcopy(json.loads((RECORDED / "always_on_timeline.json").read_text())["timeline"]) + + +def _url(timeline: list[dict], wake: int) -> str: + return next(e["trajectory_url"] for e in timeline if e["event"] == "agent_turn_end" and e["n"] == wake) + + +def _lines(files: dict, path: str) -> list[dict]: + return [json.loads(line) for line in files[path].splitlines()] + + +GEMINI_NUDGE = json.dumps([ + {"type": "message", "role": "user", "content": "Your closing text was not delivered."}, + {"type": "tool_use", "tool_name": "message_user", "tool_id": "n1", "parameters": {"text": "Switch to veggie?"}}, + {"type": "tool_result", "tool_id": "n1", "status": "success", "output": "{\"sent\": true}"}, + {"type": "result", "status": "success"}]) + + +@pytest.fixture +def run(): + """The recorded four wakes: wake 2's turn a Claude Code one and a nudge after it, wake 3's trajectory never + fetched, wake 1's unreadable, and wake 4 with 250 app changes. ``turns`` is built in an order unlike the wakes', + so only a turn's url can match it to its wake.""" + timeline = _timeline() + wake2, wake4 = _url(timeline, 2), _url(timeline, 4) + for e in timeline: + if e["event"] == "agent_turn_end" and e["n"] == 3: + e["trajectory_url"] = None + at = next(i for i, e in enumerate(timeline) if e["event"] == "message_to_user" and e["wake"] == 2) + timeline.insert(at + 1, {"event": "nudge", "n": 1, "wake": 2, "virtual_time": "2026-11-06T21:00:00Z", + "closing_text": "I asked Nina.", "state": "completed", "outcome": "message_user", + "messages_to_user": 1, "plain_reply_after": "", "trajectory_url": "mem://nudge.json"}) + timeline.insert(at + 2, {"event": "message_to_user", "virtual_time": "2026-11-06T21:00:00Z", + "text": "Switch to veggie?", "wake": 2, "after_nudge": True}) + changes = next(e for e in timeline if e["event"] == "app_changes" and e["wake"] == 4) + changes["changes"] = [{**changes["changes"][0], "id": i, "summary": f"Updated: order {i} " + "x" * 300} + for i in range(250)] + turns = {"mem://nudge.json": turn_events(GEMINI_NUDGE), + wake4: turn_events((RECORDED / "turn_otel_wake4.json").read_bytes()), + wake2: turn_events((RECORDED / "turn_claude_code.json").read_text())} + return timeline, turns + + +def test_each_turn_is_in_its_wake_by_url_and_a_nudge_in_its_wake(run): + staged = stage_run(*run) + files = staged.files + + wake2 = _lines(files, "wakes/wake_002.jsonl") + assert [line["event"] for line in wake2[:2]] == ["woke", "wake_input"] + tools = [line["tool"] for line in wake2 if line["event"] == "tool_call"] + assert any(t.endswith("gmail_get_thread") for t in tools) + nudge_at = next(i for i, line in enumerate(wake2) if line["event"] == "nudge") + assert wake2[nudge_at + 1]["tool"] == "message_user" and wake2[nudge_at + 1]["turn"] == "nudge" + assert [line["call"] for line in wake2 if line["event"] == "tool_call"] == list(range(1, len(tools) + 1)) + wake4 = _lines(files, "wakes/wake_004.jsonl") + assert "porchbite_spending_by_month" in [line.get("tool") for line in wake4] + assert staged.record["formats"] == {"claude-code-stream-json": 1, "otel": 1, "gemini-stream-json": 1} + + +def test_a_missing_or_unreadable_trajectory_is_flagged_where_its_calls_would_be(run): + timeline, turns = run + staged = stage_run(timeline, turns) + + assert staged.record["missing_trajectories"] == [3] and staged.record["unreadable_trajectories"] == [1] + wake3 = _lines(staged.files, "wakes/wake_003.jsonl") + assert wake3[3] == {"event": "trajectory_missing", "wake": 3, "turn": "turn"} + index = staged.files[INDEX_FILE] + assert "the turn's trajectory is missing: its tool calls aren't here (wakes/wake_003.jsonl:4)" in index + assert "Wakes whose tool calls are not here (their trajectory is missing or unreadable): 1, 3." in index + + +def test_what_woke_the_assistant_comes_first_with_the_user_s_whole_message(run): + staged = stage_run(*run) + + head = _lines(staged.files, "wakes/wake_003.jsonl")[:3] + assert [line["event"] for line in head] == ["woke", "user_message", "wake_input"] + assert head[0]["woke_by"] == ["user_reply"] and head[0]["at"] == "2026-11-06T21:00:00Z" + assert head[1]["text"].startswith("omg yes good catch, he's doing no-meat november!!") + assert head[1]["text"].endswith("later is totally fine!") + assert "the user (Fri Nov 6, 9:00 PM UTC): \"omg yes good catch" in staged.files[INDEX_FILE] + + +def test_every_app_change_is_kept_with_its_whole_summary(run): + staged = stage_run(*run) + + changes = [line for line in _lines(staged.files, "wakes/wake_004.jsonl") if line["event"] == "app_change"] + assert len(changes) == 250 and all(len(c["summary"]) > 300 for c in changes) + assert "at" not in changes[0] and changes[0]["recorded_at"].startswith("2026-11-13") + + +def test_long_results_go_to_side_files_and_email_bodies_are_decoded(run): + staged = stage_run(*run) + files = staged.files + + thread = next(line for line in _lines(files, "wakes/wake_002.jsonl") + if line["event"] == "tool_call" and line["tool"].endswith("gmail_get_thread")) + assert thread["result_file"].startswith("wakes/results/wake_002_call_") and "result" not in thread + assert thread["result_file"].endswith("_gmail_get_thread.json") and thread["result_preview"] + assert "Hi Northwick,\n\nPlease send 4 x 50 lb boxes" in json.loads(files[thread["result_file"]])[ + "messages"][0]["payload"]["body"]["text"] + assert all(len(line) <= LINE_CHARS for path, text in files.items() if path.startswith("wakes/wake_") + for line in text.splitlines() if json.loads(line)["event"] == "tool_call") + assert staged.record["side_files"] >= 1 and staged.record["raw_bytes"] > staged.record["staged_bytes"] + + +def _changed_by(timeline: list[dict], wake: int, tool: str) -> None: + """An app change wake ``wake``'s turn made with ``tool``, as the apps' change log records it.""" + event = next(e for e in timeline if e["event"] == "app_changes" and e["wake"] == wake) + event["changes"].append({**event["changes"][0], "id": 999, "table": "gcal.events", "op": "INSERT", "by": "agent", + "actors": [{"kind": "tool_call", "tool": tool, "role": "assistant"}]}) + + +def test_the_index_points_at_the_line_each_action_and_message_is_on(run): + timeline, turns = run + _changed_by(timeline, 2, "gcal_create_event") + staged = stage_run(timeline, turns) + files = staged.files + index = files[INDEX_FILE] + + assert index.startswith("# The run, wake by wake\n\n4 wakes, from Fri Nov 6, 8:00 PM UTC to Sat Nov 7, 4:00 AM UTC.") + assert "Actions below are the assistant's tool calls of a tool that changed the apps' data" in index + block = index[index.index("## Wake 2 "):index.index("## Wake 3 ")] + for kind in ("action", "sent the user"): + ref = next(line for line in block.splitlines() if line.startswith(f"- {kind}")).rsplit("(", 1)[1].rstrip(")") + path, n = ref.split(":") + line = _lines(files, path)[int(n) - 1] + assert line["event"] == ("tool_call" if kind == "action" else "message_to_user") + actions = [line for line in block.splitlines() if line.startswith("- action")] + assert len(actions) == 1 and actions[0].startswith("- action: gcal_create_event ") + assert "- stayed quiet (wakes/wake_004.jsonl:" in index + assert "- nudged after closing text; then message_user" in block + + +def test_without_a_change_log_actions_are_guessed_from_tool_names(run): + timeline, turns = run + timeline = [e for e in timeline if e["event"] != "app_changes"] + index = stage_run(timeline, turns).files[INDEX_FILE] + + assert "this run's apps kept no change log, so this is a guess from the name" in index + block = index[index.index("## Wake 2 "):index.index("## Wake 3 ")] + actions = [line.split(" ")[2] for line in block.splitlines() if line.startswith("- action")] + assert actions == ["gsheets_values_update", "gcal_patch_event", "gcal_create_event"] + + +def test_every_line_too_long_to_read_whole_keeps_a_preview_and_a_file(run): + timeline, turns = run + reply = "please book it " * 400 + for e in timeline: + if e["event"] == "user_reply_delivered": + e["text"] = reply + if e["event"] == "message_to_user" and e["wake"] == 1: + e["text"] = "here's the plan: " * 300 + staged = stage_run(timeline, turns) + + assert all(len(line) <= LINE_CHARS for path, text in staged.files.items() if path.endswith(".jsonl") + for line in text.splitlines()) + head = _lines(staged.files, "wakes/wake_003.jsonl")[1] + assert head["event"] == "user_message" and "text" not in head + assert staged.files[head["text_file"]] == reply and head["text_file"].startswith("wakes/results/wake_003_line_002_") + assert head["text_preview"].startswith("please book it please book it") + assert "- the user (Fri Nov 6, 9:00 PM UTC): \"please book it please book it" in staged.files[INDEX_FILE] + + +def test_events_jsonl_holds_every_wake_s_lines_and_the_run_s_start_and_end(run): + timeline, turns = run + timeline.insert(-1, {"event": "settle_timeout", "wake": 4, "still_at_work": [{"id": "porchbite-update"}]}) + staged = stage_run(timeline, turns) + events = _lines(staged.files, EVENTS_FILE) + + assert events[0]["event"] == "start" and events[-1]["event"] == "end" + per_wake = sum(len(staged.files[f"wakes/wake_00{n}.jsonl"].splitlines()) for n in range(1, 5)) + assert len(events) == per_wake + 2 + assert not any("real_time" in e or "input" in e and e["event"] == "user_sim_read" for e in events) + assert {"event": "settle_timeout", "wake": 4, "still_at_work": [{"id": "porchbite-update"}]} in events + + +def test_triggers_jsonl_has_every_firing_and_the_wake_it_came_after(run): + def firing(trigger: str, at: str, ok: bool = True) -> Firing: + t = datetime.fromisoformat(at.replace("Z", "+00:00")) + return Firing(trigger_id=trigger, source="env:world", turn=1, at=t, virtual_time=t, ok=ok, provoked_by=None) + + firings = [firing("tess-text", "2026-11-06T20:30:00Z"), firing("porchbite-update", "2026-11-06T21:30:00Z", False)] + staged = stage_run(*run, firings=firings, unavailable={"user_agent": "no log"}) + + lines = _lines(staged.files, TRIGGERS_FILE) + assert [(line.get("trigger_id"), line.get("after_wake"), line.get("fired")) for line in lines[:2]] == [ + ("tess-text", 1, True), ("porchbite-update", 4, False)] + assert lines[2] == {"unavailable": "user_agent", "why": "no log"} + assert "- porchbite-update (env:world): fired 0, failed 1" in staged.files[INDEX_FILE] + + +def test_120_wakes_sort_in_wake_order_and_an_unknown_format_is_flagged(): + timeline = [{"event": "start", "virtual_time": "2026-11-06T20:00:00Z"}] + turns = {} + for n in range(1, 121): + timeline += [{"event": "wake", "n": n, "virtual_time": f"2026-11-{6 + n // 24:02d}T{n % 24:02d}:00:00Z", + "woke_by": ["notification:Gmail"], "parts_text": f"alert {n}"}, + {"event": "agent_turn_end", "n": n, "state": "completed", "trajectory_url": f"mem://{n}"}] + turns[f"mem://{n}"] = turn_events(json.dumps([{"type": "tool_use", "tool_name": f"tool_{n}", "tool_id": "t", + "parameters": {}}, + {"type": "tool_result", "tool_id": "t", "output": "ok"}])) + turns["mem://7"] = turn_events(json.dumps([{"kind": "step"}])) + + staged = stage_run(timeline, turns) + + wake_files = sorted(p for p in staged.files if p.startswith("wakes/wake_")) + assert len(wake_files) == 120 and wake_files[:2] == ["wakes/wake_001.jsonl", "wakes/wake_002.jsonl"] + assert wake_files[-1] == "wakes/wake_120.jsonl" + assert [int(p[11:14]) for p in wake_files] == list(range(1, 121)) + assert _lines(staged.files, "wakes/wake_099.jsonl")[2]["tool"] == "tool_99" + assert [e["wake"] for e in _lines(staged.files, EVENTS_FILE)[1:] if e["event"] == "woke"] == list(range(1, 121)) + assert staged.record["not_compacted"] == [{"wake": 7, "turn": "turn", "format": "unknown", + "raw_bytes": len(json.dumps([{"kind": "step"}])), + "staged_bytes": staged.record["not_compacted"][0]["staged_bytes"]}] + + +def _judge(*calls: tuple[str, dict, str]) -> list[dict]: + """A judge's Claude Code trajectory: each call's tool, input and result text.""" + blocks = [{"type": "tool_use", "id": f"t{i}", "name": tool, "input": tool_input} + for i, (tool, tool_input, _) in enumerate(calls)] + results = [{"type": "tool_result", "tool_use_id": f"t{i}", "content": text} for i, (_, _, text) in enumerate(calls)] + return [{"type": "assistant", "message": {"content": blocks}}, {"type": "user", "message": {"content": results}}] + + +def test_judge_reads_names_the_wake_files_a_judge_opened_with_any_tool(): + judge = turn_events(json.dumps(_judge( + ("Read", {"file_path": "/tmp/r/index.md"}, "## Wake 4 · … wakes/wake_004.jsonl\n## Wake 5 · … wakes/wake_005.jsonl"), + ("Grep", {"pattern": "gmail_send", "path": "/tmp/r/events.jsonl"}, '{"event": "tool_call", "wake": 23}'), + ("Read", {"file_path": "/tmp/r/wakes/wake_012.jsonl", "offset": 40}, "…"), + ("Bash", {"command": "cat /tmp/r/wakes/results/wake_003_call_004_gmail_get_thread.json"}, "{}"), + ("Bash", {"command": "cd /tmp/r; for w in 080 075; do grep tool_call wakes/wake_$w.jsonl; done"}, "…"), + ("Bash", {"command": "cd /tmp/r; grep -n -i kidsy index.md"}, "61:- action: … (wakes/wake_009.jsonl:8)")))) + + assert judge_reads(judge) == {"tool_calls": 6, "wakes_opened": [3, 12, 23, 75, 80], "read_all_wakes": False, + "read_index": True, "searched_events": True, "read_triggers": False, + "helpers_started": 0} + + +def test_judge_reads_tells_one_pass_over_every_wake_file(): + judge = turn_events(json.dumps(_judge( + ("Bash", {"command": "python3 -c \"import glob\nfor f in sorted(glob.glob('wakes/wake_*.jsonl')): print(f)\""}, + '{"event": "woke", "wake": 1}'), + ("Agent", {"prompt": "check wake 7"}, "done")))) + + reads = judge_reads(judge) + assert reads["read_all_wakes"] is True and reads["wakes_opened"] == [1] and reads["helpers_started"] == 1 From 6b5828d2a7733fcd50a41996709f971ff1abc121 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Fri, 13 Nov 2026 18:35:00 -0800 Subject: [PATCH 2/3] fix(always_on_staging): split long side-file lines, label unlogged writes in the index, compare trigger times parsed, find wakes in text results; log a count, not every untrimmed trajectory Co-Authored-By: Claude Opus 5.5 --- .../verifiers/always_on_agent_verifier.py | 5 +- .../task_steps/verifiers/always_on_staging.py | 68 +++++++++++++++---- .../verifiers/test_always_on_staging.py | 33 ++++++++- 3 files changed, 89 insertions(+), 17 deletions(-) diff --git a/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py b/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py index 42f79d7d3..e71cf5682 100644 --- a/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py @@ -71,7 +71,7 @@ it), each message the user got ("message_to_user"), a nudge and its calls, and the user's reads of its messages. - {results}/: what is too long for a line (a tool's input or result, a long message): the line names the file \ ("_file", such as "result_file" or "text_file") and shows its start ("_preview"). Email bodies are \ -decoded (body.text). +decoded (body.text). A line longer than 1,800 characters is split across lines there, nothing dropped. - {events}: every wake's events in one file, so "did it ever do X" is one search. {triggers}: every trigger firing, \ fired or failed, with its pretend time and the wake it came after. @@ -249,7 +249,8 @@ def _log_staging(self, staged: StagedRun) -> None: r["staged_bytes"]) if r.get("not_compacted"): logger.error("Verifier '%s': compaction removed nothing from %d trajectories (unknown format, or nothing " - "to drop): %s", self.verifier_id, len(r["not_compacted"]), r["not_compacted"]) + "to drop), first in wake %s; the rest are in trajectory_staging.not_compacted", + self.verifier_id, len(r["not_compacted"]), r["not_compacted"][0]["wake"]) for key in ("missing_trajectories", "unreadable_trajectories"): if r.get(key): logger.warning("Verifier '%s': %s in wakes %s", self.verifier_id, key.replace("_", " "), r[key]) diff --git a/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py b/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py index 033dd7f29..92c5d0ea2 100644 --- a/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py +++ b/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py @@ -7,7 +7,7 @@ ``turn_end`` with its closing reply, ``message_to_user``, a ``nudge`` and its calls, the user's reads). A turn's trajectory is matched to its wake by the timeline's ``trajectory_url``; one that's missing is a line saying so. - ``wakes/results/``: what is too long for a line to be read whole (a tool's input or result, a long message), named - by wake and call (or line); the line keeps a preview. + by wake and call (or line); the line keeps a preview. Lines longer than LINE_CHARS are split there, nothing dropped. - ``events.jsonl``: every line of every wake, and the run's start and end. ``triggers.jsonl``: every trigger firing. - ``index.md``: a block per wake with its time, what woke it, the user's message, its write actions (tool and key arguments) and messages with the ``file:line`` they're at, and its errors. @@ -18,6 +18,7 @@ import json import re from collections import Counter +from datetime import datetime, timezone from dataclasses import dataclass, field from typing import Any, Mapping, Optional, Sequence @@ -242,8 +243,12 @@ def fit(self, line: dict, stem: str, movable: Sequence[str]) -> tuple[dict, int] while len(_dumps(line)) > LINE_CHARS and (inline := [k for k in movable if line.get(k) not in (None, "")]): key = max(inline, key=lambda k: len(_dumps(line[k]))) value = line.pop(key) - path = f"{stem}{'' if key == 'result' else '_' + key}.{'txt' if isinstance(value, str) else 'json'}" body = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=1, default=str) + wrapped = _wrapped(body) + # A wrapped file is .txt even when it held JSON: it no longer parses as one. + ext = "txt" if isinstance(value, str) or wrapped != body else "json" + path = f"{stem}{'' if key == 'result' else '_' + key}.{ext}" + body = wrapped self.files[path] = body written += len(body.encode()) line[f"{key}_file"] = path @@ -272,8 +277,12 @@ def finish(self, firings: Sequence[Firing], unavailable: Mapping[str, str]) -> S path = wake_file(n, self.width) self.files[path] = "".join(_dumps(line) + "\n" for line in lines) everything += lines - index_blocks.append(_index_block(w, path, lines, self.by_changes)) + index_blocks.append((w, path, lines)) everything += end + # A tool no app change names anywhere in the run may be an app that keeps no change log: its writes are + # guessed from their names, labelled as such, so they aren't left out of the index. + run_changing = set().union(*(_changing_tools(lines) for _, _, lines in index_blocks)) + index_blocks = [_index_block(w, path, lines, self.by_changes, run_changing) for w, path, lines in index_blocks] self.files[EVENTS_FILE] = "".join(_dumps(line) + "\n" for line in everything) self.files[TRIGGERS_FILE] = _triggers(firings, unavailable, self.wakes) self.files[INDEX_FILE] = _index(self, index_blocks, start, end, firings) @@ -300,9 +309,11 @@ def _changing_tools(lines: list[dict]) -> set[str]: for a in line.get("actors") or [] if a.get("tool")} -def _index_block(w: _Wake, path: str, lines: list[dict], by_changes: bool) -> str: +def _index_block(w: _Wake, path: str, lines: list[dict], by_changes: bool, run_changing: set[str]) -> str: """The wake's block. With ``by_changes`` its write actions are the calls of a tool that made one of the wake's app - changes; else (a run whose apps keep no change log) calls whose tool name has no read word (``is_write_action``).""" + changes, and calls of a tool no app change in the run names (``run_changing``) whose name has no read word are + "possible actions"; else (a run whose apps keep no change log) every call whose name has no read word is an action + (``is_write_action``).""" calls = sum(1 for line in lines if line["event"] == "tool_call") changing = _changing_tools(lines) out = [f"## Wake {w.n} · {label_time(w.at)} · woken by {woke_by_text(w.woke_by)} · {path} " @@ -318,11 +329,13 @@ def _index_block(w: _Wake, path: str, lines: list[dict], by_changes: bool) -> st changes = Counter() for i, line in enumerate(lines, 1): event = line["event"] - writes = _short_tool(line.get("tool") or "") in changing if by_changes else is_write_action(line.get("tool") or "") - if event == "tool_call" and writes: - action = _cut(f"{_short_tool(line['tool'])} {_key_args(line.get('input', line.get('input_preview')))}", - ACTION_CHARS) - out.append(f"- {'FAILED ' if line.get('error') else ''}action: {action} ({path}:{i})") + tool = line.get("tool") or "" + writes = _short_tool(tool) in changing if by_changes else is_write_action(tool) + guessed = by_changes and not writes and _short_tool(tool) not in run_changing and is_write_action(tool) + if event == "tool_call" and (writes or guessed): + action = _cut(f"{_short_tool(tool)} {_key_args(line.get('input', line.get('input_preview')))}", ACTION_CHARS) + kind = "possible action (no app change names this tool)" if guessed else "action" + out.append(f"- {'FAILED ' if line.get('error') else ''}{kind}: {action} ({path}:{i})") elif event == "tool_call" and line.get("error"): out.append(f"- error: {_short_tool(line['tool'])} failed ({path}:{i})") elif event == "message_to_user": @@ -370,7 +383,9 @@ def _index(stager: _Stager, blocks: list[str], start: list[dict], end: list[dict f"fields and names the file in {RESULTS_DIR}/ that holds them (\"_file\").", "", "Actions below are the assistant's tool calls of a tool that changed the apps' data in that wake (from the " - "apps' change logs)." if stager.by_changes else + "apps' change logs). A \"possible action\" is a call of a tool no app change in the run names whose name " + "has no read word (get, list, search, …): an app that keeps no change log, or a write that changed " + "nothing." if stager.by_changes else "Actions below are the assistant's tool calls whose tool name has no read word (get, list, search, …): " "this run's apps kept no change log, so this is a guess from the name."] gaps = sorted(set(stager.missing) | set(stager.unreadable)) @@ -393,17 +408,39 @@ def _index(stager: _Stager, blocks: list[str], start: list[dict], end: list[dict def _triggers(firings: Sequence[Firing], unavailable: Mapping[str, str], wakes: Mapping[int, _Wake]) -> str: """One line per firing, oldest first, with the last wake at or before its pretend time (``after_wake``, 0 before the first); then one per firing log that couldn't be read.""" - starts = sorted((w.at, n) for n, w in wakes.items() if w.at) + starts = sorted((t, n) for n, w in wakes.items() if (t := _instant(w.at))) out = [] for f in sorted(firings, key=lambda f: f.at): vt = f.virtual_time.isoformat().replace("+00:00", "Z") if f.virtual_time else None - after = max((n for at, n in starts if vt and at <= vt), default=0) + when = _instant(f.virtual_time) + after = max((n for t, n in starts if when and t <= when), default=0) out.append(_dumps({"trigger_id": f.trigger_id, "source": f.source, "fired": f.ok, "at": vt, "recorded_at": f.at.isoformat(), "after_wake": after, "provoked_by": f.provoked_by})) out += [_dumps({"unavailable": source, "why": why}) for source, why in unavailable.items() if source != "conversation"] return "".join(line + "\n" for line in out) +def _instant(value: Any) -> Optional[datetime]: + """A time (a datetime or an ISO string, "Z" or an offset; naive is UTC) as an aware datetime, for comparing.""" + if isinstance(value, str): + try: + value = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if not isinstance(value, datetime): + return None + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + + +def _wrapped(text: str) -> str: + """``text`` with every line longer than LINE_CHARS split into lines of LINE_CHARS, nothing dropped, so a reader + that cuts long lines shows it all.""" + if all(len(line) <= LINE_CHARS for line in text.split("\n")): + return text + return "\n".join(line[i:i + LINE_CHARS] if line else "" + for line in text.split("\n") for i in range(0, max(len(line), 1), LINE_CHARS)) + + _WAKE_NAME = re.compile(r"wake_(\d+)") # A shell loop over wake numbers (``for w in 080 075; do … wake_$w.jsonl``): the numbers it names. _WAKE_LOOP = re.compile(r"\bfor\s+(\w+)\s+in\s+([\d\s]+?)\s*;") @@ -430,7 +467,10 @@ def judge_reads(judge: TurnEvents) -> dict: wakes |= {int(n) for n in numbers.split()} every = every or bool(_ALL_WAKES.search(named)) if INDEX_FILE not in named: - wakes |= {int(n) for n in _WAKE_FIELD.findall(_dumps(call.get("result")))} + # A text result (Grep's lines, several events.jsonl lines) is searched as it is: serializing it would + # escape its quotes. + result = call.get("result") + wakes |= {int(n) for n in _WAKE_FIELD.findall(result if isinstance(result, str) else _dumps(result))} text = " ".join(_dumps(c.get("input")) for c in calls) return {"tool_calls": len(calls), "wakes_opened": sorted(wakes), "read_all_wakes": every, "read_index": INDEX_FILE in text, "searched_events": EVENTS_FILE in text, diff --git a/tst/unit/task_step/verifiers/test_always_on_staging.py b/tst/unit/task_step/verifiers/test_always_on_staging.py index 36f0ee23f..9429249dc 100644 --- a/tst/unit/task_step/verifiers/test_always_on_staging.py +++ b/tst/unit/task_step/verifiers/test_always_on_staging.py @@ -158,6 +158,17 @@ def test_the_index_points_at_the_line_each_action_and_message_is_on(run): assert "- nudged after closing text; then message_user" in block +def test_a_write_tool_no_app_change_names_is_a_labelled_possible_action(run): + timeline, turns = run + _changed_by(timeline, 2, "gcal_create_event") + index = stage_run(timeline, turns).files[INDEX_FILE] + + block = index[index.index("## Wake 2 "):index.index("## Wake 3 ")] + possible = [line.split("): ", 1)[1].split(" ")[0] for line in block.splitlines() if line.startswith("- possible action")] + assert possible == ["gsheets_values_update", "gcal_patch_event"] + assert "A \"possible action\" is a call of a tool no app change in the run names" in index + + def test_without_a_change_log_actions_are_guessed_from_tool_names(run): timeline, turns = run timeline = [e for e in timeline if e["event"] != "app_changes"] @@ -183,7 +194,9 @@ def test_every_line_too_long_to_read_whole_keeps_a_preview_and_a_file(run): for line in text.splitlines()) head = _lines(staged.files, "wakes/wake_003.jsonl")[1] assert head["event"] == "user_message" and "text" not in head - assert staged.files[head["text_file"]] == reply and head["text_file"].startswith("wakes/results/wake_003_line_002_") + side = staged.files[head["text_file"]] + assert side.replace("\n", "") == reply and all(len(line) <= LINE_CHARS for line in side.splitlines()) + assert head["text_file"].startswith("wakes/results/wake_003_line_002_") assert head["text_preview"].startswith("please book it please book it") assert "- the user (Fri Nov 6, 9:00 PM UTC): \"please book it please book it" in staged.files[INDEX_FILE] @@ -249,6 +262,15 @@ def _judge(*calls: tuple[str, dict, str]) -> list[dict]: return [{"type": "assistant", "message": {"content": blocks}}, {"type": "user", "message": {"content": results}}] +def test_a_firing_a_fraction_of_a_second_after_a_wake_comes_after_it(run): + t = datetime.fromisoformat("2026-11-06T20:00:00.500000+00:00") + firing = Firing(trigger_id="tess-text", source="env:world", turn=1, at=t, virtual_time=t, ok=True, provoked_by=None) + + lines = _lines(stage_run(*run, firings=[firing]).files, TRIGGERS_FILE) + + assert lines[0]["after_wake"] == 1 + + def test_judge_reads_names_the_wake_files_a_judge_opened_with_any_tool(): judge = turn_events(json.dumps(_judge( ("Read", {"file_path": "/tmp/r/index.md"}, "## Wake 4 · … wakes/wake_004.jsonl\n## Wake 5 · … wakes/wake_005.jsonl"), @@ -263,6 +285,15 @@ def test_judge_reads_names_the_wake_files_a_judge_opened_with_any_tool(): "helpers_started": 0} +def test_judge_reads_finds_the_wakes_in_a_search_that_returned_several_lines(): + judge = turn_events(json.dumps(_judge( + ("Grep", {"pattern": "porchbite_place_order", "path": "/tmp/r/events.jsonl", "-n": True}, + '12:{"event": "tool_call", "wake": 7, "tool": "porchbite_place_order"}\n' + '40:{"event": "tool_call", "wake": 19, "tool": "porchbite_place_order"}')))) + + assert judge_reads(judge)["wakes_opened"] == [7, 19] + + def test_judge_reads_tells_one_pass_over_every_wake_file(): judge = turn_events(json.dumps(_judge( ("Bash", {"command": "python3 -c \"import glob\nfor f in sorted(glob.glob('wakes/wake_*.jsonl')): print(f)\""}, From 86408d71111dc59f757eb56f518a45b8b680963e Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Fri, 13 Nov 2026 18:51:31 -0800 Subject: [PATCH 3/3] feat(always_on_staging): the agent's thinking is an assistant_thinking line in its wake when the trajectory recorded it Co-Authored-By: Claude Opus 5.5 --- .../verifiers/always_on_agent_verifier.py | 3 ++- .../task_steps/verifiers/always_on_staging.py | 6 ++++-- .../task_step/verifiers/test_always_on_staging.py | 14 ++++++++++++++ 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py b/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py index e71cf5682..9163cd959 100644 --- a/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/always_on_agent_verifier.py @@ -66,7 +66,8 @@ `file:line` it is at, and errors. - {first} … {last}: one file per wake. First what woke the assistant ("woke", the user's own messages in full as \ "user_message", the whole text it was sent as "wake_input"); then every tool call of its turn with input and result \ -("tool_call", numbered "call") and its own text ("assistant_text", which nobody saw); then the rest of the wake in time \ +("tool_call", numbered "call"), its own text ("assistant_text", which nobody saw) and its thinking when it was \ +recorded ("assistant_thinking": why it acted; what it did is in the tool calls and app changes); then the rest of the wake in time \ order: every app change ("app_change", "by": "agent" for its own), its closing reply ("turn_end"; the user never saw \ it), each message the user got ("message_to_user"), a nudge and its calls, and the user's reads of its messages. - {results}/: what is too long for a line (a tool's input or result, a long message): the line names the file \ diff --git a/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py b/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py index 92c5d0ea2..e1e2e21b9 100644 --- a/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py +++ b/src/agent_env/task_step/task_steps/verifiers/always_on_staging.py @@ -2,8 +2,8 @@ stored timeline and each wake's trajectory: one file per wake, one event per line, so no line is a whole turn. - ``wakes/wake_NNN.jsonl``: what woke the assistant (``woke``, the user's messages in full ``user_message``, the full - text it was sent ``wake_input``), then its turn's tool calls with their inputs and results (``tool_call``) and its - own text (``assistant_text``), then the rest of the wake in time order (``app_change`` for every change, + text it was sent ``wake_input``), then its turn's tool calls with their inputs and results (``tool_call``), its + own text (``assistant_text``) and its thinking when the trajectory recorded it (``assistant_thinking``), then the rest of the wake in time order (``app_change`` for every change, ``turn_end`` with its closing reply, ``message_to_user``, a ``nudge`` and its calls, the user's reads). A turn's trajectory is matched to its wake by the timeline's ``trajectory_url``; one that's missing is a line saying so. - ``wakes/results/``: what is too long for a line to be read whole (a tool's input or result, a long message), named @@ -211,6 +211,8 @@ def turn(self, w: _Wake, url: Optional[str], which: str) -> list[dict]: side += written elif kind == "text": lines.append(_line("assistant_text", w.n, turn=which, text=e["text"], subagent=e.get("subagent"))) + elif kind == "thinking": + lines.append(_line("assistant_thinking", w.n, turn=which, text=e["text"], subagent=e.get("subagent"))) elif kind == "note": lines.append(_line("harness_note", w.n, turn=which, text=e["text"])) elif kind == "raw": diff --git a/tst/unit/task_step/verifiers/test_always_on_staging.py b/tst/unit/task_step/verifiers/test_always_on_staging.py index 9429249dc..8bf7f4056 100644 --- a/tst/unit/task_step/verifiers/test_always_on_staging.py +++ b/tst/unit/task_step/verifiers/test_always_on_staging.py @@ -107,6 +107,20 @@ def test_what_woke_the_assistant_comes_first_with_the_user_s_whole_message(run): assert "the user (Fri Nov 6, 9:00 PM UTC): \"omg yes good catch" in staged.files[INDEX_FILE] +def test_the_agent_s_thinking_is_a_line_in_its_wake(run): + timeline, turns = run + wake2 = _url(timeline, 2) + turns[wake2] = turn_events(json.dumps([{"type": "assistant", "message": {"content": [ + {"type": "thinking", "thinking": "Nina is allergic; ask before ordering.", "signature": "s"}, + {"type": "text", "text": "Asking Nina."}]}}])) + + lines = _lines(stage_run(timeline, turns).files, "wakes/wake_002.jsonl") + + kinds = [line["event"] for line in lines if line["event"] in ("assistant_thinking", "assistant_text")] + assert kinds == ["assistant_thinking", "assistant_text"] + assert next(line for line in lines if line["event"] == "assistant_thinking")["text"].startswith("Nina is allergic") + + def test_every_app_change_is_kept_with_its_whole_summary(run): staged = stage_run(*run)