fix(consolidation): eliminate duplicate observations (interleave dedup recall + tool & benchmark)

.github/workflows/perf-test.yml

@@ -34,6 +34,18 @@ on:
         description: "Skip LoComo job"
         type: boolean
         default: false
+      obs_skip:
+        description: "Skip observation-dedup benchmark job"
+        type: boolean
+        default: false
+      obs_dataset:
+        description: "Obs benchmark dataset substring (blank = English hermes transcript)."
+        type: string
+        default: ""
+      obs_fraction:
+        description: "Obs benchmark fraction (0-1] of each document to run."
+        type: string
+        default: "1.0"
       ref:
         description: "Git ref to test (branch, tag, or SHA). Defaults to main."
         type: string
@@ -199,3 +211,86 @@ jobs:
         PERF_DASHBOARD_TOKEN: ${{ secrets.PERF_DASHBOARD_TOKEN }}
         GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       run: ./scripts/benchmarks/publish-locomo-results.sh hindsight-dev/benchmarks/locomo/results/benchmark_results.json
+
+  obs:
+    # Observation-dedup quality benchmark: ingests a transcript, drains consolidation
+    # (serial SyncTaskBackend + embedded pg0 — no external DB / worker), and reports the
+    # near-duplicate observation rate. Real LLM via VertexAI, mirroring the LoComo job.
+    if: inputs.obs_skip != true
+    runs-on: ubuntu-latest
+    env:
+      HINDSIGHT_API_LLM_PROVIDER: vertexai
+      HINDSIGHT_API_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY: /tmp/gcp-credentials.json
+      HINDSIGHT_API_LLM_MODEL: google/gemini-2.5-flash-lite
+      HINDSIGHT_API_ENABLE_OBSERVATIONS: "true"
+    steps:
+    - uses: actions/checkout@v6
+      with:
+        ref: ${{ inputs.ref || github.ref }}
+
+    - name: Setup GCP credentials
+      run: |
+        printf '%s' '${{ secrets.GCP_VERTEXAI_CREDENTIALS }}' > /tmp/gcp-credentials.json
+        PROJECT_ID=$(jq -r '.project_id' /tmp/gcp-credentials.json)
+        echo "HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID=$PROJECT_ID" >> $GITHUB_ENV
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v7
+      with:
+        enable-cache: true
+        prune-cache: false
+
+    - name: Set up Python
+      uses: actions/setup-python@v6
+      with:
+        python-version-file: ".python-version"
+
+    - name: Cache HuggingFace models
+      uses: actions/cache@v5
+      with:
+        path: ~/.cache/huggingface
+        key: ${{ runner.os }}-huggingface-${{ hashFiles('hindsight-api-slim/pyproject.toml') }}
+        restore-keys: |
+          ${{ runner.os }}-huggingface-
+
+    - name: Pre-download models
+      working-directory: ./hindsight-api-slim
+      run: |
+        uv run --frozen --all-extras --index-strategy unsafe-best-match python -c "
+        from sentence_transformers import SentenceTransformer
+        print('Downloading embedding model...')
+        SentenceTransformer('BAAI/bge-small-en-v1.5')
+        print('Model downloaded successfully')
+        "
+
+    - name: Install hindsight-dev dependencies
+      run: |
+        cd hindsight-dev && uv sync --frozen --all-extras --index-strategy unsafe-best-match
+
+    - name: Run obs benchmark
+      # Default to the English hermes transcript at full fraction — a clean, deterministic
+      # consolidation-dedup signal (the Chinese variant adds a cross-lingual embedding
+      # confound). Override dataset/fraction via workflow_dispatch.
+      run: |
+        DATASET="${{ inputs.obs_dataset }}"
+        if [ -z "$DATASET" ]; then DATASET="hermes_session_2026-05-15_en"; fi
+        FRACTION="${{ inputs.obs_fraction }}"
+        if [ -z "$FRACTION" ]; then FRACTION="1.0"; fi
+        cd hindsight-dev
+        uv run python -m benchmarks.obs.obs_benchmark \
+          --dataset "$DATASET" --fraction "$FRACTION" --wipe-bank --output obs-results.json
+
+    - name: Upload obs results
+      if: always()
+      uses: actions/upload-artifact@v7
+      with:
+        name: obs-results-${{ github.sha }}
+        path: hindsight-dev/obs-results.json
+        retention-days: 90
+
+    - name: Publish obs to dashboard
+      if: success() && (github.event_name == 'schedule' || github.event_name == 'push' || github.event_name == 'workflow_dispatch')
+      env:
+        PERF_DASHBOARD_TOKEN: ${{ secrets.PERF_DASHBOARD_TOKEN }}
+        GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      run: ./scripts/benchmarks/publish-obs-results.sh hindsight-dev/obs-results.json

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

@@ -53,6 +53,38 @@
 logger = logging.getLogger(__name__)
 
 
+def _norm_obs_text(text: str) -> str:
+    """Whitespace-normalised observation text for exact-duplicate matching.
+
+    Collapses runs of whitespace only; case is preserved. The reconciliation guard
+    drops a CREATE on the premise that an exact-text match loses no information — but
+    case-folding would also drop a create differing only in case (e.g. "TLS" vs "tls"),
+    which *does* lose information, so we match case-sensitively.
+    """
+    return " ".join((text or "").split()).strip()
+
+
+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.
@@ -174,6 +206,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
@@ -185,6 +220,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
@@ -194,6 +230,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):
@@ -1142,16 +1179,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,
@@ -1442,6 +1507,13 @@ 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,
+            # Round-robin interleave fusion (no cross-encoder): consolidation is looking
+            # for an existing near-identical observation to merge into. Both the
+            # cross-encoder (semantic #1 -> reranked #37) and RRF (semantic #1 -> outside
+            # the 512-token budget) were measured to bury that twin; interleave guarantees
+            # each retrieval arm's top hits a slot, so the semantic-#1 twin is always shown
+            # to the LLM, which then UPDATEs instead of creating a duplicate.
+            reranking="interleave",
             _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."""