Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 24 additions & 4 deletions hermes_cli/console_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -585,24 +585,44 @@ def _logs(_engine: HermesConsoleEngine, args: list[str]) -> str:


def _session_db():
"""``with _session_db() as db:`` — SessionDB closed on exit."""
"""``with _session_db() as db:`` — writable SessionDB closed on exit.

For mutating commands only (rename, optimize, repair). Observational
commands use ``_session_db_readonly`` so a nested inspection never takes
writer privileges on a live store."""
from hermes_state import SessionDB
return closing(SessionDB())


def _session_db_readonly():
"""``with _session_db_readonly() as db:`` — read-only SessionDB closed on exit.

Raises ``ConsoleCommandError`` when no database exists yet: a read-only
open must never mint a store as a side effect (the writable default
scaffolds schema on first open). Path resolves at call time so a runtime
``HERMES_HOME`` redirect is honored.
"""
from hermes_state import SessionDB, _default_db_path

db_path = _default_db_path()
if not db_path.exists():
raise ConsoleCommandError(f"No session database at {db_path} yet.")
return closing(SessionDB(read_only=True))


def _sessions_list(_engine: HermesConsoleEngine, args: list[str]) -> str:
ns = _parse("sessions list", args, (("--limit",), dict(type=int, default=20)))
if ns.limit < 1 or ns.limit > 200:
raise ConsoleCommandError("sessions list --limit must be between 1 and 200")
with _session_db() as db:
with _session_db_readonly() as db:
sessions = db.list_sessions_rich(
exclude_sources=["kanban", "tool"], limit=ns.limit, order_by_last_active=True)
return _format_sessions(sessions)


