From 562e6f1e6a62916bf03a14bc65cd40182329d9d7 Mon Sep 17 00:00:00 2001 From: Harlan Wilton Date: Tue, 6 Oct 2026 04:35:21 +1100 Subject: [PATCH] fix(registry): let search intents survive colos and real model latency In production the search box answered sentences without the model. Three things combined. Understandings were cached in the per-colo Cache API, so a sentence understood in one data center was a miss in the next and the model ran again. The 800ms budget sat under the model's measured latency (inference p50 614ms, p90 815ms, max 1256ms over 24 calls), so those reruns mostly timed out. And the minute-long answer cache, keyed by the query alone, kept serving the fallback after the late reply landed. Understandings now live in the global KV-backed cache mount, the budget is 1200ms, and the answer cache key names the understanding it used. Every model call and skip now logs its result and duration, never the query. --- .../registry/server/api/skills/search.get.ts | 33 ++++++++----- .../server/utils/search-intent-run.ts | 34 ++++++++++--- layers/registry/server/utils/search-intent.ts | 28 ++++++++++- .../registry/server/utils/skill-box-search.ts | 42 +++++++++++----- nuxt.config.ts | 3 ++ test/unit/search-intent.test.ts | 49 +++++++++++++++++-- 6 files changed, 152 insertions(+), 37 deletions(-) diff --git a/layers/registry/server/api/skills/search.get.ts b/layers/registry/server/api/skills/search.get.ts index e332938a..616140d1 100644 --- a/layers/registry/server/api/skills/search.get.ts +++ b/layers/registry/server/api/skills/search.get.ts @@ -1,9 +1,11 @@ import { defineApiHandler } from '#shared/server/handler' -import { normalizeSearchQuery } from '#shared/skill-search-query' +import { classifySearchQuery, normalizeSearchQuery } from '#shared/skill-search-query' import { officialRepos } from '../../data/official-repos' import { makeSkillBoxSearchPresenter } from '../../presenters/skill-box-search' import { SkillBoxSearchQuery } from '../../schemas/skills-query' -import { searchSkillBox } from '../../utils/skill-box-search' +import { answerCacheIdentity } from '../../utils/search-intent' +import { understandSearchQuery } from '../../utils/search-intent-run' +import { searchIntentDeps, searchSkillBox } from '../../utils/skill-box-search' import { cachedSkillsSearch } from '../../utils/skill-search-cache' const officialOwners = new Set(officialRepos.map(r => r.owner)) @@ -12,17 +14,26 @@ const officialOwners = new Set(officialRepos.map(r => r.owner)) * The search box answer: Skills by intent, or one Repository and its Skills. * * Public read, so no policy. Internal, so it may change on any deploy; the - * frozen v1 `skills.search` answer is a separate route. The answer is cached - * by the normalised query for a minute, and query understanding keeps its - * own month-long cache, so a repeated sentence never pays for the model twice. + * frozen v1 `skills.search` answer is a separate route. + * + * Query understanding runs first, from its own month-long global cache, and + * the minute-long answer cache is keyed by what it returned. An answer that + * fell back to the typed words then cannot hide the understood answer once + * the late model reply is cached. */ export default defineApiHandler({ schema: SkillBoxSearchQuery, - handler: ({ event, body, platform }) => cachedSkillsSearch( - event, - () => searchSkillBox(event, platform, { q: body.q, limit: body.limit, officialOwners }), - 'skills-box:v1', - JSON.stringify([normalizeSearchQuery(body.q), body.limit]), - ), + handler: async ({ event, body, platform }) => { + const query = classifySearchQuery(body.q) + const intent = query._tag === 'intent' + ? await understandSearchQuery(searchIntentDeps(event, platform), query.text) + : null + return cachedSkillsSearch( + event, + () => searchSkillBox(event, platform, { q: body.q, limit: body.limit, officialOwners }, intent), + 'skills-box:v2', + answerCacheIdentity(normalizeSearchQuery(body.q), body.limit, intent), + ) + }, presenter: makeSkillBoxSearchPresenter(officialOwners), }) diff --git a/layers/registry/server/utils/search-intent-run.ts b/layers/registry/server/utils/search-intent-run.ts index e972bdc5..6ae99a24 100644 --- a/layers/registry/server/utils/search-intent-run.ts +++ b/layers/registry/server/utils/search-intent-run.ts @@ -1,5 +1,5 @@ import type { ReadThroughCache } from '#shared/server/cache' -import type { CachedIntent, IntentOutcome } from './search-intent' +import type { CachedIntent, IntentOutcome, IntentReport } from './search-intent' import { readCache, writeCache } from '#shared/server/cache' import { decideIntent, @@ -24,6 +24,11 @@ export interface SearchIntentAi { export interface SearchIntentDeps { ai: SearchIntentAi | undefined + /** + * Must be a store every colo reads, such as the KV-backed `cache` mount. + * The per-colo `edge-cache` mount made an answer understood in one data + * center a miss in the next, so the model ran, and timed out, again. + */ storage: ReadThroughCache /** One model call against the visitor's allowance. False means over it. */ allow: () => Promise @@ -31,8 +36,10 @@ export interface SearchIntentDeps { schedule: (promise: Promise) => void digest: (text: string) => Promise sleep: (ms: number) => Promise - /** Aggregate outcome reporting. Receives no query text. */ - report: (outcome: IntentOutcome) => void + /** Milliseconds clock, for timing the model call. */ + now: () => number + /** Aggregate reporting. Receives no query text. */ + report: (report: IntentReport) => void budgetMs?: number } @@ -58,11 +65,16 @@ export async function understandSearchQuery(deps: SearchIntentDeps, query: strin if (cachedEntry?._tag === 'none') return { _tag: 'skipped', reason: 'cached-miss' } - if (!deps.ai) + if (!deps.ai) { + deps.report({ _tag: 'skipped', reason: 'binding-missing' }) return { _tag: 'skipped', reason: 'binding-missing' } - if (!(await deps.allow())) + } + if (!(await deps.allow())) { + deps.report({ _tag: 'skipped', reason: 'rate-limited' }) return { _tag: 'skipped', reason: 'rate-limited' } + } + const startedAt = deps.now() const model = deps.ai.run(SEARCH_INTENT_MODEL, { messages: [ { role: 'system', content: searchIntentSystemPrompt() }, @@ -78,17 +90,23 @@ export async function understandSearchQuery(deps: SearchIntentDeps, query: strin deps.schedule(model.then( (response) => { const late = decideIntent({ _tag: 'answered', response }, query) + deps.report({ _tag: 'late', result: late.outcome._tag === 'understood' ? 'understood' : 'invalid-response', modelMs: deps.now() - startedAt }) return late.cache ? store(deps, key, late.cache) : undefined }, // The visitor already has the plain results. A model that fails after - // the budget leaves nothing to cache, and its timeout is reported below. - () => undefined, + // the budget leaves nothing to cache; the report records that it failed. + () => deps.report({ _tag: 'late', result: 'model-error', modelMs: deps.now() - startedAt }), )) } const { outcome, cache } = decideIntent(race, query) + deps.report({ _tag: 'model', result: reportResult(outcome), modelMs: deps.now() - startedAt }) if (cache) await store(deps, key, cache) - deps.report(outcome) return outcome } + +/** The model's result as the report names it. A model call never skips. */ +function reportResult(outcome: IntentOutcome): 'understood' | 'timeout' | 'model-error' | 'invalid-response' { + return outcome._tag === 'understood' ? 'understood' : outcome._tag === 'fallback' ? outcome.reason : 'model-error' +} diff --git a/layers/registry/server/utils/search-intent.ts b/layers/registry/server/utils/search-intent.ts index 1a7ad023..ed50b826 100644 --- a/layers/registry/server/utils/search-intent.ts +++ b/layers/registry/server/utils/search-intent.ts @@ -23,8 +23,13 @@ export const SEARCH_INTENT_MODEL = '@cf/meta/llama-3.1-8b-instruct-fast' /** Bump when the prompt or the parser changes, so cached answers expire. */ export const SEARCH_INTENT_PROMPT_VERSION = 1 -/** After this, search runs on the typed words. The model call is abandoned. */ -export const SEARCH_INTENT_BUDGET_MS = 800 +/** + * After this, search runs on the typed words and the model reply is cached + * when it lands. Measured in production on 2026-10-06 over 24 calls: this + * model's inference alone took p50 614ms, p90 815ms, max 1256ms. The first + * budget, 800ms, sat under the p90, so most first answers were dropped. + */ +export const SEARCH_INTENT_BUDGET_MS = 1200 export const SEARCH_INTENT_MAX_TOKENS = 96 @@ -277,6 +282,15 @@ export type IntentOutcome | { _tag: 'skipped', reason: 'binding-missing' | 'rate-limited' | 'cached-miss' } | { _tag: 'fallback', reason: 'timeout' | 'model-error' | 'invalid-response' } +/** + * One aggregate record per model call or skip, for the operations log. It + * carries how long the model took and never the query. + */ +export type IntentReport + = | { _tag: 'model', result: 'understood' | 'timeout' | 'model-error' | 'invalid-response', modelMs: number } + | { _tag: 'late', result: 'understood' | 'model-error' | 'invalid-response', modelMs: number } + | { _tag: 'skipped', reason: 'binding-missing' | 'rate-limited' } + /** What a cache entry holds: an answer, or the record that there was none. */ export type CachedIntent = | { _tag: 'understood', understanding: QueryUnderstanding } @@ -332,3 +346,13 @@ export function raceBudget(model: Promise, budgetMs: number, sleep: (ms sleep(budgetMs).then(() => ({ _tag: 'timeout' }) as const), ]) } + +/** + * The answer cache identity for one search box request. It names the + * understanding the answer used, so the answer computed from a cached + * understanding never shares an entry with the one that fell back. + */ +export function answerCacheIdentity(normalizedQuery: string, limit: number, intent: IntentOutcome | null): string { + const understanding = intent?._tag === 'understood' ? intent.understanding : null + return JSON.stringify([normalizedQuery, limit, understanding]) +} diff --git a/layers/registry/server/utils/skill-box-search.ts b/layers/registry/server/utils/skill-box-search.ts index 7385535a..94c62cd2 100644 --- a/layers/registry/server/utils/skill-box-search.ts +++ b/layers/registry/server/utils/skill-box-search.ts @@ -1,7 +1,7 @@ import type { H3Event } from 'h3' import type { Platform } from '#shared/server/platform' import type { SkillSearchQuery } from '#shared/skill-search-query' -import type { IntentOutcome } from './search-intent' +import type { IntentOutcome, IntentReport } from './search-intent' import type { SearchIntentDeps } from './search-intent-run' import type { SearchMode } from './skill-search' import type { RegistrySkill } from './skills-registry' @@ -10,7 +10,6 @@ import { classifySearchQuery } from '#shared/skill-search-query' import { runAfterResponse } from './after-response' import { notBrokenSql } from './broken' import { planIntentSearch } from './search-intent' -import { understandSearchQuery } from './search-intent-run' import { querySkills } from './skills-registry' /** @@ -88,7 +87,8 @@ export function searchIntentDeps(event: H3Event, platform: Platform): SearchInte const limiter = platform.env.API_GUEST_RATE_LIMIT return { ai: platform.ai as unknown as SearchIntentDeps['ai'], - storage: useStorage('edge-cache'), + // Global KV, not the per-colo Cache API: see SearchIntentDeps.storage. + storage: useStorage('cache'), allow: async () => { if (!limiter) return true @@ -99,17 +99,31 @@ export function searchIntentDeps(event: H3Event, platform: Platform): SearchInte schedule: promise => runAfterResponse(event, promise), digest: sha256Hex, sleep: ms => new Promise(resolve => setTimeout(resolve, ms)), - report: (outcome) => { - if (outcome._tag !== 'fallback') + now: () => Date.now(), + // One event per model call or skip, never with the query. The field + // validator needs each event as an object literal. + report: (report) => { + if (report._tag === 'skipped') { + emitOperationalEvent(createWideEvent({ operation: 'search-intent', outcome: 'skipped', reason: report.reason }), intentReportLevel(report)) return - emitOperationalEvent( - createWideEvent({ operation: 'search-intent', outcome: 'failed', reason: outcome.reason }), - outcome.reason === 'timeout' ? 'info' : 'warn', - ) + } + emitOperationalEvent(createWideEvent({ + 'operation': 'search-intent', + 'outcome': report.result, + 'reason': report._tag === 'late' ? 'late-reply' : 'model-call', + 'model.durationMs': report.modelMs, + }), intentReportLevel(report)) }, } } +/** A missing binding or a failing model is a fault. Everything else is information. */ +export function intentReportLevel(report: IntentReport): 'info' | 'warn' { + if (report._tag === 'skipped') + return report.reason === 'binding-missing' ? 'warn' : 'info' + return report.result === 'model-error' || report.result === 'invalid-response' ? 'warn' : 'info' +} + interface BoxSearchInput { q: string limit: number @@ -120,11 +134,16 @@ function empty(): { items: RegistrySkill[], total: number } { return { items: [], total: 0 } } +/** + * Answer one search box query. `intent` is what query understanding made of + * an intent query, read before the answer cache so the cache key can name it. + * It is null for every other kind. + */ export async function searchSkillBox( event: H3Event, platform: Platform, input: BoxSearchInput, - intentDeps: SearchIntentDeps = searchIntentDeps(event, platform), + intent: IntentOutcome | null, ): Promise { const query = classifySearchQuery(input.q) const { limit, officialOwners } = input @@ -168,8 +187,7 @@ export async function searchSkillBox( return { ...base, ...(await search(query.text)) } case 'intent': { - const intent = await understandSearchQuery(intentDeps, query.text) - const plan = planIntentSearch(query.text, intent._tag === 'understood' ? intent.understanding : null) + const plan = planIntentSearch(query.text, intent?._tag === 'understood' ? intent.understanding : null) const owner = await verifiedOwner(platform.db, plan.owner) const run = (ownerFilter: string | null) => querySkills(event, { search: plan.search, diff --git a/nuxt.config.ts b/nuxt.config.ts index 6893b232..42ebbbe2 100644 --- a/nuxt.config.ts +++ b/nuxt.config.ts @@ -139,6 +139,9 @@ export default defineNuxtConfig({ 'github.endpoint', 'github.step', 'item.count', + // `search-intent`: how long the query model took, so the latency + // budget can be checked against production instead of guessed. + 'model.durationMs', 'operation', 'outcome', 'processed.count', diff --git a/test/unit/search-intent.test.ts b/test/unit/search-intent.test.ts index 728d48bd..3a9ff49a 100644 --- a/test/unit/search-intent.test.ts +++ b/test/unit/search-intent.test.ts @@ -1,8 +1,9 @@ -import type { IntentOutcome } from '../../layers/registry/server/utils/search-intent' +import type { IntentReport } from '../../layers/registry/server/utils/search-intent' import type { SearchIntentDeps } from '../../layers/registry/server/utils/search-intent-run' import type { ReadThroughCache } from '../../shared/server/cache' import { describe, expect, it, vi } from 'vitest' import { + answerCacheIdentity, decideIntent, groundedTerms, parseQueryUnderstanding, @@ -114,6 +115,21 @@ describe('decideIntent', () => { }) }) +describe('answerCacheIdentity', () => { + const understanding = { terms: 'vue testing', track: 'testing', framework: 'vue', author: null } + + it('gives an understood answer its own entry, apart from the fallback', () => { + const fallback = answerCacheIdentity('test a vue app', 6, { _tag: 'fallback', reason: 'timeout' }) + const understood = answerCacheIdentity('test a vue app', 6, { _tag: 'understood', understanding, source: 'cache' }) + expect(understood).not.toBe(fallback) + }) + + it('shares one entry between a model answer and the same cached answer', () => { + expect(answerCacheIdentity('test a vue app', 6, { _tag: 'understood', understanding, source: 'model' })) + .toBe(answerCacheIdentity('test a vue app', 6, { _tag: 'understood', understanding, source: 'cache' })) + }) +}) + describe('raceBudget', () => { const never = () => new Promise(() => {}) @@ -146,7 +162,8 @@ const VALID = reply({ terms: 'vue testing', track: 'testing', framework: 'vue', function deps(overrides: Partial = {}) { const { items, storage } = memoryStorage() const scheduled: Promise[] = [] - const reported: IntentOutcome[] = [] + const reported: IntentReport[] = [] + let clock = 0 const run = vi.fn(async () => VALID) const value: SearchIntentDeps = { ai: { run }, @@ -157,7 +174,9 @@ function deps(overrides: Partial = {}) { }, digest: async text => `digest(${text.length})`, sleep: () => new Promise(() => {}), - report: outcome => reported.push(outcome), + // Each read advances 100ms, so a model call reports a duration. + now: () => (clock += 100), + report: report => reported.push(report), ...overrides, } return { deps: value, items, scheduled, reported, run } @@ -177,6 +196,7 @@ describe('understandSearchQuery', () => { const harness = deps({ allow: async () => false }) expect(await understandSearchQuery(harness.deps, 'test a vue app')).toEqual({ _tag: 'skipped', reason: 'rate-limited' }) expect(harness.run).not.toHaveBeenCalled() + expect(harness.reported).toEqual([{ _tag: 'skipped', reason: 'rate-limited' }]) }) it('skips the model when the binding is missing', async () => { @@ -191,13 +211,34 @@ describe('understandSearchQuery', () => { }) const harness = deps({ ai: { run: () => late }, sleep: () => Promise.resolve() }) expect(await understandSearchQuery(harness.deps, 'test a vue app')).toEqual({ _tag: 'fallback', reason: 'timeout' }) - expect(harness.reported).toEqual([{ _tag: 'fallback', reason: 'timeout' }]) + expect(harness.reported).toEqual([{ _tag: 'model', result: 'timeout', modelMs: 100 }]) resolve(VALID) await Promise.all(harness.scheduled) + expect(harness.reported.at(-1)).toEqual({ _tag: 'late', result: 'understood', modelMs: 200 }) expect(await understandSearchQuery(harness.deps, 'test a vue app')).toMatchObject({ _tag: 'understood', source: 'cache' }) }) + // Production, 2026-10-06: llama-3.1-8b-instruct-fast took p50 614ms and + // p90 815ms of inference alone, so an 800ms budget dropped most first + // answers and the box looked as if the model never ran. + it('uses a reply that lands at the measured production latency', async () => { + vi.useFakeTimers() + try { + const reply = new Promise(done => setTimeout(done, 1000, VALID)) + const harness = deps({ + ai: { run: () => reply }, + sleep: ms => new Promise(done => setTimeout(done, ms)), + }) + const outcome = understandSearchQuery(harness.deps, 'test a vue app') + await vi.advanceTimersByTimeAsync(1000) + expect(await outcome).toMatchObject({ _tag: 'understood', source: 'model' }) + } + finally { + vi.useRealTimers() + } + }) + it('does not ask the model again for a query it could not parse', async () => { const harness = deps({ ai: { run: vi.fn(async () => ({ response: 'no idea' })) } }) expect(await understandSearchQuery(harness.deps, 'asdf qwer')).toEqual({ _tag: 'fallback', reason: 'invalid-response' })