tests/plugins/test_honcho_session_cache_bounds.py

"""Regression tests for bounded growth of the Honcho local session cache.

Covers the fix for unbounded RSS growth in long-running gateways: prior to
this, ``HonchoSession.messages`` grew forever (never trimmed after a sync),
and ``HonchoSessionManager``'s ``_cache``/``_sessions_cache``/``_context_cache``
had no eviction path short of an explicit ``/new`` reset.
"""

import threading
import time
from datetime import datetime, timedelta
from types import SimpleNamespace

import pytest

from plugins.memory.honcho.session import (
    HonchoSession,
    HonchoSessionManager,
    _PEERS_CACHE_MAX_SIZE,
    _SESSION_CACHE_MAX_SIZE,
    _SESSION_IDLE_TTL_SECONDS,
    _SESSION_MESSAGE_RETENTION,
)


def _session(key="k", honcho_session_id=None):
    return HonchoSession(
        key=key,
        user_peer_id="user",
        assistant_peer_id="assistant",
        honcho_session_id=honcho_session_id or f"hs-{key}",
    )


def _manager():
    cfg = SimpleNamespace(
        write_frequency="turn",
        dialectic_reasoning_level="low",
        dialectic_dynamic=True,
        dialectic_max_chars=600,
        observation_mode="directional",
        user_observe_me=True,
        user_observe_others=True,
        ai_observe_me=True,
        ai_observe_others=True,
        message_max_chars=25000,
        dialectic_max_input_chars=10000,
    )
    return HonchoSessionManager(honcho=SimpleNamespace(), config=cfg)


def test_trim_synced_messages_caps_total_length():
    session = _session()
    for i in range(250):
        session.add_message("user", f"msg{i}", _synced=True)
    for i in range(250, 255):
        session.add_message("user", f"msg{i}", _synced=False)
    assert len(session.messages) == 255

    HonchoSessionManager._trim_synced_messages(session)

    assert len(session.messages) == _SESSION_MESSAGE_RETENTION
    # oldest synced messages are the ones dropped; the unsynced tail survives intact
    assert not any(m.get("_synced") is False for m in session.messages[:-5])
    assert all(m.get("_synced") is False for m in session.messages[-5:])


def test_trim_synced_messages_never_drops_an_unsynced_message():
    session = _session()
    for i in range(300):
        session.add_message("user", f"m{i}", _synced=(i != 10))

    HonchoSessionManager._trim_synced_messages(session)

    contents = [m["content"] for m in session.messages]
    assert "m10" in contents
    idx = contents.index("m10")
    assert session.messages[idx].get("_synced") is False
    # trimming stops at the first unsynced message from the front — it does
    # not skip past it to keep reducing, so everything from there on survives
    assert contents[idx:] == [f"m{i}" for i in range(10, 300)]


def test_trim_synced_messages_is_a_noop_under_the_cap():
    session = _session()
    for i in range(10):
        session.add_message("user", f"m{i}", _synced=True)

    HonchoSessionManager._trim_synced_messages(session)

    assert len(session.messages) == 10


def test_sweep_idle_sessions_evicts_stale_entries_across_all_caches():
    mgr = _manager()
    stale = _session(key="stale", honcho_session_id="hs-stale")
    stale.updated_at = datetime.now() - timedelta(seconds=_SESSION_IDLE_TTL_SECONDS + 1)
    fresh = _session(key="fresh", honcho_session_id="hs-fresh")

    mgr._cache = {"stale": stale, "fresh": fresh}
    mgr._sessions_cache = {"hs-stale": object(), "hs-fresh": object()}
    mgr._context_cache = {"stale": {"x": 1}, "fresh": {"x": 1}}

    evicted = mgr._sweep_idle_sessions_locked()

    assert evicted == 1
    assert set(mgr._cache) == {"fresh"}
    assert set(mgr._sessions_cache) == {"hs-fresh"}
    assert set(mgr._context_cache) == {"fresh"}


def test_sweep_idle_sessions_keeps_fresh_entries():
    mgr = _manager()
    fresh = _session(key="fresh")
    mgr._cache = {"fresh": fresh}
    mgr._sessions_cache = {fresh.honcho_session_id: object()}
    mgr._context_cache = {"fresh": {}}

    evicted = mgr._sweep_idle_sessions_locked()

    assert evicted == 0
    assert "fresh" in mgr._cache


