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

@@ -199,7 +199,8 @@ def set_time(self, virtual_time: str, rate=1.0) -> dict:
 
     def set_rate(self, rate) -> dict:
         """Change an armed clock's rate from its time at the call, keeping t0 and generation as an advance does, so time
-        triggers keep their marks and a rate change can't step the clock back by a time read a moment earlier."""
+        triggers keep their marks and a rate change can't step the clock back by a time read a moment earlier; 0 freezes
+        the clock."""
         r = _parse_rate(rate)
         with self._lock:
             if not self._armed or self._t0 is None:

src/agent_env/env/gateway/constants.py

@@ -88,6 +88,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"
@@ -139,9 +140,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"
@@ -230,21 +233,22 @@ 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 setting concurrent_fires (a repeat trigger whose actions are all contact replies (nl) or notifications (notify) runs its fires side by side instead of queueing).",
         "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"}}}},
             "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/set_rate/advance/clear are the harness control path, each changing the clock's opaque version and, given wait_followers_s, answering 504 naming the followers that don't show it in time (set_rate keeps t0, as advance does); a follower follows it through watch, a long poll naming the version it shows and held up to hold_s; state's followers block is diagnostic, not part of the contract.",
+        "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set_time/set_rate/advance/clear are the harness control path, each changing the clock's opaque version and, given wait_followers_s, answering 504 naming the followers that don't show it in time (set_rate keeps t0, as advance does, so time triggers keep their marks; rate 0 freezes the clock); a follower follows it through watch, a long poll naming the version it shows and held up to hold_s; state's followers block is diagnostic, not part of the contract.",
         "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"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}},
-            "set_rate": {"method": "PUT", "endpoint": "/clock/set-rate", "request": {"type": "object", "properties": {"virtual_seconds_per_real_second": {"type": "number"}, **_WAIT_FOLLOWERS}, "required": ["virtual_seconds_per_real_second"]}},
+            "set_rate": {"method": "PUT", "endpoint": "/clock/set-rate", "request": {"type": "object", "properties": {"virtual_seconds_per_real_second": {"type": "number", "minimum": 0}, **_WAIT_FOLLOWERS}, "required": ["virtual_seconds_per_real_second"]}},
             "advance": {"method": "PUT", "endpoint": "/clock/advance", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}},
             "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {**_WAIT_FOLLOWERS}}},
             "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}},
@@ -295,4 +299,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

@@ -97,6 +97,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__)
 
@@ -228,6 +230,7 @@ async def list_website_urls() -> list[str]:
         self._clock = Clock()
         self._clock_followers = ClockFollowers(lambda: self._clock, clock_followers or {})
         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()
 
@@ -322,6 +325,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)
@@ -437,6 +444,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()
@@ -563,9 +615,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, {})
@@ -654,12 +715,15 @@ async def _change_clock(self, change, wait_s: float = 0.0) -> tuple[dict, list[s
         """Every change of the clock goes through here. ``change`` applied, then: the time triggers re-evaluated when it
         moved time on the current arm (it kept the clock's generation, as an advance or a rate change does; a re-arm or
         clear re-anchors them by its new one), every held watch answered, and, given ``wait_s``, a wait for each declared
-        follower to show the new version. The state ``change`` left, with the triggers it made ``due``, and the
+        follower to show the new version. Either way, the fires waiting out virtual time are woken to read the clock
+        again (clock_advanced does it too). The state ``change`` left, with the triggers it made ``due``, and the
         followers still behind."""
         generation = self._clock.generation
         state = change()
         if self._clock.generation == generation:
             state = {**state, "due": self._trigger_engine.clock_advanced()}
+        else:
+            self._trigger_engine.clock_changed()
         self._clock_followers.changed()
         return state, await self._clock_followers.stale_after(state["version"], wait_s) if wait_s else []
 
@@ -1036,7 +1100,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)
@@ -1330,6 +1395,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)