feat(always_on): a step that runs an assistant over days of task time, with a simulated user (FD-5364)

src/agent_env/env/gateway/clock.py

@@ -186,6 +186,21 @@ def advance(self, virtual_time: str) -> dict:
             self._lived += jump
         return self.state()
 
+    def set_rate(self, rate) -> dict:
+        """Change the rate on the current arm from now on, keeping the virtual time, t0 and generation, so time
+        triggers keep their marks (a re-arm with set_time would re-anchor them); 0 freezes the clock."""
+        r = _parse_rate(rate)
+        with self._lock:
+            if not self._armed or self._t0 is None:
+                raise ClockError("clock is not armed")
+            current = _advance(self._t0, (self._now() - self._anchor) * self._rate + self._ahead)
+            if current > _MAX_T0 - timedelta(seconds=_OVERFLOW_HORIZON_S * r):
+                raise ClockError(f"rate {r:g} would overflow datetime near the max from {_iso(current)}")
+            self._bank()
+            self._ahead = (current - self._t0).total_seconds()
+            self._rate = r
+        return self.state()
+
     def clear(self) -> dict:
         """Disarm (idempotent). Afterwards reads 404."""
         with self._lock:

src/agent_env/env/gateway/constants.py

@@ -79,6 +79,7 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int:
 EXT_STATE_URI = "urn:agentenv:state/v1"
 EXT_CHANGES_URI = "urn:agentenv:changes/v1"
 EXT_NOTIFICATIONS_URI = "urn:agentenv:notifications/v1"
+EXT_USER_CHAT_URI = "urn:agentenv:user-chat/v1"
 EXT_WEBSITE_URI = "urn:agentenv:website/v1"
 EXT_BROWSER_URI = "urn:agentenv:browser/v1"
 EXT_COMPUTER_APP_URI = "urn:agentenv:computer-app/v1"
@@ -130,9 +131,11 @@ def sources(names: list[str]) -> str:
 TRIGGER_NEXT_EVENT_AT = "next_event_at"
 TRIGGER_NOTIFY = "notify"
 TRIGGER_WEBSITE_OP = "website_op"
+# state reports each fire in flight (running, waits) and each executor call in progress (executor_calls).
+TRIGGER_FIRES_IN_STATE = "fires_in_state"
 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_NOTIFY, TRIGGER_WEBSITE_OP)
+                    TRIGGER_NOTIFY, TRIGGER_WEBSITE_OP, TRIGGER_FIRES_IN_STATE)
 
 # 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"
@@ -207,21 +210,23 @@ def random_mcp_server_name() -> str:
     },
     {
         "uri": EXT_TRIGGERS_URI,
-        "description": "Register (additive add-by-id), remove, clear, or read (state) typed Action/State triggers + the firing log (harness control path).",
+        "description": "Register (additive add-by-id), remove, clear, or read (state) typed Action/State triggers + the firing log (harness control path). config sets the engine's opt-in settings: concurrent_fires (a repeat trigger whose actions are all contact replies (nl) or notifications (notify) runs its fires side by side instead of queueing) and executor_timeout_seconds (the time limit of an executor that sets none).",
         "params": {"endpoint": "/triggers/register", "features": list(TRIGGER_FEATURES), "methods": {
             "register": {"method": "POST", "endpoint": "/triggers/register", "request": {"type": "object", "properties": {"watch_roles": {"type": "array", "items": {"type": "string"}}, "executor": {"type": "object"}, "executors": {"type": "object"}, "triggers": {"type": "array"}}, "required": ["triggers"]}},
+            "config": {"method": "POST", "endpoint": "/triggers/config", "request": {"type": "object", "properties": {"concurrent_fires": {"type": "boolean"}, "executor_timeout_seconds": {"type": "number", "exclusiveMinimum": 0}}}},
             "remove": {"method": "POST", "endpoint": "/triggers/remove", "request": {"type": "object", "properties": {"ids": {"type": "array", "items": {"type": "string"}}}, "required": ["ids"]}},
             "clear": {"method": "POST", "endpoint": "/triggers/clear", "request": {"type": "object", "properties": {}}},
             "state": {"method": "GET", "endpoint": "/triggers/state", "request": {"type": "object", "properties": {}}}}},
         "required": False,
     },
     {
         "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/advance/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/set_rate/clear are the harness control path (set_rate changes the rate on the current arm, keeping time triggers' marks; 0 freezes the clock).",
         "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"]}},