def test_maybe_sweep_idle_sessions_is_rate_limited():
    mgr = _manager()
    stale = _session(key="stale")
    stale.updated_at = datetime.now() - timedelta(seconds=_SESSION_IDLE_TTL_SECONDS + 1)
    mgr._cache = {"stale": stale}

    mgr._last_idle_sweep_ts = time.time()  # just swept — this call should no-op
    mgr._maybe_sweep_idle_sessions()
    assert "stale" in mgr._cache

    mgr._last_idle_sweep_ts = 0.0  # force the interval to have elapsed
    mgr._maybe_sweep_idle_sessions()
    assert "stale" not in mgr._cache


def test_get_or_create_triggers_sweep_without_blocking_on_lock_reentrancy():
    """`_cache_lock` is an RLock specifically so a sweep triggered from inside
    `get_or_create` (which also takes the lock) can't deadlock the manager's
    own thread. Guard against that regressing silently.
    """
    mgr = _manager()
    stale = _session(key="stale")
    stale.updated_at = datetime.now() - timedelta(seconds=_SESSION_IDLE_TTL_SECONDS + 1)
    mgr._cache = {"stale": stale}
    mgr._last_idle_sweep_ts = 0.0

    done = threading.Event()

    def call_it():
        mgr._maybe_sweep_idle_sessions()
        done.set()

    t = threading.Thread(target=call_it)
    t.start()
    t.join(timeout=5)
    assert done.is_set(), "sweep did not complete — possible deadlock"
    assert "stale" not in mgr._cache


# ---------------------------------------------------------------------------
# hard caps, unsynced buffers, peers, and read activity (follows #71463)
# ---------------------------------------------------------------------------



def _fill_sessions(mgr, count, unsynced_keys=()):
    for i in range(count):
        key = f"k{i}"
        session = _session(key=key)
        if key in unsynced_keys:
            session.add_message("user", "pending", _synced=False)
        mgr._cache[key] = session
        mgr._sessions_cache[session.honcho_session_id] = object()
        mgr._session_observation[session.honcho_session_id] = {"ai_observe_others": False}
        mgr._context_cache[key] = {"representation": "r"}


def test_size_cap_evicts_least_recently_used_sessions_with_their_entries():
    mgr = _manager()
    _fill_sessions(mgr, _SESSION_CACHE_MAX_SIZE + 2)

    with mgr._cache_lock:
        mgr._enforce_cache_caps_locked()

    assert len(mgr._cache) == _SESSION_CACHE_MAX_SIZE
    assert "k0" not in mgr._cache and "k1" not in mgr._cache
    assert "k2" in mgr._cache
    for gone in ("k0", "k1"):
        assert f"hs-{gone}" not in mgr._sessions_cache
        assert f"hs-{gone}" not in mgr._session_observation
        assert gone not in mgr._context_cache
    assert "hs-k2" in mgr._sessions_cache and "hs-k2" in mgr._session_observation


def test_size_cap_never_evicts_a_session_with_unsynced_messages():
    mgr = _manager()
    _fill_sessions(mgr, _SESSION_CACHE_MAX_SIZE + 1, unsynced_keys={"k0"})

    with mgr._cache_lock:
        mgr._enforce_cache_caps_locked()

    assert "k0" in mgr._cache  # the only copy until its flush lands
    assert "k1" not in mgr._cache
    assert len(mgr._cache) == _SESSION_CACHE_MAX_SIZE


def test_idle_sweep_keeps_sessions_with_unsynced_messages():
    mgr = _manager()
    stale = _session(key="stale")
    stale.add_message("user", "pending", _synced=False)
    stale.updated_at = datetime.now() - timedelta(seconds=_SESSION_IDLE_TTL_SECONDS + 1)
    mgr._cache = {"stale": stale}

    with mgr._cache_lock:
        evicted = mgr._sweep_idle_sessions_locked()

    assert evicted == 0
    assert "stale" in mgr._cache


