fix(data-access): cap embedQueries texts, surface cacheError, keep cache helpers internal (LLMO-7445)

Author: Tanase Butcaru

docs/adr/0002-embedding-client-and-semantic-index-utils.md

@@ -45,13 +45,13 @@ Following ADR-0001's test (*true for every consumer → shared; specific to one
 
 ### 4. Semantic-index helpers live in `data-access`, mirroring the URL index
 
-`semantic-index.utils.js` mirrors `url-index.utils.js`:_Superseded by the amendment below:_  `syncOpportunitySemantic` (full-replace per `source_type`), `lookupOpportunitiesByVectors` (ANN read RPC wrapper, many query vectors per call), and the batched `semantic_query_embedding` cache helpers (`getQueryEmbeddings` / `upsertQueryEmbeddings` / `touchQueryEmbeddings`). (`copyEntityVectors`, a wrapper over the `wrpc_copy_opportunity_semantic_vectors` RPC, was removed with its only caller, the offsite snapshot vector copy.) Text normalization + hashing live in this layer so a value written always matches a later read/re-sync. **The shared layer stores and reads pre-embedded vectors; it never embeds** — embedding is the writer's (audit-worker's) job via the client above.
+_Superseded by the amendment below:_ `semantic-index.utils.js` mirrors `url-index.utils.js`: `syncOpportunitySemantic` (full-replace per `source_type`), `lookupOpportunitiesByVectors` (ANN read RPC wrapper, many query vectors per call), and the batched `semantic_query_embedding` cache helpers (`getQueryEmbeddings` / `upsertQueryEmbeddings` / `touchQueryEmbeddings`). (`copyEntityVectors`, a wrapper over the `wrpc_copy_opportunity_semantic_vectors` RPC, was removed with its only caller, the offsite snapshot vector copy.) Text normalization + hashing live in this layer so a value written always matches a later read/re-sync. **The shared layer stores and reads pre-embedded vectors; it never embeds** — embedding is the writer's (audit-worker's) job via the client above.
 
-**Type registries.**_Superseded by the amendment below:_  `OPPORTUNITY_SEMANTIC_SOURCE_TYPES` (kinds of embedded text: `topic`, later `claim`) and `OPPORTUNITY_SEMANTIC_ENTITY_TYPES` (opportunity types, from `OPPORTUNITY_TYPES`) are defined in `spacecat-shared-utils` (so non-Node consumers such as the UI can import them and their types) and re-exported from this layer, and this library's writer stores and reader accepts only their values; anything else is a `ValidationError`. The reader takes `sourceTypes` (required, non-empty) and optional `entityTypes` (omitted or empty = all), deduped, so a list can never exceed its registry. The DB does not duplicate the registries: the RPC only rejects empty, multi-dimensional, or NULL-element lists and lists over 100 elements (a length bound, not a registry mirror; `NULL` entity types = all; the reader sends `NULL` when `entityTypes` is omitted or empty). A team that wants its opportunities searchable adds its source kind and/or opportunity type to the registries, indexes with them, and callers opt in by passing them.
+_Superseded by the amendment below:_ **Type registries.** `OPPORTUNITY_SEMANTIC_SOURCE_TYPES` (kinds of embedded text: `topic`, later `claim`) and `OPPORTUNITY_SEMANTIC_ENTITY_TYPES` (opportunity types, from `OPPORTUNITY_TYPES`) are defined in `spacecat-shared-utils` (so non-Node consumers such as the UI can import them and their types) and re-exported from this layer, and this library's writer stores and reader accepts only their values; anything else is a `ValidationError`. The reader takes `sourceTypes` (required, non-empty) and optional `entityTypes` (omitted or empty = all), deduped, so a list can never exceed its registry. The DB does not duplicate the registries: the RPC only rejects empty, multi-dimensional, or NULL-element lists and lists over 100 elements (a length bound, not a registry mirror; `NULL` entity types = all; the reader sends `NULL` when `entityTypes` is omitted or empty). A team that wants its opportunities searchable adds its source kind and/or opportunity type to the registries, indexes with them, and callers opt in by passing them.
 
 **Status filter.** The reader also takes optional `statuses`, checked against `Opportunity.STATUSES` and deduped like the type lists. The RPC applies it before the per-query limit, so filtered-out opportunities never take result slots. Which statuses count as visible is the caller's choice. The data-service casts the list to its status enum, so a value it doesn't know is an error, not an empty result. Sending `statuses` needs a data-service with `p_statuses` (mysticat-data-service PR 1132).
 
