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)

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

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