def _sessions_stats(_engine: HermesConsoleEngine, args: list[str]) -> str:
_expect_no_args(args, "sessions stats")
with _session_db() as db:
with _session_db_readonly() as db:
total = db.session_count()
listable = db.session_count(exclude_children=True, exclude_sources=["kanban", "tool"])
lines = [
Expand Down Expand Up @@ -663,7 +683,7 @@ def _guard_exports(db, session_ids: list[str]) -> None:
@_captured
def _sessions_export(_engine: HermesConsoleEngine, args: list[str]) -> None:
ns = _parse("sessions export", args, "output", "--source", "--session-id")
with _session_db() as db:
with _session_db_readonly() as db:
if ns.session_id:
resolved_session_id = db.resolve_session_id(ns.session_id)
if not resolved_session_id:
Expand Down
8 changes: 5 additions & 3 deletions hermes_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -1155,14 +1155,16 @@ def _resolve_workspace_key() -> Optional[str]:

@contextlib.contextmanager
def _session_db():
"""Yield a ``SessionDB`` (lazy import, so test patches on ``hermes_state``
"""Yield a read-only ``SessionDB`` (lazy import, so test patches on ``hermes_state``
intercept). Open failures yield None and any error raised by the ``with``
body is swallowed — callers fall through to their ``return None``."""
body is swallowed — callers fall through to their ``return None``. Read-only
on purpose: every caller only queries (MRU search, title/ID resolve, cwd
restore), so this handle must never take writer privileges on a live store."""
db = None
try:
from hermes_state import SessionDB

db = SessionDB()
db = SessionDB(read_only=True)
except Exception:
pass
try:
Expand Down
2 changes: 1 addition & 1 deletion hermes_cli/main_tui_launch.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ def _print_tui_exit_summary(session_id: Optional[str], active_session_file: Opti
db = None
try:
from hermes_state import SessionDB
db = SessionDB()
db = SessionDB(read_only=True)
session = db.get_session(target)
if not session:
return
Expand Down
2 changes: 1 addition & 1 deletion hermes_cli/terminal_breadcrumbs.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ def resolve_breadcrumb_session() -> Optional[str]:
try:
from hermes_state import SessionDB

db = SessionDB()
db = SessionDB(read_only=True)
except Exception:
return None
try:
Expand Down
213 changes: 213 additions & 0 deletions tests/hermes_cli/test_observational_readonly_remaining.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,213 @@
"""Read-only observational openers outside the #109725/#110026/#110186 family.

``main._session_db``, terminal breadcrumb resolution, the TUI exit summary,
and console ``sessions list/stats/export`` only query state: they must open
``SessionDB(read_only=True)`` so a nested inspection never mints a second
writable WAL handle while a gateway owns the store. Console mutating paths
(``rename``/``optimize``/``repair``) intentionally stay writable.

Each mode test pins the constructor kwargs, so reverting any opener to a
bare ``SessionDB()`` turns its test red (sabotage-checked).
"""

import time

import pytest

import hermes_state
from hermes_state import SessionDB as _RealSessionDB
from hermes_cli import console_engine as ce
from hermes_cli import main as cli_main
from hermes_cli import main_tui_launch as tui_launch
from hermes_cli import terminal_breadcrumbs as tb


class _RecordingDB:
"""Fake SessionDB recording constructor kwargs; stub read/write surface."""

instances = []
seed = {}

def __init__(self, *args, **kwargs):
type(self).instances.append(kwargs)
self._sessions = dict(type(self).seed)

def close(self):
pass

# -- read surface used by the observational paths ---------------------
def get_session(self, session_id):
return self._sessions.get(session_id)

def get_session_title(self, session_id):
session = self._sessions.get(session_id) or {}
return session.get("title")

def get_compression_tip(self, session_id):
return session_id

def resolve_session_by_title(self, title):
return None

def resolve_session_id(self, session_id):
return session_id if session_id in self._sessions else None

def search_sessions(self, **kwargs):
return [{"id": sid} for sid in self._sessions]

def session_count(self, **kwargs):
return len(self._sessions)

def message_count(self, **kwargs):
return 0

def list_sessions_rich(self, **kwargs):
return [{"id": sid} for sid in self._sessions]

def export_session(self, session_id):
return {"id": session_id} if session_id in self._sessions else None

def export_all(self, **kwargs):
return [{"id": sid} for sid in self._sessions]

def assert_export_safe(self, session_id, max_messages=None):
return 0

# -- write surface (console rename must reach it through a WRITER) ----
def set_session_title(self, session_id, title):
return True


@pytest.fixture
def recording_db(monkeypatch):
_RecordingDB.instances.clear()
_RecordingDB.seed = {}
monkeypatch.setattr(hermes_state, "SessionDB", _RecordingDB)
return _RecordingDB


def _last_kwargs():
assert _RecordingDB.instances, "expected SessionDB to be constructed"
return _RecordingDB.instances[-1]


def _ensure_real_store():
"""Production precondition for readers: a real store file exists.

Called before the class is faked, so the missing-db guard passes and
the mode assertion observes the real open path selection.
"""
db = _RealSessionDB()
try:
db.create_session("seed-1", source="cli")
finally:
db.close()


# ------------------------------------------------------------------ modes

def test_main_session_db_opens_readonly(recording_db, _isolate_hermes_home):
with cli_main._session_db():
pass
assert _last_kwargs().get("read_only") is True


def test_breadcrumb_resolve_opens_readonly(recording_db, _isolate_hermes_home, monkeypatch):
_RecordingDB.seed = {"sid-1": {"id": "sid-1"}}
monkeypatch.setattr(tb, "is_enabled", lambda: True)
monkeypatch.setattr(
tb, "read_breadcrumb", lambda: {"session_id": "sid-1", "ts": time.time()}
)
assert tb.resolve_breadcrumb_session() == "sid-1"
assert _last_kwargs().get("read_only") is True


def test_tui_exit_summary_opens_readonly(
recording_db, _isolate_hermes_home, monkeypatch, capsys
):
# Seed one visible session through the fake before the summary runs.
_RecordingDB.seed = {"sid-9": {"title": "hello", "message_count": 3}}
tui_launch._print_tui_exit_summary("sid-9")
out = capsys.readouterr().out
assert "sid-9" in out
assert _last_kwargs().get("read_only") is True


@pytest.mark.parametrize("line", ["sessions list", "sessions stats"])
def test_console_list_stats_open_readonly(
recording_db, _isolate_hermes_home, line
):
_ensure_real_store()
engine = ce.HermesConsoleEngine()
result = engine.execute(line)
assert result.status == "ok"
assert _last_kwargs().get("read_only") is True


def test_console_export_opens_readonly(recording_db, _isolate_hermes_home):
_ensure_real_store()
engine = ce.HermesConsoleEngine()
result = engine.execute("sessions export - --source cli", confirmed=True)
assert result.status == "ok"
assert _last_kwargs().get("read_only") is True


def test_console_rename_stays_writable(recording_db, _isolate_hermes_home):
_RecordingDB.seed = {"sid-1": {"id": "sid-1"}}
text = ce._sessions_rename(None, ["sid-1", "New", "Title"])
assert "renamed" in text
assert "read_only" not in _last_kwargs()


def test_console_list_on_missing_db_fails_closed_without_minting(
_isolate_hermes_home,
):
"""No database yet: friendly error, and no store file created."""
from hermes_state import _default_db_path

engine = ce.HermesConsoleEngine()
result = engine.execute("sessions list")
assert result.status == "error"
assert "No session database" in result.output
assert not _default_db_path().exists()


# ------------------------------------------------------- live-writer proof

def test_live_writer_survives_observational_readers(
_isolate_hermes_home, monkeypatch, capsys
):
"""A writable holder stays usable after every converted reader runs."""
from hermes_state import SessionDB

writer = SessionDB()
try:
writer.create_session("live-1", source="cli")
writer.append_message("live-1", "user", "hello")

# 1. breadcrumb resolve (read-only after the fix).
monkeypatch.setattr(tb, "is_enabled", lambda: True)
monkeypatch.setattr(
tb, "read_breadcrumb", lambda: {"session_id": "live-1", "ts": time.time()}
)
assert tb.resolve_breadcrumb_session() == "live-1"

# 2. main MRU/title lookups (read-only after the fix).
assert cli_main._resolve_last_session(source="cli") == "live-1"

# 3. console list + stats (read-only after the fix).
engine = ce.HermesConsoleEngine()
assert engine.execute("sessions list").status == "ok"
assert engine.execute("sessions stats").status == "ok"

# 4. TUI exit summary (read-only after the fix).
tui_launch._print_tui_exit_summary("live-1")
assert "live-1" in capsys.readouterr().out

# The live writer's generation is intact: keep writing and reading.
writer.append_message("live-1", "assistant", "still here")
writer.create_session("live-2", source="cli")
assert writer.get_session("live-2")["id"] == "live-2"
assert writer.get_session("live-1")["message_count"] == 2
finally:
writer.close()
4 changes: 3 additions & 1 deletion tests/hermes_cli/test_resolve_last_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -131,5 +131,7 @@ def test_resolve_last_session_real_db_prefers_workspace(monkeypatch, tmp_path):
cmd, 0, stdout=str(repo_a), stderr=""
),
)
monkeypatch.setattr("hermes_state.SessionDB", lambda: real_db(db_path=state_db))
monkeypatch.setattr(
"hermes_state.SessionDB", lambda *a, **k: real_db(db_path=state_db, **k)
)
assert _resolve_last_session("cli") == "repo_a"