-Adding an entity type:_Superseded by the amendment below:_  (1) add it to `OPPORTUNITY_TYPES` if it is new, and to `OPPORTUNITY_SEMANTIC_ENTITY_TYPES` (plus its literal type in `constants.d.ts`), and release `spacecat-shared-utils`; (2) bump data-access's utils pin and release data-access; (3) bump data-access in the writer (audit-worker) and the reader (api-service). The writer rejects the type until step 3, so index only after it. A new source kind is the same, in `OPPORTUNITY_SEMANTIC_SOURCE_TYPES`.
+_Superseded by the amendment below:_ Adding an entity type: (1) add it to `OPPORTUNITY_TYPES` if it is new, and to `OPPORTUNITY_SEMANTIC_ENTITY_TYPES` (plus its literal type in `constants.d.ts`), and release `spacecat-shared-utils`; (2) bump data-access's utils pin and release data-access; (3) bump data-access in the writer (audit-worker) and the reader (api-service). The writer rejects the type until step 3, so index only after it. A new source kind is the same, in `OPPORTUNITY_SEMANTIC_SOURCE_TYPES`.
 
 ### Amendment (2026-10-01): semantic-index redesign
 
@@ -63,12 +63,13 @@ The index was redesigned before it had external consumers (mysticat-data-service
 - **The shared layer now embeds.** `indexSemanticTopics` and `embedQueries` take an `EmbeddingProvider` (Decision 2) and call it. Transport, auth and retry stay in the client (Decision 3); data-access only decides which texts need a vector. That keeps the batching, dedupe and cache-first logic in one place instead of in every producer.
 - **The writer reads the query cache but never writes it.** Stored texts are normalized the same way as queries, so a topic already embedded for a query is reused. Only the read path (`embedQueries`) writes `semantic_query_embedding`, so the cache keeps meaning "texts users searched".
 - **Callers never pass the embedding generation.** The writer, `embedQueries` and both lookups read `SEMANTIC_MATCHING_CONFIG.embedding` themselves. After a model change the writer treats rows from the old generation as stale and overwrites them on the entity's next refresh.
-- **The read side is bounded.** `embedQueries` rejects texts that are empty or over `MAX_SOURCE_TEXT_LENGTH` once normalized and more than `MAX_QUERY_TEXTS` distinct texts, embeds the distinct misses in one call, and treats a cache read failure as all misses (logged), like the writer.
+- **The writer copies vectors from the query cache into index rows**, so a bad cache row would persist past cache eviction. This relies on PostgREST being reachable only internally, and on the cache helpers being internal to data-access (only `embedQueries` writes the cache).
+- **The read side is bounded.** `embedQueries` rejects texts that are empty or over `MAX_SOURCE_TEXT_LENGTH` once normalized and more than `MAX_QUERY_TEXTS` texts, embeds the distinct misses in one call, and treats a cache read failure as all misses (logged and returned as `cacheError`), like the writer.
 
 ## Consequences
 
 - `gpt-client` gains an embeddings surface and its own `AZURE_EMBEDDING_*` config (documented in its README); the intended cost, not a drawback.
-- The shared embedding model is a cross-repo contract: both the write and read paths must point at the **same deployment**. ANN search is **scoped to one generation** — `lookupOpportunitiesByVectors` (and the `rpc_opportunity_semantic_search` RPC) filter on `model`+`dims`, and the query cache is keyed on `model+dims`. This makes a **same-dims model change an online, forward-only re-embed** (chosen over an atomic swap — see Alternatives): the read path queries the new generation immediately while the write path back-populates it opportunity-by-opportunity on the normal refresh cadence. **Accepted cost:** during the migration window, opportunities not yet re-embedded don't match new-generation queries (reduced recall), self-healing as the backfill completes. A **dimension change** is out of this online path — it needs a new `vector(N)` column and an atomic swap, not a forward-only re-embed.
+- The shared embedding model is a cross-repo contract: both the write and read paths must point at the **same deployment**. ANN search is **scoped to one generation** — `lookupOpportunitiesByTopic` / `lookupSuggestionsByTopic` (and the search RPCs) filter on `model`+`dims`, read from `SEMANTIC_MATCHING_CONFIG.embedding`, and the query cache is keyed on `model+dims`. This makes a **same-dims model change an online, forward-only re-embed** (chosen over an atomic swap — see Alternatives): the read path queries the new generation immediately while the write path back-populates it entity-by-entity on the normal refresh cadence, overwriting old-generation rows in place. Entities that are never refreshed (e.g. closed opportunities) keep old-generation rows, which search ignores; the model-change runbook ends with a one-off sweep deleting rows whose `model`/`dims` is not current. **Accepted cost:** during the migration window, opportunities not yet re-embedded don't match new-generation queries (reduced recall), self-healing as the backfill completes. A **dimension change** is out of this online path — it needs a new `vector(N)` column and an atomic swap, not a forward-only re-embed.
 - _Superseded by the amendment below:_ **Deploy ordering:** `lookupOpportunitiesByVectors` calls `rpc_opportunity_semantic_search` with `p_source_types`/`p_entity_types` (lists) and `p_query_embeddings` (a list of vectors) plus `p_model`/`p_dims`, so the data-service release carrying that signature must be deployed **before** any consumer upgrades to this library version — otherwise PostgREST can't resolve the function (`PGRST202`).
 - _Superseded by the amendment below:_ **Registry version skew:** the registries check inputs, not results. A newer writer can index a type an older reader doesn't know, and with `entityTypes` omitted that `entity_type` comes back unvalidated, so a reader that must recognise every returned type passes `entityTypes` explicitly.
 - Consumers stay thin: they inject the `EmbeddingProvider`, own their cache/status mapping, and call the `data-access` helpers — no matching logic is reimplemented per consumer.
@@ -77,9 +78,9 @@ The index was redesigned before it had external consumers (mysticat-data-service
 
 - **A new `spacecat-shared-embedding-client` package.** Rejected for now: same vendor/auth as `gpt-client`; deferred behind the `EmbeddingProvider` seam so a later extraction is non-breaking.
 - **Overloading `AzureOpenAIClient`.** Rejected: embeddings are a different API surface; a distinct class keeps each client cohesive.
-- **Embedding inside the shared data-access layer.** Rejected: transport/auth belongs in a client (ADR-0001 boundary), and the write side (audit-worker) owns when/what to embed.
+- _Superseded by the amendment below:_ **Embedding inside the shared data-access layer.** Rejected: transport/auth belongs in a client (ADR-0001 boundary), and the write side (audit-worker) owns when/what to embed.
 - **Producer-namespaced source types (e.g. `offsite_topic`).** Rejected: `entity_type` already identifies the producer, so namespacing `source_type` would duplicate it. Generic kinds plus an optional `entityTypes` filter keep the two axes separate. Accepted cost: a new producer that indexes `topic` shows up in by-topics results by default unless callers pass `entityTypes`.
-- **A `*` (all types) wildcard.** Rejected: results would change silently whenever a new producer starts indexing, and an exact scan's cost grows with every type included. Callers that want every kind pass `Object.values(OPPORTUNITY_SEMANTIC_SOURCE_TYPES)` explicitly.
+- _Superseded by the amendment below:_ **A `*` (all types) wildcard.** Rejected: results would change silently whenever a new producer starts indexing, and an exact scan's cost grows with every type included. Callers that want every kind pass `Object.values(OPPORTUNITY_SEMANTIC_SOURCE_TYPES)` explicitly.
 - **Registries enforced in the DB (CHECK/enum).** Rejected: a migration for every new type; the registries live next to the helpers that enforce them.
 - **Validating `entityType` against `OPPORTUNITY_TYPES` directly.** Rejected: it saves no release steps (a new upstream type still needs the exact utils pin bumped and a data-access release) and only loses the opt-in: it would immediately admit every opportunity type already behind today's pin, so any of them could be indexed and searched without a deliberate registration.
 - **Embedding model label from env (e.g. derived from `AZURE_EMBEDDING_DEPLOYMENT`).** Rejected: writer and reader have separate env configs, so a mismatch would silently return no matches; a deployment name is an alias, not a model id; and changing the label is a re-embed, which should be a reviewed release.

packages/spacecat-shared-data-access/README.md

@@ -167,10 +167,10 @@ const result = await indexSemanticTopics(postgrestClient, embeddingClient, {
 // result: { entities: [{ entityId, fields: [{ matchFieldType, rejected, inserted, deleted, unchanged, ... }] }],
 //           embedded, cacheHits, cacheError? }
 
-// query side: validate (at most MAX_QUERY_TEXTS distinct texts, each up to 2048 chars), read the
+// query side: validate (at most MAX_QUERY_TEXTS texts, each up to 2048 chars once normalized), read the
 // cache (best-effort), embed the distinct misses in one call within a time budget.
 // cacheWrites (touch hits + store misses) is best-effort; await it or hand it to waitUntil.
-const { vectors, cacheWrites } = await embedQueries(postgrestClient, embeddingClient, texts, { log });
+const { vectors, cacheWrites, cacheError } = await embedQueries(postgrestClient, embeddingClient, texts, { log });
 
 // reader, batched: one best-first list of { entityId, entityType, score } per query vector.
 // matchFieldTypes / entityTypes / statuses are optional filters (omitted or [] means no filter).
@@ -191,7 +191,7 @@ The match type (`topic`, later `claim`) is never a parameter: each dimension has
 
 `SEMANTIC_MATCHING_CONFIG.embedding` (`model`, `dims`) is the embedding generation the writer and reader both use; the writer, `embedQueries` and the lookups read it from there rather than taking it as a parameter. It is a code constant, not env config, so writer and reader agree as long as both run the same data-access version; changing it means a re-embed plus upgrading both consumers. The re-embed is automatic: the writer treats rows from another generation as stale and overwrites them on the entity's next refresh.
 
-The writer reads the query cache but never writes it. The low-level cache helpers (`getQueryEmbeddings`, `upsertQueryEmbeddings`, `touchQueryEmbeddings`) are still exported.
+The writer reads the query cache but never writes it; only `embedQueries` does. The low-level cache helpers are internal, so nothing outside data-access can write vectors that the writer and readers trust.
 
 Calls are grouped internally (embeddings in batches of 256, writes and searches in groups of 20, id/hash reads and deletes in groups of 50), so callers pass whole lists rather than looping per item.
 

packages/spacecat-shared-data-access/src/service/index.js

@@ -76,9 +76,6 @@ export {
   embedQueries,
   lookupOpportunitiesByTopic,
   lookupSuggestionsByTopic,
-  getQueryEmbeddings,
-  upsertQueryEmbeddings,
-  touchQueryEmbeddings,
 } from '../util/semantic-index.utils.js';
 
 const createPostgrestService = (config, client = undefined) => {

packages/spacecat-shared-data-access/src/util/semantic-index.utils.js

@@ -70,8 +70,8 @@ export const QUERY_HASH_CHUNK_SIZE = 50;
 /** Texts per `createEmbeddings` call (Azure accepts up to 2048 inputs per request). */
 export const EMBEDDING_BATCH_SIZE = 256;
 
-/** Max distinct query texts per `embedQueries` call, so its misses fit one embed call. */
-export const MAX_QUERY_TEXTS = EMBEDDING_BATCH_SIZE;
+/** Max query texts per `embedQueries` call; keep <= `EMBEDDING_BATCH_SIZE` (one embed call). */
+export const MAX_QUERY_TEXTS = 256;
 
 /** Max text length; matches the `text` DB CHECKs. */
 export const MAX_SOURCE_TEXT_LENGTH = 2048;
@@ -714,18 +714,20 @@ async function embedWithinBudget(embeddingClient, texts, timeoutMs) {
 /**
  * Read side: one vector per query text (embedded in normalized form). One batched cache read, then
  * one embed call for the distinct misses within `timeoutMs`. A cache read failure only costs extra
- * embeddings (logged). The cache writes (access bump for hits, populate for misses) are best-effort
- * and come back as a promise the caller awaits alongside the search; they never fail the lookup.
+ * embeddings (logged and returned as `cacheError`). The cache writes (access bump for hits,
+ * populate for misses) are best-effort and come back as a promise the caller awaits alongside the
+ * search; they never fail the lookup.
  *
  * @param {object} postgrestClient - `@supabase/postgrest-js` client
  * @param {{createEmbeddings: (texts: string[]) => Promise<number[][]>}} embeddingClient
  * @param {string[]} texts - query texts; each must be non-empty and at most
- *   `MAX_SOURCE_TEXT_LENGTH` once normalized, and at most `MAX_QUERY_TEXTS` distinct texts
+ *   `MAX_SOURCE_TEXT_LENGTH` once normalized; at most `MAX_QUERY_TEXTS` texts
  * @param {object} [opts]
  * @param {number} [opts.timeoutMs=EMBEDDING_TIMEOUT_MS] - budget for the embed call
  * @param {object} [opts.log] - logger for non-fatal cache failures
  * @returns {Promise<{vectors: number[][], cacheWrites: Promise<unknown>, hits: number,
- *   misses: number}>} vectors in `texts` order; `hits` / `misses` count distinct texts
+ *   misses: number, cacheError?: Error}>} vectors in `texts` order; `hits` / `misses` count
+ *   distinct texts
  * @throws {EmbeddingUnavailableError} when the embed call fails, times out, or is malformed
  */
 export async function embedQueries(postgrestClient, embeddingClient, texts, {
@@ -739,19 +741,20 @@ export async function embedQueries(postgrestClient, embeddingClient, texts, {
   if (typeof timeoutMs !== 'number' || !Number.isFinite(timeoutMs) || timeoutMs <= 0) {
     throw new ValidationError('timeoutMs must be a positive finite number');
   }
+  // Total, not distinct: every entry gets a vector and a search, even a repeat.
+  if (texts.length > MAX_QUERY_TEXTS) {
+    throw new ValidationError(`texts must have at most ${MAX_QUERY_TEXTS} entries (got ${texts.length})`);
+  }
   const keys = texts.map((text, i) => {
     const cleaned = cleanTopicText(text);
     if (!cleaned) {
       throw new ValidationError(
-        `texts[${i}] must be a non-empty string of at most ${MAX_SOURCE_TEXT_LENGTH} characters`,
+        `texts[${i}] must be a non-empty string of at most ${MAX_SOURCE_TEXT_LENGTH} characters once normalized`,
       );
     }
     return cleaned.key;
   });
   const distinct = [...new Set(keys)];
-  if (distinct.length > MAX_QUERY_TEXTS) {
-    throw new ValidationError(`texts must have at most ${MAX_QUERY_TEXTS} distinct entries (got ${distinct.length})`);
-  }
   const { model, dims } = SEMANTIC_MATCHING_CONFIG.embedding;
   const cacheKey = { model, dims };
   // warn, not debug: a persistently failing cache turns every repeat query into an embed call.
@@ -761,6 +764,7 @@ export async function embedQueries(postgrestClient, embeddingClient, texts, {
 
   const vectorByKey = new Map();
   const hitHashes = [];
+  let cacheError;
   try {
     const cached = await getQueryEmbeddings(postgrestClient, { texts: distinct, ...cacheKey });
     cached.forEach((hit, i) => {
@@ -770,6 +774,7 @@ export async function embedQueries(postgrestClient, embeddingClient, texts, {
       }
     });
   } catch (e) {
+    cacheError = e;
     swallow('getQueryEmbeddings')(e);
   }
   const misses = distinct.filter((key) => !vectorByKey.has(key));
@@ -799,6 +804,7 @@ export async function embedQueries(postgrestClient, embeddingClient, texts, {
     cacheWrites,
     hits: distinct.length - misses.length,
     misses: misses.length,
+    ...(cacheError && { cacheError }),
   };
 }