resolving comments

Author: Aayush Kataria

Docs/azure_testing.md

@@ -120,7 +120,7 @@ az functionapp config appsettings set \
     MEMORY_PROCESSOR_OWNER="durable"
 ```
 
-`COSMOS_DB_THROUGHPUT_MODE=serverless` is the default and creates the `memories`, `counter`, and `leases` containers without specifying RU/s. Set `COSMOS_DB_THROUGHPUT_MODE=autoscale` to apply the shared `COSMOS_DB_AUTOSCALE_MAX_RU` cap to all required containers.
+`COSMOS_DB_THROUGHPUT_MODE=serverless` is the default and creates the `memories`, `memories_turns`, `memories_summaries`, `counter`, and `leases` containers without specifying RU/s. Set `COSMOS_DB_THROUGHPUT_MODE=autoscale` to apply the shared `COSMOS_DB_AUTOSCALE_MAX_RU` cap to all required containers.
 
 `MEMORY_PROCESSOR_OWNER=durable` tells the SDK that the deployed Function App owns processing, so any `CosmosMemoryClient` pointed at the same container will skip its in-process auto-trigger and avoid double-extraction. See the README's processor-ownership table for details.
 
@@ -258,7 +258,7 @@ await memory.connect_cosmos()
 await memory.create_memory_store()
 ```
 
-This provisions the `memories`, `counter`, and `leases` containers. `serverless` is the default throughput mode; if you set `COSMOS_DB_THROUGHPUT_MODE=autoscale`, the shared `COSMOS_DB_AUTOSCALE_MAX_RU` value is applied to all three containers.
+This provisions the `memories`, `memories_turns`, `memories_summaries`, `counter`, and `leases` containers. `serverless` is the default throughput mode; if you set `COSMOS_DB_THROUGHPUT_MODE=autoscale`, the shared `COSMOS_DB_AUTOSCALE_MAX_RU` value is applied to all five containers.
 
 ---
 

Docs/concepts.md

