feat(gateway): an nl action's action_ok counts the calls its executor made for the fire

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)")

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"]
 

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

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"):

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])],