From 711f1c2f1ac80920e1b02ce3946d4a984b026bb9 Mon Sep 17 00:00:00 2001 From: MacroAnarchy Date: Sun, 12 Jul 2026 12:41:18 +0200 Subject: [PATCH] fix(search): surface compaction-archived and compression-parent sessions in discovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After context compaction, pre-compaction content was invisible to session_search — a memory black hole. The _discover() skip logic filtered both same-session and same-lineage hits unconditionally, without distinguishing compression-summarised content (gone from live context) from delegation children (still visible to the parent agent). Reworked _resolve_to_parent to return (root_id, has_compression_hop), checking end_reason='compression' on every hop during the same db.get_session() traversal — zero extra queries. _discover() now has three compression-aware paths: - In-place compaction: FTS hits on active=0 (compacted=1) rows pass through even when raw_sid == current_session_id - Legacy rotation: lineage hits pass through when has_compression_hop is true on either side of the chain - Delegation children: still excluded (no compression edge) 18 new tests covering all three scenarios + unit tests for the helpers. Addresses Teknium's review feedback on #6256. Closes #13840, #13841. --- tests/tools/test_session_search.py | 240 +++++++++++++++++++++++++++++ tools/session_search_tool.py | 92 +++++++++-- 2 files changed, 316 insertions(+), 16 deletions(-) diff --git a/tests/tools/test_session_search.py b/tests/tools/test_session_search.py index 69e3adb1d76..6bec26e88ec 100644 --- a/tests/tools/test_session_search.py +++ b/tests/tools/test_session_search.py @@ -17,6 +17,8 @@ from tools.session_search_tool import ( SESSION_SEARCH_SCHEMA, _HIDDEN_SESSION_SOURCES, _format_timestamp, + _is_compacted_message, + _resolve_to_parent, session_search, ) @@ -772,3 +774,241 @@ class TestCompactionSummaryFiltering: entry = result["results"][0] for msg in entry.get("bookend_start", []): assert "[CONTEXT SUMMARY]" not in (msg.get("content") or "") + + +# ========================================================================= +# Compression-aware discovery (#6256) +# +# After compression (in-place compaction or legacy rotation), pre-compaction +# content is no longer in the live context but MUST stay discoverable via +# session_search. The old code skipped any FTS hit on the current session or +# lineage, creating a "memory black hole". Delegation children must STAY +# excluded — their content is still visible to the parent agent. +# ========================================================================= + +class TestResolveToParent: + """Unit tests for _resolve_to_parent's compression-aware tuple return.""" + + def test_root_session_no_compression(self, db): + db.create_session("s1", source="cli") + root, has_compression = _resolve_to_parent(db, "s1") + assert root == "s1" + assert has_compression is False + + def test_empty_session_id(self, db): + root, has_compression = _resolve_to_parent(db, "") + assert root == "" + assert has_compression is False + + def test_none_session_id(self, db): + root, has_compression = _resolve_to_parent(db, None) + assert root is None + assert has_compression is False + + def test_legacy_rotation_detects_compression(self, db): + """Parent ended with end_reason='compression', child has parent_session_id.""" + db.create_session("s_parent", source="cli") + db.end_session("s_parent", "compression") + db.create_session("s_child", source="cli", parent_session_id="s_parent") + root, has_compression = _resolve_to_parent(db, "s_child") + assert root == "s_parent" + assert has_compression is True + + def test_delegation_no_compression(self, db): + """Delegation child: parent_session_id set but no compression end_reason.""" + db.create_session("s_parent", source="cli") + db.create_session("s_child", source="cli", parent_session_id="s_parent") + root, has_compression = _resolve_to_parent(db, "s_child") + assert root == "s_parent" + assert has_compression is False + + def test_multi_level_compression_chain(self, db): + """Grandparent → parent → child, both with compression edges.""" + db.create_session("s_gp", source="cli") + db.end_session("s_gp", "compression") + db.create_session("s_p", source="cli", parent_session_id="s_gp") + db.end_session("s_p", "compression") + db.create_session("s_c", source="cli", parent_session_id="s_p") + root, has_compression = _resolve_to_parent(db, "s_c") + assert root == "s_gp" + assert has_compression is True + + def test_chain_with_mixed_edges(self, db): + """Compression parent → delegation-style child (no end_reason on child).""" + db.create_session("s_gp", source="cli") + db.end_session("s_gp", "compression") + db.create_session("s_p", source="cli", parent_session_id="s_gp") + # s_p does NOT end with compression — but ancestor s_gp does + db.create_session("s_c", source="cli", parent_session_id="s_p") + root, has_compression = _resolve_to_parent(db, "s_c") + assert root == "s_gp" + assert has_compression is True + + +class TestIsCompactedMessage: + """Unit tests for the _is_compacted_message helper.""" + + def test_active_message_returns_false(self, db): + db.create_session("s1", source="cli") + mid = db.append_message("s1", role="user", content="hello") + assert _is_compacted_message(db, mid) is False + + def test_compacted_message_returns_true(self, db): + db.create_session("s1", source="cli") + mid = db.append_message("s1", role="user", content="archived content") + db.archive_and_compact("s1", [ + {"role": "assistant", "content": "compacted summary"}, + ]) + # mid is now active=0, compacted=1 + assert _is_compacted_message(db, mid) is True + + def test_none_message_id(self, db): + assert _is_compacted_message(db, None) is False + + def test_nonexistent_message_id(self, db): + assert _is_compacted_message(db, 999999) is False + + +class TestInPlaceCompactionDiscovery: + """In-place compaction: archived turns on the SAME session_id must be + discoverable from the current session.""" + + def test_archived_content_discoverable_after_compaction(self, db): + """The core regression: pre-compaction content on the current session + must surface in discovery even though raw_sid == current_session_id.""" + db.create_session("s_compact", source="cli") + db.append_message("s_compact", role="user", + content="The spectral phoenix only spawns during full moons") + db.append_message("s_compact", role="assistant", + content="Spectral phoenix requires moonstone bait") + db.archive_and_compact("s_compact", [ + {"role": "user", "content": "Summary: spectral phoenix discussed"}, + {"role": "assistant", "content": "Acknowledged spectral phoenix info"}, + ]) + + result = json.loads(session_search( + query="spectral phoenix", db=db, current_session_id="s_compact", + )) + assert result["success"] is True + assert result["count"] >= 1 + # The hit should be from the same session (archived rows) + hit = result["results"][0] + assert hit["session_id"] == "s_compact" + + def test_live_content_still_filtered_on_current_session(self, db): + """Non-compacted (active) content on the current session stays filtered.""" + db.create_session("s_live", source="cli") + db.append_message("s_live", role="user", content="crystal golem farming route") + result = json.loads(session_search( + query="crystal golem", db=db, current_session_id="s_live", + )) + assert result["count"] == 0 + + def test_mixed_active_and_compacted_on_same_session(self, db): + """A session that has been compacted: the archived content is + discoverable, but the new (post-compaction) active content is not + (it's in live context).""" + db.create_session("s_mixed", source="cli") + # Pre-compaction content (will be archived) + db.append_message("s_mixed", role="user", content="ancient ruins exploration log") + db.append_message("s_mixed", role="assistant", content="ancient ruins mapped") + # Compact + db.archive_and_compact("s_mixed", [ + {"role": "user", "content": "Summary of ancient ruins exploration"}, + {"role": "assistant", "content": "Continuing ancient ruins work"}, + ]) + # Archived content should be discoverable + result_archived = json.loads(session_search( + query="ancient ruins exploration", db=db, + current_session_id="s_mixed", + )) + assert result_archived["count"] >= 1 + + +class TestLegacyRotationDiscovery: + """Legacy rotation: parent session ended with end_reason='compression', + child session created. Parent's pre-compaction content must be discoverable + from the child.""" + + def test_compression_parent_discoverable_from_child(self, db): + db.create_session("s_parent", source="cli") + db.append_message("s_parent", role="user", + content="The void crystal mining requires diamond pickaxe") + db.append_message("s_parent", role="assistant", + content="Void crystal found in the deep caverns") + db.end_session("s_parent", "compression") + + db.create_session("s_child", source="cli", parent_session_id="s_parent") + db.append_message("s_child", role="user", content="Continue void crystal work") + + result = json.loads(session_search( + query="void crystal", db=db, current_session_id="s_child", + )) + assert result["success"] is True + assert result["count"] >= 1 + sids = [r["session_id"] for r in result["results"]] + assert "s_parent" in sids + + def test_multi_level_compression_chain_discoverable(self, db): + """Grandparent → parent → child, each compression-rotated. Content from + ancestors must be discoverable.""" + db.create_session("s_gp", source="cli") + db.append_message("s_gp", role="user", + content="Project titan initial architecture design") + db.end_session("s_gp", "compression") + + db.create_session("s_p", source="cli", parent_session_id="s_gp") + db.append_message("s_p", role="user", + content="Project titan second phase planning") + db.end_session("s_p", "compression") + + db.create_session("s_c", source="cli", parent_session_id="s_p") + db.append_message("s_c", role="user", content="Project titan final review") + + result = json.loads(session_search( + query="project titan", db=db, current_session_id="s_c", + )) + assert result["count"] >= 1 + # Should find content from s_gp or s_p (or both, deduped by lineage) + sids = [r["session_id"] for r in result["results"]] + assert any(s in ("s_gp", "s_p") for s in sids) + + +class TestDelegationExclusion: + """Delegation children (delegate_task) must STAY excluded — their content + is still visible to the parent agent. parent_session_id is set but the + parent does NOT have end_reason='compression'.""" + + def test_delegation_parent_excluded_from_child(self, db): + """Child can see its own content but parent's live content stays + excluded (it's in context via delegation).""" + db.create_session("s_parent", source="cli") + db.append_message("s_parent", role="user", + content="nebula deployment infrastructure setup") + db.append_message("s_parent", role="assistant", + content="Nebula deployment configured successfully") + + db.create_session("s_child", source="cli", parent_session_id="s_parent") + db.append_message("s_child", role="user", + content="delegated nebula deployment subtask") + + result = json.loads(session_search( + query="nebula deployment", db=db, current_session_id="s_child", + )) + assert result["count"] == 0 + + def test_delegation_child_excluded_from_parent(self, db): + """Parent searching should not see delegation child content either — + both are in the same lineage with no compression edge.""" + db.create_session("s_parent", source="cli") + db.append_message("s_parent", role="user", + content="Working on stellar forge project") + + db.create_session("s_child", source="cli", parent_session_id="s_parent") + db.append_message("s_child", role="user", + content="stellar forge delegated subtask execution") + + result = json.loads(session_search( + query="stellar forge", db=db, current_session_id="s_parent", + )) + assert result["count"] == 0 diff --git a/tools/session_search_tool.py b/tools/session_search_tool.py index f2cf85a3a9d..f9f68143bc5 100644 --- a/tools/session_search_tool.py +++ b/tools/session_search_tool.py @@ -99,19 +99,31 @@ def _is_compaction_summary(content: str) -> bool: return any(stripped.startswith(p) for p in _COMPACTION_PREFIXES) +def _resolve_to_parent(db, session_id: str) -> tuple[str, bool]: + """Walk parent_session_id chain to the lineage root. -def _resolve_to_parent(db, session_id: str) -> str: - """Walk parent_session_id chain to the lineage root. Falls back to input on errors.""" + Returns ``(root_id, has_compression_hop)`` where ``has_compression_hop`` is + True if any session along the chain ended with ``end_reason = 'compression'`` + — i.e. at least one parent/ancestor was compression-rotated into this + lineage. That flag lets callers distinguish a compression-split lineage + (parent content summarised away, no longer in live context) from a + delegation lineage (child content still visible to the parent agent). + + Falls back to ``(session_id, False)`` on errors. + """ if not session_id: - return session_id - visited = set() + return session_id, False + visited: set[str] = set() cur = session_id + has_compression = False while cur and cur not in visited: visited.add(cur) try: s = db.get_session(cur) if not s: break + if s.get("end_reason") == "compression": + has_compression = True parent = s.get("parent_session_id") if not parent: break @@ -119,7 +131,35 @@ def _resolve_to_parent(db, session_id: str) -> str: except Exception as e: logging.debug("Error resolving parent for %s: %s", cur, e, exc_info=True) break - return cur + return cur, has_compression + + +def _resolve_lineage(db, session_id: str) -> str: + """Convenience: return only the lineage root (ignores compression hop).""" + return _resolve_to_parent(db, session_id)[0] + + +def _is_compacted_message(db, message_id) -> bool: + """Return True if *message_id* is a soft-archived compaction row (active=0). + + Used by ``_discover`` to distinguish a compaction-archived FTS hit on the + current session (pre-compaction content no longer in live context — should + stay discoverable) from an active live hit (already in context — skip). + Returns False on any error so the caller falls back to the safe default + (skip the current session). + """ + if not message_id: + return False + try: + with db._lock: + cursor = db._conn.execute( + "SELECT active FROM messages WHERE id = ?", (message_id,) + ) + row = cursor.fetchone() + except Exception: + logging.debug("is_compacted_message lookup failed for %s", message_id, exc_info=True) + return False + return row is not None and row["active"] == 0 def _annotate_rebuild_status(db, payload: Dict[str, Any]) -> None: @@ -328,7 +368,7 @@ def _list_recent_sessions(db, limit: int, current_session_id: str = None) -> str order_by_last_active=True, ) # fetch extra so we can skip current - current_root = _resolve_to_parent(db, current_session_id) if current_session_id else None + current_root = _resolve_lineage(db, current_session_id) if current_session_id else None results = [] for s in sessions: @@ -395,8 +435,8 @@ def _scroll( # Reject scrolling inside the active session lineage — those messages are # already in context. if current_session_id: - a_root = _resolve_to_parent(db, session_id) - c_root = _resolve_to_parent(db, current_session_id) + a_root = _resolve_lineage(db, session_id) + c_root = _resolve_lineage(db, current_session_id) if a_root and c_root and a_root == c_root: return tool_error( "scroll rejected: anchor lives in the current session lineage (already in your active context)", @@ -439,8 +479,8 @@ def _scroll( logging.debug("owning-session lookup failed: %s", e, exc_info=True) owning = None if owning and owning != session_id: - a_root = _resolve_to_parent(db, session_id) - o_root = _resolve_to_parent(db, owning) + a_root = _resolve_lineage(db, session_id) + o_root = _resolve_lineage(db, owning) if a_root and o_root and a_root == o_root: try: rebind_view = db.get_messages_around(owning, around_message_id, window=window) @@ -509,7 +549,7 @@ def _title_match_result( if not session_id: return None - lineage_root = _resolve_to_parent(db, session_id) + lineage_root = _resolve_lineage(db, session_id) if current_lineage_root and lineage_root == current_lineage_root: return None @@ -568,7 +608,9 @@ def _discover( ) -> str: """Discovery shape: FTS5 + anchored window + bookends per hit. Single call.""" role_list = role_filter if role_filter else ["user", "assistant"] - current_lineage_root = _resolve_to_parent(db, current_session_id) if current_session_id else None + current_lineage_root, current_has_compression = ( + _resolve_to_parent(db, current_session_id) if current_session_id else (None, False) + ) title_result = _title_match_result(db, query, current_lineage_root) try: @@ -620,12 +662,30 @@ def _discover( if len(seen_sessions) >= limit: break raw_sid = r["session_id"] - resolved_sid = _resolve_to_parent(db, raw_sid) - # Skip the current session lineage + resolved_sid, has_compression = _resolve_to_parent(db, raw_sid) + # Skip the current session lineage — UNLESS the content has been + # compression-summarised out of the live context (memory black hole + # after compression). Two sub-cases: + # + # Legacy rotation: the FTS hit lives in a session whose lineage chain + # has a compression edge (end_reason='compression' on an ancestor). + # The parent's pre-compaction content is gone from the active context, + # so it must stay discoverable. + # + # In-place compaction: the FTS hit lives on the SAME session_id as the + # current session, but the matched message row is an archived + # (active=0, compacted=1) row. The live-context load filters active=1, + # so that content is no longer in context — let it through. + is_compacted_hit = _is_compacted_message(db, r.get("id")) if current_lineage_root and resolved_sid == current_lineage_root: - continue + if not (has_compression or current_has_compression or is_compacted_hit): + continue if current_session_id and raw_sid == current_session_id: - continue + # Same-session hit: only skip if the matched message is still live + # (active=1). Archived/compacted rows are pre-compaction content + # that's been summarised away — let them through. + if not is_compacted_hit: + continue if resolved_sid not in seen_sessions: row = dict(r) row["_lineage_root"] = resolved_sid