feat(prompt_agent): post-turn hooks, and an example hook that runs a conversation from env event to env event on simulated time

packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/_triggers.py

@@ -206,6 +206,7 @@ def decide(
         solver_message: str = "",
         context_id: str = "default",
         env_triggers: Any = None,
+        record_unfired: bool = True,
     ) -> dict[str, Any]:
         if isinstance(turn, bool) or not isinstance(turn, int) or turn < 1:
             raise TriggerError("decide: 'turn' must be a positive integer")
@@ -256,6 +257,8 @@ def decide(
                     elif action["type"] == "end":
                         done = True
 
+        if not fired_ids and not record_unfired:
+            return {"parts": [], "done": False, "fired": []}
         # Commit only after every predicate succeeds. A timed-out decision can then
         # be retried without consuming the turn or partially firing once-only rules.
         context = self._context_state(context_id)

packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/extensions.py

@@ -285,6 +285,7 @@ class TriggerDecideRequest(ExtensionRequest):
     solver_message: str = Field(default="", max_length=MAX_SOLVER_MESSAGE_LENGTH)
     context_id: str = "default"
     env_triggers: dict[str, dict[str, Any]] | None = None
+    record_unfired: bool = True  # False: a decision where nothing fires changes no state, so it can be repeated
 
 
 PROXY_TOKEN_MIN_LENGTH = 22

packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/framework.py

@@ -836,6 +836,7 @@ async def triggers_decide(self, request: TriggerDecideRequest) -> dict[str, Any]
             solver_message=request.solver_message,
             context_id=request.context_id,
             env_triggers=request.env_triggers,
+            record_unfired=request.record_unfired,
         )
 
     async def triggers_state(self) -> dict[str, Any]:

packages/agentenv-protocol/tests/test_a2a_agent.py

@@ -302,6 +302,23 @@ def trigger(trigger_id: str) -> dict[str, Any]:
         engine.decide(turn=1, context_id="x" * 257)
 
 
+def test_a_decision_where_nothing_fires_can_leave_no_trace() -> None:
+    engine = TriggerEngine()
+    engine.register({"triggers": [{
+        "id": "julia",
+        "when": {"type": "env_trigger", "env_id": "mail", "trigger_id": "school", "status": "fired"},
+        "actions": [{"type": "say", "text": "did you see it?"}],
+    }]})
+    fired = {"mail": {"school": "fired"}}
+
+    assert engine.decide(turn=1, record_unfired=False)["fired"] == []
+    assert engine.decide(turn=1, env_triggers=fired)["fired"] == ["julia"]
+    assert engine.decide(turn=2, env_triggers=fired, record_unfired=False)["fired"] == []
+    # Recorded, turn 2 would now repeat a turn and reset, firing the once-only trigger again.
+    assert engine.decide(turn=2, env_triggers=fired)["fired"] == []
+    assert [entry["kind"] for entry in engine.state()["firing_log"]] == ["registered", "fired"]
+
+
 def test_trigger_engine_bounds_retained_trigger_content() -> None:
     engine = TriggerEngine()
 

pyproject.toml

@@ -122,6 +122,7 @@ drivers = "agent_env"
 hello = "agent_env.examples"
 echo-mcp = "agent_env.examples"
 mail-mcp = "agent_env.examples"
+next-env-event = "agent_env.examples"
 
 [project.urls]
 Homepage = "https://www.agentenvframework.com"

src/agent_env/a2a_agent/conversation_store.py

@@ -109,11 +109,20 @@ def find_by_a2a_task_id(a2a_task_id: str) -> Optional[dict]:
     )
 
 
+def _message(role: str, parts: list[dict], now: datetime, env_clock: Optional[list[dict]]) -> dict:
+    """A conversation message; ``env_clock`` holds each armed env clock's virtual time when it was recorded."""
+    message = {"role": role, "parts": parts, "ts": now.isoformat()}
+    if env_clock is not None:
+        message["env_clock"] = env_clock
+    return message
+
+
 def add_a2a_task(
     conversation_id: str,
     parts: list[dict],
     a2a_task_id: str,
     role: str,
+    env_clock: Optional[list[dict]] = None,
 ) -> Optional[dict]:
     """Append a caller-side message and create a new working a2a_task.
 
@@ -133,7 +142,7 @@ def add_a2a_task(
     doesn't exist.
     """
     now = datetime.now(timezone.utc)
-    new_message = {"role": role, "parts": parts, "ts": now.isoformat()}
+    new_message = _message(role, parts, now, env_clock)
 
     def _mutate(doc: dict) -> UpdateSpec:
         idx = len(doc.get("messages", []))
