Add procedural synthesis, strip per-turn procedural, fact_count rename

Brings forward the procedural-synthesis feature on top of the post-PR-#13
baseline and applies hygiene cleanup.

- Adds synthesize_procedural() on pipeline + processors (sync/async) and
  the Durable Functions orchestrator. Procedural memory is now produced
  via this dedicated reflection flow over a user's full history (system-
  prompt-style), not as a per-turn extraction.
- Strips procedural emission from extract_memories: removes the procedural
  type from the extraction prompt, drops procedural_count from result
  dicts and docstrings, and trims procedural from _load_existing_memories.
- Renames facts_count -> fact_count across pipeline, processors, FA
  orchestrator docstring, and all tests.
- Cosmetic enum rename: supersede_reason value 'contradiction' ->
  'contradict' for verb-tense consistency with 'update'. README and
  Docs/concepts.md updated accordingly.

Unit suite: 531 passed.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: 2 people

Docs/concepts.md

@@ -119,7 +119,7 @@ Prompts for summarization and fact extraction live in `azure_functions/prompts/`
 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.
+- **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="contradict"` and `superseded_by` set to the winner.
 
 ### Why one pass
 

README.md

@@ -170,14 +170,14 @@ high_conf_facts = memory.get_memories(user_id="u1", memory_types=["fact"], min_c
 
 ### 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.
+`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 `"contradict"`. 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) |
+| `supersede_reason` | `"duplicate"` or `"contradict"` (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) |
 

agent_memory_toolkit/aio/cosmos_memory_client.py

