fix(consolidation): eliminate duplicate observations

Consolidation produced duplicate observations two ways: (1) the cross-encoder
reranker demoted a near-identical existing observation below the recall budget,
hiding it from the LLM; (2) even when shown the twin, the model emitted an
identical CREATE anyway (often while also UPDATE-ing it).

Changes:
- recall: add a 'rerank' flag; consolidation recall ranks by RRF (rerank=False)
  so a near-identical existing observation stays near the top instead of being
  demoted by the cross-encoder. Default recall path is unchanged.
- consolidation: drop a CREATE whose normalized text exactly matches an
  observation shown to the LLM, or the text of an UPDATE issued in the same
  response (exact match => no information loss). See _duplicate_create_target;
  dropped creates are logged at WARNING.
- add a 'reason' field to consolidation create/update/delete actions so the
  model explains its choices (surfaced when a duplicate create is dropped).
- tests/eval: deterministic unit test for the dedup guard
  (test_consolidation_dedup.py); the duplicate-observation rate is tracked by a
  new quality benchmark (hindsight-dev/benchmarks/obs) over a synthetic
  herb-garden transcript and the real PII-free hermes session that surfaced the
  bug, reusing the obs-dedup tool — instead of a flaky real-LLM pytest.

Author: nicoloboschi

hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py

@@ -46,6 +46,32 @@
 logger = logging.getLogger(__name__)
 
 
