Skip to content
This repository was archived by the owner on Oct 14, 2026. It is now read-only.

Commit 878e170

Browse files
Tanase Butcaruclaude
andcommitted
fix: strict write-path validation + generation-scoped ANN search (LLMO-7445)
Address PR review: - toRows validates model/dims + vector.length===dims and runs cleanTopicText (2048 guard); upsertQueryEmbedding length check - lookupOpportunitiesByVector scopes ANN search to model/dims (passes p_model/p_dims) - README named import; retryMaxDelayMs lower clamp; validateEmbeddingResponse integer index; truncate error body - tests: pagination >1000, chunk >20 + exact-20, full row shape, too-many/missing-index, deterministic retry-delay assertion Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent a953b87 commit 878e170

6 files changed

Lines changed: 215 additions & 30 deletions

File tree

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ Following ADR-0001's test (*true for every consumer → shared; specific to one
5050
## Consequences
5151

5252
- `gpt-client` gains an embeddings surface and its own `AZURE_EMBEDDING_*` config (documented in its README); the intended cost, not a drawback.
53-
- The shared embedding model is a cross-repo contract: both the write and read paths must point at the **same deployment**. A model change is a coordinated re-embed of the opportunity index; the query cache self-invalidates via its `model+dims` key.
53+
- The shared embedding model is a cross-repo contract: both the write and read paths must point at the **same deployment**. A model change is a coordinated re-embed of the opportunity index; the query cache self-invalidates via its `model+dims` key. ANN search is **scoped to one generation** — `lookupOpportunitiesByVector` (and the `rpc_opportunity_semantic_search` RPC) filter on `model`+`dims`, so an in-progress re-embed to a new model (same dims) never mixes generations in the ranking; not-yet-re-embedded opportunities simply don't match until the write path refreshes them (forward-only, self-healing).
5454
- 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.
5555

5656
## Alternatives considered

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

