fix(cascade): embed a file's changed entries in one batched call

CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- **`[embedding].max_concurrent` defaults to 10 instead of 5.** The
+  `EmbeddingSettings` class default (50) is aligned to the same value; the
+  shipped `default.toml` always overrode it, so the effective default moves
+  from 5 to 10. (#486)
+
 ### Fixed
 
 - **The same owner's memories in two projects no longer overwrite each other
@@ -44,6 +51,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   nothing pending, waiting for the worker's in-flight batch, and
   `everos cascade sync` / `rebuild` use the same loop.
 
+- **A cascade pass embeds a file's changed entries in one batched request,
+  not one request per entry.** The four daily-log handlers (episode, atomic
+  fact, foresight, agent case) each called the embedding endpoint once per
+  entry, so a turn that wrote ~20 atomic facts queued ~20 single-input
+  requests on the process-wide embedding semaphore, and the next search's
+  query embedding waited behind all of them: about five times the
+  endpoint's latency at the previous default `max_concurrent = 5`. The
+  handlers now
+  send every text of a file through `embed_batch`, which
+  `[embedding].batch_size` splits into requests (23 entries become 3
+  requests at the default 10), so after a single turn's writes the query
+  embedding waits for one request; a large backlog still queues ahead of
+  it (#473). (#486)
+
 - **Changing a knowledge document's category keeps it inside `knowledge/`.**
   `PATCH /knowledge/documents/{doc_id}` with `category_id` set to `..` moved
   the document's directory up into the project directory, where the markdown

config.example.toml

@@ -34,7 +34,7 @@ base_url = "https://api.deepinfra.com/v1/openai"
 timeout_seconds = 30.0
 max_retries     = 3
 batch_size      = 10
-max_concurrent  = 5
+max_concurrent  = 10
 
 # ── Rerank ────────────────────────────────────────────
 [rerank]

docs/configuration.md

@@ -151,7 +151,13 @@ Zilliz Cloud endpoint; a Milvus Lite filesystem path is rejected.
 | `timeout_seconds` | float | `30.0` | No | Request timeout. |
 | `max_retries` | int | `3` | No | Retry count on failure. |
 | `batch_size` | int | `10` | No | Texts per batch request. |
-| `max_concurrent` | int | `5` | No | Max parallel batch requests. |
+| `max_concurrent` | int | `10` | No | Max parallel batch requests. |
+
+`timeout_seconds` bounds one request, and one request carries up to
+`batch_size` texts (the cascade embeds a file's changed entries in one
+batched call). On a slow or serialized local endpoint (CPU Ollama, a single
+GPU) queued batch requests can exceed the timeout: lower `batch_size` and
+`max_concurrent` together, or raise `timeout_seconds`.
 
 ### `[rerank]`
 

src/everos/component/embedding/capability.py

@@ -3,8 +3,9 @@
 The wrapper encapsulates ``Optional[EmbeddingProvider]`` so consumers
 never see ``None`` directly. Two consumption modes:
 
-- :meth:`embed_or_none` for soft-degrade write paths (cascade handlers
-  that write ``vector=NULL`` when embedding is unavailable).
+- :meth:`embed_or_none` / :meth:`embed_batch_or_none` for soft-degrade
+  write paths (cascade handlers that write ``vector=NULL`` when embedding
+  is unavailable).
 - :meth:`require` for hard-required paths (search VECTOR/HYBRID/AGENTIC)
   that raise :class:`ProviderNotConfiguredError` -> HTTP 422 when missing.
 
@@ -17,6 +18,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from dataclasses import dataclass
 
 from everos.core.errors import ProviderNotConfiguredError
@@ -45,6 +47,18 @@ async def embed_or_none(self, text: str) -> list[float] | None:
             return None
         return list(await self.provider.embed(text))
 
+    async def embed_batch_or_none(
+        self, texts: Sequence[str]
+    ) -> list[list[float]] | None:
+        """Embed ``texts`` in one batched call if available, else ``None``.
+
+        Same soft-degrade contract as :meth:`embed_or_none`; the provider
+        chunks the request by its configured ``batch_size``.
+        """
+        if self.provider is None:
+            return None
+        return [list(v) for v in await self.provider.embed_batch(texts)]
+
     def require(self) -> EmbeddingProvider:
         """Return the provider, or raise :class:`ProviderNotConfiguredError`.
 

src/everos/component/embedding/openai_provider.py

@@ -132,9 +132,12 @@ async def _embed_chunk(self, chunk: list[str]) -> list[list[float]]:
                     )
                 except openai.OpenAIError as exc:
                     raise _classify(exc) from exc
-            if not response.data:
+            # Callers slice the batch back per input, so a short (or empty)
+            # list would silently shift every later vector onto the wrong text.
+            if len(response.data) != len(chunk):
                 raise EmbeddingServiceError(
-                    f"Embedding API returned empty data for {len(chunk)} inputs"
+                    f"Embedding API returned {len(response.data)} vectors "
+                    f"for {len(chunk)} inputs"
                 )
             # Embeddings report only input (prompt) tokens.
             usage = getattr(response, "usage", None)

src/everos/config/default.toml

@@ -135,7 +135,7 @@ base_url  = "https://api.deepinfra.com/v1/openai"
 timeout_seconds = 30.0
 max_retries     = 3
 batch_size      = 10
-max_concurrent  = 5
+max_concurrent  = 10
 
 [rerank]
 # Rerank provider. Override via env:

src/everos/config/settings.py

@@ -377,7 +377,7 @@ class EmbeddingSettings(BaseModel):
     timeout_seconds: float = Field(default=30.0, gt=0)
     max_retries: int = Field(default=3, ge=0)
     batch_size: int = Field(default=10, ge=1)
-    max_concurrent: int = Field(default=50, ge=1)
+    max_concurrent: int = Field(default=10, ge=1)
 
 
 class RerankSettings(BaseModel):

src/everos/memory/cascade/handlers/__init__.py

@@ -2,8 +2,9 @@
 
 Four daily-log handlers (episode / atomic_fact / foresight /
 agent_case) inherit :class:`BaseDailyLogHandler` for the shared
-read / diff / upsert / delete loop; the per-kind subclass only
-declares its repo binding and ``_build_row`` mapping. ``agent_skill``,
+read / diff / embed / upsert / delete loop; the per-kind subclass only
+declares its repo binding, the texts to embed (``_embed_texts``) and
+the ``_build_row`` mapping. ``agent_skill``,
 ``user_profile``, and ``knowledge_topic`` stand alone — they're
 single-file kinds (no entries, no per-entry diff), so they implement
 :class:`Handler` directly and own their reconcile loop.

src/everos/memory/cascade/handlers/_daily_log_base.py

@@ -26,8 +26,10 @@
 import abc
 import asyncio
 import dataclasses
+from collections.abc import Sequence
 from typing import Any, ClassVar
 
+from everos.component.embedding import get_embedding_capability
 from everos.core.observability.logging import get_logger
 from everos.core.persistence import MarkdownReader, StructuredEntry
 from everos.infra.persistence.index import all_of, eq, one_of
@@ -65,9 +67,11 @@ class BaseDailyLogHandler(Handler):
       subset of inline + section fields whose changes should trigger
       re-upsert + re-embed. Each key is ``"section:Name"`` or
       ``"inline:name"``.
+    - :meth:`_embed_texts` (override) — the texts one entry needs embedded.
     - :meth:`_build_row` (override) — turn a :class:`ParsedEntry` plus
-      common context (owner_id / owner_type / md_path) into a typed
-      LanceDB row. Tokenisation + embedding live in the subclass.
+      common context (owner_id / owner_type / md_path) and the vectors
+      for its texts into a typed LanceDB row. Tokenisation lives in the
+      subclass; embedding is batched here, one call per file.
     """
 
     kind: ClassVar[str] = ""
@@ -170,24 +174,46 @@ async def _embed_entries(
         project_id: str,
         md_path: str,
     ) -> list[Any]:
-        """Build LanceDB rows for changed entries (embed concurrently)."""
+        """Build LanceDB rows for changed entries.
+
+        Every text the changed entries need goes out in ONE ``embed_batch``
+        call (the provider chunks it by ``batch_size``), not one ``embed``
+        call per entry: the per-entry fan-out queued N single-input requests
+        on the process-wide embedding semaphore ahead of the interactive
+        query embedding (issue #486).
+        """
         if not to_build:
             return []
-        return list(
-            await asyncio.gather(
-                *(
-                    self._build_row(
-                        owner_id=owner_id,
-                        owner_type=owner_type,
-                        app_id=app_id,
-                        project_id=project_id,
-                        md_path=md_path,
-                        entry=entry,
-                    )
-                    for entry in to_build
+        texts_per_entry = [self._embed_texts(entry) for entry in to_build]
+        flat = [text for texts in texts_per_entry for text in texts]
+        vectors: Sequence[list[float] | None] | None
+        vectors = await get_embedding_capability().embed_batch_or_none(flat)
+        if vectors is None:
+            logger.debug(
+                "cascade_handler_embed_skipped",
+                kind=self.kind,
+                md_path=md_path,
+                entries=len(to_build),
+                reason="embedding_capability_unavailable",
+            )
+            vectors = [None] * len(flat)
+        rows: list[Any] = []
+        cursor = 0
+        for entry, texts in zip(to_build, texts_per_entry, strict=True):
+            entry_vectors = vectors[cursor : cursor + len(texts)]
+            cursor += len(texts)
+            rows.append(
+                await self._build_row(
+                    owner_id=owner_id,
+                    owner_type=owner_type,
+                    app_id=app_id,
+                    project_id=project_id,
+                    md_path=md_path,
+                    entry=entry,
+                    vectors=entry_vectors,
                 )
             )
-        )
+        return rows
 
     async def _apply_lance_changes(
         self,
@@ -272,6 +298,11 @@ async def _mark_deprecated(
                 exc_info=True,
             )
 
+    @abc.abstractmethod
+    def _embed_texts(self, entry: ParsedEntry) -> tuple[str, ...]:
+        """Subclass: the texts one entry needs embedded, in the order
+        :meth:`_build_row` receives their ``vectors`` back."""
+
     @abc.abstractmethod
     async def _build_row(
         self,
@@ -282,9 +313,14 @@ async def _build_row(
         project_id: str = "default",
         md_path: str,
         entry: ParsedEntry,
+        vectors: Sequence[list[float] | None] = (),
     ) -> Any:
         """Subclass: build the typed LanceDB row for one parsed entry.
 
+        ``vectors`` aligns with :meth:`_embed_texts`; an empty sequence or
+        ``None`` items mean "no vector" (embedding unavailable) and the row
+        is written with ``vector=None``.
+
         ``app_id`` / ``project_id`` carry the path-derived scope; the base
         always supplies them (via :func:`resolve_scope`). They default to
         ``"default"`` so white-box callers exercising only the field mapping

src/everos/memory/cascade/handlers/agent_case.py

@@ -28,15 +28,13 @@
 
 from __future__ import annotations
 
-from everos.component.embedding import get_embedding_capability
-from everos.core.observability.logging import get_logger
+from collections.abc import Sequence
+
 from everos.infra.persistence.index import AgentCase, ParentType, agent_case_repo
 
 from ._common import require_float, require_iso_timestamp
 from ._daily_log_base import BaseDailyLogHandler, ParsedEntry
 
-logger = get_logger(__name__)
-
 
 class AgentCaseHandler(BaseDailyLogHandler):
     """Cascade handler for ``agents/<a>/.cases/agent_case-*.md``."""
@@ -52,6 +50,9 @@ class AgentCaseHandler(BaseDailyLogHandler):
     """Includes quality_score (semantic score, not audit). Audit
     inline (owner_id / session_id / timestamp / parent_id) is excluded."""
 
+    def _embed_texts(self, entry: ParsedEntry) -> tuple[str, ...]:
+        return (entry.structured.sections.get("TaskIntent", "").strip(),)
+
     async def _build_row(
         self,
         *,
@@ -61,21 +62,14 @@ async def _build_row(
         project_id: str = "default",
         md_path: str,
         entry: ParsedEntry,
+        vectors: Sequence[list[float] | None] = (),
     ) -> AgentCase:
         s = entry.structured
         task_intent = s.sections.get("TaskIntent", "").strip()
         approach = s.sections.get("Approach", "").strip()
         key_insight = (s.sections.get("KeyInsight") or "").strip() or None
         intent_tokens = self._deps.tokenizer.tokenize(task_intent)
         approach_tokens = self._deps.tokenizer.tokenize(approach)
-        vector = await get_embedding_capability().embed_or_none(task_intent)
-        if vector is None:
-            logger.debug(
-                "cascade_handler_embed_skipped",
-                kind=self.kind,
-                entry_id=entry.entry_id,
-                reason="embedding_capability_unavailable",
-            )
         return AgentCase(
             id=f"{owner_id}_{entry.entry_id}",
             entry_id=entry.entry_id,
@@ -97,5 +91,5 @@ async def _build_row(
             key_insight=key_insight,
             md_path=md_path,
             content_sha256=entry.content_sha256,
-            vector=vector,
+            vector=vectors[0] if vectors else None,
         )

src/everos/memory/cascade/handlers/atomic_fact.py

@@ -21,15 +21,13 @@
 
 from __future__ import annotations
 
-from everos.component.embedding import get_embedding_capability
-from everos.core.observability.logging import get_logger
+from collections.abc import Sequence
+
 from everos.infra.persistence.index import AtomicFact, ParentType, atomic_fact_repo
 
 from ._common import parse_inline_list, require_iso_timestamp
 from ._daily_log_base import BaseDailyLogHandler, ParsedEntry
 
-logger = get_logger(__name__)
-
 
 class AtomicFactHandler(BaseDailyLogHandler):
     """Cascade handler for ``users/<u>/.atomic_facts/atomic_fact-*.md``."""
@@ -40,6 +38,9 @@ class AtomicFactHandler(BaseDailyLogHandler):
     """Only ``Fact`` matters — it's both the embedded text AND the
     BM25 source. Audit inline is excluded."""
 
+    def _embed_texts(self, entry: ParsedEntry) -> tuple[str, ...]:
+        return (entry.structured.sections.get("Fact", "").strip(),)
+
     async def _build_row(
         self,
         *,
@@ -49,18 +50,11 @@ async def _build_row(
         project_id: str = "default",
         md_path: str,
         entry: ParsedEntry,
+        vectors: Sequence[list[float] | None] = (),
     ) -> AtomicFact:
         s = entry.structured
         text = s.sections.get("Fact", "").strip()
         tokens = self._deps.tokenizer.tokenize(text)
-        vector = await get_embedding_capability().embed_or_none(text)
-        if vector is None:
-            logger.debug(
-                "cascade_handler_embed_skipped",
-                kind=self.kind,
-                entry_id=entry.entry_id,
-                reason="embedding_capability_unavailable",
-            )
         return AtomicFact(
             id=f"{owner_id}_{entry.entry_id}",
             entry_id=entry.entry_id,
@@ -77,5 +71,5 @@ async def _build_row(
             fact_tokens=" ".join(tokens),
             md_path=md_path,
             content_sha256=entry.content_sha256,
-            vector=vector,
+            vector=vectors[0] if vectors else None,
         )