From 96033aac481fc15e06afdc7278e205c882796491 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 15 Oct 2026 14:56:12 -0700 Subject: [PATCH 1/6] register_env_triggers: optional await of start-of-run triggers (FD-3998) The step returns when its triggers are armed, so a task whose world is seeded by absolute-time triggers had to carry a run_code warm-up script that polled the gateway until they fired. `await_fired` moves that guard into the step: after registering, it polls the triggers extension's `state` method until every named trigger reads `await_status` (default `fired`), stops early on `failed` or `cancelled`, and raises with the per-trigger statuses and the gateway's failure events for them when the deadline passes. The audit entry gains `awaited`. `preflight` and `execute` refuse an awaited id the step does not register, and a gateway whose card lacks `state` is refused before any request. Co-Authored-By: Claude Fable 5.1 --- .../task_steps/register_env_triggers.py | 91 ++++++++- .../test_register_env_triggers_local.py | 32 ++++ .../task_step/test_register_env_triggers.py | 176 +++++++++++++++++- 3 files changed, 294 insertions(+), 5 deletions(-) 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..b45a4eba 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 @@ -15,6 +17,7 @@ TRIGGER_EXECUTOR_HOURS, TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES, + TRIGGER_STATUSES, TRIGGER_UNLESS, ) from agent_env.task_step.context import TaskStepContext @@ -28,7 +31,9 @@ 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 reach `await_status` (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 +48,31 @@ 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_status: str = "fired", + 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 await_status not in TRIGGER_STATUSES: + raise ValueError(f"await_status must be one of {list(TRIGGER_STATUSES)}, not {await_status!r}") + if 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_status = await_status + self.await_timeout_seconds = float(await_timeout_seconds) def to_dict(self) -> dict: base = super().to_dict() @@ -64,6 +82,9 @@ 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_status"] = self.await_status + base["await_timeout_seconds"] = self.await_timeout_seconds return base @classmethod @@ -76,8 +97,21 @@ 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_status=data.get("await_status", "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: deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None) if deployed is None: @@ -91,6 +125,13 @@ 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") + unknown_awaited = self._unknown_awaited() + if unknown_awaited: + raise RuntimeError(f"await_fired names {unknown_awaited}, which this step does not register") + 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 +150,38 @@ 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, - }) + } + if self.await_fired: + entry["awaited"] = await self._await_statuses(deployed) + context.metadata.setdefault("env_trigger_registrations", []).append(entry) return context + async def _await_statuses(self, deployed: DeployedEnv) -> dict[str, Optional[str]]: + """Poll the gateway until every awaited trigger reads `await_status`; raise on a terminal miss or the deadline.""" + terminal = {"failed", "cancelled"} - {self.await_status} + deadline = time.monotonic() + self.await_timeout_seconds + while True: + state = await _trigger_state(deployed) + statuses = {t.get("id"): t.get("status") for t in state.get("triggers", []) if isinstance(t, dict)} + seen = {tid: statuses.get(tid) for tid in self.await_fired} + if all(status == self.await_status for status in seen.values()): + logger.info(f"awaited triggers on env={self.env_id} reached {self.await_status!r}: {sorted(seen)}") + return seen + stuck = sorted(tid for tid, status in seen.items() if status in terminal) + timed_out = time.monotonic() >= deadline + if stuck or timed_out: + why = (f"did not all reach {self.await_status!r} within {self.await_timeout_seconds:g}s" if timed_out + else f"reached a terminal status before {self.await_status!r}: {stuck}") + raise RuntimeError(f"Env '{self.env_id}': awaited triggers {why}; statuses {seen}" + + _failure_detail(state, set(seen))) + await asyncio.sleep(min(1.0, max(0.0, deadline - time.monotonic()))) + 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 +204,27 @@ def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agen return config +async def _trigger_state(deployed: DeployedEnv) -> dict: + last_exc: Optional[Exception] = None + for attempt in range(3): + try: + return await deployed.invoke(EXT_TRIGGERS_URI, "state", timeout=15) + except Exception as e: # noqa: BLE001 - any read failure is retried, then reported with its cause + last_exc = e + await asyncio.sleep(0.5 * (attempt + 1)) + raise RuntimeError(f"could not read /triggers/state for env '{deployed.env_id}': {last_exc}") + + +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[-6:]] + 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..c9620303 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,38 @@ 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 reach 'fired' 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() + + [audit] = context.metadata["env_trigger_registrations"] + assert audit["awaited"] == {"wf-1": "fired", "wf-2": "fired"} + 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..531d10a4 100644 --- a/tst/unit/task_step/test_register_env_triggers.py +++ b/tst/unit/task_step/test_register_env_triggers.py @@ -1,5 +1,5 @@ -"""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 @@ -48,6 +48,14 @@ 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_status, rebuilt.await_timeout_seconds) == (None, "fired", 120) + + awaiting = RegisterEnvTriggersStep.from_dict(_step(await_fired=["t1", "t1"], await_status="armed", + await_timeout_seconds=7).to_dict()) + assert (awaiting.await_fired, awaiting.await_status, awaiting.await_timeout_seconds) == (["t1"], "armed", 7.0) + stored_before_await = {k: v for k, v in _step().to_dict().items() + if k not in ("await_fired", "await_status", "await_timeout_seconds")} + assert RegisterEnvTriggersStep.from_dict(stored_before_await).await_fired is None def test_bad_triggers_rejected_at_construction(): @@ -304,3 +312,167 @@ 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_status must be one of"): + _step(await_fired=["t1"], await_status="done") + with pytest.raises(ValueError, match="await_timeout_seconds must be a positive number"): + _step(await_fired=["t1"], await_timeout_seconds=0) + + +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"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): + _no_sleep(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"}])], + }) + with pytest.raises(RuntimeError) as err: + await _step(triggers=_TWO, await_fired=["t1", "t2"], await_timeout_seconds=0.01).execute(_context()) + message = str(err.value) + assert "did not all reach 'fired' within 0.01s" 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_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 'fired': \['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): + _no_sleep(monkeypatch) + _mock_gateway_by_path(monkeypatch, { + "/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}], + "/triggers/state": [_state({})], + }) + with pytest.raises(RuntimeError, match=r"statuses \{'t1': None\}"): + await _step(await_fired=["t1"], await_timeout_seconds=0.01).execute(_context()) + + +@pytest.mark.asyncio +async def test_a_flaky_state_read_is_retried_then_reported(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(monkeypatch): + _no_sleep(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="could not read /triggers/state for env 'env-x'.*503"): + await _step(await_fired=["t1"]).execute(_context()) + assert [str(r.url.path) for r in sent].count("/triggers/state") == 3 + + +_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 From 449a57507c8a4e641255d52ab0fc93b4b5c6e543 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 15 Oct 2026 15:05:34 -0700 Subject: [PATCH 2/6] register_env_triggers: bound the await by its budget, record the registration before waiting, name the wait settings Each trigger-state read and each retry backoff is now cut to what is left of `await_timeout_seconds`, so a stalled gateway cannot hold the step past its deadline. The registration audit entry is appended before the wait, with `awaited` carrying the last statuses seen, so a failed wait no longer hides a registration that later steps look up. The poll interval, read timeout, attempt count, backoff and the number of failure events quoted are named module constants. Co-Authored-By: Claude Fable 5.1 --- .../task_steps/register_env_triggers.py | 57 +++++++++++++------ .../task_step/test_register_env_triggers.py | 40 +++++++++++++ 2 files changed, 79 insertions(+), 18 deletions(-) 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 b45a4eba..dc1fa62a 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 @@ -26,6 +26,15 @@ logger = logging.getLogger(__name__) +# The await polls the gateway's trigger state once per interval; a read gets this long, retried this often with this +# backoff, and never past the step's own deadline. A failure message quotes at most this many gateway events. +_AWAIT_POLL_INTERVAL_S = 1.0 +_STATE_READ_TIMEOUT_S = 15 +_STATE_READ_ATTEMPTS = 3 +_STATE_READ_BACKOFF_S = 0.5 +_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 @@ -157,30 +166,35 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: "executor_agent_name": self.executor_agent_name, "executor_agent_names": named, } - if self.await_fired: - entry["awaited"] = await self._await_statuses(deployed) + # 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) -> dict[str, Optional[str]]: - """Poll the gateway until every awaited trigger reads `await_status`; raise on a terminal miss or the deadline.""" + async def _await_statuses(self, deployed: DeployedEnv, seen: dict[str, Optional[str]]) -> None: + """Poll the gateway, updating `seen` in place, until every awaited trigger reads `await_status`; raise on a + terminal miss or when the deadline passes. Every read and retry is bounded by what is left of the deadline.""" terminal = {"failed", "cancelled"} - {self.await_status} deadline = time.monotonic() + self.await_timeout_seconds while True: - state = await _trigger_state(deployed) + remaining = deadline - time.monotonic() + state = await _trigger_state(deployed, remaining) statuses = {t.get("id"): t.get("status") for t in state.get("triggers", []) if isinstance(t, dict)} - seen = {tid: statuses.get(tid) for tid in self.await_fired} + seen.update({tid: statuses.get(tid) for tid in self.await_fired}) if all(status == self.await_status for status in seen.values()): logger.info(f"awaited triggers on env={self.env_id} reached {self.await_status!r}: {sorted(seen)}") - return seen + return stuck = sorted(tid for tid, status in seen.items() if status in terminal) - timed_out = time.monotonic() >= deadline - if stuck or timed_out: - why = (f"did not all reach {self.await_status!r} within {self.await_timeout_seconds:g}s" if timed_out - else f"reached a terminal status before {self.await_status!r}: {stuck}") - raise RuntimeError(f"Env '{self.env_id}': awaited triggers {why}; statuses {seen}" + remaining = deadline - time.monotonic() + if stuck or remaining <= 0: + why = (f"reached a terminal status before {self.await_status!r}: {stuck}" if stuck + else f"did not all reach {self.await_status!r} within {self.await_timeout_seconds:g}s") + raise RuntimeError(f"Env '{self.env_id}': awaited triggers {why}; statuses {dict(seen)}" + _failure_detail(state, set(seen))) - await asyncio.sleep(min(1.0, max(0.0, deadline - time.monotonic()))) + await asyncio.sleep(min(_AWAIT_POLL_INTERVAL_S, remaining)) 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) @@ -204,14 +218,20 @@ def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agen return config -async def _trigger_state(deployed: DeployedEnv) -> dict: +async def _trigger_state(deployed: DeployedEnv, budget_s: float) -> dict: + """One trigger-state read, retried on failure, with each read and backoff cut to the remaining `budget_s`.""" + deadline = time.monotonic() + budget_s last_exc: Optional[Exception] = None - for attempt in range(3): + for attempt in range(_STATE_READ_ATTEMPTS): + remaining = deadline - time.monotonic() + if attempt and remaining <= 0: + break + timeout = max(_STATE_READ_MIN_TIMEOUT_S, min(_STATE_READ_TIMEOUT_S, remaining)) try: - return await deployed.invoke(EXT_TRIGGERS_URI, "state", timeout=15) + return await deployed.invoke(EXT_TRIGGERS_URI, "state", timeout=timeout) except Exception as e: # noqa: BLE001 - any read failure is retried, then reported with its cause last_exc = e - await asyncio.sleep(0.5 * (attempt + 1)) + await asyncio.sleep(max(0.0, min(_STATE_READ_BACKOFF_S * (attempt + 1), deadline - time.monotonic()))) raise RuntimeError(f"could not read /triggers/state for env '{deployed.env_id}': {last_exc}") @@ -221,7 +241,8 @@ def _failure_detail(state: dict, trigger_ids: set) -> str: 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[-6:]] + 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}" diff --git a/tst/unit/task_step/test_register_env_triggers.py b/tst/unit/task_step/test_register_env_triggers.py index 531d10a4..7a9d5bd7 100644 --- a/tst/unit/task_step/test_register_env_triggers.py +++ b/tst/unit/task_step/test_register_env_triggers.py @@ -5,6 +5,7 @@ import copy import json +import time import httpx import pytest @@ -389,6 +390,45 @@ async def test_await_times_out_naming_statuses_and_failure_events(monkeypatch): 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): + _no_sleep(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=0.01).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 and each retry's backoff are cut to the time left, so a stalled gateway cannot hold the step + for three full reads 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="could not read /triggers/state for env 'env-x'.*gateway stalled"): + await _step(await_fired=["t1"], await_timeout_seconds=0.3).execute(_context()) + assert time.monotonic() - started < 2.0 + assert timeouts and all(t <= 1 for t in timeouts), timeouts + + @pytest.mark.asyncio async def test_await_stops_early_on_a_failed_trigger(monkeypatch): _no_sleep(monkeypatch) From 292bfadbda2d92f5f2a78bb17aadbed2d446ef50 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 15 Oct 2026 15:06:57 -0700 Subject: [PATCH 3/6] test: the failed wait in the local await test now leaves its own registration entry Co-Authored-By: Claude Fable 5.1 --- .../task_step/test_register_env_triggers_local.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 c9620303..771d6586 100644 --- a/tst/integration/task_step/test_register_env_triggers_local.py +++ b/tst/integration/task_step/test_register_env_triggers_local.py @@ -221,8 +221,9 @@ async def test_the_step_can_wait_for_start_of_run_time_triggers(local_stack): finally: await env.close() - [audit] = context.metadata["env_trigger_registrations"] - assert audit["awaited"] == {"wf-1": "fired", "wf-2": "fired"} + 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:] From a2b94202c9f13887535a892f2c9a246021178cbc Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 15 Oct 2026 15:43:10 -0700 Subject: [PATCH 4/6] register_env_triggers: the await never outlives its deadline A state read now starts only with a full minimum read window left before the deadline and is cut to that remainder, so neither a slow read nor a late reply can carry the step past `await_timeout_seconds`; a budget shorter than one read fails at once with the statuses last seen. The failure message is built in one place. Co-Authored-By: Claude Fable 5.1 --- .../task_steps/register_env_triggers.py | 31 +++++++++------- .../task_step/test_register_env_triggers.py | 36 ++++++++++++------- 2 files changed, 41 insertions(+), 26 deletions(-) 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 dc1fa62a..2abb57a7 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 @@ -176,11 +176,15 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: async def _await_statuses(self, deployed: DeployedEnv, seen: dict[str, Optional[str]]) -> None: """Poll the gateway, updating `seen` in place, until every awaited trigger reads `await_status`; raise on a - terminal miss or when the deadline passes. Every read and retry is bounded by what is left of the deadline.""" + terminal miss 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`.""" terminal = {"failed", "cancelled"} - {self.await_status} deadline = time.monotonic() + self.await_timeout_seconds + state: dict = {} while True: remaining = deadline - time.monotonic() + if remaining < _STATE_READ_MIN_TIMEOUT_S: + raise RuntimeError(self._await_failure(seen, state, stuck=[])) state = await _trigger_state(deployed, remaining) 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}) @@ -188,13 +192,14 @@ async def _await_statuses(self, deployed: DeployedEnv, seen: dict[str, Optional[ logger.info(f"awaited triggers on env={self.env_id} reached {self.await_status!r}: {sorted(seen)}") return stuck = sorted(tid for tid, status in seen.items() if status in terminal) - remaining = deadline - time.monotonic() - if stuck or remaining <= 0: - why = (f"reached a terminal status before {self.await_status!r}: {stuck}" if stuck - else f"did not all reach {self.await_status!r} within {self.await_timeout_seconds:g}s") - raise RuntimeError(f"Env '{self.env_id}': awaited triggers {why}; statuses {dict(seen)}" - + _failure_detail(state, set(seen))) - await asyncio.sleep(min(_AWAIT_POLL_INTERVAL_S, remaining)) + 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]) -> str: + why = (f"reached a terminal status before {self.await_status!r}: {stuck}" if stuck + else f"did not all reach {self.await_status!r} within {self.await_timeout_seconds:g}s") + return f"Env '{self.env_id}': awaited triggers {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) @@ -219,20 +224,20 @@ def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agen async def _trigger_state(deployed: DeployedEnv, budget_s: float) -> dict: - """One trigger-state read, retried on failure, with each read and backoff cut to the remaining `budget_s`.""" + """One trigger-state read, retried on failure; each read and backoff is cut to the remaining `budget_s`, and no + read starts without a full minimum read window left.""" deadline = time.monotonic() + budget_s last_exc: Optional[Exception] = None for attempt in range(_STATE_READ_ATTEMPTS): remaining = deadline - time.monotonic() - if attempt and remaining <= 0: + if remaining < _STATE_READ_MIN_TIMEOUT_S: break - timeout = max(_STATE_READ_MIN_TIMEOUT_S, min(_STATE_READ_TIMEOUT_S, remaining)) try: - return await deployed.invoke(EXT_TRIGGERS_URI, "state", timeout=timeout) + return await deployed.invoke(EXT_TRIGGERS_URI, "state", timeout=min(_STATE_READ_TIMEOUT_S, remaining)) except Exception as e: # noqa: BLE001 - any read failure is retried, then reported with its cause last_exc = e await asyncio.sleep(max(0.0, min(_STATE_READ_BACKOFF_S * (attempt + 1), deadline - time.monotonic()))) - raise RuntimeError(f"could not read /triggers/state for env '{deployed.env_id}': {last_exc}") + raise RuntimeError(f"could not read /triggers/state for env '{deployed.env_id}' within the await budget: {last_exc}") def _failure_detail(state: dict, trigger_ids: set) -> str: diff --git a/tst/unit/task_step/test_register_env_triggers.py b/tst/unit/task_step/test_register_env_triggers.py index 7a9d5bd7..9fef3e60 100644 --- a/tst/unit/task_step/test_register_env_triggers.py +++ b/tst/unit/task_step/test_register_env_triggers.py @@ -373,7 +373,6 @@ async def test_no_await_means_no_state_read_and_no_awaited_key(monkeypatch): @pytest.mark.asyncio async def test_await_times_out_naming_statuses_and_failure_events(monkeypatch): - _no_sleep(monkeypatch) _mock_gateway_by_path(monkeypatch, { "/triggers/register": [{"ok": True, "added": ["t1", "t2"], "all": ["t1", "t2"]}], "/triggers/state": [_state({"t1": "fired", "t2": "armed"}, @@ -381,10 +380,12 @@ async def test_await_times_out_naming_statuses_and_failure_events(monkeypatch): "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=0.01).execute(_context()) + 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 reach 'fired' within 0.01s" in message + assert "did not all reach 'fired' 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 @@ -392,14 +393,13 @@ async def test_await_times_out_naming_statuses_and_failure_events(monkeypatch): @pytest.mark.asyncio async def test_a_failed_wait_still_records_the_registration_with_the_last_statuses(monkeypatch): - _no_sleep(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=0.01).execute(ctx) + 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"} @@ -423,10 +423,10 @@ async def stalled(self, uri, method, params=None, *, environment_name=None, time monkeypatch.setattr(DeployedGatewayEnv, "invoke", stalled) started = time.monotonic() - with pytest.raises(RuntimeError, match="could not read /triggers/state for env 'env-x'.*gateway stalled"): - await _step(await_fired=["t1"], await_timeout_seconds=0.3).execute(_context()) - assert time.monotonic() - started < 2.0 - assert timeouts and all(t <= 1 for t in timeouts), timeouts + with pytest.raises(RuntimeError, match="could not read /triggers/state for env 'env-x' within the await budget.*gateway stalled"): + 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 @@ -444,13 +444,23 @@ async def test_await_stops_early_on_a_failed_trigger(monkeypatch): @pytest.mark.asyncio async def test_a_trigger_missing_from_state_keeps_waiting_until_the_deadline(monkeypatch): - _no_sleep(monkeypatch) - _mock_gateway_by_path(monkeypatch, { + sent = _mock_gateway_by_path(monkeypatch, { "/triggers/register": [{"ok": True, "added": ["t1"], "all": ["t1"]}], "/triggers/state": [_state({})], }) - with pytest.raises(RuntimeError, match=r"statuses \{'t1': None\}"): - await _step(await_fired=["t1"], await_timeout_seconds=0.01).execute(_context()) + 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 From e7ba56060675ae735264790e8b6e6d219592a401 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 15 Oct 2026 16:57:19 -0700 Subject: [PATCH 5/6] register_env_triggers: the await has one target, fired `await_status` is gone: the warm-up only ever waits for triggers to fire, and a configurable target brought a validation branch, a computed terminal set and odd cases such as awaiting `failed`. The target and the terminal statuses are constants; the messages say "fire" instead of quoting a status. Co-Authored-By: Claude Fable 5.1 --- .../task_steps/register_env_triggers.py | 33 ++++++++----------- .../test_register_env_triggers_local.py | 2 +- .../task_step/test_register_env_triggers.py | 16 ++++----- 3 files changed, 21 insertions(+), 30 deletions(-) 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 2abb57a7..39e4e319 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 @@ -17,7 +17,6 @@ TRIGGER_EXECUTOR_HOURS, TRIGGER_NAMED_EXECUTORS, TRIGGER_PROVOKING_ROLES, - TRIGGER_STATUSES, TRIGGER_UNLESS, ) from agent_env.task_step.context import TaskStepContext @@ -27,7 +26,10 @@ logger = logging.getLogger(__name__) # The await polls the gateway's trigger state once per interval; a read gets this long, retried this often with this -# backoff, and never past the step's own deadline. A failure message quotes at most this many gateway events. +# backoff, and never past the step's own deadline. A failure message quotes at most this many gateway events. A trigger +# that reads one of the terminal statuses can no longer fire, so the wait stops there. +_AWAIT_TARGET_STATUS = "fired" +_AWAIT_TERMINAL_STATUSES = ("failed", "cancelled") _AWAIT_POLL_INTERVAL_S = 1.0 _STATE_READ_TIMEOUT_S = 15 _STATE_READ_ATTEMPTS = 3 @@ -41,8 +43,8 @@ class RegisterEnvTriggersStep(TaskStep): 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. - `await_fired` holds the step until the named one-shot triggers reach `await_status` (a recurrence re-arms after each - fire), so a dependent step starts only once start-of-run time triggers have landed in the world.""" + `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"),) @@ -58,7 +60,6 @@ def __init__( executor_timeout_seconds: float = 120, executor_hours: Optional[dict[str, dict]] = None, await_fired: Optional[list[str]] = None, - await_status: str = "fired", await_timeout_seconds: float = 120, depends_on: Optional[list[TaskStepDependency]] = None, fail_task_on_error: bool = True, @@ -69,8 +70,6 @@ def __init__( 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 await_status not in TRIGGER_STATUSES: - raise ValueError(f"await_status must be one of {list(TRIGGER_STATUSES)}, not {await_status!r}") if 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 @@ -80,7 +79,6 @@ def __init__( 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_status = await_status self.await_timeout_seconds = float(await_timeout_seconds) def to_dict(self) -> dict: @@ -92,7 +90,6 @@ def to_dict(self) -> dict: base["executor_timeout_seconds"] = self.executor_timeout_seconds base["executor_hours"] = self.executor_hours base["await_fired"] = self.await_fired - base["await_status"] = self.await_status base["await_timeout_seconds"] = self.await_timeout_seconds return base @@ -107,7 +104,6 @@ def from_dict(cls, data: dict) -> RegisterEnvTriggersStep: executor_timeout_seconds=data.get("executor_timeout_seconds", 120), executor_hours=data.get("executor_hours"), await_fired=data.get("await_fired"), - await_status=data.get("await_status", "fired"), await_timeout_seconds=data.get("await_timeout_seconds", 120), ) @@ -175,10 +171,9 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: 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 reads `await_status`; raise on a - terminal miss 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`.""" - terminal = {"failed", "cancelled"} - {self.await_status} + """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`.""" deadline = time.monotonic() + self.await_timeout_seconds state: dict = {} while True: @@ -188,17 +183,17 @@ async def _await_statuses(self, deployed: DeployedEnv, seen: dict[str, Optional[ state = await _trigger_state(deployed, remaining) 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 == self.await_status for status in seen.values()): - logger.info(f"awaited triggers on env={self.env_id} reached {self.await_status!r}: {sorted(seen)}") + 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 terminal) + 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]) -> str: - why = (f"reached a terminal status before {self.await_status!r}: {stuck}" if stuck - else f"did not all reach {self.await_status!r} within {self.await_timeout_seconds:g}s") + why = (f"reached a terminal status before firing: {stuck}" if stuck + else f"did not all fire within {self.await_timeout_seconds:g}s") return f"Env '{self.env_id}': awaited triggers {why}; statuses {dict(seen)}" + _failure_detail(state, set(seen)) def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agent_name: str) -> dict: 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 771d6586..dfd738f7 100644 --- a/tst/integration/task_step/test_register_env_triggers_local.py +++ b/tst/integration/task_step/test_register_env_triggers_local.py @@ -214,7 +214,7 @@ async def test_the_step_can_wait_for_start_of_run_time_triggers(local_stack): 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 reach 'fired' within 3s; statuses \{'later': 'armed'\}"): + 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) diff --git a/tst/unit/task_step/test_register_env_triggers.py b/tst/unit/task_step/test_register_env_triggers.py index 9fef3e60..64b4eee5 100644 --- a/tst/unit/task_step/test_register_env_triggers.py +++ b/tst/unit/task_step/test_register_env_triggers.py @@ -49,13 +49,11 @@ 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_status, rebuilt.await_timeout_seconds) == (None, "fired", 120) + assert (rebuilt.await_fired, rebuilt.await_timeout_seconds) == (None, 120) - awaiting = RegisterEnvTriggersStep.from_dict(_step(await_fired=["t1", "t1"], await_status="armed", - await_timeout_seconds=7).to_dict()) - assert (awaiting.await_fired, awaiting.await_status, awaiting.await_timeout_seconds) == (["t1"], "armed", 7.0) - stored_before_await = {k: v for k, v in _step().to_dict().items() - if k not in ("await_fired", "await_status", "await_timeout_seconds")} + 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 @@ -318,8 +316,6 @@ def handle(request: httpx.Request) -> httpx.Response: 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_status must be one of"): - _step(await_fired=["t1"], await_status="done") with pytest.raises(ValueError, match="await_timeout_seconds must be a positive number"): _step(await_fired=["t1"], await_timeout_seconds=0) @@ -385,7 +381,7 @@ async def test_await_times_out_naming_statuses_and_failure_events(monkeypatch): 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 reach 'fired' within 1.2s" in message + 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 @@ -437,7 +433,7 @@ async def test_await_stops_early_on_a_failed_trigger(monkeypatch): "/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 'fired': \['t2'\].*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"] From ebb27bfc0785b4042a074ffbaa8aa9f2b03fb494 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 15 Oct 2026 21:26:28 -0700 Subject: [PATCH 6/6] register_env_triggers: one deadline for the await, preflight enforced in execute The state read no longer has its own three-attempt retry with backoff; a failed read is retried on the next poll under the step's single deadline, and the last failure is reported if the deadline passes. execute raises preflight's own problems instead of a second copy of the unknown-id message, and await_timeout_seconds rejects booleans. Co-Authored-By: Claude Fable 5.1 --- .../task_steps/register_env_triggers.py | 77 +++++++++---------- .../task_step/test_register_env_triggers.py | 21 ++--- 2 files changed, 47 insertions(+), 51 deletions(-) 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 39e4e319..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 @@ -25,15 +25,14 @@ logger = logging.getLogger(__name__) -# The await polls the gateway's trigger state once per interval; a read gets this long, retried this often with this -# backoff, and never past the step's own deadline. A failure message quotes at most this many gateway events. A trigger -# that reads one of the terminal statuses can no longer fire, so the wait stops there. +# 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_ATTEMPTS = 3 -_STATE_READ_BACKOFF_S = 0.5 _STATE_READ_MIN_TIMEOUT_S = 1 _AWAIT_FAILURE_EVENTS_SHOWN = 6 @@ -70,7 +69,8 @@ def __init__( 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 not isinstance(await_timeout_seconds, (int, float)) or await_timeout_seconds <= 0: + 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 @@ -118,6 +118,9 @@ def _unknown_awaited(self) -> list[str]: 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") @@ -130,9 +133,6 @@ 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") - unknown_awaited = self._unknown_awaited() - if unknown_awaited: - raise RuntimeError(f"await_fired names {unknown_awaited}, which this step does not register") 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; " @@ -173,28 +173,40 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: 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`.""" + 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=[])) - state = await _trigger_state(deployed, remaining) - 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)) + 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]) -> str: - why = (f"reached a terminal status before firing: {stuck}" if stuck - else f"did not all fire within {self.await_timeout_seconds:g}s") - return f"Env '{self.env_id}': awaited triggers {why}; statuses {dict(seen)}" + _failure_detail(state, set(seen)) + 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) @@ -218,23 +230,6 @@ def _executor_config(self, context: TaskStepContext, deployed: DeployedEnv, agen return config -async def _trigger_state(deployed: DeployedEnv, budget_s: float) -> dict: - """One trigger-state read, retried on failure; each read and backoff is cut to the remaining `budget_s`, and no - read starts without a full minimum read window left.""" - deadline = time.monotonic() + budget_s - last_exc: Optional[Exception] = None - for attempt in range(_STATE_READ_ATTEMPTS): - remaining = deadline - time.monotonic() - if remaining < _STATE_READ_MIN_TIMEOUT_S: - break - try: - return await deployed.invoke(EXT_TRIGGERS_URI, "state", timeout=min(_STATE_READ_TIMEOUT_S, remaining)) - except Exception as e: # noqa: BLE001 - any read failure is retried, then reported with its cause - last_exc = e - await asyncio.sleep(max(0.0, min(_STATE_READ_BACKOFF_S * (attempt + 1), deadline - time.monotonic()))) - raise RuntimeError(f"could not read /triggers/state for env '{deployed.env_id}' within the await budget: {last_exc}") - - 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 diff --git a/tst/unit/task_step/test_register_env_triggers.py b/tst/unit/task_step/test_register_env_triggers.py index 64b4eee5..9e11b4bf 100644 --- a/tst/unit/task_step/test_register_env_triggers.py +++ b/tst/unit/task_step/test_register_env_triggers.py @@ -318,6 +318,8 @@ def test_bad_await_config_rejected_at_construction(): _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(): @@ -329,7 +331,7 @@ def test_preflight_names_an_awaited_trigger_this_step_does_not_register(): @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"await_fired names \['nope'\]"): + with pytest.raises(RuntimeError, match=r"register_env_triggers 'reg-1': await_fired names \['nope'\]"): await _step(await_fired=["nope"]).execute(_context()) assert sent == [] @@ -403,8 +405,8 @@ async def test_a_failed_wait_still_records_the_registration_with_the_last_status @pytest.mark.asyncio async def test_slow_state_reads_cannot_outlive_the_await_budget(monkeypatch): - """Each read's timeout and each retry's backoff are cut to the time left, so a stalled gateway cannot hold the step - for three full reads past the deadline.""" + """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"]}]}) @@ -419,7 +421,7 @@ async def stalled(self, uri, method, params=None, *, environment_name=None, time monkeypatch.setattr(DeployedGatewayEnv, "invoke", stalled) started = time.monotonic() - with pytest.raises(RuntimeError, match="could not read /triggers/state for env 'env-x' within the await budget.*gateway stalled"): + 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 @@ -460,7 +462,7 @@ async def test_a_budget_shorter_than_one_read_fails_without_reading(monkeypatch) @pytest.mark.asyncio -async def test_a_flaky_state_read_is_retried_then_reported(monkeypatch): +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"]}], @@ -472,15 +474,14 @@ async def test_a_flaky_state_read_is_retried_then_reported(monkeypatch): @pytest.mark.asyncio -async def test_a_state_read_that_keeps_failing_is_reported_with_its_cause(monkeypatch): - _no_sleep(monkeypatch) +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="could not read /triggers/state for env 'env-x'.*503"): - await _step(await_fired=["t1"]).execute(_context()) - assert [str(r.url.path) for r in sent].count("/triggers/state") == 3 + 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"},