fix: re-ingest row inflation — dedup by content_hash (#219)

docs/design/content-hash-dedup.md

@@ -0,0 +1,116 @@
+# Content-hash deduplication contract (#219)
+
+## Problem
+
+`belief_id` is `sha256(source + NUL + sentence)[:16]`. The same sentence
+arriving from two different sources (e.g. a git commit message and a
+filesystem scan of the same file) produces **different** `belief_id` values
+but **identical** `content_hash` values. Without a UNIQUE constraint on
+`content_hash`, repeated ingest inflates the table — a 5.3x blow-up was
+observed on real stores.
+
+## Fix (v1.x, ships in the release that closes #219)
+
+Two layers of protection, both in `MemoryStore`:
+
+### 1. `insert_or_corroborate()` — live dedup at ingest time
+
+```python
+def insert_or_corroborate(
+    self, b: Belief, *, source_type: str,
+    session_id: str | None = None,
+    source_path_hash: str | None = None,
+) -> tuple[str, bool]:
+```
+
+All ingest call sites use this instead of `insert_belief()` directly.
+
+- If a belief with the same `content_hash` already exists: records a
+  corroboration row on the existing belief and returns `(existing_id, False)`.
+- Otherwise: inserts the new belief and returns `(b.id, True)`.
+
+The returned `bool` tells the caller whether a new row was created so it can
+maintain accurate `beliefs_inserted` counts.
+
+### 2. One-shot migrations on store open
+
+Both migrations are idempotent via `schema_meta` markers. They run on every
+store open until their respective markers are set, then become a no-op
+single-row read.
+
+#### `_maybe_consolidate_content_hash_duplicates()`
+
+Marker: `SCHEMA_META_CONTENT_HASH_DEDUP_COMPLETE = "content_hash_dedup_complete"`
+
+Runs **before** the UNIQUE migration. For each duplicate group (beliefs
+sharing a `content_hash`) it:
+
+- Picks the **canonical** row: `ORDER BY created_at ASC, id ASC` (oldest first).
+- Sums `alpha` and `beta` across the group (each row carries independent
+  Bayesian evidence accumulated from a distinct source).
+- Propagates `lock_level = 'user'` if any member holds it.
+- Propagates the highest-precedence `origin` (`user_stated` > `user_corrected`
+  > `user_validated` > `agent_remembered` > `agent_inferred` > `document_recent`
+  > `unknown`).
+- Takes `MAX(last_retrieved_at)` across the group.
+- Rewrites FK references: `feedback_history`, `belief_corroborations`, `edges`
+  (both `src` and `dst`).
+- Drops `belief_entities` and `belief_versions` rows for duplicates (same
+  content → same entities; the canonical row already has them).
+- Inserts one `belief_corroborations` row per duplicate consumed with
+  `source_type = 'consolidation_migration'` to preserve the count signal.
+- Deletes all duplicate rows from `beliefs` and `beliefs_fts` in a single
+  bulk `DELETE ... WHERE id IN (...)`.
+
+All writes are inside one transaction. On a 20 K-belief store with ~2 K
+duplicate groups the pass completes in under 2 seconds.
+
+#### `_maybe_apply_content_hash_unique()`
+
+Marker: `SCHEMA_META_CONTENT_HASH_UNIQUE_APPLIED = "content_hash_unique_applied"`
+
+Adds `UNIQUE(content_hash)` to the `beliefs` table via a SQLite table-swap
+(SQLite does not support `ALTER TABLE ADD CONSTRAINT`):
+
+1. Read column definitions via `PRAGMA table_info(beliefs)` — preserves any
+   columns added by prior `ALTER TABLE` migrations (e.g. `hibernation_score`,
+   `activation_condition`).
+2. `DROP TABLE IF EXISTS beliefs_new` — clears any partial state from a prior
+   failed attempt.
+3. `CREATE TABLE beliefs_new` with `UNIQUE` added to `content_hash`.
+4. `INSERT INTO beliefs_new SELECT ... FROM beliefs`.
+5. `DROP TABLE beliefs`.
+6. `ALTER TABLE beliefs_new RENAME TO beliefs`.
+7. Recreate `idx_beliefs_session` and `idx_beliefs_origin`.
+
+Fresh stores (created with the new `_SCHEMA`) already have the constraint in
+DDL; they skip the swap and only stamp the marker.
+
+## Corroboration source types added
+
+| Constant | Value | Used by |
+|---|---|---|
+| `CORROBORATION_SOURCE_FILESYSTEM_INGEST` | `"filesystem_ingest"` | `scanner.py`, `classification.py` |
+| `CORROBORATION_SOURCE_CLI_REMEMBER` | `"cli_remember"` | `cli.py` |
+| `CORROBORATION_SOURCE_CONSOLIDATION_MIGRATION` | `"consolidation_migration"` | migration pass |
+
+## Ingest call sites migrated
+
+| Module | Old call | New call |
+|---|---|---|
+| `ingest.py` | `store.insert_belief(out.belief)` | `store.insert_or_corroborate(..., source_type=CORROBORATION_SOURCE_TRANSCRIPT_INGEST)` |
+| `scanner.py` (LLM + regex paths) | `store.insert_belief(b)` | `store.insert_or_corroborate(..., source_type=CORROBORATION_SOURCE_FILESYSTEM_INGEST)` |
+| `classification.py` | `store.insert_belief(b)` | `store.insert_or_corroborate(..., source_type=CORROBORATION_SOURCE_FILESYSTEM_INGEST)` |
+| `cli.py` (`_cmd_lock`) | `store.insert_belief(b)` | `store.insert_or_corroborate(..., source_type=CORROBORATION_SOURCE_CLI_REMEMBER)` |
+| `triple_extractor.py` | `store.insert_belief(b)` | `store.insert_or_corroborate(..., source_type=CORROBORATION_SOURCE_COMMIT_INGEST)` |
+| `mcp_server.py` (`tool_lock`) | `store.insert_belief(b)` | `store.insert_or_corroborate(..., source_type=CORROBORATION_SOURCE_MCP_REMEMBER)` |
+
+## Invariants
+
+- `beliefs.content_hash` is UNIQUE. Any attempt to INSERT a duplicate via raw
+  SQL raises `sqlite3.IntegrityError`.
+- `insert_or_corroborate()` is the only sanctioned way to insert a belief from
+  ingest paths. `insert_belief()` remains for test fixtures and the migration
+  pass itself.
+- The consolidation migration always runs before the UNIQUE migration on the
+  same store open, guaranteeing no UNIQUE violation can occur during the swap.

src/aelfrice/classification.py

@@ -40,6 +40,7 @@
     BELIEF_PREFERENCE,
     BELIEF_REQUIREMENT,
     BELIEF_TYPES,
+    CORROBORATION_SOURCE_FILESYSTEM_INGEST,
     INGEST_SOURCE_FILESYSTEM,
     ONBOARD_STATE_PENDING,
     OnboardSession,
@@ -519,8 +520,12 @@ def accept_classifications(
             derived_belief_ids=[bid],
             ts=timestamp,
         )
-        store.insert_belief(out.belief)
-        inserted += 1
+        _, was_inserted = store.insert_or_corroborate(
+            out.belief,
+            source_type=CORROBORATION_SOURCE_FILESYSTEM_INGEST,
+        )
+        if was_inserted:
+            inserted += 1
 
     store.complete_onboard_session(session_id, timestamp)
     return AcceptOnboardResult(

src/aelfrice/cli.py

@@ -58,6 +58,7 @@
     regime_description,
 )
 from aelfrice.models import (
+    CORROBORATION_SOURCE_CLI_REMEMBER,
     INGEST_SOURCE_CLI_REMEMBER,
     LOCK_NONE,
     LOCK_USER,
@@ -804,8 +805,14 @@ def _cmd_lock(args: argparse.Namespace, out: object) -> int:
                 derived_belief_ids=[bid],
                 ts=now,
             )
-            store.insert_belief(derived.belief)
-            print(f"locked: {bid}", file=out)  # type: ignore[arg-type]
+            actual_id, was_inserted = store.insert_or_corroborate(
+                derived.belief,
+                source_type=CORROBORATION_SOURCE_CLI_REMEMBER,
+            )
+            if was_inserted:
+                print(f"locked: {actual_id}", file=out)  # type: ignore[arg-type]
+            else:
+                print(f"locked: {actual_id} (corroborated existing)", file=out)  # type: ignore[arg-type]
         else:
             existing.lock_level = LOCK_USER
             existing.locked_at = now

src/aelfrice/ingest.py

@@ -146,8 +146,17 @@ def _ingest_turn_ids(
             session_id=session_id,
             ts=ts,
         )
-        store.insert_belief(out.belief)
-        inserted.append(belief_id)
+        # Cross-source dedup: same content from a different source
+        # produces the same content_hash but a different belief_id.
+        # insert_or_corroborate records a corroboration row on hit
+        # instead of silently inserting a duplicate belief row.
+        actual_id, was_inserted = store.insert_or_corroborate(
+            out.belief,
+            source_type=CORROBORATION_SOURCE_TRANSCRIPT_INGEST,
+            session_id=session_id,
+        )
+        if was_inserted:
+            inserted.append(actual_id)
     return inserted
 
 

src/aelfrice/mcp_server.py

@@ -218,8 +218,13 @@ def tool_lock(store: MemoryStore, *, statement: str) -> dict[str, Any]:
             derived_belief_ids=[bid],
             ts=now,
         )
