fix(gateway): isolate kanban dispatcher to_thread context
Spawn-time Context isolation cannot rewrite an already-running watcher task. Run dispatcher SQLite offloads in an empty Context so write_txn no longer false-trips after delegate_task, while real child callers still hit the mutation guard.
This commit is contained in:
@@ -15,6 +15,7 @@ import logging
|
||||
import os
|
||||
import sqlite3
|
||||
import time
|
||||
from contextvars import Context
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
@@ -73,6 +74,23 @@ def _kanban_dispatch_allowed() -> bool:
|
||||
return not check_paused("kanban", logger)
|
||||
|
||||
|
||||
def _run_in_fresh_context(func: Callable[..., Any], /, *args: Any) -> Any:
|
||||
"""Run *func* in an empty ``Context`` so request-local ContextVars stay behind.
|
||||
|
||||
``asyncio.to_thread`` copies the calling task's context onto the worker
|
||||
thread. Supervised Kanban ticks are process-owned writers; if that copy
|
||||
still carries a ``delegate_task`` child marker (spawn-time snapshot or a
|
||||
later polluted task), ``write_txn`` false-trips. An empty Context keeps
|
||||
the DB guard intact for real children without exempting dispatcher writes.
|
||||
"""
|
||||
return Context().run(func, *args)
|
||||
|
||||
|
||||
async def _to_thread_process_service(func: Callable[..., Any], /, *args: Any) -> Any:
|
||||
"""Offload blocking dispatcher work without inheriting request-local ContextVars."""
|
||||
return await asyncio.to_thread(_run_in_fresh_context, func, *args)
|
||||
|
||||
|
||||
def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]":
|
||||
"""Take an exclusive, non-blocking advisory lock for the sole dispatcher.
|
||||
|
||||
@@ -1681,7 +1699,7 @@ class GatewayKanbanWatchersMixin:
|
||||
try:
|
||||
# Reap zombie children before per-board work so a board DB
|
||||
# failure cannot block cleanup of unrelated workers.
|
||||
pids = await asyncio.to_thread(_kb.reap_worker_zombies)
|
||||
pids = await _to_thread_process_service(_kb.reap_worker_zombies)
|
||||
if pids:
|
||||
logger.info(
|
||||
"kanban dispatcher: reaped %d zombie worker(s), pids=%s",
|
||||
@@ -1704,8 +1722,8 @@ class GatewayKanbanWatchersMixin:
|
||||
# takes effect on the next tick, not on gateway restart (#49638).
|
||||
_ad_enabled, _ad_per_tick = _read_auto_decompose_settings()
|
||||
if _ad_enabled:
|
||||
await asyncio.to_thread(_auto_decompose_tick, _ad_per_tick)
|
||||
results = await asyncio.to_thread(_tick_once)
|
||||
await _to_thread_process_service(_auto_decompose_tick, _ad_per_tick)
|
||||
results = await _to_thread_process_service(_tick_once)
|
||||
any_spawned = False
|
||||
for slug, res in (results or []):
|
||||
if res is not None and getattr(res, "spawned", None):
|
||||
@@ -1724,7 +1742,7 @@ class GatewayKanbanWatchersMixin:
|
||||
len(res.auto_blocked) if hasattr(res.auto_blocked, "__len__") else 0,
|
||||
)
|
||||
# Health telemetry (aggregate across boards)
|
||||
ready_pending = await asyncio.to_thread(_ready_nonempty)
|
||||
ready_pending = await _to_thread_process_service(_ready_nonempty)
|
||||
if ready_pending and not any_spawned:
|
||||
bad_ticks += 1
|
||||
else:
|
||||
|
||||
@@ -467,6 +467,30 @@ class TestSpawnSupervised:
|
||||
|
||||
assert observed == [False]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatcher_to_thread_does_not_trip_kanban_guard(self):
|
||||
"""Already-running ticks copy the caller task context into to_thread.
|
||||
|
||||
Issue #91958 is a dispatcher that was spawned at boot, then a later
|
||||
``delegate_task`` leaves the worker copy marked as a child. Spawn
|
||||
isolation does not rewrite that frozen task context; the offload
|
||||
helper must scrub it at the tick boundary. The in-task child guard
|
||||
stays closed.
|
||||
"""
|
||||
from agent.delegation_context import (
|
||||
delegated_child_context,
|
||||
is_delegated_child_context,
|
||||
)
|
||||
from gateway.kanban_watchers import _to_thread_process_service
|
||||
from hermes_cli.kanban_db import _assert_not_delegated_child_mutation
|
||||
|
||||
with delegated_child_context():
|
||||
assert is_delegated_child_context() is True
|
||||
with pytest.raises(PermissionError, match="delegate_task child"):
|
||||
_assert_not_delegated_child_mutation()
|
||||
await _to_thread_process_service(_assert_not_delegated_child_mutation)
|
||||
assert is_delegated_child_context() is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clean_synchronous_return_is_not_respawned(self):
|
||||
# A supervised coro that returns immediately (clean exit) must be
|
||||
|
||||
Reference in New Issue
Block a user