From 1ed48117d2a8b7084865b7766cec429edededdbe Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Mon, 19 Oct 2026 16:37:51 -0700 Subject: [PATCH 1/8] feat(prompt_agent): run a multi-turn conversation from one env event to the next on simulated time prompt_agent's opt-in between_turns: next_env_event makes a multi-turn run event-driven, for long-running and background-agent evals whose events are hours or days apart. After each agent turn it reads the envs' trigger state, brings the armed env clocks up to the earliest pending time-trigger mark, waits for the triggers due there to fire, fail or be cancelled (with a timeout), and wakes the agent with a neutral message or the user simulator's agent-trigger decide step. It stops when no time trigger is pending, at end_time, or at max_conversation_turns. The user side is never asked. The gateway clock gains PUT /clock/advance (clock/v1 "advance"): a forward jump on the current arm that keeps t0, rate and generation, so relative and recurring time-trigger marks keep their meaning and delays count the jumped time. set_time is a re-arm that re-anchors them; against a gateway without advance, prompt_agent re-arms at the env's own rate and refuses a run whose pending time triggers are relative or recurring. With between_turns set, or record_env_clock, every conversation message prompt_agent records carries env_clock, each env's virtual time when it was sent or received, so a grader can place the agent's messages on the same axis as the trigger firing log and the gateway trajectory. Both options are off by default; unset, prompt_agent, the conversation store and the clock behave as before. Co-Authored-By: Claude Opus 5.5 --- src/agent_env/a2a_agent/conversation_store.py | 14 +- src/agent_env/env/gateway/clock.py | 27 +- src/agent_env/env/gateway/constants.py | 3 +- src/agent_env/env/gateway/gateway.py | 12 + .../task_step/task_steps/prompt_agent.py | 188 ++++++++++- .../test_prompt_agent_between_turns_local.py | 84 +++++ tst/unit/env/gateway/clock_advance_test.py | 164 ++++++++++ .../test_prompt_agent_between_turns.py | 305 ++++++++++++++++++ 8 files changed, 788 insertions(+), 9 deletions(-) create mode 100644 tst/integration/task_step/test_prompt_agent_between_turns_local.py create mode 100644 tst/unit/env/gateway/clock_advance_test.py create mode 100644 tst/unit/task_step/test_prompt_agent_between_turns.py diff --git a/src/agent_env/a2a_agent/conversation_store.py b/src/agent_env/a2a_agent/conversation_store.py index d9ad7afb..7f4781df 100644 --- a/src/agent_env/a2a_agent/conversation_store.py +++ b/src/agent_env/a2a_agent/conversation_store.py @@ -109,11 +109,20 @@ def find_by_a2a_task_id(a2a_task_id: str) -> Optional[dict]: ) +def _message(role: str, parts: list[dict], now: datetime, env_clock: Optional[dict]) -> dict: + """A conversation message; ``env_clock`` maps each env to its virtual time when the message was recorded.""" + message = {"role": role, "parts": parts, "ts": now.isoformat()} + if env_clock is not None: + message["env_clock"] = env_clock + return message + + def add_a2a_task( conversation_id: str, parts: list[dict], a2a_task_id: str, role: str, + env_clock: Optional[dict] = None, ) -> Optional[dict]: """Append a caller-side message and create a new working a2a_task. @@ -133,7 +142,7 @@ def add_a2a_task( doesn't exist. """ now = datetime.now(timezone.utc) - new_message = {"role": role, "parts": parts, "ts": now.isoformat()} + new_message = _message(role, parts, now, env_clock) def _mutate(doc: dict) -> UpdateSpec: idx = len(doc.get("messages", [])) @@ -161,6 +170,7 @@ def complete_a2a_task( conversation_id: str, parts: list[dict], role: str, + env_clock: Optional[dict] = None, ) -> Optional[dict]: """Append a responder-side message and complete the oldest working a2a_task. @@ -185,7 +195,7 @@ def complete_a2a_task( doesn't exist (or the target task lost the state race). """ now = datetime.now(timezone.utc) - new_message = {"role": role, "parts": parts, "ts": now.isoformat()} + new_message = _message(role, parts, now, env_clock) # One backend for the whole retry loop: a reset between the read and the CAS would # decide against one database and write to another. docs = _doc_store() diff --git a/src/agent_env/env/gateway/clock.py b/src/agent_env/env/gateway/clock.py index 44913026..2b2118ad 100644 --- a/src/agent_env/env/gateway/clock.py +++ b/src/agent_env/env/gateway/clock.py @@ -123,6 +123,7 @@ def __init__(self, now_fn=time.monotonic): self._rate = 1.0 self._generation = 0 self._lived = 0.0 # virtual seconds run before the current arm/clear + self._ahead = 0.0 # virtual seconds the current arm was advanced by @property def armed(self) -> bool: @@ -153,10 +154,27 @@ def set_time(self, virtual_time: str, rate=1.0) -> dict: self._bank(t0) self._t0 = t0 self._rate = r + self._ahead = 0.0 self._armed = True self._generation += 1 return self.state() + def advance(self, virtual_time: str) -> dict: + """Jump forward to virtual_time on the current arm: t0, rate and generation stay, so time triggers keep their marks.""" + target = _parse_rfc3339(virtual_time) + with self._lock: + if not self._armed or self._t0 is None: + raise ClockError("clock is not armed") + if target > _MAX_T0 - timedelta(seconds=_OVERFLOW_HORIZON_S * self._rate): + raise ClockError(f"virtual_time {virtual_time!r} at rate {self._rate:g} would overflow datetime near the max") + current = _advance(self._t0, (self._now() - self._anchor) * self._rate + self._ahead) + jump = (target - current).total_seconds() + if jump < 0: + raise ClockError(f"advance only moves forward: {virtual_time!r} is before the current {_iso(current)}") + self._ahead += jump + self._lived += jump + return self.state() + def clear(self) -> dict: """Disarm (idempotent). Afterwards reads 404.""" with self._lock: @@ -164,12 +182,13 @@ def clear(self) -> dict: self._armed = False self._t0 = None self._rate = 1.0 + self._ahead = 0.0 self._generation += 1 return {"ok": True, "armed": False} def elapsed(self) -> float: """Virtual seconds run so far: real time at each arm's rate (1 while unarmed), plus the jump of each re-arm - forward; one back counts nothing.""" + or advance forward; one back counts nothing.""" with self._lock: return self._lived + (self._now() - self._anchor) * self._rate @@ -178,7 +197,7 @@ def now(self) -> datetime | None: if not self._armed or self._t0 is None: return None elapsed = self._now() - self._anchor - return _advance(self._t0, elapsed * self._rate) + return _advance(self._t0, elapsed * self._rate + self._ahead) def read(self) -> dict: """The frozen consumer contract: {"virtual_time": ""} — pure read, never advances.""" @@ -193,7 +212,7 @@ def state(self) -> dict: return {"armed": False} elapsed = self._now() - self._anchor return {"armed": True, "t0": _iso(self._t0), "virtual_seconds_per_real_second": self._rate, - "virtual_time": _iso(_advance(self._t0, elapsed * self._rate))} + "virtual_time": _iso(_advance(self._t0, elapsed * self._rate + self._ahead))} def _bank(self, t0: datetime | None = None) -> None: """Fold the stretch since the last arm/clear into `_lived` at its rate, and start the next one now.""" @@ -201,5 +220,5 @@ def _bank(self, t0: datetime | None = None) -> None: ran = (now - self._anchor) * self._rate self._lived += ran if t0 is not None and self._armed: # a re-arm forward is time the world lived through; one back adds nothing - self._lived += max(0.0, (t0 - _advance(self._t0, ran)).total_seconds()) + self._lived += max(0.0, (t0 - _advance(self._t0, ran + self._ahead)).total_seconds()) self._anchor = now diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py index 6354dcd1..6c70de34 100644 --- a/src/agent_env/env/gateway/constants.py +++ b/src/agent_env/env/gateway/constants.py @@ -190,10 +190,11 @@ def random_mcp_server_name() -> str: }, { "uri": EXT_CLOCK_URI, - "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set-time/clear are the harness control path.", + "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set-time/advance/clear are the harness control path.", "params": {"endpoint": "/clock/time", "methods": { "get_time": {"method": "GET", "endpoint": "/clock/time"}, "set_time": {"method": "PUT", "endpoint": "/clock/set-time", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "virtual_seconds_per_real_second": {"type": "number"}}, "required": ["virtual_time"]}}, + "advance": {"method": "PUT", "endpoint": "/clock/advance", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}}, "required": ["virtual_time"]}}, "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {}}}, "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}}}}, "required": False, diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index 43c4410a..d3c8b19a 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -269,6 +269,18 @@ async def clock_set_time(request: Request) -> Response: self._register_get_time_tool() return self._json_response(state, 200) + @self._mcp.custom_route("/clock/advance", methods=["PUT"]) + async def clock_advance(request: Request) -> Response: + try: + body = await request.json() + except Exception: + return self._json_response({"ok": False, "error": "Invalid JSON body"}, 400) + try: + state = self._clock.advance(body.get("virtual_time")) + except ClockError as e: + return self._json_response({"ok": False, "error": str(e)}, 400) + return self._json_response(state, 200) + @self._mcp.custom_route("/clock/clear", methods=["POST"]) async def clock_clear(request: Request) -> Response: self._unregister_get_time_tool() diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 9d732679..cd06db75 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -5,6 +5,7 @@ import copy import json import logging +import math import time import uuid from contextlib import nullcontext @@ -115,6 +116,66 @@ def _file_uris(parts: list[dict]) -> list[str]: } +NEXT_ENV_EVENT = "next_env_event" +DEFAULT_WAKE_MESSAGE = "[{virtual_time}] Time has passed. Check for anything new and handle it." +_BETWEEN_TURNS_KEYS = ("mode", "wake", "wake_message", "end_time", "fire_timeout_seconds") +_WAKES = ("message", "decide") +_SETTLED_STATUSES = ("fired", "failed", "cancelled") +_FIRE_POLL_SECONDS = 0.5 + + +def _utc(text: str) -> datetime: + """An RFC 3339 instant as an aware UTC datetime; ValueError without an offset.""" + parsed = datetime.fromisoformat(text.replace("Z", "+00:00").replace("z", "+00:00")) + if parsed.tzinfo is None: + raise ValueError(f"{text!r} has no UTC offset") + return parsed.astimezone(timezone.utc) + + +def _iso(value: datetime) -> str: + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _between_turns_spec(raw: Any) -> Optional[dict]: + """``between_turns`` with its defaults filled in, refusing a shape the turn loop can't follow; None when unset.""" + if raw is None: + return None + spec = {"mode": raw} if isinstance(raw, str) else raw + if not isinstance(spec, dict): + raise ValueError(f"between_turns must be {NEXT_ENV_EVENT!r} or an object with a mode") + unknown = sorted(set(spec) - set(_BETWEEN_TURNS_KEYS)) + if unknown: + raise ValueError(f"between_turns has unknown key(s) {unknown}; expected {list(_BETWEEN_TURNS_KEYS)}") + if spec.get("mode") != NEXT_ENV_EVENT: + raise ValueError(f"between_turns.mode must be {NEXT_ENV_EVENT!r}, got {spec.get('mode')!r}") + out = {"mode": NEXT_ENV_EVENT, "wake": spec.get("wake", "message"), + "wake_message": spec.get("wake_message", DEFAULT_WAKE_MESSAGE), "end_time": spec.get("end_time"), + "fire_timeout_seconds": spec.get("fire_timeout_seconds", 60)} + if out["wake"] not in _WAKES: + raise ValueError(f"between_turns.wake must be one of {list(_WAKES)}, got {out['wake']!r}") + if not isinstance(out["wake_message"], str) or not out["wake_message"].strip(): + raise ValueError("between_turns.wake_message must be a non-empty string") + try: + out["wake_message"].format(virtual_time="", trigger_ids="") + except (KeyError, IndexError, ValueError) as e: + raise ValueError(f"between_turns.wake_message may use only {{virtual_time}} and {{trigger_ids}}: {e!r}") from e + if out["end_time"] is not None: + try: + _utc(out["end_time"]) + except (TypeError, AttributeError, ValueError) as e: + raise ValueError(f"between_turns.end_time must be an RFC 3339 instant: {e}") from e + timeout = out["fire_timeout_seconds"] + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or not math.isfinite(timeout) or timeout <= 0: + raise ValueError(f"between_turns.fire_timeout_seconds must be a positive number, got {timeout!r}") + return out + + +def _rearm_safe(when: dict) -> bool: + """Whether a time trigger keeps its mark across a clock re-arm: event-anchored, or one absolute `at`.""" + at = when.get("at") + return "after" in when or ("every" not in when and isinstance(at, str) and not at.startswith("P")) + + class PromptAgentTaskStep(TaskStep): type: ClassVar[str] = "prompt_agent" entity_refs = (EntityRef.env("snapshot_config.env_id"),) @@ -158,6 +219,8 @@ def __init__( *, snapshot_config: Optional[SnapshotConfig] = None, live_trajectory: bool = False, + between_turns: Optional[str | dict] = None, + record_env_clock: bool = False, ): # `prompt` is the legacy text-only field; `parts` is the multimodal # form that's the authoritative A2A wire shape. Exactly one must be set: @@ -171,6 +234,9 @@ def __init__( self._validate_parts(parts) if max_conversation_turns < 1: raise ValueError("max_conversation_turns must be >= 1") + advance = _between_turns_spec(between_turns) + if advance is not None and max_conversation_turns < 2: + raise ValueError("between_turns needs max_conversation_turns >= 2: it acts between turns") if model_params: reserved = set(model_params) & MODEL_PARAMS_RESERVED if reserved: @@ -201,6 +267,9 @@ def __init__( self.user_model = user_model self.snapshot_config = snapshot_config self.live_trajectory = live_trajectory + self.between_turns = between_turns + self.record_env_clock = record_env_clock + self._advance = advance @staticmethod def _validate_parts(parts: list[dict]) -> None: @@ -276,6 +345,10 @@ def to_dict(self) -> dict: base["snapshot_config"] = self.snapshot_config.to_dict() if self.live_trajectory: base["live_trajectory"] = True + if self.between_turns is not None: + base["between_turns"] = self.between_turns + if self.record_env_clock: + base["record_env_clock"] = True return base @classmethod @@ -312,6 +385,8 @@ def from_dict(cls, data: dict) -> PromptAgentTaskStep: else None ), live_trajectory=data.get("live_trajectory", False), + between_turns=data.get("between_turns"), + record_env_clock=data.get("record_env_clock", False), ) def _trajectory_prefix(self) -> str: @@ -418,7 +493,7 @@ async def _execute_conversation( user_url = self.user_a2a_url or ( (user_sim.a2a_url or user_sim.api_url) if user_sim else None ) - if user_url is None and user_sim is None and self.max_conversation_turns > 1: + if user_url is None and user_sim is None and self.max_conversation_turns > 1 and self._advance is None: user_url = get_config().get_default_human_a2a_url() trajectory_output_prefix = self._trajectory_prefix() @@ -481,6 +556,12 @@ async def _execute_conversation( logger.warning( f"user agent '{self.user_agent_name}' advertises {A2AAgent.EXT_TRIGGERS} but " f"max_conversation_turns={self.max_conversation_turns} (<2); the trigger engine is never invoked") + advance = self._advance + if advance is not None and advance["wake"] == "decide" and not (trig_ext and triggers_registered): + raise RuntimeError(f"{self.id}: between_turns wake 'decide' needs user agent '{self.user_agent_name}' to be a " + f"user simulator with agent triggers registered") + clock_envs = ([d for d in context.deployed_envs if d.supports(EXT_CLOCK_URI, "state")] + if advance is not None or self.record_env_clock else None) solver_context_id = a2a_context_id if self.context_id: @@ -529,6 +610,7 @@ async def _execute_conversation( # agent and recorded on the conversation as `a2a_task_id`. # This id is also used as part of the key name for the trajectory S3 object. target_a2a_task_id = uuid.uuid4().hex + sent_clock = await self._env_clock_kwargs(clock_envs) # The turn records the objects' own URLs, and only once the parts are ready to send, so an # object the agent can't be sent leaves no turn waiting. @@ -539,6 +621,7 @@ def record_turn() -> None: parts=current_user_parts, a2a_task_id=target_a2a_task_id, role="user", + **sent_clock, ) # Turn 0 of a prompt-mode step re-sends exactly the text already # persisted as PromptResponse.prompt_text; store a None placeholder @@ -571,6 +654,7 @@ def record_turn() -> None: conversation_id=conversation_id, parts=agent_response_parts, role="agent", + **(await self._env_clock_kwargs(clock_envs)), ) if traj_ext_cached: @@ -604,7 +688,16 @@ def record_turn() -> None: if series is not None: await series.capture_turn(context, turn + 1) - if trig_ext and triggers_registered: + if advance is not None: + event = await self._advance_to_next_env_event(context, advance) + (context.metadata.setdefault("env_event_advances", {}) + .setdefault(self.id, []).append({"turn": turn + 1, **event})) + if "stopped" in event: + logger.info(f"Turn {turn+1}: no further env event ({event['stopped']}); ending multi-turn") + conversation_store.mark_closed(conversation_id) + break + + if trig_ext and triggers_registered and (advance is None or advance["wake"] == "decide"): env_triggers = await self._read_env_triggers(context) (context.metadata.setdefault("env_trigger_snapshots", {}) .setdefault(self.id, []).append({"turn": turn + 1, "envs": env_triggers})) @@ -623,6 +716,12 @@ def record_turn() -> None: continue # no trigger fired -> fall through to the LLM user-sim for this turn (hybrid) + if advance is not None: + text = advance["wake_message"].format(virtual_time=event["mark"], + trigger_ids=", ".join(event["trigger_ids"])) + current_user_parts = [{"kind": "text", "text": text}] + continue + user_a2a_task_id = uuid.uuid4().hex try: # A user-sim runs in a sandbox agent-env deployed; a human peer, registered or named by @@ -845,6 +944,91 @@ async def _decide(self, decide_url: str, turn: int, solver_message: str, context raise RuntimeError(f"agent-trigger /decide failed (HTTP {resp.status_code}): {resp.text}") return resp.json() + async def _env_clock_kwargs(self, envs: Optional[list]) -> dict: + """``{"env_clock": {env_id: virtual_time or None}}`` for a conversation message, or {} when not recording.""" + if envs is None: + return {} + states = await asyncio.gather(*(self._get_clock_state(d) for d in envs)) + return {"env_clock": {d.env_id: (s or {}).get("virtual_time") if (s or {}).get("armed") else None + for d, s in zip(envs, states)}} + + async def _advance_to_next_env_event(self, context: TaskStepContext, advance: dict) -> dict: + """Move armed env clocks to the earliest pending time-trigger mark and await its triggers; ``stopped`` once none is due.""" + envs = [] + for d in context.deployed_envs: + if not (d.supports(EXT_TRIGGERS_URI, "state") and d.supports(EXT_CLOCK_URI, "state")): + continue + clock = await self._get_clock_state(d) + if clock and clock.get("armed"): + envs.append((d, clock, (await self._get_trigger_state(d)).get("triggers", []))) + pending = [(_utc(t["next_mark"]), d, t) for d, _, rows in envs for t in rows + if t.get("type") == "time" and t.get("status") == "armed" and t.get("next_mark")] + if not envs: + return {"stopped": "no_armed_env_clock", "virtual_time": None} + end = _utc(advance["end_time"]) if advance["end_time"] else None + mark = min((m for m, _, _ in pending), default=None) + if mark is None or (end is not None and mark > end): + if end is not None: + await self._move_env_clocks(envs, end, check_marks=False) + return {"stopped": "end_time" if mark is not None else "no_pending_time_triggers", + "virtual_time": _iso(end) if end is not None else None} + due = [(d, t) for m, d, t in pending if m == mark] + moved = await self._move_env_clocks(envs, mark, check_marks=True) + statuses = await self._wait_fired(context, due, advance["fire_timeout_seconds"]) + return {"mark": _iso(mark), "trigger_ids": [t["id"] for _, t in due], "statuses": statuses, "moved": moved} + + async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: bool) -> dict: + """Bring each env clock behind ``target`` up to it, by advance where offered, else a re-arm only fixed marks survive.""" + moved = {} + for d, clock, rows in envs: + if _utc(clock["virtual_time"]) >= target: + continue + if d.supports(EXT_CLOCK_URI, "advance"): + await d.invoke(EXT_CLOCK_URI, "advance", {"virtual_time": _iso(target)}, timeout=15) + moved[d.env_id] = "advance" + continue + unsafe = [t["id"] for t in rows if check_marks and t.get("type") == "time" + and t.get("status") == "armed" and not _rearm_safe(t.get("when") or {})] + if unsafe: + raise RuntimeError(f"{self.id}: env '{d.env_id}' has no clock advance, and a re-arm would re-anchor the " + f"relative or recurring time trigger(s) {unsafe}; give them absolute marks") + await d.invoke(EXT_CLOCK_URI, "set_time", {"virtual_time": _iso(target), "virtual_seconds_per_real_second": + clock.get("virtual_seconds_per_real_second", 1.0)}, timeout=15) + moved[d.env_id] = "set_time" + return moved + + async def _wait_fired(self, context: TaskStepContext, due: list, timeout_seconds: float) -> dict: + """Each due trigger's outcome once it has fired, failed or been cancelled; "timeout" for one still unsettled.""" + before = {t["id"]: (t.get("fire_count") or 0, t.get("failure_count") or 0) for _, t in due} + by_env: dict[str, tuple[Any, list[str]]] = {} + for d, t in due: + by_env.setdefault(d.env_id, (d, []))[1].append(t["id"]) + outcome: dict[str, str] = {} + deadline = time.monotonic() + timeout_seconds + while True: + for d, ids in by_env.values(): + rows = {t["id"]: t for t in (await self._get_trigger_state(d)).get("triggers", [])} + for tid in (i for i in ids if i not in outcome): + row = rows.get(tid) + fires, failures = before[tid] + if row is None: + outcome[tid] = "removed" + elif row.get("status") in TRIGGER_IN_FLIGHT_STATUSES: + continue + elif (row.get("failure_count") or 0) > failures: + outcome[tid] = "failed" + elif (row.get("fire_count") or 0) > fires or row.get("status") in _SETTLED_STATUSES: + outcome[tid] = "fired" if row.get("status") == "armed" else row["status"] + if len(outcome) == len(due) or time.monotonic() >= deadline: + break + context.heartbeat() + await asyncio.sleep(_FIRE_POLL_SECONDS) + for _, t in due: + if t["id"] not in outcome: + logger.warning(f"{self.id}: time trigger '{t['id']}' had not settled {timeout_seconds}s after its mark") + outcome[t["id"]] = "timeout" + return outcome + _CAPTURE_BUDGET_SECONDS: ClassVar[float] = 90 async def _persist_env_trigger_state(self, context: TaskStepContext) -> None: diff --git a/tst/integration/task_step/test_prompt_agent_between_turns_local.py b/tst/integration/task_step/test_prompt_agent_between_turns_local.py new file mode 100644 index 00000000..f29ce54c --- /dev/null +++ b/tst/integration/task_step/test_prompt_agent_between_turns_local.py @@ -0,0 +1,84 @@ +"""An event-driven multi-day run on the local defaults: a gateway env whose frozen clock starts on a Monday, two time +triggers days apart that write into its items server, and the echo agent (tst/data/a2a_agent) prompted with +between_turns next_env_event. Between turns prompt_agent advances the gateway clock to each mark, the trigger writes +land, the agent is woken once per event with no user side, and every message carries the env clock, all in seconds of +wall time. Needs Docker and a throwaway local registry.""" + +import json +import time +import uuid +from datetime import datetime + +import httpx +import pytest + +from agent_env.a2a_agent import conversation_store +from agent_env.env.gateway import AGENT_ENV_ROLE_HEADER +from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.task_steps.prompt_agent import DEFAULT_WAKE_MESSAGE, PromptAgentTaskStep +from agent_env.task_step.task_steps.register_env_triggers import RegisterEnvTriggersStep +from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep +from tst.integration.store.local_parallel_deploy_test import _seed_bootstrap_envs +from tst.integration.task_step.test_prompt_agent_live_trajectory_local import _deploy, local_registry as local_registry +from tst.integration.task_step.test_register_env_triggers_local import _deploy_items +from tst.util.a2a_test_agent import put_test_agent + +pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] + +MONDAY = "2026-03-02T13:00:00Z" +TUESDAY = "2026-03-03T16:40:00Z" +THURSDAY = "2026-03-05T03:15:00Z" +FRIDAY = "2026-03-06T12:00:00Z" + + +def _vt(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +@pytest.mark.asyncio +async def test_each_event_days_apart_wakes_the_agent_once_in_one_run(local_registry): + _seed_bootstrap_envs() + env, deployed, _ = await _deploy_items("multiday") + agent = put_test_agent(f"multiday-agent-{uuid.uuid4().hex[:8]}") + triggers = [{"id": "school-email", "when": {"type": "time", "at": TUESDAY}, + "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "school email"}}]}, + {"id": "calendar-clash", "when": {"type": "time", "at": "P2DT14H15M"}, + "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "calendar clash"}}]}] + context = TaskStepContext(deployed_envs=[deployed], deployed_agents=[], metadata={}) + try: + await SyncEnvClockTaskStep(id="clock", version=None, env_id=env.id, virtual_time=MONDAY, + virtual_seconds_per_real_second=0).execute(context) + await RegisterEnvTriggersStep(id="events", version=None, env_id=env.id, triggers=triggers, + watch_roles=["default"]).execute(context) + await _deploy(agent, "solver").execute(context) + started = time.monotonic() + await PromptAgentTaskStep(id="watch", version=None, prompt="Watch the inbox.", agent_name="solver", + max_conversation_turns=6, poll_interval_seconds=1, + between_turns={"mode": "next_env_event", "end_time": FRIDAY}).execute(context) + wall = time.monotonic() - started + async with httpx.AsyncClient(base_url=deployed.gateway_url, timeout=60) as client: + state = (await client.get("/triggers/state")).raise_for_status().json() + clock = (await client.get("/clock/state")).raise_for_status().json() + stored = await client.post("/step", json={"action": "call_tool", "tool_name": "list_items", "arguments": {}}, + headers={AGENT_ENV_ROLE_HEADER: "harness"}) + finally: + await env.close() + + conversation = conversation_store.get_conversation(context.metadata["a2a_conversations"]["watch"]) + sent = [m["parts"][0]["text"] for m in conversation["messages"] if m["role"] == "user"] + assert sent == ["Watch the inbox.", DEFAULT_WAKE_MESSAGE.format(virtual_time=TUESDAY), + DEFAULT_WAKE_MESSAGE.format(virtual_time=THURSDAY)] + assert [m["env_clock"][env.id] for m in conversation["messages"]] == [MONDAY, MONDAY, TUESDAY, TUESDAY, + THURSDAY, THURSDAY] + assert {t["id"]: (t["status"], t["fire_count"]) for t in state["triggers"]} == { + "school-email": ("fired", 1), "calendar-clash": ("fired", 1)} + fired = {e["trigger_id"]: e["virtual_time"] for e in state["events"] if e["kind"] == "fired"} + assert {k: _vt(v) for k, v in fired.items()} == {"school-email": _vt(TUESDAY), "calendar-clash": _vt(THURSDAY)} + assert "reanchored" not in {e["kind"] for e in state["events"]} + assert json.loads(stored.json()["content"][0]["text"])["items"][-2:] == ["school email", "calendar clash"] + assert _vt(clock["virtual_time"]) == _vt(FRIDAY) and clock["t0"] == MONDAY + advances = context.metadata["env_event_advances"]["watch"] + assert [a.get("trigger_ids") for a in advances[:2]] == [["school-email"], ["calendar-clash"]] + assert all(a["moved"] == {env.id: "advance"} for a in advances[:2]) + assert advances[2] == {"turn": 3, "stopped": "no_pending_time_triggers", "virtual_time": FRIDAY} + assert wall < 90, f"four simulated days took {wall:.0f}s of wall time" diff --git a/tst/unit/env/gateway/clock_advance_test.py b/tst/unit/env/gateway/clock_advance_test.py new file mode 100644 index 00000000..df7e625b --- /dev/null +++ b/tst/unit/env/gateway/clock_advance_test.py @@ -0,0 +1,164 @@ +"""Clock.advance: a forward jump on the current arm, so time triggers keep their marks, and its PUT /clock/advance route.""" +from __future__ import annotations + +import asyncio + +import httpx +import pytest + +from agent_env.env.gateway import gateway as gw_mod +from agent_env.env.gateway import triggers as trig_mod +from agent_env.env.gateway.clock import Clock, ClockError +from agent_env.env.gateway.constants import EXT_CLOCK_URI, GATEWAY_EXTENSIONS +from agent_env.env.gateway.gateway import Gateway +from agent_env.env.gateway.triggers import TriggerEngine +from tst.unit.env.gateway.triggers_time_test import FakeNow, _FakeGateway, _perm, _settle, _tick_call + +T0 = "2026-01-01T00:00:00Z" + + +def _armed(rate: float = 1.0) -> tuple[Clock, FakeNow]: + now = FakeNow() + clock = Clock(now_fn=now) + clock.set_time(T0, rate=rate) + return clock, now + + +def test_advance_jumps_forward_and_keeps_t0_rate_and_generation(): + clock, now = _armed(rate=2.0) + gen = clock.generation + state = clock.advance("2026-01-03T00:00:00Z") + assert state["virtual_time"] == "2026-01-03T00:00:00Z" + assert (state["t0"], state["virtual_seconds_per_real_second"], clock.generation) == (T0, 2.0, gen) + now.tick(30) + assert clock.read()["virtual_time"] == "2026-01-03T00:01:00Z" + + +def test_advance_on_a_frozen_clock_holds_there(): + clock, now = _armed(rate=0) + clock.advance("2026-01-02T12:00:00Z") + now.tick(3600) + assert clock.read()["virtual_time"] == "2026-01-02T12:00:00Z" + + +def test_elapsed_counts_the_jump(): + clock, now = _armed() + now.tick(10) + clock.advance("2026-01-01T01:00:10Z") + assert clock.elapsed() == pytest.approx(3610) + + +def test_advance_refuses_backwards_and_unarmed(): + clock, now = _armed() + now.tick(60) + with pytest.raises(ClockError, match="only moves forward"): + clock.advance("2026-01-01T00:00:30Z") + with pytest.raises(ClockError, match="not armed"): + Clock(now_fn=FakeNow()).advance(T0) + with pytest.raises(ClockError, match="RFC3339"): + clock.advance("tomorrow") + + +def test_a_rearm_after_an_advance_starts_from_its_own_t0_and_banks_from_the_advanced_time(): + clock, now = _armed() + clock.advance("2026-01-02T00:00:00Z") + lived = clock.elapsed() + clock.set_time("2026-01-02T06:00:00Z", rate=1.0) + assert clock.read()["virtual_time"] == "2026-01-02T06:00:00Z" + assert clock.elapsed() == pytest.approx(lived + 6 * 3600) + clock.set_time("2026-01-01T00:00:00Z", rate=1.0) + assert clock.read()["virtual_time"] == T0 + + +def test_clear_drops_the_advance(): + clock, _ = _armed() + clock.advance("2026-01-05T00:00:00Z") + clock.clear() + clock.set_time(T0, rate=0) + assert clock.read()["virtual_time"] == T0 + + +@pytest.mark.asyncio +async def test_a_relative_mark_comes_due_after_an_advance_with_no_reanchor(): + now = FakeNow() + clock = Clock(now_fn=now) + engine = TriggerEngine(_FakeGateway(clock)) + clock.set_time(T0, rate=0) + engine.register({"watch_roles": ["default"], "triggers": [ + {"id": "rel", "when": {"type": "time", "at": "P2D"}, "actions": [_perm()]}]}) + _tick_call(engine, now, 1) + clock.advance("2026-01-03T00:00:00Z") + _tick_call(engine, now, 1) + await _settle(engine) + trig = engine.state()["triggers"][0] + assert (trig["status"], trig["fire_count"]) == ("fired", 1) + assert "reanchored" not in {e["kind"] for e in engine.state()["events"]} + + +@pytest.mark.asyncio +async def test_a_delay_counts_the_jumped_time(monkeypatch): + monkeypatch.setattr(trig_mod, "_VIRTUAL_RECHECK_S", 0.01) + now = FakeNow() + clock = Clock(now_fn=now) + engine = TriggerEngine(_FakeGateway(clock)) + clock.set_time(T0, rate=0) + engine.register({"watch_roles": ["default"], "triggers": [ + {"id": "slow", "when": {"type": "time", "at": "PT1M"}, "actions": [{**_perm(), "delay": "PT6H"}]}]}) + clock.advance("2026-01-01T00:01:00Z") + _tick_call(engine, now, 1) + await asyncio.sleep(0.05) + assert engine.state()["triggers"][0]["waiting"] is True + clock.advance("2026-01-01T06:01:00Z") + for _ in range(100): + if engine.state()["triggers"][0]["status"] == "fired": + break + await asyncio.sleep(0.05) + assert engine.state()["triggers"][0]["status"] == "fired" + + +def test_the_card_offers_advance(): + clock_ext = next(e for e in GATEWAY_EXTENSIONS if e["uri"] == EXT_CLOCK_URI) + assert clock_ext["params"]["methods"]["advance"] == { + "method": "PUT", "endpoint": "/clock/advance", + "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}}, "required": ["virtual_time"]}} + + +@pytest.fixture +def gateway(tmp_path, monkeypatch) -> Gateway: + monkeypatch.setattr(gw_mod, "GATEWAY_TRAJECTORY_FILE", str(tmp_path / "trajectory.jsonl")) + gw = Gateway(host="127.0.0.1", port=0, server_name="AgentEnvGateway", internal_mcp_servers=[]) + gw._clock = Clock(now_fn=FakeNow()) + yield gw + gw._close_trajectory_file() + + +def _put(gw: Gateway, path: str, body) -> httpx.Response: + async def _do() -> httpx.Response: + transport = httpx.ASGITransport(app=gw._mcp.streamable_http_app()) + async with httpx.AsyncClient(transport=transport, base_url="http://gateway") as client: + return await client.put(path, content=body if isinstance(body, str) else None, + json=None if isinstance(body, str) else body) + + return asyncio.run(_do()) + + +def test_route_advances_an_armed_clock(gateway): + gateway._clock.set_time(T0, rate=0) + resp = _put(gateway, "/clock/advance", {"virtual_time": "2026-01-04T08:00:00Z"}) + assert resp.status_code == 200 + assert resp.json()["virtual_time"] == "2026-01-04T08:00:00Z" + + +@pytest.mark.parametrize("body,fragment", [ + ({"virtual_time": "2026-01-04T08:00:00Z"}, "not armed"), + ("{not json", "Invalid JSON body"), +]) +def test_route_refuses_with_400(gateway, body, fragment): + resp = _put(gateway, "/clock/advance", body) + assert resp.status_code == 400 and fragment in resp.json()["error"] + + +def test_route_refuses_a_backwards_advance(gateway): + gateway._clock.set_time("2026-01-04T08:00:00Z", rate=0) + resp = _put(gateway, "/clock/advance", {"virtual_time": T0}) + assert resp.status_code == 400 and "only moves forward" in resp.json()["error"] diff --git a/tst/unit/task_step/test_prompt_agent_between_turns.py b/tst/unit/task_step/test_prompt_agent_between_turns.py new file mode 100644 index 00000000..47621670 --- /dev/null +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -0,0 +1,305 @@ +"""prompt_agent between_turns next_env_event: between turns the env clock jumps to the next pending time-trigger mark, +the run waits for those triggers, and the agent is woken; with the option unset nothing changes.""" +from __future__ import annotations + +import copy +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +import httpx +import pytest +from a2a.types import TaskState + +from agent_env.a2a_agent import A2AAgent +from agent_env.env.gateway.constants import EXT_CLOCK_URI, EXT_TRIGGERS_URI +from agent_env.task_step.context import DeployedAgent, TaskStepContext +from agent_env.task_step.task_steps import prompt_agent as pa_mod +from agent_env.task_step.task_steps.prompt_agent import DEFAULT_WAKE_MESSAGE, PromptAgentTaskStep + +START = "2026-03-02T13:00:00Z" +DAY1 = "2026-03-03T09:30:00Z" +DAY2 = "2026-03-04T21:15:00Z" + + +def _dt(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +def _z(value: datetime) -> str: + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +class _Env: + """A deployed env whose clock and time triggers behave as a gateway's: a trigger fires once the clock reaches its mark.""" + + def __init__(self, marks: dict, *, rate: float = 0.0, advance: bool = True, outcome: str = "fired", + env_id: str = "env-1"): + self.env_id, self.rate, self.offers_advance, self.outcome = env_id, rate, advance, outcome + self.now = _dt(START) + self.calls: list[tuple[str, dict | None]] = [] + self.rows = {tid: {"id": tid, "type": "time", "status": "armed", "next_mark": at if not at.startswith("P") else + _z(self.now + timedelta(days=1)), "fire_count": 0, "failure_count": 0, + "when": {"type": "time", "at": at}} for tid, at in marks.items()} + + def supports(self, uri: str, method: str) -> bool: + if uri == EXT_CLOCK_URI: + return method in ("state", "set_time") or (method == "advance" and self.offers_advance) + return uri == EXT_TRIGGERS_URI and method == "state" + + async def invoke(self, uri, method, params=None, *, timeout=30, environment_name=None): + self.calls.append((method, params)) + if uri == EXT_CLOCK_URI and method in ("advance", "set_time"): + self.now = _dt(params["virtual_time"]) + return {} + if uri == EXT_CLOCK_URI: + return {"armed": True, "virtual_time": _z(self.now), "virtual_seconds_per_real_second": self.rate} + for row in self.rows.values(): + if row["status"] == "armed" and row["next_mark"] and _dt(row["next_mark"]) <= self.now: + if self.outcome == "fired": + row.update(status="fired", fire_count=1, next_mark=None) + elif self.outcome == "failed": + row.update(status="failed", failure_count=1, next_mark=None) + return {"triggers": copy.deepcopy(list(self.rows.values()))} + + def moves(self) -> list[tuple[str, dict]]: + return [(m, p) for m, p in self.calls if m in ("advance", "set_time")] + + +def _wire(monkeypatch, *, decide=None): + """Mock the conversation's HTTP and store boundary; returns what the solver was sent and what the store recorded.""" + cfg = MagicMock() + cfg.get_model_params.return_value = {} + cfg.get_default_human_a2a_url.side_effect = AssertionError("no human peer in an event-driven run") + monkeypatch.setattr(pa_mod, "get_config", lambda: cfg) + seen = {"solver": [], "stored": [], "closed": 0} + monkeypatch.setattr(pa_mod.conversation_store, "create_conversation", lambda *a, **kw: None) + monkeypatch.setattr(pa_mod.conversation_store, "get_conversation", lambda *a, **kw: None) + monkeypatch.setattr(pa_mod.conversation_store, "add_a2a_task", lambda *a, **kw: seen["stored"].append(("add", kw))) + monkeypatch.setattr(pa_mod.conversation_store, "complete_a2a_task", + lambda *a, **kw: seen["stored"].append(("complete", kw))) + + def closed(*a, **kw): + seen["closed"] += 1 + + monkeypatch.setattr(pa_mod.conversation_store, "mark_closed", closed) + + async def send(url, parts, message_id, context_id, timeout): + if "user" in url: + raise AssertionError("the user side is never asked in an event-driven run") + seen["solver"].append(parts[0]["text"]) + return f"task-{message_id}", None + + async def poll(url, task_id, timeout, interval, **_): + return {"status": {"state": TaskState.completed, "message": {"parts": [{"kind": "text", "text": "ok"}]}}} + + monkeypatch.setattr(pa_mod.protocol, "send_a2a_message", send) + monkeypatch.setattr(pa_mod.protocol, "poll_a2a_task", poll) + monkeypatch.setattr(pa_mod.asyncio, "sleep", _no_sleep) + if decide is not None: + monkeypatch.setattr(PromptAgentTaskStep, "_decide", decide) + return seen + + +async def _no_sleep(_seconds): + return None + + +def _context(*envs, user_sim_triggers: bool = False) -> TaskStepContext: + ctx = TaskStepContext(instance_id="ti-1", deployed_envs=list(envs)) + ctx.deployed_agents.append(DeployedAgent(agent_name="solver", api_url="http://solver", a2a_url="http://solver", + a2a_card={})) + if user_sim_triggers: + card = {"capabilities": {"extensions": [{"uri": A2AAgent.EXT_TRIGGERS, "params": { + "endpoint": "/ext/triggers", "methods": {"decide": {"method": "POST", "endpoint": "/ext/triggers/decide"}}}}]}} + ctx.deployed_agents.append(DeployedAgent(agent_name="stakeholder", api_url="http://user-sim", + a2a_url="http://user-sim", a2a_card=card)) + ctx.metadata["agent_trigger_registrations"] = [{"agent_name": "stakeholder", "added": ["news"]}] + return ctx + + +def _step(between_turns, turns: int = 6, **kw) -> PromptAgentTaskStep: + return PromptAgentTaskStep(id="watch", version=None, prompt="Watch my day.", agent_name="solver", + max_conversation_turns=turns, between_turns=between_turns, **kw) + + +def test_unset_adds_nothing_to_the_stored_step(): + data = PromptAgentTaskStep(id="s", version=None, prompt="hi").to_dict() + assert "between_turns" not in data and "record_env_clock" not in data + again = PromptAgentTaskStep.from_dict(data) + assert (again.between_turns, again.record_env_clock) == (None, False) + + +@pytest.mark.parametrize("given", ["next_env_event", {"mode": "next_env_event", "end_time": DAY2, "wake": "message"}]) +def test_the_option_round_trips_as_given(given): + step = PromptAgentTaskStep.from_dict(_step(given, record_env_clock=True).to_dict()) + assert (step.between_turns, step.record_env_clock) == (given, True) + + +@pytest.mark.parametrize("given,fragment", [ + ("every_turn", "mode must be"), + ({"mode": "next_env_event", "until": DAY1}, "unknown key"), + ({"mode": "next_env_event", "wake": "user"}, "wake must be one of"), + ({"mode": "next_env_event", "wake_message": "{now}"}, "may use only"), + ({"mode": "next_env_event", "wake_message": " "}, "non-empty"), + ({"mode": "next_env_event", "end_time": "Tuesday"}, "end_time"), + ({"mode": "next_env_event", "end_time": "2026-03-04T21:15:00"}, "end_time"), + ({"mode": "next_env_event", "fire_timeout_seconds": 0}, "positive"), + (["next_env_event"], "or an object with a mode"), +]) +def test_a_bad_option_is_refused_at_construction(given, fragment): + with pytest.raises(ValueError, match=fragment): + _step(given) + + +def test_a_single_turn_step_cannot_act_between_turns(): + with pytest.raises(ValueError, match="max_conversation_turns >= 2"): + _step("next_env_event", turns=1) + + +@pytest.mark.asyncio +async def test_two_events_days_apart_each_wake_the_agent_once_then_the_run_ends(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({"school-email": DAY1, "calendar-clash": DAY2}) + + ctx = await _step("next_env_event").execute(_context(env)) + + assert seen["solver"] == ["Watch my day.", DEFAULT_WAKE_MESSAGE.format(virtual_time=DAY1), + DEFAULT_WAKE_MESSAGE.format(virtual_time=DAY2)] + assert env.moves() == [("advance", {"virtual_time": DAY1}), ("advance", {"virtual_time": DAY2})] + advances = ctx.metadata["env_event_advances"]["watch"] + assert advances[0] == {"turn": 1, "mark": DAY1, "trigger_ids": ["school-email"], "statuses": {"school-email": "fired"}, + "moved": {"env-1": "advance"}} + assert advances[1]["trigger_ids"] == ["calendar-clash"] + assert advances[2] == {"turn": 3, "stopped": "no_pending_time_triggers", "virtual_time": None} + assert seen["closed"] >= 1 + + +@pytest.mark.asyncio +async def test_every_message_carries_the_env_clock_when_the_run_is_event_driven(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({"school-email": DAY1}) + + await _step("next_env_event").execute(_context(env)) + + stamps = [(kind, kw["env_clock"]) for kind, kw in seen["stored"]] + assert stamps == [("add", {"env-1": START}), ("complete", {"env-1": START}), + ("add", {"env-1": DAY1}), ("complete", {"env-1": DAY1})] + + +@pytest.mark.asyncio +async def test_end_time_stops_the_run_and_brings_the_clock_to_it(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({"day-1": DAY1, "day-5": "2026-03-07T09:00:00Z"}) + end = "2026-03-05T07:00:00Z" + + ctx = await _step({"mode": "next_env_event", "end_time": end}).execute(_context(env)) + + assert len(seen["solver"]) == 2 + assert env.moves()[-1] == ("advance", {"virtual_time": end}) + assert ctx.metadata["env_event_advances"]["watch"][-1] == {"turn": 2, "stopped": "end_time", "virtual_time": end} + assert env.rows["day-5"]["status"] == "armed" + + +@pytest.mark.asyncio +async def test_max_conversation_turns_still_bounds_the_run(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({"a": DAY1, "b": DAY2, "c": "2026-03-06T00:00:00Z"}) + + await _step("next_env_event", turns=2).execute(_context(env)) + + assert len(seen["solver"]) == 2 and env.moves() == [("advance", {"virtual_time": DAY1})] + + +@pytest.mark.asyncio +async def test_triggers_due_at_one_mark_share_one_wake(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({"sale": DAY1, "text": DAY1}) + + await _step({"mode": "next_env_event", "wake_message": "ping {virtual_time}: {trigger_ids}"}).execute(_context(env)) + + assert seen["solver"][1:] == [f"ping {DAY1}: sale, text"] + + +@pytest.mark.asyncio +async def test_a_gateway_without_advance_is_rearmed_at_its_own_rate(monkeypatch): + _wire(monkeypatch) + env = _Env({"school-email": DAY1}, rate=60.0, advance=False) + + await _step("next_env_event").execute(_context(env)) + + assert env.moves() == [("set_time", {"virtual_time": DAY1, "virtual_seconds_per_real_second": 60.0})] + + +@pytest.mark.asyncio +async def test_a_gateway_without_advance_refuses_to_rearm_over_a_relative_mark(monkeypatch): + _wire(monkeypatch) + env = _Env({"school-email": DAY1, "reminder": "P1D"}, advance=False) + + with pytest.raises(RuntimeError, match=r"re-anchor .*\['reminder'\]"): + await _step("next_env_event").execute(_context(env)) + assert env.moves() == [] + + +@pytest.mark.asyncio +async def test_a_failed_or_stuck_trigger_is_reported_and_the_agent_still_woken(monkeypatch): + seen = _wire(monkeypatch) + failed = await _step("next_env_event", turns=2).execute(_context(_Env({"e": DAY1}, outcome="failed"))) + stuck = await _step({"mode": "next_env_event", "fire_timeout_seconds": 0.01}, turns=2).execute( + _context(_Env({"e": DAY1}, outcome="stuck"))) + + assert failed.metadata["env_event_advances"]["watch"][0]["statuses"] == {"e": "failed"} + assert stuck.metadata["env_event_advances"]["watch"][0]["statuses"] == {"e": "timeout"} + assert len(seen["solver"]) == 4 + + +@pytest.mark.asyncio +async def test_wake_decide_lets_agent_triggers_speak_and_falls_back_to_the_wake_message(monkeypatch): + calls = [] + + async def decide(self, url, turn, message, context_id, env_triggers): + calls.append((turn, env_triggers)) + return {"fired": ["news"], "done": False, "parts": [{"kind": "text", "text": "[news] the school wrote"}]} \ + if turn == 1 else {"fired": [], "done": False, "parts": []} + + async def readback(self, url, timeout): + return httpx.Response(200, json={"firing_log": []}, request=httpx.Request("GET", url)) + + seen = _wire(monkeypatch, decide=decide) + monkeypatch.setattr(httpx.AsyncClient, "get", readback) + env = _Env({"school-email": DAY1, "calendar-clash": DAY2}) + + await _step({"mode": "next_env_event", "wake": "decide"}, user_agent_name="stakeholder").execute( + _context(env, user_sim_triggers=True)) + + assert seen["solver"][1:] == ["[news] the school wrote", DEFAULT_WAKE_MESSAGE.format(virtual_time=DAY2)] + assert calls[0] == (1, {"env-1": {"school-email": "fired", "calendar-clash": "armed"}}) + + +@pytest.mark.asyncio +async def test_a_run_with_no_armed_env_clock_ends_after_the_first_turn(monkeypatch): + seen = _wire(monkeypatch) + + ctx = await _step("next_env_event").execute(_context()) + + assert len(seen["solver"]) == 1 + assert ctx.metadata["env_event_advances"]["watch"] == [{"turn": 1, "stopped": "no_armed_env_clock", "virtual_time": None}] + + +@pytest.mark.asyncio +async def test_wake_decide_without_agent_triggers_is_refused(monkeypatch): + _wire(monkeypatch) + with pytest.raises(RuntimeError, match="wake 'decide' needs"): + await _step({"mode": "next_env_event", "wake": "decide"}).execute(_context(_Env({"e": DAY1}))) + + +@pytest.mark.asyncio +async def test_record_env_clock_stamps_a_plain_conversation_and_unset_stamps_nothing(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({}) + await PromptAgentTaskStep(id="s", version=None, prompt="hi", agent_name="solver", + record_env_clock=True).execute(_context(env)) + await PromptAgentTaskStep(id="s", version=None, prompt="hi", agent_name="solver").execute(_context(env)) + + assert [kw.get("env_clock", "absent") for _, kw in seen["stored"]] == [{"env-1": START}, {"env-1": START}, + "absent", "absent"] + assert env.calls == [("state", None), ("state", None)] From 9dc9ddf263498cb03bbb16b2c2f806386ab974e7 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 09:42:13 -0700 Subject: [PATCH 2/8] fix(prompt_agent): await a due trigger on each env by env and trigger id Two envs can each have a trigger with the same id due at one mark; keyed by id alone, the second was skipped and the wait ran to its timeout. Outcomes are now keyed by (env id, trigger id) and recorded as {env_id: {trigger_id: outcome}}. Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/prompt_agent.py | 34 ++++++++++--------- .../test_prompt_agent_between_turns.py | 22 ++++++++++-- 2 files changed, 37 insertions(+), 19 deletions(-) diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index cd06db75..87839df6 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -975,7 +975,8 @@ async def _advance_to_next_env_event(self, context: TaskStepContext, advance: di due = [(d, t) for m, d, t in pending if m == mark] moved = await self._move_env_clocks(envs, mark, check_marks=True) statuses = await self._wait_fired(context, due, advance["fire_timeout_seconds"]) - return {"mark": _iso(mark), "trigger_ids": [t["id"] for _, t in due], "statuses": statuses, "moved": moved} + trigger_ids = list(dict.fromkeys(t["id"] for _, t in due)) + return {"mark": _iso(mark), "trigger_ids": trigger_ids, "statuses": statuses, "moved": moved} async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: bool) -> dict: """Bring each env clock behind ``target`` up to it, by advance where offered, else a re-arm only fixed marks survive.""" @@ -998,36 +999,37 @@ async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: b return moved async def _wait_fired(self, context: TaskStepContext, due: list, timeout_seconds: float) -> dict: - """Each due trigger's outcome once it has fired, failed or been cancelled; "timeout" for one still unsettled.""" - before = {t["id"]: (t.get("fire_count") or 0, t.get("failure_count") or 0) for _, t in due} + """``{env_id: {trigger_id: outcome}}`` once each due trigger has fired, failed or been cancelled; "timeout" if not.""" + before = {(d.env_id, t["id"]): (t.get("fire_count") or 0, t.get("failure_count") or 0) for d, t in due} by_env: dict[str, tuple[Any, list[str]]] = {} for d, t in due: by_env.setdefault(d.env_id, (d, []))[1].append(t["id"]) - outcome: dict[str, str] = {} + outcome: dict[tuple[str, str], str] = {} deadline = time.monotonic() + timeout_seconds while True: - for d, ids in by_env.values(): + for env_id, (d, ids) in by_env.items(): rows = {t["id"]: t for t in (await self._get_trigger_state(d)).get("triggers", [])} - for tid in (i for i in ids if i not in outcome): + for tid in (i for i in ids if (env_id, i) not in outcome): row = rows.get(tid) - fires, failures = before[tid] + fires, failures = before[(env_id, tid)] if row is None: - outcome[tid] = "removed" + outcome[(env_id, tid)] = "removed" elif row.get("status") in TRIGGER_IN_FLIGHT_STATUSES: continue elif (row.get("failure_count") or 0) > failures: - outcome[tid] = "failed" + outcome[(env_id, tid)] = "failed" elif (row.get("fire_count") or 0) > fires or row.get("status") in _SETTLED_STATUSES: - outcome[tid] = "fired" if row.get("status") == "armed" else row["status"] - if len(outcome) == len(due) or time.monotonic() >= deadline: + outcome[(env_id, tid)] = "fired" if row.get("status") == "armed" else row["status"] + if len(outcome) == len(before) or time.monotonic() >= deadline: break context.heartbeat() await asyncio.sleep(_FIRE_POLL_SECONDS) - for _, t in due: - if t["id"] not in outcome: - logger.warning(f"{self.id}: time trigger '{t['id']}' had not settled {timeout_seconds}s after its mark") - outcome[t["id"]] = "timeout" - return outcome + statuses: dict[str, dict[str, str]] = {} + for env_id, tid in before: + if (env_id, tid) not in outcome: + logger.warning(f"{self.id}: time trigger '{tid}' on env '{env_id}' had not settled {timeout_seconds}s after its mark") + statuses.setdefault(env_id, {})[tid] = outcome.get((env_id, tid), "timeout") + return statuses _CAPTURE_BUDGET_SECONDS: ClassVar[float] = 90 diff --git a/tst/unit/task_step/test_prompt_agent_between_turns.py b/tst/unit/task_step/test_prompt_agent_between_turns.py index 47621670..88c19185 100644 --- a/tst/unit/task_step/test_prompt_agent_between_turns.py +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy +import time from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock @@ -167,7 +168,7 @@ async def test_two_events_days_apart_each_wake_the_agent_once_then_the_run_ends( DEFAULT_WAKE_MESSAGE.format(virtual_time=DAY2)] assert env.moves() == [("advance", {"virtual_time": DAY1}), ("advance", {"virtual_time": DAY2})] advances = ctx.metadata["env_event_advances"]["watch"] - assert advances[0] == {"turn": 1, "mark": DAY1, "trigger_ids": ["school-email"], "statuses": {"school-email": "fired"}, + assert advances[0] == {"turn": 1, "mark": DAY1, "trigger_ids": ["school-email"], "statuses": {"env-1": {"school-email": "fired"}}, "moved": {"env-1": "advance"}} assert advances[1]["trigger_ids"] == ["calendar-clash"] assert advances[2] == {"turn": 3, "stopped": "no_pending_time_triggers", "virtual_time": None} @@ -220,6 +221,21 @@ async def test_triggers_due_at_one_mark_share_one_wake(monkeypatch): assert seen["solver"][1:] == [f"ping {DAY1}: sale, text"] +@pytest.mark.asyncio +async def test_one_trigger_id_on_two_envs_is_awaited_on_each_without_waiting_out_the_timeout(monkeypatch): + seen = _wire(monkeypatch) + envs = (_Env({"news": DAY1}, env_id="env-1"), _Env({"news": DAY1}, env_id="env-2")) + started = time.monotonic() + + ctx = await _step({"mode": "next_env_event", "fire_timeout_seconds": 5}, turns=2).execute(_context(*envs)) + + assert time.monotonic() - started < 2 + advance = ctx.metadata["env_event_advances"]["watch"][0] + assert advance["statuses"] == {"env-1": {"news": "fired"}, "env-2": {"news": "fired"}} + assert (advance["trigger_ids"], advance["moved"]) == (["news"], {"env-1": "advance", "env-2": "advance"}) + assert seen["solver"][1] == DEFAULT_WAKE_MESSAGE.format(virtual_time=DAY1) + + @pytest.mark.asyncio async def test_a_gateway_without_advance_is_rearmed_at_its_own_rate(monkeypatch): _wire(monkeypatch) @@ -247,8 +263,8 @@ async def test_a_failed_or_stuck_trigger_is_reported_and_the_agent_still_woken(m stuck = await _step({"mode": "next_env_event", "fire_timeout_seconds": 0.01}, turns=2).execute( _context(_Env({"e": DAY1}, outcome="stuck"))) - assert failed.metadata["env_event_advances"]["watch"][0]["statuses"] == {"e": "failed"} - assert stuck.metadata["env_event_advances"]["watch"][0]["statuses"] == {"e": "timeout"} + assert failed.metadata["env_event_advances"]["watch"][0]["statuses"] == {"env-1": {"e": "failed"}} + assert stuck.metadata["env_event_advances"]["watch"][0]["statuses"] == {"env-1": {"e": "timeout"}} assert len(seen["solver"]) == 4 From 974d115fdb827953f66a93e79e9360b1bb2b86e3 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 09:42:40 -0700 Subject: [PATCH 3/8] fix(prompt_agent): a running clock that reaches the mark before the advance lands is not an error With a running clock the mark can pass between the clock read and the advance, which the gateway then refuses as backwards. On that 400 the clock is read again, and if it has reached the mark the move is recorded as "passed"; any other refusal still fails the step. Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/prompt_agent.py | 14 ++++++++-- .../test_prompt_agent_between_turns.py | 28 ++++++++++++++++++- 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 87839df6..6dd75d50 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -985,8 +985,7 @@ async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: b if _utc(clock["virtual_time"]) >= target: continue if d.supports(EXT_CLOCK_URI, "advance"): - await d.invoke(EXT_CLOCK_URI, "advance", {"virtual_time": _iso(target)}, timeout=15) - moved[d.env_id] = "advance" + moved[d.env_id] = await self._advance_clock(d, target) continue unsafe = [t["id"] for t in rows if check_marks and t.get("type") == "time" and t.get("status") == "armed" and not _rearm_safe(t.get("when") or {})] @@ -998,6 +997,17 @@ async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: b moved[d.env_id] = "set_time" return moved + async def _advance_clock(self, d, target: datetime) -> str: + """Advance one env clock to ``target``: "advance", or "passed" when a running clock got there before the call did.""" + try: + await d.invoke(EXT_CLOCK_URI, "advance", {"virtual_time": _iso(target)}, timeout=15) + return "advance" + except httpx.HTTPStatusError as e: + now = await self._get_clock_state(d) if e.response.status_code == 400 else None + if now and now.get("armed") and _utc(now["virtual_time"]) >= target: + return "passed" + raise + async def _wait_fired(self, context: TaskStepContext, due: list, timeout_seconds: float) -> dict: """``{env_id: {trigger_id: outcome}}`` once each due trigger has fired, failed or been cancelled; "timeout" if not.""" before = {(d.env_id, t["id"]): (t.get("fire_count") or 0, t.get("failure_count") or 0) for d, t in due} diff --git a/tst/unit/task_step/test_prompt_agent_between_turns.py b/tst/unit/task_step/test_prompt_agent_between_turns.py index 88c19185..b3f9f639 100644 --- a/tst/unit/task_step/test_prompt_agent_between_turns.py +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -34,8 +34,9 @@ class _Env: """A deployed env whose clock and time triggers behave as a gateway's: a trigger fires once the clock reaches its mark.""" def __init__(self, marks: dict, *, rate: float = 0.0, advance: bool = True, outcome: str = "fired", - env_id: str = "env-1"): + env_id: str = "env-1", overrun: timedelta | None = None): self.env_id, self.rate, self.offers_advance, self.outcome = env_id, rate, advance, outcome + self.overrun = overrun self.now = _dt(START) self.calls: list[tuple[str, dict | None]] = [] self.rows = {tid: {"id": tid, "type": "time", "status": "armed", "next_mark": at if not at.startswith("P") else @@ -49,6 +50,10 @@ def supports(self, uri: str, method: str) -> bool: async def invoke(self, uri, method, params=None, *, timeout=30, environment_name=None): self.calls.append((method, params)) + if uri == EXT_CLOCK_URI and method == "advance" and self.overrun is not None: + self.now = _dt(params["virtual_time"]) + self.overrun + raise httpx.HTTPStatusError("advance only moves forward", request=httpx.Request("PUT", "http://gw"), + response=httpx.Response(400, request=httpx.Request("PUT", "http://gw"))) if uri == EXT_CLOCK_URI and method in ("advance", "set_time"): self.now = _dt(params["virtual_time"]) return {} @@ -236,6 +241,27 @@ async def test_one_trigger_id_on_two_envs_is_awaited_on_each_without_waiting_out assert seen["solver"][1] == DEFAULT_WAKE_MESSAGE.format(virtual_time=DAY1) +@pytest.mark.asyncio +async def test_a_running_clock_that_passes_the_mark_before_the_advance_lands_is_not_an_error(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({"school-email": DAY1}, rate=60.0, overrun=timedelta(seconds=2)) + + ctx = await _step("next_env_event", turns=2).execute(_context(env)) + + advance = ctx.metadata["env_event_advances"]["watch"][0] + assert (advance["moved"], advance["statuses"]) == ({"env-1": "passed"}, {"env-1": {"school-email": "fired"}}) + assert len(seen["solver"]) == 2 + + +@pytest.mark.asyncio +async def test_an_advance_refused_for_another_reason_still_fails_the_step(monkeypatch): + _wire(monkeypatch) + env = _Env({"school-email": DAY1}, overrun=timedelta(seconds=-2)) + + with pytest.raises(httpx.HTTPStatusError): + await _step("next_env_event", turns=2).execute(_context(env)) + + @pytest.mark.asyncio async def test_a_gateway_without_advance_is_rearmed_at_its_own_rate(monkeypatch): _wire(monkeypatch) From e6b4d0e732387250ab5e86f70bf6e579209ce5c2 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 09:43:05 -0700 Subject: [PATCH 4/8] fix(prompt_agent): a recurring time trigger still mid-fire keeps its next mark in the schedule A recurring trigger whose fire outlived the wait (an action delay on a frozen clock, say) stays firing with its next mark already stamped; only armed ones were scheduled, so the run could end with "no_pending_time_triggers" despite future marks. Firing and queued triggers with a next mark are now scheduled too. Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/prompt_agent.py | 5 +++-- .../test_prompt_agent_between_turns.py | 22 +++++++++++++++++-- 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 6dd75d50..18b47e40 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -121,6 +121,7 @@ def _file_uris(parts: list[dict]) -> list[str]: _BETWEEN_TURNS_KEYS = ("mode", "wake", "wake_message", "end_time", "fire_timeout_seconds") _WAKES = ("message", "decide") _SETTLED_STATUSES = ("fired", "failed", "cancelled") +_SCHEDULED_STATUSES = ("armed", *TRIGGER_IN_FLIGHT_STATUSES) _FIRE_POLL_SECONDS = 0.5 @@ -962,7 +963,7 @@ async def _advance_to_next_env_event(self, context: TaskStepContext, advance: di if clock and clock.get("armed"): envs.append((d, clock, (await self._get_trigger_state(d)).get("triggers", []))) pending = [(_utc(t["next_mark"]), d, t) for d, _, rows in envs for t in rows - if t.get("type") == "time" and t.get("status") == "armed" and t.get("next_mark")] + if t.get("type") == "time" and t.get("status") in _SCHEDULED_STATUSES and t.get("next_mark")] if not envs: return {"stopped": "no_armed_env_clock", "virtual_time": None} end = _utc(advance["end_time"]) if advance["end_time"] else None @@ -988,7 +989,7 @@ async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: b moved[d.env_id] = await self._advance_clock(d, target) continue unsafe = [t["id"] for t in rows if check_marks and t.get("type") == "time" - and t.get("status") == "armed" and not _rearm_safe(t.get("when") or {})] + and t.get("status") in _SCHEDULED_STATUSES and not _rearm_safe(t.get("when") or {})] if unsafe: raise RuntimeError(f"{self.id}: env '{d.env_id}' has no clock advance, and a re-arm would re-anchor the " f"relative or recurring time trigger(s) {unsafe}; give them absolute marks") diff --git a/tst/unit/task_step/test_prompt_agent_between_turns.py b/tst/unit/task_step/test_prompt_agent_between_turns.py index b3f9f639..6dfbf9cd 100644 --- a/tst/unit/task_step/test_prompt_agent_between_turns.py +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -60,8 +60,12 @@ async def invoke(self, uri, method, params=None, *, timeout=30, environment_name if uri == EXT_CLOCK_URI: return {"armed": True, "virtual_time": _z(self.now), "virtual_seconds_per_real_second": self.rate} for row in self.rows.values(): - if row["status"] == "armed" and row["next_mark"] and _dt(row["next_mark"]) <= self.now: - if self.outcome == "fired": + if row["status"] == "firing" and self.outcome == "fired": + row.update(status="armed", fire_count=row["fire_count"] + 1) + elif row["status"] == "armed" and row["next_mark"] and _dt(row["next_mark"]) <= self.now: + if "every" in row["when"] and self.outcome == "fired": + row.update(fire_count=row["fire_count"] + 1, next_mark=_z(_dt(row["next_mark"]) + timedelta(days=1))) + elif self.outcome == "fired": row.update(status="fired", fire_count=1, next_mark=None) elif self.outcome == "failed": row.update(status="failed", failure_count=1, next_mark=None) @@ -262,6 +266,20 @@ async def test_an_advance_refused_for_another_reason_still_fails_the_step(monkey await _step("next_env_event", turns=2).execute(_context(env)) +@pytest.mark.asyncio +async def test_a_recurring_trigger_still_mid_fire_keeps_its_next_mark_in_the_schedule(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({}) + env.rows["daily"] = {"id": "daily", "type": "time", "status": "firing", "next_mark": DAY1, "fire_count": 3, + "failure_count": 0, "when": {"type": "time", "every": "P1D"}} + + ctx = await _step("next_env_event", turns=3).execute(_context(env)) + + assert env.moves() == [("advance", {"virtual_time": DAY1}), ("advance", {"virtual_time": "2026-03-04T09:30:00Z"})] + assert [a.get("trigger_ids") for a in ctx.metadata["env_event_advances"]["watch"]] == [["daily"], ["daily"]] + assert len(seen["solver"]) == 3 + + @pytest.mark.asyncio async def test_a_gateway_without_advance_is_rearmed_at_its_own_rate(monkeypatch): _wire(monkeypatch) From c42e8539f56a2ec7ef8b6648eaabab564f5f02a4 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 09:43:25 -0700 Subject: [PATCH 5/8] fix(prompt_agent): the message an agent trigger ends the conversation with carries the env clock too Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/prompt_agent.py | 3 ++- .../test_prompt_agent_between_turns.py | 19 +++++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 18b47e40..258f1df8 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -709,7 +709,8 @@ def record_turn() -> None: if bool(resp.get("done")): current_user_parts = resp.get("parts") or [] conversation_store.add_a2a_task(conversation_id, parts=current_user_parts, - a2a_task_id=uuid.uuid4().hex, role="user") + a2a_task_id=uuid.uuid4().hex, role="user", + **(await self._env_clock_kwargs(clock_envs))) conversation_store.mark_closed(conversation_id) break if resp.get("fired"): diff --git a/tst/unit/task_step/test_prompt_agent_between_turns.py b/tst/unit/task_step/test_prompt_agent_between_turns.py index 6dfbf9cd..9b455155 100644 --- a/tst/unit/task_step/test_prompt_agent_between_turns.py +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -345,6 +345,25 @@ async def test_a_run_with_no_armed_env_clock_ends_after_the_first_turn(monkeypat assert ctx.metadata["env_event_advances"]["watch"] == [{"turn": 1, "stopped": "no_armed_env_clock", "virtual_time": None}] +@pytest.mark.asyncio +async def test_an_agent_trigger_that_ends_the_run_records_its_message_with_the_env_clock(monkeypatch): + async def decide(self, url, turn, message, context_id, env_triggers): + return {"fired": ["bye"], "done": True, "parts": [{"kind": "text", "text": "[news] that's all"}]} + + async def readback(self, url, timeout): + return httpx.Response(200, json={"firing_log": []}, request=httpx.Request("GET", url)) + + seen = _wire(monkeypatch, decide=decide) + monkeypatch.setattr(httpx.AsyncClient, "get", readback) + + await _step({"mode": "next_env_event", "wake": "decide"}, user_agent_name="stakeholder").execute( + _context(_Env({"school-email": DAY1}), user_sim_triggers=True)) + + assert seen["stored"][-1] == ("add", {"parts": [{"kind": "text", "text": "[news] that's all"}], + "a2a_task_id": seen["stored"][-1][1]["a2a_task_id"], "role": "user", + "env_clock": {"env-1": DAY1}}) + + @pytest.mark.asyncio async def test_wake_decide_without_agent_triggers_is_refused(monkeypatch): _wire(monkeypatch) From 810b448fcd93dda2802cb3061d0d7d3b322e4c00 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 09:43:47 -0700 Subject: [PATCH 6/8] fix(gateway): PUT /clock/advance answers a body that isn't a JSON object with 400, not a server error Co-Authored-By: Claude Opus 5.5 --- src/agent_env/env/gateway/gateway.py | 2 ++ tst/unit/env/gateway/clock_advance_test.py | 3 +++ 2 files changed, 5 insertions(+) diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index d3c8b19a..435fe349 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -275,6 +275,8 @@ async def clock_advance(request: Request) -> Response: body = await request.json() except Exception: return self._json_response({"ok": False, "error": "Invalid JSON body"}, 400) + if not isinstance(body, dict): + return self._json_response({"ok": False, "error": "body must be a JSON object"}, 400) try: state = self._clock.advance(body.get("virtual_time")) except ClockError as e: diff --git a/tst/unit/env/gateway/clock_advance_test.py b/tst/unit/env/gateway/clock_advance_test.py index df7e625b..0a76031f 100644 --- a/tst/unit/env/gateway/clock_advance_test.py +++ b/tst/unit/env/gateway/clock_advance_test.py @@ -152,6 +152,9 @@ def test_route_advances_an_armed_clock(gateway): @pytest.mark.parametrize("body,fragment", [ ({"virtual_time": "2026-01-04T08:00:00Z"}, "not armed"), ("{not json", "Invalid JSON body"), + ("null", "must be a JSON object"), + ("[]", "must be a JSON object"), + ('"2026-01-04T08:00:00Z"', "must be a JSON object"), ]) def test_route_refuses_with_400(gateway, body, fragment): resp = _put(gateway, "/clock/advance", body) From 8c1a11ae756a591cf25f1a4df7ce52fd357ad153 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 09:52:40 -0700 Subject: [PATCH 7/8] fix(prompt_agent): a mark whose trigger didn't settle is not chosen again A recurring trigger whose fire outlives its interval on a frozen clock stays firing with the same next mark, so every later turn chose that mark again, waited out the timeout and woke the agent without moving time. Marks that timed out are now skipped, so the run moves on to the next event or end_time. Co-Authored-By: Claude Opus 5.5 --- .../task_step/task_steps/prompt_agent.py | 11 ++++++++--- .../test_prompt_agent_between_turns.py | 18 ++++++++++++++++++ 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 258f1df8..02c5a8e6 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -563,6 +563,7 @@ async def _execute_conversation( f"user simulator with agent triggers registered") clock_envs = ([d for d in context.deployed_envs if d.supports(EXT_CLOCK_URI, "state")] if advance is not None or self.record_env_clock else None) + timed_out: set[tuple[str, str, str]] = set() solver_context_id = a2a_context_id if self.context_id: @@ -690,7 +691,7 @@ def record_turn() -> None: await series.capture_turn(context, turn + 1) if advance is not None: - event = await self._advance_to_next_env_event(context, advance) + event = await self._advance_to_next_env_event(context, advance, timed_out) (context.metadata.setdefault("env_event_advances", {}) .setdefault(self.id, []).append({"turn": turn + 1, **event})) if "stopped" in event: @@ -954,7 +955,8 @@ async def _env_clock_kwargs(self, envs: Optional[list]) -> dict: return {"env_clock": {d.env_id: (s or {}).get("virtual_time") if (s or {}).get("armed") else None for d, s in zip(envs, states)}} - async def _advance_to_next_env_event(self, context: TaskStepContext, advance: dict) -> dict: + async def _advance_to_next_env_event(self, context: TaskStepContext, advance: dict, + timed_out: set[tuple[str, str, str]]) -> dict: """Move armed env clocks to the earliest pending time-trigger mark and await its triggers; ``stopped`` once none is due.""" envs = [] for d in context.deployed_envs: @@ -964,7 +966,8 @@ async def _advance_to_next_env_event(self, context: TaskStepContext, advance: di if clock and clock.get("armed"): envs.append((d, clock, (await self._get_trigger_state(d)).get("triggers", []))) pending = [(_utc(t["next_mark"]), d, t) for d, _, rows in envs for t in rows - if t.get("type") == "time" and t.get("status") in _SCHEDULED_STATUSES and t.get("next_mark")] + if t.get("type") == "time" and t.get("status") in _SCHEDULED_STATUSES and t.get("next_mark") + and (d.env_id, t["id"], t["next_mark"]) not in timed_out] if not envs: return {"stopped": "no_armed_env_clock", "virtual_time": None} end = _utc(advance["end_time"]) if advance["end_time"] else None @@ -977,6 +980,8 @@ async def _advance_to_next_env_event(self, context: TaskStepContext, advance: di due = [(d, t) for m, d, t in pending if m == mark] moved = await self._move_env_clocks(envs, mark, check_marks=True) statuses = await self._wait_fired(context, due, advance["fire_timeout_seconds"]) + # A mark whose trigger didn't settle is not chosen again, so an unfinished fire can't hold time still. + timed_out.update((d.env_id, t["id"], t["next_mark"]) for d, t in due if statuses[d.env_id][t["id"]] == "timeout") trigger_ids = list(dict.fromkeys(t["id"] for _, t in due)) return {"mark": _iso(mark), "trigger_ids": trigger_ids, "statuses": statuses, "moved": moved} diff --git a/tst/unit/task_step/test_prompt_agent_between_turns.py b/tst/unit/task_step/test_prompt_agent_between_turns.py index 9b455155..60195eb1 100644 --- a/tst/unit/task_step/test_prompt_agent_between_turns.py +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -60,6 +60,8 @@ async def invoke(self, uri, method, params=None, *, timeout=30, environment_name if uri == EXT_CLOCK_URI: return {"armed": True, "virtual_time": _z(self.now), "virtual_seconds_per_real_second": self.rate} for row in self.rows.values(): + if row.get("stuck"): + continue if row["status"] == "firing" and self.outcome == "fired": row.update(status="armed", fire_count=row["fire_count"] + 1) elif row["status"] == "armed" and row["next_mark"] and _dt(row["next_mark"]) <= self.now: @@ -280,6 +282,22 @@ async def test_a_recurring_trigger_still_mid_fire_keeps_its_next_mark_in_the_sch assert len(seen["solver"]) == 3 +@pytest.mark.asyncio +async def test_an_unfinished_recurring_fire_does_not_hold_time_at_its_mark(monkeypatch): + seen = _wire(monkeypatch) + env = _Env({"later": DAY2}) + env.rows["daily"] = {"id": "daily", "type": "time", "status": "firing", "next_mark": DAY1, "fire_count": 1, + "failure_count": 0, "when": {"type": "time", "every": "P1D"}, "stuck": True} + + ctx = await _step({"mode": "next_env_event", "fire_timeout_seconds": 0.01}, turns=5).execute(_context(env)) + + assert env.moves() == [("advance", {"virtual_time": DAY1}), ("advance", {"virtual_time": DAY2})] + advances = ctx.metadata["env_event_advances"]["watch"] + assert advances[0]["statuses"] == {"env-1": {"daily": "timeout"}} + assert advances[1]["statuses"] == {"env-1": {"later": "fired"}} + assert advances[2]["stopped"] == "no_pending_time_triggers" and len(seen["solver"]) == 3 + + @pytest.mark.asyncio async def test_a_gateway_without_advance_is_rearmed_at_its_own_rate(monkeypatch): _wire(monkeypatch) From eda2fe5031b8af78dfcd06d2526aca03ae9861d8 Mon Sep 17 00:00:00 2001 From: Sarah Chitty Date: Tue, 20 Oct 2026 18:00:43 -0700 Subject: [PATCH 8/8] review: the jump to end_time refuses a re-arm over a relative or recurring mark, as every other jump does Against a gateway without clock advance, stopping at end_time re-armed the clock unchecked, re-anchoring a pending relative or recurring time trigger past end_time for the steps after it. That jump now runs the same check. Co-Authored-By: Claude Opus 5.5 --- src/agent_env/task_step/task_steps/prompt_agent.py | 8 ++++---- tst/unit/task_step/test_prompt_agent_between_turns.py | 10 ++++++++++ 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 02c5a8e6..0d4aa397 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -974,18 +974,18 @@ async def _advance_to_next_env_event(self, context: TaskStepContext, advance: di mark = min((m for m, _, _ in pending), default=None) if mark is None or (end is not None and mark > end): if end is not None: - await self._move_env_clocks(envs, end, check_marks=False) + await self._move_env_clocks(envs, end) return {"stopped": "end_time" if mark is not None else "no_pending_time_triggers", "virtual_time": _iso(end) if end is not None else None} due = [(d, t) for m, d, t in pending if m == mark] - moved = await self._move_env_clocks(envs, mark, check_marks=True) + moved = await self._move_env_clocks(envs, mark) statuses = await self._wait_fired(context, due, advance["fire_timeout_seconds"]) # A mark whose trigger didn't settle is not chosen again, so an unfinished fire can't hold time still. timed_out.update((d.env_id, t["id"], t["next_mark"]) for d, t in due if statuses[d.env_id][t["id"]] == "timeout") trigger_ids = list(dict.fromkeys(t["id"] for _, t in due)) return {"mark": _iso(mark), "trigger_ids": trigger_ids, "statuses": statuses, "moved": moved} - async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: bool) -> dict: + async def _move_env_clocks(self, envs: list, target: datetime) -> dict: """Bring each env clock behind ``target`` up to it, by advance where offered, else a re-arm only fixed marks survive.""" moved = {} for d, clock, rows in envs: @@ -994,7 +994,7 @@ async def _move_env_clocks(self, envs: list, target: datetime, *, check_marks: b if d.supports(EXT_CLOCK_URI, "advance"): moved[d.env_id] = await self._advance_clock(d, target) continue - unsafe = [t["id"] for t in rows if check_marks and t.get("type") == "time" + unsafe = [t["id"] for t in rows if t.get("type") == "time" and t.get("status") in _SCHEDULED_STATUSES and not _rearm_safe(t.get("when") or {})] if unsafe: raise RuntimeError(f"{self.id}: env '{d.env_id}' has no clock advance, and a re-arm would re-anchor the " diff --git a/tst/unit/task_step/test_prompt_agent_between_turns.py b/tst/unit/task_step/test_prompt_agent_between_turns.py index 60195eb1..1ce4948e 100644 --- a/tst/unit/task_step/test_prompt_agent_between_turns.py +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -318,6 +318,16 @@ async def test_a_gateway_without_advance_refuses_to_rearm_over_a_relative_mark(m assert env.moves() == [] +@pytest.mark.asyncio +async def test_a_gateway_without_advance_refuses_to_rearm_to_end_time_over_a_relative_mark(monkeypatch): + _wire(monkeypatch) + env = _Env({"reminder": "P1D"}, advance=False) + + with pytest.raises(RuntimeError, match=r"re-anchor .*\['reminder'\]"): + await _step({"mode": "next_env_event", "end_time": "2026-03-02T20:00:00Z"}).execute(_context(env)) + assert env.moves() == [] + + @pytest.mark.asyncio async def test_a_failed_or_stuck_trigger_is_reported_and_the_agent_still_woken(monkeypatch): seen = _wire(monkeypatch)