feat(gateway): wakeups — checks an agent schedules for itself, and notify_agent

src/agent_env/env/gateway/constants.py

@@ -78,6 +78,7 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int:
 EXT_STEP_URI = "urn:agentenv:step/v1"
 EXT_STATE_URI = "urn:agentenv:state/v1"
 EXT_CHANGES_URI = "urn:agentenv:changes/v1"
+EXT_WAKEUPS_URI = "urn:agentenv:wakeups/v1"
 
 # Rows per GET /changes page: the default, and the most a client may ask for.
 CHANGES_PAGE_DEFAULT = 1000
@@ -122,8 +123,10 @@ def sources(names: list[str]) -> str:
 TRIGGER_WEBSITE = "website"
 TRIGGER_EXECUTOR_HOURS_TIME_ZONE = "executor_hours_time_zone"
 TRIGGER_NEXT_EVENT_AT = "next_event_at"
+TRIGGER_NOTIFY_AGENT = "notify_agent"
 TRIGGER_FEATURES = (TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES, TRIGGER_ACTION_DELAY, TRIGGER_EXECUTOR_HOURS,
-                    TRIGGER_UNLESS, TRIGGER_WEBSITE, TRIGGER_EXECUTOR_HOURS_TIME_ZONE, TRIGGER_NEXT_EVENT_AT)
+                    TRIGGER_UNLESS, TRIGGER_WEBSITE, TRIGGER_EXECUTOR_HOURS_TIME_ZONE, TRIGGER_NEXT_EVENT_AT,
+                    TRIGGER_NOTIFY_AGENT)
 
 # The one `source` a /step call may name instead of step_api, with the id of the hook artifact that made it.
 STEP_SOURCE_POST_TURN_HOOK = "post_turn_hook"
@@ -244,4 +247,17 @@ def random_mcp_server_name() -> str:
                 "limit": {"type": "integer", "minimum": 1, "maximum": CHANGES_PAGE_MAX}}}}}},
         "required": False,
     },
+    {
+        "uri": EXT_WAKEUPS_URI,
+        "description": "Wakeups (urn:agentenv:wakeups/v1): what is due to wake the agent between its turns, on the virtual clock: checks it scheduled for itself (schedule_check, which config turns on) and notifications triggers sent it (the notify_agent action). state says when the next is due; take hands out each due one once (harness control path).",
+        "params": {"endpoint": "/wakeups/state", "methods": {
+            "config": {"method": "POST", "endpoint": "/wakeups/config", "request": {"type": "object", "properties": {
+                "scheduled_checks": {"type": "boolean"}, "max_active": {"type": "integer", "minimum": 1},
+                "min_interval": {"type": "string"}}}},
+            "state": {"method": "GET", "endpoint": "/wakeups/state", "request": {"type": "object", "properties": {}}},
+            "take": {"method": "POST", "endpoint": "/wakeups/take", "request": {"type": "object", "properties": {
+                "until": {"type": "string"}}}},
+            "clear": {"method": "POST", "endpoint": "/wakeups/clear", "request": {"type": "object", "properties": {}}}}},
+        "required": False,
+    },
 ]

src/agent_env/env/gateway/gateway.py

@@ -73,6 +73,7 @@
 from .get_time import GET_TIME_TOOL_NAME, build_get_time_tool
 from .rest_proxy import ROLE_HEADER_BYTES, UpstreamResponse, fetch_child_card, forward, rewrite_child_card
 from .triggers import TriggerEngine, TriggerError
+from .wakeups import WAKEUP_TOOL_NAMES, WakeupError, Wakeups, build_wakeup_tools
 
 logger = logging.getLogger(__name__)
 
@@ -173,6 +174,7 @@ async def list_website_urls() -> list[str]:
                 )
             )
         self._clock = Clock()
+        self._wakeups = Wakeups(lambda: self._clock)
         self._env_get_time_url_cache: str | None = None
         self._install_role_filters()
 
@@ -314,6 +316,47 @@ async def clock_state(request: Request) -> Response:
                 st["env_get_time_url"] = await self._env_get_time_url()
             return self._json_response(st, 200)
 