@@ -161,6 +170,7 @@ def complete_a2a_task(
     conversation_id: str,
     parts: list[dict],
     role: str,
+    env_clock: Optional[list[dict]] = None,
 ) -> Optional[dict]:
     """Append a responder-side message and complete the oldest working a2a_task.
 
@@ -185,7 +195,7 @@ def complete_a2a_task(
     doesn't exist (or the target task lost the state race).
     """
     now = datetime.now(timezone.utc)
-    new_message = {"role": role, "parts": parts, "ts": now.isoformat()}
+    new_message = _message(role, parts, now, env_clock)
     # One backend for the whole retry loop: a reset between the read and the CAS would
     # decide against one database and write to another.
     docs = _doc_store()

src/agent_env/env/gateway/clock.py

@@ -134,6 +134,7 @@ def __init__(self, now_fn=time.monotonic):
         self._rate = 1.0
         self._generation = 0
         self._lived = 0.0  # virtual seconds run before the current arm/clear
+        self._ahead = 0.0  # virtual seconds the current arm was advanced by
 
     @property
     def armed(self) -> bool:
@@ -164,23 +165,40 @@ def set_time(self, virtual_time: str, rate=1.0) -> dict:
             self._bank(t0)
             self._t0 = t0
             self._rate = r
+            self._ahead = 0.0
             self._armed = True
             self._generation += 1
         return self.state()
 
+    def advance(self, virtual_time: str) -> dict:
+        """Jump forward to virtual_time on the current arm, keeping t0, rate and generation, so marks stay put."""
+        target = _parse_rfc3339(virtual_time)
+        with self._lock:
+            if not self._armed or self._t0 is None:
+                raise ClockError("clock is not armed")
+            if target > _MAX_T0 - timedelta(seconds=_OVERFLOW_HORIZON_S * self._rate):
+                raise ClockError(f"virtual_time {virtual_time!r} at rate {self._rate:g} would overflow datetime near the max")
+            current = _advance(self._t0, (self._now() - self._anchor) * self._rate + self._ahead)
+            jump = (target - current).total_seconds()
+            if jump < 0:
+                raise ClockError(f"advance only moves forward: {virtual_time!r} is before the current {_iso(current)}")
+            self._ahead += jump
+            self._lived += jump
+        return self.state()
+
     def clear(self) -> dict:
         """Disarm (idempotent). Afterwards reads 404."""
         with self._lock:
             self._bank()
             self._armed = False
             self._t0 = None
             self._rate = 1.0
+            self._ahead = 0.0
             self._generation += 1
         return {"ok": True, "armed": False}
 
     def elapsed(self) -> float:
-        """Virtual seconds run so far: real time at each arm's rate (1 while unarmed), plus the jump of each re-arm
-        forward; one back counts nothing."""
+        """Virtual seconds run so far: real time at each arm's rate (1 unarmed), plus each forward re-arm or advance."""
         with self._lock:
             return self._lived + (self._now() - self._anchor) * self._rate
 
@@ -189,7 +207,14 @@ def now(self) -> datetime | None:
             if not self._armed or self._t0 is None:
                 return None
             elapsed = self._now() - self._anchor
-            return _advance(self._t0, elapsed * self._rate)
+            return _advance(self._t0, elapsed * self._rate + self._ahead)
+
+    def at_elapsed(self, elapsed: float) -> datetime | None:
+        """The virtual time at which elapsed() reaches `elapsed` on the current arm; None while unarmed."""
+        with self._lock:
+            if not self._armed or self._t0 is None:
+                return None
+            return _advance(self._t0, elapsed - self._lived + self._ahead)
 
     def read(self) -> dict:
         """The frozen consumer contract: {"virtual_time": "<RFC3339>"} — pure read, never advances."""
@@ -204,13 +229,13 @@ def state(self) -> dict:
                 return {"armed": False}
             elapsed = self._now() - self._anchor
             return {"armed": True, "t0": _iso(self._t0), "virtual_seconds_per_real_second": self._rate,
-                    "virtual_time": _iso(_advance(self._t0, elapsed * self._rate))}
+                    "virtual_time": _iso(_advance(self._t0, elapsed * self._rate + self._ahead))}
 
     def _bank(self, t0: datetime | None = None) -> None:
         """Fold the stretch since the last arm/clear into `_lived` at its rate, and start the next one now."""
         now = self._now()
         ran = (now - self._anchor) * self._rate
         self._lived += ran
         if t0 is not None and self._armed:  # a re-arm forward is time the world lived through; one back adds nothing
-            self._lived += max(0.0, (t0 - _advance(self._t0, ran)).total_seconds())
+            self._lived += max(0.0, (t0 - _advance(self._t0, ran + self._ahead)).total_seconds())
         self._anchor = now