def test_peers_cap_evicts_unreferenced_peers_oldest_first():
    mgr = _manager()
    live = _session(key="live")
    mgr._cache = {"live": live}
    mgr._peers_cache[live.user_peer_id] = object()
    mgr._peers_cache[live.assistant_peer_id] = object()
    for i in range(_PEERS_CACHE_MAX_SIZE):
        mgr._peers_cache[f"guest{i}"] = object()

    with mgr._cache_lock:
        mgr._enforce_cache_caps_locked()

    assert len(mgr._peers_cache) == _PEERS_CACHE_MAX_SIZE
    assert live.user_peer_id in mgr._peers_cache and live.assistant_peer_id in mgr._peers_cache
    assert "guest0" not in mgr._peers_cache and "guest1" not in mgr._peers_cache
    assert f"guest{_PEERS_CACHE_MAX_SIZE - 1}" in mgr._peers_cache


def test_recall_read_counts_as_activity_for_the_idle_sweep():
    mgr = _manager()
    session = _session(key="read-only")
    session.updated_at = datetime.now() - timedelta(seconds=_SESSION_IDLE_TTL_SECONDS + 1)
    mgr._cache = {"read-only": session}

    assert mgr._cached_session("read-only") is session
    with mgr._cache_lock:
        evicted = mgr._sweep_idle_sessions_locked()

    assert evicted == 0
    assert "read-only" in mgr._cache


def test_sdk_object_hit_moves_the_key_to_the_recent_end():
    mgr = _manager()
    mgr._peers_cache = {"a": object(), "b": object()}

    mgr._cached_sdk_object(mgr._peers_cache, "a", lambda: None)

    assert list(mgr._peers_cache) == ["b", "a"]


def test_get_or_create_stores_observation_flags_with_the_entry_and_eviction_drops_them():
    mgr = _manager()
    mgr._config.ai_peer = "hermes"
    mgr._config.peer_name = "operator"  # unnamed peers now fail closed instead of minting a fallback
    flags = {"user_observe_me": True, "user_observe_others": True, "ai_observe_me": True, "ai_observe_others": False}
    mgr._get_or_create_peer = lambda peer_id: object()
    mgr._get_or_create_honcho_session = lambda sid, user, assistant: (object(), [], dict(flags))

    session = mgr.get_or_create("cli:one")

    assert mgr._session_observation[session.honcho_session_id] == flags
    assert mgr._ai_observes_others(session) is False
    with mgr._cache_lock:
        mgr._evict_session_locked("cli:one", session)
    assert session.honcho_session_id not in mgr._session_observation


def test_cap_enforcement_drops_observation_flags_a_post_eviction_flush_stored():
    """A flush that rebuilds an evicted session's SDK session stores its flags again; the next cap pass
    must prune that orphan like every other per-session entry, or the dict grows one entry per evicted-then-
    flushed session."""
    mgr = _manager()
    live = _session(key="live")
    mgr._cache = {"live": live}
    mgr._session_observation = {live.honcho_session_id: {"ai_observe_others": True}, "hs-gone": {"ai_observe_others": False}}

    with mgr._cache_lock:
        mgr._enforce_cache_caps_locked()

    assert set(mgr._session_observation) == {live.honcho_session_id}


def test_flush_does_not_resurrect_an_evicted_session():
    mgr = _manager()
    session = _session(key="gone")
    session.add_message("user", "late", _synced=False)
    peer = SimpleNamespace(message=lambda content: content)
    mgr._get_or_create_peer = lambda peer_id: peer
    mgr._sessions_cache[session.honcho_session_id] = SimpleNamespace(add_messages=lambda messages: None)

    assert mgr._flush_session(session) is True
    assert "gone" not in mgr._cache
    assert all(m["_synced"] for m in session.messages)


def _sdk_session():
    return SimpleNamespace(add_messages=lambda messages: None)


