diff --git a/src/agent_env/env/gateway/triggers.py b/src/agent_env/env/gateway/triggers.py index a9a70bcb7..a493c0b3e 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,12 @@ 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 {}), + **({"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")): @@ -1442,38 +1448,42 @@ 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 = _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): 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.""" @@ -1590,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/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..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] + gw._trigger_engine._in_flight["bob@example.com"] = [_in_flight_fire(cause)] async def body(): yield b"a" @@ -304,7 +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"] = [{"trigger_id": "reply", "action_index": 0, "fire": 2}] + 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 10b8127af..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] + 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 d28b5ee17..f6897f64f 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 @@ -901,6 +901,54 @@ 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") == [] 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): + 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()) + await called.arrive() # and neither ends before both calls are made + 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): r = engine.register({"watch_roles": ["default"], "triggers": [ {"when": {"type": "action", "tool": "t"}, "actions": []}]}) @@ -1130,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])],