fix(gateway): isolate supervised watcher contexts
This commit is contained in:
+16
-5
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user