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 56d6d192..17761124 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
@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
+import time
 from typing import ClassVar, Optional
 
 import httpx
@@ -23,12 +25,25 @@
 
 logger = logging.getLogger(__name__)
 
+# The await reads the gateway's trigger state once per poll interval until the step's deadline; a read gets at most the
+# read timeout, never starts without the minimum window left, and when it fails is simply retried on the next poll. A
+# trigger that reads one of the terminal statuses can no longer fire, so the wait stops there. A failure message quotes
+# at most this many gateway events.
+_AWAIT_TARGET_STATUS = "fired"
+_AWAIT_TERMINAL_STATUSES = ("failed", "cancelled")
+_AWAIT_POLL_INTERVAL_S = 1.0
+_STATE_READ_TIMEOUT_S = 15
+_STATE_READ_MIN_TIMEOUT_S = 1
+_AWAIT_FAILURE_EVENTS_SHOWN = 6
+
 
 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. An
-    action's `delay` and `executor_hours` pace replies in virtual time, and a time trigger's `unless` makes it a deadline."""
+    action's `delay` and `executor_hours` pace replies in virtual time, and a time trigger's `unless` makes it a deadline.
+    `await_fired` holds the step until the named one-shot triggers have fired (a recurrence re-arms after each fire), so
+    a dependent step starts only once start-of-run time triggers have landed in the world."""
 
     type: ClassVar[str] = "register_env_triggers"
     entity_refs = (EntityRef.env("env_id"),)
@@ -43,18 +58,28 @@ def __init__(
         executor_agent_name: Optional[str] = None,
         executor_timeout_seconds: float = 120,
         executor_hours: Optional[dict[str, dict]] = None,
+        await_fired: Optional[list[str]] = None,
+        await_timeout_seconds: float = 120,
         depends_on: Optional[list[TaskStepDependency]] = None,
         fail_task_on_error: bool = True,
     ):
         super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
         if not isinstance(triggers, list) or not all(isinstance(t, dict) for t in triggers):
             raise ValueError("triggers must be a list of objects")
+        if await_fired is not None and (not isinstance(await_fired, list)
+                                        or not all(isinstance(i, str) and i for i in await_fired)):
+            raise ValueError("await_fired must be a list of trigger ids")
+        if (isinstance(await_timeout_seconds, bool) or not isinstance(await_timeout_seconds, (int, float))
+                or await_timeout_seconds <= 0):
+            raise ValueError("await_timeout_seconds must be a positive number")
         self.env_id = env_id
         self.triggers = triggers
         self.watch_roles = watch_roles
         self.executor_agent_name = executor_agent_name
         self.executor_timeout_seconds = executor_timeout_seconds
         self.executor_hours = executor_hours
+        self.await_fired = list(dict.fromkeys(await_fired)) if await_fired else None
+        self.await_timeout_seconds = float(await_timeout_seconds)
 
     def to_dict(self) -> dict:
         base = super().to_dict()
@@ -64,6 +89,8 @@ def to_dict(self) -> dict:
         base["executor_agent_name"] = self.executor_agent_name
         base["executor_timeout_seconds"] = self.executor_timeout_seconds
         base["executor_hours"] = self.executor_hours
+        base["await_fired"] = self.await_fired
+        base["await_timeout_seconds"] = self.await_timeout_seconds
         return base
 
     @classmethod
@@ -76,9 +103,24 @@ def from_dict(cls, data: dict) -> RegisterEnvTriggersStep:
             executor_agent_name=data.get("executor_agent_name"),
             executor_timeout_seconds=data.get("executor_timeout_seconds", 120),
             executor_hours=data.get("executor_hours"),
+            await_fired=data.get("await_fired"),
+            await_timeout_seconds=data.get("await_timeout_seconds", 120),
         )
 
+    def preflight(self) -> list[str]:
+        unknown = self._unknown_awaited()
+        if unknown:
+            return [f"register_env_triggers '{self.id}': await_fired names {unknown}, which this step does not register"]
+        return []
+
+    def _unknown_awaited(self) -> list[str]:
+        registered = {t.get("id") for t in self.triggers}
+        return [tid for tid in (self.await_fired or []) if tid not in registered]
+
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
+        problems = self.preflight()
+        if problems:
+            raise RuntimeError("; ".join(problems))
         deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None)
         if deployed is None:
             raise RuntimeError(f"Env '{self.env_id}' not found in context.deployed_envs")