@@ -187,8 +187,8 @@ Set any value to `0` to disable that processing type. For example, setting `THRE
 
 The toolkit provisions all required Cosmos containers under one shared throughput mode:
 
-- `serverless` is the default. The toolkit creates the `memories`, `counter`, and `leases` containers without specifying RU/s.
-- `autoscale` applies the shared `COSMOS_DB_AUTOSCALE_MAX_RU` cap to all three containers.
+- `serverless` is the default. The toolkit creates the `memories`, `memories_turns`, `memories_summaries`, `counter`, and `leases` containers without specifying RU/s.
+- `autoscale` applies the shared `COSMOS_DB_AUTOSCALE_MAX_RU` cap to all five containers.
 
 This keeps the change feed dependencies aligned with the main memory store instead of letting the Functions trigger create the lease container independently.
 

Docs/troubleshooting.md

@@ -94,7 +94,7 @@ RBAC changes can take several minutes to propagate.
 
 ## 3. Cosmos DB Store Creation
 
-Run `create_memory_store()` before relying on cloud operations. It creates the database plus the `memories`, `counter`, and `leases` containers.
+Run `create_memory_store()` before relying on cloud operations. It creates the database plus the `memories`, `memories_turns`, `memories_summaries`, `counter`, and `leases` containers.
 
 The memories container is created with:
 

agent_memory_toolkit/__init__.py

@@ -11,6 +11,7 @@
     LLMError,
     MemoryConflictError,
     MemoryNotFoundError,
+    MemoryTypeMismatchError,
     ValidationError,
 )
 from agent_memory_toolkit.models import MemoryRecord, MemoryRole, MemoryType, SearchResult
@@ -53,6 +54,7 @@
     "LLMError",
     "MemoryConflictError",
     "MemoryNotFoundError",
+    "MemoryTypeMismatchError",
     "ValidationError",
     "DEFAULT_FACT_EXTRACTION_EVERY_N",
     "DEFAULT_THREAD_SUMMARY_EVERY_N",

agent_memory_toolkit/aio/store/memory_store.py

@@ -8,6 +8,7 @@
 from typing import Any, Optional
 
 from agent_memory_toolkit._container_routing import (
+    _CONTAINER_FOR_TYPE,
     ContainerKey,
     container_key_for_type,
 )
@@ -24,6 +25,7 @@
     CosmosOperationError,
     MemoryConflictError,
     MemoryNotFoundError,
+    MemoryTypeMismatchError,
 )
 from agent_memory_toolkit.logging import get_logger
 from agent_memory_toolkit.models import MemoryRecord
@@ -131,7 +133,13 @@ async def _query_items(
     async def add_cosmos(self, record: dict[str, Any]) -> dict[str, Any]:
         """Upsert a pre-built Cosmos memory document and return the stored body."""
         body = self._prepare_doc(record)
-        container = self._container_for_type(body.get("type"))
+        memory_type = body.get("type")
+        if memory_type not in _CONTAINER_FOR_TYPE:
+            raise ValueError(
+                f"add_cosmos: record id={body.get('id')!r} has invalid type={memory_type!r}. "
+                f"Set 'type' to one of {sorted(_CONTAINER_FOR_TYPE)} before calling add_cosmos."
+            )
+        container = self._container_for_type(memory_type)
         try:
             response = await container.upsert_item(body=body)
         except Exception as exc:
@@ -251,6 +259,14 @@ async def push(self, local_memory: list[dict[str, Any]], batch_size: int = 25) -
                     )
 
             bodies = [self._prepare_doc(body) for body in bodies]
+            for body in bodies:
+                memory_type = body.get("type")
+                if memory_type not in _CONTAINER_FOR_TYPE:
+                    raise ValueError(
+                        f"push: record id={body.get('id')!r} has invalid type={memory_type!r}. "
+                        f"Set 'type' to one of {sorted(_CONTAINER_FOR_TYPE)} on every local "
+                        f"memory before calling push_to_cosmos."
+                    )
             tasks = [self._container_for_type(b.get("type")).upsert_item(body=b) for b in bodies]
             try:
                 await asyncio.gather(*tasks)
@@ -355,6 +371,10 @@ async def update(
         except Exception as exc:
             raise _wrap_cosmos_exception(exc, message=f"async update read failed for {memory_id}: {exc}") from exc
 
+        actual_type = doc.get("type")
+        if actual_type != memory_type:
+            raise MemoryTypeMismatchError(memory_id=memory_id, expected=memory_type, actual=actual_type)
+
         if content is not None:
             doc["content"] = content
         if role is not None:
@@ -378,10 +398,26 @@ async def delete(
         thread_id: str,
         memory_type: str,
     ) -> None:
-        """Delete a memory document from the container for ``memory_type``."""
+        """Delete a memory document from the container for ``memory_type``.
+
+        Reads the doc first to verify its ``type`` matches ``memory_type`` so a
+        wrong routing key cannot silently delete the wrong document (within
+        MEMORIES, fact/episodic/procedural share the same partition key).
+        """
         from azure.cosmos.exceptions import CosmosResourceNotFoundError
 
         container = self._container_for_type(memory_type)
+        try:
+            doc = await container.read_item(item=memory_id, partition_key=[user_id, thread_id])
+        except CosmosResourceNotFoundError as exc:
+            raise MemoryNotFoundError(memory_id=memory_id, user_id=user_id, thread_id=thread_id) from exc
+        except Exception as exc:
+            raise _wrap_cosmos_exception(exc, message=f"async delete read failed for {memory_id}: {exc}") from exc
+
+        actual_type = doc.get("type")
+        if actual_type != memory_type:
+            raise MemoryTypeMismatchError(memory_id=memory_id, expected=memory_type, actual=actual_type)
+
         try:
             await container.delete_item(item=memory_id, partition_key=[user_id, thread_id])
         except CosmosResourceNotFoundError as exc:
@@ -595,7 +631,10 @@ async def mark_superseded(
     ) -> bool:
         """Set supersession audit fields using ETag protection when available."""
         from azure.core import MatchConditions
-        from azure.cosmos.exceptions import CosmosAccessConditionFailedError
+        from azure.cosmos.exceptions import (
+            CosmosAccessConditionFailedError,
+            CosmosHttpResponseError,
+        )
 
         etag = old_doc.get("_etag")
         new_doc = {
@@ -625,8 +664,12 @@ async def mark_superseded(
             )
             del exc
             return False
-        except Exception:
-            logger.exception("supersede failed id=%s superseder=%s", old_doc.get("id"), superseder_id)
+        except CosmosHttpResponseError:
+            logger.exception(
+                "supersede transient failure id=%s superseder=%s",
+                old_doc.get("id"),
+                superseder_id,
+            )
             return False
 
     async def get_procedural_prompt(self, user_id: str) -> Optional[str]:

agent_memory_toolkit/exceptions.py

@@ -103,6 +103,33 @@ def _build_message(self) -> str:
         return "Memory not found"
 
 
+class MemoryTypeMismatchError(AgentMemoryError):
+    """Raised when an operation's ``memory_type`` argument does not match the
+    actual type of the stored document. Prevents silent cross-type mutation
+    or deletion when callers pass the wrong routing key.
+    """
+
+    error_code = "memory_type_mismatch"
+
+    def __init__(
+        self,
+        message: str | None = None,
+        *,
+        memory_id: str | None = None,
+        expected: str | None = None,
+        actual: str | None = None,
+    ):
+        self.memory_id = memory_id
+        self.expected = expected
+        self.actual = actual
+        if message is None:
+            message = (
+                f"Memory type mismatch for memory_id={memory_id!r}: "
+                f"caller passed memory_type={expected!r}, actual stored type is {actual!r}."
+            )
+        super().__init__(message)
+
+
 class LLMError(AgentMemoryError):
     """Raised when the LLM returns a response shape the SDK does not surface
     as an exception (no choices, empty content, invalid JSON)."""

agent_memory_toolkit/store/memory_store.py

@@ -6,6 +6,7 @@
 from typing import Any, Optional
 
 from agent_memory_toolkit._container_routing import (
+    _CONTAINER_FOR_TYPE,
     ContainerKey,
     container_key_for_type,
 )
@@ -22,6 +23,7 @@
     CosmosOperationError,
     MemoryConflictError,
     MemoryNotFoundError,
+    MemoryTypeMismatchError,
 )
 from agent_memory_toolkit.logging import get_logger
 from agent_memory_toolkit.models import MemoryRecord
@@ -148,7 +150,13 @@ def _query_items(
     def add_cosmos(self, record: dict[str, Any]) -> dict[str, Any]:
         """Upsert a pre-built Cosmos memory document and return the stored body."""
         body = self._prepare_doc(record)
-        container = self._container_for_type(body.get("type"))
+        memory_type = body.get("type")
+        if memory_type not in _CONTAINER_FOR_TYPE:
+            raise ValueError(
+                f"add_cosmos: record id={body.get('id')!r} has invalid type={memory_type!r}. "
+                f"Set 'type' to one of {sorted(_CONTAINER_FOR_TYPE)} before calling add_cosmos."
+            )
+        container = self._container_for_type(memory_type)
         try:
             response = container.upsert_item(body=body)
         except Exception as exc:
@@ -265,7 +273,14 @@ def push(self, local_memory: list[dict[str, Any]], batch_size: int = 25) -> None
 
             bodies = [self._prepare_doc(body) for body in bodies]
             for record, body in zip(batch, bodies):
-                container = self._container_for_type(body.get("type"))
+                memory_type = body.get("type")
+                if memory_type not in _CONTAINER_FOR_TYPE:
+                    raise ValueError(
+                        f"push: record id={body.get('id')!r} has invalid type={memory_type!r}. "
+                        f"Set 'type' to one of {sorted(_CONTAINER_FOR_TYPE)} on every local "
+                        f"memory before calling push_to_cosmos."
+                    )
+                container = self._container_for_type(memory_type)
                 try:
                     container.upsert_item(body=body)
                 except Exception as exc:
@@ -370,6 +385,10 @@ def update(
         except Exception as exc:
             raise _wrap_cosmos_exception(exc, message=f"update read failed for {memory_id}: {exc}") from exc
 
+        actual_type = doc.get("type")
+        if actual_type != memory_type:
+            raise MemoryTypeMismatchError(memory_id=memory_id, expected=memory_type, actual=actual_type)
+
         if content is not None:
             doc["content"] = content
         if role is not None:
@@ -393,10 +412,26 @@ def delete(
         thread_id: str,
         memory_type: str,
     ) -> None:
-        """Delete a memory document from the container for ``memory_type``."""
+        """Delete a memory document from the container for ``memory_type``.
+
+        Reads the doc first to verify its ``type`` matches ``memory_type`` so a
+        wrong routing key cannot silently delete the wrong document (within
+        MEMORIES, fact/episodic/procedural share the same partition key).
+        """
         from azure.cosmos.exceptions import CosmosResourceNotFoundError
 
         container = self._container_for_type(memory_type)
+        try:
+            doc = container.read_item(item=memory_id, partition_key=[user_id, thread_id])
+        except CosmosResourceNotFoundError as exc:
+            raise MemoryNotFoundError(memory_id=memory_id, user_id=user_id, thread_id=thread_id) from exc
+        except Exception as exc:
+            raise _wrap_cosmos_exception(exc, message=f"delete read failed for {memory_id}: {exc}") from exc
+
+        actual_type = doc.get("type")
+        if actual_type != memory_type:
+            raise MemoryTypeMismatchError(memory_id=memory_id, expected=memory_type, actual=actual_type)
+
         try:
             container.delete_item(item=memory_id, partition_key=[user_id, thread_id])
         except CosmosResourceNotFoundError as exc:
@@ -612,7 +647,10 @@ def mark_superseded(
     ) -> bool:
         """Set supersession audit fields using ETag protection when available."""
         from azure.core import MatchConditions
-        from azure.cosmos.exceptions import CosmosAccessConditionFailedError
+        from azure.cosmos.exceptions import (
+            CosmosAccessConditionFailedError,
+            CosmosHttpResponseError,
+        )
 
         etag = old_doc.get("_etag")
         new_doc = {
@@ -642,8 +680,12 @@ def mark_superseded(
             )
             del exc
             return False
-        except Exception:
-            logger.exception("supersede failed id=%s superseder=%s", old_doc.get("id"), superseder_id)
+        except CosmosHttpResponseError:
+            logger.exception(
+                "supersede transient failure id=%s superseder=%s",
+                old_doc.get("id"),
+                superseder_id,
+            )
             return False
 
     def get_procedural_prompt(self, user_id: str) -> Optional[str]:

infra/README.md

@@ -78,6 +78,12 @@ azd provision   # skips Function app, Storage, App Insights, Log Analytics, stor
 
 > Use **`azd provision`** (not `azd up`) for SDK-only. `azd up` always invokes `azd deploy --all`, which looks for a resource tagged `azd-service-name: function_app`. With the Function app turned off there is no such resource and the deploy step fails. `azd provision` runs Bicep without the deploy step.
 
+When `DEPLOY_FUNCTION_APP=false`, the Bicep template automatically sets `MEMORY_PROCESSOR_OWNER=inprocess` in the generated `.env` so the in-process processor's auto-trigger fires. If you previously deployed with `DEPLOY_FUNCTION_APP=true` and are switching to SDK-only mode, also clear the override explicitly:
+
+```bash
+azd env set MEMORY_PROCESSOR_OWNER inprocess
+```
+
 ## Model / deployment names
 
 Two concepts kept separate:

infra/main.bicep

@@ -266,4 +266,4 @@ output AZURE_USER_ASSIGNED_IDENTITY_ID string = identity.outputs.id
 output FUNCTION_APP_NAME string = deployFunctionApp ? functions!.outputs.functionAppName : ''
 output FUNCTION_APP_URL string = deployFunctionApp ? functions!.outputs.functionAppUrl : ''
 
-output MEMORY_PROCESSOR_OWNER string = memoryProcessorOwner
+output MEMORY_PROCESSOR_OWNER string = deployFunctionApp ? memoryProcessorOwner : 'inprocess'

tests/unit/aio/store/test_memory_store.py

@@ -7,7 +7,7 @@
 
 from agent_memory_toolkit._container_routing import ContainerKey
 from agent_memory_toolkit.aio.store import AsyncMemoryStore
-from agent_memory_toolkit.exceptions import MemoryNotFoundError
+from agent_memory_toolkit.exceptions import MemoryNotFoundError, MemoryTypeMismatchError
 
 
 class AsyncIterator:
@@ -153,20 +153,21 @@ async def test_update_raises_when_missing():
         await store.update("missing", user_id="u1", thread_id="t1", memory_type="fact")
 
 
-async def test_update_does_not_mutate_type_field():
+async def test_update_raises_on_type_mismatch():
     memories = MagicMock()
     memories.read_item = AsyncMock(return_value=_doc(id="m1", type="fact"))
     memories.replace_item = AsyncMock()
     store = AsyncMemoryStore(containers=_containers(memories=memories))
 
-    await store.update("m1", user_id="u1", thread_id="t1", memory_type="episodic", content="x")
+    with pytest.raises(MemoryTypeMismatchError):
+        await store.update("m1", user_id="u1", thread_id="t1", memory_type="episodic", content="x")
 
-    body = memories.replace_item.call_args.kwargs["body"]
-    assert body["type"] == "fact"
+    memories.replace_item.assert_not_awaited()
 
 
 async def test_delete_calls_delete_item_directly():
     memories = MagicMock()
+    memories.read_item = AsyncMock(return_value=_doc(id="m1", type="fact"))
     memories.delete_item = AsyncMock()
     store = AsyncMemoryStore(containers=_containers(memories=memories))
 
@@ -179,12 +180,27 @@ async def test_delete_raises_when_missing():
     from azure.cosmos.exceptions import CosmosResourceNotFoundError
 
     memories = MagicMock()
-    memories.delete_item = AsyncMock(side_effect=CosmosResourceNotFoundError(message="404"))
+    memories.read_item = AsyncMock(side_effect=CosmosResourceNotFoundError(message="404"))
+    memories.delete_item = AsyncMock()
     store = AsyncMemoryStore(containers=_containers(memories=memories))
 
     with pytest.raises(MemoryNotFoundError):
         await store.delete("m1", user_id="u1", thread_id="t1", memory_type="fact")
 
+    memories.delete_item.assert_not_awaited()
+
+
+async def test_delete_raises_on_type_mismatch():
+    memories = MagicMock()
+    memories.read_item = AsyncMock(return_value=_doc(id="m1", type="fact"))
+    memories.delete_item = AsyncMock()
+    store = AsyncMemoryStore(containers=_containers(memories=memories))
+
+    with pytest.raises(MemoryTypeMismatchError):
+        await store.delete("m1", user_id="u1", thread_id="t1", memory_type="episodic")
+
+    memories.delete_item.assert_not_awaited()
+
 
 async def test_read_and_tag_mutation_use_point_reads():
     memories = MagicMock()