From bf3a0bb99d348416fd17cf3574da3d1d730fb88f Mon Sep 17 00:00:00 2001 From: fangliquan Date: Sat, 22 Aug 2026 09:13:01 +0800 Subject: [PATCH] fix(gateway): isolate supervised watcher contexts --- gateway/run.py | 21 ++++++++++++++++----- tests/gateway/test_platform_reconnect.py | 22 ++++++++++++++++++++++ 2 files changed, 38 insertions(+), 5 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index c973c2a029..90de3e880f 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -42,7 +42,7 @@ import threading import time import traceback from collections import OrderedDict -from contextvars import copy_context +from contextvars import Context, copy_context from pathlib import Path from datetime import datetime, timedelta, timezone from typing import Awaitable, Callable, Dict, Optional, Any, List, Tuple, Union, cast @@ -13483,6 +13483,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ``_SUPERVISED_HEALTHY_SECS`` — so a long-lived daemon that crashes occasionally over days is never permanently abandoned. + Each watcher starts in a fresh ``Context``. These are process-level + services, not continuations of whichever message turn happened to + spawn them; inheriting a delegated-child marker would make the Kanban + dispatcher reject its own writes when ``asyncio.to_thread`` copies the + watcher's context. + ``on_spawn`` (optional) is invoked with the freshly-created task on every spawn, INCLUDING internal backoff respawns. Callers that also track the live handle elsewhere (e.g. ``self._reconnect_watcher_task`` @@ -13509,9 +13515,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # uses it to distinguish a rapid crash-loop from a healthy-run-then-crash. _started = time.monotonic() - # Deliberately do NOT pass name= to create_task — some test doubles mock - # create_task with a signature that rejects the name kwarg. - task = asyncio.create_task(coro_factory()) + # Deliberately do NOT pass kwargs to create_task — some test doubles + # mock it with a narrow signature. Calling it from a fresh Context has + # the same isolation semantics as create_task(..., context=Context()) + # while preserving that compatibility. + task = Context().run(lambda: asyncio.create_task(coro_factory())) # Mark this as a PERMANENT supervised watcher, not transient background # WORK. The scale-to-zero idle check must ignore these: supervised # watchers (session-expiry, kanban, reconnect, the scale-to-zero watcher @@ -13587,7 +13595,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew on_give_up=on_give_up, ) - respawn_task = asyncio.create_task(_respawn()) + # The done callback retains the context in which it was + # registered, so isolate the backoff task too; otherwise a + # restart could reintroduce the original caller's turn scope. + respawn_task = Context().run(lambda: asyncio.create_task(_respawn())) self._background_tasks.add(respawn_task) respawn_task.add_done_callback(self._background_tasks.discard) diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index 8b100eebcf..7564f99ffe 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -445,6 +445,28 @@ class TestPlatformSlashCommand: class TestSpawnSupervised: """Verify the task-level supervision wrapper around watcher launches.""" + @pytest.mark.asyncio + async def test_watcher_does_not_inherit_delegated_child_context(self): + """Long-lived gateway services must not inherit one turn's child scope.""" + from agent.delegation_context import ( + delegated_child_context, + is_delegated_child_context, + ) + + runner = _make_runner() + observed = [] + + async def _watcher(): + observed.append(is_delegated_child_context()) + + with delegated_child_context(): + assert is_delegated_child_context() is True + task = runner._spawn_supervised(_watcher, "isolated_watcher") + await task + assert is_delegated_child_context() is True + + assert observed == [False] + @pytest.mark.asyncio async def test_clean_synchronous_return_is_not_respawned(self): # A supervised coro that returns immediately (clean exit) must be