-        store.insert_belief(out.belief)
-        return {"kind": "lock.created", "id": bid, "action": "locked"}
+        actual_id, was_inserted = store.insert_or_corroborate(
+            out.belief,
+            source_type=CORROBORATION_SOURCE_MCP_REMEMBER,
+        )
+        if was_inserted:
+            return {"kind": "lock.created", "id": actual_id, "action": "locked"}
+        return {"kind": "lock.corroborated", "id": actual_id, "action": "corroborated"}
     existing.lock_level = LOCK_USER
     existing.locked_at = now
     existing.demotion_pressure = 0

src/aelfrice/models.py

@@ -82,12 +82,28 @@
 CORROBORATION_SOURCE_TRANSCRIPT_INGEST: Final[str] = "transcript_ingest"
 CORROBORATION_SOURCE_MCP_REMEMBER: Final[str] = "mcp_remember"
 CORROBORATION_SOURCE_HOOK_INGEST: Final[str] = "hook_ingest"
+# filesystem_ingest: scanner / onboard paths that read local files
+# (not git history). Distinct from transcript_ingest (conversation
+# turns) and commit_ingest (git history).
+CORROBORATION_SOURCE_FILESYSTEM_INGEST: Final[str] = "filesystem_ingest"
+# cli_remember: `aelf lock` / `aelf remember` terminal command, where
+# the user explicitly types a statement to lock.
+CORROBORATION_SOURCE_CLI_REMEMBER: Final[str] = "cli_remember"
+# consolidation_migration: synthetic row inserted by the one-shot
+# content_hash dedup pass (commit 3). Records that a duplicate row was
+# absorbed into the canonical belief; the source_type is preserved so
+# downstream consumers know the row is migration-produced, not a live
+# re-ingest.
+CORROBORATION_SOURCE_CONSOLIDATION_MIGRATION: Final[str] = "consolidation_migration"
 
 CORROBORATION_SOURCE_TYPES: Final[frozenset[str]] = frozenset({
     CORROBORATION_SOURCE_COMMIT_INGEST,
     CORROBORATION_SOURCE_TRANSCRIPT_INGEST,
     CORROBORATION_SOURCE_MCP_REMEMBER,
     CORROBORATION_SOURCE_HOOK_INGEST,
+    CORROBORATION_SOURCE_FILESYSTEM_INGEST,
+    CORROBORATION_SOURCE_CLI_REMEMBER,
+    CORROBORATION_SOURCE_CONSOLIDATION_MIGRATION,
 })
 
 # v2.0 #205 ingest_log source_kind enum. Wire-format strings; do not

