diff --git a/src/agent_env/env/gateway/clock.py b/src/agent_env/env/gateway/clock.py
index d9e9b369..44913026 100644
--- a/src/agent_env/env/gateway/clock.py
+++ b/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
diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py
index c0e27eff..cdc8420c 100644
--- a/src/agent_env/env/gateway/constants.py
+++ b/src/agent_env/env/gateway/constants.py
@@ -103,7 +103,7 @@ 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")
@@ -111,7 +111,11 @@ def sources(names: list[str]) -> str:
 # 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",
diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py
index c6e898b6..2e86d3a8 100644
--- a/src/agent_env/env/gateway/gateway.py
+++ b/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
diff --git a/src/agent_env/env/gateway/triggers.py b/src/agent_env/env/gateway/triggers.py
index 6d1eea42..41da5e1d 100644
--- a/src/agent_env/env/gateway/triggers.py
+++ b/src/agent_env/env/gateway/triggers.py
@@ -41,6 +41,8 @@
 _BARRIER_AT = ("provoking_call",)
 _BARRIER_KEYS = ("at", "timeout_seconds")
 _DROP = object()  # an unresolved `${x?}` used as a whole value: drop the key / element
+# Real seconds between re-reads of the clock while waiting out virtual time, so a rate change, re-arm or clear is followed.
+_VIRTUAL_RECHECK_S = 5.0
 # An allowlist, not a size bound: identifier/metadata keys only, since bodies and addresses are PII.
 _ECHO_KEYS = frozenset({
     "action", "createdTime", "fileName", "kind", "mimeType", "modifiedTime", "name", "role",
@@ -59,6 +61,8 @@
     "action_ok", "action_failed",     # one fire-action within a trigger
     "verify_ok", "verify_failed",     # typed acceptance of an action
     "fired", "failed", "eval_error",  # per trigger; `failed` re-arms an action/state trigger, retires a time one
+    "delayed", "deferred",            # an action waited out its delay / its executor's working hours
+    "cancelled",                      # a deadline's `unless` call came before it was due
 })
 # Trigger-level failures; the per-action kinds are details of one of these, so this counts once per failure.
 _FAILURE_KINDS = ("failed", "eval_error")
@@ -385,6 +389,7 @@ class TriggerEngine:
     def __init__(self, gateway):
         self._gw = gateway
         self._executors: dict[str | None, dict] = {}  # `executor` under None, `executors` by name
+        self._in_flight: dict[str, list[dict]] = {}  # executor role -> the fires it is realizing, for attribution
         self._watch_roles: set[str] = set()
         self._config_set = False
         self._triggers: dict[str, dict] = {}
@@ -443,7 +448,8 @@ def register(self, body: Any) -> dict:
                                    "next_mark": None, "resolved": False, "fire_count": 0,
                                    "failure_count": 0, "last_failure_at": None,
                                    "rng": _make_rng(spec), "mark_gen": None,
-                                   "pending": deque(), "task": None}
+                                   "pending": deque(), "task": None, "delay_rngs": _delay_rngs(spec),
+                                   "attempts": 0, "waiting": False}
             self._emit("added", tid)
         self._recompute_time_flag()
         return {"ok": True, "added": batch_order, "all": list(self._triggers)}
@@ -468,6 +474,8 @@ def _check_tools_known(spec: dict, known: set[str]) -> None:
             named.append((when["tool"], "when.tool"))
         elif when["type"] == "state":
             named.extend((t, "when.check") for t in _check_tools(when["check"]))
+        elif "unless" in when:
+            named.append((when["unless"]["tool"], "when.unless.tool"))
         for i, action in enumerate(spec["actions"]):
             if action["type"] == "tool":
                 named.append((action["tool"], f"actions[{i}].tool"))
@@ -486,7 +494,10 @@ def remove(self, body: Any) -> dict:
             raise TriggerError("ids must be a list of non-empty strings")
         removed = []
         for tid in ids:
-            if self._triggers.pop(tid, None) is not None:
+            trig = self._triggers.pop(tid, None)
+            if trig is not None:
+                if trig["task"] is not None:
+                    trig["task"].cancel()  # a fire waiting out a delay or hours would otherwise still act
                 removed.append(tid)
                 self._emit("removed", tid)
         self._recompute_time_flag()
@@ -503,6 +514,7 @@ def clear(self) -> dict:
         self._seq = 0
         self._watch_roles = set()
         self._executors = {}
+        self._in_flight = {}
         self._config_set = False
         self._readonly_tools = None
         self._recompute_time_flag()
@@ -593,9 +605,12 @@ def _normalize_executor(raw: Any, where: str, watch_roles: set[str]) -> dict:
             raise TriggerError(f"{where}.role must be a non-empty string (the deployed executor's AgentEnv-Role, kept out of watch_roles so its own tool calls never fire triggers)")
         if role in watch_roles:
             raise TriggerError(f"{where}.role {role!r} must not be in watch_roles {sorted(watch_roles)} — the executor's own tool calls would fire triggers (cascade)")
-        return {"a2a_url": raw["a2a_url"].rstrip("/"),
-                "timeout_seconds": float(raw.get("timeout_seconds", 120)),
-                "role": role}
+        executor = {"a2a_url": raw["a2a_url"].rstrip("/"),
+                    "timeout_seconds": float(raw.get("timeout_seconds", 120)),
+                    "role": role}
+        if raw.get("hours") is not None:
+            executor["hours"] = _normalize_hours(raw["hours"], where)
+        return executor
 
     @staticmethod
     def _normalize_barrier(spec: dict, tid: str, has_ctx: bool) -> None:
@@ -629,6 +644,10 @@ def _normalize_trigger(self, spec: dict, executors: dict[str | None, dict]) -> d
         if not isinstance(when, dict) or when.get("type") not in ("action", "state", "time"):
             raise TriggerError(f"trigger '{tid}': when.type must be 'action', 'state', or 'time'")
         self._normalize_roles(when, tid)
+        if when.get("unless") is None:
+            when.pop("unless", None)
+        if "unless" in when and when["type"] != "time":
+            raise TriggerError(f"trigger '{tid}': when.unless is only valid on time triggers")
         if when["type"] == "action":
             self._normalize_action_when(when, tid)
         elif when["type"] == "state":
@@ -652,6 +671,10 @@ def _normalize_trigger(self, spec: dict, executors: dict[str | None, dict]) -> d
             if action["type"] == "nl" and target is not None and target["role"] in when.get("roles", ()):
                 raise TriggerError(f"trigger '{tid}': when.roles includes {target['role']!r}, the role of the executor "
                                    f"actions[{i}] goes to, so each of its replies would provoke it again")
+        if spec.get("barrier") and any("delay" in action or (action["type"] == "nl" and "hours" in (
+                executors.get(action.get("executor")) or {})) for action in actions):
+            raise TriggerError(f"trigger '{tid}': a delayed action, or an nl action whose executor keeps working hours, "
+                               f"can't run under a barrier, which holds the provoking call until the actions finish")
         return spec
 
     @staticmethod
@@ -691,6 +714,10 @@ def _normalize_action(action: Any, w: str, has_ctx: bool, executors: dict[str |
             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 action.get("delay") is None:
+            action.pop("delay", None)
+        else:
+            _normalize_delay(action["delay"], w)
         if atype == "nl":
             if not isinstance(action.get("instruction"), str) or not action["instruction"]:
                 raise TriggerError(f"{w}: instruction must be a non-empty string")
@@ -743,6 +770,10 @@ def _normalize_time_when(self, when: dict, tid: str) -> None:
             raise TriggerError(f"{w}: when.type=='time' needs at least one of 'at', 'after', or 'every'")
         if has_at and has_after:
             raise TriggerError(f"{w}: when may have at most one of 'at' / 'after'")
+        if "unless" in when:
+            if has_every:
+                raise TriggerError(f"{w}: when.unless is only valid on a one-shot time trigger ('at' or 'after')")
+            _normalize_unless(when["unless"], w)
         if has_after:
             if not isinstance(when["after"], str) or not when["after"]:
                 raise TriggerError(f"{w}: when.after must be a non-empty trigger id")
@@ -793,7 +824,7 @@ def state(self) -> dict:
                  "fired_at": t["fired_at"], "fire_count": t["fire_count"],
                  "failure_count": t["failure_count"], "last_failure_at": t["last_failure_at"],
                  "barrier": t["spec"].get("barrier"),
-                 "pending": len(t["pending"]),
+                 "pending": len(t["pending"]), "waiting": t["waiting"],
                  "next_mark": _iso(t["next_mark"]) if t["next_mark"] else None}
                 for tid, t in self._triggers.items()
             ],
