diff --git a/docs/sessions.md b/docs/sessions.md index 6d81553..c0b3a68 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -75,8 +75,10 @@ - `get(session_id)` → `Session | None` - `save(session)` — serializes with `model_dump(mode='json')` (required for datetime serialization) - `list_all(user_id=None, is_admin=False)` → if `is_admin`: all sessions; else: sessions for `user_id` or legacy (`user_id IS NULL`) sessions +- `list_all_summaries(user_id=None, is_admin=False, special=None)` → the same rows and order as `list_all`, but history is not loaded: `display_message_count` and `preview` come from one aggregate query. List views use it, so `GET /sessions` costs ~10–40 ms instead of ~300 ms and never allocates the hot history. - `count_all(user_id=None, is_admin=False, search=None)` → total matching sessions -- `search_list(limit, offset, user_id=None, is_admin=False, search=None, sort_by="last_active", sort_order="desc")` → paginated, filtered, sorted sessions +- `search_list(limit, offset, user_id=None, is_admin=False, search=None, sort_by="last_active", sort_order="desc")` → paginated, filtered, sorted sessions. Like `list_page`, it returns **summaries**: callers that need "where does the term appear" ask `search_matches`. +- `search_matches(session_ids, search)` → `{session_id: (display indices, snippet)}`, one aggregate over the given sessions instead of their loaded histories. Indices are positions in the display history, the numbering clients scroll to. - `delete(session_id)` → `bool` - `list_page(user_id=None, is_admin=False, limit=50, offset=0, profile_id=None)` → paginated list with `has_more` flag; `profile_id` filters to one profile - `set_pinned(session_id, pinned)` → `bool` @@ -86,6 +88,14 @@ Requires `DATABASE_URL` env variable (e.g. `postgresql://user:pass@localhost/navi`). +**Search speed.** `search_list` matches `content ILIKE '%q%'` over `session_messages` and `session_messages_archive`, which without an index is a sequential scan of every message (~400 ms on a 29k-row dev clone, ~1 s on the 391 MB prod table). The store creates GIN trigram indexes on both `content` columns when `pg_trgm` is **already installed** — it does not create the extension itself, because that needs privileges the service user may not have. Install it once per database: + +```sql +CREATE EXTENSION IF NOT EXISTS pg_trgm; +``` + +The indexes are built on the next store init (`CREATE INDEX IF NOT EXISTS`, a few seconds on 150 MB) and session search drops to tens of milliseconds. + **Ownership:** Legacy sessions (`user_id IS NULL`) are accessible only to admins. New sessions created by authenticated users carry `user_id`. The `list_all()` method respects the `is_admin` flag to filter appropriately. When `NAVI_AUTH_ENABLED=false`, every session is created with `user_id = 'anonymous'` and all access checks are bypassed. Querying `sessions.user_id = 'anonymous'` is a reliable way to identify sessions created in no-auth mode. diff --git a/navi/api/routes/admin/sessions.py b/navi/api/routes/admin/sessions.py index a0d45a3..880be5f 100644 --- a/navi/api/routes/admin/sessions.py +++ b/navi/api/routes/admin/sessions.py @@ -43,7 +43,7 @@ "profile_id": s.profile_id, "user_id": s.user_id, "name": s.name, - "message_count": len(s.messages), + "message_count": s.display_count, "pinned": s.pinned, "created_at": s.created_at.isoformat(), "last_active": s.last_active.isoformat(), diff --git a/navi/api/routes/admin/users.py b/navi/api/routes/admin/users.py index d2bbf6f..68ab09e 100644 --- a/navi/api/routes/admin/users.py +++ b/navi/api/routes/admin/users.py @@ -70,14 +70,14 @@ user: Annotated[User, Depends(require_admin)], ): """Return sessions owned by a specific user.""" - sessions = await store.list_all(user_id=user.id, is_admin=True) + sessions = await store.list_all_summaries(user_id=user.id, is_admin=True) user_sessions = [s for s in sessions if s.user_id == user_id] return [ { "session_id": s.id, "profile_id": s.profile_id, "name": s.name, - "message_count": len(s.messages), + "message_count": s.display_count, "pinned": s.pinned, "created_at": s.created_at.isoformat(), "last_active": s.last_active.isoformat(), diff --git a/navi/api/routes/sessions/listing.py b/navi/api/routes/sessions/listing.py index 2bf91aa..84b9331 100644 --- a/navi/api/routes/sessions/listing.py +++ b/navi/api/routes/sessions/listing.py @@ -72,21 +72,28 @@ cutoff = datetime.now(timezone.utc) - timedelta(minutes=30) try: - sessions = await store.list_all(user_id=user_id, is_admin=False) + # Candidates first, histories later. Listing used to hydrate every + # session's messages on every POST /sessions just to decide which ones + # were idle; a summary says the same thing and costs nothing. + candidates = await store.list_all_summaries(user_id=user_id, is_admin=False) backend = get_backend_registry().get("ollama") except Exception: return - for session in sessions: - if not session.messages: + for candidate in candidates: + if not candidate.display_count: continue - if session.user_id is None: + if candidate.user_id is None: continue # skip legacy sessions — no multi-user memory for unowned sessions - if session.last_active >= cutoff: + if candidate.last_active >= cutoff: continue # still active - extracted_at = await memory.get_extracted_at(session.id) - if extracted_at and extracted_at >= session.last_active.isoformat(): + extracted_at = await memory.get_extracted_at(candidate.id) + if extracted_at and extracted_at >= candidate.last_active.isoformat(): continue # already up to date + # Only a session that actually needs extraction pays for its history. + session = await store.get(candidate.id) + if session is None: + continue try: await extract_and_update(session, backend, settings.ollama_default_model, memory) except Exception: @@ -111,11 +118,24 @@ is_admin = _session_admin(user) if limit is None and offset is None: - sessions = await store.list_all(user_id=_session_owner(user), is_admin=is_admin, special=special_filter) + # No paging means "the whole list" (the TUI picker asks this way). The + # sidebar only needs a count and a snippet, and both come from + # aggregates: hydrating every session's history to produce them cost + # ~300 ms a call and allocated the entire hot history for nothing. + sessions = await store.list_all_summaries( + user_id=_session_owner(user), is_admin=is_admin, special=special_filter + ) if profile_id and not search: sessions = [s for s in sessions if s.profile_id == profile_id] - pending_ids = await scheduler.get_pending_session_ids([s.id for s in sessions]) - return [_session_summary(s, search=search, has_pending_recall=s.id in pending_ids) for s in sessions] + ids = [s.id for s in sessions] + matches = await store.search_matches(ids, search) if search else {} + pending_ids = await scheduler.get_pending_session_ids(ids) + return [ + _session_summary( + s, match=matches.get(s.id, ([], "")), has_pending_recall=s.id in pending_ids + ) + for s in sessions + ] page_limit = limit or 30 page_offset = offset or 0 @@ -138,9 +158,17 @@ special=special_filter, ) items = sessions[:page_limit] + # Search results are summaries too; "where does the term appear" is one + # aggregate over the page's sessions, not a reason to load their history. + matches = await store.search_matches([s.id for s in items], search) if search else {} pending_ids = await scheduler.get_pending_session_ids([s.id for s in items]) return { - "items": [_session_summary(s, search=search, has_pending_recall=s.id in pending_ids) for s in items], + "items": [ + _session_summary( + s, match=matches.get(s.id, ([], "")), has_pending_recall=s.id in pending_ids + ) + for s in items + ], "limit": page_limit, "offset": page_offset, "has_more": len(sessions) > page_limit, @@ -148,12 +176,16 @@ } -def _session_summary(session, search: str | None = None, has_pending_recall: bool = False) -> dict: +def _session_summary( + session, + match: tuple[list[int], str] | None = None, + has_pending_recall: bool = False, +) -> dict: summary = { "session_id": session.id, "profile_id": session.profile_id, "name": session.name, - "message_count": session.display_message_count, + "message_count": session.display_count, "preview": session.preview, "pinned": session.pinned, "special": session.special, @@ -161,28 +193,6 @@ "created_at": session.created_at.isoformat(), "last_active": session.last_active.isoformat(), } - if search: - summary["match_indices"], summary["match_preview"] = _match_info(session, search) + if match is not None: + summary["match_indices"], summary["match_preview"] = match return summary - - -def _match_info(session, search: str) -> tuple[list[int], str]: - """Return message indices that contain the search term and a preview snippet.""" - q = search.lower() - indices = [] - preview = "" - for i, msg in enumerate(session.messages): - content = msg.content or "" - if q in content.lower(): - indices.append(i) - if not preview: - idx = content.lower().find(q) - start = max(0, idx - 30) - end = min(len(content), idx + len(search) + 30) - snippet = content[start:end] - if start > 0: - snippet = "…" + snippet - if end < len(content): - snippet = snippet + "…" - preview = snippet - return indices, preview diff --git a/navi/core/session.py b/navi/core/session.py index eacdd9d..5db86db 100644 --- a/navi/core/session.py +++ b/navi/core/session.py @@ -35,6 +35,56 @@ # in-memory baseline it compares against (never serialized). _saved_fingerprints: dict = PrivateAttr(default_factory=dict) + @property + def display_count(self) -> int: + """How many display messages the session has, cheaply when possible. + + List views fill ``display_message_count`` from an aggregate and leave + ``messages`` empty; a store that hands back a hydrated session fills + ``messages`` instead. Either answer is the same number. + """ + return self.display_message_count or len(self.messages) + + +def match_snippet(content: str, search: str) -> str: + """The sidebar snippet: ~30 chars either side of the first occurrence. + + Falls back to the head of the content when the term cannot be located by a + plain lowercase scan (SQL matching and Python differ on a few exotic + case-folding pairs, and a hit is still a hit). + """ + idx = content.lower().find(search.lower()) + if idx < 0: + return content[:60] + ("…" if len(content) > 60 else "") + start = max(0, idx - 30) + end = min(len(content), idx + len(search) + 30) + snippet = content[start:end] + if start > 0: + snippet = "…" + snippet + if end < len(content): + snippet = snippet + "…" + return snippet + + +def match_info_from_messages(messages: list[Message], search: str) -> tuple[list[int], str]: + """Message positions containing ``search``, plus a preview of the first hit. + + Indices are positions in the session's *display* history — the numbering + ``GET /sessions`` returns and clients scroll to. A store that can answer + from SQL (``search_matches``) produces the same two values without loading + messages. + """ + q = search.lower() + indices: list[int] = [] + preview = "" + for i, msg in enumerate(messages): + content = msg.content or "" + if q in content.lower(): + indices.append(i) + if not preview: + preview = match_snippet(content, search) + return indices, preview + class SessionStore(ABC): @abstractmethod @@ -67,6 +117,30 @@ special: bool | None = None, ) -> list[Session]: ... + async def list_all_summaries( + self, + user_id: str | None = None, + is_admin: bool = False, + special: bool | None = None, + ) -> list[Session]: + """``list_all`` for list views: counts and previews, no message payload. + + Optional because it is an optimization, not a capability: a store that + holds its sessions in memory answers with ``list_all`` and loses + nothing. A database-backed store overrides it and reads aggregates. + """ + return await self.list_all(user_id=user_id, is_admin=is_admin, special=special) + + async def search_matches( + self, session_ids: list[str], search: str + ) -> dict[str, tuple[list[int], str]]: + """``{session_id: (display indices, snippet)}`` for a search result page. + + Default: no match information, which clients treat as "cannot say + where" and fall back to showing the session without a jump target. + """ + return {} + @abstractmethod async def count_all( self, @@ -290,6 +364,16 @@ sessions = sorted(sessions, key=key, reverse=reverse) return sessions[offset:offset + limit] + async def search_matches( + self, session_ids: list[str], search: str + ) -> dict[str, tuple[list[int], str]]: + # Everything is already in memory here, so the scan is the whole job. + return { + sid: match_info_from_messages(self._sessions[sid].messages, search) + for sid in session_ids + if sid in self._sessions + } + async def delete(self, session_id: str) -> bool: if session_id in self._sessions: del self._sessions[session_id] diff --git a/navi/core/sessions/rows.py b/navi/core/sessions/rows.py index 106258c..45e2670 100644 --- a/navi/core/sessions/rows.py +++ b/navi/core/sessions/rows.py @@ -11,7 +11,7 @@ from navi.llm.base import Message, ToolCallRequest -from ..session import Session +from ..session import Session, match_snippet def _serialize(messages: list[Message]) -> str: @@ -162,6 +162,53 @@ return "" +async def _load_match_info( + conn: asyncpg.Connection, session_ids: list[str], search: str +) -> dict[str, tuple[list[int], str]]: + """Where the search term appears in each session, without loading history. + + ``idx`` reproduces the index the caller would get from the hydrated display + list: display rows only, in ``sequence_number`` order, above the session's + archive threshold. The preview comes from the first hit only, so nothing + but one content per session crosses the wire. + """ + if not session_ids or not search: + return {} + rows = await conn.fetch( + """ + WITH hot AS ( + SELECT m.session_id, + m.content, + row_number() OVER ( + PARTITION BY m.session_id ORDER BY m.sequence_number + ) - 1 AS idx + FROM session_messages m + JOIN sessions s ON s.id = m.session_id + WHERE m.session_id = ANY($1) + AND m.is_display + AND m.sequence_number >= coalesce(s.archive_threshold, 0) + ), matched AS ( + SELECT * FROM hot WHERE content ILIKE $2 + ) + SELECT agg.session_id, + agg.indices, + first.content AS first_content + FROM (SELECT session_id, array_agg(idx ORDER BY idx) AS indices + FROM matched + GROUP BY session_id) agg + JOIN matched first + ON first.session_id = agg.session_id + AND first.idx = agg.indices[1] + """, + session_ids, + f"%{search}%", + ) + return { + r["session_id"]: (list(r["indices"]), match_snippet(r["first_content"] or "", search)) + for r in rows + } + + async def _load_messages_map(conn: asyncpg.Connection, session_ids: list[str]) -> dict[str, list[Message]]: """Batch-load all messages for the given session IDs.""" if not session_ids: diff --git a/navi/core/sessions/store.py b/navi/core/sessions/store.py index 9c18941..212db3a 100644 --- a/navi/core/sessions/store.py +++ b/navi/core/sessions/store.py @@ -15,6 +15,7 @@ from .rows import ( _build_session_summaries, _build_sessions, + _load_match_info, _row_to_message, _special_and, _special_frag, @@ -599,33 +600,56 @@ async def list_all( self, user_id: str | None = None, is_admin: bool = False, special: bool | None = None ) -> list[Session]: - """Sessions visible to the caller. + """Sessions visible to the caller, with their hot history hydrated. Scoping: admins see everything; a named owner sees their own rows; ``user_id=None`` (local mode, the anonymous local user) sees only NULL-owner sessions — never other users' chats. ``special``: None — no filter, False — regular only, True — special only. + + Callers that only need name/count/preview should ask for + ``list_all_summaries`` instead — this one loads every message. """ pool = await self._get_pool() async with pool.acquire() as conn: - if is_admin: - rows = await conn.fetch( - "SELECT id, profile_id, user_id, pinned, special, created_at, last_active, context_token_count, name, planning_logs, next_sequence, archive_threshold, session_metadata " - f"FROM sessions{_special_where(special)} ORDER BY pinned DESC, last_active DESC" - ) - elif user_id is not None: - rows = await conn.fetch( - "SELECT id, profile_id, user_id, pinned, special, created_at, last_active, context_token_count, name, planning_logs, next_sequence, archive_threshold, session_metadata " - f"FROM sessions WHERE user_id = $1{_special_and(special)} ORDER BY pinned DESC, last_active DESC", - user_id, - ) - else: - rows = await conn.fetch( - "SELECT id, profile_id, user_id, pinned, special, created_at, last_active, context_token_count, name, planning_logs, next_sequence, archive_threshold, session_metadata " - f"FROM sessions WHERE user_id IS NULL{_special_and(special)} ORDER BY pinned DESC, last_active DESC" - ) + rows = await self._list_all_rows(conn, user_id=user_id, is_admin=is_admin, special=special) return await _build_sessions(conn, rows) + async def list_all_summaries( + self, user_id: str | None = None, is_admin: bool = False, special: bool | None = None + ) -> list[Session]: + """``list_all`` without the history: counts and previews come from SQL. + + Same rows, same order, same scoping — the difference is that no message + body is read or deserialized, which is what the sidebar needs. + """ + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await self._list_all_rows(conn, user_id=user_id, is_admin=is_admin, special=special) + return await _build_session_summaries(conn, rows) + + async def _list_all_rows( + self, conn: asyncpg.Connection, user_id: str | None, is_admin: bool, special: bool | None + ) -> list[asyncpg.Record]: + select = ( + "SELECT id, profile_id, user_id, pinned, special, created_at, last_active, " + "context_token_count, name, planning_logs, next_sequence, archive_threshold, session_metadata " + ) + if is_admin: + return await conn.fetch( + select + f"FROM sessions{_special_where(special)} ORDER BY pinned DESC, last_active DESC" + ) + if user_id is not None: + return await conn.fetch( + select + f"FROM sessions WHERE user_id = $1{_special_and(special)} " + "ORDER BY pinned DESC, last_active DESC", + user_id, + ) + return await conn.fetch( + select + f"FROM sessions WHERE user_id IS NULL{_special_and(special)} " + "ORDER BY pinned DESC, last_active DESC" + ) + async def list_page( self, *, @@ -823,7 +847,18 @@ f"FROM sessions {where} {order_clause}", *params, ) - return await _build_sessions(conn, rows) + # Search results are a list view like any other: the caller needs + # counts and previews, and `search_matches` answers "where" from + # SQL. Hydrating every matched history is what made this ~500 ms. + return await _build_session_summaries(conn, rows) + + async def search_matches( + self, session_ids: list[str], search: str + ) -> dict[str, tuple[list[int], str]]: + """``{session_id: (display indices, snippet)}`` for a search result page.""" + pool = await self._get_pool() + async with pool.acquire() as conn: + return await _load_match_info(conn, session_ids, search) async def pending_sweep_loop(store: "PgSessionStore", interval: int = 3600) -> None: diff --git a/tests/unit/core/test_session_list_summaries.py b/tests/unit/core/test_session_list_summaries.py new file mode 100644 index 0000000..217ebf9 --- /dev/null +++ b/tests/unit/core/test_session_list_summaries.py @@ -0,0 +1,269 @@ +"""Session lists are cheap: summaries from SQL, match info from one query. + +``GET /sessions`` used to hydrate every hot message of every listed session to +produce a count, a one-line preview and "where does the search term appear" — +313 ms and the whole hot history allocated for a sidebar. These tests pin the +cheap shape so it does not creep back: list views read aggregates only, and a +search page asks one extra question instead of loading histories. +""" + +from datetime import UTC, datetime, timedelta + +import pytest + +from navi.core.pg_session_store import PgSessionStore +from navi.core.session import Session, match_info_from_messages, match_snippet +from navi.llm.base import Message +from tests.conftest_factory import FakeConnection, FakePool + +SESSION_COLS = { + "id": None, # per-row + "profile_id": "navi_code", + "user_id": None, + "pinned": False, + "special": False, + "created_at": datetime(2026, 1, 1, tzinfo=UTC), + "last_active": datetime(2026, 1, 2, tzinfo=UTC), + "context_token_count": 0, + "name": None, + "planning_logs": None, + "next_sequence": 4, + "archive_threshold": 0, + "session_metadata": None, +} + + +def _session_row(sid: str, **overrides): + from tests.conftest_factory import FakeRecord + + return FakeRecord(**{**SESSION_COLS, "id": sid, **overrides}) + + +def _stats_row(sid: str, total: int, display: int, preview: str): + from tests.conftest_factory import FakeRecord + + return FakeRecord( + session_id=sid, total_count=total, display_count=display, preview=preview + ) + + +def _hydrating_queries(conn: FakeConnection) -> list[str]: + return [q for kind, q, _ in conn.calls if kind == "fetch" and "SELECT * FROM session_messages" in q] + + +# --- match info ------------------------------------------------------------- + + +def test_match_info_indexes_display_messages_and_previews_the_first_hit(): + messages = [ + Message(role="user", content="hello there"), + Message(role="assistant", content="we talked about NGINX earlier"), + Message(role="user", content="nginx again, and nginx twice"), + ] + + indices, preview = match_info_from_messages(messages, "nginx") + + assert indices == [1, 2] + assert preview.startswith("we talked about NGINX") + assert preview.endswith("earlier") + + +def test_match_snippet_marks_trimmed_edges(): + content = "x" * 40 + " needle " + "y" * 40 + + snippet = match_snippet(content, "needle") + + assert snippet.startswith("…") + assert snippet.endswith("…") + assert "needle" in snippet + + +def test_match_snippet_falls_back_to_the_head_when_the_term_is_not_findable(): + assert match_snippet("plain text", "zzz") == "plain text" + + +def test_no_hit_returns_no_indices_and_no_preview(): + assert match_info_from_messages([Message(role="user", content="abc")], "zzz") == ([], "") + + +# --- display_count ---------------------------------------------------------- + + +def test_display_count_prefers_the_aggregate(): + session = Session(profile_id="p") + session.display_message_count = 7 + session.messages = [Message(role="user", content="a")] + + assert session.display_count == 7 + + +def test_display_count_falls_back_to_loaded_messages(): + session = Session(profile_id="p") + session.messages = [Message(role="user", content="a"), Message(role="user", content="b")] + + assert session.display_count == 2 + + +# --- the store -------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_all_summaries_reads_aggregates_not_histories(): + conn = FakeConnection() + conn.enqueue([_session_row("s1"), _session_row("s2")]) + conn.enqueue([_stats_row("s1", 5, 4, "last line"), _stats_row("s2", 2, 2, "other")]) + store = PgSessionStore(FakePool(conn)) + store._initialized = True + + sessions = await store.list_all_summaries(is_admin=True) + + assert [s.id for s in sessions] == ["s1", "s2"] + assert sessions[0].display_message_count == 4 + assert sessions[0].preview == "last line" + assert sessions[0].messages == [] + assert _hydrating_queries(conn) == [] + + +@pytest.mark.asyncio +async def test_search_list_returns_summaries_without_history(): + conn = FakeConnection() + conn.enqueue([_session_row("s1")]) + conn.enqueue([_stats_row("s1", 9, 9, "match here")]) + store = PgSessionStore(FakePool(conn)) + store._initialized = True + + sessions = await store.search_list(limit=10, offset=0, is_admin=True, search="match") + + assert len(sessions) == 1 + assert sessions[0].display_count == 9 + assert sessions[0].messages == [] + assert _hydrating_queries(conn) == [] + + +@pytest.mark.asyncio +async def test_search_matches_maps_rows_to_indices_and_snippet(): + conn = FakeConnection() + conn.enqueue([{"session_id": "s1", "indices": [1, 4], "first_content": "a needle here"}]) + store = PgSessionStore(FakePool(conn)) + store._initialized = True + + matches = await store.search_matches(["s1"], "needle") + + assert matches["s1"][0] == [1, 4] + assert "needle" in matches["s1"][1] + + +@pytest.mark.asyncio +async def test_search_matches_skips_the_query_when_there_is_nothing_to_do(): + conn = FakeConnection() + store = PgSessionStore(FakePool(conn)) + store._initialized = True + + assert await store.search_matches([], "needle") == {} + assert await store.search_matches(["s1"], "") == {} + assert conn.calls == [] + + +# --- the route -------------------------------------------------------------- + + +def test_session_summary_uses_the_cheap_count_and_carries_match_info(): + from navi.api.routes.sessions.listing import _session_summary + + session = Session(profile_id="navi_code") + session.display_message_count = 12 + + summary = _session_summary(session, match=([3], "…needle…")) + + assert summary["message_count"] == 12 + assert summary["match_indices"] == [3] + assert summary["match_preview"] == "…needle…" + + +def test_session_summary_omits_match_keys_without_a_search(): + from navi.api.routes.sessions.listing import _session_summary + + summary = _session_summary(Session(profile_id="navi_code")) + + assert "match_indices" not in summary + assert "match_preview" not in summary + + +class _StaleStore: + """Store double: summaries for the scan, full loads only when asked.""" + + def __init__(self, sessions): + self._sessions = {s.id: s for s in sessions} + self.gets: list[str] = [] + + async def list_all_summaries(self, user_id=None, is_admin=False, special=None): + listed = [] + for s in self._sessions.values(): + summary = Session(profile_id=s.profile_id, user_id=s.user_id, id=s.id) + summary.last_active = s.last_active + summary.display_message_count = len(s.messages) + listed.append(summary) + return listed + + async def get(self, session_id): + self.gets.append(session_id) + return self._sessions.get(session_id) + + +class _ExtractedMemory: + def __init__(self, already: set[str]): + self._already = already + self.calls: list[str] = [] + + async def get_extracted_at(self, session_id): + self.calls.append(session_id) + if session_id in self._already: + return datetime.now(UTC).isoformat() + return None + + +@pytest.mark.asyncio +async def test_stale_sweep_loads_history_only_for_sessions_that_need_extraction(monkeypatch): + from navi.api import deps + from navi.api.routes.sessions import listing + from navi.memory import extractor + + now = datetime.now(UTC) + stale = Session(profile_id="p", user_id="u1") + stale.messages = [Message(role="user", content="remember this")] + stale.last_active = now - timedelta(hours=2) + + fresh = Session(profile_id="p", user_id="u1") + fresh.messages = [Message(role="user", content="still talking")] + fresh.last_active = now + + done = Session(profile_id="p", user_id="u1") + done.messages = [Message(role="user", content="already extracted")] + done.last_active = now - timedelta(hours=3) + + unowned = Session(profile_id="p", user_id=None) + unowned.messages = [Message(role="user", content="legacy")] + unowned.last_active = now - timedelta(hours=4) + + empty = Session(profile_id="p", user_id="u1") + empty.last_active = now - timedelta(hours=5) + + store = _StaleStore([stale, fresh, done, unowned, empty]) + memory = _ExtractedMemory(already={done.id}) + extracted: list[str] = [] + + async def fake_extract(session, backend, model, mem): + extracted.append(session.id) + + class _Registry: + def get(self, name): + return object() + + monkeypatch.setattr(extractor, "extract_and_update", fake_extract) + monkeypatch.setattr(deps, "get_backend_registry", lambda: _Registry()) + + await listing._process_stale_sessions(store, memory, user_id="u1") + + assert extracted == [stale.id] + # Only the candidate that actually needed extraction paid for its history. + assert store.gets == [stale.id]