common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(gateway): replace os.environ session state with contextvars for concurrency safety

Browse Source 
When two gateway messages arrived concurrently, _set_session_env wrote
HERMES_SESSION_PLATFORM/CHAT_ID/CHAT_NAME/THREAD_ID into the process-global
os.environ. Because asyncio tasks share the same process, Message B would
overwrite Message A's values mid-flight, causing background-task notifications
and tool calls to route to the wrong thread/chat.

Replace os.environ with Python's contextvars.ContextVar. Each asyncio task
(and any run_in_executor thread it spawns) gets its own copy, so concurrent
messages never interfere.

Changes:
- New gateway/session_context.py with ContextVar definitions, set/clear/get
  helpers, and os.environ fallback for CLI/cron/test backward compatibility
- gateway/run.py: _set_session_env returns reset tokens, _clear_session_env
  accepts them for proper cleanup in finally blocks
- All tool consumers updated: cronjob_tools, send_message_tool, skills_tool,
  terminal_tool (both notify_on_complete AND check_interval blocks), tts_tool,
  agent/skill_utils, agent/prompt_builder
- Tests updated for new contextvar-based API

Fixes #7358

Co-authored-by: teknium1 <127238744+teknium1@users.noreply.github.com>
This commit is contained in:
0xFrank-eth
2026-04-10 16:50:56 -07:00
committed by Teknium
parent dab5ec8245
commit e8034e2f6a
10 changed files with 255 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
tools/cronjob_tools.py
View File

@@ -64,14 +64,15 @@ def _scan_cron_prompt(prompt: str) -> str:
def _origin_from_env() -> Optional[Dict[str, str]]:
origin_platform = os.getenv("HERMES_SESSION_PLATFORM")
origin_chat_id = os.getenv("HERMES_SESSION_CHAT_ID")
from gateway.session_context import get_session_env
origin_platform = get_session_env("HERMES_SESSION_PLATFORM")
origin_chat_id = get_session_env("HERMES_SESSION_CHAT_ID")
if origin_platform and origin_chat_id:
return {
"platform": origin_platform,
"chat_id": origin_chat_id,
"chat_name": os.getenv("HERMES_SESSION_CHAT_NAME"),
"thread_id": os.getenv("HERMES_SESSION_THREAD_ID"),
"chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None,
"thread_id": get_session_env("HERMES_SESSION_THREAD_ID") or None,
}
return None

6
tools/send_message_tool.py
View File

@@ -212,7 +212,8 @@ def _handle_send(args):
if isinstance(result, dict) and result.get("success") and mirror_text:
try:
from gateway.mirror import mirror_to_session
source_label = os.getenv("HERMES_SESSION_PLATFORM", "cli")
from gateway.session_context import get_session_env
source_label = get_session_env("HERMES_SESSION_PLATFORM", "cli")
if mirror_to_session(platform_name, chat_id, mirror_text, source_label=source_label, thread_id=thread_id):
result["mirrored"] = True
except Exception:
@@ -1023,7 +1024,8 @@ async def _send_feishu(pconfig, chat_id, message, media_files=None, thread_id=No
def _check_send_message():
"""Gate send_message on gateway running (always available on messaging platforms)."""
platform = os.getenv("HERMES_SESSION_PLATFORM", "")
from gateway.session_context import get_session_env
platform = get_session_env("HERMES_SESSION_PLATFORM", "")
if platform and platform != "local":
return True
try:

3
tools/skills_tool.py
View File

@@ -347,7 +347,8 @@ def _capture_required_environment_variables(
def _is_gateway_surface() -> bool:
if os.getenv("HERMES_GATEWAY_SESSION"):
return True
return bool(os.getenv("HERMES_SESSION_PLATFORM"))
from gateway.session_context import get_session_env
return bool(get_session_env("HERMES_SESSION_PLATFORM"))
def _get_terminal_backend_name() -> str:

14
tools/terminal_tool.py
View File

@@ -1420,10 +1420,11 @@ def terminal_tool(
# In gateway mode, auto-register a fast watcher so the
# gateway can detect completion and trigger a new agent
# turn. CLI mode uses the completion_queue directly.
_gw_platform = os.getenv("HERMES_SESSION_PLATFORM", "")
from gateway.session_context import get_session_env as _gse
_gw_platform = _gse("HERMES_SESSION_PLATFORM", "")
if _gw_platform and not check_interval:
_gw_chat_id = os.getenv("HERMES_SESSION_CHAT_ID", "")
_gw_thread_id = os.getenv("HERMES_SESSION_THREAD_ID", "")
_gw_chat_id = _gse("HERMES_SESSION_CHAT_ID", "")
_gw_thread_id = _gse("HERMES_SESSION_THREAD_ID", "")
proc_session.watcher_platform = _gw_platform
proc_session.watcher_chat_id = _gw_chat_id
proc_session.watcher_thread_id = _gw_thread_id
@@ -1445,9 +1446,10 @@ def terminal_tool(
result_data["check_interval_note"] = (
f"Requested {check_interval}s raised to minimum 30s"
)
watcher_platform = os.getenv("HERMES_SESSION_PLATFORM", "")
watcher_chat_id = os.getenv("HERMES_SESSION_CHAT_ID", "")
watcher_thread_id = os.getenv("HERMES_SESSION_THREAD_ID", "")
from gateway.session_context import get_session_env as _gse2
watcher_platform = _gse2("HERMES_SESSION_PLATFORM", "")
watcher_chat_id = _gse2("HERMES_SESSION_CHAT_ID", "")
watcher_thread_id = _gse2("HERMES_SESSION_THREAD_ID", "")
# Store on session for checkpoint persistence
proc_session.watcher_platform = watcher_platform

3
tools/tts_tool.py
View File

@@ -480,7 +480,8 @@ def text_to_speech_tool(
# Telegram voice bubbles require Opus (.ogg); OpenAI and ElevenLabs can
# produce Opus natively (no ffmpeg needed). Edge TTS always outputs MP3
# and needs ffmpeg for conversion.
platform = os.getenv("HERMES_SESSION_PLATFORM", "").lower()
from gateway.session_context import get_session_env
platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower()
want_opus = (platform == "telegram")
# Determine output path