+            "set_rate": {"method": "PUT", "endpoint": "/clock/rate", "request": {"type": "object", "properties": {"virtual_seconds_per_real_second": {"type": "number", "minimum": 0}}, "required": ["virtual_seconds_per_real_second"]}},
             "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {}}},
             "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}}}},
         "required": False,
@@ -270,4 +275,17 @@ def random_mcp_server_name() -> str:
             "clear": {"method": "POST", "endpoint": "/notifications/clear", "request": {"type": "object", "properties": {}}}}},
         "required": False,
     },
+    {
+        "uri": EXT_USER_CHAT_URI,
+        "description": "User chat (urn:agentenv:user-chat/v1): the message_user tool, for the roles config names, is how an always-on agent reaches its user; each message is queued with its virtual time until the harness takes it. With stay_quiet the roles also get stay_quiet, which sends nothing and is recorded for the next take. Built in: role rules don't hide them from a configured role.",
+        "params": {"endpoint": "/user_chat/state", "methods": {
+            "config": {"method": "POST", "endpoint": "/user_chat/config", "request": {"type": "object", "properties": {
+                "roles": {"type": "array", "items": {"type": "string"}}, "user_name": {"type": "string"},
+                "stay_quiet": {"type": "boolean"}}}},
+            "state": {"method": "GET", "endpoint": "/user_chat/state", "request": {"type": "object", "properties": {}}},
+            "take": {"method": "POST", "endpoint": "/user_chat/take", "request": {"type": "object", "properties": {
+                "take_id": {"type": "string"}, "from": {"type": "string"}}}},
+            "clear": {"method": "POST", "endpoint": "/user_chat/clear", "request": {"type": "object", "properties": {}}}}},
+        "required": False,
+    },
 ]

src/agent_env/env/gateway/gateway.py

@@ -90,6 +90,8 @@
 )
 from .triggers import TriggerEngine, TriggerError
 from .notifications import CHECK_TOOL_NAMES, NotificationError, Notifications, build_check_tools
+from .user_chat import MESSAGE_USER_TOOL_NAME, UserChat, UserChatError, build_user_chat_tool
+from .user_chat import STAY_QUIET_TOOL_NAME, build_stay_quiet_tool
 
 logger = logging.getLogger(__name__)
 
@@ -201,6 +203,7 @@ async def list_website_urls() -> list[str]:
             )
         self._clock = Clock()
         self._notifications = Notifications(lambda: self._clock)
+        self._user_chat = UserChat(lambda: self._clock)
         self._env_get_time_url_cache: str | None = None
         self._install_role_filters()
 
@@ -295,6 +298,10 @@ async def triggers_register(request: Request) -> Response:
         async def triggers_remove(request: Request) -> Response:
             return await _trigger_write(request, self._trigger_engine.remove)
 
+        @self._mcp.custom_route("/triggers/config", methods=["POST"])
+        async def triggers_config(request: Request) -> Response:
+            return await _trigger_write(request, self._trigger_engine.config)
+
         @self._mcp.custom_route("/triggers/clear", methods=["POST"])
         async def triggers_clear(request: Request) -> Response:
             return self._json_response(self._trigger_engine.clear(), 200)
@@ -316,6 +323,7 @@ async def clock_set_time(request: Request) -> Response:
                 state = self._clock.set_time(body.get("virtual_time"), body.get("virtual_seconds_per_real_second"))
             except ClockError as e:
                 return self._json_response({"ok": False, "error": str(e)}, 400)