+def _norm_obs_text(text: str) -> str:
+    """Whitespace/case-normalised observation text for exact-duplicate matching."""
+    return " ".join((text or "").split()).strip().lower()
+
+
+def _duplicate_create_target(
+    create_text: str,
+    shown_obs_by_text: "dict[str, MemoryFact]",
+    update_texts: set[str],
+) -> str | None:
+    """Return a human label for what ``create_text`` duplicates, or None if novel.
+
+    A CREATE is a duplicate when its normalised text matches an observation that was
+    already shown to the LLM, or the text of an UPDATE issued in the same response
+    (the model occasionally UPDATEs the twin to text X and also CREATEs X). Exact-text
+    match means no information is lost by dropping the CREATE.
+    """
+    norm = _norm_obs_text(create_text)
+    matched = shown_obs_by_text.get(norm)
+    if matched is not None:
+        return f"shown observation {str(matched.id)[:8]}"
+    if norm in update_texts:
+        return "an UPDATE in this response"
+    return None
+
+
 @dataclass
 class _BatchDeltas:
     """Per-LLM-batch deltas, merged into the job's running stats after dispatch.
@@ -167,6 +193,9 @@ async def _filter_live_source_memories(
 class _CreateAction(BaseModel):
     text: str
     source_fact_ids: list[str]  # memory UUIDs from the NEW FACTS list
+    # One-sentence justification from the LLM (why CREATE vs UPDATE). Diagnostic
+    # only — surfaced in the consolidation trace to explain duplicate creates.
+    reason: str = ""
 
     @field_validator("text", mode="before")
     @classmethod
@@ -178,6 +207,7 @@ class _UpdateAction(BaseModel):
     text: str
     observation_id: str  # UUID of the existing observation to update
     source_fact_ids: list[str]  # memory UUIDs from the NEW FACTS list
+    reason: str = ""  # LLM's one-sentence justification (diagnostic only)
 
     @field_validator("text", mode="before")
     @classmethod
@@ -187,6 +217,7 @@ def sanitize_text(cls, v: str) -> str:
 
 class _DeleteAction(BaseModel):
     observation_id: str  # UUID of the observation to remove
+    reason: str = ""  # LLM's one-sentence justification (diagnostic only)
 
 
 class _ConsolidationBatchResponse(BaseModel):
@@ -1104,16 +1135,44 @@ async def _process_memory_batch(
         for m in source_mems:
             per_memory_updated.add(str(m["id"]))
 
+    # Deterministic dedup guard: map the observations the LLM was SHOWN by their
+    # normalised text. The model intermittently emits a CREATE whose text is identical
+    # to an observation already in its context (over-aggregation / incoherence — it even
+    # UPDATEs the twin and creates a sibling). When that happens we drop the duplicate
+    # CREATE instead of inserting a redundant row. No extra LLM/embedding cost — the
+    # match is exact text against the in-memory set.
+    shown_obs_by_text = {_norm_obs_text(o.text): o for o in union_observations}
+    # Also collapse a CREATE that reproduces the text of an UPDATE issued in the SAME
+    # response (the model occasionally UPDATEs the twin to text X and also CREATEs X).
+    update_texts = {_norm_obs_text(u.text) for u in llm_result.updates if u.text}
+
     for create in llm_result.creates:
         source_mems = [mem_by_id[fid] for fid in create.source_fact_ids if fid in mem_by_id]
         if not source_mems:
             continue
         agg = _aggregate_source_fields(source_mems, tags=fact_tags)
+        create_source_ids = [m["id"] for m in source_mems]
+
+        # Reconcile against observations shown to the LLM: an exact-text match means
+        # this CREATE reproduces verbatim an observation the model already had in context.
+        # Since that observation already carries this exact text, drop the duplicate CREATE
+        # — no row is inserted, nothing is lost. We deliberately do NOT also UPDATE the twin
+        # here: the LLM frequently UPDATEd it earlier in this same batch, and a second update
+        # would run off the pre-LLM snapshot and clobber that change (see _dedupe_updates).
+        duplicate_of = _duplicate_create_target(create.text, shown_obs_by_text, update_texts)
+        if duplicate_of is not None:
+            logger.warning(
+                "[CONSOLIDATION] dropped duplicate observation CREATE — verbatim match of %s; llm_reason=%r",
+                duplicate_of,
+                create.reason or "(none given)",
+            )
+            continue
+
         await _execute_create_action(
             conn=conn,
             memory_engine=memory_engine,
             bank_id=bank_id,
-            source_memory_ids=[m["id"] for m in source_mems],
+            source_memory_ids=create_source_ids,
             text=create.text,
             source_fact_tags=agg.tags,
             event_date=agg.event_date,
@@ -1398,6 +1457,12 @@ async def _find_related_observations(
             include_source_facts=True,  # Embed source facts so we avoid a separate DB fetch
             max_source_facts_tokens=config.consolidation_source_facts_max_tokens,
             max_source_facts_tokens_per_observation=config.consolidation_source_facts_max_tokens_per_observation,
+            # Rank by RRF, skipping the cross-encoder reranker: for consolidation we are
+            # looking for an existing near-identical observation to merge into, and the
+            # cross-encoder was observed to demote that twin far below the budget cutoff
+            # (semantic rank #1 -> reranked #37). RRF keeps strong lexical/semantic
+            # matches near the top so the LLM is shown the twin and UPDATEs it.
+            rerank=False,
             _quiet=True,  # Suppress logging
         )
     finally:

hindsight-api-slim/hindsight_api/engine/consolidation/prompts.py

@@ -65,7 +65,7 @@
 # Output format — JSON braces escaped as {{ }} so .format() leaves them literal
 _OUTPUT_SECTION = """## OUTPUT FORMAT
 
-Return a JSON object with three arrays: `creates`, `updates`, `deletes`.
+Return a JSON object with three arrays: `creates`, `updates`, `deletes`. Every entry must include a `reason`.
 
 ### Example 1 — Merging recurring claims into an existing observation
 
