95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
"""Liveness pass: `sessions_alive` tracks live tmux panes, not disk JSONLs.
|
|
|
|
Regression for the over-count bug — the old pass marked *every* session at
|
|
a cwd with any live pane as `alive`, so a busy workspace with one live pane
|
|
and N historical session JSONLs reported N "alive". The fix ranks sessions
|
|
by recency per (host, cwd-slug) and marks only the freshest N, where N is
|
|
the number of live panes at that workspace.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from uuid import UUID
|
|
|
|
from claire.pull import pull
|
|
from claire.rclaude import SessionRow, TmuxRow
|
|
|
|
_CWD = "/var/home/lilith/Code/@projects/@claire"
|
|
_SLUG = "var-home-lilith-Code--projects--claire"
|
|
|
|
|
|
class _FakeRclaude:
|
|
def __init__(self, sessions: list[SessionRow], tmux: list[TmuxRow]):
|
|
self._sessions = sessions
|
|
self._tmux = tmux
|
|
|
|
def list_sessions(self) -> list[SessionRow]:
|
|
return list(self._sessions)
|
|
|
|
def list_tmux(self) -> list[TmuxRow]:
|
|
return list(self._tmux)
|
|
|
|
def triage(self) -> list:
|
|
return []
|
|
|
|
|
|
def _session(uuid_hex: str, mtime: int) -> SessionRow:
|
|
return SessionRow(
|
|
host="apricot", uuid=UUID(uuid_hex), snippet="x", cwd=_CWD, mtime_epoch=mtime,
|
|
)
|
|
|
|
|
|
def _liveness(conn, uuid_hex: str) -> str:
|
|
return conn.execute(
|
|
"SELECT liveness FROM sessions WHERE uuid = ?", (uuid_hex,)
|
|
).fetchone()["liveness"]
|
|
|
|
|
|
def test_one_live_pane_marks_only_freshest_session_alive(conn, gen) -> None:
|
|
"""3 disk sessions at one workspace, 1 live pane → only the newest alive."""
|
|
old = "11111111-1111-1111-1111-111111111111"
|
|
mid = "22222222-2222-2222-2222-222222222222"
|
|
new = "33333333-3333-3333-3333-333333333333"
|
|
fake = _FakeRclaude(
|
|
sessions=[
|
|
_session(old, 1_700_000_000),
|
|
_session(mid, 1_700_000_500),
|
|
_session(new, 1_700_001_000),
|
|
],
|
|
tmux=[TmuxRow(host="apricot", session_name=f"claude-natalie-{_SLUG}-1779326883", detail="")],
|
|
)
|
|
stats = pull(conn, gen, rclaude=fake)
|
|
assert stats.sessions_alive == 1
|
|
assert stats.sessions_closed == 2
|
|
assert _liveness(conn, new) == "alive"
|
|
assert _liveness(conn, mid) == "closed"
|
|
assert _liveness(conn, old) == "closed"
|
|
|
|
|
|
def test_alive_count_equals_live_pane_count(conn, gen) -> None:
|
|
"""4 disk sessions, 2 live panes → the 2 freshest are alive."""
|
|
uuids = [f"{i}{i}{i}{i}{i}{i}{i}{i}-0000-0000-0000-000000000000"[:36] for i in range(4)]
|
|
fake = _FakeRclaude(
|
|
sessions=[_session(u, 1_700_000_000 + i * 100) for i, u in enumerate(uuids)],
|
|
tmux=[
|
|
TmuxRow(host="apricot", session_name=f"claude-natalie-{_SLUG}-1779320000", detail=""),
|
|
TmuxRow(host="apricot", session_name=f"claude-natalie-{_SLUG}-1779320001", detail=""),
|
|
],
|
|
)
|
|
stats = pull(conn, gen, rclaude=fake)
|
|
assert stats.sessions_alive == 2
|
|
assert stats.sessions_closed == 2
|
|
# Freshest two (indices 3, 2) alive; oldest two (1, 0) closed.
|
|
assert _liveness(conn, uuids[3]) == "alive"
|
|
assert _liveness(conn, uuids[2]) == "alive"
|
|
assert _liveness(conn, uuids[1]) == "closed"
|
|
assert _liveness(conn, uuids[0]) == "closed"
|
|
|
|
|
|
def test_no_live_pane_marks_all_closed(conn, gen) -> None:
|
|
"""A workspace with disk sessions but no live pane → all closed."""
|
|
u = "44444444-4444-4444-4444-444444444444"
|
|
fake = _FakeRclaude(sessions=[_session(u, 1_700_000_000)], tmux=[])
|
|
stats = pull(conn, gen, rclaude=fake)
|
|
assert stats.sessions_alive == 0
|
|
assert _liveness(conn, u) == "closed"
|