feat(examples): agent-wakeups hook — wake the agent only for its own checks, notifications and the person

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"
+agent-wakeups = "agent_env.examples"
 
 [project.urls]
 Homepage = "https://www.agentenvframework.com"

src/agent_env/examples/agent-wakeups/README.md

@@ -0,0 +1,42 @@
+A prompt_agent post-turn hook that runs one conversation over simulated days and wakes the agent only for what
+would really reach it.
+
+`artifacts/agent-wakeups/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 jumps the clock, silently, to the next thing that happens: a
+world event (a trigger's mark, a delay ending, an executor's hours opening), a check the agent scheduled for itself,
+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` (turned on by a `configure_wakeups` step),
+- a notification a trigger sent it with a `notify_agent` action, through a channel the task declares,
+- a message from `person_messages`.
+
+Everything due goes into one message, each line saying when and through what, so nothing poses as the person:
+
+```
+[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_wakeups`:
+
+```json
+{"id": "week", "type": "prompt_agent", "prompt": "Keep an eye on curbfinch this week for an 8x10 rug.",
+ "max_conversation_turns": 20,
+ "post_turn_hooks": [{"file_artifact_id": "@local/agentenv-framework/agent-wakeups/agent-wakeups",
+                      "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`).
+- `time_zone`: the IANA zone the wake lines show 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 wakeups/v1. A hook runs
+in the control-plane process with the trust of an `env_outcome_verifier` script.

src/agent_env/examples/agent-wakeups/artifacts/agent-wakeups/hook.py

@@ -0,0 +1,213 @@
+"""A prompt_agent post-turn hook that runs a conversation over simulated days, waking the agent only for what would
+really reach it: a check it scheduled for itself, a notification a trigger sent it (notify_agent), or a message the
+person writes at a set time. Everything else the world does happens silently while the agent is away."""
+import asyncio
+import logging
+import re
+import time
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+CLOCK = "urn:agentenv:clock/v1"
+TRIGGERS = "urn:agentenv:triggers/v1"
+WAKEUPS = "urn:agentenv:wakeups/v1"
+IN_FLIGHT = ("firing", "queued")
+PARAMS = ("end_time", "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 jump, silently, to the next thing that would wake the agent."""
+    if turn.last:
+        return None
+    end_time, zone, person, settle_timeout, max_jumps = _params(params)
+    envs = [env for env in turn.envs if env.supports(CLOCK, "advance") and env.supports(WAKEUPS, "take")]
+    if not envs:
+        return _end(turn, "no_env_with_clock_and_wakeups")
+    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)
+        items = await _take(envs, now)
+        due_person = [m for m in person if m["at"] <= now and _iso(m["at"]) not in sent]
+        if items or due_person:
+            turn.record({"woke_at": _iso(now), "for": [i["kind"] for i in items] + ["person"] * len(due_person),
+                         **({"person_at": [_iso(m["at"]) for m in due_person]} if due_person else {})})
+            return {"message": _message(items, due_person, zone)}
+        nxt = await _next(envs, 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"agent-wakeups takes {list(PARAMS)}, not {unknown}")
+    end_time = params.get("end_time")
+    if end_time is not None:
+        _end_spec(end_time)
+    zone_name = params.get("time_zone", "UTC")
+    try:
+        zone = ZoneInfo(zone_name)
+    except Exception as e:
+        raise ValueError(f"time_zone must be an IANA name such as 'America/Los_Angeles': {e!r}") from e
+    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, 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 _take(envs, now):
+    items = []
+    for env in envs:
+        items += (await env.invoke(WAKEUPS, "take", {"until": _iso(now)})).get("items", [])
+    return sorted(items, key=lambda i: i.get("at") or "")
+
+
+async def _next(envs, person, now, sent):
+    """The earliest instant after now at which anything happens: a world event (silent), a scheduled check, a
+    notification, 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:
+        wake = await env.invoke(WAKEUPS, "state")
+        if wake.get("next_check_at"):
+            candidates.append(_utc(wake["next_check_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 _message(items, person, zone):
+    """One line per thing, oldest first, each saying when and through what, so nothing poses as the person."""
+    lines = []
+    for item in items:
+        when = _local(item.get("at"), zone)
+        if item["kind"] == "scheduled_check":
+            again = f" (it came due {item['times']} times while you were away)" if item.get("times", 1) > 1 else ""
+            lines.append((item.get("at") or "", f"[{when} · your scheduled check] {item['note']}{again}"))
+        else:
+            lines.append((item.get("at") or "", f"[{when} · {item['channel']}] {item['text']}"))
+    for m in person:
+        lines.append((_iso(m["at"]), f"[{_local(_iso(m['at']), zone)}] {m['text']}"))
+    return "\n".join(text for _, text in sorted(lines))
+
+
+def _local(at, zone):
+    if not at:
+        return "now"
+    local = _utc(at).astimezone(zone)
+    return f"{local.strftime('%a %b')} {local.day}, {local.strftime('%I:%M %p').lstrip('0')}"
+
+
+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")

src/agent_env/task_step/task_steps/post_turn_hooks.py

@@ -26,14 +26,15 @@
     EXT_CLOCK_URI,
     EXT_STEP_URI,
     EXT_TRIGGERS_URI,
+    EXT_WAKEUPS_URI,
     STEP_SOURCE,
     STEP_SOURCE_POST_TURN_HOOK,
 )
 from agent_env.task_step.context import TaskStepContext, _strip_redacted_keys
 
 _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_WAKEUPS_URI)
 _TOOL_CALL_TIMEOUT_S = 600
 
 Answer = dict[str, Any]
@@ -118,7 +119,7 @@ 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 wakeups 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

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

tst/integration/task_step/test_agent_wakeups_hook_local.py

@@ -0,0 +1,80 @@
+"""The agent-wakeups example hook on a real gateway: a world event lands silently, and only the agent's own check and
+a notify_agent notification wake it, 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_wakeups import ConfigureWakeupsTaskStep
+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" / "agent-wakeups" / "artifacts" / "agent-wakeups" / "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="agent-wakeups", 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_agent", "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 ConfigureWakeupsTaskStep(id="wakeups", version=None, env_id=env.id).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,
+            post_turn_hooks=[{"file_artifact_id": hook.id, "file_artifact_version": hook.version,
+                              "params": {"end_time": "P5D", "time_zone": "America/Los_Angeles"}}],
+        ).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"})
+            wake = (await client.get("/wakeups/state")).raise_for_status().json()
+    finally:
+        await env.close()
+
+    messages = conversation_store.get_conversation(context.metadata["a2a_conversations"]["watch"])["messages"]
+    assert [m["parts"][0]["text"] 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 wake["pending_notifications"] == [] and wake["next_check_at"] is None
+    records = context.metadata["post_turn_hooks"]["watch"]
+    assert [r.get("for") for r in records] == [["scheduled_check"], ["notification"], None]
+    assert records[-1]["end"] == "nothing_can_wake_the_agent"

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, "agent-wakeups", "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