feat: expose HERMES_SESSION_ID to agent tools via ContextVar + env (#23847)

Set HERMES_SESSION_ID using the existing session_context.py ContextVar
system for concurrency safety (multiple gateway sessions in one process
won't cross-talk). Also writes os.environ as fallback for CLI mode.

Touchpoints:
- gateway/session_context.py: Add _SESSION_ID ContextVar + _VAR_MAP entry
- run_agent.py: Set both ContextVar and os.environ at init and on
  context-compression rotation
- tools/environments/local.py: Bridge ContextVars into subprocess env
  in _make_run_env() (ContextVars don't propagate to child processes)
- tests/run_agent/test_session_id_env.py: 3 tests covering env, provided
  ID, and ContextVar paths

execute_code subprocess already passes HERMES_* prefixed vars through
_scrub_child_env (line 82: _SAFE_ENV_PREFIXES includes 'HERMES_').

Primary use case: webhook-triggered agents that need to include a
`--resume <session_id>` takeover command in their output.

gateway/session_context.py

@@ -55,6 +55,7 @@ _SESSION_THREAD_ID: ContextVar = ContextVar("HERMES_SESSION_THREAD_ID", default=
 _SESSION_USER_ID: ContextVar = ContextVar("HERMES_SESSION_USER_ID", default=_UNSET)
 _SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET)
 _SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET)
+_SESSION_ID: ContextVar = ContextVar("HERMES_SESSION_ID", default=_UNSET)
 
 # Cron auto-delivery vars — set per-job in run_job() so concurrent jobs
 # don't clobber each other's delivery targets.
@@ -70,6 +71,7 @@ _VAR_MAP = {
     "HERMES_SESSION_USER_ID": _SESSION_USER_ID,
     "HERMES_SESSION_USER_NAME": _SESSION_USER_NAME,
     "HERMES_SESSION_KEY": _SESSION_KEY,
+    "HERMES_SESSION_ID": _SESSION_ID,
     "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM,
     "HERMES_CRON_AUTO_DELIVER_CHAT_ID": _CRON_AUTO_DELIVER_CHAT_ID,
     "HERMES_CRON_AUTO_DELIVER_THREAD_ID": _CRON_AUTO_DELIVER_THREAD_ID,

run_agent.py

@@ -1837,7 +1837,20 @@ class AIAgent:
             timestamp_str = self.session_start.strftime("%Y%m%d_%H%M%S")
             short_uuid = uuid.uuid4().hex[:6]
             self.session_id = f"{timestamp_str}_{short_uuid}"
-        
+
+        # Expose session ID to tools (terminal, execute_code) so agents can
+        # reference their own session for --resume commands, cross-session
+        # coordination, and logging.  Uses the ContextVar system from
+        # session_context.py for concurrency safety (gateway runs multiple
+        # sessions in one process).  Also writes os.environ as fallback for
+        # CLI mode where ContextVars aren't used.
+        os.environ["HERMES_SESSION_ID"] = self.session_id
+        try:
+            from gateway.session_context import _SESSION_ID
+            _SESSION_ID.set(self.session_id)
+        except Exception:
+            pass  # CLI/test mode — ContextVar not needed
+
         # Session logs go into ~/.hermes/sessions/ alongside gateway sessions
         hermes_home = get_hermes_home()
         self.logs_dir = hermes_home / "sessions"
@@ -10233,6 +10246,12 @@ class AIAgent:
                 self._session_db.end_session(self.session_id, "compression")
                 old_session_id = self.session_id
                 self.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
+                os.environ["HERMES_SESSION_ID"] = self.session_id
+                try:
+                    from gateway.session_context import _SESSION_ID
+                    _SESSION_ID.set(self.session_id)
+                except Exception:
+                    pass
                 # Update session_log_file to point to the new session's JSON file
                 self.session_log_file = self.logs_dir / f"session_{self.session_id}.json"
                 self._session_db_created = False

tests/run_agent/test_session_id_env.py

@@ -0,0 +1,61 @@
+"""Test that HERMES_SESSION_ID is exposed as an env var and ContextVar."""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../.."))
+
+from run_agent import AIAgent
+
+
+@pytest.fixture(autouse=True)
+def _cleanup_env():
+    """Remove HERMES_SESSION_ID before/after each test."""
+    os.environ.pop("HERMES_SESSION_ID", None)
+    yield
+    os.environ.pop("HERMES_SESSION_ID", None)
+
+
+def test_session_id_env_set_on_init():
+    """AIAgent.__init__ sets HERMES_SESSION_ID in the environment."""
+    agent = AIAgent(
+        api_key="test-key",
+        base_url="https://openrouter.ai/api/v1",
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+    )
+    assert os.environ.get("HERMES_SESSION_ID") == agent.session_id
+    assert len(agent.session_id) > 0
+
+
+def test_session_id_env_uses_provided_id():
+    """When session_id is passed explicitly, HERMES_SESSION_ID reflects it."""
+    custom_id = "20260511_120000_abc12345"
+    agent = AIAgent(
+        api_key="test-key",
+        base_url="https://openrouter.ai/api/v1",
+        session_id=custom_id,
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+    )
+    assert os.environ["HERMES_SESSION_ID"] == custom_id
+    assert agent.session_id == custom_id
+
+
+def test_session_id_contextvar_set():
+    """AIAgent.__init__ also sets the ContextVar for concurrency safety."""
+    custom_id = "20260511_130000_def67890"
+    AIAgent(
+        api_key="test-key",
+        base_url="https://openrouter.ai/api/v1",
+        session_id=custom_id,
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+    )
+    from gateway.session_context import get_session_env
+    assert get_session_env("HERMES_SESSION_ID") == custom_id

tools/environments/local.py

@@ -274,6 +274,17 @@ def _make_run_env(env: dict) -> dict:
     if _profile_home:
         run_env["HOME"] = _profile_home
 
+    # Inject ContextVar-based session vars into subprocess env.
+    # ContextVars don't propagate to child processes, so we bridge them here.
+    try:
+        from gateway.session_context import get_session_env, _UNSET, _VAR_MAP
+        for var_name, var in _VAR_MAP.items():
+            value = var.get()
+            if value is not _UNSET and value:
+                run_env[var_name] = value
+    except Exception:
+        pass
+
     return run_env