53db597201
hermes_state.py: delete every '# noqa: F401 (re-exported...)' import block (hermes_state_common/errors/guard/
readpool/sessions/fts/dbfile/wal/repair/registry + agent.context_compressor _DB_PERSISTED_MARKER_KEY); keep
only the names hermes_state.py itself uses, without noqa.
hermes_state_registry.py: drop get_shared_session_db/release_shared_session_db/close_shared_session_dbs
aliases; every caller (gateway/, tools/, tui_gateway/, cron/, mcp_serve, run_agent, tests) now imports
acquire/release/close_all/release_or_close from hermes_state_registry.
hermes_state_titles.py: drop set_auto_title_if_empty shim (title_generator keeps its getattr fallback).
Re-remove shim-only names restored by 34abf954bd: latest_user_message_row_id (tests call
latest_message_row_id(key, role='user'); role-targeting assertions kept) and get_session_activity (tests
build the snapshot via agent.session_activity.build_activity_snapshot over db.get_session(sid)).
hermes_state_wal._log_once resolves its dedupe sets as module globals instead of via hermes_state;
hermes_state_repair helpers call module globals directly (tests patch hermes_state_repair.<name>).
Frozen updater surface untouched (update_cmd_maint imports only SessionDB from hermes_state).
44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
"""Recovery for legacy compression parents with no continuation child."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from agent.conversation_compression import recover_rotated_compression_session
|
|
from hermes_state import SessionDB
|
|
from hermes_state_errors import CompressionSessionClosedError
|
|
|
|
|
|
def test_recover_rotated_compression_session_reopens_legacy_orphan(tmp_path):
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
try:
|
|
db.create_session("orphan", source="cli")
|
|
db.append_message("orphan", "user", "before compression")
|
|
db.end_session("orphan", "compression")
|
|
agent = SimpleNamespace(_session_db=db, session_id="orphan")
|
|
|
|
assert recover_rotated_compression_session(agent) is None
|
|
db.append_message("orphan", "user", "after recovery")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_recover_rotated_compression_session_keeps_parent_closed_with_child(
|
|
tmp_path,
|
|
):
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
try:
|
|
db.create_session("parent", source="cli")
|
|
db.append_message("parent", "user", "before compression")
|
|
db.end_session("parent", "compression")
|
|
db.create_session("child", source="cli", parent_session_id="parent")
|
|
agent = SimpleNamespace(_session_db=db, session_id="parent")
|
|
|
|
assert recover_rotated_compression_session(agent) is None
|
|
try:
|
|
db.append_message("parent", "user", "must stay closed")
|
|
except CompressionSessionClosedError:
|
|
pass
|
|
else:
|
|
raise AssertionError("compression parent with child was reopened")
|
|
finally:
|
|
db.close()
|