@@ -79,7 +79,7 @@
 Expected output (one UPDATE, no creates — both new facts are additional evidence for the same canonical decision):
 
 {{"creates": [],
-  "updates": [{{"text": "Donald named Athena's sovereignty as a foundational principle of the Janus architecture.", "observation_id": "11111111-1111-1111-1111-111111111111", "source_fact_ids": ["a1b2c3d4-e5f6-7890-abcd-ef1234567890", "b2c3d4e5-f6a7-8901-bcde-f12345678901"]}}],
+  "updates": [{{"text": "Donald named Athena's sovereignty as a foundational principle of the Janus architecture.", "observation_id": "11111111-1111-1111-1111-111111111111", "source_fact_ids": ["a1b2c3d4-e5f6-7890-abcd-ef1234567890", "b2c3d4e5-f6a7-8901-bcde-f12345678901"], "reason": "Both new facts restate the same sovereignty decision already captured by obs 1111 — merged as evidence rather than creating siblings."}}],
   "deletes": []}}
 
 ### Example 2 — State change updates one observation; unrelated fact creates a new one
@@ -93,8 +93,8 @@
 
 Expected output (UPDATE for the state change; CREATE for the unrelated work-hours facet):
 
-{{"creates": [{{"text": "Alice works long hours, often past midnight.", "source_fact_ids": ["d4e5f6a7-b8c9-0123-defa-234567890123"]}}],
-  "updates": [{{"text": "Alice owned a 2019 Honda Civic; sold it on March 15, 2025.", "observation_id": "22222222-2222-2222-2222-222222222222", "source_fact_ids": ["c3d4e5f6-a7b8-9012-cdef-123456789012"]}}],
+{{"creates": [{{"text": "Alice works long hours, often past midnight.", "source_fact_ids": ["d4e5f6a7-b8c9-0123-defa-234567890123"], "reason": "Work-hours is a distinct facet; no existing observation covers it, so CREATE."}}],
+  "updates": [{{"text": "Alice owned a 2019 Honda Civic; sold it on March 15, 2025.", "observation_id": "22222222-2222-2222-2222-222222222222", "source_fact_ids": ["c3d4e5f6-a7b8-9012-cdef-123456789012"], "reason": "State change to the existing Honda Civic observation 2222 — UPDATE, not a new sibling."}}],
   "deletes": []}}
 
 ### Observation text rules
@@ -110,6 +110,7 @@
 - One create or update may reference multiple facts when they jointly support the observation.
 - **AT MOST ONE UPDATE PER `observation_id`**: if several new facts all update the same existing observation, emit a single `updates` entry that lists all contributing `source_fact_ids` and a single consolidated `text`. Never emit two `updates` entries with the same `observation_id` in one response — they would silently overwrite each other.
 - `deletes`: only when an observation is directly superseded or contradicted by new facts.
+- `reason`: REQUIRED on every create/update/delete — one sentence explaining the choice. For a CREATE, state which existing observation(s) you considered and why none matched (a near-identical existing observation means you should UPDATE, not CREATE). This is audited to catch duplicate creates.
 - Do NOT include `tags` — handled automatically.
 - Return `{{"creates": [], "updates": [], "deletes": []}}` if nothing durable is found."""
 

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -242,6 +242,7 @@ def validate_sql_schema(sql: str) -> None:
 from .search import think_utils
 from .search.reranking import CrossEncoderReranker, apply_combined_scoring
 from .search.tags import TagGroup, TagsMatch, build_tag_groups_where_clause, build_tags_where_clause
+from .search.types import ScoredResult
 from .task_backend import TaskBackend
 from .token_encoding import get_token_encoding
 
@@ -3141,6 +3142,7 @@ async def recall_async(
         created_before: datetime | None = None,
         _connection_budget: int | None = None,
         _quiet: bool = False,
+        rerank: bool = True,
     ) -> RecallResultModel:
         """
         Recall memories using N*4-way parallel retrieval (N fact types × 4 retrieval methods).
@@ -3297,6 +3299,7 @@ async def recall_async(
                             include_source_facts=include_source_facts,
                             max_source_facts_tokens=max_source_facts_tokens,
                             max_source_facts_tokens_per_observation=max_source_facts_tokens_per_observation,
+                            rerank=rerank,
                         )
                         break  # Success - exit retry loop
                     except Exception as e:
@@ -3428,6 +3431,7 @@ async def _search_with_retries(
         include_source_facts: bool = False,
         max_source_facts_tokens: int = 4096,
         max_source_facts_tokens_per_observation: int = -1,
+        rerank: bool = True,
     ) -> RecallResultModel:
         """
         Search implementation with modular retrieval and reranking.
@@ -3791,26 +3795,43 @@ def to_tuple_format(results):
 
             scored_results: list = []
             pre_filtered_count = 0
+            rerank_kind = "cross-encoder"
             try:
-                # Ensure reranker is initialized (for lazy initialization mode)
-                await reranker_instance.ensure_initialized()
-
-                # Pre-filter candidates to reduce reranking cost (RRF already provides good ranking)
-                # This is especially important for remote rerankers with network latency
+                # Pre-filter candidates by RRF before the (optional) cross-encoder.
+                # RRF already provides good ranking; this caps cross-encoder cost.
                 reranker_max_candidates = get_config().reranker_max_candidates
                 if len(merged_candidates) > reranker_max_candidates:
-                    # Sort by RRF score and take top candidates
                     merged_candidates.sort(key=lambda mc: mc.rrf_score, reverse=True)
                     pre_filtered_count = len(merged_candidates) - reranker_max_candidates
                     merged_candidates = merged_candidates[:reranker_max_candidates]
 
-                # Rerank using cross-encoder
-                scored_results = await reranker_instance.rerank(query, merged_candidates)
+                if rerank:
+                    # Ensure reranker is initialized (for lazy initialization mode)
+                    await reranker_instance.ensure_initialized()
+                    scored_results = await reranker_instance.rerank(query, merged_candidates)
+                else:
+                    # RRF-only: skip the cross-encoder and rank by RRF score. Used by
+                    # consolidation recall, where the cross-encoder was observed to demote
+                    # a near-identical existing observation (the dedup "twin") far below the
+                    # budget cutoff (semantic rank #1 -> reranked #37), causing the LLM to
+                    # never see it and create a duplicate. Passthrough ScoredResults keep the
+                    # downstream recency/temporal boosts working, seeded from RRF rank.
+                    rerank_kind = "rrf-passthrough"
+                    scored_results = [
+                        ScoredResult(
+                            candidate=mc,
+                            cross_encoder_score=0.0,
+                            cross_encoder_score_normalized=0.0,
+                            weight=0.0,
+                        )
+                        for mc in sorted(merged_candidates, key=lambda mc: mc.rrf_score, reverse=True)
+                    ]
 
                 step_duration = time.time() - step_start
                 pre_filter_note = f" (pre-filtered {pre_filtered_count})" if pre_filtered_count > 0 else ""
                 log_buffer.append(
-                    f"  [4] Reranking: {len(scored_results)} candidates scored in {step_duration:.3f}s{pre_filter_note}"
+                    f"  [4] Reranking [{rerank_kind}]: {len(scored_results)} candidates "
+                    f"scored in {step_duration:.3f}s{pre_filter_note}"
                 )
             finally:
                 rerank_span.set_attribute("hindsight.scored_count", len(scored_results))
@@ -3825,7 +3846,8 @@ def to_tuple_format(results):
             # the slim/passthrough one that returns a constant score per pair.
             if scored_results:
                 ce = reranker_instance.cross_encoder
-                is_passthrough = ce is not None and ce.provider_name == "rrf"
+                # RRF-only mode is passthrough by construction; so is a configured "rrf" CE.
+                is_passthrough = (not rerank) or (ce is not None and ce.provider_name == "rrf")
                 apply_combined_scoring(
                     scored_results,
                     now=_recall_scoring_now(question_date),
@@ -3845,7 +3867,7 @@ def to_tuple_format(results):
                 tracer.add_phase_metric(
                     "reranking",
                     step_duration,
-                    {"reranker_type": "cross-encoder", "candidates_reranked": len(scored_results)},
+                    {"reranker_type": rerank_kind, "candidates_reranked": len(scored_results)},
                 )
 
             # Step 5: Truncate to thinking_budget * 2 for token filtering

hindsight-api-slim/tests/test_consolidation_dedup.py

@@ -0,0 +1,45 @@
+"""Deterministic unit tests for the consolidation duplicate-create guard.
+
+These exercise the dedup decision directly (no LLM, no DB), so they reliably
+guard the fix in CI — unlike the real-LLM integration test, which only triggers
+the path stochastically.
+"""
+
+from dataclasses import dataclass
+
+from hindsight_api.engine.consolidation.consolidator import _duplicate_create_target, _norm_obs_text
+
+
+@dataclass
+class _FakeObs:
+    id: str
+    text: str
+
+
+def _shown(*observations: _FakeObs) -> dict[str, _FakeObs]:
+    return {_norm_obs_text(o.text): o for o in observations}
+
+
+def test_norm_obs_text_collapses_whitespace_and_case() -> None:
+    assert _norm_obs_text("  The  User  likes BASIL.\n") == "the user likes basil."
+    assert _norm_obs_text(None) == ""
+
+
+def test_create_matching_shown_observation_is_duplicate() -> None:
+    shown = _shown(_FakeObs(id="11111111-aaaa", text="User waters the herbs early in the morning."))
+    # Same text with different whitespace/case still matches.
+    target = _duplicate_create_target("user waters the herbs early in the morning.", shown, set())
+    assert target is not None
+    assert target.startswith("shown observation 11111111")
+
+
+def test_create_matching_inresponse_update_is_duplicate() -> None:
+    update_texts = {_norm_obs_text("Mint is kept in its own separate bed.")}
+    target = _duplicate_create_target("Mint is kept in its own separate bed.", {}, update_texts)
+    assert target == "an UPDATE in this response"
+
+
+def test_novel_create_is_not_duplicate() -> None:
+    shown = _shown(_FakeObs(id="22222222-bbbb", text="User waters the herbs early in the morning."))
+    assert _duplicate_create_target("Rosemary is drought-tolerant.", shown, set()) is None
+    assert _duplicate_create_target("", {}, set()) is None

hindsight-dev/benchmarks/obs/README.md

@@ -0,0 +1,32 @@
+# Observation duplication benchmark
+
+Measures how many **duplicate observations** consolidation produces — a quality
+signal for the consolidation pipeline (complements the perf-focused
+`consolidation` benchmark).
+
+For each document under `datasets/`, it ingests the content into a fresh bank,
+runs consolidation, then reuses the observation-dedup tool
+(`hindsight_dev.obs_dedup`) to score:
+
+- **exact duplicates** — observations with identical (normalised) text in a scope
+- **near duplicates** — cosine-similarity clusters at thresholds (0.97, 0.92)
+
+Headline metric: **duplication rate** = redundant observations / total. Lower is
+better.
+
+## Run
+
+```bash
+./scripts/benchmarks/run-obs.sh
+# or
+cd hindsight-dev && uv run python -m benchmarks.obs.obs_benchmark
+```
+
+Requires a real LLM (set `HINDSIGHT_API_LLM_PROVIDER` / `_MODEL` / `_API_KEY`).
+Results are written to `benchmarks/results/obs_benchmark_<ts>.json`.
+
+## Extending
+
+Drop more `*.txt` transcripts into `datasets/` — one file per scenario. Keep them
+synthetic and PII-free. Documents that restate the same durable facts across many
+turns are the ones that stress consolidation's dedup behaviour.

hindsight-dev/benchmarks/obs/__init__.py

@@ -0,0 +1 @@
+"""Observation-quality benchmark: measures duplicate-observation rate after consolidation."""