+        async def _wakeup_write(request: Request, method) -> Response:
+            try:
+                body = await request.json()
+            except Exception:
+                return self._json_response({"ok": False, "error": "Invalid JSON body"}, 400)
+            try:
+                return self._json_response(method(body), 200)
+            except WakeupError as e:
+                return self._json_response({"ok": False, "error": str(e)}, 400)
+
+        @self._mcp.custom_route("/wakeups/config", methods=["POST"])
+        async def wakeups_config(request: Request) -> Response:
+            try:  # the backing servers' tools must be known, or a later discovery would replace ours unseen
+                await self._ensure_tools_discovered()
+            except Exception as e:
+                return self._json_response(
+                    {"ok": False, "error": f"tool discovery has not completed ({type(e).__name__}); retry"}, 503)
+            taken = sorted(set(WAKEUP_TOOL_NAMES) & set(self._tool_server_urls))
+            if taken:
+                return self._json_response(
+                    {"ok": False, "error": f"{taken} already registered by a backing server"}, 409)
+            response = await _wakeup_write(request, self._wakeups.configure)
+            if self._wakeups.enabled:
+                self._register_wakeup_tools()
+            else:
+                self._unregister_wakeup_tools()
+            return response
+
+        @self._mcp.custom_route("/wakeups/state", methods=["GET"])
+        async def wakeups_state(request: Request) -> Response:
+            return self._json_response(self._wakeups.state(), 200)
+
+        @self._mcp.custom_route("/wakeups/take", methods=["POST"])
+        async def wakeups_take(request: Request) -> Response:
+            return await _wakeup_write(request, self._wakeups.take)
+
+        @self._mcp.custom_route("/wakeups/clear", methods=["POST"])
+        async def wakeups_clear(request: Request) -> Response:
+            self._unregister_wakeup_tools()
+            return self._json_response(self._wakeups.clear(), 200)
+
         @self._mcp.custom_route("/trajectory", methods=["GET"])
         async def env_trajectory(request: Request) -> Response:
             return self._serve_trajectory()
@@ -901,9 +944,11 @@ async def _discover_and_register_tools(self, sessions: dict[str, ClientSession])
                 raise
             logger.info(f"Found {len(tools_result.tools)} tool(s) from {server.name}")
             listed.append((server, tools_result.tools))
-        # get_time is left out: a backing server's tool of that name wins, and the clock refuses to arm over it
+        # get_time and the wakeup tools are left out: a backing server's tool of that name wins, and the clock and
+        # wakeups refuse to turn on over it
         sources: dict[str, list[str]] = {
-            tool.name: ["gateway"] for tool in self._server_tools.get("gateway", []) if tool.name != GET_TIME_TOOL_NAME}
+            tool.name: ["gateway"] for tool in self._server_tools.get("gateway", [])
+            if tool.name != GET_TIME_TOOL_NAME and tool.name not in WAKEUP_TOOL_NAMES}
         for server, tools in listed:
             for tool in tools:
                 sources.setdefault(tool.name, []).append(server.name)
@@ -923,6 +968,8 @@ async def _discover_and_register_tools(self, sessions: dict[str, ClientSession])
         self._tools_discovered = True
         if self._mirror_get_time_descriptor():
             logger.info(f"Re-mirrored '{GET_TIME_TOOL_NAME}' descriptor after tool discovery")
+        if self._mirror_wakeup_descriptors():
+            logger.info("Re-mirrored the wakeup tool descriptors after tool discovery")
         # list_website_urls sits in _server_tools from construction, so a memo may predate the backing tools
         self._trigger_engine.invalidate_readonly_cache()
 
@@ -1148,6 +1195,46 @@ def _unregister_get_time_tool(self) -> None:
             self._server_tools.pop("gateway", None)
         self._trigger_engine.invalidate_readonly_cache()
 
