resolving comments

Author: Aayush Kataria

Docs/public_api.md

@@ -14,10 +14,10 @@
 
 ### Connection
 
-- `__init__(cosmos_endpoint=None, cosmos_credential=None, cosmos_key=None, cosmos_database=None, cosmos_container=None, cosmos_counter_container=None, cosmos_lease_container=None, cosmos_throughput_mode=None, cosmos_autoscale_max_ru=None, ai_foundry_endpoint=None, ai_foundry_credential=None, ai_foundry_api_key=None, embedding_deployment_name='text-embedding-3-large', embedding_dimensions=None, chat_deployment_name='gpt-4o-mini', use_default_credential=True, processor=None) -> None` — configure local state, model clients, optional Cosmos auto-connect, and optional processing backend.
+- `__init__(cosmos_endpoint=None, cosmos_credential=None, cosmos_key=None, cosmos_database=None, cosmos_container=None, cosmos_turns_container=None, cosmos_counter_container=None, cosmos_lease_container=None, cosmos_throughput_mode=None, cosmos_autoscale_max_ru=None, ai_foundry_endpoint=None, ai_foundry_credential=None, ai_foundry_api_key=None, embedding_deployment_name='text-embedding-3-large', embedding_dimensions=None, chat_deployment_name='gpt-4o-mini', use_default_credential=True, processor=None) -> None` — configure local state, model clients, optional Cosmos auto-connect, and optional processing backend. When `cosmos_turns_container` is set, turn-type documents land in a dedicated container so the main `memories` container only fires the Durable change-feed trigger for processed memory writes.
 - `close() -> None` — close Cosmos/model clients and owned credentials.
-- `connect_cosmos(endpoint=None, credential=None, key=None, database=None, container=None) -> None` — connect to an existing memory container.
-- `create_memory_store(database=None, container=None, counter_container=None, lease_container=None, endpoint=None, credential=None, key=None, embedding_dimensions=None, embedding_data_type=None, distance_function=None, full_text_language=None, throughput_mode=None, autoscale_max_ru=None) -> None` — create/connect the memory, counter, and lease containers.
+- `connect_cosmos(endpoint=None, credential=None, key=None, database=None, container=None, turns_container=None) -> None` — connect to an existing memory container.
+- `create_memory_store(database=None, container=None, turns_container=None, counter_container=None, lease_container=None, endpoint=None, credential=None, key=None, embedding_dimensions=None, embedding_data_type=None, distance_function=None, full_text_language=None, throughput_mode=None, autoscale_max_ru=None) -> None` — create/connect the memory, optional turns, counter, and lease containers.
 
 ### Memory CRUD
 
@@ -65,10 +65,10 @@ Local-buffer methods remain synchronous in-memory operations; Cosmos, retrieval,
 
 ### Connection
 
-- `__init__(cosmos_endpoint=None, cosmos_credential=None, cosmos_key=None, cosmos_database=None, cosmos_container=None, cosmos_counter_container=None, cosmos_lease_container=None, cosmos_throughput_mode=None, cosmos_autoscale_max_ru=None, ai_foundry_endpoint=None, ai_foundry_credential=None, ai_foundry_api_key=None, embedding_deployment_name='text-embedding-3-large', embedding_dimensions=None, chat_deployment_name='gpt-4o-mini', use_default_credential=True, processor=None) -> None` — configure async local state, model clients, and optional processing backend.
+- `__init__(cosmos_endpoint=None, cosmos_credential=None, cosmos_key=None, cosmos_database=None, cosmos_container=None, cosmos_turns_container=None, cosmos_counter_container=None, cosmos_lease_container=None, cosmos_throughput_mode=None, cosmos_autoscale_max_ru=None, ai_foundry_endpoint=None, ai_foundry_credential=None, ai_foundry_api_key=None, embedding_deployment_name='text-embedding-3-large', embedding_dimensions=None, chat_deployment_name='gpt-4o-mini', use_default_credential=True, processor=None) -> None` — configure async local state, model clients, and optional processing backend. When `cosmos_turns_container` is set, turn-type documents land in a dedicated container so the main `memories` container only fires the Durable change-feed trigger for processed memory writes.
 - `async close() -> None` — close async/sync resources and owned credentials.