@@ -1179,6 +1179,58 @@ async def remove_tags(self, memory_id: str, user_id: str, thread_id: str, tags:
     # Procedural and episodic memory retrieval
     # ------------------------------------------------------------------
 
+    async def get_procedural_prompt(self, user_id: str) -> Optional[str]:
+        """Return the active synthesized procedural prompt for a user."""
+        await self._require_cosmos()
+
+        qb = _QueryBuilder()
+        qb.add_filter("c.user_id", "@user_id", user_id)
+        qb.add_filter("c.thread_id", "@thread_id", "__procedural__")
+        qb.add_filter("c.type", "@type", "procedural")
+        qb.add_is_null_or_undefined("c.superseded_by")
+
+        query = f"SELECT TOP 1 c.content, c.version FROM c{qb.build_where()} ORDER BY c.version DESC"
+        try:
+            items_iter = self._container_client.query_items(query=query, parameters=qb.get_parameters())
+            items = [item async for item in items_iter]
+        except Exception as exc:
+            raise CosmosOperationError(f"async get_procedural_prompt query failed: {exc}") from exc
+
+        if not items:
+            return None
+        return items[0].get("content")
+
+    async def get_procedural_history(self, user_id: str, limit: int = 10) -> list[dict[str, Any]]:
+        """Return synthesized procedural docs for a user, newest first."""
+        await self._require_cosmos()
+        if limit <= 0:
+            return []
+
+        qb = _QueryBuilder()
+        qb.add_filter("c.user_id", "@user_id", user_id)
+        qb.add_filter("c.thread_id", "@thread_id", "__procedural__")
+        qb.add_filter("c.type", "@type", "procedural")
+
+        query = f"SELECT * FROM c{qb.build_where()} ORDER BY c.version DESC"
+        try:
+            items_iter = self._container_client.query_items(query=query, parameters=qb.get_parameters())
+            items = [item async for item in items_iter]
+        except Exception as exc:
+            raise CosmosOperationError(f"async get_procedural_history query failed: {exc}") from exc
+
+        def _is_active(doc: dict[str, Any]) -> bool:
+            return not doc.get("superseded_by")
+
+        items.sort(
+            key=lambda doc: (
+                1 if _is_active(doc) else 0,
+                int(doc.get("version") or 0),
+                int(doc.get("_ts") or 0),
+            ),
+            reverse=True,
+        )
+        return items[:limit]
+
     async def get_procedural_memories(
         self,
         user_id: str,
@@ -1244,14 +1296,7 @@ async def search_episodic_memories(
 
     async def build_procedural_context(self, user_id: str) -> str:
         """Build formatted text for system prompt injection."""
-        memories = await self.get_procedural_memories(user_id)
-        if not memories:
-            return ""
-        lines = ["## Learned User Preferences"]
-        for m in memories:
-            priority = m.get("metadata", {}).get("priority", "should")
-            lines.append(f"- {m['content']} [{priority}]")
-        return "\n".join(lines)
+        return await self.get_procedural_prompt(user_id) or ""
 
     async def build_episodic_context(self, user_id: str, query: str, top_k: int = 3) -> str:
         """Build formatted context of relevant past experiences."""
@@ -1680,7 +1725,7 @@ async def extract_memories(
         thread_id: str,
         recent_k: Optional[int] = None,
     ) -> dict[str, int]:
-        """Extract facts, procedural, and episodic memories from a thread.
+        """Extract facts and episodic memories from a thread.
 
         Pipeline calls are dispatched to a worker thread via
         :func:`asyncio.to_thread` to avoid blocking the event loop on
@@ -1690,6 +1735,34 @@ async def extract_memories(
         self._require_pipeline()
         return await asyncio.to_thread(self._pipeline.extract_memories, user_id, thread_id, recent_k)
 
+    async def synthesize_procedural(
+        self,
+        user_id: str,
+        *,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """Trigger synthesized procedural prompt generation for a user.
+
+        For DurableFunctionProcessor this returns a deferred status; synthesis
+        is auto-driven inside the Function App. ``force=True`` is only honored
+        by AsyncInProcessProcessor.
+        """
+        await self._require_cosmos()
+        processor = self._get_processor()
+        if not isinstance(processor, AsyncInProcessProcessor):
+            logger.debug("synthesize_procedural deferred to Function App auto-trigger user_id=%s", user_id)
+            return {
+                "status": "deferred",
+                "reason": "durable_auto_trigger",
+                "message": (
+                    "Procedural synthesis runs reactively in the Function App after each "
+                    "ExtractMemoriesOrchestrator pass. Use get_procedural_prompt() to read "
+                    "the synthesized prompt once it has been generated."
+                ),
+            }
+
+        return await processor.synthesize_procedural(user_id=user_id, force=force)
+
     async def generate_thread_summary(
         self,
         user_id: str,

agent_memory_toolkit/aio/processors/durable.py

@@ -93,6 +93,19 @@ async def generate_user_summary(
         )
         return UserSummaryResult(summary=None)
 
+    async def synthesize_procedural(
+        self,
+        *,
+        user_id: str,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        logger.debug(
+            "AsyncDurableFunctionProcessor.synthesize_procedural deferred user_id=%s force=%s",
+            user_id,
+            force,
+        )
+        return {"status": "deferred", "reason": "durable_auto_trigger"}
+
     async def close(self) -> None:
         logger.debug("AsyncDurableFunctionProcessor.close no-op")
         return None

agent_memory_toolkit/aio/processors/inprocess.py

@@ -146,6 +146,15 @@ async def generate_user_summary(
         summary = await asyncio.to_thread(self._pipeline.generate_user_summary, user_id, thread_ids)
         return UserSummaryResult(summary=summary if isinstance(summary, dict) else None)
 
+    async def synthesize_procedural(
+        self,
+        *,
+        user_id: str,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """Run procedural prompt synthesis through the in-process pipeline."""
+        return await asyncio.to_thread(self._pipeline.synthesize_procedural, user_id, force=force)
+
     async def close(self) -> None:
         return None
 

agent_memory_toolkit/cosmos_memory_client.py

@@ -1429,6 +1429,68 @@ def remove_tags(self, memory_id: str, user_id: str, thread_id: str, tags: list[s
     # Procedural and episodic memory retrieval
     # ------------------------------------------------------------------
 
+    def get_procedural_prompt(self, user_id: str) -> Optional[str]:
+        """Return the active synthesized procedural prompt for a user."""
+        self._require_cosmos()
+
+        qb = _QueryBuilder()
+        qb.add_filter("c.user_id", "@user_id", user_id)
+        qb.add_filter("c.thread_id", "@thread_id", "__procedural__")
+        qb.add_filter("c.type", "@type", "procedural")
+        qb.add_is_null_or_undefined("c.superseded_by")
+
+        query = f"SELECT TOP 1 c.content, c.version FROM c{qb.build_where()} ORDER BY c.version DESC"
+        try:
+            items = list(
+                self._container_client.query_items(
+                    query=query,
+                    parameters=qb.get_parameters(),
+                    enable_cross_partition_query=True,
+                )
+            )
+        except Exception as exc:
+            raise CosmosOperationError(f"get_procedural_prompt query failed: {exc}") from exc
+
+        if not items:
+            return None
+        return items[0].get("content")
+
+    def get_procedural_history(self, user_id: str, limit: int = 10) -> list[dict[str, Any]]:
+        """Return synthesized procedural docs for a user, newest first."""
+        self._require_cosmos()
+        if limit <= 0:
+            return []
+
+        qb = _QueryBuilder()
+        qb.add_filter("c.user_id", "@user_id", user_id)
+        qb.add_filter("c.thread_id", "@thread_id", "__procedural__")
+        qb.add_filter("c.type", "@type", "procedural")
+
+        query = f"SELECT * FROM c{qb.build_where()} ORDER BY c.version DESC"
+        try:
+            items = list(
+                self._container_client.query_items(
+                    query=query,
+                    parameters=qb.get_parameters(),
+                    enable_cross_partition_query=True,
+                )
+            )
+        except Exception as exc:
+            raise CosmosOperationError(f"get_procedural_history query failed: {exc}") from exc
+
+        def _is_active(doc: dict[str, Any]) -> bool:
+            return not doc.get("superseded_by")
+
+        items.sort(
+            key=lambda doc: (
+                1 if _is_active(doc) else 0,
+                int(doc.get("version") or 0),
+                int(doc.get("_ts") or 0),
+            ),
+            reverse=True,
+        )
+        return items[:limit]
+
     def get_procedural_memories(
         self,
         user_id: str,
@@ -1496,14 +1558,7 @@ def search_episodic_memories(
 
     def build_procedural_context(self, user_id: str) -> str:
         """Build formatted text for system prompt injection."""
-        memories = self.get_procedural_memories(user_id)
-        if not memories:
-            return ""
-        lines = ["## Learned User Preferences"]
-        for m in memories:
-            priority = m.get("metadata", {}).get("priority", "should")
-            lines.append(f"- {m['content']} [{priority}]")
-        return "\n".join(lines)
+        return self.get_procedural_prompt(user_id) or ""
 
     def build_episodic_context(self, user_id: str, query: str, top_k: int = 3) -> str:
         """Build formatted context of relevant past experiences."""
@@ -1527,10 +1582,38 @@ def extract_memories(
         thread_id: str,
         recent_k: Optional[int] = None,
     ) -> dict[str, int]:
-        """Extract facts, procedural, and episodic memories from a thread."""
+        """Extract facts and episodic memories from a thread."""
         self._require_cosmos()
         return self._pipeline.extract_memories(user_id, thread_id, recent_k)
 
+    def synthesize_procedural(
+        self,
+        user_id: str,
+        *,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """Trigger synthesized procedural prompt generation for a user.
+
+        For DurableFunctionProcessor this returns a deferred status; synthesis
+        is auto-driven inside the Function App. ``force=True`` is only honored
+        by InProcessProcessor.
+        """
+        self._require_cosmos()
+        processor = self._get_processor()
+        if not isinstance(processor, InProcessProcessor):
+            logger.debug("synthesize_procedural deferred to Function App auto-trigger user_id=%s", user_id)
+            return {
+                "status": "deferred",
+                "reason": "durable_auto_trigger",
+                "message": (
+                    "Procedural synthesis runs reactively in the Function App after each "
+                    "ExtractMemoriesOrchestrator pass. Use get_procedural_prompt() to read "
+                    "the synthesized prompt once it has been generated."
+                ),
+            }
+
+        return processor.synthesize_procedural(user_id=user_id, force=force)
+
     def generate_thread_summary(
         self,
         user_id: str,

agent_memory_toolkit/models.py

@@ -95,7 +95,7 @@ class MemoryRecord(BaseModel):
     confidence: Optional[float] = None
     content_hash: Optional[str] = None
     superseded_by: Optional[str] = None
-    supersede_reason: Optional[Literal["duplicate", "contradiction", "update"]] = None
+    supersede_reason: Optional[Literal["duplicate", "contradict", "update"]] = None
     superseded_at: Optional[str] = None
     supersedes_ids: list[str] = Field(default_factory=list)
     source_memory_ids: list[str] = Field(default_factory=list)