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
25 changes: 17 additions & 8 deletions tests/tools/test_session_search.py
Original file line number Diff line number Diff line change
Expand Up @@ -523,27 +523,36 @@ def _patch_profiles(self, monkeypatch, home, exists=True):
monkeypatch.setattr(profiles_mod, "profile_exists", lambda n: exists)
monkeypatch.setattr(profiles_mod, "get_profile_dir", lambda n: home)

def test_bare_id_locates_across_profiles(self, db, tmp_path, monkeypatch):
def test_bare_id_miss_is_generic_no_owner_oracle(self, db, tmp_path, monkeypatch):
# The real-world failure: model dropped the owning profile and passed a
# bare id. The tool must scan profiles and find it anyway.
# bare id. Fail closed with a generic miss β€” no cross-profile scan and
# no owner oracle; naming the profile is the authorized read path (#106761).
other_home = tmp_path / "asdf_home"
other_home.mkdir()
other = SessionDB(other_home / "state.db")
other.create_session("s_far", source="cli")
other.append_message("s_far", role="user", content="hi")
other._conn.commit()

from collections import namedtuple
from hermes_cli import profiles as profiles_mod
Info = namedtuple("Info", "name path")
monkeypatch.setattr(profiles_mod, "get_profile_dir", lambda n: tmp_path / "default_home")
monkeypatch.setattr(profiles_mod, "list_profiles", lambda: [Info("asdf", other_home)])
monkeypatch.setattr(profiles_mod, "get_profile_dir",
lambda n: other_home if n == "asdf" else tmp_path / "default_home")
monkeypatch.setattr(profiles_mod, "normalize_profile_name", lambda n: n)
monkeypatch.setattr(profiles_mod, "validate_profile_name", lambda n: None)
monkeypatch.setattr(profiles_mod, "profile_exists", lambda n: n == "asdf")

# `db` (current profile) lacks s_far; no profile passed β†’ scan finds it.
# `db` (current profile) lacks s_far; no profile passed β†’ generic miss.
# The error must not reveal which (if any) profile owns the id.
result = json.loads(session_search(session_id="s_far", db=db))
assert result["success"] is False
assert "asdf" not in result["error"]
assert "messages" not in result

# The named re-run is the authorized path and must still read it.
result = json.loads(session_search(session_id="s_far", profile="asdf", db=db))
assert result["success"] is True
assert result["mode"] == "read"
assert result["profile"] == "asdf"
assert result["session_id"] == "s_far"


def test_combined_value_autosplits(self, db, tmp_path, monkeypatch):
Expand Down
41 changes: 4 additions & 37 deletions tools/session_search_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -339,32 +339,6 @@ def _resolve_profile_db(profile: str):
return SessionDB(db_path=profiles_mod.get_profile_dir(canon) / "state.db", read_only=True)


def _locate_session_db(session_id: str):
"""Scan every profile's ``state.db`` -> ``(db, profile_name)`` or ``(None, None)``.
Ids are globally unique, so the first hit is authoritative."""
from pathlib import Path
try:
from hermes_cli import profiles as profiles_mod
from hermes_state import SessionDB
except Exception:
return None, None
targets = [("default", profiles_mod.get_profile_dir("default"))] + _quiet(
lambda: [(info.name, info.path) for info in profiles_mod.list_profiles()], [],
"list_profiles failed during session locate")
seen: set = set()
for name, home in targets:
db_path = Path(home) / "state.db"
if str(db_path) in seen or not db_path.exists():
continue
seen.add(str(db_path))
pdb = _quiet(lambda: SessionDB(db_path=db_path, read_only=True), None, "open %s failed", db_path)
if pdb and _get_session_meta(pdb, session_id):
return pdb, name
if pdb:
pdb.close()
return None, None


def _read_session(db, session_id: str, head: int = 20, tail: int = 10, link_profile: str = None) -> str:
"""Read shape: whole session, or ``head`` + ``tail`` messages with a scroll pointer."""
meta = _get_session_meta(db, session_id)
Expand All @@ -384,17 +358,10 @@ def _read_session(db, session_id: str, head: int = 20, tail: int = 10, link_prof


def _read_with_profile_fallback(db, sid: str, profile: Optional[str]) -> str:
"""Read shape; on a miss scan every profile (the model may have dropped the owning
profile from the link) and tag the result with where it was found."""
result = _read_session(db, sid, link_profile=profile)
located, owner = (None, None) if json.loads(result).get("success") else _locate_session_db(sid)
if located is None:
return result
try:
found = json.loads(_read_session(located, sid, link_profile=owner))
finally:
located.close()
return json.dumps({**found, "profile": owner}, ensure_ascii=False) if found.get("success") else result
"""Read shape; a bare id never scans other profiles' databases and a miss never
names an owner β€” another profile's transcript is only served when the caller
names that profile."""
return _read_session(db, sid, link_profile=profile)


def _list_recent_sessions(db, limit: int, current_session_id: str = None, link_profile: str = None) -> str:
Expand Down