+            self._trigger_engine.clock_changed()
             self._register_get_time_tool()
             return self._json_response(state, 200)
 
@@ -333,10 +341,26 @@ async def clock_advance(request: Request) -> Response:
                 return self._json_response({"ok": False, "error": str(e)}, 400)
             return self._json_response({**state, "due": self._trigger_engine.clock_advanced()}, 200)
 
+        @self._mcp.custom_route("/clock/rate", methods=["PUT"])
+        async def clock_rate(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.set_rate(body.get("virtual_seconds_per_real_second"))
+            except ClockError as e:
+                return self._json_response({"ok": False, "error": str(e)}, 400)
+            self._trigger_engine.clock_changed()
+            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()
             state = self._clock.clear()
+            self._trigger_engine.clock_changed()
             return self._json_response(state, 200)
 
         @self._mcp.custom_route("/clock/time", methods=["GET"])
@@ -408,6 +432,51 @@ async def notifications_clear(request: Request) -> Response:
             self._unregister_check_tools()
             return self._json_response(self._notifications.clear(), 200)
 
+        @self._mcp.custom_route("/user_chat/config", methods=["POST"])
+        async def user_chat_config(request: Request) -> Response:
+            try:
+                body = await request.json()
+            except Exception:
+                return self._json_response({"ok": False, "error": "Invalid JSON body"}, 400)
+            if isinstance(body, dict) and body.get("roles"):
+                try:
+                    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)
+                if MESSAGE_USER_TOOL_NAME in self._tool_server_urls:
+                    return self._json_response(
+                        {"ok": False, "error": f"'{MESSAGE_USER_TOOL_NAME}' already registered by a backing server"}, 409)
+            try:
+                state = self._user_chat.configure(body)
+            except UserChatError as e:
+                return self._json_response({"ok": False, "error": str(e)}, 400)
+            if self._user_chat.enabled:
+                self._register_user_chat_tool()
+            else:
+                self._unregister_user_chat_tool()
+            return self._json_response(state, 200)
+
+        @self._mcp.custom_route("/user_chat/state", methods=["GET"])
+        async def user_chat_state(request: Request) -> Response:
+            return self._json_response(self._user_chat.state(), 200)
+
+        @self._mcp.custom_route("/user_chat/take", methods=["POST"])
+        async def user_chat_take(request: Request) -> Response:
+            try:
+                body = await request.json()
+            except Exception:
+                body = {}
+            try:
+                return self._json_response(self._user_chat.take(body), 200)
+            except UserChatError as e:
+                return self._json_response({"ok": False, "error": str(e)}, 400)
+
+        @self._mcp.custom_route("/user_chat/clear", methods=["POST"])
+        async def user_chat_clear(request: Request) -> Response:
+            self._unregister_user_chat_tool()
+            return self._json_response(self._user_chat.clear(), 200)
+
         @self._mcp.custom_route("/trajectory", methods=["GET"])
         async def env_trajectory(request: Request) -> Response:
             return self._serve_trajectory()
