From df08578643273f5e0815ab426600bc56a9cc6976 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 18 Oct 2026 20:50:03 -0700 Subject: [PATCH] fix(worker): route the async-delegation ledger's dispatch writes off Relay's loop External review of #141334: `_persist_dispatch` -> `WorkerDelegationLedger.dispatch` (and the `unscheduled` mark) run inside a tool callback on Relay's loop, so the `synchronous_rpc_on_event_loop` guard still fired and left the store failed. Both now go through `call_off_event_loop`, like the activity and aux-usage writers. Tests: a red/green test drives `_persist_dispatch` + `unscheduled` inside `relay_tools.execute`; one more drives the real `tool_executor._begin_tool_execution` frame from the reporter's ThreadPoolExecutor stack (passes at the prior head, guards against new store calls on that frame). The label-clear test's docstring now says it is defence in depth, not a reproduced caller. Refs #139029 Refs #134889 (KB-2) --- .../test_activity_persist_off_relay_loop.py | 53 ++++++++++++++++++- tools/async_delegation_worker.py | 7 +-- 2 files changed, 55 insertions(+), 5 deletions(-) diff --git a/tests/agent/test_activity_persist_off_relay_loop.py b/tests/agent/test_activity_persist_off_relay_loop.py index f2930023f7..2b0593a10d 100644 --- a/tests/agent/test_activity_persist_off_relay_loop.py +++ b/tests/agent/test_activity_persist_off_relay_loop.py @@ -43,7 +43,7 @@ class _OwnerReplies: self.on_loop.append(False) frame = json.loads(message) self.persisted.append(frame["params"]["operation"]) - self._reply = {"jsonrpc": "2.0", "id": request_id, "result": {"ok": True}} + self._reply = {"jsonrpc": "2.0", "id": request_id, "result": {"ok": True, "value": None}} def receive(self, timeout): return self._reply @@ -153,7 +153,10 @@ def test_later_turn_writes_still_land_after_an_on_loop_heartbeat(relay_turn, wor def test_label_reset_on_a_running_loop_persists_off_loop(worker_agent, caplog): - """Sibling writer in the same mixin: turn-teardown label clear reached on a loop.""" + """Sibling writer in the same mixin: turn-teardown label clear reached on a loop. + + Defence in depth, not a reproduced caller: no reported stack shows this path on a loop, so + this drives it from a synthetic ``asyncio.run``.""" agent, store, owner = worker_agent caplog.set_level(logging.WARNING, logger="agent.runtime_session_store") @@ -188,3 +191,49 @@ def test_sync_aux_usage_inside_managed_tool_persists_off_relay_loop(relay_turn, finally: reset_accounting_context(token) _assert_persisted_off_loop(store, owner, caplog, "usage.auxiliary") + + +def test_begin_tool_execution_inside_relay_tool_persists_off_relay_loop(relay_turn, worker_agent, caplog): + """Reporter stack 3 through the real frame: tool_executor._begin_tool_execution on a + ThreadPoolExecutor thread inside relay_tools.execute.""" + from agent import tool_executor + + agent, store, owner = worker_agent + caplog.set_level(logging.WARNING, logger="agent.runtime_session_store") + agent.quiet_mode, agent.tool_progress_callback, agent.tool_start_callback = True, None, None + agent._checkpoint_mgr = SimpleNamespace(enabled=False) + ref = tool_executor._ToolCallRef(name="image_generate", args={"prompt": "x"}, task_id="task-1", + call_id="call-1", trace=[]) + + def tool(args): + assert asyncio.get_running_loop() is not None + tool_executor._begin_tool_execution(agent, ref, 1) + return {"ok": True} + + with ThreadPoolExecutor(max_workers=1) as pool: + pool.submit(relay_tools.execute, "image_generate", {"prompt": "x"}, tool, + session_id=SESSION, tool_call_id="call-1").result(timeout=30) + _assert_persisted_off_loop(store, owner, caplog, "session.activity") + + +def test_async_delegation_ledger_writes_inside_relay_tool_persist_off_relay_loop( + relay_turn, worker_agent, caplog, monkeypatch): + """Reviewer probe: ``_persist_dispatch`` -> ``WorkerDelegationLedger.dispatch`` (and the + ``unscheduled`` mark) run inside the tool callback, on Relay's loop.""" + from tools import async_delegation + from tools.async_delegation_worker import WorkerDelegationLedger + + _agent, store, owner = worker_agent + caplog.set_level(logging.WARNING, logger="agent.runtime_session_store") + ledger = WorkerDelegationLedger(store) + monkeypatch.setattr(async_delegation, "worker_ledger", lambda: ledger) + + def tool(args): + assert asyncio.get_running_loop() is not None + async_delegation._persist_dispatch({"delegation_id": "d-1", "parent_session_id": SESSION, + "goal": "g", "dispatched_at": 1.0}) + ledger.unscheduled("d-1") + return {"ok": True} + + relay_tools.execute("cronjob", {"action": "run"}, tool, session_id=SESSION) + _assert_persisted_off_loop(store, owner, caplog, "delegation.dispatch", "delegation.unscheduled") diff --git a/tools/async_delegation_worker.py b/tools/async_delegation_worker.py index 605398ea76..2f686aa6b7 100644 --- a/tools/async_delegation_worker.py +++ b/tools/async_delegation_worker.py @@ -6,7 +6,7 @@ Fresh executor/helper threads retain the registered interpreter's owner boundary import os import threading -from agent.runtime_session_store import WorkerPersistenceError, is_worker_process +from agent.runtime_session_store import WorkerPersistenceError, call_off_event_loop, is_worker_process _store = None _lock = threading.Lock() @@ -46,7 +46,8 @@ class WorkerDelegationLedger: from tools.async_delegation import _ROUTING_KEYS if any(record.get(key) for key in _ROUTING_KEYS): raise WorkerPersistenceError('unsupported_producer') - return self.store._apply('delegation.dispatch', { + # Reached from a tool callback, on Relay's loop (#139029). + return call_off_event_loop(self.store._apply, 'delegation.dispatch', { key: record.get(key, '') for key in ('delegation_id', 'parent_session_id', 'session_key', 'origin_ui_session_id', 'origin_session_id', 'dispatched_at')} | {'task': { key: record[key] for key in ('goal', 'goals', 'context', 'toolsets', 'role', 'model', 'is_batch', 'task_indexes') @@ -77,4 +78,4 @@ class WorkerDelegationLedger: return self.store._apply('delegation.drop', {'delegation_id': delegation_id, 'claim_id': claim_id})['value'] def unscheduled(self, delegation_id): - return self.store._apply('delegation.unscheduled', {'delegation_id': delegation_id})['value'] + return call_off_event_loop(self.store._apply, 'delegation.unscheduled', {'delegation_id': delegation_id})['value']