+    def _register_wakeup_tools(self) -> None:
+        """Expose schedule_check and its siblings (idempotent). Called when a task turns scheduled checks on."""
+        tools = self._mcp._tool_manager._tools
+        for tool in build_wakeup_tools(self._wakeups, self._create_arg_model_from_schema):
+            if tool.name not in self._tool_server_urls:
+                tools.setdefault(tool.name, tool)
+        if self._mirror_wakeup_descriptors():
+            self._trigger_engine.invalidate_readonly_cache()
+
+    def _mirror_wakeup_descriptors(self) -> bool:
+        """Mirror the registered wakeup tools into `_server_tools`, as `_mirror_get_time_descriptor` does for
+        get_time. Returns whether it appended."""
+        if not self._tools_discovered or not self._wakeups.enabled:
+            return False
+        mirrored = self._server_tools.setdefault("gateway", [])
+        present = {t.name for t in mirrored}
+        added = False
+        for name in WAKEUP_TOOL_NAMES:
+            tool = self._mcp._tool_manager._tools.get(name)
+            if tool is None or name in present or name in self._tool_server_urls:
+                continue
+            mirrored.append(MCPTool(name=tool.name, description=tool.description,
+                                    inputSchema=tool.parameters, annotations=tool.annotations))
+            added = True
+        return added
+
+    def _unregister_wakeup_tools(self) -> None:
+        """Remove them when a task turns scheduled checks off or clears (idempotent); a backing server's are left."""
+        tools = self._mcp._tool_manager._tools
+        for name in WAKEUP_TOOL_NAMES:
+            if name not in self._tool_server_urls:
+                tools.pop(name, None)
+        remaining = [t for t in self._server_tools.get("gateway", []) if t.name not in WAKEUP_TOOL_NAMES
+                     or t.name in self._tool_server_urls]
+        if remaining:
+            self._server_tools["gateway"] = remaining
+        else:
+            self._server_tools.pop("gateway", None)
+        self._trigger_engine.invalidate_readonly_cache()
+
     def _create_and_register_proxy_tool(self, server_url: str, tool: MCPTool) -> None:
         self._mcp._tool_manager._tools[tool.name] = self._proxy_tool(server_url, tool)
 

src/agent_env/env/gateway/triggers.py

@@ -23,6 +23,7 @@
 from mcp.types import CallToolResult, TextContent
 
 from .clock import ClockError, _advance, _iso, _parse_duration, _parse_rfc3339, _zone
+from .wakeups import validate_channel
 from .caller import Caller
 from .constants import DEFAULT_ROLE, TRIGGER_IN_FLIGHT_STATUSES, TRIGGER_STATUSES
 
@@ -34,7 +35,7 @@
 _CTX_ROOTS = {"action": ("args", "result"), "website": ("args",)}
 # RFC 9110's safe methods: a request using one asks for no change, so it prompts no state re-evaluation.
 _SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE"})
-_ACTION_TYPES = ("nl", "tool", "permission")
+_ACTION_TYPES = ("nl", "tool", "permission", "notify_agent")
 _TERMINAL_A2A_STATES = ("completed", "failed", "canceled", "rejected")
 # Bounds the per-evaluation catch-up burst; the backlog drains on later evaluations.
 _MAX_CATCHUP = 1000
@@ -742,6 +743,8 @@ def _normalize_action(action: Any, w: str, roots: tuple[str, ...], executors: di
             raise TriggerError(f"{w}: 'as' is only valid on tool actions")
         if "executor" in action and atype != "nl":
             raise TriggerError(f"{w}: 'executor' is only valid on nl actions")
+        if "verify" in action and atype == "notify_agent":
+            raise TriggerError(f"{w}: a notify_agent action has nothing to verify")
         if action.get("delay") is None:
             action.pop("delay", None)
         else:
@@ -761,6 +764,14 @@ def _normalize_action(action: Any, w: str, roots: tuple[str, ...], executors: di
             elif not isinstance(named, str) or named not in executors:
                 raise TriggerError(f"{w}: executor {named!r} is not a registered executor name "
                                    f"(registered: {sorted(n for n in executors if n is not None)})")
+        elif atype == "notify_agent":
+            try:
+                validate_channel(action.get("channel"), w)
+            except ValueError as e:
+                raise TriggerError(str(e))
+            if not isinstance(action.get("text"), str) or not action["text"].strip():
+                raise TriggerError(f"{w}: text must be the notification's non-empty text")
+            _validate_placeholders(action["text"], f"{w}.text", roots)
         elif atype == "tool":
             if not isinstance(action.get("tool"), str) or not action["tool"]:
                 raise TriggerError(f"{w}: tool must be a non-empty string")
@@ -1362,6 +1373,15 @@ async def _run_action(self, tid: str, trig: dict, index: int, action: dict, ctx:
                                tool=action["tool"], args=_echo_args(args), **acting)
                     return False
                 detail = {"tool": action["tool"], "args": _echo_args(args), **acting}
+            elif action["type"] == "notify_agent":
+                try:
+                    text = _template(action["text"], {}, ctx)
+                except TemplateError as e:
+                    self._emit("action_failed", tid, action_index=index, detail=str(e)[:300])
+                    return False
+                note = self._gw._wakeups.notify(action["channel"], str(text), trigger_id=tid)
+                # the text may carry what a person wrote, so the log keeps its size, not its words
+                detail = {"notify_agent": action["channel"], "notification_id": note["id"], "text_len": len(str(text))}
             else:
                 if not await self._run_nl(tid, trig, index, action, ctx):
                     return False