diff --git a/navi/core/compressor.py b/navi/core/compressor.py index 1b4dee5..b6f55ea 100644 --- a/navi/core/compressor.py +++ b/navi/core/compressor.py @@ -107,6 +107,12 @@ return max(0, score) +def _order_like(reference: list[Message], subset: list[Message]) -> list[Message]: + """Reorder `subset` into the order its members appear in `reference`.""" + wanted = {id(m) for m in subset} + return [m for m in reference if id(m) in wanted] + + def partition_messages( messages: list[Message], keep_recent: int, @@ -121,26 +127,43 @@ Tool call groups (assistant + tool results) always stay together. When adaptive=True, important turns (user corrections, errors, critical tools) are kept longer and social/filler turns are compressed sooner. + + Summary messages (is_summary=True) are not part of any turn: they carry + already-compressed history, not a new user request. They always go into + to_summarize so the fresh summary replaces them (the single-summary + invariant) — an old summary can never act as a turn's user-request head + (which the midturn split would keep verbatim beside the fresh summary) or + be rescued by the adaptive swap. This is what used to make repeated + midturn compressions pile stale summary copies into the context. """ non_system = [m for m in messages if m.role != "system"] + summary_msgs = [m for m in non_system if m.is_summary] + convo = [m for m in non_system if not m.is_summary] - # Group into turns: each turn starts with a user message + # Group into turns: each turn starts with a user message. A non-user + # message with no open turn (can only follow a stripped summary) trails + # the user request that preceded it, so it attaches to the previous turn + # instead of forming an orphan turn without a head. turns: list[list[Message]] = [] - current: list[Message] = [] - for msg in non_system: - if msg.role == "user" and current: - turns.append(current) - current = [msg] + for msg in convo: + if msg.role == "user" or not turns: + turns.append([msg]) else: - current.append(msg) - if current: - turns.append(current) + turns[-1].append(msg) if len(turns) <= keep_recent: if keep_recent_messages is not None: intra_turn = partition_current_turn_messages(turns, keep_recent_messages) if intra_turn is not None: - return intra_turn + old, to_keep = intra_turn + # A compression pass replaces every old summary, never keeps + # one beside the fresh one (single-summary invariant). + return _order_like(non_system, old + summary_msgs), to_keep + if summary_msgs: + # Even when nothing else has aged out, summaries are never kept: + # two or more of them must consolidate (the meta-summary path in + # compress_context) instead of sitting in the context together. + return _order_like(non_system, summary_msgs), convo return [], non_system # nothing old enough to compress # Adaptive: pull important older turns into the kept region and push @@ -192,6 +215,10 @@ to_keep = [m for turn in recent_turns for m in turn] else: to_keep = [m for turn in recent_turns for m in turn] + + # Fold every old summary into this pass; to_summarize is emitted in + # chronological order so the summarizer sees history in sequence. + to_summarize = _order_like(non_system, to_summarize + summary_msgs) return to_summarize, to_keep @@ -838,8 +865,13 @@ # Mark messages in session.messages as not context if they are no longer in new_context # and are not system messages. + # Iterate session.context too: after a session reload, context-only rows + # (old summaries — is_display=False) are present in session.context but + # NOT in session.messages, so demoting only the latter left stale + # summaries with is_context=True in the DB — they piled up in the + # context at every reload (the half-the-context-is-summaries symptom). new_context_ids = {id(m) for m in new_context} - for msg in session.messages: + for msg in [*session.messages, *session.context]: if msg.role != "system" and id(msg) not in new_context_ids: msg.is_context = False diff --git a/tests/unit/core/test_compressor.py b/tests/unit/core/test_compressor.py index 04e5a6d..f09b217 100644 --- a/tests/unit/core/test_compressor.py +++ b/tests/unit/core/test_compressor.py @@ -129,6 +129,114 @@ # Nothing is dropped: old + recent partition the whole non-system set. assert len(old) + len(recent) == len(msgs) + def test_summary_in_inflight_turn_is_folded_not_kept(self): + """Regression: a midturn compression inserts its summary right before + the kept head of the in-flight turn. The summary's role=user made it + the head of that turn, so the next midturn pass kept the old summary + verbatim beside the fresh one — a new stale copy at every compression + (six summary copies built up to half the context in one session).""" + msgs = [] + for i in range(14): + msgs.append(Message(role="user", content=f"u{i}")) + msgs.append(Message(role="assistant", content=f"a{i}")) + msgs.append(Message(role="user", content="old summary text", is_summary=True)) + msgs.append(Message(role="user", content="inflight request")) + for i in range(10): + msgs.append(Message(role="assistant", content=f"step {i}")) + msgs.append(Message(role="tool", content=f"res {i}", name="fs", tool_call_id=str(i))) + + old, recent = partition_messages(msgs, keep_recent=12, keep_recent_messages=4) + + old_contents = {m.content for m in old} + recent_contents = {m.content for m in recent} + assert "old summary text" in old_contents + assert "old summary text" not in recent_contents + # The real user request stays as the verbatim head, not the summary. + assert "inflight request" in recent_contents + assert len(old) + len(recent) == len(msgs) + + def test_summary_in_recent_window_is_folded(self): + """An existing summary inside the turn-based keep window still goes to + to_summarize — the new summary must replace it (single-summary invariant), + never sit beside it.""" + msgs = [ + Message(role="user", content="u0"), + Message(role="assistant", content="a0"), + Message(role="user", content="old summary text", is_summary=True), + ] + for i in range(1, 7): + msgs.append(Message(role="user", content=f"u{i}")) + msgs.append(Message(role="assistant", content=f"a{i}")) + + old, recent = partition_messages(msgs, keep_recent=6) + + old_contents = {m.content for m in old} + recent_contents = {m.content for m in recent} + assert "old summary text" in old_contents + assert "old summary text" not in recent_contents + assert "u1" in recent_contents + + def test_adaptive_swap_does_not_rescue_summary(self): + """Adaptive swap must not pull a summary-carrying turn back into the + keep window: the summary text pattern-matches importance heuristics + ("wrong", "fix"), which used to resurrect stale summaries every pass.""" + summary_text = "summary: the first approach was wrong, fix the parser" + msgs = [ + Message(role="user", content="setup the project"), + Message(role="assistant", content="ok"), + Message(role="user", content=summary_text, is_summary=True), + ] + for i in range(6): + msgs.append(Message(role="user", content=f"task {i}")) + msgs.append(Message(role="assistant", content=f"answer {i}")) + + old, recent = partition_messages(msgs, keep_recent=2) # adaptive=True + + old_contents = {m.content for m in old} + recent_contents = {m.content for m in recent} + assert summary_text in old_contents + assert summary_text not in recent_contents + assert len(old) + len(recent) == len(msgs) + + def test_summary_stays_chronological_in_to_summarize(self): + """Summaries folded into to_summarize keep their position between the + surrounding turns (the summarizer reads history in sequence).""" + msgs = [ + Message(role="user", content="u0"), + Message(role="assistant", content="a0"), + ] + msgs.append(Message(role="user", content="old summary text", is_summary=True)) + for i in range(2, 7): + msgs.append(Message(role="user", content=f"u{i}")) + msgs.append(Message(role="assistant", content=f"a{i}")) + + old, _ = partition_messages(msgs, keep_recent=2) + + idx = [m.content for m in old].index("old summary text") + assert [m.content for m in old] == [ + "u0", "a0", "old summary text", "u2", "a2", "u3", "a3", "u4", "a4", + ] + + def test_midturn_fallback_folds_summary_in_short_history(self): + """Early-exit path (turns <= keep_recent) with midturn mode: a summary + inside the in-flight turn is folded, and the kept head is the real + user request.""" + msgs = [ + Message(role="user", content="old summary text", is_summary=True), + Message(role="user", content="long request"), + ] + for i in range(6): + msgs.append(Message(role="assistant", content=f"step {i}")) + msgs.append(Message(role="tool", content=f"res {i}", name="fs", tool_call_id=str(i))) + + old, recent = partition_messages(msgs, keep_recent=8, keep_recent_messages=4) + + old_contents = {m.content for m in old} + assert "old summary text" in old_contents + assert recent[0].role == "user" + assert recent[0].content == "long request" + assert len(old) + len(recent) == len(msgs) + class TestFormatForSummary: def test_user_message(self): diff --git a/tests/unit/workers/test_compression_worker.py b/tests/unit/workers/test_compression_worker.py index b100123..ed3ad10 100644 --- a/tests/unit/workers/test_compression_worker.py +++ b/tests/unit/workers/test_compression_worker.py @@ -134,6 +134,41 @@ assert all(m.is_context is False for m in dropped) +@pytest.mark.asyncio +async def test_worker_demotes_context_only_summaries_after_reload(): + """After a session reload, old summaries (is_display=False) exist in + session.context but NOT in session.messages. The demotion loop must cover + both lists — otherwise stale summaries keep is_context=True in the DB and + pile up in the context (they once built up to half of one session's + context: 6 copies, 120k of 239k chars).""" + worker = CompressionWorker() + backend = FakeLLMBackend(responses=["fresh summary"]) + store = InMemorySessionStore() + session = Session(profile_id="test") + store._sessions[session.id] = session + + old_summary = Message( + role="user", content="old summary text", is_summary=True, + is_display=False, is_context=True, + ) + context = [old_summary] + for i in range(10): + context.append(Message(role="user", content=f"task {i}")) + context.append(Message(role="assistant", content=f"answer {i}")) + session.context = list(context) + # Reload state: display history has no context-only rows. + session.messages = [m for m in context if not m.is_summary] + + result = await worker.run(session, _ctx(session.id, backend, store)) + assert any(isinstance(ev, ContextCompressed) for ev in result.events) + # The stale summary is demoted even though it is absent from session.messages. + assert old_summary.is_context is False + kept_summaries = [m for m in session.context if m.is_summary] + assert len(kept_summaries) == 1 + assert kept_summaries[0] is not old_summary + assert "fresh summary" in kept_summaries[0].content + + def test_worker_keep_recent_messages_mirrors_midturn(): """The worker passes keep_recent_messages=max(12, context_keep_recent*2), matching the midturn auto-compress path (source of truth for the intra-turn