Adding conflict resolution

Docs/concepts.md

@@ -114,6 +114,33 @@ Prompts for summarization and fact extraction live in `azure_functions/prompts/`
 
 ---
 
+## Memory Reconciliation
+
+The `reconcile_memories(user_id, n=50)` pipeline step reads up to N most-recent active facts for a user and asks the LLM to identify two orthogonal outcomes in one pass:
+
+- **Duplicates** — two or more facts that restate the same claim in different words. Resolution: collapse into one merged fact; the originals are soft-deleted with `supersede_reason="duplicate"` and `superseded_by` set to the merged fact's id.
+- **Contradictions** — two facts that assert opposing claims about the same subject. Resolution: keep the winner (more recent first, higher confidence as tiebreaker), soft-delete the loser with `supersede_reason="contradiction"` and `superseded_by` set to the winner.
+
+### Why one pass
+
+Detecting contradictions semantically requires the LLM to see the candidate pool as a whole — paraphrased ("user prefers aisle seats") and contradictory ("user is vegetarian" vs "user loves steak") facts often have very different embedding vectors and would never co-occur in any cosine cluster. Putting all N candidates into one prompt lets the LLM do the semantic reasoning across both axes simultaneously. The pipeline returns `{"kept": int, "merged": int, "contradicted": int}`.
+
+### Loser preservation
+
+Soft-deleted facts stay in the container with their `supersede_reason`, `superseded_at`, and `superseded_by` fields populated. Default reads (`get_memories`, `search_cosmos`) filter them out via `superseded_by IS NULL`. To inspect the audit trail (e.g. "show everything that ever applied to this user"), opt out of the filter at the query level.
+
+### Write-time exact dedup
+
+Each fact written by `extract_memories` carries a `content_hash` (SHA-256 of normalized content, truncated to 32 hex chars; lowercase, whitespace-collapsed). Before upserting a freshly-extracted fact, the pipeline checks the hash against existing active facts and short-circuits if a match exists, incrementing the `exact_dedup_skipped` metric. This catches identical re-extractions cheaply without an LLM call.
+
+### Tunable
+
+`DEDUP_EVERY_N` (default 5) controls how often `reconcile_memories` runs in the auto-trigger path. Set to `0` to disable. The candidate cap `n` (default 50) is tunable per call; larger values give the LLM a wider view at higher token cost.
+
+> **Indexing note.** The reconcile pool query orders by `created_at` (matching the prompt's "more recent first" tiebreaker). Cosmos's default indexing policy includes every property, so this works out of the box. If you customize the indexing policy to reduce write RU, ensure `/created_at/?` remains indexed or the query will fail with a 400 (`Order-by over a non-indexed path`).
+
+---
+
 ## Automatic Processing (Change Feed)
 
 In addition to on-demand processing via the SDK, the toolkit includes a Cosmos DB change feed trigger that **automatically** starts processing orchestrations when enough new turns have been written.

README.md

@@ -205,14 +205,28 @@ results = memory.search_cosmos("user preferences", user_id="u1", min_confidence=
 high_conf_facts = memory.get_memories(user_id="u1", memory_type="fact", min_confidence=0.7)
 ```
 
+### Memory Reconciliation
+
+`reconcile(user_id, n=50)` (on the public client; underlying pipeline method is `ProcessingPipeline.reconcile_memories`) collapses paraphrased duplicates and resolves semantic contradictions in a single LLM pass over the N most-recent active facts. Both outcomes soft-delete the loser with a `supersede_reason` of `"duplicate"` or `"contradiction"`. See [Docs/concepts.md](Docs/concepts.md#memory-reconciliation) for details.
+
+> **Cost note.** Each reconciliation makes one LLM call covering up to `n` facts (default 50, hard cap 500). With auto-trigger, this fires every `FACT_EXTRACTION_EVERY_N × DEDUP_EVERY_N` turns per user, with `n` taken from `DEDUP_POOL_SIZE`. The previous cosine-cluster pre-filter was removed deliberately — it could not catch semantic contradictions like "vegetarian" vs "ribeye steak" — so the LLM is now invoked whenever there are ≥ 2 active facts. To bound LLM cost more tightly: raise `DEDUP_EVERY_N` (lower frequency — reconcile fires every Nth extraction, so a *higher* N means *less often*), lower `DEDUP_POOL_SIZE` (smaller per-call pool), or override `n` per call when invoking `reconcile()` directly.
+
+| New `MemoryRecord` field | Meaning |
+|---|---|
+| `content_hash` | SHA-256 of normalized content; enables write-time exact-dedup short-circuit |
+| `supersede_reason` | `"duplicate"` or `"contradiction"` (None for live records) |
+| `superseded_at` | ISO timestamp when the supersede happened (None for live records) |
+| `superseded_by` | Id of the record that replaced this one (existing field) |
+
 ### Auto-trigger (per-turn extraction)
 
 By default, the **InProcess processor** runs each pipeline step independently as its own threshold trips inside `push_to_cosmos()`:
 
 | Env var | Default | Step that fires | Async behavior |
 |---|---|---|---|
 | `FACT_EXTRACTION_EVERY_N` | `1` (every turn) | `process_extract_memories` | scheduled via `asyncio.create_task` |
-| `DEDUP_EVERY_N` | `5` | `process_dedup` (fires every Nth extract → effectively every `FACT_EXTRACTION_EVERY_N × DEDUP_EVERY_N` turns) | scheduled via `asyncio.create_task` |
+| `DEDUP_EVERY_N` | `5` | `process_reconcile` (fires every Nth extract → effectively every `FACT_EXTRACTION_EVERY_N × DEDUP_EVERY_N` turns) | scheduled via `asyncio.create_task` |
+| `DEDUP_POOL_SIZE` | `50` | pool size (`n`) passed to `process_reconcile` from the auto-trigger; hard-capped at `500` | n/a (per-call) |
 | `THREAD_SUMMARY_EVERY_N` | `10` | `process_thread_summary` | scheduled via `asyncio.create_task` |
 | `USER_SUMMARY_EVERY_N` | `20` | `process_user_summary` | scheduled via `asyncio.create_task` |
 

Samples/Demo.ipynb

@@ -16,7 +16,7 @@
     "5. **Memory Extraction** – `extract_memories()` (facts + episodic + procedural)\n",
     "6. **User Summary** – `generate_user_summary()` (cross-thread profile)\n",
     "7. **Vector / hybrid search** – `search_cosmos()`\n",
-    "8. **Tagging, salience & deduplication** – tag mutation, salience filter, `deduplicate_facts()`\n",
+    "8. **Tagging, salience & deduplication** – tag mutation, salience filter, `reconcile()`\n",
     "9. **Automatic processing (Change Feed)** – optional Azure Function for background processing\n",
     "\n",
     "> 💡 **Tip:** the synchronous `CosmosMemoryClient` accepts an optional `processor=` kwarg (defaults to `InProcessProcessor`). Pass `DurableFunctionProcessor()` to delegate summarization to the sibling Azure Function app — see `Samples/scenario_remote_processor.py`.\n"

Samples/scenario_memory_reconciliation.py

@@ -0,0 +1,178 @@
+"""Scenario: Memory reconciliation — duplicate merging and contradiction resolution.
+
+Demonstrates how `reconcile_memories` collapses paraphrased facts and resolves
+semantic contradictions in a single LLM pass:
+
+1. Seed paraphrased facts about a user (will be merged).
+2. Seed contradicting facts about the same subject (one will win, one will lose).
+3. Run reconcile() and print the {kept, merged, contradicted} stats.
+4. Show the live state — paraphrased duplicates collapsed, contradiction loser hidden.
+5. Show the audit trail (include_superseded=True) — soft-deleted records carry
+   supersede_reason, superseded_at, and superseded_by pointing at the survivor.
+
+Requirements:
+    - Azure Cosmos DB account with vector-search enabled.
+    - Azure AI Foundry endpoint with chat + embeddings deployments.
+    - Environment variables (.env supported via python-dotenv):
+        COSMOS_DB_ENDPOINT
+        COSMOS_DB_KEY                          (optional, falls back to DefaultAzureCredential)
+        AI_FOUNDRY_ENDPOINT
+        AI_FOUNDRY_EMBEDDING_DEPLOYMENT_NAME
+        AI_FOUNDRY_CHAT_DEPLOYMENT_NAME
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import uuid
+
+from dotenv import load_dotenv
+
+from agent_memory_toolkit import CosmosMemoryClient
+
+load_dotenv()
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+DIVIDER = "-" * 60
+
+
+def banner(title: str) -> None:
+    """Print a section banner."""
+    print(f"\n{DIVIDER}")
+    print(f"  {title}")
+    print(DIVIDER)
+
+
+def print_facts(facts: list[dict]) -> None:
+    """Pretty-print fact records (id + content)."""
+    if not facts:
+        print("  (none)")
+        return
+    for f in facts:
+        print(f"  id={f['id']}  content: {f.get('content', '')}")
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+PARAPHRASED_FACTS = [
+    "User prefers aisle seats on flights.",
+    "User likes aisle seats when flying.",
+    "User always picks aisle when booking flights.",
+]
+
+CONTRADICTING_FACTS = [
+    "User is strictly vegetarian and avoids all meat.",
+    "User loves a good ribeye steak.",
+]
+
+
+def main() -> None:
+    required = ["COSMOS_DB_ENDPOINT", "AI_FOUNDRY_ENDPOINT"]
+    missing = [v for v in required if not os.environ.get(v)]
+    if missing:
+        print(f"ERROR: missing env vars: {', '.join(missing)}")
+        sys.exit(1)
+
+    mem = CosmosMemoryClient(
+        cosmos_endpoint=os.environ["COSMOS_DB_ENDPOINT"],
+        cosmos_key=os.environ.get("COSMOS_DB_KEY") or None,
+        cosmos_database=os.environ.get("COSMOS_DB_DATABASE", "ai_memory"),
+        cosmos_container=os.environ.get("COSMOS_DB_CONTAINER", "memories"),
+        ai_foundry_endpoint=os.environ["AI_FOUNDRY_ENDPOINT"],
+        ai_foundry_api_key=os.environ.get("AI_FOUNDRY_API_KEY") or None,
+        embedding_deployment_name=os.environ.get("AI_FOUNDRY_EMBEDDING_DEPLOYMENT_NAME", "text-embedding-3-large"),
+        chat_deployment_name=os.environ.get("AI_FOUNDRY_CHAT_DEPLOYMENT_NAME", "gpt-4o-mini"),
+        use_default_credential=True,
+    )
+    print("Connected to Cosmos DB.")
+
+    unique_user_id = f"reconcile-demo-{uuid.uuid4().hex[:8]}"
+    unique_thread_id = f"reconcile-demo-thread-{uuid.uuid4().hex[:8]}"
+    print(f"User ID:   {unique_user_id}")
+    print(f"Thread ID: {unique_thread_id}")
+
+    try:
+        banner("1. Seeding paraphrased facts (duplicates)")
+        for content in PARAPHRASED_FACTS:
+            mem.add_cosmos(
+                user_id=unique_user_id,
+                role="user",
+                content=content,
+                memory_type="fact",
+                thread_id=unique_thread_id,
+                salience=0.7,
+            )
+            print(f"  + {content}")
+
+        banner("2. Seeding contradicting facts")
+        for content in CONTRADICTING_FACTS:
+            mem.add_cosmos(
+                user_id=unique_user_id,
+                role="user",
+                content=content,
+                memory_type="fact",
+                thread_id=unique_thread_id,
+                salience=0.7,
+            )
+            print(f"  + {content}")
+
+        banner("3. Active facts before reconcile")
+        before = mem.get_memories(user_id=unique_user_id, memory_type="fact")
+        print_facts(before)
+
+        banner("4. Running reconcile_memories")
+        stats = mem.reconcile(user_id=unique_user_id)
+        print(f"  stats: {dict(stats)}")
+
+        banner("5. Active facts after reconcile (duplicates merged, contradictions resolved)")
+        after = mem.get_memories(user_id=unique_user_id, memory_type="fact")
+        print_facts(after)
+
+        banner("6. Audit trail (soft-deleted records)")
+        all_facts = mem.get_memories(
+            user_id=unique_user_id,
+            memory_type="fact",
+            include_superseded=True,
+        )
+        soft_deleted = [f for f in all_facts if f.get("supersede_reason")]
+        if not soft_deleted:
+            print("  (no soft-deleted records)")
+        for f in soft_deleted:
+            print(
+                f"  id={f['id']}  reason={f.get('supersede_reason')}  "
+                f"superseded_at={f.get('superseded_at')}  "
+                f"superseded_by={f.get('superseded_by')}"
+            )
+            print(f"  content: {f.get('content', '')}")
+
+    finally:
+        banner("7. Cleanup")
+        try:
+            all_records = mem.get_memories(
+                user_id=unique_user_id,
+                include_superseded=True,
+            )
+            deleted = 0
+            for rec in all_records:
+                try:
+                    mem.delete_cosmos(
+                        memory_id=rec["id"],
+                        thread_id=rec.get("thread_id", unique_thread_id),
+                        user_id=unique_user_id,
+                    )
+                    deleted += 1
+                except Exception as exc:  # pragma: no cover - best effort cleanup
+                    print(f"  WARN: failed to delete {rec.get('id')}: {exc}")
+            print(f"  Deleted {deleted} record(s) for user {unique_user_id}")
+        except Exception as exc:  # pragma: no cover - best effort cleanup
+            print(f"  WARN: cleanup failed: {exc}")
+
+
+if __name__ == "__main__":
+    main()

agent_memory_toolkit/_utils.py

@@ -8,6 +8,7 @@
 
 import hashlib
 import os
+import re
 import uuid
 from datetime import datetime, timezone
 from typing import Any, Optional
@@ -37,10 +38,30 @@
 # ---------------------------------------------------------------------------
 
 
+_WHITESPACE_RE = re.compile(r"\s+")
+
+
+def _normalize_for_hash(text: str) -> str:
+    """Lowercase + collapse whitespace for write-time exact-dedup.
+
+    Deliberately conservative: lowercase, strip, and collapse internal runs
+    of whitespace to a single space. Punctuation and word order still matter.
+    The point is to catch *identical* re-extractions cheaply — paraphrases
+    are handled by the reconciliation LLM pass.
+    """
+    return _WHITESPACE_RE.sub(" ", text.strip().lower())
+
+
 def compute_content_hash(content: str) -> str:
-    """SHA-256 of whitespace-normalized content. Does NOT lowercase."""
-    normalized = " ".join(content.split())
-    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
+    """SHA-256 of normalized text, truncated to 32 hex chars.
+
+    Normalization: lowercase + whitespace collapse (see ``_normalize_for_hash``).
+    32 chars (128 bits) is plenty for collision avoidance within a single
+    user's memory set and keeps the field compact in Cosmos documents.
+    Used uniformly across facts, procedural, and episodic memories so the
+    ``content_hash`` field has a single, stable shape regardless of type.
+    """
+    return hashlib.sha256(_normalize_for_hash(content).encode("utf-8")).hexdigest()[:32]
 
 
 # ---------------------------------------------------------------------------

agent_memory_toolkit/aio/cosmos_memory_client.py

@@ -1607,10 +1607,10 @@ async def _run_auto_trigger_steps(
 
             if fire_dedup:
                 try:
-                    await processor.process_dedup(user_id=user_id)
+                    await processor.process_reconcile(user_id=user_id)
                 except Exception as exc:
                     logger.warning(
-                        "Auto-trigger process_dedup failed for %s: %s",
+                        "Auto-trigger process_reconcile failed for %s: %s",
                         user_id,
                         exc,
                     )
@@ -1619,7 +1619,7 @@ async def _run_auto_trigger_steps(
                         thread_counter_id(user_id, thread_id),
                         user_id,
                         thread_id,
-                        f"process_dedup: {exc!r}",
+                        f"process_reconcile: {exc!r}",
                     )
 
             if fire_summary:
@@ -1721,20 +1721,27 @@ async def generate_user_summary(
         self._require_pipeline()
         return await asyncio.to_thread(self._pipeline.generate_user_summary, user_id, thread_ids, recent_k)
 
-    async def deduplicate_facts(
+    async def reconcile(
         self,
         user_id: str,
-        similarity_threshold: float = 0.9,
-        max_facts: int = 200,
+        n: Optional[int] = None,
     ) -> dict[str, int]:
-        """Run LLM-based dedup on user's facts.
+        """Reconcile a user's facts via the contradiction-aware dedup pass.
+
+        ``n`` defaults to the ``DEDUP_POOL_SIZE`` env var (via
+        :func:`agent_memory_toolkit.thresholds.get_dedup_pool_size`) so
+        explicit calls honour the same operator knob the auto-trigger
+        path uses. Pass an integer to override.
 
         Pipeline calls are dispatched to a worker thread via
         :func:`asyncio.to_thread` to avoid blocking the event loop.
         """
+        from ..thresholds import get_dedup_pool_size
+
         await self._require_cosmos()
         self._require_pipeline()
-        return await asyncio.to_thread(self._pipeline.deduplicate_facts, user_id, similarity_threshold, max_facts)
+        pool = n if n is not None else get_dedup_pool_size()
+        return await asyncio.to_thread(self._pipeline.reconcile_memories, user_id, pool)
 
     # ------------------------------------------------------------------
     # Processor delegation

agent_memory_toolkit/aio/processors/base.py

@@ -49,7 +49,7 @@ async def process_user_summary(
         thread_ids: Optional[list[str]] = None,
     ) -> UserSummaryResult: ...
 
-    async def process_dedup(
+    async def process_reconcile(
         self,
         *,
         user_id: str,

agent_memory_toolkit/aio/processors/durable.py

@@ -33,7 +33,7 @@ async def process_thread(
             thread_id,
             len(turns) if turns else 0,
         )
-        return ProcessThreadResult(thread_summary=None, extracted_counts={}, deduplicated_count=0, elapsed_ms=0)
+        return ProcessThreadResult(thread_summary=None, extracted_counts={}, reconciled_count=0, elapsed_ms=0)
 
     async def process_extract_memories(
         self,
@@ -73,8 +73,11 @@ async def process_user_summary(
         )
         return UserSummaryResult(summary=None)
 
-    async def process_dedup(self, *, user_id: str) -> int:
-        logger.debug("AsyncDurableFunctionProcessor.process_dedup no-op user_id=%s", user_id)
+    async def process_reconcile(self, *, user_id: str) -> int:
+        logger.debug(
+            "AsyncDurableFunctionProcessor.process_reconcile no-op user_id=%s",
+            user_id,
+        )
         return 0
 
     async def generate_user_summary(