From f12cd040151ec485b953ae0a9f30abdcbb260f70 Mon Sep 17 00:00:00 2001
From: fangliquan <fangliquan@qq.com>
Date: Sat, 22 Aug 2026 13:25:14 +0800
Subject: [PATCH] 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.
---
 gateway/kanban_watchers.py               | 26 ++++++++++++++++++++----
 tests/gateway/test_platform_reconnect.py | 24 ++++++++++++++++++++++
 2 files changed, 46 insertions(+), 4 deletions(-)

diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py
index eb267c10ac..45504eb866 100644
--- a/gateway/kanban_watchers.py
+++ b/gateway/kanban_watchers.py
@@ -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:
diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py
index 7564f99ffe..e0c4e4a955 100644
--- a/tests/gateway/test_platform_reconnect.py
+++ b/tests/gateway/test_platform_reconnect.py
@@ -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