-- `async connect_cosmos(endpoint=None, credential=None, key=None, database=None, container=None) -> None` — connect to an existing memory container.
-- `async create_memory_store(database=None, container=None, counter_container=None, lease_container=None, endpoint=None, credential=None, key=None, embedding_dimensions=None, embedding_data_type=None, distance_function=None, full_text_language=None, throughput_mode=None, autoscale_max_ru=None) -> None` — create/connect memory, counter, and lease containers.
+- `async connect_cosmos(endpoint=None, credential=None, key=None, database=None, container=None, turns_container=None) -> None` — connect to an existing memory container.
+- `async create_memory_store(database=None, container=None, turns_container=None, counter_container=None, lease_container=None, endpoint=None, credential=None, key=None, embedding_dimensions=None, embedding_data_type=None, distance_function=None, full_text_language=None, throughput_mode=None, autoscale_max_ru=None) -> None` — create/connect memory, optional turns, counter, and lease containers.
 
 ### Memory CRUD
 
@@ -119,5 +119,5 @@ Sync extension protocols live in `agent_memory_toolkit.services`; async variants
 Concrete service classes are exported from their respective packages:
 
 - Sync: `RetrievalService`, `PipelineService` from `agent_memory_toolkit.services` (sub-modules `retrieval`, `pipeline`).
-- Async: `AsyncRetrievalService` from `agent_memory_toolkit.aio.services` (sub-module `retrieval`). The pipeline is sync-only and is driven through `asyncio.to_thread` by the async client.
+- Async: `AsyncRetrievalService` and `AsyncPipelineService` from `agent_memory_toolkit.aio.services` (sub-modules `retrieval`, `pipeline`). The async pipeline is a fully-native asyncio implementation — not an `asyncio.to_thread` shim over the sync pipeline.
 - Threshold-driven auto-trigger: `maybe_trigger_steps` from `agent_memory_toolkit.auto_trigger` (sync) and `agent_memory_toolkit.aio.auto_trigger` (async).

agent_memory_toolkit/aio/processors/base.py

@@ -62,6 +62,13 @@ async def generate_user_summary(
         thread_summaries: list[dict[str, Any]],
     ) -> UserSummaryResult: ...
 
+    async def synthesize_procedural(
+        self,
+        *,
+        user_id: str,
+        force: bool = False,
+    ) -> dict[str, Any]: ...
+
     async def close(self) -> None: ...
 
 

agent_memory_toolkit/aio/services/pipeline.py

@@ -64,6 +64,9 @@
 _coerce_valence = coerce_valence
 _cap_structured_summary = cap_structured_summary
 
