kbdb(embed): add batch backfill endpoint for pre-Vectorize entries (issue #7 / T2.4 缺口)

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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BDtnGPJpAzp8UqHfAo8o1s
2026-07-05 04:37:04 +00:00
parent 1d19d46161
commit 55a47d7c18
5 changed files with 304 additions and 0 deletions
@@ -79,6 +79,107 @@ function parseMeta(json: string | null): Record<string, unknown> | 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<BackfillResult> {
+  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<Entry>();
+  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;
@@ -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;
@@ -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;
@@ -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<T>() {
+        // 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<T>() {
+        // 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');
+  });
+});
@@ -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',
+  },
+});