@@ -524,9 +593,18 @@ def _listen_socket(self, family: int, host: str) -> socket.socket:
     def _is_disabled(self, role: str, name: str) -> bool:
         """Walk the rule chain (role first, then WILDCARD) returning the first concrete True/False; default False.
         A tool listed hidden by default skips "*"-tool rules: only rules naming it apply, and with none it is hidden.
-        The scheduled-check tools exist only for the roles the notifications config names, whatever the rules say."""
+        The scheduled-check tools exist only for the roles the notifications config names, whatever the rules say.
+        A role the user-chat config names always has message_user, get_time and the scheduled-check tools the
+        notifications config gives it, whatever the rules say."""
         if name in CHECK_TOOL_NAMES and name not in self._tool_server_urls and role not in self._notifications.roles:
             return True
+        if name == MESSAGE_USER_TOOL_NAME and name not in self._tool_server_urls:
+            return role not in self._user_chat.roles  # built in: role rules don't hide it from a configured role
+        if (name == GET_TIME_TOOL_NAME or name in CHECK_TOOL_NAMES) and name not in self._tool_server_urls \
+                and role in self._user_chat.roles:
+            return False
+        if name == STAY_QUIET_TOOL_NAME and name not in self._tool_server_urls:
+            return not (self._user_chat.stay_quiet and role in self._user_chat.roles)  # opt-in, like message_user
         hidden = name in self._hidden_by_default
         for r in (role, WILDCARD):
             rules = self._role_rules.get(r, {})
@@ -937,7 +1015,8 @@ async def _discover_and_register_tools(self, sessions: dict[str, ClientSession])
         # clock and the checks 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 and tool.name not in CHECK_TOOL_NAMES}
+            if tool.name != GET_TIME_TOOL_NAME and tool.name not in CHECK_TOOL_NAMES
+            and tool.name != MESSAGE_USER_TOOL_NAME and tool.name != STAY_QUIET_TOOL_NAME}
         for server, tools in listed:
             for tool in tools:
                 sources.setdefault(tool.name, []).append(server.name)
@@ -1231,6 +1310,54 @@ def _unregister_check_tools(self) -> None:
             self._server_tools.pop("gateway", None)
         self._trigger_engine.invalidate_readonly_cache()
 
+    def _register_user_chat_tool(self) -> None:
+        """Expose message_user to the roles the user-chat config names (rebuilt so it names the user)."""
+        tools = self._mcp._tool_manager._tools
+        if MESSAGE_USER_TOOL_NAME in self._tool_server_urls:
+            return
+        tool = build_user_chat_tool(self._user_chat, self._create_arg_model_from_schema)
+        tools[MESSAGE_USER_TOOL_NAME] = tool
+        mirrored = [t for t in self._server_tools.get("gateway", []) if t.name != MESSAGE_USER_TOOL_NAME]
+        mirrored.append(MCPTool(name=tool.name, description=tool.description, inputSchema=tool.parameters,
+                                annotations=tool.annotations))
+        self._server_tools["gateway"] = mirrored
+        self._sync_stay_quiet_tool()
+        self._trigger_engine.invalidate_readonly_cache()
+
+    def _sync_stay_quiet_tool(self) -> None:
+        """stay_quiet exists only while the user-chat config turns it on (opt-in); otherwise it is absent."""
+        tools = self._mcp._tool_manager._tools
+        if STAY_QUIET_TOOL_NAME in self._tool_server_urls:
+            return
+        mirrored = [t for t in self._server_tools.get("gateway", []) if t.name != STAY_QUIET_TOOL_NAME]
+        if self._user_chat.stay_quiet:
+            tool = build_stay_quiet_tool(self._user_chat, self._create_arg_model_from_schema)
+            tools[STAY_QUIET_TOOL_NAME] = tool
+            mirrored.append(MCPTool(name=tool.name, description=tool.description, inputSchema=tool.parameters,
+                                    annotations=tool.annotations))
+        else:
+            tools.pop(STAY_QUIET_TOOL_NAME, None)
+        if mirrored:
+            self._server_tools["gateway"] = mirrored
+        else:
+            self._server_tools.pop("gateway", None)
+
+    def _unregister_user_chat_tool(self) -> None:
+        """Remove message_user and stay_quiet when the chat is turned off or cleared; a backing server's are left."""
+        tools = self._mcp._tool_manager._tools
+        if MESSAGE_USER_TOOL_NAME in self._tool_server_urls:
+            return
+        tools.pop(MESSAGE_USER_TOOL_NAME, None)
+        if STAY_QUIET_TOOL_NAME not in self._tool_server_urls:
+            tools.pop(STAY_QUIET_TOOL_NAME, None)
+        remaining = [t for t in self._server_tools.get("gateway", [])
+                     if t.name not in (MESSAGE_USER_TOOL_NAME, STAY_QUIET_TOOL_NAME)]
+        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)