Skip to content

Commit 6293aad

Browse files
tofikwestclaude
andauthored
fix(risk-treatment): rank tasks by exact in-process cosine, not filtered ANN (#3359)
Draft treatment plans returned "0 tasks and 0 controls" because findSimilarTasks issued a metadata-filtered ANN query (organizationId AND sourceType = "task") against a single 180k+ vector shared-namespace index. Upstash applies metadata filters during approximate HNSW traversal, so a highly selective per-org filter makes the traversal exhaust its candidate budget on nearer, non-matching vectors before reaching the org's tasks — returning 0 candidates even at topK=1000 when relevant tasks exist. It gets worse as the shared index grows. An org holds at most low-hundreds of tasks, so enumerate the org's task vectors by their `task_${org}_` id prefix and score them exactly in-process. This is the right-sized tool at this scale: no recall loss, no data migration. Controls are derived from the suggested tasks, so restoring task recall restores both. Scores use `(1 + cos) / 2` to match Upstash's COSINE score scale, so the downstream department boost / threshold in linkSuggestions and the reranker's cosineScore hint are unchanged. Verified end-to-end against production data. Isolated to findSimilarTasks; the orphan prune and in-scope filtering in runLinkage are unchanged and still compose correctly. Fixes CS-681 Claude-Session: https://claude.ai/code/session_014LMSrxyX5U6gz8QZqJAGhc Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 73ece08 commit 6293aad

2 files changed

Lines changed: 201 additions & 44 deletions

File tree

‎apps/app/src/lib/embedding/embedding.spec.ts‎

Lines changed: 98 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,11 @@ vi.mock('ai', () => ({
2929
})),
3030
}));
3131

32+
import { embedMany } from 'ai';
3233
import {
3334
upsertEntityEmbeddings,
3435
findSimilarTasks,
36+
cosineToUnitScore,
3537
waitForIndexed,
3638
pruneOrphanTaskVectors,
3739
} from './index';
@@ -168,11 +170,42 @@ describe('upsertEntityEmbeddings', () => {
168170
});
169171
});
170172

