Merge 62d6f5e44f into 2ad56a4d71

src/reconciler/sync_vectors.py

@@ -426,43 +426,67 @@ async def _sync_message_embeddings(
         emb for emb in embeddings if emb.embedding is None
     ]
     freshly_embedded: dict[int, list[float]] = {}
+    # Rows rejected with a permanent validation error (ValueError) — not
+    # retryable, so mark them failed immediately instead of burning
+    # MAX_SYNC_ATTEMPTS retries.
+    permanently_failed: set[int] = set()
 
     if embs_needing_embed:
-        try:
-            contents = [emb.content for emb in embs_needing_embed]
-            # MESSAGE_CREATE (not VECTOR_SYNC): these rows come from create_messages
-            # as pending chunks; document re-embeds stay on VECTOR_SYNC below.
-            workspaces = {emb.workspace_name for emb in embs_needing_embed}
-            with embedding_call_purpose(
-                EmbeddingCallPurpose.MESSAGE_CREATE.value,
-                workspace_name=workspaces.pop() if len(workspaces) == 1 else None,
-                parent_category="reconciliation",
-            ):
-                new_embeddings = await embedding_client.simple_batch_embed(contents)
-
-            if len(new_embeddings) != len(embs_needing_embed):
-                logger.warning(
-                    "Re-embedded %s/%s message embeddings; remaining will be retried",
-                    len(new_embeddings),
-                    len(embs_needing_embed),
-                )
-
-            for emb, new_emb in zip(embs_needing_embed, new_embeddings, strict=False):
-                freshly_embedded[emb.id] = new_emb
-                if store_in_postgres:
-                    emb.embedding = new_emb
-        except Exception:
-            logger.exception(
-                "Failed to re-embed %s message embeddings", len(embs_needing_embed)
-            )
+        # Embed each text individually so a single oversized text (rejected by
+        # the provider) can't poison the whole batch. The chunking tokenizer
+        # can undercount a provider's real token count (e.g. tiktoken o200k vs
+        # nomic-embed-text), so oversized texts can still slip through the chunk
+        # cap and get rejected by the provider. Isolating per-text keeps the
+        # rest of the batch embeddable and lets genuinely-oversized rows fail
+        # on their own instead of taking the whole batch down with them.
+        workspaces = {emb.workspace_name for emb in embs_needing_embed}
+        with embedding_call_purpose(
+            EmbeddingCallPurpose.MESSAGE_CREATE.value,
+            workspace_name=workspaces.pop() if len(workspaces) == 1 else None,
+            parent_category="reconciliation",
+        ):
+            for emb in embs_needing_embed:
+                try:
+                    new_emb = await embedding_client.simple_batch_embed([emb.content])
+                    freshly_embedded[emb.id] = new_emb[0]
+                    if store_in_postgres:
+                        emb.embedding = new_emb[0]
+                except ValueError as e:
+                    # Expected validation failure (oversized input, dimension
+                    # mismatch). Not retryable — mark failed immediately.
+                    logger.warning(
+                        "Message %s chunk %s rejected by embedding provider: %s",
+                        emb.message_id,
+                        emb.id,
+                        e,
+                    )
+                    permanently_failed.add(emb.id)
+                except Exception:
+                    logger.exception(
+                        "Unexpected error embedding message %s chunk %s; will retry",
+                        emb.message_id,
+                        emb.id,
+                    )
 
     # Mark embeddings that failed to get a vector
     failed_to_embed: list[models.MessageEmbedding] = [
         emb for emb in embs_needing_embed if emb.id not in freshly_embedded
     ]
     if failed_to_embed:
-        await _bump_message_embedding_sync_attempts(db, failed_to_embed)
-        failed_count += len(failed_to_embed)
+        # Expected validation failures are permanent — mark them failed
+        # directly instead of retrying MAX_SYNC_ATTEMPTS times.
+        permanent = [emb for emb in failed_to_embed if emb.id in permanently_failed]
+        if permanent:
+            await db.execute(
+                update(models.MessageEmbedding)
+                .where(models.MessageEmbedding.id.in_([emb.id for emb in permanent]))
+                .values(sync_state="failed", last_sync_at=func.now())
+            )
+            failed_count += len(permanent)
+        retryable = [emb for emb in failed_to_embed if emb.id not in permanently_failed]
+        if retryable:
+            await _bump_message_embedding_sync_attempts(db, retryable)
+            failed_count += len(retryable)
 
     # pgvector-only mode: no external store to upsert to. Any row that now
     # has an embedding (either pre-existing or freshly embedded) is fully