From b4afd954538a406e674f5e64194259a93850069c Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 4 Nov 2026 12:51:14 -0800 Subject: [PATCH 1/5] feat(gateway): an nl action's action_ok counts the calls its executor made for the fire `action_ok` for an nl action meant only that the executor's A2A task completed; whether the colleague posted, or said nothing, took reading the env and a global changelog bracket that concurrent writes also move. Each fire in flight now counts its executor's tool calls by tool, so the detail reads `calls: {"slack_send_message": 1}` for a reply, `calls: {}` for a silence, and `failed_calls` for calls that came back as errors. Counts, not a list, so the event's size is bounded by the env's tools, not by how many calls an executor makes. Co-Authored-By: Claude Opus 5.5 --- src/agent_env/env/gateway/triggers.py | 33 +++++++++++-------- .../test_register_env_triggers_local.py | 3 ++ tst/unit/env/gateway/changes_test.py | 5 +-- tst/unit/env/gateway/role_forwarding_test.py | 2 +- tst/unit/env/gateway/triggers_test.py | 26 +++++++++++++++ 5 files changed, 53 insertions(+), 16 deletions(-) diff --git a/src/agent_env/env/gateway/triggers.py b/src/agent_env/env/gateway/triggers.py index a9a70bcb7..690be9b36 100644 --- a/src/agent_env/env/gateway/triggers.py +++ b/src/agent_env/env/gateway/triggers.py @@ -427,7 +427,7 @@ class TriggerEngine: def __init__(self, gateway): self._gw = gateway self._executors: dict[str | None, dict] = {} # `executor` under None, `executors` by name - self._in_flight: dict[str, list[dict]] = {} # executor role -> the fires it is realizing, for attribution + self._in_flight: dict[str, list[dict]] = {} # executor role -> the fires it is realizing: cause and calls made self._watch_roles: set[str] = set() self._config_set = False self._triggers: dict[str, dict] = {} @@ -913,6 +913,9 @@ def on_tool_call(self, role: str, name: str, arguments: dict, hold the call for: the widest of their timeouts, counting an omitted one as the gateway default so a neighbour's tighter ask cannot cut it short. """ + for fire in self._in_flight.get(role, ()): # an executor's call counts toward the fires it is realizing + counts = fire["failed_calls" if getattr(result, "isError", False) else "calls"] + counts[name] = counts.get(name, 0) + 1 provokable = [(tid, trig) for tid, trig in self._triggers.items() if self._provoked_by(trig, role)] if not provokable: return [], None @@ -1418,9 +1421,11 @@ async def _run_action(self, tid: str, trig: dict, index: int, action: dict, ctx: detail = {"notify": action["channel"], "to": note["to"], "notification_id": note["id"], "text_len": len(text)} else: - if not await self._run_nl(tid, trig, index, action, ctx): + fire = await self._run_nl(tid, trig, index, action, ctx) + if fire is None: return False - detail = {"nl": action["instruction"][:120], **_named_executor(action)} + detail = {"nl": action["instruction"][:120], **_named_executor(action), "calls": fire["calls"], + **({"failed_calls": fire["failed_calls"]} if fire["failed_calls"] else {})} if "verify" in action and action["type"] != "nl": # only an action's own verify inherits its `as`; when.check state probes read as the pin if not await self._run_verify(tid, index, action["verify"], ctx, role=action.get("as")): @@ -1442,38 +1447,40 @@ async def _run_verify(self, tid: str, index: int, verify: dict, ctx: dict | None self._emit("verify_failed", tid, action_index=index) return False - async def _run_nl(self, tid: str, trig: dict, index: int, action: dict, ctx: dict | None = None) -> bool: + async def _run_nl(self, tid: str, trig: dict, index: int, action: dict, ctx: dict | None = None) -> dict | None: + """The fire with the calls its executor made for it, by tool, or None when the action failed.""" instruction = action["instruction"] named = _named_executor(action) executor = self._executors.get(action.get("executor")) if executor is not None and "hours" in executor: await self._await_hours(tid, trig, index, executor["hours"], named) - cause = {"trigger_id": tid, "action_index": index, "fire": trig["attempts"]} - causes = self._in_flight.setdefault(executor["role"], []) if executor is not None else [] - causes.append(cause) + fire = {"cause": {"trigger_id": tid, "action_index": index, "fire": trig["attempts"]}, + "calls": {}, "failed_calls": {}} + fires = self._in_flight.setdefault(executor["role"], []) if executor is not None else [] + fires.append(fire) try: for attempt in (1, 2): reply = await self._call_executor(executor, instruction, context_id=f"trigger-{tid}") if reply is None: self._emit("action_failed", tid, action_index=index, detail=f"executor attempt {attempt} did not complete", **named) - return False + return None if "verify" not in action: - return True + return fire if await self._run_verify(tid, index, action["verify"], ctx): - return True + return fire if attempt == 1: instruction = (f"{action['instruction']}\n\nYour previous attempt did not pass verification — " f"the expected change is not observable in the environment. Do it again exactly as described.") finally: - causes.remove(cause) + fires.remove(fire) self._emit("action_failed", tid, action_index=index, detail="verification failed after retry", **named) - return False + return None def caused_by(self, role: str) -> list[dict]: """The fires in flight for the executor with ``role``, all of them when several overlap; its tool calls are attributed to them. A fire is the trigger's n-th, failed ones counted.""" - return list(self._in_flight.get(role, ())) + return [fire["cause"] for fire in self._in_flight.get(role, ())] async def _delay(self, tid: str, trig: dict, index: int, action: dict, since: float) -> None: """Wait until the action's delay has run since ``since``, the Clock.elapsed() reading of its cause.""" 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 a02d17fdd..a2a023a8c 100644 --- a/tst/integration/task_step/test_register_env_triggers_local.py +++ b/tst/integration/task_step/test_register_env_triggers_local.py @@ -89,6 +89,9 @@ async def test_each_nl_action_reaches_the_executor_it_names(items_env): "ALICE-ACTS by alice@example.com", "BOB-ACTS by bob@example.com"] named = {e["trigger_id"]: e["detail"].get("executor") for e in state["events"] if e["kind"] == "action_ok"} assert named == {"to-alice": "alice", "to-bob": "bob"} + acted = {"items_whoami": 1, "items_add_item": 1} + assert {e["trigger_id"]: e["detail"]["calls"] for e in state["events"] if e["kind"] == "action_ok"} == { + "to-alice": acted, "to-bob": acted} assert state["config"]["executors"] == ["alice", "bob"] assert context.metadata["env_trigger_registrations"][0]["executor_agent_names"] == ["alice", "bob"] diff --git a/tst/unit/env/gateway/changes_test.py b/tst/unit/env/gateway/changes_test.py index c7d068a57..be165bc9a 100644 --- a/tst/unit/env/gateway/changes_test.py +++ b/tst/unit/env/gateway/changes_test.py @@ -271,7 +271,7 @@ async def test_a_request_keeps_the_trigger_fire_it_started_under(monkeypatch): changelog = _Changelog() cause = {"trigger_id": "reply", "action_index": 0, "fire": 1} gw = _gateway(changelog, rest_proxy_urls={"mail": "http://mail-backend"}) - gw._trigger_engine._in_flight["bob@example.com"] = [cause] + gw._trigger_engine._in_flight["bob@example.com"] = [{"cause": cause, "calls": {}, "failed_calls": {}}] async def body(): yield b"a" @@ -304,7 +304,8 @@ def website(request: httpx.Request) -> httpx.Response: transport = httpx.MockTransport(website) monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw)) gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"}, rest_proxy_urls={"mail": "http://mail-backend"}) - gw._trigger_engine._in_flight["bob@example.com"] = [{"trigger_id": "reply", "action_index": 0, "fire": 2}] + gw._trigger_engine._in_flight["bob@example.com"] = [ + {"cause": {"trigger_id": "reply", "action_index": 0, "fire": 2}, "calls": {}, "failed_calls": {}}] before = len(_trajectory()) assert (await _http(gw, "GET", "/website/mail/inbox"))[0] == 200 diff --git a/tst/unit/env/gateway/role_forwarding_test.py b/tst/unit/env/gateway/role_forwarding_test.py index 10b8127af..1dd8dea75 100644 --- a/tst/unit/env/gateway/role_forwarding_test.py +++ b/tst/unit/env/gateway/role_forwarding_test.py @@ -88,7 +88,7 @@ def test_mcp_tool_calls_record_who_made_them_and_the_fires_in_flight_for_that_ex gw._clock.set_time("2019-06-15T12:00:00Z", 1) gw._register_get_time_tool() cause = {"trigger_id": "x", "action_index": 0, "fire": 1} - gw._trigger_engine._in_flight["bob@example.com"] = [cause] + gw._trigger_engine._in_flight["bob@example.com"] = [{"cause": cause, "calls": {}, "failed_calls": {}}] before = len(_trajectory(gw)) for role in ("bob@example.com", "alice@example.com"): diff --git a/tst/unit/env/gateway/triggers_test.py b/tst/unit/env/gateway/triggers_test.py index d28b5ee17..485863f7e 100644 --- a/tst/unit/env/gateway/triggers_test.py +++ b/tst/unit/env/gateway/triggers_test.py @@ -901,6 +901,32 @@ async def executor(self, executor, instruction, context_id): assert [causes[0]["fire"] for causes in seen] == [1, 2, 3] and engine._triggers["x"]["fire_count"] == 2 +@pytest.mark.asyncio +async def test_an_nl_action_ok_counts_the_calls_its_executor_made_for_the_fire(engine, monkeypatch): + acts = iter([[("slack_conversations_history", False), ("slack_send_message", True), ("slack_send_message", False)], + []]) + + async def executor(self, executor, instruction, context_id): + for tool, failed in next(acts): + engine.on_tool_call("bob", tool, {}, _result(is_error=failed)) + engine.on_tool_call("default", "gdocs_search_documents", {}, _result()) # someone else's call + return "done" + + monkeypatch.setattr(TriggerEngine, "_call_executor", executor) + engine.register({"executors": {"bob": {"a2a_url": "http://bob", "role": "bob"}}, + "triggers": [{"id": "x", "when": {"type": "action", "tool": "t", "repeat": True}, + "actions": [{"type": "nl", "executor": "bob", "instruction": "reply"}]}]}) + for _ in range(2): + engine.on_tool_call("default", "t", {}, _result()) + await asyncio.sleep(0.05) + engine.on_tool_call("bob", "slack_send_message", {}, _result()) # after its fires ended + + oks = [e["detail"] for e in engine._events if e["kind"] == "action_ok"] + assert [(d["calls"], d.get("failed_calls")) for d in oks] == [ + ({"slack_conversations_history": 1, "slack_send_message": 1}, {"slack_send_message": 1}), ({}, None)] + assert engine.caused_by("bob") == [] + + def test_generated_id_on_omit(engine): r = engine.register({"watch_roles": ["default"], "triggers": [ {"when": {"type": "action", "tool": "t"}, "actions": []}]}) From 30ab11189ccd37663773f9612b5fba9f9b0763a8 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 4 Nov 2026 12:52:46 -0800 Subject: [PATCH 2/5] test(gateway): the named-executors test expects the calls an nl action_ok now carries Co-Authored-By: Claude Opus 5.5 --- tst/unit/env/gateway/triggers_test.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tst/unit/env/gateway/triggers_test.py b/tst/unit/env/gateway/triggers_test.py index 485863f7e..b5c620ef8 100644 --- a/tst/unit/env/gateway/triggers_test.py +++ b/tst/unit/env/gateway/triggers_test.py @@ -350,8 +350,8 @@ async def test_each_nl_action_goes_to_the_executor_it_names(engine, executor_cal assert sorted((e["a2a_url"], i, c) for e, i, c in executor_calls) == [ ("http://alice", "as a", "trigger-a"), ("http://bob", "as b", "trigger-b"), ("http://default", "as d", "trigger-d")] details = {e["trigger_id"]: e["detail"] for e in engine._events if e["kind"] == "action_ok"} - assert details == {"d": {"nl": "as d"}, "a": {"nl": "as a", "executor": "alice"}, - "b": {"nl": "as b", "executor": "bob"}} + assert details == {"d": {"nl": "as d", "calls": {}}, "a": {"nl": "as a", "executor": "alice", "calls": {}}, + "b": {"nl": "as b", "executor": "bob", "calls": {}}} @pytest.mark.asyncio From 76f066a3e5950870aae561029bfc6fd519624da9 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 4 Nov 2026 13:57:54 -0800 Subject: [PATCH 3/5] feat(gateway): an nl fire names the fires it shared its executor's calls with An executor's calls carry only its role, so two fires it realizes at once both count every call made during either. `calls_shared_with` now names the overlapping triggers beside the counts, instead of leaving two identical tallies to read as twice the work. The in-flight record is built by `_in_flight_fire`, which the gateway tests that seed `_in_flight` use too. Co-Authored-By: Claude Opus 5.5 --- src/agent_env/env/gateway/triggers.py | 14 ++++++++++--- tst/unit/env/gateway/changes_test.py | 6 +++--- tst/unit/env/gateway/role_forwarding_test.py | 3 ++- tst/unit/env/gateway/triggers_test.py | 21 +++++++++++++++++++- 4 files changed, 36 insertions(+), 8 deletions(-) diff --git a/src/agent_env/env/gateway/triggers.py b/src/agent_env/env/gateway/triggers.py index 690be9b36..a493c0b3e 100644 --- a/src/agent_env/env/gateway/triggers.py +++ b/src/agent_env/env/gateway/triggers.py @@ -1425,7 +1425,8 @@ async def _run_action(self, tid: str, trig: dict, index: int, action: dict, ctx: if fire is None: return False detail = {"nl": action["instruction"][:120], **_named_executor(action), "calls": fire["calls"], - **({"failed_calls": fire["failed_calls"]} if fire["failed_calls"] else {})} + **({"failed_calls": fire["failed_calls"]} if fire["failed_calls"] else {}), + **({"calls_shared_with": sorted(fire["shared_with"])} if fire["shared_with"] else {})} if "verify" in action and action["type"] != "nl": # only an action's own verify inherits its `as`; when.check state probes read as the pin if not await self._run_verify(tid, index, action["verify"], ctx, role=action.get("as")): @@ -1454,9 +1455,11 @@ async def _run_nl(self, tid: str, trig: dict, index: int, action: dict, ctx: dic executor = self._executors.get(action.get("executor")) if executor is not None and "hours" in executor: await self._await_hours(tid, trig, index, executor["hours"], named) - fire = {"cause": {"trigger_id": tid, "action_index": index, "fire": trig["attempts"]}, - "calls": {}, "failed_calls": {}} + fire = _in_flight_fire({"trigger_id": tid, "action_index": index, "fire": trig["attempts"]}) fires = self._in_flight.setdefault(executor["role"], []) if executor is not None else [] + for other in fires: # an executor's calls carry only its role, so fires it realizes at once share their counts + other["shared_with"].add(tid) + fire["shared_with"].add(other["cause"]["trigger_id"]) fires.append(fire) try: for attempt in (1, 2): @@ -1597,6 +1600,11 @@ def _named_executor(action: dict) -> dict: return {"executor": action["executor"]} if action.get("executor") is not None else {} +def _in_flight_fire(cause: dict) -> dict: + """An nl fire as `_in_flight` holds it: its cause, its executor's calls by tool, and the triggers it overlapped.""" + return {"cause": cause, "calls": {}, "failed_calls": {}, "shared_with": set()} + + _DAYS = ("mon", "tue", "wed", "thu", "fri", "sat", "sun") _HHMM_RE = re.compile(r"([01]\d|2[0-3]):([0-5]\d)") _UTC_OFFSET_RE = re.compile(r"([+-])([01]\d|2[0-3]):([0-5]\d)") diff --git a/tst/unit/env/gateway/changes_test.py b/tst/unit/env/gateway/changes_test.py index be165bc9a..a4ca4c873 100644 --- a/tst/unit/env/gateway/changes_test.py +++ b/tst/unit/env/gateway/changes_test.py @@ -17,6 +17,7 @@ from agent_env.env.gateway.caller import Caller from agent_env.env.gateway.changes import ChangeAttribution from agent_env.env.gateway.gateway import GATEWAY_TRAJECTORY_FILE, Gateway +from agent_env.env.gateway.triggers import _in_flight_fire URL = "http://backing/mcp" TOOL = MCPTool(name="mail_send_message", description="", inputSchema={"type": "object", "properties": {}}) @@ -271,7 +272,7 @@ async def test_a_request_keeps_the_trigger_fire_it_started_under(monkeypatch): changelog = _Changelog() cause = {"trigger_id": "reply", "action_index": 0, "fire": 1} gw = _gateway(changelog, rest_proxy_urls={"mail": "http://mail-backend"}) - gw._trigger_engine._in_flight["bob@example.com"] = [{"cause": cause, "calls": {}, "failed_calls": {}}] + gw._trigger_engine._in_flight["bob@example.com"] = [_in_flight_fire(cause)] async def body(): yield b"a" @@ -304,8 +305,7 @@ def website(request: httpx.Request) -> httpx.Response: transport = httpx.MockTransport(website) monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw)) gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"}, rest_proxy_urls={"mail": "http://mail-backend"}) - gw._trigger_engine._in_flight["bob@example.com"] = [ - {"cause": {"trigger_id": "reply", "action_index": 0, "fire": 2}, "calls": {}, "failed_calls": {}}] + gw._trigger_engine._in_flight["bob@example.com"] = [_in_flight_fire({"trigger_id": "reply", "action_index": 0, "fire": 2})] before = len(_trajectory()) assert (await _http(gw, "GET", "/website/mail/inbox"))[0] == 200 diff --git a/tst/unit/env/gateway/role_forwarding_test.py b/tst/unit/env/gateway/role_forwarding_test.py index 1dd8dea75..8de3a8749 100644 --- a/tst/unit/env/gateway/role_forwarding_test.py +++ b/tst/unit/env/gateway/role_forwarding_test.py @@ -18,6 +18,7 @@ from agent_env.env.gateway.constants import DEFAULT_ROLE from agent_env.env.gateway.gateway import GATEWAY_TRAJECTORY_FILE, Gateway, _call_never_reached_child, _is_dead_session from agent_env.env.gateway.get_time import GET_TIME_TOOL_NAME +from agent_env.env.gateway.triggers import _in_flight_fire URL = "http://backing/mcp" TOOL = MCPTool(name="svc_thing", description="", inputSchema={"type": "object", "properties": {}}) @@ -88,7 +89,7 @@ def test_mcp_tool_calls_record_who_made_them_and_the_fires_in_flight_for_that_ex gw._clock.set_time("2019-06-15T12:00:00Z", 1) gw._register_get_time_tool() cause = {"trigger_id": "x", "action_index": 0, "fire": 1} - gw._trigger_engine._in_flight["bob@example.com"] = [{"cause": cause, "calls": {}, "failed_calls": {}}] + gw._trigger_engine._in_flight["bob@example.com"] = [_in_flight_fire(cause)] before = len(_trajectory(gw)) for role in ("bob@example.com", "alice@example.com"): diff --git a/tst/unit/env/gateway/triggers_test.py b/tst/unit/env/gateway/triggers_test.py index b5c620ef8..3cdd27bc0 100644 --- a/tst/unit/env/gateway/triggers_test.py +++ b/tst/unit/env/gateway/triggers_test.py @@ -924,7 +924,26 @@ async def executor(self, executor, instruction, context_id): oks = [e["detail"] for e in engine._events if e["kind"] == "action_ok"] assert [(d["calls"], d.get("failed_calls")) for d in oks] == [ ({"slack_conversations_history": 1, "slack_send_message": 1}, {"slack_send_message": 1}), ({}, None)] - assert engine.caused_by("bob") == [] + assert engine.caused_by("bob") == [] and not any("calls_shared_with" in d for d in oks) + + +@pytest.mark.asyncio +async def test_fires_one_executor_realizes_at_once_name_each_other_beside_the_calls_they_share(engine, monkeypatch): + async def executor(self, executor, instruction, context_id): + await asyncio.sleep(0.05) # both fires are in flight before either replies + engine.on_tool_call("bob", "slack_send_message", {}, _result()) + return "done" + + monkeypatch.setattr(TriggerEngine, "_call_executor", executor) + engine.register({"executors": {"bob": {"a2a_url": "http://bob", "role": "bob"}}, "triggers": [ + {"id": tid, "when": {"type": "action", "tool": "t"}, "actions": [{"type": "nl", "executor": "bob", "instruction": tid}]} + for tid in ("chan", "dm")]}) + engine.on_tool_call("default", "t", {}, _result()) + await asyncio.sleep(0.2) + + oks = {e["trigger_id"]: e["detail"] for e in engine._events if e["kind"] == "action_ok"} + assert {tid: (d["calls"], d["calls_shared_with"]) for tid, d in oks.items()} == { + "chan": ({"slack_send_message": 2}, ["dm"]), "dm": ({"slack_send_message": 2}, ["chan"])} def test_generated_id_on_omit(engine): From fc138a4e548d4c78788bc667dc3975d39e5ab8bd Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 4 Nov 2026 14:00:21 -0800 Subject: [PATCH 4/5] test(gateway): hold both fires open until both calls land in the overlap test Co-Authored-By: Claude Opus 5.5 --- tst/unit/env/gateway/triggers_test.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tst/unit/env/gateway/triggers_test.py b/tst/unit/env/gateway/triggers_test.py index 3cdd27bc0..c3815d673 100644 --- a/tst/unit/env/gateway/triggers_test.py +++ b/tst/unit/env/gateway/triggers_test.py @@ -929,9 +929,15 @@ async def executor(self, executor, instruction, context_id): @pytest.mark.asyncio async def test_fires_one_executor_realizes_at_once_name_each_other_beside_the_calls_they_share(engine, monkeypatch): + both_called = asyncio.Event() + called = [] + async def executor(self, executor, instruction, context_id): - await asyncio.sleep(0.05) # both fires are in flight before either replies engine.on_tool_call("bob", "slack_send_message", {}, _result()) + called.append(instruction) + if len(called) == 2: + both_called.set() + await both_called.wait() # neither fire ends before both calls are made return "done" monkeypatch.setattr(TriggerEngine, "_call_executor", executor) From c1ebff1be5f6ec371bfa055fa01024929a051484 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 4 Nov 2026 14:02:26 -0800 Subject: [PATCH 5/5] test(gateway): the overlap test opens both fires before either executor calls Co-Authored-By: Claude Opus 5.5 --- tst/unit/env/gateway/triggers_test.py | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/tst/unit/env/gateway/triggers_test.py b/tst/unit/env/gateway/triggers_test.py index c3815d673..f6897f64f 100644 --- a/tst/unit/env/gateway/triggers_test.py +++ b/tst/unit/env/gateway/triggers_test.py @@ -929,15 +929,12 @@ async def executor(self, executor, instruction, context_id): @pytest.mark.asyncio async def test_fires_one_executor_realizes_at_once_name_each_other_beside_the_calls_they_share(engine, monkeypatch): - both_called = asyncio.Event() - called = [] + started, called = _Gate(2), _Gate(2) async def executor(self, executor, instruction, context_id): + await started.arrive() # both fires are in flight before either executor calls engine.on_tool_call("bob", "slack_send_message", {}, _result()) - called.append(instruction) - if len(called) == 2: - both_called.set() - await both_called.wait() # neither fire ends before both calls are made + await called.arrive() # and neither ends before both calls are made return "done" monkeypatch.setattr(TriggerEngine, "_call_executor", executor) @@ -1181,6 +1178,19 @@ class _ExecutorCalls(list): reply: object = "done" +class _Gate: + """Holds each arrival until ``n`` have arrived.""" + + def __init__(self, n: int) -> None: + self.n, self.arrived, self.open = n, 0, asyncio.Event() + + async def arrive(self) -> None: + self.arrived += 1 + if self.arrived == self.n: + self.open.set() + await self.open.wait() + + @pytest.mark.asyncio @pytest.mark.parametrize("rate, virtual_seconds, steps", [(60, 1200, [5.0, 5.0, 5.0, 5.0]), (None, 12, [5.0, 5.0, 2.0])],