Fix #2461: [Bug] Content between embedder limit and chat_window_max_tokens is silently stor

src/memos/api/config.py

@@ -685,6 +685,32 @@ def get_reader_config() -> dict[str, Any]:
             },
         }
 
+    @staticmethod
+    def get_chat_window_max_tokens() -> int:
+        """Get the mem_reader per-item token budget.
+
+        Items above this size are split before embedding. Deployments whose embedder has a
+        lower per-item limit than the default (1024) should lower this too, otherwise
+        content in between is stored without a vector (issue #2461).
+        """
+        raw = os.getenv("MEM_READER_CHAT_WINDOW_MAX_TOKENS", "")
+        if not raw.strip():
+            return 1024
+        try:
+            value = int(raw)
+        except ValueError:
+            logger.warning(
+                "Invalid MEM_READER_CHAT_WINDOW_MAX_TOKENS=%r; falling back to 1024", raw
+            )
+            return 1024
+        if value <= 0:
+            logger.warning(
+                "MEM_READER_CHAT_WINDOW_MAX_TOKENS must be positive (got %r); falling back to 1024",
+                raw,
+            )
+            return 1024
+        return value
+
     @staticmethod
     def get_oss_config() -> dict[str, Any] | None:
         """Get OSS configuration and validate connection."""
@@ -775,6 +801,7 @@ def get_internet_config() -> dict[str, Any]:
                             },
                         },
                         "chat_chunker": reader_config,
+                        "chat_window_max_tokens": APIConfig.get_chat_window_max_tokens(),
                     },
                 },
             },
@@ -1054,6 +1081,7 @@ def get_product_default_config() -> dict[str, Any]:
                         },
                     },
                     "chat_chunker": reader_config,
+                    "chat_window_max_tokens": APIConfig.get_chat_window_max_tokens(),
                     "direct_markdown_hostnames": [
                         h.strip()
                         for h in os.getenv(
@@ -1189,6 +1217,7 @@ def create_user_config(user_name: str, user_id: str) -> tuple["MOSConfig", "Gene
                         },
                     },
                     "chat_chunker": reader_config,
+                    "chat_window_max_tokens": APIConfig.get_chat_window_max_tokens(),
                 },
             },
             "enable_textual_memory": True,

src/memos/configs/mem_reader.py

@@ -59,6 +59,23 @@ def parse_datetime(cls, value):
     chat_chunker: dict[str, Any] = Field(
         default=None, description="Configuration for the MemReader chat chunk strategy"
     )
+    chat_window_max_tokens: int = Field(
+        default=1024,
+        description=(
+            "Maximum number of tokens per memory item. Items above this size are split "
+            "before embedding. The effective split budget is the minimum of this value "
+            "and the embedder's per-item limit, so texts that the embedder would reject "
+            "are never handed to it whole (issue #2461)."
+        ),
+    )
+
+    @field_validator("chat_window_max_tokens")
+    @classmethod
+    def validate_chat_window_max_tokens(cls, value: int) -> int:
+        """Reject non-positive windows; they would make every item unsplittable."""
+        if value <= 0:
+            raise ValueError("chat_window_max_tokens must be a positive integer")
+        return value
 
 
 class SimpleStructMemReaderConfig(BaseMemReaderConfig):

src/memos/mem_reader/multi_modal_struct.py

@@ -92,7 +92,9 @@ def _embed_memory_items(self, items: list[TextualMemoryItem]) -> None:
         """Compute embeddings for a list of memory items in-place.
 
         Attempts a single batch call first; falls back to per-item calls if the
-        batch fails.  Errors are logged but never raised so callers always
+        batch fails. The per-item retry truncates each text to the embedder budget before
+        calling, so a single over-limit item cannot leave the whole batch without
+        embeddings (issue #2461). Errors are logged but never raised so callers always
         continue normally.
         """
         valid = [w for w in items if w and w.memory]
@@ -106,9 +108,18 @@ def _embed_memory_items(self, items: list[TextualMemoryItem]) -> None:
         except Exception as e:
             logger.error(f"[MultiModalStruct] Error batch computing embeddings: {e}")
             logger.warning("[EMBED_FALLBACK] batch_size=%d", len(texts))
+            budget = self._get_embed_budget()
             for w in valid:
+                text = w.memory or ""
+                payload = self._truncate_to_budget(text, budget)
+                if payload != text:
+                    logger.warning(
+                        "[EMBED_TRUNCATE] fallback truncated %d-token text to budget %d",
+                        self._count_tokens_safe(text),
+                        budget,
+                    )
                 try:
-                    w.metadata.embedding = self.embedder.embed([w.memory])[0]
+                    w.metadata.embedding = self.embedder.embed([payload])[0]
                 except Exception as e2:
                     logger.error(f"[MultiModalStruct] Error computing embedding for item: {e2}")
 