@@ -823,6 +854,19 @@ def on_tool_call(self, role: str, name: str, arguments: dict,
         data = _result_data(result)
         ctx = {"args": arguments, "result": data}
         provoking = {"tool": name, "role": role}
+        # Before detection, so the call that sets off a deadline's anchor can't also be the call that cancels it.
+        for tid, trig in provokable:
+            when = trig["spec"]["when"]
+            unless = when.get("unless")
+            if unless is None or unless["tool"] != name or trig["status"] != "armed":
+                continue
+            # An `after` deadline's window opens when its anchor starts firing, so a reply to the anchor's own action counts.
+            if ("after" in when and not trig["resolved"]
+                    and self._triggers.get(when["after"], {}).get("status") not in TRIGGER_IN_FLIGHT_STATUSES):
+                continue
+            if all(_eval_predicate(pred, _where_value(ctx, field)) for field, pred in unless.get("where", {}).items()):
+                self._cancel(tid, trig, provoking=provoking)
+                self._recompute_time_flag()
         for tid, trig in provokable:
             when = trig["spec"]["when"]
             if when["type"] != "action" or when["tool"] != name:
@@ -846,9 +890,10 @@ def on_tool_call(self, role: str, name: str, arguments: dict,
         return pending, timeout_s
 
     def _provoked_by(self, trig: dict, role: str) -> bool:
-        """Whether a call by ``role`` can provoke ``trig``: the roles its `when` lists, else the watched ones. Unlisted,
-        an executor's own calls never provoke anything, so an executor cannot set off a cascade unasked."""
-        roles = trig["spec"]["when"].get("roles")
+        """Whether a call by ``role`` can provoke ``trig``: the roles its `when` (or a deadline's `unless`) lists, else the
+        watched ones. Unlisted, an executor's own calls never provoke anything, so it cannot set off a cascade unasked."""
+        when = trig["spec"]["when"]
+        roles = when.get("roles", (when.get("unless") or {}).get("roles"))
         if roles is not None:
             return role in roles
         return role in self._watch_roles and role not in self._executor_roles
@@ -859,16 +904,17 @@ def _executor_roles(self) -> set[str]:
 
     def _mark_detected(self, tid: str, trig: dict, provoking: dict, ctx: dict | None = None) -> asyncio.Task | None:
         ctx = ctx if ctx is not None else {}
+        since = self._gw._clock.elapsed()  # a delay runs from the provoking call, not from when its turn comes
         if trig["status"] in TRIGGER_IN_FLIGHT_STATUSES:
             # Mid-fire: queue it; the running fire drains the queue before it settles.
-            trig["pending"].append(ctx)
+            trig["pending"].append((ctx, since))
             trig["status"] = "queued"
             self._emit("detected", tid, provoking=provoking, queued=len(trig["pending"]))
             return trig["task"]
         trig["status"] = "firing"
         trig["detected_at"] = _now()
         self._emit("detected", tid, provoking=provoking)
-        trig["task"] = self._schedule(self._fire(tid, trig, ctx))
+        trig["task"] = self._schedule(self._fire(tid, trig, ctx, since))
         trig["task"].set_name(tid)
         return trig["task"]
 
@@ -933,7 +979,7 @@ def _resolve_dependents(self, anchor_tid: str) -> None:
             return
         for tid, trig in self._triggers.items():
             when = trig["spec"]["when"]
-            if when["type"] != "time" or when.get("after") != anchor_tid or trig["resolved"]:
+            if when["type"] != "time" or when.get("after") != anchor_tid or trig["resolved"] or trig["status"] != "armed":
                 continue
             trig["next_mark"] = _advance(now, _parse_duration(when["offset"]).total_seconds())
             trig["resolved"] = True
@@ -999,15 +1045,21 @@ def _mark_time_detected(self, tid: str, trig: dict, marks: list[datetime], cappe
         # first/last only: a burst can carry _MAX_CATCHUP marks, each with its own `fired` event.
         self._emit("detected", tid, provoking=provoking, due=len(marks), capped=capped,
                    first_mark=_iso(marks[0]), last_mark=_iso(marks[-1]))
-        self._schedule(self._fire_time(tid, trig, marks, capped, terminated))
+        trig["task"] = self._schedule(self._fire_time(tid, trig, marks, capped, terminated))
 
     async def _fire_time(self, tid: str, trig: dict, marks: list[datetime], capped: bool, terminated: bool) -> None:
         """Run the actions once per due mark, then re-arm (recurring) or retire (one-shot / terminated)."""
         try:
             await self._gw._log_event({"event_type": "trigger_fired", "source": "trigger_engine", "trigger_id": tid})
-            for mark in marks:
+            clock = self._gw._clock
+            now, elapsed = clock.now(), clock.elapsed()
+            # A mark's delay runs from the mark, so a backlog doesn't push each reply back further; reckoned before any
+            # wait, so a re-arm during one doesn't shorten the rest.
+            sinces = [elapsed - (max(0.0, (now - mark).total_seconds()) if now is not None else 0.0) for mark in marks]
+            for mark, since in zip(marks, sinces):
+                trig["attempts"] += 1
                 for index, action in enumerate(trig["spec"]["actions"]):
-                    if not await self._run_action(tid, index, action, {}):
+                    if not await self._run_action(tid, trig, index, action, {}, since):
                         trig["status"] = "failed"
                         trig["next_mark"] = None
                         self._emit("failed", tid, action_index=index)
@@ -1119,7 +1171,7 @@ async def _internal_call(self, tool_name: str, arguments: dict, role: str | None
         caller = Caller(role=role) if role is not None else None
         return await gw._call_child_tool(server_url, tool_name, arguments, caller=caller)
 
-    async def _fire(self, tid: str, trig: dict, ctx: dict) -> None:
+    async def _fire(self, tid: str, trig: dict, ctx: dict, since: float) -> None:
         """Run the actions for one provoking call, then drain any calls queued while firing (repeat
         triggers). A failure always re-arms: a mirror states a standing invariant, so one transient
         error must not silence it for the rest of the run. Ends armed or fired (one-shot success)."""
@@ -1128,11 +1180,12 @@ async def _fire(self, tid: str, trig: dict, ctx: dict) -> None:
         try:
             while True:
                 ok = True
+                trig["attempts"] += 1
                 try:
                     await self._gw._log_event({"event_type": "trigger_fired", "source": "trigger_engine", "trigger_id": tid,
                                                "provoking_tool": (ctx.get("args") is not None and spec["when"].get("tool")) or None})
                     for index, action in enumerate(spec["actions"]):
-                        if not await self._run_action(tid, index, action, ctx):
+                        if not await self._run_action(tid, trig, index, action, ctx, since):
                             ok = False
                             self._emit("failed", tid, action_index=index)
                             break
@@ -1147,21 +1200,24 @@ async def _fire(self, tid: str, trig: dict, ctx: dict) -> None:
                 if ok:
                     self._resolve_dependents(tid)  # success-only, outside the try so it can't flip a fired trigger to failed
                 if trig["pending"]:
-                    ctx = trig["pending"].popleft()
+                    ctx, since = trig["pending"].popleft()
                     trig["status"] = "firing"
                     trig["detected_at"] = _now()
                     continue
                 trig["status"] = "armed" if (repeat or not ok) else "fired"
                 return
         finally:
-            # An abnormal exit must not wedge the trigger in firing/queued, nor drop its queue silently.
-            if trig["status"] in TRIGGER_IN_FLIGHT_STATUSES:
+            # An abnormal exit must not wedge the trigger in firing/queued, nor drop its queue silently; a trigger removed
+            # or cleared meanwhile is gone, and has nothing left to report.
+            if trig["status"] in TRIGGER_IN_FLIGHT_STATUSES and self._triggers.get(tid) is trig:
                 self._emit("failed", tid, detail="firing did not settle", dropped=len(trig["pending"]))
                 trig["status"] = "armed"
                 trig["pending"].clear()
             trig["task"] = None
 
-    async def _run_action(self, tid: str, index: int, action: dict, ctx: dict) -> bool:
+    async def _run_action(self, tid: str, trig: dict, index: int, action: dict, ctx: dict, since: float) -> bool:
+        if "delay" in action:
+            await self._delay(tid, trig, index, action, since)
         changelog_before = self._gw._query_changelog_id()
         try:
             if action["type"] == "permission":
@@ -1195,7 +1251,7 @@ async def _run_action(self, tid: str, index: int, action: dict, ctx: dict) -> bo
                     return False
                 detail = {"tool": action["tool"], "args": _echo_args(args), **acting}
             else:
-                if not await self._run_nl(tid, index, action, ctx):
+                if not await self._run_nl(tid, trig, index, action, ctx):
                     return False
                 detail = {"nl": action["instruction"][:120], **_named_executor(action)}
             if "verify" in action and action["type"] != "nl":
@@ -1219,26 +1275,84 @@ async def _run_verify(self, tid: str, index: int, verify: dict, ctx: dict | None
         self._emit("verify_failed", tid, action_index=index)
         return False
 
-    async def _run_nl(self, tid: str, index: int, action: dict, ctx: dict | None = None) -> bool:
+    async def _run_nl(self, tid: str, trig: dict, index: int, action: dict, ctx: dict | None = None) -> bool:
         instruction = action["instruction"]
         named = _named_executor(action)
         executor = self._executors.get(action.get("executor"))
-        for attempt in (1, 2):
-            reply = await self._call_executor(executor, instruction, context_id=f"trigger-{tid}")
-            if reply is None:
-                self._emit("action_failed", tid, action_index=index,
-                           detail=f"executor attempt {attempt} did not complete", **named)
-                return False
-            if "verify" not in action:
-                return True
-            if await self._run_verify(tid, index, action["verify"], ctx):
-                return True
-            if attempt == 1:
-                instruction = (f"{action['instruction']}\n\nYour previous attempt did not pass verification — "
-                               f"the expected change is not observable in the environment. Do it again exactly as described.")
+        if executor is not None and "hours" in executor:
+            await self._await_hours(tid, trig, index, executor["hours"], named)
+        cause = {"trigger_id": tid, "action_index": index, "fire": trig["attempts"]}
+        causes = self._in_flight.setdefault(executor["role"], []) if executor is not None else []
+        causes.append(cause)
+        try:
+            for attempt in (1, 2):
+                reply = await self._call_executor(executor, instruction, context_id=f"trigger-{tid}")
+                if reply is None:
+                    self._emit("action_failed", tid, action_index=index,
+                               detail=f"executor attempt {attempt} did not complete", **named)
+                    return False
+                if "verify" not in action:
+                    return True
+                if await self._run_verify(tid, index, action["verify"], ctx):
+                    return True
+                if attempt == 1:
+                    instruction = (f"{action['instruction']}\n\nYour previous attempt did not pass verification — "
+                                   f"the expected change is not observable in the environment. Do it again exactly as described.")
+        finally:
+            causes.remove(cause)
         self._emit("action_failed", tid, action_index=index, detail="verification failed after retry", **named)
         return False
 
+    def caused_by(self, role: str) -> list[dict]:
+        """The fires in flight for the executor with ``role``, all of them when several overlap; its tool calls are
+        attributed to them. A fire is the trigger's n-th, failed ones counted."""
+        return list(self._in_flight.get(role, ()))
+
+    async def _delay(self, tid: str, trig: dict, index: int, action: dict, since: float) -> None:
+        """Wait until the action's delay has run since ``since``, the Clock.elapsed() reading of its cause."""
+        delay = action["delay"]
+        if isinstance(delay, str):
+            seconds = _parse_duration(delay).total_seconds()
+        else:
+            mean = _parse_duration(delay["mean"]).total_seconds()
+            seconds = trig["delay_rngs"][index].expovariate(1.0 / mean)
+        self._emit("delayed", tid, action_index=index, virtual_seconds=round(seconds, 3), **_named_executor(action))
+        trig["waiting"] = True
+        try:
+            await self._sleep_virtual(seconds - (self._gw._clock.elapsed() - since))
+        finally:
+            trig["waiting"] = False
+
+    async def _await_hours(self, tid: str, trig: dict, index: int, hours: dict, named: dict) -> None:
+        """Wait until the executor's working hours are open, reckoned again after a re-arm; they don't apply while the
+        clock is unarmed."""
+        clock = self._gw._clock
+        gen = opens = None
+        trig["waiting"] = True
+        try:
+            while True:
+                current = clock.generation  # read before now(), so a re-arm in between is caught on the next pass
+                if (now := clock.now()) is None:
+                    return
+                if current != gen:
+                    gen, opens = current, _next_opening(now, hours)
+                    if opens is not None:
+                        self._emit("deferred", tid, action_index=index, until=_iso(opens), **named)
+                if opens is None or now >= opens:
+                    return
+                await self._sleep_virtual((opens - now).total_seconds(), generation=gen)
+        finally:
+            trig["waiting"] = False
+
+    async def _sleep_virtual(self, seconds: float, generation: int | None = None) -> None:
+        """Wait out ``seconds`` of virtual time as the clock runs (real time at its rate, 1 while unarmed, none while
+        frozen) or jumps ahead. Given the clock's ``generation``, stop early once a re-arm or clear moves it on."""
+        clock = self._gw._clock
+        until = clock.elapsed() + seconds
+        while (left := until - clock.elapsed()) > 0 and (generation is None or clock.generation == generation):
+            rate = clock.rate
+            await asyncio.sleep(min(left / rate, _VIRTUAL_RECHECK_S) if rate else _VIRTUAL_RECHECK_S)
+
     async def _call_executor(self, executor: dict | None, instruction: str, context_id: str) -> str | None:
         if executor is None:
             return None
@@ -1268,6 +1382,15 @@ async def _call_executor(self, executor: dict | None, instruction: str, context_
                     return "\n".join(p.get("text", "") for p in parts if p.get("kind") == "text")
         return None
 
+    def _cancel(self, tid: str, trig: dict, **cause) -> None:
+        """Retire a deadline its `unless` call answered, and the time triggers anchored on it, which can no longer fire."""
+        trig["status"] = "cancelled"
+        trig["next_mark"] = None
+        self._emit("cancelled", tid, **cause)
+        for dep_id, dep in self._triggers.items():
+            if dep["spec"]["when"].get("after") == tid and dep["status"] == "armed":
+                self._cancel(dep_id, dep, anchor=tid)
+
 
 def _result_text(result: CallToolResult) -> str:
     return "\n".join(c.text for c in result.content if getattr(c, "type", None) == "text")
@@ -1276,3 +1399,85 @@ def _result_text(result: CallToolResult) -> str:
 def _named_executor(action: dict) -> dict:
     """``{"executor": name}`` for an nl action that names its executor, for its firing-log events; else empty."""
     return {"executor": action["executor"]} if action.get("executor") is not None else {}
+
+
+_DAYS = ("mon", "tue", "wed", "thu", "fri", "sat", "sun")
+_HHMM_RE = re.compile(r"([01]\d|2[0-3]):([0-5]\d)")
+_UTC_OFFSET_RE = re.compile(r"([+-])([01]\d|2[0-3]):([0-5]\d)")
+
+
+def _normalize_delay(delay: Any, w: str) -> None:
+    """Validate an action's `delay`: a positive ISO-8601 duration, or a seeded exponential like `every`'s."""
+    if isinstance(delay, str):
+        if _dur_or_400(delay, f"{w}: delay") <= timedelta(0):
+            raise TriggerError(f"{w}: delay must be a positive duration")
+        return
+    if not isinstance(delay, dict) or delay.get("dist") != "exp":
+        raise TriggerError(f"{w}: delay must be an ISO-8601 duration or {{dist: 'exp', mean, seed}}")
+    if not isinstance(delay.get("mean"), str) or _dur_or_400(delay["mean"], f"{w}: delay.mean") <= timedelta(0):
+        raise TriggerError(f"{w}: delay.mean must be a positive ISO-8601 duration")
+    if not isinstance(delay.get("seed"), int) or isinstance(delay.get("seed"), bool):
+        raise TriggerError(f"{w}: a stochastic delay requires an integer 'seed'")
+
+
+def _delay_rngs(spec: dict) -> dict[int, random.Random]:
+    """A seeded RNG per stochastic delay, so its draws repeat run to run like a stochastic `every`'s."""
+    return {i: random.Random(action["delay"]["seed"]) for i, action in enumerate(spec["actions"])
+            if isinstance(action.get("delay"), dict)}
+
+
+def _normalize_unless(unless: Any, w: str) -> None:
+    """Validate a deadline's `unless`: the call that cancels it, matched like an action trigger's."""
+    if not isinstance(unless, dict) or not isinstance(unless.get("tool"), str) or not unless["tool"]:
+        raise TriggerError(f"{w}: when.unless must be an object with a non-empty 'tool'")
+    where = unless.get("where", {})
+    if not isinstance(where, dict):
+        raise TriggerError(f"{w}: when.unless.where must be an object")
+    for field, pred in where.items():
+        _validate_predicate(pred, f"{w}: unless.where.{field}")
+        _, path = _ctx_path(field)
+        if path:
+            _validate_path(path, f"{w}: unless.where.{field}")
+    roles = unless.get("roles")
+    if roles is not None and (not isinstance(roles, list) or not roles
+                              or not all(isinstance(r, str) and r for r in roles)):
+        raise TriggerError(f"{w}: when.unless.roles must be a non-empty list of non-empty AgentEnv-Roles")
+
+
+def _normalize_hours(raw: Any, where: str) -> dict:
+    """Validate an executor's working hours, returning them with `utc_offset` and `days` filled in."""
+    if not isinstance(raw, dict):
+        raise TriggerError(f"{where}.hours must be an object {{start, end, utc_offset?, days?}}")
+    for key in ("start", "end"):
+        if not isinstance(raw.get(key), str) or not _HHMM_RE.fullmatch(raw[key]):
+            raise TriggerError(f"{where}.hours.{key} must be a 24-hour 'HH:MM'")
+    if raw["start"] == raw["end"]:
+        raise TriggerError(f"{where}.hours: start equals end, so the executor would never be available")
+    offset = raw.get("utc_offset", "+00:00")
+    if not isinstance(offset, str) or not _UTC_OFFSET_RE.fullmatch(offset):
+        raise TriggerError(f"{where}.hours.utc_offset must look like '-07:00' or '+05:30'")
+    days = raw.get("days", list(_DAYS))
+    if not isinstance(days, list) or not days or not all(day in _DAYS for day in days):
+        raise TriggerError(f"{where}.hours.days must be a non-empty list of {list(_DAYS)}")
+    return {"start": raw["start"], "end": raw["end"], "utc_offset": offset, "days": [d for d in _DAYS if d in days]}
+
+
+def _next_opening(now: datetime, hours: dict) -> datetime | None:
+    """None if ``now`` is within the working hours, else when they next open. A window starts on each listed day and
+    runs to `end`, past midnight when `end` is earlier than `start`."""
+    sign, hh, mm = _UTC_OFFSET_RE.fullmatch(hours["utc_offset"]).groups()
+    local = now.astimezone(timezone((1 if sign == "+" else -1) * timedelta(hours=int(hh), minutes=int(mm))))
+    start_h, start_m = map(int, hours["start"].split(":"))
+    end_h, end_m = map(int, hours["end"].split(":"))
+    length = (timedelta(hours=end_h, minutes=end_m) - timedelta(hours=start_h, minutes=start_m)) % timedelta(days=1)
+    midnight = local.replace(hour=0, minute=0, second=0, microsecond=0)
+    for days_ahead in range(-1, 8):
+        day = midnight + timedelta(days=days_ahead)
+        if _DAYS[day.weekday()] not in hours["days"]:
+            continue
+        opens = day.replace(hour=start_h, minute=start_m)
+        if opens <= local < opens + length:
+            return None
+        if opens > local:
+            return opens
+    return None
diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py
index 3528a0df..d0e7185c 100644
--- a/src/agent_env/task_step/task_steps/prompt_agent.py
+++ b/src/agent_env/task_step/task_steps/prompt_agent.py
@@ -913,6 +913,8 @@ def _is_settling(state: dict) -> bool:
             # `pending` too: the queue itself, not a label, so it outlives a status we do not know.
             if t.get("status") not in TRIGGER_IN_FLIGHT_STATUSES and not t.get("pending"):
                 continue
+            if t.get("waiting"):  # waiting out a delay or its executor's hours, which can outlast the poll
+                continue
             when = t.get("when")
             if when is None:
                 # Older gateway with no ``when``: a scheduled next arrival is the only tell
diff --git a/src/agent_env/task_step/task_steps/register_env_triggers.py b/src/agent_env/task_step/task_steps/register_env_triggers.py
index 78b1f7c3..56d6d192 100644
--- a/src/agent_env/task_step/task_steps/register_env_triggers.py
+++ b/src/agent_env/task_step/task_steps/register_env_triggers.py
@@ -9,7 +9,14 @@
 from agentenv_protocol import client as protocol_v1
 
 from agent_env.env.env import DeployedEnv, DeployedSandboxEnv
-from agent_env.env.gateway.constants import EXT_TRIGGERS_URI, TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES
+from agent_env.env.gateway.constants import (
+    EXT_TRIGGERS_URI,
+    TRIGGER_ACTION_DELAY,
+    TRIGGER_EXECUTOR_HOURS,
+    TRIGGER_NAMED_EXECUTORS,
+    TRIGGER_PROVOKING_ROLES,
+    TRIGGER_UNLESS,
+)
 from agent_env.task_step.context import TaskStepContext
 from agent_env.entity_refs import EntityRef
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
@@ -20,7 +27,8 @@
 class RegisterEnvTriggersStep(TaskStep):
     """An nl action may name its `executor`, and a trigger's `when.roles` may list executor roles so personas react to
     each other. Such a chain ends when a persona's turn makes no matching write, so give conversational nl actions no
-    `verify` (its retry forces a write), give persona prompts a stop condition, and prefer a `where` off-ramp."""
+    `verify` (its retry forces a write), give persona prompts a stop condition, and prefer a `where` off-ramp. An
+    action's `delay` and `executor_hours` pace replies in virtual time, and a time trigger's `unless` makes it a deadline."""
 
     type: ClassVar[str] = "register_env_triggers"
     entity_refs = (EntityRef.env("env_id"),)
@@ -34,6 +42,7 @@ def __init__(
         watch_roles: Optional[list[str]] = None,
         executor_agent_name: Optional[str] = None,
         executor_timeout_seconds: float = 120,
+        executor_hours: Optional[dict[str, dict]] = None,
         depends_on: Optional[list[TaskStepDependency]] = None,
         fail_task_on_error: bool = True,
     ):
@@ -45,6 +54,7 @@ def __init__(
         self.watch_roles = watch_roles
         self.executor_agent_name = executor_agent_name
         self.executor_timeout_seconds = executor_timeout_seconds
+        self.executor_hours = executor_hours
 
     def to_dict(self) -> dict:
         base = super().to_dict()
@@ -53,6 +63,7 @@ def to_dict(self) -> dict:
         base["watch_roles"] = self.watch_roles
         base["executor_agent_name"] = self.executor_agent_name
         base["executor_timeout_seconds"] = self.executor_timeout_seconds
+        base["executor_hours"] = self.executor_hours
         return base
 
     @classmethod
@@ -64,6 +75,7 @@ def from_dict(cls, data: dict) -> RegisterEnvTriggersStep:
             watch_roles=data.get("watch_roles"),
             executor_agent_name=data.get("executor_agent_name"),
             executor_timeout_seconds=data.get("executor_timeout_seconds", 120),
+            executor_hours=data.get("executor_hours"),
         )
 
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
@@ -76,9 +88,15 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         if self.executor_agent_name is not None:
             body["executor"] = self._executor_config(context, deployed, self.executor_agent_name)
         named = _named_executors(self.triggers)
-        # An older gateway ignores what it does not declare: named actions would go to the default executor, and a
-        # trigger's when.roles would leave it to the watched roles, so a persona never provokes it.
-        used = {TRIGGER_NAMED_EXECUTORS: bool(named), TRIGGER_PROVOKING_ROLES: _lists_roles(self.triggers)}
+        unused_hours = sorted(set(self.executor_hours or {}) - {self.executor_agent_name, *named})
+        if unused_hours:
+            raise RuntimeError(f"executor_hours names {unused_hours}, which no nl action or executor_agent_name uses")
+        # An older gateway ignores what it does not declare: named actions would go to the default executor, a
+        # trigger's when.roles would leave it to the watched roles, delays and hours would not wait, and a deadline
+        # would fire even after its `unless` call.
+        used = {TRIGGER_NAMED_EXECUTORS: bool(named), TRIGGER_PROVOKING_ROLES: _when_has(self.triggers, "roles"),
+                TRIGGER_ACTION_DELAY: _delays(self.triggers), TRIGGER_EXECUTOR_HOURS: bool(self.executor_hours),
+                TRIGGER_UNLESS: _when_has(self.triggers, "unless")}
         missing = [feature for feature, needed in used.items() if needed and feature not in _trigger_features(deployed)]
         if missing:
             raise RuntimeError(
@@ -116,7 +134,10 @@ def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agen
         # The env's gateway calls the executor; an env outside our sandboxes gets its URL as it is.
         executor_url = (agent.url_for(deployed.sandbox_type) if isinstance(deployed, DeployedSandboxEnv)
                         else agent.a2a_url or agent.api_url)
-        return {"a2a_url": executor_url, "timeout_seconds": self.executor_timeout_seconds, "role": agent.role}
+        config = {"a2a_url": executor_url, "timeout_seconds": self.executor_timeout_seconds, "role": agent.role}
+        if agent_name in (self.executor_hours or {}):
+            config["hours"] = self.executor_hours[agent_name]
+        return config
 
 
 def _named_executors(triggers: list[dict]) -> list[str]:
@@ -126,8 +147,13 @@ def _named_executors(triggers: list[dict]) -> list[str]:
     return list(dict.fromkeys(name for name in names if isinstance(name, str) and name))
 
 
-def _lists_roles(triggers: list[dict]) -> bool:
-    return any(isinstance(trigger.get("when"), dict) and trigger["when"].get("roles") for trigger in triggers)
+def _when_has(triggers: list[dict], key: str) -> bool:
+    return any(isinstance(trigger.get("when"), dict) and trigger["when"].get(key) for trigger in triggers)
+
+
+def _delays(triggers: list[dict]) -> bool:
+    return any(isinstance(action, dict) and action.get("delay") is not None
+               for trigger in triggers for action in trigger.get("actions") or [])
 
 
 def _trigger_features(deployed: DeployedEnv) -> list:
diff --git a/tst/env/gateway/clock_test.py b/tst/env/gateway/clock_test.py
index 4500c63a..a431238f 100644
--- a/tst/env/gateway/clock_test.py
+++ b/tst/env/gateway/clock_test.py
@@ -260,3 +260,19 @@ def test_saturated_clock_stays_monotonic():
         seen.append(c.read()["virtual_time"])
     assert seen == sorted(seen)
     assert seen[-1] == "9999-12-31T23:59:59.999999Z"
+
+
+def test_elapsed_runs_at_each_arm_s_rate_and_counts_a_re_arm_forward_but_not_back():
+    now = FakeNow()
+    c = Clock(now_fn=now)
+    now.tick(10)  # unarmed: real time
+    c.set_time(T0, rate=60)
+    now.tick(2)  # 120 virtual s, to 00:02:00
+    c.set_time("2026-01-01T01:00:00Z", rate=60)  # 58 virtual min ahead
+    now.tick(1)
+    c.set_time("2025-12-31T00:00:00Z", rate=0)  # a day back, frozen
+    now.tick(100)
+    c.clear()
+    now.tick(5)  # unarmed again
+    assert c.elapsed() == 10 + 120 + 58 * 60 + 60 + 0 + 5
+
diff --git a/tst/integration/task_step/test_register_env_triggers_local.py b/tst/integration/task_step/test_register_env_triggers_local.py
index eec5c734..557c1ba9 100644
--- a/tst/integration/task_step/test_register_env_triggers_local.py
+++ b/tst/integration/task_step/test_register_env_triggers_local.py
@@ -1,6 +1,7 @@
 """Named nl-trigger executors on a local gateway: each nl action reaches only the executor it names, that executor acts
-in the env as its own role, and personas react to each other through `when.roles`. Requires a docker daemon; spins up a
-throwaway ``registry:2``."""
+in the env as its own role, personas react to each other through `when.roles`, replies wait out their delay and working
+hours in virtual time, deadlines fire on silence, and the trajectory names who made each call. Requires a docker daemon;
+spins up a throwaway ``registry:2``."""
 
 import asyncio
 import functools
@@ -8,6 +9,7 @@
 import threading
 import time
 import uuid
+from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 import httpx
@@ -119,6 +121,80 @@ async def test_personas_react_to_each_other_until_one_stays_quiet(local_stack):
                          ("denise-hears-bob", "bob@example.com"), ("bob-hears-denise", "denise@example.com")]
 
 
+@pytest.mark.asyncio
+async def test_a_reply_waits_its_delay_and_hours_and_its_writes_name_the_fire(local_stack):
+    _seed_bootstrap_envs()
+    env, deployed, host = await _deploy_items("pace")
+    started: list[_Executor] = []
+    triggers = [{"id": "bob-hears-solver", "when": {"type": "action", "tool": "list_items"},
+                 "actions": [{"type": "nl", "executor": "bob", "instruction": "BOB-ACTS", "delay": "PT2M"}]}]
+    try:
+        bob = _Executor("bob", "bob@example.com", deployed.gateway_url, host, started)
+        context = TaskStepContext(deployed_envs=[deployed], deployed_agents=[bob.agent], metadata={})
+        async with httpx.AsyncClient(base_url=deployed.gateway_url, timeout=60) as client:
+            await RegisterEnvTriggersStep(id="reg", version=None, env_id=env.id, triggers=triggers, watch_roles=["default"],
+                                          executor_timeout_seconds=60,
+                                          executor_hours={"bob": {"start": "09:00", "end": "17:00"}}).execute(context)
+            # At 60x the delay ends about 08:52, and bob's hours hold him to 09:00: some 10 real seconds in all.
+            await _arm_clock(client, "2026-10-14T08:50:00Z", 60)
+            (await client.post("/step", json={"action": "call_tool", "tool_name": "list_items", "arguments": {}})).raise_for_status()
+            state = await _quiet(client, {"bob-hears-solver": 1})
+            trajectory = [json.loads(line) for line in (await client.get("/trajectory")).raise_for_status().text.splitlines()]
+    finally:
+        for executor in started:
+            executor.stop()
+        await env.close()
+
+    paced = [e for e in state["events"] if e["kind"] in ("delayed", "deferred", "action_ok")]
+    assert [e["kind"] for e in paced] == ["delayed", "deferred", "action_ok"]
+    delayed, deferred, _ = paced
+    assert deferred["until"] == "2026-10-14T09:00:00Z"
+    assert (_virtual(deferred) - _virtual(delayed)).total_seconds() >= 120
+    calls = [e for e in trajectory if e.get("event_type") == "tool_call"]
+    cause = [{"trigger_id": "bob-hears-solver", "action_index": 0, "fire": 1}]
+    assert ("list_items", "default", None) in [(e["tool_call"]["function_name"], e["role"], e.get("caused_by"))
+                                               for e in calls]
+    bobs = [e for e in calls if e["role"] == "bob@example.com"]
+    assert [(e["tool_call"]["function_name"], e.get("caused_by")) for e in bobs] == [("items_whoami", cause),
+                                                                                     ("items_add_item", cause)]
+    assert all(_virtual(e) >= datetime(2026, 10, 14, 9, tzinfo=timezone.utc) for e in bobs)
+
+
+@pytest.mark.asyncio
+async def test_a_deadline_fires_on_silence_and_a_reply_cancels_it(local_stack):
+    _seed_bootstrap_envs()
+    env, deployed, host = await _deploy_items("deadline")
+    started: list[_Executor] = []
+    triggers = []
+    for n in (1, 2):
+        triggers += [{"id": f"dm-{n}", "when": {"type": "action", "tool": "list_items"}, "actions": []},
+                     {"id": f"deadline-{n}", "when": {"type": "time", "after": f"dm-{n}", "offset": "PT2M",
+                                                      "unless": {"tool": "items_add_item", "roles": ["default"],
+                                                                 "where": {"item": {"regex": f"^REPLY-{n}"}}}},
+                      "actions": [{"type": "nl", "executor": "bob", "instruction": f"ESCALATE-{n}"}]}]
+    try:
+        bob = _Executor("bob", "bob@example.com", deployed.gateway_url, host, started)
+        context = TaskStepContext(deployed_envs=[deployed], deployed_agents=[bob.agent], metadata={})
+        async with httpx.AsyncClient(base_url=deployed.gateway_url, timeout=60) as client:
+            await _arm_clock(client, "2026-10-14T10:00:00Z", 60)
+            await RegisterEnvTriggersStep(id="reg", version=None, env_id=env.id, triggers=triggers, watch_roles=["default"],
+                                          executor_timeout_seconds=60).execute(context)
+            (await client.post("/step", json={"action": "call_tool", "tool_name": "list_items", "arguments": {}})).raise_for_status()
+            (await client.post("/step", json={"action": "call_tool", "tool_name": "items_add_item",
+                                              "arguments": {"item": "REPLY-2"}})).raise_for_status()
+            state = await _quiet(client, {"dm-1": 1, "dm-2": 1, "deadline-1": 1, "deadline-2": 0})
+    finally:
+        for executor in started:
+            executor.stop()
+        await env.close()
+
+    statuses = {t["id"]: t["status"] for t in state["triggers"]}
+    assert (statuses["deadline-1"], statuses["deadline-2"]) == ("fired", "cancelled"), state["events"][-8:]
+    assert bob.instructions == ["ESCALATE-1"]
+    [cancelled] = [e for e in state["events"] if e["kind"] == "cancelled"]
+    assert (cancelled["trigger_id"], cancelled["provoking"]["role"]) == ("deadline-2", "default")
+
+
 class _Executor:
     """An A2A executor on this machine that acts on each instruction through the gateway, under its own role; with
     ``replies``, it stops acting after that many."""
@@ -193,3 +269,22 @@ async def _quiet(client: httpx.AsyncClient, fires: dict[str, int], timeout_s: fl
         if time.monotonic() > deadline:
             raise AssertionError(f"chain did not settle: { {tid: t['fire_count'] for tid, t in triggers.items()} }")
         await asyncio.sleep(1)
+
+
+async def _deploy_items(label: str):
+    """A MultiEnv of the in-repo items server on local compose, and the host address its gateway reaches this machine at."""
+    uid = uuid.uuid4().hex[:8]
+    items = MCPServerEnv.put(id=f"{label}-items-{uid}", environment_name="items",
+                             docker_image_artifact=_put_items_image(f"{label}-items-{uid}"))
+    env = MultiEnv.put(id=f"{label}-{uid}", mcp_server_envs=[items], name="suite")
+    deployed = await env.deploy(sandbox_type="local", ttl_seconds=900)
+    return env, deployed, env._sandbox.host_ips[-1]
+
+
+def _virtual(event: dict) -> datetime:
+    return datetime.fromisoformat(event["virtual_time"].replace("Z", "+00:00"))
+
+
+async def _arm_clock(client: httpx.AsyncClient, virtual_time: str, rate: float) -> None:
+    (await client.put("/clock/set-time", json={"virtual_time": virtual_time,
+                                                "virtual_seconds_per_real_second": rate})).raise_for_status()
diff --git a/tst/unit/env/gateway/role_forwarding_test.py b/tst/unit/env/gateway/role_forwarding_test.py
index a00ceb93..619ed1bd 100644
--- a/tst/unit/env/gateway/role_forwarding_test.py
+++ b/tst/unit/env/gateway/role_forwarding_test.py
@@ -81,6 +81,29 @@ def call(role=None, key=None):
     ]
 
 
+def test_mcp_tool_calls_record_who_made_them_and_the_fires_in_flight_for_that_executor():
+    session = _RecordingSession()
+    gw = _gateway(session)
+    gw._create_and_register_proxy_tool(URL, TOOL)
+    gw._tool_server_urls[TOOL.name] = URL
+    gw._clock.set_time("2019-06-15T12:00:00Z", 1)
+    gw._register_get_time_tool()
+    cause = {"trigger_id": "x", "action_index": 0, "fire": 1}
+    gw._trigger_engine._in_flight["bob@example.com"] = [cause]
+    before = len(_trajectory(gw))
+
+    for role in ("bob@example.com", "alice@example.com"):
+        for tool in (TOOL.name, GET_TIME_TOOL_NAME):  # a proxied tool, and one the gateway serves itself
+            ctx = contextvars.copy_context()
+            ctx.run(Caller(role=role).bind)
+            ctx.run(asyncio.run, gw._mcp._tool_manager.call_tool(tool, {}))
+
+    calls = [e for e in _trajectory(gw)[before:] if e["event_type"] == "tool_call"]
+    assert [(e["tool_call"]["function_name"], e["role"], e.get("caused_by")) for e in calls] == [
+        (TOOL.name, "bob@example.com", [cause]), (GET_TIME_TOOL_NAME, "bob@example.com", [cause]),
+        (TOOL.name, "alice@example.com", None), (GET_TIME_TOOL_NAME, "alice@example.com", None)]
+
+
 @pytest.mark.asyncio
 async def test_step_stamps_the_header_role_in_meta_including_on_the_reconnect():
     session = _RecordingSession(failures=[anyio.ClosedResourceError()])
diff --git a/tst/unit/env/gateway/triggers_test.py b/tst/unit/env/gateway/triggers_test.py
index a70cd54b..5336a48f 100644
--- a/tst/unit/env/gateway/triggers_test.py
+++ b/tst/unit/env/gateway/triggers_test.py
@@ -3,13 +3,16 @@
 
 import asyncio
 import json
+import random
+from datetime import datetime, timezone
 
 import httpx
 import pytest
 from mcp.types import CallToolResult, TextContent
 
 from agent_env.env.gateway.clock import Clock
-from agent_env.env.gateway.triggers import TriggerEngine, TriggerError
+from agent_env.env.gateway import triggers as triggers_module
+from agent_env.env.gateway.triggers import TriggerEngine, TriggerError, _next_opening
 
 
 class _FakeGateway:
@@ -89,6 +92,20 @@ async def fake_executor(self, executor, instruction, context_id):
     return calls
 
 
+@pytest.fixture
+def waits(engine, monkeypatch):
+    """Every virtual-time wait the engine makes, in virtual seconds, on an unarmed clock whose source only the waits
+    advance, so none of them actually sleeps."""
+    return _fake_waits(engine, monkeypatch)
+
+
+@pytest.fixture
+def virtual_time(engine, monkeypatch):
+    """The engine's clock armed at 2026-10-14T08:00Z (a Wednesday) on a fake source; each wait advances it at once
+    and is recorded in virtual seconds, and `.advance(s)` moves it by `s` virtual seconds."""
+    return _fake_waits(engine, monkeypatch, armed_at="2026-10-14T08:00:00Z")
+
+
 def _registration(**overrides):
     base = {
         "watch_roles": ["default"],
@@ -162,6 +179,31 @@ def _registration(**overrides):
     ({"executors": {"bob": {"a2a_url": "http://b", "role": "bob"}},
       "triggers": [{"id": "x", "when": {"type": "action", "tool": "t"},
                     "actions": [{"type": "tool", "tool": "t2", "executor": "bob"}]}]}, "only valid on nl actions"),
+    *[({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t"},
+                      "actions": [{"type": "tool", "tool": "t2", "delay": delay}]}]}, fragment)
+      for delay, fragment in [("soon", "delay"), ("PT0S", "delay must be a positive duration"),
+                              ({"dist": "uniform"}, "delay must be an ISO-8601 duration or"),
+                              ({"dist": "exp", "mean": "PT1M"}, "requires an integer 'seed'"),
+                              ({"dist": "exp", "mean": "PT0S", "seed": 1}, "delay.mean must be a positive")]],
+    ({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t"}, "barrier": {"at": "provoking_call"},
+                    "actions": [{"type": "tool", "tool": "t2", "delay": "PT1M"}]}]}, "can't run under a barrier"),
+    ({"executors": {"bob": {"a2a_url": "http://b", "role": "bob", "hours": {"start": "09:00", "end": "17:00"}}},
+      "triggers": [{"id": "x", "when": {"type": "action", "tool": "t"}, "barrier": {"at": "provoking_call"},
+                    "actions": [{"type": "nl", "executor": "bob", "instruction": "hi"}]}]}, "can't run under a barrier"),
+    *[({"executors": {"bob": {"a2a_url": "http://b", "role": "bob", "hours": hours}}, "triggers": []}, fragment)
+      for hours, fragment in [({"start": "9:00", "end": "17:00"}, "hours.start must be a 24-hour"),
+                              ({"start": "09:00", "end": "09:00"}, "would never be available"),
+                              ({"start": "09:00", "end": "17:00", "utc_offset": "PST"}, "utc_offset must look like"),
+                              ({"start": "09:00", "end": "17:00", "days": ["monday"]}, "hours.days must be"),
+                              ({"start": "09:00", "end": "17:00", "days": [["mon"]]}, "hours.days must be")]],
+    ({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "unless": {"tool": "t2"}}, "actions": []}]},
+     "when.unless is only valid on time triggers"),
+    ({"triggers": [{"id": "x", "when": {"type": "time", "every": "PT1H", "unless": {"tool": "t2"}}, "actions": []}]},
+     "only valid on a one-shot time trigger"),
+    ({"triggers": [{"id": "x", "when": {"type": "time", "at": "PT1H", "unless": {"where": {}}}, "actions": []}]},
+     "must be an object with a non-empty 'tool'"),
+    ({"triggers": [{"id": "x", "when": {"type": "time", "at": "PT1H", "unless": {"tool": "t2", "roles": []}},
+                    "actions": []}]}, "when.unless.roles must be a non-empty list"),
     *[({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "roles": roles}, "actions": []}]},
        "when.roles must be a non-empty list") for roles in ("denise", [], [""], [5])],
     ({"triggers": [{"id": "x", "when": {"type": "time", "after": "y", "offset": "PT1S", "roles": ["denise"]},
@@ -389,6 +431,301 @@ async def test_an_executor_write_prompts_only_the_state_triggers_that_list_it(en
     assert {tid: t["status"] for tid, t in engine._triggers.items()} == {"listed": "fired", "unlisted": "armed"}
 
 
+@pytest.mark.asyncio
+async def test_a_delay_waits_its_virtual_time_before_the_action(engine, executor_calls, virtual_time):
+    engine.register({"executors": {"bob": {"a2a_url": "http://bob", "role": "bob"}},
+                     "triggers": [{"id": "x", "when": {"type": "action", "tool": "t"},
+                                   "actions": [{"type": "nl", "executor": "bob", "instruction": "reply",
+                                                "delay": "PT20M"}]}]})
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.2)
+
+    assert virtual_time == [1200.0] and len(executor_calls) == 1
+    [delayed] = [e for e in engine._events if e["kind"] == "delayed"]
+    [acted] = [e for e in engine._events if e["kind"] == "action_ok"]
+    assert (delayed["virtual_seconds"], delayed["executor"]) == (1200.0, "bob")
+    assert (delayed["virtual_time"], acted["virtual_time"]) == ("2026-10-14T08:00:00Z", "2026-10-14T08:20:00Z")
+
+
+@pytest.mark.asyncio
+async def test_queued_calls_each_wait_out_their_delay_from_their_own_call(engine, virtual_time):
+    engine.register({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "repeat": True},
+                                   "actions": [{"type": "tool", "tool": "slack_send_message",
+                                                "args": {"channel": "#c", "text": "hi"}, "delay": "PT1H"}]}]})
+    for _ in range(3):  # 08:00, 08:01 and 08:02, all before the first reply's turn
+        engine.on_tool_call("default", "t", {}, _result())
+        virtual_time.advance(60)
+    await asyncio.sleep(0.2)
+
+    replies = [e["virtual_time"] for e in engine._events if e["kind"] == "action_ok"]
+    assert replies == ["2026-10-14T09:00:00Z", "2026-10-14T09:01:00Z", "2026-10-14T09:02:00Z"]
+
+
+@pytest.mark.asyncio
+async def test_due_marks_wait_their_delay_from_the_mark_not_from_the_last_reply(engine, virtual_time):
+    engine.register({"triggers": [{"id": "x", "when": {"type": "time", "every": "PT1H"},
+                                   "actions": [{"type": "tool", "tool": "slack_send_message",
+                                                "args": {"channel": "#c", "text": "hi"}, "delay": "PT30M"}]}]})
+    engine._eval_time_triggers()
+    virtual_time.advance(3 * 3600)  # 11:00: the 09:00, 10:00 and 11:00 marks come due together
+    engine._eval_time_triggers()
+    await asyncio.sleep(0.2)
+
+    replies = [e["virtual_time"] for e in engine._events if e["kind"] == "action_ok"]
+    assert replies == ["2026-10-14T11:00:00Z", "2026-10-14T11:00:00Z", "2026-10-14T11:30:00Z"]
+
+
+@pytest.mark.asyncio
+async def test_a_stochastic_delay_draws_the_same_sequence_from_its_seed(engine, waits):
+    engine.register({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "repeat": True},
+                                   "actions": [{"type": "tool", "tool": "slack_send_message",
+                                                "args": {"channel": "#c", "text": "hi"},
+                                                "delay": {"dist": "exp", "mean": "PT10M", "seed": 7}}]}]})
+    for _ in range(3):
+        engine.on_tool_call("default", "t", {}, _result())
+        await asyncio.sleep(0.1)
+
+    rng = random.Random(7)
+    assert waits == [rng.expovariate(1.0 / 600) for _ in range(3)]
+    assert len(engine._test_state["slack"]) == 3
+
+
+@pytest.mark.asyncio
+async def test_a_re_arm_during_one_mark_s_delay_does_not_shorten_the_next(engine, virtual_time, monkeypatch):
+    engine.register({"triggers": [{"id": "x", "when": {"type": "time", "every": "PT1H"},
+                                   "actions": [{"type": "tool", "tool": "slack_send_message",
+                                                "args": {"channel": "#c", "text": "hi"}, "delay": "PT90M"}]}]})
+    record = TriggerEngine._sleep_virtual
+
+    async def re_arm_a_day_back_after_the_first_wait(self, seconds, generation=None):
+        await record(self, seconds, generation)
+        if len(virtual_time) == 1:
+            self._gw._clock.set_time("2026-10-13T10:30:00Z")
+
+    monkeypatch.setattr(TriggerEngine, "_sleep_virtual", re_arm_a_day_back_after_the_first_wait)
+    engine._eval_time_triggers()
+    virtual_time.advance(2 * 3600)  # 10:00: the 09:00 and 10:00 marks come due together
+    engine._eval_time_triggers()
+    await asyncio.sleep(0.2)
+
+    assert virtual_time == [1800.0, 3600.0]
+
+
+@pytest.mark.asyncio
+async def test_working_hours_defer_an_nl_action_until_they_open(engine, executor_calls, virtual_time):
+    engine.register({"executors": {"bob": {"a2a_url": "http://bob", "role": "bob",
+                                           "hours": {"start": "09:00", "end": "17:00"}}},
+                     "triggers": [{"id": "x", "when": {"type": "action", "tool": "t"},
+                                   "actions": [{"type": "nl", "executor": "bob", "instruction": "reply"}]}]})
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.2)
+
+    [deferred] = [e for e in engine._events if e["kind"] == "deferred"]
+    [acted] = [e for e in engine._events if e["kind"] == "action_ok"]
+    assert (deferred["until"], deferred["executor"], acted["virtual_time"]) == (
+        "2026-10-14T09:00:00Z", "bob", "2026-10-14T09:00:00Z")
+    assert virtual_time == [3600.0] and len(executor_calls) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("change, until", [
+    (lambda c: c.set_time("2026-10-19T10:00:00Z", rate=3600), ["2026-10-19T09:00:00Z"]),
+    (lambda c: c.set_time("2026-10-17T12:00:00Z", rate=3600), ["2026-10-19T09:00:00Z"] * 2),
+    (lambda c: c.clear(), ["2026-10-19T09:00:00Z"]),
+], ids=["re-armed-into-hours", "re-armed-still-closed", "cleared"])
+async def test_working_hours_are_reckoned_again_after_a_re_arm_and_lifted_by_a_clear(engine, monkeypatch, change,
+                                                                                      until):
+    stepped = _SteppedClock(engine, monkeypatch)
+    stepped.clock.set_time("2026-10-16T18:00:00Z", rate=3600)  # a Friday evening, closed until Monday 09:00
+    stepped.at[1] = lambda: change(stepped.clock)
+    trig = {"waiting": False}
+    hours = {"start": "09:00", "end": "17:00", "utc_offset": "+00:00", "days": ["mon", "tue", "wed", "thu", "fri"]}
+
+    await engine._await_hours("x", trig, 0, hours, {})
+
+    assert [e["until"] for e in engine._events if e["kind"] == "deferred"] == until and not trig["waiting"]
+    now = stepped.clock.now()
+    assert now is None or now.strftime("%a %H:%M") in ("Mon 09:00", "Mon 10:00")
+
+
+@pytest.mark.asyncio
+async def test_working_hours_do_not_apply_while_the_clock_is_unarmed(engine, executor_calls, waits):
+    engine.register({"executors": {"bob": {"a2a_url": "http://bob", "role": "bob",
+                                           "hours": {"start": "09:00", "end": "09:01"}}},
+                     "triggers": [{"id": "x", "when": {"type": "action", "tool": "t"},
+                                   "actions": [{"type": "nl", "executor": "bob", "instruction": "reply"}]}]})
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.2)
+
+    assert waits == [] and len(executor_calls) == 1
+    assert not [e for e in engine._events if e["kind"] == "deferred"]
+
+
+@pytest.mark.parametrize("now, hours, opens", [
+    ("2026-10-14T08:00:00Z", {"start": "09:00", "end": "17:00"}, "2026-10-14T09:00:00Z"),
+    ("2026-10-14T12:00:00Z", {"start": "09:00", "end": "17:00"}, None),
+    ("2026-10-14T17:00:00Z", {"start": "09:00", "end": "17:00"}, "2026-10-15T09:00:00Z"),
+    ("2026-10-14T15:00:00Z", {"start": "09:00", "end": "17:00", "utc_offset": "-07:00"}, "2026-10-14T16:00:00Z"),
+    ("2026-10-14T16:30:00Z", {"start": "09:00", "end": "17:00", "utc_offset": "-07:00"}, None),
+    ("2026-10-14T23:00:00Z", {"start": "22:00", "end": "06:00"}, None),
+    ("2026-10-15T03:00:00Z", {"start": "22:00", "end": "06:00"}, None),
+    ("2026-10-14T12:00:00Z", {"start": "22:00", "end": "06:00"}, "2026-10-14T22:00:00Z"),
+    ("2026-10-17T12:00:00Z", {"start": "09:00", "end": "17:00", "days": ["mon", "tue", "wed", "thu", "fri"]},
+     "2026-10-19T09:00:00Z"),
+    ("2026-10-17T03:00:00Z", {"start": "22:00", "end": "06:00", "days": ["fri"]}, None),
+], ids=["before", "inside", "at-end", "offset-before", "offset-inside", "wrap-evening", "wrap-morning",
+        "wrap-midday", "weekend", "wrap-from-a-listed-day"])
+def test_next_opening(now, hours, opens):
+    hours = {"utc_offset": "+00:00", "days": ["mon", "tue", "wed", "thu", "fri", "sat", "sun"], **hours}
+    got = _next_opening(datetime.fromisoformat(now.replace("Z", "+00:00")), hours)
+    assert (got.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") if got else None) == opens
+
+
+def _deadline(**unless) -> dict:
+    """An anchor `dm` (fired by a call to `t`) and a deadline 30 virtual minutes after it, cancelled by `unless`."""
+    return {"executors": {"bob": {"a2a_url": "http://bob", "role": "bob"}},
+            "triggers": [{"id": "dm", "when": {"type": "action", "tool": "t"}, "actions": []},
+                         {"id": "deadline", "when": {"type": "time", "after": "dm", "offset": "PT30M",
+                                                     "unless": {"tool": "slack_send_message", **unless}},
+                          "actions": []}]}
+
+
+@pytest.mark.asyncio
+async def test_a_matching_call_after_the_anchor_cancels_the_deadline(engine, virtual_time):
+    engine.register(_deadline(where={"channel_id": {"equals": "D1"}}))
+    engine.on_tool_call("default", "slack_send_message", {"channel_id": "D1"}, _result())  # before the anchor
+    assert engine._triggers["deadline"]["status"] == "armed"
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.1)
+    for call, error in ((("default", "slack_send_message", {"channel_id": "D2"}), False),
+                        (("default", "slack_send_message", {"channel_id": "D1"}), True),
+                        (("bob", "slack_send_message", {"channel_id": "D1"}), False)):
+        engine.on_tool_call(*call, _result(is_error=error))
+    assert engine._triggers["deadline"]["status"] == "armed"
+
+    engine.on_tool_call("default", "slack_send_message", {"channel_id": "D1"}, _result())
+    assert engine.state()["triggers"][1]["status"] == "cancelled"
+    [cancelled] = [e for e in engine._events if e["kind"] == "cancelled"]
+    assert cancelled["provoking"] == {"tool": "slack_send_message", "role": "default"}
+
+    virtual_time.advance(3600)
+    engine._eval_time_triggers()
+    await asyncio.sleep(0.1)
+    assert engine._triggers["deadline"]["status"] == "cancelled" and engine._triggers["deadline"]["fire_count"] == 0
+
+
+@pytest.mark.asyncio
+async def test_a_deadline_lists_the_executor_roles_that_cancel_it(engine, virtual_time):
+    engine.register(_deadline(roles=["bob"]))
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.1)
+    engine.on_tool_call("default", "slack_send_message", {}, _result())
+    assert engine._triggers["deadline"]["status"] == "armed"
+    engine.on_tool_call("bob", "slack_send_message", {}, _result())
+    assert engine._triggers["deadline"]["status"] == "cancelled"
+
+
+@pytest.mark.asyncio
+async def test_a_deadline_nobody_cancels_fires_when_due(engine, virtual_time):
+    engine.register(_deadline())
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.1)
+    virtual_time.advance(29 * 60)
+    engine._eval_time_triggers()
+    assert engine._triggers["deadline"]["status"] == "armed"
+    virtual_time.advance(2 * 60)
+    engine._eval_time_triggers()
+    await asyncio.sleep(0.1)
+    assert engine._triggers["deadline"]["status"] == "fired"
+
+
+@pytest.mark.asyncio
+async def test_a_reply_while_the_anchor_is_still_acting_cancels_the_deadline(engine, monkeypatch, virtual_time):
+    async def bob_dms_and_is_answered_before_he_reports_back(self, executor, instruction, context_id):
+        engine.on_tool_call("default", "slack_send_message", {}, _result())
+        return "done"
+
+    monkeypatch.setattr(TriggerEngine, "_call_executor", bob_dms_and_is_answered_before_he_reports_back)
+    registration = _deadline()
+    registration["triggers"][0]["actions"] = [{"type": "nl", "executor": "bob", "instruction": "DM the solver"}]
+    engine.register(registration)
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.1)
+    assert (engine._triggers["dm"]["status"], engine._triggers["deadline"]["status"]) == ("fired", "cancelled")
+    assert not [e for e in engine._events if e["kind"] == "anchored"]
+
+    virtual_time.advance(3600)
+    engine._eval_time_triggers()
+    await asyncio.sleep(0.1)
+    assert engine._triggers["deadline"]["fire_count"] == 0
+
+
+@pytest.mark.asyncio
+async def test_a_reply_after_the_deadline_is_due_comes_too_late(engine, virtual_time):
+    engine.register(_deadline())
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.1)
+    virtual_time.advance(31 * 60)  # due, though nothing has looked yet
+    engine.on_tool_call("default", "slack_send_message", {}, _result())
+    await asyncio.sleep(0.1)
+    assert engine._triggers["deadline"]["status"] == "fired"
+    assert not [e for e in engine._events if e["kind"] == "cancelled"]
+
+
+@pytest.mark.asyncio
+async def test_cancelling_a_deadline_retires_the_triggers_anchored_on_it(engine, virtual_time):
+    engine.register({"triggers": [
+        {"id": "nudge", "when": {"type": "time", "at": "PT1H", "unless": {"tool": "slack_send_message"}}, "actions": []},
+        {"id": "escalate", "when": {"type": "time", "after": "nudge", "offset": "PT1H"}, "actions": []}]})
+    engine.on_tool_call("default", "slack_send_message", {}, _result())
+
+    assert {t["id"]: t["status"] for t in engine.state()["triggers"]} == {"nudge": "cancelled", "escalate": "cancelled"}
+    assert not engine._has_time_triggers
+    cancelled = [e for e in engine._events if e["kind"] == "cancelled"]
+    assert [(e["trigger_id"], e.get("anchor")) for e in cancelled] == [("nudge", None), ("escalate", "nudge")]
+
+
+def test_a_null_unless_means_none(engine):
+    engine.register({"triggers": [{"id": "a", "when": {"type": "action", "tool": "t", "unless": None}, "actions": []},
+                                  {"id": "b", "when": {"type": "time", "at": "PT1H", "unless": None}, "actions": []}]})
+    assert all("unless" not in t["spec"]["when"] for t in engine._triggers.values())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("reset", ["remove", "clear"])
+async def test_removing_or_clearing_a_trigger_mid_delay_ends_its_fire_without_a_trace(engine, reset):
+    engine.register({"triggers": [{"id": "x", "when": {"type": "action", "tool": "t"},
+                                   "actions": [{"type": "tool", "tool": "slack_send_message",
+                                                "args": {"channel": "#c", "text": "hi"}, "delay": "PT1H"}]}]})
+    engine.on_tool_call("default", "t", {}, _result())
+    await asyncio.sleep(0.05)
+    assert engine.state()["triggers"][0]["waiting"] is True
+
+    engine.remove({"ids": ["x"]}) if reset == "remove" else engine.clear()
+    await asyncio.sleep(0.05)
+    assert engine._test_state["slack"] == []
+    expected = ["added", "detected", "delayed", "removed"] if reset == "remove" else []
+    assert [e["kind"] for e in engine._events] == expected
+
+
+@pytest.mark.asyncio
+async def test_each_fire_is_numbered_failed_ones_included(engine, monkeypatch):
+    seen, replies = [], iter([None, "done", "done"])
+
+    async def executor(self, executor, instruction, context_id):
+        seen.append(engine.caused_by("bob"))
+        return next(replies)
+
+    monkeypatch.setattr(TriggerEngine, "_call_executor", executor)
+    engine.register({"executors": {"bob": {"a2a_url": "http://bob", "role": "bob"}},
+                     "triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "repeat": True},
+                                   "actions": [{"type": "nl", "executor": "bob", "instruction": "reply"}]}]})
+    for _ in range(3):
+        engine.on_tool_call("default", "t", {}, _result())
+        await asyncio.sleep(0.05)
+    assert [causes[0]["fire"] for causes in seen] == [1, 2, 3] and engine._triggers["x"]["fire_count"] == 2
+
+
 def test_generated_id_on_omit(engine):
     r = engine.register({"watch_roles": ["default"], "triggers": [
         {"when": {"type": "action", "tool": "t"}, "actions": []}]})