src/agent_env/env/gateway/constants.py

@@ -121,8 +121,15 @@ def sources(names: list[str]) -> str:
 TRIGGER_UNLESS = "unless"
 TRIGGER_WEBSITE = "website"
 TRIGGER_EXECUTOR_HOURS_TIME_ZONE = "executor_hours_time_zone"
+TRIGGER_NEXT_EVENT_AT = "next_event_at"
 TRIGGER_FEATURES = (TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES, TRIGGER_ACTION_DELAY, TRIGGER_EXECUTOR_HOURS,
-                    TRIGGER_UNLESS, TRIGGER_WEBSITE, TRIGGER_EXECUTOR_HOURS_TIME_ZONE)
+                    TRIGGER_UNLESS, TRIGGER_WEBSITE, TRIGGER_EXECUTOR_HOURS_TIME_ZONE, TRIGGER_NEXT_EVENT_AT)
+
+# 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"
+# What step/v1 takes beyond its first release, declared on the card as triggers/v1 declares its features.
+STEP_SOURCE = "source"
+STEP_FEATURES = (STEP_SOURCE,)
 
 GATEWAY_TRAJECTORY_FILE = os.environ.get(
     "GATEWAY_TRAJECTORY_FILE",
@@ -167,7 +174,8 @@ def random_mcp_server_name() -> str:
 # call_tool also needs a tool_name; if/then rather than oneOf keeps the fields top-level, where card viewers read them
 _STEP_REQUEST = {
     "type": "object",
-    "properties": {"action": {"enum": ["list_tools", "call_tool"]}, "tool_name": {"type": "string", "minLength": 1}, "arguments": {"type": "object"}},
+    "properties": {"action": {"enum": ["list_tools", "call_tool"]}, "tool_name": {"type": "string", "minLength": 1}, "arguments": {"type": "object"},
+                   "source": {"enum": [STEP_SOURCE_POST_TURN_HOOK]}, "hook_artifact_id": {"type": "string", "minLength": 1}},
     "required": ["action"],
     "if": {"properties": {"action": {"const": "call_tool"}}},
     "then": {"required": ["tool_name"]},
@@ -197,10 +205,11 @@ def random_mcp_server_name() -> str:
     },
     {
         "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/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/clear are the harness control path.",
         "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"]}},
             "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {}}},
             "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}}}},
         "required": False,
