feat(gateway): pace trigger replies in virtual time, add deadlines, and attribute tool calls

src/agent_env/env/gateway/clock.py

@@ -119,14 +119,20 @@ def __init__(self, now_fn=time.monotonic):
         self._lock = threading.Lock()
         self._armed = False
         self._t0: datetime | None = None
-        self._anchor: float | None = None
+        self._anchor = now_fn()  # real time the current arm (or unarmed stretch) began
         self._rate = 1.0
         self._generation = 0
+        self._lived = 0.0  # virtual seconds run before the current arm/clear
 
     @property
     def armed(self) -> bool:
         return self._armed
 
+    @property
+    def rate(self) -> float:
+        """Virtual seconds per real second; 1.0 while unarmed."""
+        return self._rate
+
     @property
     def generation(self) -> int:
         """Bumped on every arm/clear — lets consumers detect a re-anchor and drop stale derived state."""
@@ -144,8 +150,8 @@ def set_time(self, virtual_time: str, rate=1.0) -> dict:
         if t0 > _MAX_T0 - timedelta(seconds=_OVERFLOW_HORIZON_S * r):
             raise ClockError(f"virtual_time {virtual_time!r} at rate {r:g} would overflow datetime near the max")
         with self._lock:
+            self._bank(t0)
             self._t0 = t0
-            self._anchor = self._now()
             self._rate = r
             self._armed = True
             self._generation += 1
@@ -154,13 +160,19 @@ def set_time(self, virtual_time: str, rate=1.0) -> dict:
     def clear(self) -> dict:
         """Disarm (idempotent). Afterwards reads 404."""
         with self._lock:
+            self._bank()
             self._armed = False
             self._t0 = None
-            self._anchor = None
             self._rate = 1.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."""
+        with self._lock:
+            return self._lived + (self._now() - self._anchor) * self._rate
+
     def now(self) -> datetime | None:
         with self._lock:
             if not self._armed or self._t0 is None:
@@ -182,3 +194,12 @@ def state(self) -> dict:
             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))}
+
+    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._anchor = now

src/agent_env/env/gateway/constants.py

@@ -103,15 +103,19 @@ def sources(names: list[str]) -> str:
     return "; ".join(f"'{name}' from {sources(names)}" for name, names in clashes.items())
 
 # Trigger status vocabulary, part of the GET /triggers/state contract; `state` asserts against it.
-TRIGGER_STATUSES = ("armed", "firing", "queued", "fired", "failed")
+TRIGGER_STATUSES = ("armed", "firing", "queued", "fired", "failed", "cancelled")
 
 # A fire is in flight; `queued` also has further provoking calls already waiting to be drained.
 TRIGGER_IN_FLIGHT_STATUSES = ("firing", "queued")
 
 # What triggers/v1 takes beyond its first release, declared on the card so a client can refuse an older gateway.
 TRIGGER_NAMED_EXECUTORS = "named_executors"
 TRIGGER_PROVOKING_ROLES = "provoking_roles"
-TRIGGER_FEATURES = (TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES)
+TRIGGER_ACTION_DELAY = "action_delay"
+TRIGGER_EXECUTOR_HOURS = "executor_hours"
+TRIGGER_UNLESS = "unless"
+TRIGGER_FEATURES = (TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES, TRIGGER_ACTION_DELAY, TRIGGER_EXECUTOR_HOURS,
+                    TRIGGER_UNLESS)
 
 GATEWAY_TRAJECTORY_FILE = os.environ.get(
     "GATEWAY_TRAJECTORY_FILE",

src/agent_env/env/gateway/gateway.py

@@ -417,7 +417,7 @@ async def filtered_call_tool(name, arguments, context=None, convert_result=False
             if self._is_disabled(role, name):
                 raise ToolError(f"Tool '{name}' is disabled for role '{role}'")
             native = name not in self._tool_server_urls
-            event_id = await self._log_native_tool_call(name, arguments) if native else None
+            event_id = await self._log_native_tool_call(name, arguments, role) if native else None
             result = await original_call_tool(name, arguments, context=context, convert_result=convert_result)
             if native:
                 await self._log_native_tool_result(event_id, result)
@@ -946,6 +946,7 @@ async def _step_call_tool(self, body: dict, role: str) -> Response:
             "event_type": "tool_call",
             "source": "step_api",
             "tool_call": {"function_name": tool_name, "arguments": arguments},
+            **self._attribution(role),
         })
         try:
             async with self._tool_call_guard():
@@ -980,6 +981,7 @@ async def _step_call_gateway_tool(self, tool_name: str, arguments: dict, role: s
             "event_type": "tool_call",
             "source": "step_api",
             "tool_call": {"function_name": tool_name, "arguments": arguments},
+            **self._attribution(role),
         })
         try:
             async with self._tool_call_guard():
@@ -1067,6 +1069,7 @@ async def proxy(**kwargs) -> CallToolResult:
                     "function_name": tool_name,
                     "arguments": kwargs,
                 },
+                **self._attribution(caller.role),
             })
 
             async with self._tool_call_guard():
@@ -1109,12 +1112,13 @@ def _close_trajectory_file(self) -> None:
         if self._trajectory_file and not self._trajectory_file.closed:
             self._trajectory_file.close()
 
-    async def _log_native_tool_call(self, name: str, arguments: dict | None) -> str | None:
+    async def _log_native_tool_call(self, name: str, arguments: dict | None, role: str) -> str | None:
         """Trajectory record for a gateway-owned tool on the MCP path (fail-open)."""
         try:
             return await self._log_event({
                 "event_type": "tool_call",
                 "tool_call": {"function_name": name, "arguments": arguments or {}},
+                **self._attribution(role),
             })
         except Exception:
             logger.exception(f"Failed to log tool_call for gateway tool '{name}'")
@@ -1132,6 +1136,12 @@ async def _log_native_tool_result(self, event_id: str | None, result: Any) -> No
         except Exception:
             logger.exception("Failed to log tool_call_result for a gateway tool")
 
+    def _attribution(self, role: str) -> dict:
+        """Who made a tool call, for its trajectory event: the caller's role, and the trigger fires it was made for
+        when that role is an executor realizing them."""
+        causes = self._trigger_engine.caused_by(role)
+        return {"role": role, **({"caused_by": causes} if causes else {})}
+
     async def _log_event(self, event: dict) -> str:
         async with self._event_lock:
             self._event_counter += 1