@@ -91,6 +133,10 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         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")
+        if self.await_fired and not deployed.supports(EXT_TRIGGERS_URI, "state"):
+            raise RuntimeError(
+                f"Env '{self.env_id}' runs a gateway whose triggers/v1 does not offer 'state', which await_fired needs; "
+                f"redeploy it on a gateway image that does.")
         # 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.
@@ -109,15 +155,59 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         except httpx.HTTPStatusError as e:
             raise RuntimeError(f"trigger registration failed (HTTP {e.response.status_code}): {e.response.text}") from e
         logger.info(f"registered triggers on env={self.env_id}: {result}")
-        context.metadata.setdefault("env_trigger_registrations", []).append({
+        entry = {
             "step_id": self.id,
             "env_id": self.env_id,
             "added": result.get("added", []),
             "executor_agent_name": self.executor_agent_name,
             "executor_agent_names": named,
-        })
+        }
+        # Recorded before the wait: the triggers are on the gateway whether or not the wait succeeds, and later steps
+        # look them up here. `awaited` then carries the last statuses seen, complete on success.
+        context.metadata.setdefault("env_trigger_registrations", []).append(entry)
+        if self.await_fired:
+            entry["awaited"] = {tid: None for tid in self.await_fired}
+            await self._await_statuses(deployed, entry["awaited"])
         return context
 
+    async def _await_statuses(self, deployed: DeployedEnv, seen: dict[str, Optional[str]]) -> None:
+        """Poll the gateway, updating `seen` in place, until every awaited trigger has fired; raise when one reaches a
+        terminal status first or when the deadline passes. A read starts only with a full minimum read window left
+        before the deadline and is cut to that remainder, so the step never waits past `await_timeout_seconds`; a read
+        that fails is retried on the next poll, and the last failure is reported if the deadline passes."""
+        deadline = time.monotonic() + self.await_timeout_seconds
+        state: dict = {}
+        read_error: Optional[Exception] = None
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining < _STATE_READ_MIN_TIMEOUT_S:
+                raise RuntimeError(self._await_failure(seen, state, stuck=[], read_error=read_error))
+            try:
+                state = await deployed.invoke(EXT_TRIGGERS_URI, "state", timeout=min(_STATE_READ_TIMEOUT_S, remaining))
+                read_error = None
+            except Exception as e:  # noqa: BLE001 - any read failure is retried on the next poll
+                read_error = e
+            else:
+                statuses = {t.get("id"): t.get("status") for t in state.get("triggers", []) if isinstance(t, dict)}
+                seen.update({tid: statuses.get(tid) for tid in self.await_fired})
+                if all(status == _AWAIT_TARGET_STATUS for status in seen.values()):
+                    logger.info(f"awaited triggers on env={self.env_id} fired: {sorted(seen)}")
+                    return
+                stuck = sorted(tid for tid, status in seen.items() if status in _AWAIT_TERMINAL_STATUSES)
+                if stuck:
+                    raise RuntimeError(self._await_failure(seen, state, stuck))
+            await asyncio.sleep(max(0.0, min(_AWAIT_POLL_INTERVAL_S, deadline - time.monotonic())))
+
+    def _await_failure(self, seen: dict[str, Optional[str]], state: dict, stuck: list[str],
+                       read_error: Optional[Exception] = None) -> str:
+        if stuck:
+            why = f"awaited triggers reached a terminal status before firing: {stuck}"
+        elif read_error is not None:
+            why = f"could not read /triggers/state within the {self.await_timeout_seconds:g}s await budget: {read_error}"
+        else:
+            why = f"awaited triggers did not all fire within {self.await_timeout_seconds:g}s"
+        return f"Env '{self.env_id}': {why}; statuses {dict(seen)}" + _failure_detail(state, set(seen))
+
     def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agent_name: str) -> dict:
         agent = next((a for a in context.deployed_agents if a.agent_name == agent_name), None)
         if agent is None:
@@ -140,6 +230,17 @@ def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agen
         return config
 
 