@@ -214,8 +223,8 @@ def random_mcp_server_name() -> str:
     },
     {
         "uri": EXT_STEP_URI,
-        "description": "Step (urn:agentenv:step/v1): list or call the env's tools over REST, filtered by the AgentEnv-Role header.",
-        "params": {"endpoint": "/step", "methods": {
+        "description": "Step (urn:agentenv:step/v1): list or call the env's tools over REST, filtered by the AgentEnv-Role header; a call a post-turn hook makes names it, and the trajectory records it as the hook's.",
+        "params": {"endpoint": "/step", "features": list(STEP_FEATURES), "methods": {
             "step": {"method": "POST", "endpoint": "/step", "request": _STEP_REQUEST}}},
         "required": False,
     },

src/agent_env/env/gateway/gateway.py

@@ -58,6 +58,7 @@
     PROTOCOL_VERSION,
     RPC_PATH,
     STATE_TOOL_NAME_CLASHES_KEY,
+    STEP_SOURCE_POST_TURN_HOOK,
     TOOL_DISABLE_ACTION,
     TOOL_ENABLE_ACTION,
     WELL_KNOWN_PATH,
@@ -77,6 +78,7 @@
 
 # RFC 9110's safe methods: a request using one asks for no change, so the proxy measures no write window for it.
 _SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE"})
+_STEP_API = {"source": "step_api"}
 
 
 class ToolNameClash(ValueError):
@@ -278,6 +280,20 @@ async def clock_set_time(request: Request) -> Response:
             self._register_get_time_tool()
             return self._json_response(state, 200)
 
+        @self._mcp.custom_route("/clock/advance", methods=["PUT"])
+        async def clock_advance(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.advance(body.get("virtual_time"))
+            except ClockError as e:
+                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/clear", methods=["POST"])
         async def clock_clear(request: Request) -> Response:
             self._unregister_get_time_tool()
@@ -975,10 +991,14 @@ async def _handle_step(self, request: Request) -> Response:
             return self._json_response({"error": "Invalid JSON body"}, 400)
         action = body.get("action")
         role = self._request_role(request)
+        source = _step_source(body)
+        if source is None:
+            return self._json_response({"error": f"'source' may only be '{STEP_SOURCE_POST_TURN_HOOK}', with a "
+                                                 "non-empty 'hook_artifact_id' naming the hook"}, 400)
         if action == "list_tools":
             return self._step_list_tools(role)
         elif action == "call_tool":
-            return await self._step_call_tool(body, role)
+            return await self._step_call_tool(body, role, source)
         else:
             return self._json_response({"error": f"Unknown action: {action}. Expected 'list_tools' or 'call_tool'."}, 400)
 
@@ -996,7 +1016,7 @@ def _step_list_tools(self, role: str) -> Response:
         ]
         return self._json_response({"tools": tools}, 200)
 
-    async def _step_call_tool(self, body: dict, role: str) -> Response:
+    async def _step_call_tool(self, body: dict, role: str, source: dict = _STEP_API) -> Response:
         tool_name = body.get("tool_name")
         if not tool_name or not isinstance(tool_name, str):
             return self._json_response({"error": "Missing or invalid 'tool_name' field"}, 400)
@@ -1012,10 +1032,10 @@ async def _step_call_tool(self, body: dict, role: str) -> Response:
 
         server_url = self._tool_server_urls.get(tool_name)
         if server_url is None:
-            return await self._step_call_gateway_tool(tool_name, arguments, role)
+            return await self._step_call_gateway_tool(tool_name, arguments, role, source)
         tool_call_event_id = await self._log_event({
             "event_type": "tool_call",
-            "source": "step_api",
+            **source,
             "tool_call": {"function_name": tool_name, "arguments": arguments},
             **self._attribution(role),
         })
@@ -1025,13 +1045,13 @@ async def _step_call_tool(self, body: dict, role: str) -> Response:
                 try:
                     result = await self._call_child_tool(server_url, tool_name, arguments, caller=Caller(role=role))
                 except Exception as e:
-                    await self._log_failed_call(tool_call_event_id, changelog_before, e, source="step_api")
+                    await self._log_failed_call(tool_call_event_id, changelog_before, e, **source)
                     raise
                 changelog_id = self._query_changelog_id()
 
             await self._log_event({
                 "event_type": "tool_call_result",
-                "source": "step_api",
+                **source,
                 "tool_call_event_id": tool_call_event_id,
                 "tool_call_result": result.model_dump(mode="json"),
                 "changelog_id_before": changelog_before,
@@ -1049,14 +1069,15 @@ async def _step_call_tool(self, body: dict, role: str) -> Response:
             error_detail = str(e) or repr(e)
             return self._json_response({"error": f"Tool execution failed ({type(e).__name__}): {error_detail}"}, 500)
 
-    async def _step_call_gateway_tool(self, tool_name: str, arguments: dict, role: str = DEFAULT_ROLE) -> Response:
+    async def _step_call_gateway_tool(self, tool_name: str, arguments: dict, role: str = DEFAULT_ROLE,
+                                      source: dict = _STEP_API) -> Response:
         """Call a FastMCP-registered tool (e.g. list_website_urls) for /step."""
         tool = self._mcp._tool_manager._tools.get(tool_name)
         if tool is None:
             return self._json_response({"error": f"Unknown tool: {tool_name}"}, 404)
         tool_call_event_id = await self._log_event({
             "event_type": "tool_call",
-            "source": "step_api",
+            **source,
             "tool_call": {"function_name": tool_name, "arguments": arguments},
             **self._attribution(role),
         })
@@ -1072,7 +1093,7 @@ async def _step_call_gateway_tool(self, tool_name: str, arguments: dict, role: s
                 response_data = result.model_dump(mode="json", exclude_none=True)
             await self._log_event({
                 "event_type": "tool_call_result",
-                "source": "step_api",
+                **source,
                 "tool_call_event_id": tool_call_event_id,
                 "tool_call_result": response_data,
                 "changelog_id": changelog_id,
@@ -1319,6 +1340,16 @@ def model_dump_one_level(self) -> dict[str, Any]:
         return kwargs
 
 
+def _step_source(body: dict) -> dict | None:
+    """The trajectory tag of a /step call: step_api, or the post-turn hook its body names; None for any other source."""
+    if "source" not in body:
+        return _STEP_API
+    hook = body.get("hook_artifact_id")
+    if body["source"] != STEP_SOURCE_POST_TURN_HOOK or not isinstance(hook, str) or not hook:
+        return None
+    return {"source": STEP_SOURCE_POST_TURN_HOOK, "hook_artifact_id": hook}
+
+
 def _request_args(request: Request, body: bytes) -> dict:
     """A request's query parameters and form or JSON-object body fields (the body wins a clash; repeats become lists)."""
     args = _fields(parse_qsl(request.url.query, keep_blank_values=True))