1  """Verify `hermes -c` picks the session the user most recently used."""
  2  
  3  from __future__ import annotations
  4  
  5  from hermes_cli.main import _resolve_last_session
  6  
  7  
  8  class _FakeDB:
  9      def __init__(self, rows):
 10          self._rows = rows
 11          self.closed = False
 12  
 13      def search_sessions(self, source=None, limit=20, **_kw):
 14          rows = [r for r in self._rows if r.get("source") == source] if source else list(self._rows)
 15          rows.sort(
 16              key=lambda r: float(r.get("last_active") or r.get("started_at") or 0),
 17              reverse=True,
 18          )
 19          return rows[:limit]
 20  
 21      def close(self):
 22          self.closed = True
 23  
 24  
 25  def test_resolve_last_session_prefers_last_active_over_started_at(monkeypatch):
 26      # `search_sessions` should return in MRU order, so -c can trust row 0.
 27      rows = [
 28          {
 29              "id": "new_started_old_active",
 30              "source": "cli",
 31              "started_at": 1000.0,
 32              "last_active": 100.0,
 33          },
 34          {
 35              "id": "old_started_recently_active",
 36              "source": "cli",
 37              "started_at": 500.0,
 38              "last_active": 999.0,
 39          },
 40      ]
 41  
 42      fake_db = _FakeDB(rows)
 43      monkeypatch.setattr("hermes_state.SessionDB", lambda: fake_db)
 44  
 45      assert _resolve_last_session("cli") == "old_started_recently_active"
 46      assert fake_db.closed
 47  
 48  
 49  def test_search_sessions_exposes_last_active_column(tmp_path, monkeypatch):
 50      # End-to-end: SessionDB must surface last_active and order by MRU.
 51      monkeypatch.setenv("HERMES_HOME", str(tmp_path))
 52      monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
 53  
 54      import hermes_state
 55  
 56      from pathlib import Path
 57  
 58      db = hermes_state.SessionDB(db_path=Path(tmp_path / "state.db"))
 59      try:
 60          db.create_session("s_started_later", source="cli")
 61          db.create_session("s_active_later", source="cli")
 62          # Force started_at ordering so the test is deterministic regardless
 63          # of how quickly the two inserts land.
 64          with db._lock:
 65              db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (2000.0, "s_started_later"))
 66              db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (1000.0, "s_active_later"))
 67              db._conn.commit()
 68  
 69          db.append_message("s_active_later", role="user", content="hi")
 70          with db._lock:
 71              db._conn.execute(
 72                  "UPDATE messages SET timestamp=? WHERE session_id=?",
 73                  (3000.0, "s_active_later"),
 74              )
 75              db._conn.commit()
 76  
 77          rows = db.search_sessions(source="cli", limit=5)
 78          ids = {r["id"]: r.get("last_active") for r in rows}
 79  
 80          assert ids["s_started_later"] == 2000.0
 81          assert ids["s_active_later"] == 3000.0
 82          assert rows[0]["id"] == "s_active_later"
 83      finally:
 84          db.close()
 85  
 86  
 87  def test_resolve_last_session_returns_none_when_empty(monkeypatch):
 88      monkeypatch.setattr("hermes_state.SessionDB", lambda: _FakeDB([]))
 89      assert _resolve_last_session("cli") is None
 90  
 91  
 92  def test_resolve_last_session_closes_db_on_search_error(monkeypatch):
 93      class _FailingDB:
 94          def __init__(self):
 95              self.closed = False
 96  
 97          def search_sessions(self, source=None, limit=20, **_kw):
 98              raise RuntimeError("boom")
 99  
100          def close(self):
101              self.closed = True
102  
103      db = _FailingDB()
104      monkeypatch.setattr("hermes_state.SessionDB", lambda: db)
105  
106      assert _resolve_last_session("cli") is None
107      assert db.closed is True
108  
109  
110  def test_resolve_last_session_falls_back_to_started_at(monkeypatch):
111      # When last_active is missing entirely (legacy row), fall back to
112      # started_at so the helper still picks the newest session.
113      rows = [
114          {"id": "older", "source": "cli", "started_at": 10.0},
115          {"id": "newer", "source": "cli", "started_at": 20.0},
116      ]
117      monkeypatch.setattr("hermes_state.SessionDB", lambda: _FakeDB(rows))
118      assert _resolve_last_session("cli") == "newer"
119  
120  
121  def test_resolve_last_session_not_limited_to_newest_started_20(tmp_path, monkeypatch):
122      # Regression: when sampling by started_at, -c could miss the true MRU if
123      # it was older than the newest 20 started sessions.
124      monkeypatch.setenv("HERMES_HOME", str(tmp_path))
125      monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
126  
127      import hermes_state
128  
129      from pathlib import Path
130  
131      state_db = Path(tmp_path / "state.db")
132      real_session_db = hermes_state.SessionDB
133      db = real_session_db(db_path=state_db)
134      try:
135          for i in range(25):
136              sid = f"s_{i:02d}"
137              db.create_session(sid, source="cli")
138              with db._lock:
139                  db._conn.execute(
140                      "UPDATE sessions SET started_at=? WHERE id=?",
141                      (10_000.0 - i, sid),
142                  )
143                  db._conn.commit()
144  
145          target = "s_24"
146          db.append_message(target, role="user", content="latest activity")
147          with db._lock:
148              db._conn.execute(
149                  "UPDATE messages SET timestamp=? WHERE session_id=?",
150                  (20_000.0, target),
151              )
152              db._conn.commit()
153      finally:
154          db.close()
155  
156      monkeypatch.setattr("hermes_state.SessionDB", lambda: real_session_db(db_path=state_db))
157      assert _resolve_last_session("cli") == target