diff --git a/onlion/backend/monitoring.mjs b/onlion/backend/monitoring.mjs index acc0a597..b9e0c71b 100644 --- a/onlion/backend/monitoring.mjs +++ b/onlion/backend/monitoring.mjs @@ -1,87 +1,233 @@ /* eslint-disable no-console -- deliberate test/diagnostic output */ -import {AsyncLocalStorage} from 'node:async_hooks'; -import {createHash} from 'node:crypto'; -const requests=new AsyncLocalStorage(),stages=new WeakMap(); -const ERROR_KEY='monitoring/recent-failures.json',MAX_ERRORS=20,RETENTION_MS=86400000; -const localFailures=[]; -let waiting=0,backlogSince=null,lastEmbedding=null; -const running=new Map(); -export const withRequest=(context,operation)=>requests.run(context,operation); -export function requestIdentity(userId,requestId){ - const context=requests.getStore();if(!context)return; - for(const [name,value] of Object.entries({user_id:userId,request_id:requestId})){ - if(typeof value==='string')context[name+'_hash']=createHash('sha256').update(value).digest('hex').slice(0,24); +import { AsyncLocalStorage } from "node:async_hooks"; +import { createHash } from "node:crypto"; +const requests = new AsyncLocalStorage(), + stages = new WeakMap(); +const ERROR_KEY = "monitoring/recent-failures.json", + MAX_ERRORS = 20, + RETENTION_MS = 86400000; +const localFailures = []; +let waiting = 0, + backlogSince = null, + lastEmbedding = null; +const running = new Map(); +export const withRequest = (context, operation) => requests.run(context, operation); +export function requestIdentity(userId, requestId) { + const context = requests.getStore(); + if (!context) return; + for (const [name, value] of Object.entries({ user_id: userId, request_id: requestId })) { + if (typeof value === "string") + context[name + "_hash"] = createHash("sha256").update(value).digest("hex").slice(0, 24); } } -export function observedStage(name,operation){ - try{ - const result=operation(); - if(result&&typeof result.then==='function')return result.catch(error=>{if(error&&typeof error==='object'&&!stages.has(error))stages.set(error,name);throw error;}); +export function observedStage(name, operation) { + try { + const result = operation(); + if (result && typeof result.then === "function") + return result.catch((error) => { + if (error && typeof error === "object" && !stages.has(error)) stages.set(error, name); + throw error; + }); return result; - }catch(error){if(error&&typeof error==='object'&&!stages.has(error))stages.set(error,name);throw error;} + } catch (error) { + if (error && typeof error === "object" && !stages.has(error)) stages.set(error, name); + throw error; + } +} +export function queueAccepted() { + if (waiting++ === 0) backlogSince = Date.now(); +} +export function queueStarted(context) { + waiting--; + if (waiting === 0) backlogSince = null; + running.set(context.trace_id, { + trace_id: context.trace_id, + route: context.route, + started_at: Date.now(), + }); } -export function queueAccepted(){if(waiting++===0)backlogSince=Date.now();} -export function queueStarted(context){waiting--;if(waiting===0)backlogSince=null;running.set(context.trace_id,{trace_id:context.trace_id,route:context.route,started_at:Date.now()});} -export function queueFinished(){const context=requests.getStore();if(context)running.delete(context.trace_id);} -export function queueStatus(){ - const now=Date.now(),active=running.values().next().value;return {scope:'current_isolate',waiting,running:running.size, - backlog_ms:backlogSince===null?0:now-backlogSince, - active:active?{...active,elapsed_ms:now-active.started_at}:null}; +export function queueFinished() { + const context = requests.getStore(); + if (context) running.delete(context.trace_id); } -export function classifyError(error){ - const message=String(error?.message??''); - const upstream=/embedding server returned\s+(\d{3})/i.exec(message); - const sqlite=/SQLITE_[A-Z_]+/.exec(message)?.[0]; - let code=typeof error?.code==='string'&&/^[a-zA-Z0-9_]{1,64}$/.test(error.code)?error.code:null; - code??=upstream?'embedding_http_'+upstream[1]:sqlite==='SQLITE_FULL'?'user_storage_limit':sqlite?'sqlite_error':error?.name==='TimeoutError'?'dependency_timeout':error?.name==='AbortError'?'operation_aborted':'service_unavailable'; - const summaries={user_storage_limit:'数据库容量或工作分片已达上限',part_storage_limit:'本批数据超过分片限制',request_receipt_limit:'写入回执数量已达上限',write_conflict:'并发提交冲突，重试次数已用完',snapshot_changed:'读取的快照已被替换',storage_unavailable:'持久存储不可用',storage_corrupt:'存储格式异常',not_configured:'服务鉴权未配置',reclamation_in_progress:'正在完全回收，记忆存取暂停',memory_reclaimed:'回收期间的旧请求已停止',user_reset:'用户数据已被重置',sqlite_error:'SQLite 执行失败',dependency_timeout:'外部依赖超时',operation_aborted:'操作被中止',embedding_timeout:'远程向量模型超时',embedding_network_error:'远程向量模型连接失败',embedding_failed:'远程向量模型调用失败',service_unavailable:'服务处理失败，需要凭关联编号查询平台日志'}; - const summary=upstream?'远程向量模型返回 HTTP '+upstream[1]:(summaries[code]??'服务操作失败：'+code); - return {code,summary,...(upstream?{upstream_status:Number(upstream[1])}:{}),...(sqlite?{sqlite_code:sqlite}:{})}; +export function queueStatus() { + const now = Date.now(), + active = running.values().next().value; + return { + scope: "current_isolate", + waiting, + running: running.size, + backlog_ms: backlogSince === null ? 0 : now - backlogSince, + active: active ? { ...active, elapsed_ms: now - active.started_at } : null, + }; } -async function persistFailure(bucket,entry){ - for(let attempt=0;attempt<2;attempt++){ - const old=await bucket.get(ERROR_KEY); - const previous=old?await old.json():[]; - const entries=[entry,...previous.filter(e=>e.id!==entry.id&&Date.now()-Date.parse(e.at) e.id !== entry.id && Date.now() - Date.parse(e.at) < RETENTION_MS), + ].slice(0, MAX_ERRORS); + const saved = await bucket.put(ERROR_KEY, JSON.stringify(entries), { + onlyIf: old ? { etagMatches: old.etag } : { etagDoesNotMatch: "*" }, + httpMetadata: { contentType: "application/json" }, + }); + if (saved) return; } } -export function reportFailure(error,status,kind='request_failure'){ - const context=requests.getStore();if(!context)return; - const entry={id:crypto.randomUUID(),at:new Date().toISOString(),kind,status, - trace_id:context.trace_id,route:context.route,stage:stages.get(error)??'handler', - ...classifyError(error),...(context.user_id_hash?{user_id_hash:context.user_id_hash}:{}), - ...(context.request_id_hash?{request_id_hash:context.request_id_hash}:{}), +export function reportFailure(error, status, kind = "request_failure") { + const context = requests.getStore(); + if (!context) return; + const entry = { + id: crypto.randomUUID(), + at: new Date().toISOString(), + kind, + status, + trace_id: context.trace_id, + route: context.route, + stage: stages.get(error) ?? "handler", + ...classifyError(error), + ...(context.user_id_hash ? { user_id_hash: context.user_id_hash } : {}), + ...(context.request_id_hash ? { request_id_hash: context.request_id_hash } : {}), // NMG_DEBUG_ERRORS=1 surfaces the exception name/message for diagnosis; the // default record still carries no exception text, stack or request content. - ...(context.env.NMG_DEBUG_ERRORS==='1'?{debug:{name:error?.name??null,message:String(error?.message??'').slice(0,300)}}:{})}; - localFailures.unshift(entry);localFailures.length=Math.min(localFailures.length,MAX_ERRORS); + ...(context.env.NMG_DEBUG_ERRORS === "1" + ? { + debug: { name: error?.name ?? null, message: String(error?.message ?? "").slice(0, 300) }, + } + : {}), + }; + localFailures.unshift(entry); + localFailures.length = Math.min(localFailures.length, MAX_ERRORS); // No exception text, stack, request body, source statements or credentials. - console.error(JSON.stringify({event:'nmg_failure',...entry})); - if(context.env.MEMORY&&context.execution?.waitUntil){ - context.execution.waitUntil(persistFailure(context.env.MEMORY,entry).catch(()=>console.error(JSON.stringify({event:'nmg_monitor_persistence_failed',trace_id:entry.trace_id})))); + console.error(JSON.stringify({ event: "nmg_failure", ...entry })); + if (context.env.MEMORY && context.execution?.waitUntil) { + context.execution.waitUntil( + persistFailure(context.env.MEMORY, entry).catch(() => + console.error( + JSON.stringify({ event: "nmg_monitor_persistence_failed", trace_id: entry.trace_id }), + ), + ), + ); } } -export async function embeddingOperation(operation){ - try{const value=await observedStage('embedding',operation);lastEmbedding={status:'ok',at:new Date().toISOString()};return value;} - catch(caught){ - const error=caught&&typeof caught==='object'?caught:new Error('Embedding failed'); - if(!error.code){error.code=/embedding server returned/.test(String(error.message))?classifyError(error).code:error.name==='TimeoutError'?'embedding_timeout':error.name==='TypeError'?'embedding_network_error':'embedding_failed';} - lastEmbedding={status:'failed',at:new Date().toISOString(),...classifyError(error)}; - reportFailure(error,null,'embedding_failure');throw error; +export async function embeddingOperation(operation) { + // The call crosses a region boundary (function in ap-hongkong, provider elsewhere), so its + // duration is the number that decides whether a per-attempt timeout is sane. Recording only + // the status left that unknown through two incidents. + const started = performance.now(); + try { + const value = await observedStage("embedding", operation); + lastEmbedding = { + status: "ok", + at: new Date().toISOString(), + ms: Math.round(performance.now() - started), + }; + return value; + } catch (caught) { + const error = caught && typeof caught === "object" ? caught : new Error("Embedding failed"); + if (!error.code) { + error.code = /embedding server returned/.test(String(error.message)) + ? classifyError(error).code + : error.name === "TimeoutError" + ? "embedding_timeout" + : error.name === "TypeError" + ? "embedding_network_error" + : "embedding_failed"; + } + lastEmbedding = { + status: "failed", + at: new Date().toISOString(), + ms: Math.round(performance.now() - started), + ...classifyError(error), + }; + reportFailure(error, null, "embedding_failure"); + throw error; } } -export async function recentFailures(bucket){ - const object=await bucket.get(ERROR_KEY); - const persisted=object?await object.json():[]; - return [...new Map([...localFailures,...persisted].filter(e=>Date.now()-Date.parse(e.at)[e.id,e])).values()] - .sort((a,b)=>Date.parse(b.at)-Date.parse(a.at)).slice(0,MAX_ERRORS); +export async function recentFailures(bucket) { + const object = await bucket.get(ERROR_KEY); + const persisted = object ? await object.json() : []; + return [ + ...new Map( + [...localFailures, ...persisted] + .filter((e) => Date.now() - Date.parse(e.at) < RETENTION_MS) + .map((e) => [e.id, e]), + ).values(), + ] + .sort((a, b) => Date.parse(b.at) - Date.parse(a.at)) + .slice(0, MAX_ERRORS); } -export const embeddingObservation=()=>lastEmbedding; -export async function check(operation,timeoutMs=3000){ - let timer;const start=performance.now(); - try{const value=await Promise.race([operation(),new Promise((_,reject)=>{timer=setTimeout(()=>reject(Object.assign(new Error('Probe timeout'),{code:'probe_timeout'})),timeoutMs);})]);return {status:'ok',ms:performance.now()-start,value};} - catch(error){return {status:'failed',ms:performance.now()-start,code:error.code==='probe_timeout'?'probe_timeout':classifyError(error).code};} - finally{clearTimeout(timer);} +export const embeddingObservation = () => lastEmbedding; +export async function check(operation, timeoutMs = 3000) { + let timer; + const start = performance.now(); + try { + const value = await Promise.race([ + operation(), + new Promise((_, reject) => { + timer = setTimeout( + () => reject(Object.assign(new Error("Probe timeout"), { code: "probe_timeout" })), + timeoutMs, + ); + }), + ]); + return { status: "ok", ms: performance.now() - start, value }; + } catch (error) { + return { + status: "failed", + ms: performance.now() - start, + code: error.code === "probe_timeout" ? "probe_timeout" : classifyError(error).code, + }; + } finally { + clearTimeout(timer); + } } diff --git a/onlion/backend/worker.mjs b/onlion/backend/worker.mjs index 964d720d..96bf3b56 100644 --- a/onlion/backend/worker.mjs +++ b/onlion/backend/worker.mjs @@ -61,7 +61,6 @@ import { } from "./storage.mjs"; import { deltaBytes, USER_DELTA_LIMIT } from "./page-delta.mjs"; import { deferredDocuments } from "./embedding-session.mjs"; -import { queryVector, rememberQueryVector } from "./query-cache.mjs"; import { LOG_ENTRY_VERSION, MAX_PENDING_READ, @@ -342,6 +341,19 @@ async function mutateUser(env, userId, operation) { * timeouts of the 2026-10-29 incident look like. Ten seconds is the provider client's * own default and leaves room for a round trip to be slow ten times over. */ const EMBED_TIMEOUT_MS = 10_000; +/** Per-attempt timeout for documents. + * + * A query is small and the provider is remote from the function, so ten seconds is already + * generous for it. A document batch is neither: it carries a whole message, and the calls + * cross a region boundary, where a healthy round trip and a stalled one differ by an order of + * magnitude. Cutting a document at ten seconds turned "the provider was slow" into a failed + * Add, which is the one failure the write path cannot absorb. */ +const EMBED_DOCUMENT_TIMEOUT_MS = 25_000; +/** Total embedding work one write may spend. The invocation is killed at 120 s, and the write + * still needs room for its partition round trips after the vectors land, so the budget is + * well inside it: past this point the Add fails with a clear code instead of running out the + * clock with no answer. */ +const EMBED_WRITE_BUDGET_MS = 45_000; /** Total embedding work one search may spend, across every call site. * * A single search embeds its clauses, their windows, every pending record and each @@ -355,13 +367,26 @@ function embeddingClient(env) { NMG_EMBED_TIMEOUT_MS: env.NMG_EMBED_TIMEOUT_MS ?? String(EMBED_TIMEOUT_MS), }); if (!client) return undefined; - const queries = new Map(); - const provider = hash(JSON.stringify([client.indexId, client.baseUrl, client.apiKey])); + /** text -> vector, for every feature this request has needed so far. */ + const vectorsByText = new Map(); const budgetMs = Number(env.NMG_EMBED_QUERY_BUDGET_MS) > 0 ? Number(env.NMG_EMBED_QUERY_BUDGET_MS) : EMBED_QUERY_BUDGET_MS; const deadline = Date.now() + budgetMs; + const writeBudgetMs = + Number(env.NMG_EMBED_WRITE_BUDGET_MS) > 0 + ? Number(env.NMG_EMBED_WRITE_BUDGET_MS) + : EMBED_WRITE_BUDGET_MS; + const writeDeadline = Date.now() + writeBudgetMs; + // A second client, not a second knob on the first: the provider client keeps one timeout, + // and documents need a longer one than queries. + let documentClient; + const documents = () => + (documentClient ??= createEmbeddingClientFromEnv({ + ...env, + NMG_EMBED_TIMEOUT_MS: env.NMG_EMBED_TIMEOUT_MS ?? String(EMBED_DOCUMENT_TIMEOUT_MS), + })); // Thrown from inside `measure`'s async callback this becomes a rejection, which is what // every call site is written against (`catch(() => [])`). A synchronous throw would // escape that and fail the request instead of degrading it. @@ -380,31 +405,41 @@ function embeddingClient(env) { spend(); return embeddingOperation(() => client.embed(inputs)); }), - embedDocuments: deferredDocuments((inputs) => - // Documents are the write path: they carry the memory itself, so they are not - // budgeted. The per-attempt timeout and the batch size bound them instead. - embeddingOperation(() => client.embedDocuments(inputs)), - ), + embedDocuments: deferredDocuments((inputs) => { + // Bounded by a total, not by an attempt: without it one slow provider turned a single + // Add into a 55-120 s request that the platform killed with no answer. + if (Date.now() >= writeDeadline) + throw Object.assign(new Error("embedding budget exhausted for this request"), { + code: "embedding_budget_exhausted", + }); + return embeddingOperation(() => documents().embedDocuments(inputs)); + }), + // Cached per TEXT, not per input array. The retrieval engine embeds one feature at a + // time (`embedQueries([text])`), so a cache keyed by the array only ever hits when the + // caller happens to ask for the same list: pre-warming a batch would populate nothing the + // engine could use, and it would embed every feature separately instead. Against a + // provider across a region boundary that is up to twenty-four round trips per search + // instead of one. embedQueries(inputs) { - const key = hash(JSON.stringify([provider, inputs])); - if (!queries.has(key)) - queries.set( - key, - measure("embedding", async () => { - const cached = queryVector(key); - if (cached) return cached; + const unique = [...new Set(inputs)]; + const missing = unique.filter((text) => !vectorsByText.has(text)); + const fetched = missing.length + ? measure("embedding", async () => { spend(); // Embedded exactly as given. Long-query handling lives in // `queryFeatures`, which searches the whole clause and its windows as // separate features: pooling them here would average the query's // aspects into one compromise vector, which measured worse than either // the whole text or the per-feature maximum. - const vectors = await embeddingOperation(() => client.embedQueries(inputs)); - rememberQueryVector(key, vectors); + const vectors = await embeddingOperation(() => client.embedQueries(missing)); + if (vectors.length !== missing.length) + throw new Error("embedding provider returned an invalid query batch"); return vectors; - }), - ); - return queries.get(key); + }).then((vectors) => { + missing.forEach((text, index) => vectorsByText.set(text, vectors[index])); + }) + : undefined; + return Promise.resolve(fetched).then(() => inputs.map((text) => vectorsByText.get(text))); }, }; } @@ -1097,24 +1132,14 @@ function addAppended(appended, memories) { * clauses balanced instead of finishing one clause's windows before starting the * next. Best-effort: a failed embedding is simply not cached, and the feature falls * back to whatever the retrieval layer manages, so the request still answers. */ -const WARM_CONCURRENCY = 4; async function warmClauseVectors(client, features, parts) { if (!client || parts.length === 0) return; - const depth = Math.max(0, ...features.map((list) => list.length)); - for (let round = 0; round < depth; round += 1) { - const batch = []; - features.forEach((list) => { - if (round < list.length) batch.push(list[round]); - }); - let next = 0; - const warm = async () => { - while (next < batch.length) { - const text = batch[next++]; - await client.embedQueries([text]).catch(() => undefined); - } - }; - await Promise.all(Array.from({ length: Math.min(WARM_CONCURRENCY, batch.length) }, warm)); - } + // One call for every distinct feature text. The engine then looks each one up by text, so + // this pre-warms exactly the entries it will ask for; doing it per text meant up to + // twenty-four cross-region round trips before the first partition was even opened. + const texts = [...new Set(features.flat())]; + if (texts.length === 0) return; + await client.embedQueries(texts).catch(() => undefined); } /** Reports a request that is still running after `NMG_SLOW_REPORT_MS`. The platform * kills an invocation at 120 s and a killed invocation leaves nothing behind, so an @@ -1137,6 +1162,7 @@ function watchSlowRequest(env, context, started) { // the request is still alive: memory, how many requests this instance has served, // and how many SQLite images it has held at once. ...memFields({ sqlite: sqliteStats() }), + last_embedding: embeddingObservation(), }), ); }, afterMs); @@ -1474,6 +1500,14 @@ async function healthResponse(env, path) { Number(env.NMG_EMBED_QUERY_BUDGET_MS) > 0 ? Number(env.NMG_EMBED_QUERY_BUDGET_MS) : EMBED_QUERY_BUDGET_MS, + embed_document_timeout_ms: + Number(env.NMG_EMBED_TIMEOUT_MS) > 0 + ? Number(env.NMG_EMBED_TIMEOUT_MS) + : EMBED_DOCUMENT_TIMEOUT_MS, + embed_write_budget_ms: + Number(env.NMG_EMBED_WRITE_BUDGET_MS) > 0 + ? Number(env.NMG_EMBED_WRITE_BUDGET_MS) + : EMBED_WRITE_BUDGET_MS, slow_report_ms: Number(env.NMG_SLOW_REPORT_MS) > 0 ? Number(env.NMG_SLOW_REPORT_MS) : 60_000, }, upstream_commit: SOURCE_REVISION.split("+")[0], diff --git a/onlion/scripts/test-add-log-mode.mjs b/onlion/scripts/test-add-log-mode.mjs index 13d2de46..a5feb548 100644 --- a/onlion/scripts/test-add-log-mode.mjs +++ b/onlion/scripts/test-add-log-mode.mjs @@ -311,6 +311,36 @@ try { ); } + // A search must embed its features in one call, not one call per feature. The engine asks + // for a single feature at a time, so the batch only helps if the cache is keyed by text - + // and against a provider across a region boundary, twenty-four round trips per search is + // the difference between a search that finishes and one that hits the invocation ceiling. + { + const featureUser = "features-" + process.pid; + for (let index = 0; index < 3; index += 1) { + const added = await call( + "/add", + addBody(featureUser, `feat-${index}`, `feature marker ${index} about harbour tide ${index}`), + ); + assert.equal(added.status, 200, JSON.stringify(added.body)); + } + embedCalls.length = 0; + const found = await call("/search", { + user_id: featureUser, + query: "harbour tide levels and the engineering notes from the second session", + top_k: 20, + }); + assert.equal(found.status, 200, JSON.stringify(found.body)); + // Two calls at most: the feature batch and the pending-record batch. Before the cache was + // keyed by text this was one call per feature plus one per pending record. + assert.ok( + embedCalls.length <= 2, + `one batch per purpose; saw ${embedCalls.length} calls with ${embedCalls + .map((inputs) => inputs.length) + .join("/")} inputs`, + ); + } + // Reset drains the log as well as the partitions. const reset = await call("/reset", { user_id: "alice", diff --git a/onlion/scripts/test-backend.mjs b/onlion/scripts/test-backend.mjs index 9095120f..f665e517 100644 --- a/onlion/scripts/test-backend.mjs +++ b/onlion/scripts/test-backend.mjs @@ -901,6 +901,8 @@ print(json.dumps(r))`, // or above the 120 s platform ceiling turns a slow provider into a dead request. embed_timeout_ms: 10_000, embed_query_budget_ms: 25_000, + embed_document_timeout_ms: 25_000, + embed_write_budget_ms: 45_000, slow_report_ms: 60_000, }); const longQuery = await call("a", "/search", { diff --git a/src/core/openai-embedding.ts b/src/core/openai-embedding.ts index 4dadb7c3..bcfc83e9 100644 --- a/src/core/openai-embedding.ts +++ b/src/core/openai-embedding.ts @@ -91,6 +91,12 @@ export class OpenAIEmbeddingClient { this.documentBody = options.documentBody ?? {}; this.indexId = indexIdentity({ model: this.model, + // The endpoint takes part in the identity. Two deployments of the same model name can + // serve different weights, and a vector from one is not comparable with a query + // embedded by the other - but nothing else in the identity would notice, so switching + // endpoints used to leave the stored vectors silently in place and mix two spaces. + // With it here, a switch makes the index stale and the existing backfill re-embeds. + providerEndpoint: endpointIdentity(options.baseUrl), ...(options.indexNamespace ? { indexNamespace: options.indexNamespace } : {}), dimensions: this.dimensions, documentTemplate: this.documentTemplate, @@ -232,8 +238,22 @@ export class OpenAIEmbeddingClient { * per-call bodies means existing embeddings cannot be reused. Serialisation order * is part of the contract, so keep the spread order identical when editing. */ +/** The part of the endpoint that identifies the deployment: scheme, host and path, without + * a trailing slash or any query, so an equivalent spelling of the same endpoint keeps the + * same identity and a genuinely different one does not. */ +function endpointIdentity(baseUrl: string | undefined): string | null { + if (!baseUrl) return null; + try { + const url = new URL(baseUrl); + return `${url.protocol}//${url.host}${url.pathname.replace(/\/+$/, "")}`; + } catch { + return baseUrl.replace(/\/+$/, ""); + } +} + function indexIdentity(input: { model: string; + providerEndpoint?: string | null; indexNamespace?: string; dimensions?: number; documentTemplate: string; @@ -246,6 +266,7 @@ function indexIdentity(input: { const fingerprint = createHash("sha256") .update( JSON.stringify({ + ...(input.providerEndpoint ? { providerEndpoint: input.providerEndpoint } : {}), ...(input.indexNamespace ? { indexNamespace: input.indexNamespace } : {}), dimensions: input.dimensions ?? null, documentTemplate: input.documentTemplate, diff --git a/tests/core/openai-embedding.test.ts b/tests/core/openai-embedding.test.ts index 3bbe46a4..234bb49c 100644 --- a/tests/core/openai-embedding.test.ts +++ b/tests/core/openai-embedding.test.ts @@ -286,3 +286,28 @@ test("a dropped connection is still retried", async () => { ); } }); + +test("the embedding index identity separates deployments of the same model", () => { + // Two regions of the same provider serve the same model name. A vector from one is not + // comparable with a query embedded by the other, and nothing else in the identity - model, + // dimensions, templates - differs, so the endpoint has to be part of it. Without this, a + // switch left every stored vector in place and silently mixed the two spaces. + const mainland = new OpenAIEmbeddingClient({ + baseUrl: "https://embed.example.cn/compatible-mode/v1", + model: "text-embedding-v4", + dimensions: 1024, + }); + const singapore = new OpenAIEmbeddingClient({ + baseUrl: "https://embed.example.sg/compatible-mode/v1", + model: "text-embedding-v4", + dimensions: 1024, + }); + const spelling = new OpenAIEmbeddingClient({ + baseUrl: "https://embed.example.cn/compatible-mode/v1/", + model: "text-embedding-v4", + dimensions: 1024, + }); + assert.notEqual(mainland.indexId, singapore.indexId, "a different endpoint is a different index"); + assert.equal(spelling.indexId, mainland.indexId, "a trailing slash is the same endpoint"); + assert.match(mainland.indexId, /^text-embedding-v4@[0-9a-f]{12}$/); +});