Collapse byte-identical turn hits at recall time

CHANGELOG.md

@@ -6,6 +6,18 @@ All notable changes to this project are documented here. The format follows
 
 ## [Unreleased]
 
+## [0.2.4] - 2026-09-03
+
+### Fixed
+- Recall now collapses turns whose quoted text is **byte-identical** to a higher-ranked hit, so a
+  scheduled-automation prompt captured on many days takes one result slot instead of several. The
+  static skip-prefix list (a schema-v2 sweep) only catches harness-injected boilerplate; a user's
+  own repeated cron/system prompts are real turns it can never match. Collapse is a recall-time
+  view — every turn stays in the store, a session still reads back whole, and distinct findings
+  inside those sessions still surface. Deterministic, model-free; opt out with
+  `collapse_duplicates=False`.
+
+
 ## [0.2.3] - 2026-09-03
 
 ### Fixed

README.md

@@ -133,6 +133,8 @@ expects), the same way it would issue a few grep or web-search queries:
 recall(queries=["which port does the api listen on", "api port", "8443", "gateway listen port"])
 ```
 
+Byte-identical hits collapse to a single slot, so a prompt captured on many days — a scheduled job's own preamble, say — never crowds out distinct evidence; the turns stay in the store and a session still reads back whole.
+
 Prompt-time injection (the hooks) stays deterministic and only injects beliefs whose
 *subject* the prompt names.
 

src/memware/__init__.py

@@ -22,4 +22,4 @@
     "history",
     "reject",
 ]
-__version__ = "0.2.3"
+__version__ = "0.2.4"

src/memware/index.py

@@ -14,6 +14,11 @@
 activation stays the turn's — recency and use belong to the conversation. Hits
 collapse to one per turn and quote only the matching passages, so recall costs
 roughly a third fewer tokens; reading a session back still returns whole turns.
+
+Results also collapse across turns whose quoted text is byte-identical — the same
+scheduled-automation prompt captured on many days would otherwise take several slots
+with copies of one string. The highest-ranked copy is kept; the turns themselves stay
+in the store, so a session still reads back whole and distinct findings still surface.
 """
 
 from __future__ import annotations
@@ -174,6 +179,7 @@ def search_turns(
     record_use: bool = True,
     snippet_tokens: int = 96,
     passages_per_turn: int = 3,
+    collapse_duplicates: bool = True,
 ) -> list[Hit]:
     """Top-k turns by BM25 x activation, each ranked on its best passage.
 
@@ -208,16 +214,25 @@ def search_turns(
         score = r["rel"] * activation(r["ts"], r["use_count"], decay=decay, use_weight=use_weight)
         scored.setdefault(r["turn_id"], []).append((score, r))
     hits: list[Hit] = []
-    for turn_id in sorted(scored, key=lambda t: -max(s for s, _ in scored[t]))[:k]:
+    seen_text: set[str] = set()
+    for turn_id in sorted(scored, key=lambda t: -max(s for s, _ in scored[t])):
+        if len(hits) >= k:
+            break
         by_score = sorted(scored[turn_id], key=lambda sr: -sr[0])
         keep = sorted(by_score[: max(1, passages_per_turn)], key=lambda sr: sr[1]["ord"])
+        text = _join_passages([r for _, r in keep])
+        if collapse_duplicates:
+            sig = text.strip()
+            if sig in seen_text:
+                continue  # byte-identical to a higher-ranked hit (e.g. a repeated cron prompt)
+            seen_text.add(sig)
         best = by_score[0][1]
         hits.append(
             Hit(
                 id=turn_id,
                 kind="turn",
                 score=by_score[0][0],
-                text=_join_passages([r for _, r in keep]),
+                text=text,
                 session=best["session"],
                 ts=best["ts"],
                 role=best["role"],
@@ -313,6 +328,25 @@ def _rrf(ranked_lists: list[list[Hit]], k: int, c: int = 60) -> list[Hit]:
     return [best[kk] for kk in order]
 
 
+def _collapse_identical(hits: list[Hit], k: int) -> list[Hit]:
+    """Drop turn hits whose quoted text is byte-identical to a higher-ranked one, then take the
+    first ``k``. Non-turn hits pass through. This is the cross-phrasing counterpart to the
+    per-query collapse in :func:`search_turns`: reciprocal-rank fusion keys on ``(kind, id)``,
+    so two different turns holding the same repeated text would otherwise both survive."""
+    seen: set[str] = set()
+    out: list[Hit] = []
+    for h in hits:
+        if h.kind == "turn":
+            sig = h.text.strip()
+            if sig in seen:
+                continue
+            seen.add(sig)
+        out.append(h)
+        if len(out) >= k:
+            break
+    return out
+
+
 def search_turns_multi(
     store: Store,
     queries: list[str],
@@ -334,8 +368,10 @@ def search_turns_multi(
         return []
     if len(seen) == 1:
         return search_turns(store, seen[0], k=k, record_use=record_use, **kw)  # type: ignore[arg-type]
+    collapse = bool(kw.get("collapse_duplicates", True))
     lists = [search_turns(store, q, k=max(k, 20), record_use=False, **kw) for q in seen]  # type: ignore[arg-type]
-    fused = _rrf(lists, k)
+    fused = _rrf(lists, max(k * 3, 30) if collapse else k)
+    fused = _collapse_identical(fused, k) if collapse else fused[:k]
     if record_use and fused:
         ts = now_iso()
         store.conn.executemany(

tests/test_hermes_upstream_plugin.py

@@ -131,10 +131,12 @@ def test_a_turn_is_captured_and_recallable(provider, tmp_path):
 
 def test_a_session_switch_keeps_each_session_separate(provider, tmp_path):
     """Turns must land in the session that was live when they happened."""
-    provider.sync_turn("how is the scheduler wired", "the scheduler polls a work queue")
+    provider.sync_turn(
+        "how is the scheduler wired", "the scheduler polls a work queue every second"
+    )
     provider.shutdown()
     provider.on_session_switch("sess-2")
-    provider.sync_turn("how is the scheduler wired", "the scheduler polls a work queue")
+    provider.sync_turn("how is the scheduler wired", "the scheduler drains a work queue in batches")
     provider.shutdown()
 
     hits = json.loads(

tests/test_index.py

@@ -21,12 +21,12 @@ def test_turn_recall_ranks_recent_higher_and_records_use(store, tmp_path):
     write_claude_jsonl(
         old,
         "old",
-        [("assistant", "2024-01-01T00:00:00Z", "deploy uses the blue-green rollout script")],
+        [("assistant", "2024-01-01T00:00:00Z", "the old deploy ran a blue-green rollout script")],
     )
     write_claude_jsonl(
         new,
         "new",
-        [("assistant", "2026-08-30T00:00:00Z", "deploy uses the blue-green rollout script")],
+        [("assistant", "2026-08-30T00:00:00Z", "the new deploy runs a blue-green rollout script")],
     )
     sync_file(store, old, harness="claude-code")
     sync_file(store, new, harness="claude-code")
@@ -191,3 +191,36 @@ def test_multi_query_fusion_finds_what_single_phrasings_rank_low(store, tmp_path
     )
     assert "8443" in fused[0].text
     assert search_turns_multi(store, ["", "  "], k=3) == []
+
+
+def _write_turn(tmp_path, session, text, ts="2026-09-01T00:00:00Z"):
+    from tests.conftest import write_claude_jsonl
+
+    p = tmp_path / f"{session}.jsonl"
+    write_claude_jsonl(p, session, [("assistant", ts, text)])
+    return p
+
+
+def test_recall_collapses_byte_identical_turns_but_keeps_distinct_findings(store, tmp_path):
+    """A scheduled prompt captured on several days is one string repeated across turns; it
+    should take a single result slot, not crowd out real evidence. Distinct findings that merely
+    share vocabulary must still surface, and every turn stays in the store."""
+    boiler = "you are the nightly drift scan for the pcc runbooks, review each entry"
+    for day in ("d1", "d2", "d3"):  # byte-identical cron prompt, three days
+        sync_file(store, _write_turn(tmp_path, day, boiler), harness="claude-code")
+    sync_file(  # a real finding from one of those runs — shares vocabulary, different text
+        store,
+        _write_turn(
+            tmp_path, "finding", "the nightly drift scan found a stale runbooks entry to fix"
+        ),
+        harness="claude-code",
+    )
+
+    collapsed = search_turns(store, "nightly drift scan runbooks", k=5)
+    assert len(collapsed) == 2  # the repeated prompt -> one slot, plus the distinct finding
+    assert any("stale runbooks entry" in h.text for h in collapsed)  # finding not crowded out
+
+    raw = search_turns(store, "nightly drift scan runbooks", k=5, collapse_duplicates=False)
+    assert sum(1 for h in raw if h.text.strip() == boiler) == 3  # opt out -> copies return
+
+    assert store.stats()["turns"] == 4  # collapse is a recall-time view; nothing was deleted