+def _failure_detail(state: dict, trigger_ids: set) -> str:
+    """The gateway's failure events for `trigger_ids`, newest last, without their sequence and clock stamps."""
+    events = [e for e in state.get("events", []) if isinstance(e, dict) and e.get("trigger_id") in trigger_ids
+              and e.get("kind") in ("failed", "eval_error", "action_failed")]
+    if not events:
+        return ""
+    shown = [{k: v for k, v in e.items() if k not in ("seq", "ts", "virtual_time")}
+             for e in events[-_AWAIT_FAILURE_EVENTS_SHOWN:]]
+    return f"; failures: {shown}"
+
+
 def _named_executors(triggers: list[dict]) -> list[str]:
     """The executor agent names the triggers' nl actions name, in first-use order."""
     names = (action.get("executor") for trigger in triggers for action in trigger.get("actions") or []
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 f62c70da..dfd738f7 100644
--- a/tst/integration/task_step/test_register_env_triggers_local.py
+++ b/tst/integration/task_step/test_register_env_triggers_local.py
@@ -195,6 +195,39 @@ async def test_a_deadline_fires_on_silence_and_a_reply_cancels_it(local_stack):
     assert (cancelled["trigger_id"], cancelled["provoking"]["role"]) == ("deadline-2", "default")
 
 
+@pytest.mark.asyncio
+async def test_the_step_can_wait_for_start_of_run_time_triggers(local_stack):
+    _seed_bootstrap_envs()
+    env, deployed, host = await _deploy_items("await")
+    triggers = [{"id": "wf-1", "when": {"type": "time", "at": "2026-10-14T10:02:00Z"},
+                 "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "WORLD-1"}}]},
+                {"id": "wf-2", "when": {"type": "time", "at": "2026-10-14T10:03:00Z"},
+                 "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "WORLD-2"}}]},
+                {"id": "later", "when": {"type": "time", "at": "2026-10-14T18:00:00Z"},
+                 "actions": [{"type": "tool", "tool": "items_add_item", "args": {"item": "LATER"}}]}]
+    context = TaskStepContext(deployed_envs=[deployed], deployed_agents=[], metadata={})
+    try:
+        async with httpx.AsyncClient(base_url=deployed.gateway_url, timeout=60) as client:
+            await _arm_clock(client, "2026-10-14T10:00:00Z", 60)
+            started = time.monotonic()
+            await RegisterEnvTriggersStep(id="reg", version=None, env_id=env.id, triggers=triggers, watch_roles=["default"],
+                                          await_fired=["wf-1", "wf-2"], await_timeout_seconds=60).execute(context)
+            waited = time.monotonic() - started
+            state = (await client.get("/triggers/state")).raise_for_status().json()
+            with pytest.raises(RuntimeError, match=r"did not all fire within 3s; statuses \{'later': 'armed'\}"):
+                await RegisterEnvTriggersStep(id="reg-later", version=None, env_id=env.id, triggers=triggers[2:],
+                                              watch_roles=["default"], await_fired=["later"],
+                                              await_timeout_seconds=3).execute(context)
+    finally:
+        await env.close()
+
+    awaited_first, awaited_later = (e["awaited"] for e in context.metadata["env_trigger_registrations"])
+    assert awaited_first == {"wf-1": "fired", "wf-2": "fired"}
+    assert awaited_later == {"later": "armed"}  # the failed wait still records its registration and last statuses
+    assert waited >= 2.0, waited  # three virtual minutes at sixty-to-one
+    assert {t["id"]: t["fire_count"] for t in state["triggers"]} == {"wf-1": 1, "wf-2": 1, "later": 0}, state["events"][-6:]
+
+
 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."""
diff --git a/tst/unit/task_step/test_register_env_triggers.py b/tst/unit/task_step/test_register_env_triggers.py
index d6a41323..9e11b4bf 100644
--- a/tst/unit/task_step/test_register_env_triggers.py
+++ b/tst/unit/task_step/test_register_env_triggers.py
@@ -1,10 +1,11 @@
-"""Behavior tests for RegisterEnvTriggersStep: executor-name resolution, the POST body/audit, and 4xx
-propagation. Only the gateway HTTP boundary is mocked; the step logic runs real."""
+"""Behavior tests for RegisterEnvTriggersStep: executor-name resolution, the POST body/audit, 4xx propagation, and
+the optional await of trigger statuses. Only the gateway HTTP boundary is mocked; the step logic runs real."""
 
 from __future__ import annotations
 
 import copy
 import json
+import time
 
 import httpx
 import pytest
@@ -48,6 +49,12 @@ def test_round_trips_through_to_dict_from_dict():
     assert rebuilt.executor_agent_name == "exec-1"
     assert rebuilt.executor_timeout_seconds == 90
     assert rebuilt.executor_hours == {"exec-1": _HOURS}
+    assert (rebuilt.await_fired, rebuilt.await_timeout_seconds) == (None, 120)
+
+    awaiting = RegisterEnvTriggersStep.from_dict(_step(await_fired=["t1", "t1"], await_timeout_seconds=7).to_dict())
+    assert (awaiting.await_fired, awaiting.await_timeout_seconds) == (["t1"], 7.0)
+    stored_before_await = {k: v for k, v in _step().to_dict().items() if k not in ("await_fired", "await_timeout_seconds")}
+    assert RegisterEnvTriggersStep.from_dict(stored_before_await).await_fired is None
 
 
 def test_bad_triggers_rejected_at_construction():
@@ -304,3 +311,215 @@ def handle(request: httpx.Request) -> httpx.Response:
 
     monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle)))
     return sent
+
+
+def test_bad_await_config_rejected_at_construction():
+    with pytest.raises(ValueError, match="await_fired must be a list of trigger ids"):
+        _step(await_fired="t1")
+    with pytest.raises(ValueError, match="await_timeout_seconds must be a positive number"):
+        _step(await_fired=["t1"], await_timeout_seconds=0)
+    with pytest.raises(ValueError, match="await_timeout_seconds must be a positive number"):
+        _step(await_fired=["t1"], await_timeout_seconds=True)
+
+
+def test_preflight_names_an_awaited_trigger_this_step_does_not_register():
+    assert _step(await_fired=["t1"]).preflight() == []
+    [problem] = _step(await_fired=["t1", "nope"]).preflight()
+    assert "await_fired names ['nope']" in problem and "reg-1" in problem
+
+
+@pytest.mark.asyncio
+async def test_an_unregistered_awaited_trigger_raises_before_any_request(monkeypatch):
+    sent = _mock_gateway(monkeypatch, {})
+    with pytest.raises(RuntimeError, match=r"register_env_triggers 'reg-1': await_fired names \['nope'\]"):
+        await _step(await_fired=["nope"]).execute(_context())
+    assert sent == []
+
+
+@pytest.mark.asyncio
+async def test_a_card_without_triggers_state_refuses_the_await_before_any_request(monkeypatch):
+    sent = _mock_gateway(monkeypatch, {})
+    ctx = TaskStepContext(deployed_envs=[_record(extensions=_extensions_without_state())], deployed_agents=[], metadata={})
+    with pytest.raises(RuntimeError, match="does not offer 'state', which await_fired needs"):
+        await _step(await_fired=["t1"]).execute(ctx)
+    assert sent == []
+
+
+@pytest.mark.asyncio
+async def test_await_returns_once_each_named_trigger_reads_fired(monkeypatch):
+    _no_sleep(monkeypatch)
+    sent = _mock_gateway_by_path(monkeypatch, {
+        "/triggers/register": [{"ok": True, "added": ["t1", "t2"], "all": ["t1", "t2"]}],
+        "/triggers/state": [_state({"t1": "fired", "t2": "armed"}), _state({"t1": "fired", "t2": "firing"}),
+                            _state({"t1": "fired", "t2": "fired"})],
+    })
+    out = await _step(triggers=_TWO, await_fired=["t1", "t2"]).execute(_context())
+
+    assert [str(r.url.path) for r in sent] == ["/triggers/register"] + ["/triggers/state"] * 3
+    [audit] = out.metadata["env_trigger_registrations"]
+    assert audit["added"] == ["t1", "t2"]
+    assert audit["awaited"] == {"t1": "fired", "t2": "fired"}
+
+
+@pytest.mark.asyncio
+async def test_no_await_means_no_state_read_and_no_awaited_key(monkeypatch):
+    sent = _mock_gateway_by_path(monkeypatch, {"/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}]})
+    out = await _step().execute(_context())
+    assert [str(r.url.path) for r in sent] == ["/triggers/register"]
+    assert "awaited" not in out.metadata["env_trigger_registrations"][0]
+
+
+@pytest.mark.asyncio
+async def test_await_times_out_naming_statuses_and_failure_events(monkeypatch):
+    _mock_gateway_by_path(monkeypatch, {
+        "/triggers/register": [{"ok": True, "added": ["t1", "t2"], "all": ["t1", "t2"]}],
+        "/triggers/state": [_state({"t1": "fired", "t2": "armed"},
+                                   events=[{"seq": 3, "ts": "x", "kind": "action_failed", "trigger_id": "t2",
+                                            "error": "tool 'slack_send_message' refused"},
+                                           {"seq": 4, "ts": "y", "kind": "fired", "trigger_id": "t1"}])],
+    })
+    started = time.monotonic()
+    with pytest.raises(RuntimeError) as err:
+        await _step(triggers=_TWO, await_fired=["t1", "t2"], await_timeout_seconds=1.2).execute(_context())
+    assert time.monotonic() - started < 1.2
+    message = str(err.value)
+    assert "did not all fire within 1.2s" in message
+    assert "{'t1': 'fired', 't2': 'armed'}" in message
+    assert "action_failed" in message and "slack_send_message" in message
+    assert "'seq'" not in message and "'kind': 'fired'" not in message
+
+
+@pytest.mark.asyncio
+async def test_a_failed_wait_still_records_the_registration_with_the_last_statuses(monkeypatch):
+    _mock_gateway_by_path(monkeypatch, {
+        "/triggers/register": [{"ok": True, "added": ["t1", "t2"], "all": ["t1", "t2"]}],
+        "/triggers/state": [_state({"t1": "fired", "t2": "armed"})],
+    })
+    ctx = _context()
+    with pytest.raises(RuntimeError):
+        await _step(triggers=_TWO, await_fired=["t1", "t2"], await_timeout_seconds=1.2).execute(ctx)
+    [audit] = ctx.metadata["env_trigger_registrations"]
+    assert audit["added"] == ["t1", "t2"]
+    assert audit["awaited"] == {"t1": "fired", "t2": "armed"}
+
+
+@pytest.mark.asyncio
+async def test_slow_state_reads_cannot_outlive_the_await_budget(monkeypatch):
+    """Each read's timeout is cut to the time left and no read starts inside the last minimum window, so a stalled
+    gateway cannot hold the step past the deadline."""
+    import asyncio as real_asyncio
+
+    _mock_gateway_by_path(monkeypatch, {"/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}]})
+    timeouts: list[float] = []
+
+    async def stalled(self, uri, method, params=None, *, environment_name=None, timeout=30):
+        if method == "register":
+            return {"ok": True, "added": ["t1"], "all": ["t1"]}
+        timeouts.append(timeout)
+        await real_asyncio.sleep(min(timeout, 0.05))
+        raise httpx.ReadTimeout("gateway stalled")
+
+    monkeypatch.setattr(DeployedGatewayEnv, "invoke", stalled)
+    started = time.monotonic()
+    with pytest.raises(RuntimeError, match=r"could not read /triggers/state within the 2.5s await budget: gateway stalled; statuses \{'t1': None\}"):
+        await _step(await_fired=["t1"], await_timeout_seconds=2.5).execute(_context())
+    assert time.monotonic() - started < 2.5
+    assert timeouts and all(t <= 2.5 for t in timeouts) and timeouts == sorted(timeouts, reverse=True), timeouts
+
+
+@pytest.mark.asyncio
+async def test_await_stops_early_on_a_failed_trigger(monkeypatch):
+    _no_sleep(monkeypatch)
+    sent = _mock_gateway_by_path(monkeypatch, {
+        "/triggers/register": [{"ok": True, "added": ["t1", "t2"], "all": ["t1", "t2"]}],
+        "/triggers/state": [_state({"t1": "armed", "t2": "failed"},
+                                   events=[{"seq": 1, "ts": "x", "kind": "failed", "trigger_id": "t2", "error": "boom"}])],
+    })
+    with pytest.raises(RuntimeError, match=r"reached a terminal status before firing: \['t2'\].*boom"):
+        await _step(triggers=_TWO, await_fired=["t1", "t2"], await_timeout_seconds=60).execute(_context())
+    assert [str(r.url.path) for r in sent] == ["/triggers/register", "/triggers/state"]
+
+
+@pytest.mark.asyncio
+async def test_a_trigger_missing_from_state_keeps_waiting_until_the_deadline(monkeypatch):
+    sent = _mock_gateway_by_path(monkeypatch, {
+        "/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}],
+        "/triggers/state": [_state({})],
+    })
+    with pytest.raises(RuntimeError, match=r"within 1.2s; statuses \{'t1': None\}"):
+        await _step(await_fired=["t1"], await_timeout_seconds=1.2).execute(_context())
+    assert [str(r.url.path) for r in sent].count("/triggers/state") == 1  # one read, one poll interval, the deadline
+
+
+@pytest.mark.asyncio
+async def test_a_budget_shorter_than_one_read_fails_without_reading(monkeypatch):
+    sent = _mock_gateway_by_path(monkeypatch, {"/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}]})
+    ctx = _context()
+    with pytest.raises(RuntimeError, match=r"within 0.5s; statuses \{'t1': None\}"):
+        await _step(await_fired=["t1"], await_timeout_seconds=0.5).execute(ctx)
+    assert [str(r.url.path) for r in sent] == ["/triggers/register"]
+    assert ctx.metadata["env_trigger_registrations"][0]["awaited"] == {"t1": None}
+
+
+@pytest.mark.asyncio
+async def test_a_failed_state_read_is_retried_on_the_next_poll(monkeypatch):
+    _no_sleep(monkeypatch)
+    sent = _mock_gateway_by_path(monkeypatch, {
+        "/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}],
+        "/triggers/state": [(503, {"error": "busy"}), _state({"t1": "fired"})],
+    })
+    out = await _step(await_fired=["t1"]).execute(_context())
+    assert out.metadata["env_trigger_registrations"][0]["awaited"] == {"t1": "fired"}
+    assert [str(r.url.path) for r in sent].count("/triggers/state") == 2
+
+
+@pytest.mark.asyncio
+async def test_a_state_read_that_keeps_failing_is_reported_with_its_cause_at_the_deadline(monkeypatch):
+    sent = _mock_gateway_by_path(monkeypatch, {
+        "/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}],
+        "/triggers/state": [(503, {"error": "busy"})],
+    })
+    with pytest.raises(RuntimeError, match=r"(?s)could not read /triggers/state within the 2.5s await budget: .*503.*; statuses \{'t1': None\}"):
+        await _step(await_fired=["t1"], await_timeout_seconds=2.5).execute(_context())
+    assert [str(r.url.path) for r in sent].count("/triggers/state") == 2  # reads at 0 s and 1 s; 2 s leaves no full window
+
+
+_TWO = [{"id": "t1", "when": {"type": "time", "at": "2026-07-01T17:00:02-05:00"},
+         "actions": [{"type": "tool", "tool": "slack_send_message", "args": {"channel_id": "C1", "message": "hi"}}]},
+        {"id": "t2", "when": {"type": "time", "at": "2026-07-01T17:00:04-05:00"},
+         "actions": [{"type": "tool", "tool": "slack_send_message", "args": {"channel_id": "C1", "message": "hi"}}]}]
+
+
+def _state(statuses: dict[str, str], events: list | None = None) -> dict:
+    return {"config": {}, "triggers": [{"id": tid, "type": "time", "status": status, "fire_count": int(status == "fired")}
+                                       for tid, status in statuses.items()],
+            "events": events or [], "events_dropped": 0}
+
+
+def _extensions_without_state() -> list:
+    extensions = copy.deepcopy(GATEWAY_EXTENSIONS)
+    del next(e for e in extensions if e["uri"] == EXT_TRIGGERS_URI)["params"]["methods"]["state"]
+    return extensions
+
+
+def _no_sleep(monkeypatch) -> None:
+    """Polls and retries run back to back; the deadline still comes from the monotonic clock."""
+    async def instant(_seconds):
+        return None
+    monkeypatch.setattr("agent_env.task_step.task_steps.register_env_triggers.asyncio.sleep", instant)
+
+
+def _mock_gateway_by_path(monkeypatch, scripted: dict[str, list]) -> list[httpx.Request]:
+    """Answer each path from its scripted list in order, repeating the last entry; an entry may be `(status, body)`."""
+    sent, real, served = [], httpx.AsyncClient, {path: 0 for path in scripted}
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        sent.append(request)
+        answers = scripted[request.url.path]
+        entry = answers[min(served[request.url.path], len(answers) - 1)]
+        served[request.url.path] += 1
+        status, body = entry if isinstance(entry, tuple) else (200, entry)
+        return httpx.Response(status, json=body)
+
+    monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle)))
+    return sent