diff --git a/src/reconciler/sync_vectors.py b/src/reconciler/sync_vectors.py index b9b06418..87afcc2a 100644 --- a/src/reconciler/sync_vectors.py +++ b/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