diff --git a/pyproject.toml b/pyproject.toml index bd2424f40..e219facfa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -123,6 +123,7 @@ hello = "agent_env.examples" echo-mcp = "agent_env.examples" mail-mcp = "agent_env.examples" next-env-event = "agent_env.examples" +advance-to-next-event = "agent_env.examples" [project.urls] Homepage = "https://www.agentenvframework.com" diff --git a/src/agent_env/examples/advance-to-next-event/README.md b/src/agent_env/examples/advance-to-next-event/README.md new file mode 100644 index 000000000..d0334b38f --- /dev/null +++ b/src/agent_env/examples/advance-to-next-event/README.md @@ -0,0 +1,47 @@ +A prompt_agent post-turn hook that runs one conversation over simulated days and wakes the agent only for what +would really reach it. Unlike `next-env-event`, which wakes the agent with a nudge at every env event, it moves time +past world events silently. + +`artifacts/advance-to-next-event/hook.py` is the hook. After each agent turn it waits for what the turn set off to +finish, then lets a firing agent trigger answer. Otherwise it moves the frozen clock, silently, to the next thing +that happens: a world event (a trigger's mark, a delay ending, an executor's hours opening), something due for the +agent on notifications/v1, or a message the person writes at a set time. World events change the env but wake no +one. The agent is woken only when one of these has come due: + +- a check it scheduled with `schedule_check` (which a `configure_scheduled_checks` step gives its role), +- a notification a trigger sent it with a `notify` action, through a channel the task declares, +- a message from `person_messages`. + +The hook takes nothing from notifications/v1: it returns `{"message": []}` and `prompt_agent` delivers what came due, +each line saying when and through what, so nothing poses as the person. A person's message is the hook's own reply, +after those lines: + +``` +[Tue Feb 2, 9:00 AM · Curbfinch alert] New for "rug 8x10": 1970s wool rug · $360 +[Tue Feb 2, 6:00 PM · your scheduled check] Look for new 8x10 rugs under $300. +``` + +The conversation ends when nothing can wake the agent any more, or the next thing is past `end_time`. + +Reference the hook from a `prompt_agent` step, after `sync_env_clock` (frozen: `virtual_seconds_per_real_second: 0`) +and `configure_scheduled_checks` (`"roles": ["default"]`). `time_zone` on `prompt_agent` sets the zone the delivered +lines show: + +```json +{"id": "week", "type": "prompt_agent", "prompt": "Keep an eye on curbfinch this week for an 8x10 rug.", + "max_conversation_turns": 20, "time_zone": "America/Los_Angeles", + "post_turn_hooks": [{"file_artifact_id": "@local/agentenv-framework/advance-to-next-event/advance-to-next-event", + "params": {"end_time": "P5D", "time_zone": "America/Los_Angeles"}}]} +``` + +Its params, all optional: + +- `end_time`: an RFC 3339 instant with an offset, or an ISO 8601 duration from the clock's t0 (`P5D`). +- `role`: the agent's AgentEnv-Role, whose checks and notifications it moves the clock to; `default` by default. +- `time_zone`: the IANA zone the person's messages show their times in; UTC by default. +- `person_messages`: `[{"at": RFC 3339 instant, "text": "..."}]`, what the person writes and when. +- `settle_timeout_seconds`: how long to wait for triggers to settle, 300 by default. +- `max_jumps_per_turn`: silent world events it may pass between two wakes, 500 by default. + +`max_conversation_turns` caps the wakes. The env needs a gateway with clock/v1 `advance` and notifications/v1. A hook +runs in the control-plane process with the trust of an `env_outcome_verifier` script. diff --git a/src/agent_env/examples/advance-to-next-event/artifacts/advance-to-next-event/hook.py b/src/agent_env/examples/advance-to-next-event/artifacts/advance-to-next-event/hook.py new file mode 100644 index 000000000..9db614718 --- /dev/null +++ b/src/agent_env/examples/advance-to-next-event/artifacts/advance-to-next-event/hook.py @@ -0,0 +1,204 @@ +"""A prompt_agent post-turn hook that runs a conversation over simulated days by moving a frozen clock to the next +thing due for the agent: a check it scheduled for itself or a notification a trigger sent it (notifications/v1, which +prompt_agent delivers), or a message the person writes at a set time (which this hook sends). Everything else the +world does happens silently on the way, while the agent is away.""" +import asyncio +import logging +import re +import time +from datetime import datetime, timedelta, timezone + +from agent_env.task_step.task_steps.notification_delivery import label_time, label_zone + +CLOCK = "urn:agentenv:clock/v1" +TRIGGERS = "urn:agentenv:triggers/v1" +NOTIFICATIONS = "urn:agentenv:notifications/v1" +IN_FLIGHT = ("firing", "queued") +PARAMS = ("end_time", "role", "time_zone", "person_messages", "settle_timeout_seconds", "max_jumps_per_turn") +POLL_SECONDS = 0.5 + +logger = logging.getLogger(__name__) + + +async def post_turn(turn, params): + """A firing agent trigger answers first; else move the clock, silently, to the next thing due for the agent.""" + if turn.last: + return None + end_time, role, zone, person, settle_timeout, max_jumps = _params(params) + envs = [env for env in turn.envs if env.supports(CLOCK, "advance") and env.supports(NOTIFICATIONS, "state")] + if not envs: + return _end(turn, "no_env_with_clock_and_notifications") + await _settle(turn, envs, settle_timeout) + answer = await turn.user_triggers() + if answer is not None: + return answer + t0 = min([_utc((await env.invoke(CLOCK, "state"))["t0"]) for env in envs]) + end = _end_instant(end_time, t0) + sent = _delivered_person(turn) + for _ in range(max_jumps): + now = await _now(envs) + due = await _due(envs, role, now) + due_person = [m for m in person if m["at"] <= now and _iso(m["at"]) not in sent] + if due or due_person: + turn.record({"woke_at": _iso(now), "for": due + ["person"] * len(due_person), + **({"person_at": [_iso(m["at"]) for m in due_person]} if due_person else {})}) + # prompt_agent puts what came due on notifications/v1 ahead of this; [] sends only that + return {"message": _person_message(due_person, zone) if due_person else []} + nxt = await _next(envs, role, person, now, sent) + if nxt is None: + return _end(turn, "nothing_can_wake_the_agent", at=_iso(now)) + if end is not None and nxt > end: + return _end(turn, "end_time", next_at=_iso(nxt)) + for env in envs: + await _advance(env, nxt) + await _settle(turn, envs, settle_timeout) + return _end(turn, "max_jumps_per_turn", at=_iso(await _now(envs))) + + +def _params(params): + unknown = sorted(set(params) - set(PARAMS)) + if unknown: + raise ValueError(f"advance-to-next-event takes {list(PARAMS)}, not {unknown}") + end_time = params.get("end_time") + if end_time is not None: + _end_spec(end_time) + role = params.get("role", "default") + if not isinstance(role, str) or not role.strip(): + raise ValueError(f"role must name the agent's AgentEnv-Role, e.g. 'default', got {role!r}") + zone = label_zone(params.get("time_zone")) + person = [] + for m in params.get("person_messages", []): + if not isinstance(m, dict) or not isinstance(m.get("at"), str) or not isinstance(m.get("text"), str): + raise ValueError("each person_messages entry is {at: RFC 3339 instant, text: what the person writes}") + person.append({"at": _utc(m["at"]), "text": m["text"]}) + timeout = params.get("settle_timeout_seconds", 300) + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or timeout <= 0: + raise ValueError(f"settle_timeout_seconds must be a positive number, got {timeout!r}") + max_jumps = params.get("max_jumps_per_turn", 500) + if isinstance(max_jumps, bool) or not isinstance(max_jumps, int) or max_jumps < 1: + raise ValueError(f"max_jumps_per_turn must be a positive integer, got {max_jumps!r}") + return end_time, role.strip(), zone, sorted(person, key=lambda m: m["at"]), timeout, max_jumps + + +async def _now(envs): + return min([_utc((await env.invoke(CLOCK, "state"))["virtual_time"]) for env in envs]) + + +async def _due(envs, role, now): + """The source of each check and notification due for ``role`` by now, oldest first. Read, not taken: prompt_agent + takes and delivers them.""" + due = [] + for env in envs: + state = await env.invoke(NOTIFICATIONS, "state", {"to": role}) + due += [(_utc(c["next"]), "scheduled_check") for c in state.get("checks", []) + if c["status"] == "active" and _utc(c["next"]) <= now] + due += [(_utc(n["at"]) if n.get("at") else now, "trigger") for n in state.get("pending_notifications", []) + if not n.get("at") or _utc(n["at"]) <= now] + return [source for _, source in sorted(due, key=lambda d: d[0])] + + +async def _next(envs, role, person, now, sent): + """The earliest instant after now at which anything happens: a world event (silent), a check or notification due + for ``role``, or a person's message.""" + candidates = [m["at"] for m in person if m["at"] > now and _iso(m["at"]) not in sent] + for env in envs: + notes = await env.invoke(NOTIFICATIONS, "state", {"to": role}) + if notes.get("next_due_at"): + candidates.append(_utc(notes["next_due_at"])) + if env.supports(TRIGGERS, "state"): + world = await env.invoke(TRIGGERS, "state") + if world.get("next_event_at"): + candidates.append(_utc(world["next_event_at"])) + later = [c for c in candidates if c > now] + return min(later) if later else None + + +def _delivered_person(turn): + """The person's messages this hook already delivered, from its own records (a module global would outlive the run).""" + sent = set() + for entries in (turn.metadata.get("post_turn_hooks") or {}).values(): + for entry in entries: + sent.update(entry.get("person_at", [])) + return sent + + +def _person_message(person, zone): + """One line per message, oldest first, each saying when it was written.""" + return "\n".join(f"[{label_time(_iso(m['at']), zone)}] {m['text']}" for m in person) + + +async def _advance(env, target): + try: + await env.invoke(CLOCK, "advance", {"virtual_time": _iso(target)}) + except Exception: + clock = await env.invoke(CLOCK, "state") + if not (clock.get("armed") and _utc(clock["virtual_time"]) >= target): + raise + + +async def _settle(turn, envs, timeout): + """Wait until no trigger still fires (bar one waiting for a later time), so what an event set off has landed.""" + deadline = time.monotonic() + timeout + waiting = {i for i, env in enumerate(envs) if env.supports(TRIGGERS, "state")} + while waiting and time.monotonic() < deadline: + for i in sorted(waiting): + env = envs[i] + clock, state = await env.invoke(CLOCK, "state"), await env.invoke(TRIGGERS, "state") + now = _utc(clock["virtual_time"]) + if state.get("next_event_at") is not None and _utc(state["next_event_at"]) <= now: + await _advance(env, now) + elif not any(_busy(row, now) for row in state["triggers"]): + waiting.discard(i) + if waiting: + turn.heartbeat() + await asyncio.sleep(POLL_SECONDS) + if waiting: + logger.warning(f"env(s) {sorted(envs[i].env_id for i in waiting)} still had triggers firing {timeout}s on") + + +def _busy(row, now): + if row.get("evaluating"): # a state check the agent's turn set off is still running + return True + if row["status"] not in IN_FLIGHT and not row.get("pending"): + return False + return not (row.get("waiting") and row.get("resumes_at") and _utc(row["resumes_at"]) > now) + + +def _end_spec(end_time): + if isinstance(end_time, str) and end_time.startswith("P"): + return "after", _duration(end_time) + if isinstance(end_time, str): + at = _utc(end_time) + return "at", at + raise ValueError(f"end_time {end_time!r} is neither an RFC 3339 instant with an offset nor a duration like 'P4D'") + + +def _end_instant(end_time, t0): + if end_time is None: + return None + kind, value = _end_spec(end_time) + return value if kind == "at" else t0 + value + + +def _duration(text): + m = re.fullmatch(r"P(?:(\d+)W)?(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?)?", text) + if m is None or not any(m.groups()): + raise ValueError(f"invalid ISO 8601 duration {text!r}") + weeks, days, hours, minutes, seconds = (float(g or 0) for g in m.groups()) + return timedelta(weeks=weeks, days=days, hours=hours, minutes=minutes, seconds=seconds) + + +def _end(turn, reason, **detail): + turn.record({"end": reason, **detail}) + return {"end": reason} + + +def _utc(text): + at = datetime.fromisoformat(text.replace("Z", "+00:00")) + if at.tzinfo is None: + raise ValueError(f"{text!r} has no UTC offset") + return at.astimezone(timezone.utc) + + +def _iso(value): + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/src/agent_env/task_step/task_steps/post_turn_hooks.py b/src/agent_env/task_step/task_steps/post_turn_hooks.py index 7b634eee7..2a0fe3bea 100644 --- a/src/agent_env/task_step/task_steps/post_turn_hooks.py +++ b/src/agent_env/task_step/task_steps/post_turn_hooks.py @@ -24,6 +24,7 @@ from agent_env.env.gateway.constants import ( AGENT_ENV_ROLE_HEADER, EXT_CLOCK_URI, + EXT_NOTIFICATIONS_URI, EXT_STEP_URI, EXT_TRIGGERS_URI, STEP_SOURCE, @@ -33,7 +34,7 @@ _SPEC_KEYS = ("file_artifact_id", "file_artifact_version", "params") _RECORD_TAGS = ("hook", "turn") -_INVOKABLE = (EXT_CLOCK_URI, EXT_TRIGGERS_URI) +_INVOKABLE = (EXT_CLOCK_URI, EXT_TRIGGERS_URI, EXT_NOTIFICATIONS_URI) _TOOL_CALL_TIMEOUT_S = 600 Answer = dict[str, Any] @@ -118,7 +119,8 @@ def close(self) -> None: class HookEnv: - """A deployed env as a hook reaches it: clock and triggers calls, and tool calls recorded as the hook's.""" + """A deployed env as a hook reaches it: clock, triggers and notifications calls, and tool calls recorded as the + hook's.""" def __init__(self, deployed: DeployedEnv, hook_artifact_id: str) -> None: self.env_id = deployed.env_id diff --git a/tst/installer/test_container_journey.py b/tst/installer/test_container_journey.py index 0c475c11d..af104de73 100644 --- a/tst/installer/test_container_journey.py +++ b/tst/installer/test_container_journey.py @@ -133,7 +133,7 @@ def test_a_new_user_installs_agent_env_adds_plugins_runs_a_task_and_removes_them container.ok("python", "-m", "pip", "install", "-q", "--no-index", "--find-links", "/wheels", "uv") container.ok("uv", "tool", "install", "agentenv-framework") rows = container.ok("agent-env", "plugin", "list").stdout.split("STATUS\n", 1)[1] - assert re.fullmatch(r"agentenv-framework +\S+ +6 bundles +ok\n", rows), rows + assert re.fullmatch(r"agentenv-framework +\S+ +7 bundles +ok\n", rows), rows added = container.ok("agent-env", "plugin", "add", browser, grader, "--yes") assert "toy_navigate active" in added.stdout and "toy_grade active" in added.stdout diff --git a/tst/integration/task_step/test_advance_to_next_event_hook_local.py b/tst/integration/task_step/test_advance_to_next_event_hook_local.py new file mode 100644 index 000000000..05f11c39a --- /dev/null +++ b/tst/integration/task_step/test_advance_to_next_event_hook_local.py @@ -0,0 +1,82 @@ +"""The advance-to-next-event example hook on a real gateway: a world event lands silently, and the hook moves the clock only +to the agent's own check and a notify action's notification, which prompt_agent delivers, each labelled with its time +and source.""" + +import json +import uuid +from pathlib import Path + +import httpx +import pytest + +import agent_env +from agent_env.a2a_agent import conversation_store +from agent_env.artifact import FileArtifact +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.configure_scheduled_checks import ConfigureScheduledChecksTaskStep +from agent_env.task_step.task_steps.prompt_agent import 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] + +HOOK = Path(agent_env.__file__).parent / "examples" / "advance-to-next-event" / "artifacts" / "advance-to-next-event" / "hook.py" +MONDAY = "2026-03-02T13:00:00Z" + + +@pytest.mark.asyncio +async def test_only_the_agent_s_check_and_a_declared_notification_wake_it(local_registry): + _seed_bootstrap_envs() + env, deployed, _ = await _deploy_items("wakeups") + agent = put_test_agent(f"wakeups-agent-{uuid.uuid4().hex[:8]}") + hook = FileArtifact.put(f"wakeups-hook-{uuid.uuid4().hex[:8]}", description="advance-to-next-event", file_path=str(HOOK)) + context = TaskStepContext(deployed_envs=[deployed], deployed_agents=[], metadata={}) + triggers = [ + {"id": "w-1", "when": {"type": "time", "at": "2026-03-03T17:00:00Z"}, + "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "rug listed"}}]}, + {"id": "w-2", "when": {"type": "time", "at": "2026-03-04T16:00:00Z"}, + "actions": [{"type": "notify", "channel": "Messages", "text": "Jamie: did the rug thing get sorted?"}]}, + ] + try: + await SyncEnvClockTaskStep(id="clock", version=None, env_id=env.id, virtual_time=MONDAY, + virtual_seconds_per_real_second=0).execute(context) + await ConfigureScheduledChecksTaskStep(id="checks", version=None, env_id=env.id, + roles=["default"]).execute(context) + await RegisterEnvTriggersStep(id="world", version=None, env_id=env.id, triggers=triggers, + watch_roles=["default"]).execute(context) + async with httpx.AsyncClient(base_url=deployed.gateway_url, timeout=60) as client: + # what the agent would do in its first turn: schedule a check of its own, through the tool + made = await client.post("/step", headers={AGENT_ENV_ROLE_HEADER: "default"}, json={ + "action": "call_tool", "tool_name": "schedule_check", + "arguments": {"note": "look for new rugs", "at": "2026-03-04T02:00:00Z"}}) + assert made.status_code == 200 and not made.json().get("isError"), made.text + await _deploy(agent, "solver").execute(context) + await PromptAgentTaskStep( + id="watch", version=None, prompt="Keep an eye on things.", agent_name="solver", max_conversation_turns=6, + poll_interval_seconds=1, time_zone="America/Los_Angeles", + post_turn_hooks=[{"file_artifact_id": hook.id, "file_artifact_version": hook.version, + "params": {"end_time": "P5D"}}], + ).execute(context) + async with httpx.AsyncClient(base_url=deployed.gateway_url, timeout=60) as client: + stored = await client.post("/step", json={"action": "call_tool", "tool_name": "list_items", "arguments": {}}, + headers={AGENT_ENV_ROLE_HEADER: "harness"}) + state = (await client.get("/notifications/state")).raise_for_status().json() + finally: + await env.close() + + messages = conversation_store.get_conversation(context.metadata["a2a_conversations"]["watch"])["messages"] + assert [[p["text"] for p in m["parts"]] for m in messages if m["role"] == "user"] == [ + ["Keep an eye on things."], + ["[Tue Mar 3, 6:00 PM · your scheduled check] look for new rugs"], + ["[Wed Mar 4, 8:00 AM · Messages] Jamie: did the rug thing get sorted?"], + ] + assert "rug listed" in json.loads(stored.json()["content"][0]["text"])["items"] + assert state["pending_notifications"] == [] and state["next_due_at"] is None + records = context.metadata["post_turn_hooks"]["watch"] + assert [r.get("for") for r in records] == [["scheduled_check"], ["trigger"], None] + assert records[-1]["end"] == "nothing_can_wake_the_agent" diff --git a/tst/unit/bundle/installed_test.py b/tst/unit/bundle/installed_test.py index 0bf4e629d..191687a43 100644 --- a/tst/unit/bundle/installed_test.py +++ b/tst/unit/bundle/installed_test.py @@ -127,8 +127,9 @@ def test_the_plugin_report_counts_the_bundles_agent_env_ships(monkeypatch, quiet assert result.exit_code == 0, result.output (core,) = json.loads(result.stdout)["plugins"] assert [(c["group"], c["name"], c["status"]) for c in core["contributions"]] == [ - (BUNDLES, "agents", "active"), (BUNDLES, "drivers", "active"), (BUNDLES, "echo-mcp", "active"), - (BUNDLES, "hello", "active"), (BUNDLES, "mail-mcp", "active"), (BUNDLES, "next-env-event", "active")] + (BUNDLES, "advance-to-next-event", "active"), (BUNDLES, "agents", "active"), (BUNDLES, "drivers", "active"), + (BUNDLES, "echo-mcp", "active"), (BUNDLES, "hello", "active"), (BUNDLES, "mail-mcp", "active"), + (BUNDLES, "next-env-event", "active")] assert "\n agent-env itself\n" in shown and "requires:" not in shown diff --git a/tst/unit/bundle/shipped_bundles_test.py b/tst/unit/bundle/shipped_bundles_test.py index 058861fce..b2ebb71ff 100644 --- a/tst/unit/bundle/shipped_bundles_test.py +++ b/tst/unit/bundle/shipped_bundles_test.py @@ -26,7 +26,7 @@ def _shipped() -> dict: def test_agentenv_framework_ships_its_bundles(): - assert SHIPPED == ["agents", "drivers", "echo-mcp", "hello", "mail-mcp", "next-env-event"] + assert SHIPPED == ["advance-to-next-event", "agents", "drivers", "echo-mcp", "hello", "mail-mcp", "next-env-event"] @pytest.mark.parametrize("name", SHIPPED) diff --git a/tst/unit/task_step/test_advance_to_next_event_hook.py b/tst/unit/task_step/test_advance_to_next_event_hook.py new file mode 100644 index 000000000..cbc9f74e0 --- /dev/null +++ b/tst/unit/task_step/test_advance_to_next_event_hook.py @@ -0,0 +1,237 @@ +"""The advance-to-next-event example hook, turn by turn, against an in-process gateway's real clock, triggers and +notifications, with what came due delivered as prompt_agent delivers it.""" + +from __future__ import annotations + +import asyncio +import importlib.util +from pathlib import Path + +import httpx +import pytest +from mcp.types import CallToolResult, TextContent + +import agent_env +from agent_env.env.env import DeployedGatewayEnv +from agent_env.env.gateway import gateway as gw_mod +from agent_env.env.gateway.clock import Clock +from agent_env.env.gateway.constants import GATEWAY_EXTENSIONS, WELL_KNOWN_PATH +from agent_env.env.gateway.gateway import Gateway +from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.task_steps.notification_delivery import delivered_parts, label_zone, next_message, take_due +from agent_env.task_step.task_steps.post_turn_hooks import PostTurnHook, PostTurnHooks, Turn, _Answer, run_post_turn_hooks + +HOOK = Path(agent_env.__file__).parent / "examples" / "advance-to-next-event" / "artifacts" / "advance-to-next-event" / "hook.py" +GATEWAY = "http://gateway.test" +MONDAY = "2026-03-02T08:00:00Z" + + +@pytest.fixture +def hook(monkeypatch): + spec = importlib.util.spec_from_file_location("advance_to_next_event_hook", HOOK) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + monkeypatch.setattr(module, "POLL_SECONDS", 0.01) + return module + + +@pytest.fixture +def gateway(tmp_path, monkeypatch): + """A gateway on a frozen clock armed Monday 08:00 whose world writes record the virtual time they ran at.""" + 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() + gw._clock.set_time(MONDAY, rate=0) + gw._tool_server_urls = {"items_add_item": "http://items/mcp"} + gw.calls = [] + + async def call_child_tool(server_url, tool_name, arguments, *, caller=None): + gw.calls.append((arguments["item"], gw._clock.read()["virtual_time"])) + return CallToolResult(content=[TextContent(type="text", text="ok")], isError=False) + + gw._call_child_tool = call_child_tool + gw._notifications.configure({"roles": ["default"]}) + real_client = httpx.AsyncClient + transport = httpx.ASGITransport(app=gw._mcp.streamable_http_app()) + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **kw: real_client(*a, **{**kw, "transport": transport})) + yield gw + gw._close_trajectory_file() + + +def _context() -> TaskStepContext: + card = {"capabilities": {"extensions": list(GATEWAY_EXTENSIONS)}} + env = DeployedGatewayEnv(env_id="items", env_version=1, gateway_url=GATEWAY, mcp_url="", sandbox_id="sb", + environment_card_url=f"{GATEWAY}{WELL_KNOWN_PATH}", environment_card=card, + metadata={"deploy_step_id": "deploy_items"}) + return TaskStepContext(deployed_envs=[env]) + + +async def _turns(hook, context, params=None, on_turn=None, triggers=None, time_zone=None) -> list: + """What prompt_agent sends the agent after each turn until the conversation ends: each message's part texts (what + came due for the solver's role first, then the hook's or a trigger's reply), then the end. ``on_turn(n)`` plays + the agent's turn n first.""" + answers = list(triggers or []) + + async def user_triggers(): + return answers.pop(0) if answers else None + + async def ask_user(): + raise AssertionError("the hook never asks the user side") + + hooks = PostTurnHooks() + hooks.hooks.append(PostTurnHook("advance-to-next-event", None, dict(params or {}), hook.post_turn)) + sent = [] + while not sent or not isinstance(sent[-1], dict): + assert len(sent) < 12, sent + number = len(sent) + 1 + if on_turn: + on_turn(number) + answer = await run_post_turn_hooks(hooks, context=context, step_id="week", number=number, last=False, + agent_reply="ok", agent_parts=[{"kind": "text", "text": "ok"}], + ask_user=ask_user, user_triggers=user_triggers) + if "end" in answer: + sent.append(answer) + continue + due = await take_due(context.deployed_envs, "default") + message = next_message(delivered_parts(due, label_zone(time_zone)), answer["message"]) + sent.append([p["text"] for p in message] if message else {"end": "nothing_to_send"}) + return sent + + +def _turn(hook, context, number=1) -> Turn: + async def never(): + raise AssertionError("not asked") + + async def no_trigger(): + return None + + return Turn(context=context, step_id="week", hook=PostTurnHook("advance-to-next-event", None, {}, hook.post_turn), + answer=_Answer(never, no_trigger), number=number, last=False, agent_reply="ok", + agent_parts=[{"kind": "text", "text": "ok"}]) + + +def _register(gw: Gateway, *triggers: dict) -> None: + gw._trigger_engine.register({"watch_roles": ["default"], "triggers": list(triggers)}) + + +@pytest.mark.asyncio +async def test_the_hook_moves_the_clock_to_what_is_due_and_leaves_it_for_prompt_agent_to_take(hook, gateway): + gateway._notifications.schedule("check for new rugs", at="2026-03-03T18:00:00Z") + + assert await hook.post_turn(_turn(hook, _context()), {}) == {"message": []} + + assert gateway._clock.read()["virtual_time"] == "2026-03-03T18:00:00Z" + assert gateway._notifications.state("default")["next_due_at"] == "2026-03-03T18:00:00Z" + + +@pytest.mark.asyncio +async def test_a_world_event_with_no_channel_happens_silently_and_the_agent_s_own_check_wakes_it(hook, gateway): + _register(gateway, {"id": "w-1", "when": {"type": "time", "at": "2026-03-03T09:00:00Z"}, + "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "rug listed"}}]}) + + def agent(n): + if n == 1: + gateway._notifications.schedule("check for new rugs", at="2026-03-03T18:00:00Z") + + sent = await _turns(hook, _context(), on_turn=agent) + + assert sent == [["[Tue Mar 3, 6:00 PM · your scheduled check] check for new rugs"], + {"end": "nothing_can_wake_the_agent"}] + assert gateway.calls == [("rug listed", "2026-03-03T09:00:00Z")] + + +@pytest.mark.asyncio +async def test_a_notify_action_wakes_the_agent_when_it_fires_with_its_channel(hook, gateway): + _register(gateway, {"id": "w-2", "when": {"type": "time", "at": "2026-03-03T09:00:00Z"}, + "actions": [{"type": "notify", "channel": "Curbfinch alert", + "text": 'New for "rug 8x10": 1970s wool rug · $360'}]}) + sent = await _turns(hook, _context(), time_zone="America/Los_Angeles") + assert sent == [['[Tue Mar 3, 1:00 AM · Curbfinch alert] New for "rug 8x10": 1970s wool rug · $360'], + {"end": "nothing_can_wake_the_agent"}] + + +@pytest.mark.asyncio +async def test_only_what_is_due_for_the_agent_s_role_wakes_it(hook, gateway): + _register(gateway, {"id": "landlord", "when": {"type": "time", "at": "2026-03-03T09:00:00Z"}, + "actions": [{"type": "notify", "channel": "Gmail", "to": "landlord", "text": "rent is due"}]}) + gateway._notifications.schedule("check for new rugs", at="2026-03-04T09:00:00Z") + + sent = await _turns(hook, _context()) + + assert sent == [["[Wed Mar 4, 9:00 AM · your scheduled check] check for new rugs"], + {"end": "nothing_can_wake_the_agent"}] + assert [n["to"] for n in gateway._notifications.state()["pending_notifications"]] == ["landlord"] + + +@pytest.mark.asyncio +async def test_the_person_s_messages_arrive_at_their_times_once_each(hook, gateway): + params = {"person_messages": [{"at": "2026-03-04T16:30:00Z", "text": "got Leo, we're home"}, + {"at": "2026-03-03T15:00:00Z", "text": "I'm up"}]} + sent = await _turns(hook, _context(), params=params) + assert sent == [["[Tue Mar 3, 3:00 PM] I'm up"], + ["[Wed Mar 4, 4:30 PM] got Leo, we're home"], + {"end": "nothing_can_wake_the_agent"}] + + +@pytest.mark.asyncio +async def test_what_comes_due_together_goes_in_one_message_oldest_first(hook, gateway): + _register(gateway, {"id": "w-3", "when": {"type": "time", "at": "2026-03-03T09:00:00Z"}, + "actions": [{"type": "notify", "channel": "Gmail", "text": "School: early dismissal"}]}) + + def agent(n): + if n == 1: + gateway._notifications.schedule("morning sweep", at="2026-03-03T09:00:00Z") + + [first, end] = await _turns(hook, _context(), on_turn=agent, + params={"person_messages": [{"at": "2026-03-03T09:00:00Z", "text": "pick up Leo?"}]}) + assert first == ["[Tue Mar 3, 9:00 AM · your scheduled check] morning sweep", + "[Tue Mar 3, 9:00 AM · Gmail] School: early dismissal", + "[Tue Mar 3, 9:00 AM] pick up Leo?"] + + +@pytest.mark.asyncio +async def test_it_ends_at_end_time_without_waking_for_what_lies_beyond(hook, gateway): + gateway._notifications.schedule("next week", at="2026-03-10T09:00:00Z") + assert await _turns(hook, _context(), params={"end_time": "P5D"}) == [{"end": "end_time"}] + + +@pytest.mark.asyncio +async def test_a_state_check_the_agents_turn_set_off_finishes_before_the_clock_jumps(hook, gateway, monkeypatch): + async def slow_check(check, ctx=None, role=None): + await asyncio.sleep(0.3) + return True + + monkeypatch.setattr(gateway._trigger_engine, "_run_check", slow_check) + _register(gateway, {"id": "seen", "when": {"type": "state", "check": {"tool": "items_list", "predicate": {"exists": True}}}, + "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "seen"}}]}, + {"id": "tue", "when": {"type": "time", "at": "2026-03-03T09:00:00Z"}, + "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "tue"}}]}) + gateway._trigger_engine.on_tool_call("default", "send_email", {}, CallToolResult(content=[], isError=False)) + await _turns(hook, _context()) + assert gateway.calls[0] == ("seen", MONDAY) + + +@pytest.mark.asyncio +async def test_a_firing_agent_trigger_answers_before_any_jump(hook, gateway): + gateway._notifications.schedule("later", at="2026-03-03T09:00:00Z") + pushback = {"message": [{"kind": "text", "text": "the person pushes back"}]} + sent = await _turns(hook, _context(), triggers=[pushback]) + assert sent[0] == ["the person pushes back"] + assert sent[1] == ["[Tue Mar 3, 9:00 AM · your scheduled check] later"] + + +@pytest.mark.asyncio +async def test_a_repeating_check_wakes_once_per_time_and_the_record_names_why(hook, gateway): + gateway._notifications.schedule("sweep", every="P1D", until="P2D") + context = _context() + sent = await _turns(hook, context) + assert sent == [["[Tue Mar 3, 8:00 AM · your scheduled check] sweep"], + ["[Wed Mar 4, 8:00 AM · your scheduled check] sweep"], + {"end": "nothing_can_wake_the_agent"}] + records = context.metadata["post_turn_hooks"]["week"] + assert [r.get("for") for r in records[:2]] == [["scheduled_check"], ["scheduled_check"]] + + +def test_unknown_params_are_refused(hook): + with pytest.raises(ValueError, match="takes"): + hook._params({"nudge": "x"}) diff --git a/tst/unit/test_clean_install_wheel.py b/tst/unit/test_clean_install_wheel.py index 45afc9cc1..1993dd318 100644 --- a/tst/unit/test_clean_install_wheel.py +++ b/tst/unit/test_clean_install_wheel.py @@ -62,7 +62,8 @@ def test_bundle_entry_points_that_differ_from_pyproject_are_named(tmp_path, entr def test_the_gate_checks_the_folder_of_every_shipped_bundle(): roots = {name: root.relative_to(SCRIPT.parents[2]).as_posix() for name, root in clean_install.bundle_roots().items()} - assert roots == {"agents": "src/agent_env/agents", "drivers": "src/agent_env/drivers", + assert roots == {"advance-to-next-event": "src/agent_env/examples/advance-to-next-event", + "agents": "src/agent_env/agents", "drivers": "src/agent_env/drivers", "echo-mcp": "src/agent_env/examples/echo-mcp", "hello": "src/agent_env/examples/hello", "mail-mcp": "src/agent_env/examples/mail-mcp", "next-env-event": "src/agent_env/examples/next-env-event"}