def test_flush_that_recreates_the_sdk_session_stores_its_observation_flags():
    """After an eviction the flush path rebuilds the SDK session. The flags it configured must be kept."""
    mgr = _manager()
    session = _session(key="back")
    session.add_message("user", "hello", _synced=False)
    flags = {"user_observe_me": False, "user_observe_others": True, "ai_observe_me": True, "ai_observe_others": False}
    mgr._get_or_create_peer = lambda peer_id: SimpleNamespace(message=lambda content: content)
    mgr._get_or_create_honcho_session = lambda sid, user, assistant: (_sdk_session(), [], flags)

    assert mgr._flush_session(session) is True
    assert mgr._session_observation[session.honcho_session_id] == flags


def test_cached_sdk_session_returns_the_flags_stored_for_it():
    mgr = _manager()
    flags = {"user_observe_me": True, "user_observe_others": False, "ai_observe_me": True, "ai_observe_others": True}
    sdk = _sdk_session()
    mgr._sessions_cache["hs-x"] = sdk
    mgr._session_observation["hs-x"] = flags

    assert mgr._get_or_create_honcho_session("hs-x", None, None) == (sdk, [], flags)


@pytest.mark.parametrize("synced, kept", [(False, True), (True, False)])
def test_deferred_save_keeps_an_evicted_session_only_while_it_holds_unsynced_messages(synced, kept):
    """write_frequency "session" defers to flush_all(), which only sees cached sessions."""
    mgr = _manager()
    mgr._write_frequency = "session"
    session = _session(key="evicted")
    session.add_message("user", "pending", _synced=synced)

    mgr.save(session)

    assert (mgr._cache.get("evicted") is session) is kept


def test_deferred_save_flushes_inline_when_a_newer_object_owns_the_key():
    mgr = _manager()
    mgr._write_frequency = "session"
    newer = _session(key="k")
    mgr._cache["k"] = newer
    stale = _session(key="k")
    stale.add_message("user", "late", _synced=False)
    flushed = []
    mgr._flush_session = lambda s: flushed.append(s) or True

    mgr.save(stale)

    assert flushed == [stale]
    assert mgr._cache["k"] is newer


def _failing_then_recording_uploads(mgr, session):
    """The SDK session for ``session`` refuses the first batch; ``restore()`` swaps in one that records."""
    uploads = []

    def refuse(messages):
        raise ConnectionError("upload refused")

    mgr._get_or_create_peer = lambda peer_id: SimpleNamespace(message=lambda content: content)
    mgr._sessions_cache[session.honcho_session_id] = SimpleNamespace(add_messages=refuse)

    def restore():
        mgr._sessions_cache[session.honcho_session_id] = SimpleNamespace(add_messages=lambda ms: uploads.extend(ms))

    return uploads, restore


def test_failed_turn_save_after_an_eviction_is_retried_by_flush_all():
    """A clean session can be evicted while its caller still holds it. The caller's next save flushes inline in
    turn mode, and a failed upload used to leave that object nowhere flush_all() could find it."""
    mgr = _manager()
    session = _session(key="k")
    mgr._cache["k"] = session
    with mgr._cache_lock:
        mgr._evict_session_locked("k", session)
    session.add_message("user", "late")
    uploads, restore = _failing_then_recording_uploads(mgr, session)

    mgr.save(session)
    assert mgr._cache["k"] is session
    restore()
    mgr.flush_all()

    assert uploads == ["late"]
    assert session.messages[0]["_synced"] is True


def test_failed_collision_flush_waits_for_flush_all_without_displacing_the_newer_object():
    mgr = _manager()
    mgr._write_frequency = "session"
    newer = _session(key="k")
    mgr._cache["k"] = newer
    stale = _session(key="k")
    stale.add_message("user", "late")
    uploads, restore = _failing_then_recording_uploads(mgr, stale)

    mgr.save(stale)
    assert mgr._cache["k"] is newer
    assert mgr._retry_sessions == [stale]
    restore()
    mgr.flush_all()

    assert uploads == ["late"]
    assert mgr._retry_sessions == []
    assert mgr._cache["k"] is newer


def test_a_retained_session_is_listed_once_across_repeated_failures():
    mgr = _manager()
    mgr._cache["k"] = _session(key="k")
    stale = _session(key="k")
    stale.add_message("user", "late")
    _failing_then_recording_uploads(mgr, stale)

    mgr.save(stale)
    mgr.save(stale)

    assert mgr._retry_sessions == [stale]