@@ -616,3 +953,115 @@ async def test_step_path_respects_watch_roles(monkeypatch):
 
 class _ExecutorCalls(list):
     reply: object = "done"
+
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("rate, virtual_seconds, steps", [(60, 1200, [5.0, 5.0, 5.0, 5.0]), (None, 12, [5.0, 5.0, 2.0])],
+                         ids=["armed-at-60x", "unarmed"])
+async def test_waiting_virtual_time_follows_the_clock_rate(engine, monkeypatch, rate, virtual_seconds, steps):
+    stepped = _SteppedClock(engine, monkeypatch)
+    if rate is not None:
+        stepped.clock.set_time("2026-10-14T08:00:00Z", rate=rate)
+    await engine._sleep_virtual(virtual_seconds)
+    assert stepped.slept == steps
+
+
+@pytest.mark.asyncio
+async def test_waiting_virtual_time_holds_while_frozen_and_follows_a_new_rate(engine, monkeypatch):
+    stepped = _SteppedClock(engine, monkeypatch)
+    stepped.at[1] = lambda: stepped.clock.set_time("2026-10-14T08:00:00Z", rate=0)  # after 5 unarmed seconds
+    stepped.at[3] = lambda: stepped.clock.set_time("2026-10-14T08:00:00Z", rate=4)  # where it froze, now at 4x
+    await engine._sleep_virtual(13)
+    assert stepped.slept == [5.0, 5.0, 5.0, 2.0]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("re_arm_to, steps", [("2026-10-14T08:00:08Z", [5.0, 2.0]), ("2026-10-13T08:00:00Z", [5.0, 5.0])],
+                         ids=["forward-counts-its-jump", "back-counts-nothing"])
+async def test_waiting_virtual_time_counts_a_re_arm_forward_but_not_back(engine, monkeypatch, re_arm_to, steps):
+    stepped = _SteppedClock(engine, monkeypatch)
+    stepped.clock.set_time("2026-10-14T08:00:00Z")
+    stepped.at[1] = lambda: stepped.clock.set_time(re_arm_to)  # at 08:00:05: 3 s ahead, or a day back
+    await engine._sleep_virtual(10)
+    assert stepped.slept == steps
+
+
+@pytest.mark.asyncio
+async def test_a_wait_tied_to_a_clock_generation_ends_when_the_clock_moves_on(engine, monkeypatch):
+    stepped = _SteppedClock(engine, monkeypatch)
+    stepped.clock.set_time("2026-10-14T08:00:00Z")
+    stepped.at[1] = stepped.clock.clear
+    await engine._sleep_virtual(3600, generation=stepped.clock.generation)
+    assert stepped.slept == [5.0]
+
+
+@pytest.mark.asyncio
+async def test_tool_call_events_carry_the_callers_role_and_the_fire_an_executor_made_them_for(monkeypatch):
+    gw = _step_gateway()
+    events: list[dict] = []
+
+    async def capture(event):
+        events.append(dict(event))
+        return f"event_{len(events)}"
+
+    async def executor(self, executor, instruction, context_id):
+        await gw._step_call_tool({"tool_name": "gdocs_create_document", "arguments": {"title": "by bob"}}, "bob")
+        return "done"
+
+    monkeypatch.setattr(gw, "_log_event", capture)
+    monkeypatch.setattr(TriggerEngine, "_call_executor", executor)
+    gw._trigger_engine.register({
+        "watch_roles": ["default"], "executors": {"bob": {"a2a_url": "http://bob", "role": "bob"}},
+        "triggers": [{"id": "t", "when": {"type": "action", "tool": "gdocs_create_document",
+                                          "where": {"title": {"regex": "solver"}}},
+                      "actions": [{"type": "nl", "executor": "bob", "instruction": "write"}]}]})
+    await gw._step_call_tool({"tool_name": "gdocs_create_document", "arguments": {"title": "by solver"}}, "default")
+    await asyncio.sleep(0.2)
+    await gw._step_call_tool({"tool_name": "gdocs_create_document", "arguments": {"title": "later"}}, "bob")
+
+    calls = [(e["tool_call"]["arguments"]["title"], e["role"], e.get("caused_by"))
+             for e in events if e["event_type"] == "tool_call"]
+    assert calls == [("by solver", "default", None),
+                     ("by bob", "bob", [{"trigger_id": "t", "action_index": 0, "fire": 1}]),
+                     ("later", "bob", None)]
+
+
+class _Waits(list):
+    advance = None
+
+
+def _fake_waits(engine, monkeypatch, armed_at=None) -> _Waits:
+    source = {"t": 0.0}
+    clock = Clock(now_fn=lambda: source["t"])
+    if armed_at is not None:
+        clock.set_time(armed_at)
+    engine._test_gw._clock = clock
+    waited = _Waits()
+    waited.advance = lambda seconds: source.__setitem__("t", source["t"] + seconds / clock.rate)
+
+    async def fake_sleep(self, seconds, generation=None):
+        waited.append(seconds)
+        waited.advance(max(0.0, seconds))
+
+    monkeypatch.setattr(TriggerEngine, "_sleep_virtual", fake_sleep)
+    return waited
+
+
+class _SteppedClock:
+    """The engine's clock on a fake source that asyncio.sleep advances, so the real virtual-time waits run at once;
+    `.at[n]` runs after the n-th sleep."""
+
+    def __init__(self, engine, monkeypatch):
+        self.t = 0.0
+        self.slept: list[float] = []
+        self.at: dict = {}
+        self.clock = engine._test_gw._clock = Clock(now_fn=lambda: self.t)
+
+        async def fake_sleep(seconds):
+            self.slept.append(seconds)
+            assert len(self.slept) < 100, "the wait never ended"
+            self.t += seconds
+            self.at.pop(len(self.slept), lambda: None)()
+
+        monkeypatch.setattr(triggers_module.asyncio, "sleep", fake_sleep)
diff --git a/tst/unit/task_step/test_prompt_agent_triggers.py b/tst/unit/task_step/test_prompt_agent_triggers.py
index 86c55bb1..a34bda4a 100644
--- a/tst/unit/task_step/test_prompt_agent_triggers.py
+++ b/tst/unit/task_step/test_prompt_agent_triggers.py
@@ -477,6 +477,13 @@ def test_settle_poll_still_waits_for_a_bounded_recurrence():
     assert PromptAgentTaskStep._is_settling(unbounded) is False
 
 
