Fix Durable user-summary extraction race (#40), add Function-app import guard (#39), and remove dead vector-dedup scaffolding

.github/workflows/_smoke.yml

@@ -0,0 +1,42 @@
+name: smoke
+
+on:
+  workflow_call:
+
+permissions:
+  contents: read
+
+jobs:
+  function-app-import:
+    name: "function app import #${{ matrix.python-version }}"
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        # The Function app deploys on the Flex Consumption Python 3.11 runtime.
+        python-version: ["3.11"]
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+          cache-dependency-path: pyproject.toml
+
+      # Install the toolkit from this same commit so the Function app is
+      # validated against the source it ships with, rather than a possibly-older
+      # published package. The toolkit pinned in requirements.txt is then already
+      # satisfied and is not re-fetched.
+      - name: Install toolkit from source
+        run: pip install .
+
+      - name: Install Function app requirements
+        working-directory: function_app
+        run: pip install -r requirements.txt
+
+      # Fails if the Function source uses a toolkit API the in-repo toolkit does
+      # not provide - the drift that left the host unable to load the app.
+      - name: Import function_app
+        working-directory: function_app
+        run: PYTHONPATH=. python -c "import function_app; print('import function_app OK')"

.github/workflows/ci.yml

@@ -25,12 +25,17 @@ jobs:
     permissions:
       contents: read
 
+  smoke:
+    uses: ./.github/workflows/_smoke.yml
+    permissions:
+      contents: read
+
   ci-success:
     # Single required-status-check target. Branch protection should require
     # only this job; it summarizes lint + test matrix results so a failure
     # in any matrix cell blocks merge.
     name: "CI Success"
-    needs: [lint, test]
+    needs: [lint, test, smoke]
     if: always()
     runs-on: ubuntu-latest
     env:

azure/cosmos/agent_memory/_utils.py

@@ -341,9 +341,6 @@ def _resolve_vector_index_type(val: Optional[str]) -> str:
     return raw
 
 
-_SIMILARITY_DESCENDING_FUNCTIONS = frozenset({"cosine", "dotproduct"})
-
-
 def cosine_similarity(a: list[float], b: list[float]) -> float:
     """Cosine similarity of two equal-length vectors.
 
@@ -386,35 +383,6 @@ def vector_centroid(vectors: list[list[float]]) -> list[float]:
     return [value / count for value in acc]
 
 
-def vector_order_direction(distance_function: str) -> str:
-    """Return the ``ORDER BY VectorDistance(...)`` direction for most-similar-first.
-
-    ``DESC`` for cosine/dotproduct (higher score = more similar), ``ASC`` for
-    euclidean (lower distance = more similar).
-    """
-    return "DESC" if distance_function in _SIMILARITY_DESCENDING_FUNCTIONS else "ASC"
-
-
-def distance_function_from_container_properties(props: Any, *, default: str = "cosine") -> str:
-    """Read the vector embedding's ``distanceFunction`` from container properties.
-
-    The distance function (cosine/dotproduct/euclidean) is chosen at
-    ``create_memory_store`` time, written immutably into the container's vector
-    embedding policy, and read back here from the authoritative source
-    (``container.read()``) so the dedup vector-floor logic matches how the
-    container actually ranks. This SDK provisions exactly one vector embedding;
-    falls back to ``default`` (cosine) when the policy is absent or malformed
-    (e.g. ``__new__``-built test instances with mocked containers).
-    """
-    policy = props.get("vectorEmbeddingPolicy") if isinstance(props, dict) else None
-    embeddings = policy.get("vectorEmbeddings") if isinstance(policy, dict) else None
-    entry = embeddings[0] if isinstance(embeddings, list) and embeddings else None
-    fn = entry.get("distanceFunction") if isinstance(entry, dict) else None
-    if isinstance(fn, str) and fn in _ALLOWED_DISTANCE_FUNCTIONS:
-        return fn
-    return default
-
-
 def _resolve_full_text_language(val: Optional[str]) -> str:
     """Resolve full-text language from the explicit value, defaulting to ``en-US``.
 

azure/cosmos/agent_memory/aio/services/pipeline.py

@@ -29,11 +29,10 @@
 from azure.cosmos.agent_memory._utils import (
     DEFAULT_TTL_BY_TYPE,
     compute_content_hash,
-    distance_function_from_container_properties,
-    vector_order_direction,
 )
 from azure.cosmos.agent_memory.aio.store import AsyncMemoryStore
 from azure.cosmos.agent_memory.exceptions import (
+    NoSourceMemoriesError,
     ValidationError,
 )
 from azure.cosmos.agent_memory.logging import get_logger
@@ -301,118 +300,6 @@ async def _embed_one(self, text: str) -> list[float]:
     async def _embed_batch(self, texts: list[str]) -> list[list[float]]:
         return await self._embeddings.generate_batch(texts)
 
-    async def _vector_distance_function(self) -> str:
-        """Return the container's configured Cosmos ``distanceFunction`` (cached).
-
-        Read from the container's vector embedding policy (``await container.read()``)
-        - the authoritative, immutable source set when the container was created.
-        Drives the ORDER BY direction and similarity-threshold comparisons so dedup
-        never silently assumes cosine. Falls back to cosine when the policy can't be
-        read (e.g. ``__new__``-built test instances with mocked containers).
-        """
-        fn = getattr(self, "_distance_function_cache", None)
-        if fn is not None:
-            return fn
-        try:
-            props = await self._memories_container.read()
-        except Exception:
-            # See sync pipeline: don't cache a defaulted cosine, and flag the
-            # failure so the destructive in-place fold path skips this run rather
-            # than mis-applying cosine bands to (possibly euclidean) distances.
-            self._distance_function_read_failed = True
-            logger.debug(
-                "vector dedup: could not read container vector policy; defaulting to cosine (not cached)",
-                exc_info=True,
-            )
-            return "cosine"
-        fn = distance_function_from_container_properties(props)
-        self._distance_function_cache = fn
-        self._distance_function_read_failed = False
-        return fn
-
-    def _warn_distance_policy_unavailable_once(self) -> None:
-        """One-shot WARN that in-place folding was skipped (policy unreadable)."""
-        if getattr(self, "_warned_distance_policy_unavailable", False):
-            return
-        self._warned_distance_policy_unavailable = True
-        logger.warning(
-            "vector dedup: container vector policy could not be read; skipping "
-            "near-exact auto-drop this run to avoid mis-calibrated drops. Memories are "
-            "written as-is and reconciled on a later run once the policy is readable."
-        )
-
-    def _warn_euclidean_autodrop_once(self, distance_function: str) -> None:
-        """One-shot WARN that the near-exact vector auto-drop is disabled.
-
-        The near-exact threshold is cosine-calibrated; on euclidean
-        the destructive auto-drop is skipped and LLM reconcile still runs.
-        Logged once per pipeline instance to avoid hot-path spam.
-        """
-        if getattr(self, "_warned_euclidean_autodrop", False):
-            return
-        self._warned_euclidean_autodrop = True
-        logger.warning(
-            "Container distanceFunction=%r: near-exact vector auto-drop is "
-            "cosine-calibrated and has been DISABLED for this distance function. "
-            "Duplicate detection falls back to borderline tagging + LLM reconcile. "
-            "Use cosine/dotproduct embeddings for vector-floor auto-dedup.",
-            distance_function,
-        )
-
-    async def _vector_candidates(
-        self,
-        *,
-        user_id: str,
-        embedding,
-        memory_type,
-        top_k,
-        exclude_ids,
-    ) -> list[dict]:
-        """Return active same-user vector candidates from Cosmos."""
-        if not user_id or not embedding or not top_k or int(top_k) < 1:
-            return []
-        excluded = set(exclude_ids or [])
-        capped_top = top_literal(int(top_k), name="_vector_candidates.top_k")
-        distance_function = await self._vector_distance_function()
-        order_direction = vector_order_direction(distance_function)
-        field = "embedding"
-        query = (
-            f"SELECT TOP {capped_top} c.id, c.content, c.type, "
-            f"VectorDistance(c.{field}, @vec) AS score "
-            "FROM c WHERE c.user_id = @user_id "
-            "AND c.type = @memory_type "
-            f"AND {_ACTIVE_DOC_FILTER} "
-            f"AND IS_DEFINED(c.{field}) "
-            # Cosmos orders ORDER BY VectorDistance() most-similar-first per the
-            # container's distanceFunction; an explicit ASC/DESC is rejected (BadRequest).
-            f"ORDER BY VectorDistance(c.{field}, @vec)"
-        )
-        rows = await self._query_items(
-            self._memories_container,
-            query=query,
-            parameters=[
-                {"name": "@user_id", "value": user_id},
-                {"name": "@memory_type", "value": memory_type},
-                {"name": "@vec", "value": embedding},
-            ],
-        )
-        candidates = [
-            {
-                "id": row.get("id"),
-                "content": row.get("content"),
-                "type": row.get("type"),
-                "score": float(row.get("score") or 0.0),
-            }
-            for row in rows
-            if row.get("id") and row.get("id") not in excluded
-        ]
-        # Most-similar-first: descending score for cosine/dotproduct, ascending for euclidean.
-        candidates.sort(
-            key=lambda item: item.get("score", 0.0),
-            reverse=order_direction == "DESC",
-        )
-        return candidates
-
     def _prompt_lineage(self, filename: str) -> dict[str, str]:
         """Return ``{prompt_id, prompt_version}`` for stamping a doc.
 
@@ -1634,7 +1521,7 @@ async def generate_user_summary_durable(
             user_doc.pop("embedding", None)
             return user_doc
         if not existing_summary and not items:
-            raise ValidationError(f"No memories found for user_id={user_id!r}")
+            raise NoSourceMemoriesError(f"No memories found for user_id={user_id!r}")
 
         items.sort(key=lambda m: m.get("created_at", ""), reverse=True)
         if recent_k is not None:

azure/cosmos/agent_memory/exceptions.py

@@ -39,6 +39,21 @@ class ValidationError(AgentMemoryError):
     error_code = "validation"
 
 
+class NoSourceMemoriesError(ValidationError):
+    """Raised by the durable summary path when a user has no source memories yet.
+
+    A brand-new user can cross the user-summary threshold before fact extraction
+    has persisted anything, because the change feed starts the extract and
+    user-summary orchestrations independently. Subclasses :class:`ValidationError`
+    so existing ``except ValidationError`` handlers keep catching it, while
+    letting the Durable user-summary orchestrator tell this expected "extraction
+    has not landed yet" race apart from a real validation failure and wait/retry
+    instead of failing after its short activity-retry window.
+    """
+
+    error_code = "no_source_memories"
+
+
 class CosmosNotConnectedError(AgentMemoryError):
     """Raised when a Cosmos DB operation is attempted without an active connection."""