claire/tests/test_pull_liveness.py

"""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"