+def test_settle_poll_does_not_wait_out_a_delay_or_working_hours():
+    waiting = {"triggers": [{"id": "reply", "type": "action", "status": "firing", "waiting": True,
+                             "when": {"type": "action", "tool": "x"}}]}
+    assert PromptAgentTaskStep._is_settling(waiting) is False
+    assert PromptAgentTaskStep._capture_is_final(waiting) is False
+
+
 @pytest.mark.asyncio
 async def test_capture_noop_without_registrations():
     calls = {"n": 0}
@@ -588,7 +595,7 @@ def put(self, *args, **kwargs):
 
 # The other half of the partition. It lives here rather than in the engine or the step because it is
 # a ledger for the test below, not something either of them needs at runtime.
-_KNOWN_SETTLED_STATUSES = ("armed", "fired", "failed")
+_KNOWN_SETTLED_STATUSES = ("armed", "fired", "failed", "cancelled")
 
 
 def test_the_capture_path_classifies_every_status_the_engine_can_emit():
diff --git a/tst/unit/task_step/test_register_env_triggers.py b/tst/unit/task_step/test_register_env_triggers.py
index fe405c54..d6a41323 100644
--- a/tst/unit/task_step/test_register_env_triggers.py
+++ b/tst/unit/task_step/test_register_env_triggers.py
@@ -39,13 +39,15 @@ def _step(**overrides):
 
 
 def test_round_trips_through_to_dict_from_dict():