Lines changed: 33 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -148,27 +148,37 @@ function toRows({
148148
const seen = new Set();
149149
const rows = [];
150150
for (const src of list) {
151-
const { text, vector, sourceId } = src ?? {};
152-
if (typeof text !== 'string' || text.trim().length === 0) {
151+
// Text hygiene (drop-don't-throw): same rule as the read side, incl. the 2048-char bound.
152+
const cleaned = cleanTopicText(src?.text);
153+
if (cleaned === null) {
153154
// eslint-disable-next-line no-continue
154155
continue;
155156
}
156-
const normalized = normalizeText(text);
157-
const sourceHash = hashText(normalized);
157+
const { text, key } = cleaned;
158+
const sourceHash = hashText(key);
158159
if (seen.has(sourceHash)) {
159160
// eslint-disable-next-line no-continue
160161
continue;
161162
}
162163
seen.add(sourceHash);
164+
// Contract validation (this is the authoritative write side, so be at least as strict as the
165+
// query cache): a bad model/dims or a vector whose length disagrees with dims is a hard error,
166+
// not a silently-persisted row that surfaces later as wrong neighbours or an opaque DB failure.
167+
assertId(src.model, 'model');
168+
assertDims(src.dims);
169+
const embedding = serializeVector(src.vector);
170+
if (src.vector.length !== src.dims) {
171+
throw new ValidationError(`vector length ${src.vector.length} does not match dims ${src.dims}`);
172+
}
163173
rows.push({
164174
site_id: siteId,
165175
entity_id: entityId,
166176
entity_type: entityType,
167177
source_type: sourceType,
168-
source_id: sourceId ?? null,
178+
source_id: src.sourceId ?? null,
169179
source_hash: sourceHash,
170180
source_text: text,
171-
embedding: serializeVector(vector),
181+
embedding,
172182
model: src.model,
173183
dims: src.dims,
174184
});
@@ -334,28 +344,39 @@ export async function copyEntityVectors(postgrestClient, {
334344
* cannot be expressed through the PostgREST query builder). The RPC dedupes to distinct
335345
* opportunities keeping the best cosine similarity, applies the score floor, and limits to `k`.
336346
*
347+
* The search is scoped to one embedding **generation** via `model` + `dims`, so a coordinated
348+
* re-embed to a new model (same dims) never mixes generations in the ANN ranking — search only ever
349+
* compares vectors produced by the same model. Callers pass the model/dims the query was embedded
350+
* with (the same values stored on the index rows).
351+
*
337352
* @param {object} postgrestClient
338353
* @param {object} params
339354
* @param {string} params.siteId
340355
* @param {string} params.sourceType - `topic` | `claim` | …
341356
* @param {number[]} params.vector - the query embedding (same model/dims as the index)
357+
* @param {string} params.model - the embedding model the index + query share (generation scope)
358+
* @param {number} params.dims - the embedding dimension (generation scope)
342359
* @param {number} [params.k=10] - max opportunities to return
343360
* @param {number} [params.minScore=0] - cosine-similarity floor; matches below are dropped
344361
* @returns {Promise<Array<{entityId: string, entityType: string, score: number}>>}
345362
* ranked best-first
346363
*/
347364
export async function lookupOpportunitiesByVector(postgrestClient, {
348-
siteId, sourceType, vector, k = 10, minScore = 0,
365+
siteId, sourceType, vector, model, dims, k = 10, minScore = 0,
349366
} = {}) {
350367
assertClient(postgrestClient);
351368
assertId(siteId, 'siteId');
352369
assertId(sourceType, 'sourceType');
370+
assertId(model, 'model');
371+
assertDims(dims);
353372
const queryVector = serializeVector(vector);
354373

355374
const { data, error } = await postgrestClient.rpc(SEMANTIC_SEARCH_RPC, {
356375
p_site_id: siteId,
357376
p_source_type: sourceType,
358377
p_query_embedding: queryVector,
378+
p_model: model,
379+
p_dims: dims,
359380
p_limit: k,
360381
p_min_score: minScore,
361382
});
@@ -421,12 +442,16 @@ export async function upsertQueryEmbedding(postgrestClient, {
421442
throw new ValidationError('text is required');
422443
}
423444
const textHash = hashText(normalized);
445+
const embedding = serializeVector(vector);
446+
if (vector.length !== dims) {
447+
throw new ValidationError(`vector length ${vector.length} does not match dims ${dims}`);
448+
}
424449
const row = {
425450
text_hash: textHash,
426451
model,
427452
dims,
428453
normalized_text: normalized,
429-
embedding: serializeVector(vector),
454+
embedding,
430455
last_access_at: new Date().toISOString(),
431456
};
432457
const { error } = await postgrestClient

‎packages/spacecat-shared-data-access/test/unit/util/semantic-index.utils.test.js‎

Lines changed: 147 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,8 @@ function makeClient(config = {}) {
8181
state.inFilter = { column: col, values };
8282
return builder;
8383
},
84-
order() {
84+
order(col, opts) {
85+
state.order = { col, opts };
8586
return builder;
8687
},
8788
range(from, to) {
@@ -105,7 +106,12 @@ function makeClient(config = {}) {
105106
result = config.updateResult ?? { error: null };
106107
} else {
107108
calls.select.push({
108-
table, eqs: state.eqs, cols: state.cols, limited: state.limited,
109+
table,
110+
eqs: state.eqs,
111+
cols: state.cols,
112+
limited: state.limited,
113+
order: state.order,
114+
range: state.range,
109115
});
110116
if (state.limited) {
111117
result = config.getResult ?? { data: [], error: null };
@@ -226,6 +232,15 @@ describe('semantic-index.utils', () => {
226232
});
227233
});
228234

235+
it('clears the slice when sources is omitted entirely (undefined)', async () => {
236+
const client = makeClient();
237+
const count = await syncOpportunitySemantic(client, {
238+
siteId: SITE_ID, entityId: ENTITY_ID, entityType: ENTITY_TYPE, sourceType: SOURCE_TYPE,
239+
});
240+
expect(count).to.equal(0);
241+
expect(client.calls.delete).to.have.length(1);
242+
});
243+
229244
it('throws when a non-empty sources input yields no valid rows', async () => {
230245
const client = makeClient();
231246
await expect(syncOpportunitySemantic(client, {
@@ -251,12 +266,51 @@ describe('semantic-index.utils', () => {
251266
expect(client.calls.upsert).to.have.length(1);
252267
expect(client.calls.upsert[0].options).to.deep.equal({ onConflict: 'entity_id,source_type,source_hash' });
253268
const row = client.calls.upsert[0].rows[0];
254-
expect(row).to.include({ site_id: SITE_ID, entity_id: ENTITY_ID, source_type: SOURCE_TYPE });
269+
// Full row shape — cross-repo write contract (model/dims/original text/hash/null id).
270+
expect(row).to.include({
271+
site_id: SITE_ID,
272+
entity_id: ENTITY_ID,
273+
source_type: SOURCE_TYPE,
274+
entity_type: ENTITY_TYPE,
275+
model: 'azure/text-embedding-3-small',
276+
dims: 2,
277+
source_text: 'Running Shoes', // original case, not normalized
278+
source_id: null,
279+
});
255280
expect(row.embedding).to.equal('[0.1,0.2]');
281+
expect(row.source_hash).to.equal(hashText(normalizeText('Running Shoes')));
256282
expect(client.calls.delete).to.have.length(1);
257283
expect(client.calls.delete[0].inFilter.values).to.deep.equal(['stale-hash']);
258284
});
259285

286+
it('passes an explicit sourceId through and validates model/dims/vector length', async () => {
287+
const client = makeClient({ selectPages: [{ data: [], error: null }] });
288+
await syncOpportunitySemantic(client, {
289+
siteId: SITE_ID,
290+
entityId: ENTITY_ID,
291+
entityType: ENTITY_TYPE,
292+
sourceType: SOURCE_TYPE,
293+
sources: [src('Running Shoes', [0.1, 0.2], { sourceId: 'sid-1' })],
294+
});
295+
expect(client.calls.upsert[0].rows[0].source_id).to.equal('sid-1');
296+
297+
// model/dims/vector-length are hard contract errors on the authoritative write side.
298+
await expect(syncOpportunitySemantic(makeClient(), {
299+
siteId: SITE_ID,
300+
entityId: ENTITY_ID,
301+
entityType: ENTITY_TYPE,
302+
sourceType: SOURCE_TYPE,
303+
sources: [{ text: 'x', vector: [0.1, 0.2], dims: 2 }], // missing model
304+
})).to.be.rejectedWith(ValidationError, 'model is required');
305+
await expect(syncOpportunitySemantic(makeClient(), {
306+
siteId: SITE_ID,
307+
entityId: ENTITY_ID,
308+
entityType: ENTITY_TYPE,
309+
sourceType: SOURCE_TYPE,
310+
sources: [src('x', [0.1, 0.2], { dims: 1536 })], // vector length 2 != dims 1536
311+
})).to.be.rejectedWith(ValidationError, 'does not match dims');
312+
});
313+
260314
it('handles an empty read-back (nothing to prune)', async () => {
261315
const client = makeClient({ selectPages: [{ data: [], error: null }] });
262316
const count = await syncOpportunitySemantic(client, {
@@ -315,6 +369,57 @@ describe('semantic-index.utils', () => {
315369
siteId: SITE_ID, entityId: ENTITY_ID, entityType: ENTITY_TYPE, sourceType: SOURCE_TYPE, sources: [src('x')],
316370
})).to.be.rejectedWith(DataAccessError, 'Failed to prune');
317371
});
372+
373+
it('paginates the stale-hash read-back across pages (>DEFAULT_PAGE_SIZE)', async () => {
374+
const keptHash = hashText(normalizeText('kept'));
375+
const page1 = Array.from({ length: 1000 }, (_, i) => ({ source_hash: `h${i}` }));
376+
const client = makeClient({
377+
selectPages: [
378+
{ data: page1, error: null },
379+
{ data: [{ source_hash: keptHash }], error: null },
380+
],
381+
});
382+
await syncOpportunitySemantic(client, {
383+
siteId: SITE_ID, entityId: ENTITY_ID, entityType: ENTITY_TYPE, sourceType: SOURCE_TYPE, sources: [src('kept')],
384+
});
385+
// both page windows fetched, ordered by source_hash asc
386+
expect(client.calls.select).to.have.length(2);
387+
expect(client.calls.select[0].range).to.deep.equal([0, 999]);
388+
expect(client.calls.select[1].range).to.deep.equal([1000, 1999]);
389+
expect(client.calls.select[0].order).to.deep.equal({ col: 'source_hash', opts: { ascending: true } });
390+
// stale set spans page 1 (1000 hashes) -> pruned; keptHash (page 2) is retained
391+
const deleted = client.calls.delete.flatMap((d) => d.inFilter.values);
392+
expect(deleted).to.have.length(1000);
393+
expect(deleted).to.not.include(keptHash);
394+
});
395+
396+
it('chunks upsert + delete beyond SEMANTIC_CHUNK_SIZE (and single chunk at exactly the size)', async () => {
397+
// 21 sources -> 2 upsert chunks (20 + 1); read-back has 21 stale -> 2 delete chunks.
398+
const sources21 = Array.from({ length: 21 }, (_, i) => src(`topic ${i}`));
399+
const stale21 = Array.from({ length: 21 }, (_, i) => ({ source_hash: `stale-${i}` }));
400+
const client = makeClient({ selectPages: [{ data: stale21, error: null }] });
401+
await syncOpportunitySemantic(client, {
402+
siteId: SITE_ID,
403+
entityId: ENTITY_ID,
404+
entityType: ENTITY_TYPE,
405+
sourceType: SOURCE_TYPE,
406+
sources: sources21,
407+
});
408+
expect(client.calls.upsert).to.have.length(2);
409+
expect(client.calls.delete).to.have.length(2);
410+
411+
// exactly 20 sources -> a single upsert chunk (boundary).
412+
const sources20 = Array.from({ length: 20 }, (_, i) => src(`topic ${i}`));
413+
const client20 = makeClient({ selectPages: [{ data: [], error: null }] });
414+
await syncOpportunitySemantic(client20, {
415+
siteId: SITE_ID,
416+
entityId: ENTITY_ID,
417+
entityType: ENTITY_TYPE,
418+
sourceType: SOURCE_TYPE,
419+
sources: sources20,
420+
});
421+
expect(client20.calls.upsert).to.have.length(1);
422+
});
318423
});
319424

320425
describe('copyEntityVectors', () => {
@@ -358,37 +463,61 @@ describe('semantic-index.utils', () => {
358463
});
359464

360465
describe('lookupOpportunitiesByVector', () => {
361-
it('validates args + vector', async () => {
466+
const MODEL = 'azure/text-embedding-3-small';
467+
468+
it('validates args + vector + model/dims (generation scope)', async () => {
362469
await expect(lookupOpportunitiesByVector(makeClient(), {
363-
sourceType: SOURCE_TYPE, vector: [0.1],
470+
sourceType: SOURCE_TYPE, vector: [0.1], model: MODEL, dims: 2,
364471
}))
365472
.to.be.rejectedWith(ValidationError, 'siteId is required');
366-
await expect(lookupOpportunitiesByVector(makeClient(), { siteId: SITE_ID, vector: [0.1] }))
473+
await expect(lookupOpportunitiesByVector(makeClient(), {
474+
siteId: SITE_ID, vector: [0.1], model: MODEL, dims: 2,
475+
}))
367476
.to.be.rejectedWith(ValidationError, 'sourceType is required');
368477
await expect(lookupOpportunitiesByVector(makeClient(), {
369-
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [],
478+
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [0.1], dims: 2,
479+
}))
480+
.to.be.rejectedWith(ValidationError, 'model is required');
481+
await expect(lookupOpportunitiesByVector(makeClient(), {
482+
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [0.1], model: MODEL,
483+
}))
484+
.to.be.rejectedWith(ValidationError, 'dims must be a positive integer');
485+
await expect(lookupOpportunitiesByVector(makeClient(), {
486+
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [], model: MODEL, dims: 2,
370487
}))
371488
.to.be.rejectedWith(ValidationError, 'vector must be');
372489
});
373490

374-
it('calls the ANN RPC and maps rows, using defaults', async () => {
491+
it('calls the ANN RPC scoped to model/dims and maps rows, using defaults', async () => {
375492
const client = makeClient({
376493
rpcResult: { data: [{ entity_id: 'o1', entity_type: ENTITY_TYPE, score: 0.87 }], error: null },
377494
});
378495
const out = await lookupOpportunitiesByVector(client, {
379-
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [0.1, 0.2],
496+
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [0.1, 0.2], model: MODEL, dims: 2,
380497
});
381498
expect(out).to.deep.equal([{ entityId: 'o1', entityType: ENTITY_TYPE, score: 0.87 }]);
382499
expect(client.calls.rpc[0].name).to.equal(SEMANTIC_SEARCH_RPC);
383500
expect(client.calls.rpc[0].params).to.deep.equal({
384-
p_site_id: SITE_ID, p_source_type: SOURCE_TYPE, p_query_embedding: '[0.1,0.2]', p_limit: 10, p_min_score: 0,
501+
p_site_id: SITE_ID,
502+
p_source_type: SOURCE_TYPE,
503+
p_query_embedding: '[0.1,0.2]',
504+
p_model: MODEL,
505+
p_dims: 2,
506+
p_limit: 10,
507+
p_min_score: 0,
385508
});
386509
});
387510

388511
it('passes k/minScore and tolerates null data', async () => {
389512
const client = makeClient({ rpcResult: { data: null, error: null } });
390513
const out = await lookupOpportunitiesByVector(client, {
391-
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [0.1, 0.2], k: 5, minScore: 0.3,
514+
siteId: SITE_ID,
515+
sourceType: SOURCE_TYPE,
516+
vector: [0.1, 0.2],
517+
model: MODEL,
518+
dims: 2,
519+
k: 5,
520+
minScore: 0.3,
392521
});
393522
expect(out).to.deep.equal([]);
394523
expect(client.calls.rpc[0].params).to.include({ p_limit: 5, p_min_score: 0.3 });
@@ -397,7 +526,7 @@ describe('semantic-index.utils', () => {
397526
it('wraps an RPC error', async () => {
398527
const client = makeClient({ rpcResult: { data: null, error: { message: 'boom' } } });
399528
await expect(lookupOpportunitiesByVector(client, {
400-
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [0.1],
529+
siteId: SITE_ID, sourceType: SOURCE_TYPE, vector: [0.1], model: MODEL, dims: 2,
401530
})).to.be.rejectedWith(DataAccessError, 'Failed semantic search');
402531
});
403532
});
@@ -444,9 +573,13 @@ describe('semantic-index.utils', () => {
444573
await expect(upsertQueryEmbedding(makeClient(), { text: 'x', model: MODEL, vector: [0.1] }))
445574
.to.be.rejectedWith(ValidationError, 'dims must be a positive integer');
446575
await expect(upsertQueryEmbedding(makeClient(), {
447-
text: ' ', model: MODEL, dims: 2, vector: [0.1],
576+
text: ' ', model: MODEL, dims: 2, vector: [0.1, 0.2],
448577
}))
449578
.to.be.rejectedWith(ValidationError, 'text is required');
579+
await expect(upsertQueryEmbedding(makeClient(), {
580+
text: 'x', model: MODEL, dims: 2, vector: [0.1], // length 1 != dims 2
581+
}))
582+
.to.be.rejectedWith(ValidationError, 'does not match dims');
450583
const client = makeClient();
451584
const h = await upsertQueryEmbedding(client, {
452585
text: 'Running Shoes', model: MODEL, dims: 2, vector: [0.1, 0.2],
@@ -462,7 +595,7 @@ describe('semantic-index.utils', () => {
462595
it('upsertQueryEmbedding wraps an error', async () => {
463596
const client = makeClient({ upsertResult: { error: { message: 'boom' } } });
464597
await expect(upsertQueryEmbedding(client, {
465-
text: 'x', model: MODEL, dims: 2, vector: [0.1],
598+
text: 'x', model: MODEL, dims: 2, vector: [0.1, 0.2],
466599
})).to.be.rejectedWith(DataAccessError, 'Failed to upsert semantic_query_embedding');
467600
});
468601

‎packages/spacecat-shared-gpt-client/README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,7 @@ These parameters can be set through environment variables (via `AzureEmbeddingCl
155155
### Usage Example
156156

157157
```javascript
158-
import AzureEmbeddingClient from '@adobe/spacecat-shared-gpt-client';
158+
import { AzureEmbeddingClient } from '@adobe/spacecat-shared-gpt-client';
159159

160160
// Assuming AZURE_EMBEDDING_DEPLOYMENT (+ AZURE_OPENAI_* or AZURE_EMBEDDING_*) are set on context.env
161161
const client = AzureEmbeddingClient.createFrom(context);

0 commit comments

Comments
 (0)