173+
describe('cosineToUnitScore', () => {
174+
it('maps identical vectors to 1, opposite to 0, orthogonal to 0.5', () => {
175+
expect(cosineToUnitScore([1, 0], [1, 0])).toBeCloseTo(1, 10);
176+
expect(cosineToUnitScore([1, 0], [-1, 0])).toBeCloseTo(0, 10);
177+
expect(cosineToUnitScore([1, 0], [0, 1])).toBeCloseTo(0.5, 10);
178+
});
179+
180+
it('is magnitude-invariant (pure cosine) and matches Upstash (1+cos)/2', () => {
181+
// Same direction, different magnitudes → still 1.
182+
expect(cosineToUnitScore([2, 0], [5, 0])).toBeCloseTo(1, 10);
183+
// cos = 0.6 → (1 + 0.6) / 2 = 0.8
184+
expect(cosineToUnitScore([3, 4], [3, 0])).toBeCloseTo(0.8, 10);
185+
});
186+
187+
it('returns 0 for a zero vector instead of NaN', () => {
188+
expect(cosineToUnitScore([0, 0], [1, 1])).toBe(0);
189+
expect(cosineToUnitScore([1, 1], [0, 0])).toBe(0);
190+
});
191+
});
192+
171193
describe('findSimilarTasks', () => {
172-
it('queries with org + sourceType filter and returns id+score+department', async () => {
173-
queryMock.mockResolvedValueOnce([
174-
{ id: 'task_org_1_tsk_a', score: 0.82, metadata: { sourceId: 'tsk_a', department: 'hr' } },
175-
{ id: 'task_org_1_tsk_b', score: 0.71, metadata: { sourceId: 'tsk_b', department: 'none' } },
194+
// Range page helper — `nextCursor: ''` signals the enumeration is drained.
195+
function taskPage(
196+
vectors: Array<{ id: string; vector: number[]; metadata?: Record<string, unknown> }>,
197+
) {
198+
rangeMock.mockResolvedValueOnce({ nextCursor: '', vectors });
199+
}
200+
201+
it('enumerates the org task vectors by prefix and ranks them in-process by cosine', async () => {
202+
// Query points along [1,0]; task vectors are chosen so the cosine ordering
203+
// is deterministic: tsk_a (identical) > tsk_b (orthogonal) > tsk_c (opposite).
204+
vi.mocked(embedMany).mockResolvedValueOnce({ embeddings: [[1, 0]] } as never);
205+
taskPage([
206+
{ id: 'task_org_1_tsk_b', vector: [0, 1], metadata: { sourceId: 'tsk_b', department: 'none' } },
207+
{ id: 'task_org_1_tsk_a', vector: [1, 0], metadata: { sourceId: 'tsk_a', department: 'hr' } },
208+
{ id: 'task_org_1_tsk_c', vector: [-1, 0], metadata: { sourceId: 'tsk_c' } },
176209
]);
177210

178211
const results = await findSimilarTasks({
@@ -181,42 +214,88 @@ describe('findSimilarTasks', () => {
181214
topK: 10,
182215
});
183216

184-
expect(queryMock).toHaveBeenCalledWith(
217+
// Enumerates via prefix range WITH the stored vectors — never an ANN query.
218+
expect(rangeMock).toHaveBeenCalledWith(
185219
expect.objectContaining({
186-
topK: 10,
220+
cursor: '0',
221+
prefix: 'task_org_1_',
222+
includeVectors: true,
187223
includeMetadata: true,
188-
filter: 'organizationId = "org_1" AND sourceType = "task"',
189224
}),
190225
);
226+
expect(queryMock).not.toHaveBeenCalled();
227+
// Sorted by score desc, scores on Upstash's (1+cos)/2 scale.
191228
expect(results).toEqual([
192-
{ id: 'tsk_a', score: 0.82, department: 'hr' },
193-
{ id: 'tsk_b', score: 0.71, department: 'none' },
229+
{ id: 'tsk_a', score: 1, department: 'hr' },
230+
{ id: 'tsk_b', score: 0.5, department: 'none' },
231+
{ id: 'tsk_c', score: 0, department: undefined },
194232
]);
195233
});
196234

197-
it('returns empty when query text is empty', async () => {
235+
it('applies the topK cap after ranking', async () => {
236+
vi.mocked(embedMany).mockResolvedValueOnce({ embeddings: [[1, 0]] } as never);
237+
taskPage([
238+
{ id: 'task_org_1_tsk_a', vector: [1, 0], metadata: { sourceId: 'tsk_a' } },
239+
{ id: 'task_org_1_tsk_b', vector: [0.9, 0.1], metadata: { sourceId: 'tsk_b' } },
240+
{ id: 'task_org_1_tsk_c', vector: [-1, 0], metadata: { sourceId: 'tsk_c' } },
241+
]);
242+
198243
const results = await findSimilarTasks({
199244
organizationId: 'org_1',
200-
queryText: ' ',
245+
queryText: 'phishing risk',
246+
topK: 2,
201247
});
248+
249+
expect(results).toHaveLength(2);
250+
expect(results.map((r) => r.id)).toEqual(['tsk_a', 'tsk_b']);
251+
});
252+
253+
it('paginates the enumeration across cursor pages', async () => {
254+
vi.mocked(embedMany).mockResolvedValueOnce({ embeddings: [[1, 0]] } as never);
255+
rangeMock
256+
.mockResolvedValueOnce({
257+
nextCursor: 'cursor_2',
258+
vectors: [{ id: 'task_org_1_tsk_p1', vector: [1, 0], metadata: { sourceId: 'tsk_p1' } }],
259+
})
260+
.mockResolvedValueOnce({
261+
nextCursor: '',
262+
vectors: [{ id: 'task_org_1_tsk_p2', vector: [0, 1], metadata: { sourceId: 'tsk_p2' } }],
263+
});
264+
265+
const results = await findSimilarTasks({ organizationId: 'org_1', queryText: 'x' });
266+
267+
expect(rangeMock).toHaveBeenCalledTimes(2);
268+
expect(rangeMock.mock.calls[1][0]).toEqual(
269+
expect.objectContaining({ cursor: 'cursor_2', prefix: 'task_org_1_' }),
270+
);
271+
// A task from page 2 is included in the ranking.
272+
expect(results.map((r) => r.id).sort()).toEqual(['tsk_p1', 'tsk_p2']);
273+
});
274+
275+
it('returns empty when query text is empty (no embed, no range)', async () => {
276+
const results = await findSimilarTasks({ organizationId: 'org_1', queryText: ' ' });
202277
expect(results).toEqual([]);
278+
expect(rangeMock).not.toHaveBeenCalled();
203279
expect(queryMock).not.toHaveBeenCalled();
204280
});
205281

282+
it('returns empty when the org has no task vectors', async () => {
283+
vi.mocked(embedMany).mockResolvedValueOnce({ embeddings: [[1, 0]] } as never);
284+
taskPage([]);
285+
const results = await findSimilarTasks({ organizationId: 'org_1', queryText: 'x' });
286+
expect(results).toEqual([]);
287+
});
288+
206289
it('recovers the raw task id from the prefix when metadata.sourceId is missing', async () => {
207290
// Legacy vector with no sourceId metadata. Returning the prefixed embedding
208291
// id would make runLinkage drop it as "not in live scope" (taskById is keyed
209292
// by raw ids), silently starving suggestions. (cubic P1)
210-
queryMock.mockResolvedValueOnce([
211-
{ id: 'task_org_1_tsk_legacy', score: 0.8, metadata: {} },
212-
]);
293+
vi.mocked(embedMany).mockResolvedValueOnce({ embeddings: [[1, 0]] } as never);
294+
taskPage([{ id: 'task_org_1_tsk_legacy', vector: [1, 0], metadata: {} }]);
213295

214-
const results = await findSimilarTasks({
215-
organizationId: 'org_1',
216-
queryText: 'phishing risk',
217-
});
296+
const results = await findSimilarTasks({ organizationId: 'org_1', queryText: 'phishing risk' });
218297

219-
expect(results).toEqual([{ id: 'tsk_legacy', score: 0.8, department: undefined }]);
298+
expect(results).toEqual([{ id: 'tsk_legacy', score: 1, department: undefined }]);
220299
});
221300
});
222301

‎apps/app/src/lib/embedding/index.ts‎

Lines changed: 103 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -57,13 +57,14 @@ export interface SimilarTaskResult {
5757
const EMBEDDING_MODEL = 'text-embedding-3-large';
5858
const EMBEDDING_DIMENSIONS = 1536;
5959
const DEFAULT_TOP_K = 25;
60-
// Page size for the orphan-vector sweep. `pruneOrphanTaskVectors` paginates an
61-
// org's task vectors via `range` over their shared id prefix, so there is no
62-
// top-K ceiling — this only controls how many vectors come back per round trip.
63-
const ORPHAN_SCAN_PAGE_SIZE = 1000;
60+
// Pagination for enumerating an org's task vectors by their shared id prefix —
61+
// used by both `findSimilarTasks` (exact in-process ranking) and
62+
// `pruneOrphanTaskVectors` (orphan sweep). Prefix enumeration has no top-K
63+
// ceiling; page size only controls how many vectors come back per round trip.
64+
const TASK_VECTOR_PAGE_SIZE = 1000;
6465
// Backstop so a misbehaving cursor can't loop forever. 500 pages × 1000 = 500k
6566
// task vectors, far beyond any real org — hitting it signals a bug, not scale.
66-
const ORPHAN_SCAN_MAX_PAGES = 500;
67+
const TASK_VECTOR_MAX_PAGES = 500;
6768

6869
let cachedIndex: Index | null = null;
6970

@@ -187,9 +188,90 @@ export async function upsertEntityEmbeddings({
187188
};
188189
}
189190

191+
interface OrgTaskVector {
192+
/** raw task id (sourceId), not the prefixed embedding id */
193+
sourceId: string;
194+
vector: number[];
195+
department?: string;
196+
}
197+
198+
/**
199+
* Enumerate an org's task vectors (with their embeddings) by their shared id
200+
* prefix. Every task vector is keyed `task_${organizationId}_${sourceId}` (see
201+
* `embeddingId`), so a cursor-paginated `range` returns exactly this org's task
202+
* vectors — no metadata filter, no top-K ceiling.
203+
*/
204+
async function fetchOrgTaskVectors(organizationId: string): Promise<OrgTaskVector[]> {
205+
const index = getIndex();
206+
const prefix = embeddingIdPrefix('task', organizationId);
207+
208+
const out: OrgTaskVector[] = [];
209+
let cursor: string | number = '0';
210+
let enumeratedFully = false;
211+
for (let page = 0; page < TASK_VECTOR_MAX_PAGES; page++) {
212+
const { vectors, nextCursor }: RangeResult = await index.range({
213+
cursor,
214+
limit: TASK_VECTOR_PAGE_SIZE,
215+
prefix,
216+
includeVectors: true,
217+
includeMetadata: true,
218+
});
219+
for (const r of vectors) {
220+
if (!r.vector) continue; // defensive: skip any vector row missing embeddings
221+
const meta = (r.metadata ?? {}) as { sourceId?: string; department?: string };
222+
out.push({
223+
sourceId: meta.sourceId ?? sourceIdFromEmbeddingId('task', organizationId, String(r.id)),
224+
vector: r.vector as number[],
225+
department: meta.department ?? undefined,
226+
});
227+
}
228+
if (!nextCursor) {
229+
enumeratedFully = true;
230+
break;
231+
}
232+
cursor = nextCursor;
233+
}
234+
if (!enumeratedFully) {
235+
console.warn(
236+
`[embedding] task-vector enumeration hit the ${TASK_VECTOR_MAX_PAGES}-page cap for org ${organizationId} after ${out.length} vector(s); ranking may be incomplete`,
237+
);
238+
}
239+
return out;
240+
}
241+
242+
/**
243+
* Cosine similarity mapped to Upstash's COSINE score scale, `(1 + cos) / 2`, so
244+
* scores stay in [0, 1] and are drop-in compatible with the department boost /
245+
* threshold in `linkSuggestions` and the reranker's `cosineScore` hint.
246+
*/
247+
export function cosineToUnitScore(a: number[], b: number[]): number {
248+
const len = Math.min(a.length, b.length);
249+
let dot = 0;
250+
let normA = 0;
251+
let normB = 0;
252+
for (let i = 0; i < len; i++) {
253+
dot += a[i] * b[i];
254+
normA += a[i] * a[i];
255+
normB += b[i] * b[i];
256+
}
257+
if (normA === 0 || normB === 0) return 0;
258+
const cos = dot / (Math.sqrt(normA) * Math.sqrt(normB));
259+
return (1 + cos) / 2;
260+
}
261+
190262
/**
191263
* Find the top-K most similar Tasks in an org for a given query string.
192264
* Returns raw task ids (not prefixed embedding ids) along with score + department.
265+
*
266+
* Retrieval enumerates the org's task vectors by id prefix and scores them
267+
* exactly in-process, rather than issuing a metadata-filtered ANN query. A
268+
* filtered `query` (`organizationId AND sourceType`) collapses to near-zero
269+
* recall here: one org is a tiny slice of a 180k+ vector shared-namespace index,
270+
* so Upstash's approximate traversal exhausts its candidate budget on nearer,
271+
* non-matching vectors before reaching this org's tasks — returning 0 candidates
272+
* even when relevant tasks exist, which starved treatment plans of every
273+
* suggestion (CS-681). An org holds at most low-hundreds of tasks, so exact
274+
* scoring over the full set is both correct (no recall loss) and cheap.
193275
*/
194276
export async function findSimilarTasks({
195277
organizationId,
@@ -203,23 +285,19 @@ export async function findSimilarTasks({
203285
values: [queryText],
204286
providerOptions: { openai: { dimensions: EMBEDDING_DIMENSIONS } },
205287
});
206-
207-
const index = getIndex();
208-
const results = await index.query({
209-
vector: embeddings[0],
210-
topK,
211-
includeMetadata: true,
212-
filter: `organizationId = "${organizationId}" AND sourceType = "task"`,
213-
});
214-
215-
return results.map((r) => {
216-
const meta = (r.metadata ?? {}) as { sourceId?: string; department?: string };
217-
return {
218-
id: meta.sourceId ?? sourceIdFromEmbeddingId('task', organizationId, String(r.id)),
219-
score: r.score,
220-
department: meta.department ?? undefined,
221-
};
222-
});
288+
const queryVector = embeddings[0];
289+
290+
const taskVectors = await fetchOrgTaskVectors(organizationId);
291+
if (taskVectors.length === 0) return [];
292+
293+
return taskVectors
294+
.map((t) => ({
295+
id: t.sourceId,
296+
score: cosineToUnitScore(queryVector, t.vector),
297+
department: t.department,
298+
}))
299+
.sort((a, b) => b.score - a.score)
300+
.slice(0, topK);
223301
}
224302

225303
export interface PruneOrphanTaskVectorsResult {
@@ -267,12 +345,12 @@ export async function pruneOrphanTaskVectors({
267345
let scanned = 0;
268346
let cursor: string | number = '0';
269347
let enumeratedFully = false;
270-
for (let page = 0; page < ORPHAN_SCAN_MAX_PAGES; page++) {
348+
for (let page = 0; page < TASK_VECTOR_MAX_PAGES; page++) {
271349
// Annotate the result explicitly: `cursor = nextCursor` would otherwise make
272350
// TS infer `nextCursor`'s type from a binding that depends on it (TS7022).
273351
const { vectors, nextCursor }: RangeResult = await index.range({
274352
cursor,
275-
limit: ORPHAN_SCAN_PAGE_SIZE,
353+
limit: TASK_VECTOR_PAGE_SIZE,
276354
prefix,
277355
includeMetadata: true,
278356
});
@@ -292,7 +370,7 @@ export async function pruneOrphanTaskVectors({
292370
}
293371
if (!enumeratedFully) {
294372
console.warn(
295-
`[embedding] orphan sweep hit the ${ORPHAN_SCAN_MAX_PAGES}-page cap for org ${organizationId} after scanning ${scanned} vector(s); enumeration may be incomplete`,
373+
`[embedding] orphan sweep hit the ${TASK_VECTOR_MAX_PAGES}-page cap for org ${organizationId} after scanning ${scanned} vector(s); enumeration may be incomplete`,
296374
);
297375
}
298376

0 commit comments

Comments
 (0)