@@ -118,6 +129,12 @@ def _split_large_memory_item(
         """
         Split a single memory item that exceeds max_tokens into multiple chunks.
 
+        The chunker is the primary splitter, but its output is not trusted blindly: some
+        backends (e.g. ``sentence``/chonkie) do not recognise CJK sentence terminators and
+        can hand back a chunk that is still larger than the budget. Every produced chunk is
+        therefore re-checked and hard-split when it still exceeds the budget, guaranteeing
+        that nothing over the embedder's per-item limit is ever sent to it (issue #2461).
+
         Args:
             item: TextualMemoryItem to split
             max_tokens: Maximum tokens per chunk
@@ -129,68 +146,101 @@ def _split_large_memory_item(
         if not item_text:
             return [item]
 
-        item_tokens = self._count_tokens(item_text)
+        item_tokens = self._count_tokens_safe(item_text)
         if item_tokens <= max_tokens:
             return [item]
 
-        # Use chunker to split the text
-        try:
-            chunks = self.chunker.chunk(item_text)
-            split_items = []
-            source_info = dict(item.metadata.info or {})
-            source_internal_info = dict(item.metadata.internal_info or {})
-            ingest_batch_id = str(source_internal_info.get("ingest_batch_id") or uuid.uuid4())
-            chunk_total = len(chunks)
-
-            def _create_chunk_item(chunk_idx: int, chunk):
-                # Different chunkers are not fully consistent:
-                # some return Chunk-like objects with `.text`, while others return raw strings.
-                chunk_text = chunk.text if hasattr(chunk, "text") else chunk
-                if not chunk_text or not chunk_text.strip():
-                    return None
-                chunk_info = {
-                    "user_id": item.metadata.user_id,
-                    "session_id": item.metadata.session_id,
-                    **source_info,
-                }
-                chunk_internal_info = {
-                    **source_internal_info,
-                    "ingest_batch_id": ingest_batch_id,
-                    "chunk_index": chunk_idx,
-                    "chunk_total": chunk_total,
-                }
-                # Create a new memory item for each chunk, preserving original metadata
-                split_item = self._make_memory_item(
-                    value=chunk_text,
-                    info=chunk_info,
-                    memory_type=item.metadata.memory_type,
-                    tags=item.metadata.tags or [],
-                    key=item.metadata.key,
-                    sources=item.metadata.sources or [],
-                    background=item.metadata.background or "",
-                    need_embed=False,
-                )
-                split_item.metadata.internal_info = chunk_internal_info
-                return split_item
+        # The chunker is best-effort: any failure here (unsupported backend API, empty
+        # output) must NOT fall back to the original over-budget item, which is exactly the
+        # silent no-vector write of issue #2461. Hard-split instead.
+        raw_texts = self._chunk_within_budget(item_text, max_tokens)
+        if not raw_texts:
+            return [item]
 
+        split_items = []
+        source_info = dict(item.metadata.info or {})
+        source_internal_info = dict(item.metadata.internal_info or {})
+        ingest_batch_id = str(source_internal_info.get("ingest_batch_id") or uuid.uuid4())
+        chunk_total = len(raw_texts)
+
+        def _create_chunk_item(chunk_idx: int, chunk_text: str):
+            chunk_info = {
+                "user_id": item.metadata.user_id,
+                "session_id": item.metadata.session_id,
+                **source_info,
+            }
+            chunk_internal_info = {
+                **source_internal_info,
+                "ingest_batch_id": ingest_batch_id,
+                "chunk_index": chunk_idx,
+                "chunk_total": chunk_total,
+            }
+            # Create a new memory item for each chunk, preserving original metadata
+            split_item = self._make_memory_item(
+                value=chunk_text,
+                info=chunk_info,
+                memory_type=item.metadata.memory_type,
+                tags=item.metadata.tags or [],
+                key=item.metadata.key,
+                sources=item.metadata.sources or [],
+                background=item.metadata.background or "",
+                need_embed=False,
+            )
+            split_item.metadata.internal_info = chunk_internal_info
+            return split_item
+
+        try:
             # Use thread pool to parallel process chunks, but keep the original order
             with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
                 futures = [
-                    executor.submit(_create_chunk_item, chunk_idx, chunk)
-                    for chunk_idx, chunk in enumerate(chunks)
+                    executor.submit(_create_chunk_item, chunk_idx, chunk_text)
+                    for chunk_idx, chunk_text in enumerate(raw_texts)
                 ]
                 for future in futures:
                     split_item = future.result()
                     if split_item is not None:
                         split_items.append(split_item)
-
-            return split_items if split_items else [item]
         except Exception as e:
             logger.warning(
-                f"[MultiModalStruct] Failed to split large memory item: {e}. Returning original item."
+                f"[MultiModalStruct] Failed building split item: {e}. Returning original item."
             )
             return [item]
 
+        return split_items if split_items else [item]
+
+    def _chunk_within_budget(self, text: str, max_tokens: int) -> list[str]:
+        """Split ``text`` with the chunker, guaranteeing every piece fits ``max_tokens``.
+
+        Chunker output is re-checked per chunk and hard-split when it still exceeds the
+        budget. If the chunker raises or produces nothing usable, the hard split runs over
+        the raw text so the caller never receives an over-budget item (issue #2461).
+        """
+        raw_texts: list[str] = []
+        try:
+            # Materialise inside the guard: some chunkers raise lazily while iterating.
+            chunks = list(self.chunker.chunk(text))
+        except Exception as e:
+            logger.warning(
+                f"[MultiModalStruct] Chunker failed ({e}); falling back to a hard split."
+            )
+            return self._hard_split_text(text, max_tokens)
+
+        for chunk in chunks:
+            # Different chunkers are not fully consistent: some return Chunk-like
+            # objects with `.text`, while others return raw strings.
+            chunk_text = chunk.text if hasattr(chunk, "text") else chunk
+            if not chunk_text or not chunk_text.strip():
+                continue
+            if self._count_tokens_safe(chunk_text) > max_tokens:
+                raw_texts.extend(self._hard_split_text(chunk_text, max_tokens))
+            else:
+                raw_texts.append(chunk_text)
+
+        if not raw_texts:
+            # The chunker cannot segment this text at all; hard-split it directly.
+            return self._hard_split_text(text, max_tokens)
+        return raw_texts
+
     def _concat_multi_modal_memories(
         self, all_memory_items: list[TextualMemoryItem], max_tokens=None, overlap=200
     ) -> list[TextualMemoryItem]:
@@ -206,7 +256,7 @@ def _concat_multi_modal_memories(
         if not all_memory_items:
             return []
 
-        max_tokens = max_tokens or self.chat_window_max_tokens
+        max_tokens = max_tokens or self._get_embed_budget()
 
         # Split large memory items before processing
         processed_items = []
@@ -219,7 +269,7 @@ def _concat_multi_modal_memories(
                 # Create a list to hold futures with their original index
                 futures = []
                 for idx, item in enumerate(all_memory_items):
-                    if (item.memory or "") and self._count_tokens(item.memory) > max_tokens:
+                    if (item.memory or "") and self._count_tokens_safe(item.memory) > max_tokens:
                         future = executor.submit(self._split_large_memory_item, item, max_tokens)
                         futures.append(
                             (idx, future, True)
@@ -245,14 +295,34 @@ def _concat_multi_modal_memories(
             # serial chunk large memory items
             for item in all_memory_items:
                 item_text = item.memory or ""
-                item_tokens = self._count_tokens(item_text)
+                item_tokens = self._count_tokens_safe(item_text)
                 if item_tokens > max_tokens:
                     # Split the large item into multiple chunks
                     split_items = self._split_large_memory_item(item, max_tokens)
                     processed_items.extend(split_items)
                 else:
                     processed_items.append(item)
 
+        # Guard: nothing over the embed budget may proceed. If an item is still oversized
+        # after splitting (chunker produced nothing usable and the hard split could not
+        # segment it either), emit the items directly instead of aggregating them into an
+        # even larger window — that aggregation is what produced the silent no-vector write
+        # in issue #2461. The per-item embedding step truncates as a final safety net.
+        oversize = [
+            it for it in processed_items if self._count_tokens_safe(it.memory or "") > max_tokens
+        ]
+        if oversize:
+            logger.warning(
+                "[MultiModalStruct] %d item(s) still exceed the embed budget (%d tokens) "
+                "after splitting; emitting them directly so the embedder can truncate "
+                "rather than silently dropping the vector",
+                len(oversize),
+                max_tokens,
+            )
+            with timed_stage("add", "embedding", window_count=len(processed_items)):
+                self._embed_memory_items(processed_items)
+            return processed_items
+
         # If only one item after processing, compute embedding and return
         if len(processed_items) == 1:
             single_item = processed_items[0]
@@ -280,7 +350,7 @@ def _concat_multi_modal_memories(
             # Check if adding this item would exceed max_tokens (same logic as _iter_chat_windows)
             # Note: After splitting large items, each item should be <= max_tokens,
             # but we still check to handle edge cases
-            if self._count_tokens(cur_text + line) > max_tokens and cur_text:
+            if self._count_tokens_safe(cur_text + line) > max_tokens and cur_text:
                 # Yield current window
                 window = self._build_window_from_items(buf_items, info)
                 if window:
@@ -290,7 +360,8 @@ def _concat_multi_modal_memories(
                 # (same logic as _iter_chat_windows)
                 while (
                     buf_items
-                    and self._count_tokens("".join([it.memory or "" for it in buf_items])) > overlap
+                    and self._count_tokens_safe("".join([it.memory or "" for it in buf_items]))
+                    > overlap
                 ):
                     buf_items.pop(0)
                 # Recalculate cur_text from remaining items