+_ACTIVE_DOC_FILTER = "(NOT IS_DEFINED(c.superseded_by) OR IS_NULL(c.superseded_by))"
+_PROCEDURAL_MAX_CREATE_ATTEMPTS = 5
+
 
 class _AsyncStoreContainerAdapter:
     """Async equivalent of ``services.pipeline._StoreContainerAdapter``.
@@ -195,7 +198,7 @@ async def _load_existing_memories(
             f"SELECT TOP {capped_limit} * FROM c "
             f"WHERE c.user_id = @user_id "
             f"AND c.type IN ({type_placeholders}) "
-            f"AND (NOT IS_DEFINED(c.superseded_by) OR IS_NULL(c.superseded_by)) "
+            f"AND {_ACTIVE_DOC_FILTER} "
             f"ORDER BY c._ts DESC"
         )
         parameters: list[dict[str, Any]] = [
@@ -277,7 +280,7 @@ async def _mark_extracted_superseded(
         try:
             q = (
                 "SELECT * FROM c WHERE c.id = @id AND c.user_id = @uid "
-                "AND (NOT IS_DEFINED(c.superseded_by) OR IS_NULL(c.superseded_by))"
+                f"AND {_ACTIVE_DOC_FILTER}"
             )
             results = await self._container.query_items(
                 query=q,
@@ -392,8 +395,6 @@ async def extract_memories_dry(
 
         for fact in facts:
             action = fact.get("action", "ADD").upper()
-            if action == "NONE":
-                continue
             text = fact.get("text")
             if not text:
                 logger.warning("extract_memories: dropping malformed fact (missing 'text'): %r", fact)
@@ -665,15 +666,13 @@ async def synthesize_procedural(
 
         logger.info("synthesize_procedural started user_id=%s force=%s", user_id, force)
 
-        active_filter = "(NOT IS_DEFINED(c.superseded_by) OR c.superseded_by = null)"
-
         async def _read_latest_procedural() -> Optional[dict[str, Any]]:
             docs = await self._container.query_items(
                 query=(
                     "SELECT * FROM c WHERE c.user_id = @uid "
                     "AND c.thread_id = @thread_id "
                     "AND c.type = @type "
-                    f"AND {active_filter}"
+                    f"AND {_ACTIVE_DOC_FILTER}"
                 ),
                 parameters=[
                     {"name": "@uid", "value": user_id},
@@ -700,7 +699,7 @@ async def _read_latest_procedural() -> Optional[dict[str, Any]]:
             query=(
                 "SELECT TOP 50 * FROM c WHERE c.user_id = @uid "
                 "AND c.type = @type "
-                f"AND {active_filter} "
+                f"AND {_ACTIVE_DOC_FILTER} "
                 "AND ((IS_DEFINED(c.metadata.category) "
                 "AND c.metadata.category IN ('preference', 'requirement')) "
                 "OR (IS_DEFINED(c.salience) AND c.salience >= @min_salience)) "
@@ -724,7 +723,7 @@ async def _read_latest_procedural() -> Optional[dict[str, Any]]:
             query=(
                 "SELECT TOP 50 * FROM c WHERE c.user_id = @uid "
                 "AND c.type = @type "
-                f"AND {active_filter} "
+                f"AND {_ACTIVE_DOC_FILTER} "
                 "AND IS_DEFINED(c.metadata.lesson) "
                 "AND c.metadata.lesson != null "
                 "ORDER BY c.salience DESC, c.created_at ASC, c.id ASC"
@@ -744,12 +743,14 @@ async def _read_latest_procedural() -> Optional[dict[str, Any]]:
         source_episodic_ids = [doc["id"] for doc in episodic_with_lessons]
 
         current_source_ids = set(behavioral_fact_ids) | set(source_episodic_ids)
-        prior_source_ids = (
-            set(prior_doc.get("source_fact_ids") or []) | set(prior_doc.get("source_episodic_ids") or [])
-            if prior_doc
-            else set()
-        )
-        if prior_doc and not force and current_source_ids == prior_source_ids:
+
+        def _covered_by(prior: Optional[dict[str, Any]]) -> bool:
+            if prior is None:
+                return False
+            covered = set(prior.get("source_fact_ids") or []) | set(prior.get("source_episodic_ids") or [])
+            return current_source_ids.issubset(covered)
+
+        if prior_doc and not force and _covered_by(prior_doc):
             logger.info(
                 "synthesize_procedural unchanged user_id=%s fact_count=%d episodic_count=%d",
                 user_id,
@@ -769,7 +770,7 @@ async def _read_latest_procedural() -> Optional[dict[str, Any]]:
             query=(
                 "SELECT TOP 1 * FROM c WHERE c.user_id = @uid "
                 "AND c.type = @type "
-                f"AND {active_filter} "
+                f"AND {_ACTIVE_DOC_FILTER} "
                 "AND IS_DEFINED(c.metadata.category) "
                 "AND c.metadata.category = @category "
                 "AND IS_DEFINED(c.metadata.predicate) "
@@ -799,58 +800,53 @@ def _render_bullets(values: list[str]) -> str:
                 return "(none)"
             return "\n".join(f"- {value}" for value in cleaned)
 
-        response_text = await self._run_prompty(
-            "synthesize_procedural.prompty",
-            inputs={
-                "prior_prompt": (prior_doc.get("content") or "") if prior_doc else "",
-                "behavioral_facts": _render_bullets([doc.get("content", "") for doc in behavioral_fact_docs]),
-                "episodic_lessons": _render_bullets(
-                    [doc.get("metadata", {}).get("lesson", "") for doc in episodic_with_lessons]
-                ),
-                "user_name": user_name,
-            },
-        )
-
-        parsed = self._parse_llm_json(response_text)
-        system_prompt = parsed.get("system_prompt") if isinstance(parsed, dict) else None
-        if not isinstance(system_prompt, str) or not system_prompt.strip():
-            raise LLMError("synthesize_procedural returned JSON without a non-empty 'system_prompt' string")
-        system_prompt = system_prompt.strip()
-
-        new_seq = (int(prior_doc.get("version") or 0) + 1) if prior_doc else 1
-        new_doc: dict[str, Any] = {
-            "id": f"proc_{user_id}_{new_seq}",
-            "user_id": user_id,
-            "thread_id": "__procedural__",
-            "type": "procedural",
-            "version": new_seq,
-            "content": system_prompt,
-            "source_fact_ids": behavioral_fact_ids,
-            "source_episodic_ids": source_episodic_ids,
-            "supersedes_ids": [prior_doc["id"]] if prior_doc else [],
-            "created_at": datetime.now(timezone.utc).isoformat(),
-            "updated_at": datetime.now(timezone.utc).isoformat(),
-            "role": "system",
-            "tags": ["sys:procedural", "sys:synthesized"],
-            **self._prompt_lineage("synthesize_procedural.prompty"),
-            "metadata": {},
+        static_prompty_inputs = {
+            "behavioral_facts": _render_bullets([doc.get("content", "") for doc in behavioral_fact_docs]),
+            "episodic_lessons": _render_bullets(
+                [doc.get("metadata", {}).get("lesson", "") for doc in episodic_with_lessons]
+            ),
+            "user_name": user_name,
         }
-        if not new_doc["source_fact_ids"] and not new_doc["source_episodic_ids"]:
-            logger.info(
-                "synthesize_procedural skipping write user_id=%s — no source facts or episodics",
-                user_id,
-            )
-            return {"status": "unchanged", "procedural": prior_doc}
 
-        max_attempts = 5
-        new_id = new_doc["id"]
+        # Retry loop: LLM call lives inside so that on a race-induced 409
+        # we (a) check whether the winner already covers our source set and
+        # short-circuit if so, and (b) re-call the LLM with the winner as
+        # the new prior if not — keeping synthesized content monotonic in
+        # source coverage, not just version number.
         written_doc: Optional[dict[str, Any]] = None
-        for attempt in range(1, max_attempts + 1):
-            new_doc["id"] = f"proc_{user_id}_{new_seq}"
-            new_doc["version"] = new_seq
-            new_doc["supersedes_ids"] = [prior_doc["id"]] if prior_doc else []
-            new_doc["updated_at"] = datetime.now(timezone.utc).isoformat()
-            new_id = new_doc["id"]
+        for attempt in range(1, _PROCEDURAL_MAX_CREATE_ATTEMPTS + 1):
+            response_text = await self._run_prompty(
+                "synthesize_procedural.prompty",
+                inputs={
+                    "prior_prompt": (prior_doc.get("content") or "") if prior_doc else "",
+                    **static_prompty_inputs,
+                },
+            )
+
+            parsed = self._parse_llm_json(response_text)
+            system_prompt = parsed.get("system_prompt") if isinstance(parsed, dict) else None
+            if not isinstance(system_prompt, str) or not system_prompt.strip():
+                raise LLMError("synthesize_procedural returned JSON without a non-empty 'system_prompt' string")
+            system_prompt = system_prompt.strip()
+
+            new_seq = (int(prior_doc.get("version") or 0) + 1) if prior_doc else 1
+            new_doc: dict[str, Any] = {
+                "id": f"proc_{user_id}_{new_seq}",
+                "user_id": user_id,
+                "thread_id": "__procedural__",
+                "type": "procedural",
+                "version": new_seq,
+                "content": system_prompt,
+                "source_fact_ids": behavioral_fact_ids,
+                "source_episodic_ids": source_episodic_ids,
+                "supersedes_ids": [prior_doc["id"]] if prior_doc else [],
+                "created_at": datetime.now(timezone.utc).isoformat(),
+                "updated_at": datetime.now(timezone.utc).isoformat(),
+                "role": "system",
+                "tags": ["sys:procedural", "sys:synthesized"],
+                **self._prompt_lineage("synthesize_procedural.prompty"),
+                "metadata": {},
+            }
             validated = construct_internal(ProceduralRecord, new_doc).to_doc()
             try:
                 await self._container.create_item(body=validated)
@@ -862,26 +858,34 @@ def _render_bullets(values: list[str]) -> str:
                     user_id,
                     new_seq,
                     attempt,
-                    max_attempts,
+                    _PROCEDURAL_MAX_CREATE_ATTEMPTS,
                 )
                 latest = await _read_latest_procedural()
-                if latest is not None:
-                    prior_doc = latest
-                    new_seq = int(latest.get("version") or 0) + 1
-                else:
-                    new_seq += 1
+                if latest is None:
+                    continue
+                prior_doc = latest
+                if _covered_by(prior_doc):
+                    logger.info(
+                        "synthesize_procedural race resolved by coverage user_id=%s winner=%s",
+                        user_id,
+                        prior_doc["id"],
+                    )
+                    return {"status": "unchanged", "procedural": prior_doc}
         if written_doc is None:
             raise MemoryConflictError(
-                f"synthesize_procedural failed after {max_attempts} attempts due to id collisions user_id={user_id!r}"
+                "synthesize_procedural failed after "
+                f"{_PROCEDURAL_MAX_CREATE_ATTEMPTS} attempts due to id collisions "
+                f"user_id={user_id!r}"
             )
 
+        new_id = written_doc["id"]
         if prior_doc:
             await self._mark_superseded(prior_doc, new_id, reason="update")
 
         logger.info(
             "synthesize_procedural synthesized user_id=%s version=%d fact_count=%d episodic_count=%d",
             user_id,
-            new_seq,
+            written_doc["version"],
             len(behavioral_fact_ids),
             len(source_episodic_ids),
         )
@@ -1238,7 +1242,7 @@ async def reconcile_memories(self, user_id: str, n: int = 50) -> dict[str, int]:
             f"SELECT TOP {n} * FROM c "
             "WHERE c.user_id = @user_id "
             "AND c.type = 'fact' "
-            "AND (NOT IS_DEFINED(c.superseded_by) OR IS_NULL(c.superseded_by)) "
+            f"AND {_ACTIVE_DOC_FILTER} "
             "ORDER BY c.created_at DESC"
         )
         parameters: list[dict[str, Any]] = [
@@ -1643,7 +1647,7 @@ async def build_procedural_context(self, user_id: str) -> str:
         query = (
             "SELECT TOP 1 c.content, c.version FROM c WHERE c.user_id = @user_id "
             "AND c.thread_id = @thread_id AND c.type = @type "
-            "AND (NOT IS_DEFINED(c.superseded_by) OR IS_NULL(c.superseded_by)) "
+            f"AND {_ACTIVE_DOC_FILTER} "
             "ORDER BY c.version DESC"
         )
         items = await self._store.query(

agent_memory_toolkit/processors/base.py

@@ -99,6 +99,13 @@ def generate_user_summary(
         thread_summaries: list[dict[str, Any]],
     ) -> UserSummaryResult: ...
 
+    def synthesize_procedural(
+        self,
+        *,
+        user_id: str,
+        force: bool = False,
+    ) -> dict[str, Any]: ...
+
     def close(self) -> None: ...