fix: propagate caller contextvars into the Feishu adapter-owned executor

Moving the dedup flush and thread lookup from asyncio.to_thread onto the
adapter-owned pool fixed the torn-down default executor, but to_thread also
copies the caller's contextvars and run_in_executor does not. A multiplexed
profile's HERMES_HOME override and secret scope are contextvars, so those
workers silently ran under the launch profile. _run_blocking now runs the call
through contextvars.copy_context().run, matching to_thread semantics.

Review finding: _run_blocking lost the profile HERMES_HOME override / secret scope on the worker.
This commit is contained in:
teknium1
2026-09-14 21:35:30 -07:00
committed by Teknium
parent 854e02ca6a
commit 2298dc8122
2 changed files with 25 additions and 2 deletions
+7 -2
View File
@@ -1383,9 +1383,14 @@ class FeishuAdapter(BasePlatformAdapter):
return executor
async def _run_blocking(self, func, *args):
"""Run a blocking Feishu SDK call on the adapter-owned thread pool."""
"""Run a blocking Feishu SDK call on the adapter-owned thread pool.
``copy_context().run`` mirrors ``asyncio.to_thread``: the worker sees the caller's
profile HERMES_HOME override / secret scope (multiplexed dedup flush, thread lookup).
"""
loop = asyncio.get_running_loop()
return await loop.run_in_executor(self._get_sdk_executor(), func, *args)
return await loop.run_in_executor(
self._get_sdk_executor(), contextvars.copy_context().run, func, *args)
def _shutdown_sdk_executor(self) -> None:
"""Stop the adapter-owned SDK executor without touching the loop default."""
+18
View File
@@ -98,3 +98,21 @@ async def test_is_duplicate_flush_survives_default_executor_teardown(
finally:
loop._default_executor = original_executor
adapter._shutdown_sdk_executor()
@pytest.mark.asyncio
async def test_run_blocking_propagates_caller_contextvars(tmp_path, monkeypatch):
"""Call sites moved off asyncio.to_thread must keep seeing the caller's context: a
multiplexed profile's HERMES_HOME override is a contextvar, and a worker that lost it
would flush dedup state / look up threads under the wrong profile."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
from gateway.config import PlatformConfig
from hermes_constants import get_hermes_home_override, reset_hermes_home_override, set_hermes_home_override
adapter = FeishuAdapter(PlatformConfig())
token = set_hermes_home_override(str(tmp_path / "profiles" / "secondary"))
try:
assert await adapter._run_blocking(get_hermes_home_override) == str(tmp_path / "profiles" / "secondary")
finally:
reset_hermes_home_override(token)
adapter._shutdown_sdk_executor()