refactor(terminal): extract _current_session_key() helper for session-key lookups

Follow-up to the session-scoping fix: _get_sudo_password_cache_scope()
and _resolve_container_task_id() carried byte-identical copies of the
HERMES_SESSION_KEY lookup (contextvar + os.environ fallback). Collapse
both onto one helper adopting the bare-import convention approval.py
already uses — get_session_env() implements the fallback internally, so
the old try/except could only fire on import failure, where silently
degrading to process-global semantics would reintroduce exactly the
cross-session contamination the fix prevents.
This commit is contained in:
kshitijk4poor
2026-08-22 14:55:41 +05:30
committed by kshitij
parent 8a963e8512
commit 8e475ed27b
+15 -12
View File
@@ -272,6 +272,19 @@ def _get_sudo_password_callback():
return getattr(_callback_tls, "sudo_password", None)
def _current_session_key() -> str:
"""Return the active gateway/WebUI session key, or "" outside sessions.
Single lookup point for the ``HERMES_SESSION_KEY`` ContextVar with the
os.environ fallback that ``get_session_env()`` applies for CLI, cron, and
test processes. Callers scope per-session caches by prefixing the value
with ``"session:"`` so two sessions never share a cache slot.
"""
from gateway.session_context import get_session_env
return get_session_env("HERMES_SESSION_KEY", "")
def _get_approval_callback():
return getattr(_callback_tls, "approval", None)
@@ -297,12 +310,7 @@ def set_approval_callback(cb):
def _get_sudo_password_cache_scope() -> str:
"""Return the cache scope for interactive sudo passwords."""
try:
from gateway.session_context import get_session_env
session_key = get_session_env("HERMES_SESSION_KEY", "")
except Exception:
session_key = os.getenv("HERMES_SESSION_KEY", "")
session_key = _current_session_key()
if session_key:
return f"session:{session_key}"
@@ -1398,12 +1406,7 @@ def _resolve_container_task_id(task_id: Optional[str]) -> str:
# branches above: those paths already key containers per task_id, so they
# stay authoritative where they apply and this only covers the cases that
# would otherwise collapse to the shared "default" key (notably SSH).
try:
from gateway.session_context import get_session_env
session_key = get_session_env("HERMES_SESSION_KEY", "")
except Exception:
session_key = os.getenv("HERMES_SESSION_KEY", "")
session_key = _current_session_key()
if session_key:
return f"session:{session_key}"
return "default"