claire/tests/test_pull_idempotency.py

"""Regression test: repeated `pull()` against an unchanged fleet emits zero events.

The bug: when the same session UUID is visible from multiple hosts (mac-sync's
shared `~/.claude/projects/` makes plum's local sessions also visible from
apricot), the projection alternated host on each pull, emitting a spurious
`SessionObserved` every time.

The fix: dedup incoming session rows by UUID before diffing against the
projection — one canonical observation per pull.
"""

from __future__ import annotations

from uuid import UUID

from claire.pull import pull
from claire.rclaude import SessionRow, TmuxRow, TriageRow


class _FakeRclaude:
    """Deterministic rclaude stand-in returning the same rows every call."""

    def __init__(
        self,
        sessions: list[SessionRow],
        triage: list[TriageRow] | None = None,
        tmux: list[TmuxRow] | None = None,
    ):
        self._sessions = sessions
        self._triage = triage or []
        self._tmux = tmux or []

    def list_sessions(self) -> list[SessionRow]:
        return list(self._sessions)

    def list_tmux(self) -> list[TmuxRow]:
        return list(self._tmux)

    def triage(self) -> list[TriageRow]:
        return list(self._triage)


def test_pull_is_idempotent_across_macsync_hosts(conn, gen) -> None:
    """Same UUID reported by `local` + `apricot` must not oscillate."""
    shared_uuid = UUID("11111111-1111-1111-1111-111111111111")
    mtime = 1_700_000_000  # arbitrary fixed epoch
    cwd = "/Users/natalie/Code/@projects/@claire"

    rows = [
        SessionRow(host="local", uuid=shared_uuid, snippet="hi", cwd=cwd, mtime_epoch=mtime),
        SessionRow(host="apricot", uuid=shared_uuid, snippet="hi", cwd=cwd, mtime_epoch=mtime),
    ]
    fake = _FakeRclaude(sessions=rows)

    first = pull(conn, gen, rclaude=fake)
    assert first.sessions_observed >= 1, "first pull must observe the session"
    assert first.errors == []

    second = pull(conn, gen, rclaude=fake)
    assert second.sessions_observed == 0, (
        f"second pull should be a no-op, got {second.sessions_observed} observations"
    )
    assert second.errors == []

    # And a third pull also stays quiet — confirming we don't oscillate.
    third = pull(conn, gen, rclaude=fake)
    assert third.sessions_observed == 0


def test_pull_dedup_picks_highest_mtime(conn, gen) -> None:
    """When the same UUID has different mtimes per host, take the freshest."""
    shared_uuid = UUID("22222222-2222-2222-2222-222222222222")
    older = 1_700_000_000
    newer = 1_700_000_500
    cwd = "/tmp/work"

    rows_v1 = [
        SessionRow(host="apricot", uuid=shared_uuid, snippet="a", cwd=cwd, mtime_epoch=older),
        SessionRow(host="local", uuid=shared_uuid, snippet="a", cwd=cwd, mtime_epoch=newer),
    ]
    fake = _FakeRclaude(sessions=rows_v1)

    first = pull(conn, gen, rclaude=fake)
    assert first.sessions_observed == 1

    # Re-pull with the same data — must be idempotent.
    second = pull(conn, gen, rclaude=fake)
    assert second.sessions_observed == 0

    # Projection should reflect the row with the newer mtime (host=local).
    row = conn.execute(
        "SELECT host, last_seen_mtime FROM sessions WHERE uuid = ?",
        (str(shared_uuid),),
    ).fetchone()
    assert row["host"] == "local"


def test_pull_populates_tmux_name_from_resumed_uuid(conn, gen) -> None:
    """A TmuxRow with resumed_uuid maps that UUID → its tmux session name.

    This is the fix for Clare being blind to its own fleet: without
    sessions.tmux_name populated, `send_to_session` can't resolve a session
    UUID to anything `rclaude --match` can target.
    """
    sess_uuid = UUID("33333333-3333-3333-3333-333333333333")
    tmux_name = "claude-natalie-Users-natalie-Code-1779419135"
    cwd = "/Users/natalie/Code"

    # The session is on disk (list_sessions) AND backed by a live resumed
    # tmux pane (list_tmux carries the resumed_uuid).
    sessions = [
        SessionRow(host="local", uuid=sess_uuid, snippet="hi", cwd=cwd,
                   mtime_epoch=1_700_000_000),
    ]
    tmux = [
        TmuxRow(host="local", session_name=tmux_name, detail="1 windows",
                resumed_uuid=str(sess_uuid)),
    ]
    fake = _FakeRclaude(sessions=sessions, tmux=tmux)

    first = pull(conn, gen, rclaude=fake)
    assert first.errors == []
    # One observation for the disk row + one for the tmux mapping.
    assert first.sessions_observed == 2

    row = conn.execute(
        "SELECT tmux_name FROM sessions WHERE uuid = ?", (str(sess_uuid),)
    ).fetchone()
    assert row["tmux_name"] == tmux_name

    # Re-pull is idempotent — the tmux_name already matches, no new event.
    second = pull(conn, gen, rclaude=fake)
    assert second.sessions_observed == 0


def test_pull_tmux_name_ignored_for_fresh_session(conn, gen) -> None:
    """A TmuxRow without resumed_uuid (fresh spawn) yields no tmux_name mapping."""
    sess_uuid = UUID("44444444-4444-4444-4444-444444444444")
    sessions = [
        SessionRow(host="local", uuid=sess_uuid, snippet="hi",
                   cwd="/tmp/w", mtime_epoch=1_700_000_000),
    ]
    tmux = [
        TmuxRow(host="local", session_name="claude-natalie-tmp-w-1779419135",
                detail="1 windows", resumed_uuid=None),
    ]
    fake = _FakeRclaude(sessions=sessions, tmux=tmux)
    stats = pull(conn, gen, rclaude=fake)
    assert stats.errors == []
    # Only the disk-row observation — no tmux mapping for a fresh pane.
    assert stats.sessions_observed == 1
    row = conn.execute(
        "SELECT tmux_name FROM sessions WHERE uuid = ?", (str(sess_uuid),)
    ).fetchone()
    assert row["tmux_name"] is None