src/aelfrice/scanner.py

@@ -28,6 +28,7 @@
 from aelfrice.derivation import DerivationInput, derive
 from aelfrice.inedible import is_inedible
 from aelfrice.models import (
+    CORROBORATION_SOURCE_FILESYSTEM_INGEST,
     INGEST_SOURCE_FILESYSTEM,
     LOCK_NONE,
     Belief,
@@ -245,21 +246,25 @@ def scan_repo(
                 derived_belief_ids=[belief_id],
                 ts=created_at,
             )
-            store.insert_belief(Belief(
-                id=belief_id,
-                content=candidate.text,
-                content_hash=_content_hash(candidate.text),
-                alpha=route.alpha,
-                beta=route.beta,
-                type=route.belief_type,
-                lock_level=LOCK_NONE,
-                locked_at=None,
-                demotion_pressure=0,
-                created_at=created_at,
-                last_retrieved_at=None,
-                origin=route.origin,
-            ))
-            inserted += 1
+            _, was_inserted = store.insert_or_corroborate(
+                Belief(
+                    id=belief_id,
+                    content=candidate.text,
+                    content_hash=_content_hash(candidate.text),
+                    alpha=route.alpha,
+                    beta=route.beta,
+                    type=route.belief_type,
+                    lock_level=LOCK_NONE,
+                    locked_at=None,
+                    demotion_pressure=0,
+                    created_at=created_at,
+                    last_retrieved_at=None,
+                    origin=route.origin,
+                ),
+                source_type=CORROBORATION_SOURCE_FILESYSTEM_INGEST,
+            )
+            if was_inserted:
+                inserted += 1
             # Audit row for fallback insertions (spec § 7.2 step 3).
             if route.audit_source is not None:
                 store.insert_feedback_event(
@@ -290,8 +295,12 @@ def scan_repo(
                 derived_belief_ids=[belief_id],
                 ts=created_at,
             )
-            store.insert_belief(out.belief)
-            inserted += 1
+            _, was_inserted = store.insert_or_corroborate(
+                out.belief,
+                source_type=CORROBORATION_SOURCE_FILESYSTEM_INGEST,
+            )
+            if was_inserted:
+                inserted += 1
 
     return ScanResult(
         inserted=inserted,