fix(cli): synchronize HERMES_SESSION_ID across environment and contextvar during session switches

This commit is contained in:
novax635 2026-05-23 21:14:15 +03:00 committed by Teknium
parent f63ef74eaf
commit 86871ee25a
7 changed files with 109 additions and 14 deletions

View file

@ -168,6 +168,25 @@ class TestBranchCommandCLI:
assert cli_instance._resumed is True
def test_branch_rotates_hermes_session_id_env_and_context(self, cli_instance, session_db):
"""Branching must update process-local session-id readers too."""
from cli import HermesCLI
from gateway.session_context import _UNSET, _VAR_MAP, get_session_env
old_session_id = cli_instance.session_id
os.environ["HERMES_SESSION_ID"] = old_session_id
_VAR_MAP["HERMES_SESSION_ID"].set(old_session_id)
try:
HermesCLI._handle_branch_command(cli_instance, "/branch")
assert cli_instance.session_id != old_session_id
assert os.environ["HERMES_SESSION_ID"] == cli_instance.session_id
assert get_session_env("HERMES_SESSION_ID") == cli_instance.session_id
finally:
os.environ.pop("HERMES_SESSION_ID", None)
_VAR_MAP["HERMES_SESSION_ID"].set(_UNSET)
def test_branch_fires_on_session_switch_hook(self, cli_instance, session_db):
"""The /branch command must notify memory providers of the rotation.

View file

@ -333,6 +333,33 @@ class TestHistoryDisplay:
assert "Checking Running Hermes Agent" in output
assert "Use /resume <session id or title> to continue" in output
def test_resume_updates_hermes_session_id_env_and_context(self, tmp_path):
from gateway.session_context import _UNSET, _VAR_MAP, get_session_env
from hermes_state import SessionDB
cli = _make_cli()
cli.session_id = "current_session"
cli.conversation_history = []
cli.agent = None
cli._session_db = SessionDB(db_path=tmp_path / "state.db")
cli._session_db.create_session("current_session", "cli")
cli._session_db.create_session("target_session", "cli")
cli._session_db.append_message("target_session", "user", "hello from resumed session")
os.environ["HERMES_SESSION_ID"] = "current_session"
_VAR_MAP["HERMES_SESSION_ID"].set("current_session")
try:
cli._handle_resume_command("/resume target_session")
assert cli.session_id == "target_session"
assert os.environ["HERMES_SESSION_ID"] == "target_session"
assert get_session_env("HERMES_SESSION_ID") == "target_session"
finally:
cli._session_db.close()
os.environ.pop("HERMES_SESSION_ID", None)
_VAR_MAP["HERMES_SESSION_ID"].set(_UNSET)
def test_sessions_command_no_args_lists_recent_sessions(self, capsys):
"""/sessions with no args prints the recent-sessions table (TUI parity).

View file

@ -8,6 +8,8 @@ import sys
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
import pytest
from hermes_state import SessionDB
from tools.todo_tool import TodoStore
@ -138,6 +140,15 @@ def _prepare_cli_with_active_session(tmp_path):
return cli
@pytest.fixture(autouse=True)
def _reset_session_id_context():
from gateway.session_context import _UNSET, _VAR_MAP
yield
os.environ.pop("HERMES_SESSION_ID", None)
_VAR_MAP["HERMES_SESSION_ID"].set(_UNSET)
def test_new_command_creates_real_fresh_session_and_resets_agent_state(tmp_path):
cli = _prepare_cli_with_active_session(tmp_path)
old_session_id = cli.session_id
@ -164,6 +175,21 @@ def test_new_command_creates_real_fresh_session_and_resets_agent_state(tmp_path)
cli.agent._invalidate_system_prompt.assert_called_once()
def test_new_command_rotates_hermes_session_id_env_and_context(tmp_path):
from gateway.session_context import _VAR_MAP, get_session_env
cli = _prepare_cli_with_active_session(tmp_path)
old_session_id = cli.session_id
os.environ["HERMES_SESSION_ID"] = old_session_id
_VAR_MAP["HERMES_SESSION_ID"].set(old_session_id)
cli.process_command("/new")
assert cli.session_id != old_session_id
assert os.environ["HERMES_SESSION_ID"] == cli.session_id
assert get_session_env("HERMES_SESSION_ID") == cli.session_id
def test_reset_command_is_alias_for_new_session(tmp_path):
cli = _prepare_cli_with_active_session(tmp_path)
old_session_id = cli.session_id