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 4acb7b7e..7899d5a7 100644 --- a/src/agent_env/env/gateway/constants.py +++ b/src/agent_env/env/gateway/constants.py @@ -195,10 +195,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 9c732539..441bd824 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -276,6 +276,20 @@ 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) + 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: + 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..0d4aa397 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,67 @@ 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") +_SCHEDULED_STATUSES = ("armed", *TRIGGER_IN_FLIGHT_STATUSES) +_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 +220,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 +235,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 +268,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 +346,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 +386,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 +494,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 +557,13 @@ 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) + timed_out: set[tuple[str, str, str]] = set() solver_context_id = a2a_context_id if self.context_id: @@ -529,6 +612,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 +623,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 +656,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 +690,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, timed_out) + (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})) @@ -615,7 +710,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"): @@ -623,6 +719,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 +947,107 @@ 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, + 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: + 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") 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 + 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) + 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) + 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) -> 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"): + moved[d.env_id] = await self._advance_clock(d, target) + continue + 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 " + 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 _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} + 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[tuple[str, str], str] = {} + deadline = time.monotonic() + timeout_seconds + while True: + 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 (env_id, i) not in outcome): + row = rows.get(tid) + fires, failures = before[(env_id, tid)] + if row is None: + outcome[(env_id, tid)] = "removed" + elif row.get("status") in TRIGGER_IN_FLIGHT_STATUSES: + continue + elif (row.get("failure_count") or 0) > failures: + outcome[(env_id, tid)] = "failed" + elif (row.get("fire_count") or 0) > fires or row.get("status") in _SETTLED_STATUSES: + 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) + 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 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..0a76031f --- /dev/null +++ b/tst/unit/env/gateway/clock_advance_test.py @@ -0,0 +1,167 @@ +"""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"), + ("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) + 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..1ce4948e --- /dev/null +++ b/tst/unit/task_step/test_prompt_agent_between_turns.py @@ -0,0 +1,412 @@ +"""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 +import time +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", 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 + _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 == "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 {} + 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: + 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) + 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": {"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} + 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_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_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_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_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) + 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_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) + 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"] == {"env-1": {"e": "failed"}} + assert stuck.metadata["env_event_advances"]["watch"][0]["statuses"] == {"env-1": {"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_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) + 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)]