-    step = _step(watch_roles=["default"], executor_agent_name="exec-1", executor_timeout_seconds=90)
+    step = _step(watch_roles=["default"], executor_agent_name="exec-1", executor_timeout_seconds=90,
+                 executor_hours={"exec-1": _HOURS})
     rebuilt = RegisterEnvTriggersStep.from_dict(step.to_dict())
     assert rebuilt.env_id == "env-x"
     assert rebuilt.triggers == _TRIGGERS
     assert rebuilt.watch_roles == ["default"]
     assert rebuilt.executor_agent_name == "exec-1"
     assert rebuilt.executor_timeout_seconds == 90
+    assert rebuilt.executor_hours == {"exec-1": _HOURS}
 
 
 def test_bad_triggers_rejected_at_construction():
@@ -219,6 +221,52 @@ async def test_a_gateway_without_provoking_roles_refuses_a_trigger_that_lists_ro
     assert sent == []
 
 
+_HOURS = {"start": "09:00", "end": "17:00", "utc_offset": "-07:00"}
+
+
+@pytest.mark.asyncio
+async def test_executor_hours_ride_on_the_executors_they_name(monkeypatch):
+    sent = _mock_gateway(monkeypatch, {"ok": True, "added": ["a", "b", "a2", "d"], "all": ["a", "b", "a2", "d"]})
+    context = TaskStepContext(deployed_envs=[_record()], deployed_agents=[
+        DeployedAgent(agent_name="exec-1", api_url="http://exec:9000", role="executor"),
+        DeployedAgent(agent_name="alice", api_url="http://alice:9000", role="alice@example.com"),
+        DeployedAgent(agent_name="bob", api_url="http://bob:9000", role="bob@example.com"),
+    ], metadata={})
+
+    await _step(triggers=_NAMED_TRIGGERS, executor_agent_name="exec-1", executor_hours={"bob": _HOURS}).execute(context)
+
+    body = json.loads(sent[0].content)
+    assert "hours" not in body["executor"] and "hours" not in body["executors"]["alice"]
+    assert body["executors"]["bob"]["hours"] == _HOURS
+
+
+@pytest.mark.asyncio
+async def test_executor_hours_for_an_agent_no_trigger_uses_raise(monkeypatch):
+    sent = _mock_gateway(monkeypatch, {})
+    with pytest.raises(RuntimeError, match=r"executor_hours names \['carol'\]"):
+        await _step(executor_hours={"carol": _HOURS}).execute(_context())
+    assert sent == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("triggers, hours, missing", [
+    ([{"id": "x", "when": {"type": "action", "tool": "t"},
+       "actions": [{"type": "permission", "action": "enable", "role": "r", "tools": ["t"], "delay": "PT5M"}]}],
+     None, "action_delay"),
+    ([{"id": "x", "when": {"type": "time", "after": "y", "offset": "PT5M", "unless": {"tool": "t"}}, "actions": []}],
+     None, "unless"),
+    ([{"id": "x", "when": {"type": "action", "tool": "t"}, "actions": [{"type": "nl", "instruction": "x"}]}],
+     {"exec-1": _HOURS}, "executor_hours"),
+], ids=["delay", "unless", "hours"])
+async def test_a_gateway_without_a_feature_refuses_the_triggers_that_use_it(monkeypatch, triggers, hours, missing):
+    sent = _mock_gateway(monkeypatch, {})
+    context = _context(with_executor=True)
+    context.deployed_envs = [_record(_with_features("named_executors", "provoking_roles"))]
+    with pytest.raises(RuntimeError, match=rf"does not declare \['{missing}'\]"):
+        await _step(triggers=triggers, executor_agent_name="exec-1", executor_hours=hours).execute(context)
+    assert sent == []
+
+
 _NAMED_TRIGGERS = [
     {"id": tid, "when": {"type": "action", "tool": "slack_send_message"},
      "actions": [{"type": "nl", "instruction": f"reply as {tid}", **named}]}