From 55a47d7c18b79e3e52d9f7496318f5f363c7f8a5 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 04:37:04 +0000 Subject: [PATCH] =?UTF-8?q?kbdb(embed):=20add=20batch=20backfill=20endpoin?= =?UTF-8?q?t=20for=20pre-Vectorize=20entries=20(issue=20#7=20/=20T2.4=20?= =?UTF-8?q?=E7=BC=BA=E5=8F=A3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit embed 模組原本只有 embedOnWrite（寫入即嵌），對「開 Vectorize binding 之前就寫入」或 embed-on-write 當時漏掉的既有 entry 沒有回填路徑 → is_embedded=0 永遠補不回，語義查詢回 0 筆。 新增（base，對 entries 做；embedding 是 base 唯一職責，非 graph 插件）： - embed.ts: backfillEmbeddings()——找 is_embedded=0 且 isEmbeddable(metadata.embed===true) 的 entry， 批次補嵌（單次 AI.run 陣列 + 單次 VECTORIZE.upsert 陣列 + 單次 UPDATE IN，一批≈3 subrequest）、 設 is_embedded=1，冪等、分批（limit 1-100，回傳 processed/remaining，可重複呼叫直到清零）。 模組未開誠實回 enabled:false（不假綠）。backfillStatus() 回 pending/embedded 計數。 - routes/embed.ts: POST /embed/backfill、GET /embed/backfill/status；模組未開回 409 + capability_hint。 - index.ts: mount /embed。 - tests/embed-backfill.test.ts: 5 vitest（off no-op / 補嵌+標記 / 冪等 / 分批 remaining / status）。 base 維持對內容語意無知（只認通用 embed 旗標，不知 triplet/wiki）。tsc exit 0、vitest 5/5。 端到端（leo21c，wrangler 直推、非 acr update）：pending 5→processed 5→remaining 0， Vectorize vectorCount 0→5，/entries/search?mode=semantic 由 0 筆→5 筆（語義排序命中）。 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BDtnGPJpAzp8UqHfAo8o1s --- kbdb/src/embed.ts | 101 ++++++++++++++++++++++ kbdb/src/index.ts | 4 + kbdb/src/routes/embed.ts | 52 ++++++++++++ kbdb/tests/embed-backfill.test.ts | 137 ++++++++++++++++++++++++++++++ kbdb/vitest.config.ts | 10 +++ 5 files changed, 304 insertions(+) create mode 100644 kbdb/src/routes/embed.ts create mode 100644 kbdb/tests/embed-backfill.test.ts create mode 100644 kbdb/vitest.config.ts diff --git a/kbdb/src/embed.ts b/kbdb/src/embed.ts index 3276182..c86f7dd 100644 --- a/kbdb/src/embed.ts +++ b/kbdb/src/embed.ts @@ -79,6 +79,107 @@ function parseMeta(json: string | null): Record | null { } } +// SQL predicate for "an entry that SHOULD be embedded but isn't yet". +// - isEmbeddable 契約 = metadata_json.embed === true（base 通用旗標，對內容語意無知，不寫死 entry_type）。 +// SQLite json_extract 對 JSON boolean true 回整數 1 → `= 1` 精確對齊 TS 的 `=== true`。 +// - is_embedded = 0：尚未（對「當前」index）補嵌的 bookkeeping。 +// - content 非空：空字串 embedText 會回 null，排除以免變成永遠清不掉的殘留候選。 +const BACKFILL_PREDICATE = + "is_embedded = 0 AND content IS NOT NULL AND content <> '' AND json_extract(metadata_json, '$.embed') = 1"; + +export interface BackfillResult { + enabled: boolean; // 模組是否開（false → 什麼都沒做，caller 該誠實回錯，不假裝）。 + processed: number; // 本次真的嵌進 Vectorize 並標 is_embedded=1 的筆數。 + skipped: number; // 掃到但沒嵌（例如 embedText 回 null）的筆數。 + remaining: number; // 本次之後仍待補嵌的筆數（可重複呼叫直到 0）。 + scanned: number; // 本批掃出的候選筆數（受 limit 限制）。 +} + +/** + * Backfill（回填）：對「開 Vectorize 之前就寫入、或 embed-on-write 當時漏掉」的既有 entry 批次補嵌。 + * 冪等（重跑已補嵌的不會重複算，upsert 同 id 冪等）、分批（單次 limit 上限，避開 subrequest/CPU/timeout）、 + * 回傳處理筆數 + 剩餘筆數（caller 重複呼叫直到 remaining=0）。 + * - 模組未開（無 VECTORIZE+AI）→ 誠實回 { enabled:false }，不假裝成功（mindset §7 禁假綠）。 + * - 只補「isEmbeddable（metadata.embed===true）且 is_embedded=0」的 entry——與 embedOnWrite 同一契約， + * base 維持對內容語意無知（不知 triplet/wiki，只認通用 embed 旗標）。 + * - 效率：整批用「單次 AI.run（陣列輸入）+ 單次 VECTORIZE.upsert（陣列）+ 單次 UPDATE ... IN(...)」， + * 一批 ≈ 3 個 subrequest，不隨 limit 線性增長 → free/paid tier 都安全。 + */ +export async function backfillEmbeddings( + env: Bindings, + opts: { limit?: number; owner_id?: string; source?: string } = {}, +): Promise { + if (!embedEnabled(env)) return { enabled: false, processed: 0, skipped: 0, remaining: 0, scanned: 0 }; + const limit = Math.min(Math.max(opts.limit ?? 25, 1), 100); + + const conds = [BACKFILL_PREDICATE]; + const params: unknown[] = []; + if (opts.owner_id) { conds.push('owner_id = ?'); params.push(opts.owner_id); } + if (opts.source) { conds.push("json_extract(metadata_json, '$.source') = ?"); params.push(opts.source); } + const where = conds.join(' AND '); + + const res = await env.DB + .prepare(`SELECT * FROM entries WHERE ${where} ORDER BY created_at ASC LIMIT ?`) + .bind(...params, limit) + .all(); + const rows = res.results ?? []; + const scanned = rows.length; + + let processed = 0; + const embeddable = rows.filter((e) => (e.content ?? '').trim().length > 0); + if (embeddable.length > 0 && env.AI && env.VECTORIZE) { + const texts = embeddable.map((e) => (e.content ?? '').trim()); + const out = (await env.AI.run(EMBED_MODEL, { text: texts })) as { data: number[][] }; + const data = out?.data ?? []; + const vectors = embeddable + .map((e, i) => ({ e, vec: data[i] })) + .filter((x): x is { e: Entry; vec: number[] } => Array.isArray(x.vec) && x.vec.length > 0) + .map((x) => ({ + id: x.e.id, + values: x.vec, + metadata: { + owner_id: x.e.owner_id ?? '', + entry_type: x.e.entry_type, + source: readSource(x.e) ?? '', + }, + })); + if (vectors.length > 0) { + await env.VECTORIZE.upsert(vectors); + const ids = vectors.map((v) => v.id); + const placeholders = ids.map(() => '?').join(','); + await env.DB.prepare(`UPDATE entries SET is_embedded = 1 WHERE id IN (${placeholders})`).bind(...ids).run(); + processed = vectors.length; + } + } + + const remRow = await env.DB + .prepare(`SELECT COUNT(*) as c FROM entries WHERE ${where}`) + .bind(...params) + .first<{ c: number }>(); + return { enabled: true, processed, skipped: scanned - processed, remaining: remRow?.c ?? 0, scanned }; +} + +/** 補嵌進度統計（回報用；模組未開仍可查 pending 數，誠實標 enabled:false）。 */ +export async function backfillStatus( + env: Bindings, + opts: { owner_id?: string; source?: string } = {}, +): Promise<{ enabled: boolean; pending: number; embedded: number }> { + const conds: string[] = []; + const params: unknown[] = []; + if (opts.owner_id) { conds.push('owner_id = ?'); params.push(opts.owner_id); } + if (opts.source) { conds.push("json_extract(metadata_json, '$.source') = ?"); params.push(opts.source); } + const extra = conds.length ? ` AND ${conds.join(' AND ')}` : ''; + const pendingRow = await env.DB + .prepare(`SELECT COUNT(*) as c FROM entries WHERE ${BACKFILL_PREDICATE}${extra}`) + .bind(...params) + .first<{ c: number }>(); + const embeddedRow = await env.DB + .prepare(`SELECT COUNT(*) as c FROM entries WHERE is_embedded = 1 AND json_extract(metadata_json, '$.embed') = 1${extra}`) + .bind(...params) + .first<{ c: number }>(); + return { enabled: embedEnabled(env), pending: pendingRow?.c ?? 0, embedded: embeddedRow?.c ?? 0 }; +} + export interface SemanticHit { id: string; score: number; diff --git a/kbdb/src/index.ts b/kbdb/src/index.ts index f094de7..f2095ed 100644 --- a/kbdb/src/index.ts +++ b/kbdb/src/index.ts @@ -9,6 +9,7 @@ import { entryRoutes } from './routes/entries'; import { templateRoutes } from './routes/templates'; import { recordRoutes } from './routes/records'; import { recipeStatRoutes } from './routes/recipe-stats'; +import { embedRoutes } from './routes/embed'; const app = new Hono<{ Bindings: Bindings }>(); @@ -19,5 +20,8 @@ app.route('/entries', entryRoutes); app.route('/templates', templateRoutes); app.route('/records', recordRoutes); app.route('/recipe-stats', recipeStatRoutes); +// Optional embed module admin (backfill). Route mounts unconditionally; the handler +// honestly 409s when the embed binding is off (base 對內容語意無知，只認通用 embed 旗標)。 +app.route('/embed', embedRoutes); export default app; diff --git a/kbdb/src/routes/embed.ts b/kbdb/src/routes/embed.ts new file mode 100644 index 0000000..1792221 --- /dev/null +++ b/kbdb/src/routes/embed.ts @@ -0,0 +1,52 @@ +// Embed module admin route — backfill existing entries (issue #7 / mira-dissolve T2.4 缺口). +// +// 背景：embed 原本只有「寫入即嵌」（embedOnWrite），對「開 Vectorize binding 之前就寫入」或 +// 「embed-on-write 當時漏掉」的既有 entry 沒有回填路徑 → is_embedded=0 且永遠補不回。 +// 本 route = 把 embed.ts 既有 embedText/VECTORIZE.upsert 邏輯包成可重複呼叫的批次補嵌端點。 +// +// 鐵律對齊：embedding 屬 base optional 模組；模組未開（無 VECTORIZE+AI）→ 誠實回 409，不假裝（mindset §7）。 +// base 對內容語意無知：只認通用 metadata.embed===true 旗標，不知 triplet/wiki（解耦）。 +import { Hono } from 'hono'; +import type { Bindings } from '../types'; +import { embedEnabled, backfillEmbeddings, backfillStatus } from '../embed'; + +export const embedRoutes = new Hono<{ Bindings: Bindings }>(); + +const OFF_HINT = + '語義補嵌需先開 embed 模組（Vectorize+AI binding）。叫 CC「幫我開語義查詢」（設 kbdb_embed:true + redeploy 注入 binding）後再呼叫本端點。'; + +// POST /embed/backfill — batch-embed existing embeddable entries with is_embedded=0. +// body（皆選填）：{ limit?:1-100（預設25）, owner_id?, source? }。 +// 冪等：重跑不會重複嵌（已 is_embedded=1 的不再入選；upsert 同 id 冪等）。 +// 分批：單次最多 limit 筆；回傳 remaining>0 表示還有 → 重複呼叫直到 remaining=0。 +// 模組未開 → 409 + capability_hint（不假綠）。 +embedRoutes.post('/backfill', async (c) => { + if (!embedEnabled(c.env)) { + return c.json( + { success: false, error: 'embed module not enabled (need VECTORIZE + AI bindings)', capability_hint: OFF_HINT }, + 409, + ); + } + const body = (await c.req.json().catch(() => ({}))) as { + limit?: number | string; + owner_id?: string; + source?: string; + }; + const result = await backfillEmbeddings(c.env, { + limit: body.limit !== undefined ? Number(body.limit) : undefined, + owner_id: body.owner_id || undefined, + source: body.source || undefined, + }); + return c.json({ success: true, ...result }); +}); + +// GET /embed/backfill/status?owner_id=&source= — 待補嵌 / 已補嵌計數（回報 + 判斷是否清零用）。 +embedRoutes.get('/backfill/status', async (c) => { + const status = await backfillStatus(c.env, { + owner_id: c.req.query('owner_id') || undefined, + source: c.req.query('source') || undefined, + }); + return c.json({ success: true, ...status }); +}); + +export default embedRoutes; diff --git a/kbdb/tests/embed-backfill.test.ts b/kbdb/tests/embed-backfill.test.ts new file mode 100644 index 0000000..13fbc32 --- /dev/null +++ b/kbdb/tests/embed-backfill.test.ts @@ -0,0 +1,137 @@ +import { describe, it, expect } from 'vitest'; +import { backfillEmbeddings, backfillStatus, embedEnabled } from '../src/embed'; +import type { Bindings, Entry } from '../src/types'; + +// ── Minimal in-memory fakes (no Workers runtime) ───────────────────────────── +// The fake DB interprets only the 3 statement shapes backfill issues, by keyword: +// SELECT * ... LIMIT → candidate rows (embeddable & is_embedded=0 & non-empty content) +// UPDATE ... IN (...) → flip is_embedded=1 for the bound ids +// SELECT COUNT(*) → count of remaining candidates +function isCandidate(e: Entry): boolean { + if (e.is_embedded !== 0) return false; + if (!e.content || e.content.trim() === '') return false; + try { + const m = JSON.parse(e.metadata_json ?? 'null'); + return m?.embed === true; + } catch { + return false; + } +} + +function makeFakeDB(store: Entry[]) { + const prepare = (sql: string) => { + let bound: unknown[] = []; + const stmt = { + bind(...args: unknown[]) { bound = args; return stmt; }, + async all() { + // SELECT * ... LIMIT ? (limit is the last bound param) + const limit = Number(bound[bound.length - 1]); + const results = store.filter(isCandidate).slice(0, limit) as unknown as T[]; + return { results }; + }, + async first() { + // SELECT COUNT(*) as c ... + const c = store.filter(isCandidate).length; + return { c } as unknown as T; + }, + async run() { + // UPDATE entries SET is_embedded = 1 WHERE id IN (...) → bound = ids + const ids = new Set(bound.map(String)); + for (const e of store) if (ids.has(e.id)) e.is_embedded = 1; + return { success: true }; + }, + }; + return stmt; + }; + return { prepare } as unknown as D1Database; +} + +function mkEntry(id: string, content: string | null, embed: boolean, is_embedded = 0): Entry { + return { + id, content, entry_type: 'workflow', owner_id: 'leo', parent_id: null, page_name: null, + refs_json: '[]', tags_json: '[]', task_status: null, content_hash: null, is_embedded, + confidence: null, metadata_json: JSON.stringify({ embed }), created_at: 1, updated_at: 1, + }; +} + +function makeEnv(store: Entry[], withBindings: boolean): Bindings { + const upserts: { id: string }[] = []; + const aiCalls: string[][] = []; + const env = { + DB: makeFakeDB(store), + ENVIRONMENT: 'test', + ...(withBindings + ? { + AI: { async run(_m: string, i: { text: string[] }) { aiCalls.push(i.text); return { data: i.text.map(() => [0.1, 0.2, 0.3]) }; } }, + VECTORIZE: { async upsert(v: { id: string }[]) { upserts.push(...v); return { count: v.length }; } }, + } + : {}), + } as unknown as Bindings; + (env as unknown as { __upserts: unknown[]; __ai: unknown[] }).__upserts = upserts; + (env as unknown as { __upserts: unknown[]; __ai: unknown[] }).__ai = aiCalls; + return env; +} + +describe('backfillEmbeddings', () => { + it('module off → enabled:false, no-op (誠實不假綠)', async () => { + const store = [mkEntry('e1', 'hello', true)]; + const env = makeEnv(store, false); + expect(embedEnabled(env)).toBe(false); + const r = await backfillEmbeddings(env); + expect(r).toEqual({ enabled: false, processed: 0, skipped: 0, remaining: 0, scanned: 0 }); + expect(store[0].is_embedded).toBe(0); // untouched + }); + + it('embeds embeddable+is_embedded=0 entries, marks is_embedded=1, batches AI+upsert', async () => { + const store = [ + mkEntry('e1', 'doorbell workflow', true), + mkEntry('e2', 'notify workflow', true), + mkEntry('e3', 'not tagged', false), // embed:false → not a candidate + mkEntry('e4', 'already done', true, 1), // is_embedded=1 → not a candidate + mkEntry('e5', ' ', true), // empty content → not embeddable + ]; + const env = makeEnv(store, true); + const r = await backfillEmbeddings(env, { limit: 100 }); + expect(r.enabled).toBe(true); + expect(r.processed).toBe(2); // only e1,e2 + expect(r.remaining).toBe(0); // nothing left embeddable + expect(store.find((e) => e.id === 'e1')!.is_embedded).toBe(1); + expect(store.find((e) => e.id === 'e2')!.is_embedded).toBe(1); + expect(store.find((e) => e.id === 'e3')!.is_embedded).toBe(0); + const upserts = (env as unknown as { __upserts: { id: string }[] }).__upserts; + expect(upserts.map((u) => u.id).sort()).toEqual(['e1', 'e2']); + const ai = (env as unknown as { __ai: string[][] }).__ai; + expect(ai.length).toBe(1); // single batched AI.run for the whole batch + expect(ai[0].length).toBe(2); + }); + + it('idempotent: re-run after all embedded processes nothing', async () => { + const store = [mkEntry('e1', 'x', true)]; + const env = makeEnv(store, true); + await backfillEmbeddings(env); + const r2 = await backfillEmbeddings(env); + expect(r2.processed).toBe(0); + expect(r2.remaining).toBe(0); + }); + + it('batches via limit → remaining reported so caller can loop to zero', async () => { + const store = [mkEntry('a', 'x', true), mkEntry('b', 'y', true), mkEntry('c', 'z', true)]; + const env = makeEnv(store, true); + const r1 = await backfillEmbeddings(env, { limit: 2 }); + expect(r1.processed).toBe(2); + expect(r1.remaining).toBe(1); + const r2 = await backfillEmbeddings(env, { limit: 2 }); + expect(r2.processed).toBe(1); + expect(r2.remaining).toBe(0); + }); + + it('status reports pending/embedded counts', async () => { + const store = [mkEntry('e1', 'x', true), mkEntry('e2', 'y', true, 1)]; + const env = makeEnv(store, true); + const s = await backfillStatus(env); + // fake first() returns candidate count for pending; embedded query also runs through + // the same COUNT fake, so this asserts the call path works (enabled:true). + expect(s.enabled).toBe(true); + expect(typeof s.pending).toBe('number'); + }); +}); diff --git a/kbdb/vitest.config.ts b/kbdb/vitest.config.ts new file mode 100644 index 0000000..e3a4604 --- /dev/null +++ b/kbdb/vitest.config.ts @@ -0,0 +1,10 @@ +import { defineConfig } from 'vitest/config'; + +// Plain node vitest (no Workers runtime): embed backfill is tested against a fake env +// (mock DB/AI/VECTORIZE) so logic is verified without Cloudflare bindings. +export default defineConfig({ + test: { + include: ['tests/**/*.test.ts'], + environment: 'node', + }, +});