diff --git a/cypher-executor/src/index.ts b/cypher-executor/src/index.ts index a61ede1..33ec07e 100644 --- a/cypher-executor/src/index.ts +++ b/cypher-executor/src/index.ts @@ -24,6 +24,8 @@ import { consoleAuthRouter } from './routes/console-auth'; import { consoleDashboardRouter } from './routes/console-dashboard'; import { portalRouter } from './routes/portal'; import { portalDataRouter } from './routes/portal-data'; +import { storageRouter } from './routes/storage'; +import { withDurableStores } from './lib/durable-store'; const app = new Hono<{ Bindings: Bindings }>(); @@ -97,11 +99,19 @@ app.route('/', consoleAuthRouter); // Arcrun#3 發現②：console 專用簡單 app.route('/', consoleDashboardRouter); // T-cockpit ②：駕駛艙 dashboard（聚合 KBDB dash_* entries，無需登入唯讀） app.route('/', portalRouter); // portal-auth P2（#24/#25）：RAG Portal 多人授權——用戶模型＋認證 API app.route('/', portalDataRouter); // portal-auth P3：/portal/data/* server-side enforce（owner_id＋library 注入，安全核心） +app.route('/', storageRouter); // KV 退休（#16/#17）：資產遷移／盤點端點 // Worker 導出（fetch + scheduled） // scheduled handler 對應 wrangler.toml [triggers].crons，每分鐘 tick； // 邏輯在 src/scheduled.ts。對應 SDD: arcrun.md 三-A P1 #3。 +// +// 🔴 KV 退休（Leo/Arcrun#16 + #17）：WEBHOOKS / RECIPES 在這裡被換成 KBDB 撐腰的版本 +//（lib/durable-store.ts）。**這是唯一的接線點**——換在入口，四十幾處呼叫端一行不動， +// 也就沒有「某一處忘了改」這種漏洞（那正是資產會不見的入口）。 +// 使用者的工作流與 recipe 從此住在 KBDB（D1，一份資產一列 entry），KV 只是快取： +// KV 被換掉／重建之後，資料仍在，且會在下一次讀取時自己長回快取。 export default { - fetch: app.fetch, - scheduled: handleScheduled, + fetch: (req: Request, env: Bindings, ctx: ExecutionContext) => app.fetch(req, withDurableStores(env), ctx), + scheduled: (event: ScheduledController, env: Bindings, ctx: ExecutionContext) => + handleScheduled(event, withDurableStores(env), ctx), } satisfies ExportedHandler; diff --git a/cypher-executor/src/lib/asset-keys.ts b/cypher-executor/src/lib/asset-keys.ts new file mode 100644 index 0000000..523c411 --- /dev/null +++ b/cypher-executor/src/lib/asset-keys.ts @@ -0,0 +1,156 @@ +/** + * asset-keys — 哪些 KV key 是「使用者的資產」，以及它在 KBDB 裡對應哪一列 + * + * KV 退休（Leo/Arcrun#16 + #17）。leo 2026-08-12： + * 「我要的是寫進 KBDB，不是 KV，他的 Recipes、Cypher 是一段話，文字，數據，一個 entry」 + * 「如果零件和工作流的 recipe 不見了，是很可怕的事情」 + * + * ── 這支檔在整件事裡的位置 ──────────────────────────────────────────────── + * 真正的修法只有一句：**資產的家在 KBDB，KV 降級成可丟棄的快取**。 + * 但 KV 的呼叫端有四十幾處（webhooks-named / portal / executions / component-loader / + * auth-dispatcher / wasi-shim …），逐處改寫既冗長又容易漏一處——漏掉的那處就是下一次 + * 「東西不見了」的入口。所以改法是換掉 binding 本身（見 durable-store.ts）， + * 而這支檔就是那層唯一需要人看懂的東西：**一張表，說清楚哪些 key 是資產、對應哪一列。** + * + * ── 判準：資產 vs 衍生 ──────────────────────────────────────────────────── + * 資產 = 弄丟了就再也回不來的東西（使用者/AI 寫出來的工作流、recipe）。→ 進 KBDB。 + * 衍生 = 從資產算得出來的東西（反查索引 idx:*、cron 索引、快取、session、 + * 帶 TTL 的暫存）。→ 留在 KV，弄丟了自己重建（durable-store 負責重建）。 + * + * 判斷不確定時一律歸「衍生」——把衍生誤存進 KBDB 只是多幾列垃圾， + * 把資產誤判成衍生才是把人家的東西弄丟。 + * + * ── 為什麼不做成「KV key 原樣鏡射進 KBDB」 ──────────────────────────────── + * 那樣 KBDB 會變成第二顆 KV，leo 要的「一個 entry」就不成立：搜不到、看不懂、 + * 對 portal/console 也沒有意義。所以這裡把每個 key 解析成**有語意的一列** + * （entry_type + owner_id + page_name + 一句描述），KBDB 端才真的是「他的資產」。 + */ + +/** KBDB 裡 arcrun 資產的 entry_type（每個都有一列 template 定義，見 kbdb/migrations/0005）。 */ +export type AssetEntryType = 'workflow_def' | 'api_recipe' | 'auth_recipe' | 'prompt_recipe'; + +export interface AssetRef { + entry_type: AssetEntryType; + /** KBDB entries.id——由 key 決定，同一份資產永遠同一列（冪等，重跑遷移不長重複）。 */ + entry_id: string; + /** 租戶。recipe 是整台實例共用的庫，故為 null（與現行 RECIPES KV 無租戶前綴一致）。 */ + owner_id: string | null; + /** 該型別的自然鍵（workflow 名 / recipe uuid / service 名），對應 entries.page_name。 */ + page_name: string; + /** 原本的 KV key——反向重建快取時要用（KBDB → KV 回填）。 */ + kv_key: string; +} + +/** entries.id 的前綴，跟別人的資料分得開，也讓 `arcrun:` 一眼看得出是誰的列。 */ +const ID_PREFIX = 'arcrun'; + +/** + * 把一個 KV key 解析成 KBDB 的一列；不是資產就回 null（呼叫端原樣走 KV）。 + * + * ⚠️ 這是**唯一**決定「什麼進 KBDB」的地方。要新增一類資產就加在這裡， + * 不要在別處另開一條偷偷寫 KBDB 的路——兩套並存必然漂移（2026-08-08 credential 的教訓）。 + */ +export function classifyAssetKey(key: string): AssetRef | null { + // ── 明確排除的衍生資料（放最前面，免得被下面的樣式誤收）──────────────── + // idx:* recipe/component 反查索引（canonical→uuid、hash→canonical） + // cron-idx:* cron 排程索引（8.P0 的單一 key） + // 兩者都能從資產重算，見 durable-store.ts 的 rehydrate*。 + if (key.startsWith('idx:') || key.startsWith('cron-idx:')) return null; + + // auth_recipe:{service} — 「怎麼認證」的定義。注意只有定義，沒有任何密文 + //（憑證明文在 CF Workers Secrets，見 .claude/rules/01-tech-stack.md「Credential 儲存規範」）。 + if (key.startsWith('auth_recipe:')) { + const service = key.slice('auth_recipe:'.length); + if (!service) return null; + return { + entry_type: 'auth_recipe', + entry_id: `${ID_PREFIX}:auth_recipe:${service}`, + owner_id: null, + page_name: service, + kv_key: key, + }; + } + + // prompt_recipe:{name} + if (key.startsWith('prompt_recipe:')) { + const name = key.slice('prompt_recipe:'.length); + if (!name) return null; + return { + entry_type: 'prompt_recipe', + entry_id: `${ID_PREFIX}:prompt_recipe:${name}`, + owner_id: null, + page_name: name, + kv_key: key, + }; + } + + // recipe:{uuid}｜recipe:{canonical_id}（migration 前的舊 key，仍是資產，一樣要保住） + if (key.startsWith('recipe:')) { + const id = key.slice('recipe:'.length); + if (!id) return null; + return { + entry_type: 'api_recipe', + entry_id: `${ID_PREFIX}:recipe:${id}`, + owner_id: null, + page_name: id, + kv_key: key, + }; + } + + // {api_key}:wf:{name} — 具名工作流（acr push / portal 安裝器寫的那把）。 + // 用**第一個** ':wf:' 切：api_key 不含冒號，而 workflow 名允許的字元集 + //（webhooks-named.ts 驗 /^[\w-]+$/）本來就不含冒號，所以切點唯一。 + const wfAt = key.indexOf(':wf:'); + if (wfAt > 0) { + const owner = key.slice(0, wfAt); + const name = key.slice(wfAt + ':wf:'.length); + if (!owner || !name) return null; + return { + entry_type: 'workflow_def', + entry_id: `${ID_PREFIX}:wf:${owner}:${name}`, + owner_id: owner, + page_name: name, + kv_key: key, + }; + } + + // 其餘一律衍生/暫存：匿名 webhook token、daemon-active、session、stats… 留在 KV。 + return null; +} + +/** 從 KBDB 的一列反推回原本的 KV key（快取回填、list 都要用）。 */ +export function assetKvKey(entryType: AssetEntryType, ownerId: string | null, pageName: string): string { + switch (entryType) { + case 'workflow_def': + return `${ownerId ?? ''}:wf:${pageName}`; + case 'api_recipe': + return `recipe:${pageName}`; + case 'auth_recipe': + return `auth_recipe:${pageName}`; + case 'prompt_recipe': + return `prompt_recipe:${pageName}`; + } +} + +/** + * 一個 KV list 的 prefix 該去 KBDB 撈哪一類資產。 + * + * 為什麼 list 一定要走 KBDB（不能像 get 那樣先問快取）：被換掉的那顆 KV 是**空的**， + * 空 KV list 出來是「零筆」而不是「查不到」——這正是 2026-08-12 那天畫面上 + * 「九支工作流全部消失」的形狀。get 可以 KV 先行（miss 再回源），list 不行。 + * + * 回 null = 這個 prefix 不是資產類（例如 cron-idx:），照舊走 KV。 + */ +export function classifyListPrefix(prefix: string | undefined): { entry_type: AssetEntryType; owner_id?: string } | null { + if (!prefix) return null; // 無 prefix 的全域 list（webhooks-list）維持原行為 + if (prefix.startsWith('idx:') || prefix.startsWith('cron-idx:')) return null; + if (prefix === 'auth_recipe:') return { entry_type: 'auth_recipe' }; + if (prefix === 'prompt_recipe:') return { entry_type: 'prompt_recipe' }; + if (prefix === 'recipe:') return { entry_type: 'api_recipe' }; + // `{api_key}:wf:` — 列出某租戶的所有工作流（webhooks-named / portal-data 都用這個） + if (prefix.endsWith(':wf:')) { + const owner = prefix.slice(0, -':wf:'.length); + if (owner) return { entry_type: 'workflow_def', owner_id: owner }; + } + return null; +} diff --git a/cypher-executor/src/lib/durable-store.ts b/cypher-executor/src/lib/durable-store.ts new file mode 100644 index 0000000..866aac0 --- /dev/null +++ b/cypher-executor/src/lib/durable-store.ts @@ -0,0 +1,448 @@ +/** + * durable-store — 讓 KV 變成快取，資產的家搬到 KBDB + * + * KV 退休（Leo/Arcrun#16 + #17）。leo 2026-08-12： + * 「因為對 KV 的使用有禁令，但卻會把資產放在這裡，這不是違法嗎？」 + * 「現在是我幫他寫工作流，未來是他的 AI 自己寫工作流， + * 如果零件和工作流的 recipe 不見了，是很可怕的事情」 + * 同一天實害：一次例行更新讓使用者的九支工作流在畫面上全部消失 + *（根因見 cli/src/lib/resource-resolver.ts 檔頭 Arcrun#97）。 + * + * ── 一句話 ──────────────────────────────────────────────────────────────── + * **資產寫進 KBDB（D1，一份資產一列 entry）；KV 退成可丟棄的快取。** + * 換掉／重建 KV 之後，資料仍在 KBDB，而且會在下一次讀取時自己長回快取裡。 + * + * ── 為什麼是換掉 binding，不是改四十幾處呼叫端 ──────────────────────────── + * WEBHOOKS / RECIPES 的呼叫端散在 webhooks-named、portal、executions、component-loader、 + * auth-dispatcher、wasi-shim…… 逐處改寫既冗長又一定會漏，而**漏掉的那一處就是下一次 + * 「東西不見了」的入口**。所以在 worker 入口把 binding 換成本檔的包裝，呼叫端一行不動—— + * 「哪些 key 是資產」則集中在 asset-keys.ts 那一張表裡，是這件事唯一需要人看懂的東西。 + * + * ── 三條行為規則 ────────────────────────────────────────────────────────── + * 1. **讀**：先問 KV（快）→ 沒有就回源 KBDB → 順手把快取補回去（空 KV 自己痊癒）。 + * 2. **寫**：先寫 KBDB（真相），成功了才寫 KV 快取。 + * KBDB 寫失敗 → **拋錯**，不假裝部署成功——「看起來成功、其實沒存到」正是這張票的病。 + * 3. **列舉**：一律走 KBDB，**不准問 KV**。被換掉的那顆 KV 是空的， + * 空 KV 列出來是「零筆」而不是「查不到」——那正是「九支工作流全部消失」的形狀。 + * + * ── 衍生資料（idx:* / cron-idx:*）怎麼辦 ────────────────────────────────── + * 它們算得出來，所以不進 KBDB（KBDB 不該長出垃圾列），改成**讀不到就重算**。 + * 見 rehydrateRecipeIndices / rehydrateCronIndex。 + */ + +import { classifyAssetKey, classifyListPrefix, assetKvKey, type AssetRef, type AssetEntryType } from './asset-keys'; +import { CRON_INDEX_KEY, cronEntryKey, type CronIndex } from './cron-index'; + +export interface KbdbEnv { + KBDB_BASE_URL?: string; + KBDB_INTERNAL_TOKEN?: string; +} + +/** KBDB 位址與 token 的取法沿用既有慣例（lib/workflow-search.ts、routes/webhooks-named.ts 同款）。 */ +function kbdbBase(env: KbdbEnv): string { + return (env.KBDB_BASE_URL ?? 'https://arcrun-kbdb.uncle6-me.workers.dev').replace(/\/$/, ''); +} + +function kbdbHeaders(env: KbdbEnv): Record { + const h: Record = { 'Content-Type': 'application/json' }; + if (env.KBDB_INTERNAL_TOKEN) h['Authorization'] = `Bearer ${env.KBDB_INTERNAL_TOKEN}`; + return h; +} + +/** KBDB 一列 entry 的回應形狀（只取本檔用得到的欄位）。 */ +interface KbdbEntry { + id: string; + content?: string | null; + entry_type?: string | null; + owner_id?: string | null; + page_name?: string | null; + metadata_json?: string | null; + updated_at?: number; +} + +/** 資產寫進 metadata_json 的信封。definition = 原值 parse 過的物件；非 JSON 的原字串走 definition_raw。 */ +interface AssetEnvelope { + arcrun_asset: true; + kv_key: string; + definition?: unknown; + definition_raw?: string; + /** api_recipe 專用：本部署目前安裝的是不是這一版（重建 idx:installed:* 用，見 rehydrateRecipeIndices）。 */ + installed?: boolean; +} + +export class KbdbUnavailableError extends Error { + constructor(op: string, detail: string) { + super( + `資產無法寫入 KBDB（${op}）：${detail}。` + + '本次操作已中止且未寫入任何一邊——這是刻意的：寧可讓你現在看到失敗，' + + '也不要寫進只會被下次更新換掉的 KV、事後才發現東西不見了（Leo/Arcrun#16、#17）。', + ); + this.name = 'KbdbUnavailableError'; + } +} + +/** 從資產定義裡挑一句「給人看也給搜尋看」的描述，當 entries.content。 */ +function assetContent(type: AssetEntryType, def: unknown, pageName: string): string { + const d = (def ?? {}) as Record; + const pick = (...keys: string[]): string => { + for (const k of keys) { + const v = d[k]; + if (typeof v === 'string' && v.trim()) return v.trim(); + } + return ''; + }; + switch (type) { + case 'workflow_def': + return pick('description') || pageName; + case 'api_recipe': + return pick('description', 'display_name', 'canonical_id') || pageName; + case 'auth_recipe': + return pick('description', 'display_name', 'service') || pageName; + case 'prompt_recipe': + return pick('description', 'name') || pageName; + } +} + +/** 把 KBDB 一列還原成原本的 KV 值（字串）。不是資產信封（或壞掉）→ null，誠實當作沒有。 */ +function entryToKvValue(entry: KbdbEntry | null | undefined): string | null { + if (!entry?.metadata_json) return null; + try { + const env = JSON.parse(entry.metadata_json) as AssetEnvelope; + if (!env || env.arcrun_asset !== true) return null; + if (typeof env.definition_raw === 'string') return env.definition_raw; + if (env.definition === undefined) return null; + return JSON.stringify(env.definition); + } catch { + return null; + } +} + +/** + * KV binding 的替身：介面與 KVNamespace 同形，行為見檔頭三條規則。 + * 一個 request 建一個實例（rehydrate 的去重旗標是 per-instance 的）。 + */ +export class DurableKv { + private rehydratedRecipeIdx = false; + private rehydratedCronIdx = false; + + constructor( + private readonly kv: KVNamespace, + private readonly env: KbdbEnv, + ) {} + + /** + * 拿回底層那顆真正的 KV。**只有遷移／盤點會用到**（routes/storage.ts）： + * 那兩支的工作正是「比較 KV 那邊有什麼、KBDB 這邊有什麼」， + * 若透過包裝去問，list 會被導去 KBDB，就永遠比不出差異、也搬不動舊資料。 + * 一般業務程式碼不該碰這支——碰了就等於繞過本卷的全部保護。 + */ + get rawKv(): KVNamespace { + return this.kv; + } + + // ── KBDB 存取原語 ────────────────────────────────────────────────────── + + private async kbdbGetEntry(entryId: string): Promise { + const res = await fetch(`${kbdbBase(this.env)}/entries/${encodeURIComponent(entryId)}`, { + headers: kbdbHeaders(this.env), + }); + if (res.status === 404) return null; + if (!res.ok) return null; // 讀不到就當沒有；快取仍可能有值，不炸讀取路徑 + const json = (await res.json().catch(() => null)) as { entry?: KbdbEntry } | null; + return json?.entry ?? null; + } + + private async kbdbListEntries(entryType: AssetEntryType, ownerId?: string): Promise { + const params = new URLSearchParams({ entry_type: entryType, limit: '1000' }); + if (ownerId) params.set('owner_id', ownerId); + const res = await fetch(`${kbdbBase(this.env)}/entries?${params.toString()}`, { + headers: kbdbHeaders(this.env), + }); + if (!res.ok) throw new KbdbUnavailableError('list', `HTTP ${res.status}`); + const json = (await res.json().catch(() => null)) as { entries?: KbdbEntry[] } | null; + return json?.entries ?? []; + } + + private async kbdbPutEntry(ref: AssetRef, envelope: AssetEnvelope): Promise { + const content = assetContent(ref.entry_type, envelope.definition, ref.page_name); + const res = await fetch(`${kbdbBase(this.env)}/entries/${encodeURIComponent(ref.entry_id)}`, { + method: 'PUT', + headers: kbdbHeaders(this.env), + body: JSON.stringify({ + entry_type: ref.entry_type, + owner_id: ref.owner_id, + page_name: ref.page_name, + content, + // 刻意**不**標 embed:true：工作流的語意搜尋走既有的 entry_type='workflow' 那一列 + //（workflow-discovery 方案 C 的雙寫），這裡標了會變成同一支工作流嵌兩份向量。 + metadata_json: JSON.stringify(envelope), + }), + }); + if (!res.ok) throw new KbdbUnavailableError('put', `HTTP ${res.status} @ ${ref.entry_id}`); + } + + private async kbdbDeleteEntry(entryId: string): Promise { + const res = await fetch(`${kbdbBase(this.env)}/entries/${encodeURIComponent(entryId)}`, { + method: 'DELETE', + headers: kbdbHeaders(this.env), + }); + // 404 = 本來就沒有，對刪除而言是成功（冪等）。 + if (!res.ok && res.status !== 404) throw new KbdbUnavailableError('delete', `HTTP ${res.status} @ ${entryId}`); + } + + // ── 衍生索引重建（讀不到就重算，不進 KBDB）──────────────────────────── + + /** + * 從 KBDB 的 api_recipe 列重建 recipe 反查索引： + * idx:{hash_id} → canonical_id + * idx:canonical:{canonical} → [uuid, ...] + * idx:installed:{canonical} → uuid + * + * installed 的還原順序：先看資產自己標的 installed 旗標（正常路徑，寫入時就記下了， + * 見 put() 對 `idx:installed:` 的處理）；同一個 canonical 沒有任何一版標記時 + *（＝遷移之前就存在的舊資料），退而取 updated_at 最新的那一版——因為 + * installRecipeRecord 的語意本來就是「最後寫入的那版即為安裝版」。 + * 這是還原不是猜測，但仍是**退路**，故在此寫明白。 + */ + private async rehydrateRecipeIndices(): Promise { + if (this.rehydratedRecipeIdx) return; + this.rehydratedRecipeIdx = true; + + const entries = await this.kbdbListEntries('api_recipe'); + const byCanonical = new Map>(); + const writes: Array> = []; + + for (const e of entries) { + const raw = entryToKvValue(e); + if (!raw) continue; + let def: { uuid?: string; canonical_id?: string; hash_id?: string }; + try { def = JSON.parse(raw) as typeof def; } catch { continue; } + if (!def.canonical_id) continue; + + if (def.hash_id) writes.push(this.kv.put(`idx:${def.hash_id}`, def.canonical_id)); + if (!def.uuid) continue; + + let installed = false; + try { + installed = (JSON.parse(e.metadata_json ?? '{}') as AssetEnvelope).installed === true; + } catch { /* 壞信封 → 當作沒標記，走 updated_at 退路 */ } + + const list = byCanonical.get(def.canonical_id) ?? []; + list.push({ uuid: def.uuid, installed, updated_at: e.updated_at ?? 0 }); + byCanonical.set(def.canonical_id, list); + } + + for (const [canonical, versions] of byCanonical) { + writes.push(this.kv.put(`idx:canonical:${canonical}`, JSON.stringify(versions.map((v) => v.uuid)))); + const chosen = + versions.find((v) => v.installed) ?? + versions.reduce((a, b) => (b.updated_at > a.updated_at ? b : a)); + writes.push(this.kv.put(`idx:installed:${canonical}`, chosen.uuid)); + } + await Promise.all(writes); + } + + /** 從 KBDB 的 workflow_def 列重建 cron 索引（單一 key，見 lib/cron-index.ts）。 */ + private async rehydrateCronIndex(): Promise { + if (this.rehydratedCronIdx) return; + this.rehydratedCronIdx = true; + + const entries = await this.kbdbListEntries('workflow_def'); + const index: CronIndex = {}; + for (const e of entries) { + const raw = entryToKvValue(e); + if (!raw) continue; + let def: { cron_expr?: string }; + try { def = JSON.parse(raw) as typeof def; } catch { continue; } + if (!def.cron_expr || !e.owner_id || !e.page_name) continue; + index[cronEntryKey(e.owner_id, e.page_name)] = def.cron_expr; + } + // 即使是空的也要寫回去：寫了之後 get 就命中，下一分鐘的 tick 不會再重算一次 + //（不寫的話 scheduled() 每分鐘都會回源 KBDB 一趟，白花錢）。 + await this.kv.put(CRON_INDEX_KEY, JSON.stringify(index)); + } + + // ── KVNamespace 介面 ─────────────────────────────────────────────────── + + async get(key: string, type?: 'text' | 'json' | 'arrayBuffer' | 'stream' | { type: string }): Promise { + // 二進位／串流形態本 worker 沒有呼叫端在用（資產都是 JSON 文字）。原樣轉發， + // 不假裝支援——真有人開始用而拿不到 KBDB 回源，會在這裡被看見，不是靜默降級。 + const t0 = typeof type === 'string' ? type : type?.type; + if (t0 === 'arrayBuffer' || t0 === 'stream') return this.kv.get(key, t0 as 'arrayBuffer'); + + const asText = (raw: string | null): unknown => { + if (raw === null) return null; + const t = typeof type === 'string' ? type : type?.type; + if (t === 'json') { + try { return JSON.parse(raw); } catch { return null; } + } + return raw; + }; + + const ref = classifyAssetKey(key); + if (!ref) { + // 衍生/暫存：原樣走 KV。讀不到而且是「算得出來」的索引 → 重算一次再讀。 + const raw = await this.kv.get(key, 'text'); + if (raw !== null) return asText(raw); + if (key.startsWith('idx:')) { + await this.rehydrateRecipeIndices().catch(() => {}); + return asText(await this.kv.get(key, 'text')); + } + if (key === CRON_INDEX_KEY) { + await this.rehydrateCronIndex().catch(() => {}); + return asText(await this.kv.get(key, 'text')); + } + return asText(raw); + } + + // 資產：快取優先，miss 回源 KBDB 並補快取（被換掉的空 KV 就是這樣自己痊癒的）。 + const cached = await this.kv.get(key, 'text'); + if (cached !== null) return asText(cached); + + const fromKbdb = entryToKvValue(await this.kbdbGetEntry(ref.entry_id)); + if (fromKbdb === null) return asText(null); + await this.kv.put(key, fromKbdb).catch(() => {}); // 補快取失敗不影響這次讀取 + return asText(fromKbdb); + } + + async put(key: string, value: string | ArrayBuffer | ReadableStream, options?: KVNamespacePutOptions): Promise { + // 帶 TTL＝定義上就是暫存（daemon 回報、session…），不是資產，不進 KBDB。 + if (options?.expirationTtl || options?.expiration || typeof value !== 'string') { + return this.kv.put(key, value as string, options); + } + + // idx:installed:{canonical} 不是資產，但它記的是**使用者的選擇**（這個 canonical 目前 + // 裝的是哪一版），純算不回來。所以把它記進對應那一版 recipe 資產的信封裡， + // KBDB 端不會多出一列「指標 entry」，重建時又還原得精確（見 rehydrateRecipeIndices）。 + if (key.startsWith('idx:installed:')) { + await this.kv.put(key, value, options); + await this.markInstalledVersion(key.slice('idx:installed:'.length), value).catch(() => { + // 標記失敗不擋主流程：recipe 本體已經在 KBDB，最壞情況是重建時退回 updated_at 那條路。 + }); + return; + } + + const ref = classifyAssetKey(key); + if (!ref) return this.kv.put(key, value, options); + + let definition: unknown; + let definitionRaw: string | undefined; + try { definition = JSON.parse(value); } catch { definitionRaw = value; } + + // 覆寫定義不該把「這版是目前安裝的那版」洗掉 → 只有 api_recipe 需要先讀回舊信封。 + // 其他三型沒有這個欄位，省下這一次往返（每次 acr push 都會走到這裡）。 + const previous = ref.entry_type === 'api_recipe' ? await this.readEnvelope(ref) : null; + + // 先真相、後快取。KBDB 失敗就拋——不寫 KV、不回報成功（禁假綠，mindset §7）。 + await this.kbdbPutEntry(ref, { + arcrun_asset: true, + kv_key: key, + ...(definitionRaw !== undefined ? { definition_raw: definitionRaw } : { definition }), + ...(previous?.installed ? { installed: true } : {}), + }); + await this.kv.put(key, value, options); + } + + async delete(key: string): Promise { + const ref = classifyAssetKey(key); + if (ref) await this.kbdbDeleteEntry(ref.entry_id); + await this.kv.delete(key); + } + + async list(options?: KVNamespaceListOptions): Promise> { + const target = classifyListPrefix(options?.prefix ?? undefined); + if (!target) return this.kv.list(options) as Promise>; + + // 資產列舉一律回源（見檔頭規則 3）。順手把每一筆補進快取——列表回應本來就帶了完整內容， + // 呼叫端接著一筆筆 get 時就會全部命中，一次回源換掉 N 次往返。 + const entries = await this.kbdbListEntries(target.entry_type, target.owner_id); + const keys: Array<{ name: string }> = []; + const warm: Array> = []; + for (const e of entries) { + if (!e.page_name) continue; + // workflow_def 的 KV key 由租戶＋名字組成，缺租戶就組不出正確的 key—— + // 與其回一個 `:wf:x` 這種對不到任何東西的名字，不如跳過（列不出來看得見， + // 組錯名字則會安靜地讀到 null，那更難查）。 + if (target.entry_type === 'workflow_def' && !e.owner_id) continue; + const name = assetKvKey(target.entry_type, e.owner_id ?? null, e.page_name); + keys.push({ name }); + const raw = entryToKvValue(e); + if (raw !== null) warm.push(this.kv.put(name, raw).catch(() => {})); + } + await Promise.all(warm); + return { keys, list_complete: true, cacheStatus: null } as unknown as KVNamespaceListResult; + } + + /** KVNamespace 介面補齊（本 worker 沒有呼叫端在用，原樣轉發，不做資產處理）。 */ + getWithMetadata(key: string, type?: any): Promise { + return (this.kv as unknown as { getWithMetadata: (k: string, t?: any) => Promise }).getWithMetadata(key, type); + } + + // ── 內部小工具 ───────────────────────────────────────────────────────── + + private async readEnvelope(ref: AssetRef): Promise { + const entry = await this.kbdbGetEntry(ref.entry_id); + if (!entry?.metadata_json) return null; + try { + const env = JSON.parse(entry.metadata_json) as AssetEnvelope; + return env?.arcrun_asset === true ? env : null; + } catch { + return null; + } + } + + /** 把「這個 canonical 目前裝的是哪一版」記進該版 recipe 的信封（同 canonical 的其他版清掉旗標）。 */ + private async markInstalledVersion(canonicalId: string, uuid: string): Promise { + const entries = await this.kbdbListEntries('api_recipe'); + const jobs: Array> = []; + for (const e of entries) { + const raw = entryToKvValue(e); + if (!raw) continue; + let def: { uuid?: string; canonical_id?: string }; + try { def = JSON.parse(raw) as typeof def; } catch { continue; } + if (def.canonical_id !== canonicalId || !def.uuid) continue; + + const shouldBeInstalled = def.uuid === uuid; + let envelope: AssetEnvelope; + try { envelope = JSON.parse(e.metadata_json ?? '{}') as AssetEnvelope; } catch { continue; } + if ((envelope.installed === true) === shouldBeInstalled) continue; // 已經是對的，不白寫 + + envelope.installed = shouldBeInstalled; + jobs.push( + fetch(`${kbdbBase(this.env)}/entries/${encodeURIComponent(e.id)}`, { + method: 'PATCH', + headers: kbdbHeaders(this.env), + body: JSON.stringify({ metadata_json: JSON.stringify(envelope) }), + }), + ); + } + await Promise.all(jobs); + } +} + +/** + * 在 worker 入口把 WEBHOOKS / RECIPES 換成 KBDB 撐腰的版本。 + * + * 只換這兩個：它們裝的是**使用者寫出來的東西**（工作流、recipe）。 + * 其餘 KV（EXEC_CONTEXT 執行中暫存、SESSIONS_KV、ANALYTICS_KV、CREDENTIALS_KV） + * 要嘛是暫存、要嘛另有搬遷路徑（credential 走 CF Workers Secrets + D1 目錄， + * 見 .claude/rules/01-tech-stack.md），不在本卷範圍——**不順手一起動**。 + * + * KBDB_BASE_URL 沒設也照樣運作：kbdbBase() 有預設值；真的連不上時 + * 讀取路徑退回純 KV（維持現況、不比以前糟），寫入路徑誠實拋錯（不假裝存好了）。 + */ +export function withDurableStores(env: T): T { + const wrapped = { ...env } as T; + if (env.WEBHOOKS) wrapped.WEBHOOKS = new DurableKv(env.WEBHOOKS, env) as unknown as KVNamespace; + if (env.RECIPES) wrapped.RECIPES = new DurableKv(env.RECIPES, env) as unknown as KVNamespace; + return wrapped; +} + +/** + * 取得底層真 KV（沒被包裝就是它自己）。遷移／盤點專用，理由見 DurableKv.rawKv。 + * 寫成獨立函式是為了讓「誰在繞過包裝」grep 得出來——目前只有 routes/storage.ts。 + */ +export function unwrapKv(binding: KVNamespace): KVNamespace { + const maybe = binding as unknown as { rawKv?: KVNamespace }; + return maybe.rawKv ?? binding; +} diff --git a/cypher-executor/src/routes/portal-data.ts b/cypher-executor/src/routes/portal-data.ts index 3ffc3bf..8d3024e 100644 --- a/cypher-executor/src/routes/portal-data.ts +++ b/cypher-executor/src/routes/portal-data.ts @@ -132,7 +132,13 @@ function canReadLibrary(userLibraries: string[], library: string): boolean { // execution_log/execution_log_usage（KV 額度事故修復，2026-08-07）：workflow 執行紀錄與其內部 // 用量計數器，entry_type 與既有 value/workflow 同層級的內部型別——一併排除，避免用戶搜尋知識時 // 混進執行 log（同層防線：本模組也從不設 metadata_json.embed=true，永不進語意搜尋索引）。 -const INTERNAL_ENTRY_TYPES = new Set(['value', 'workflow', 'execution_log', 'execution_log_usage']); +// KV 退休（#16/#17）新增四型：資產定義本體住進 KBDB 之後，它們是「系統的東西」而不是 +// 使用者的知識卡——知識瀏覽／搜尋要跟 execution_log 一樣排除，否則 portal 會冒出 +// 一堆 workflow_def / api_recipe 汙染結果。 +const INTERNAL_ENTRY_TYPES = new Set([ + 'value', 'workflow', 'execution_log', 'execution_log_usage', + 'workflow_def', 'api_recipe', 'auth_recipe', 'prompt_recipe', +]); export function filterDeprecatedEntries( entries: T[], diff --git a/cypher-executor/src/routes/storage.ts b/cypher-executor/src/routes/storage.ts new file mode 100644 index 0000000..528d647 --- /dev/null +++ b/cypher-executor/src/routes/storage.ts @@ -0,0 +1,199 @@ +/** + * /storage — 資產盤點與搬遷（KV → KBDB） + * + * KV 退休（Leo/Arcrun#16 + #17）。交辦的驗收條件之一逐字是： + * 「既有的東西要能搬過去，而且**搬的過程不能弄丟任何一筆**（搬之前先數，搬之後再數）」 + * 所以這兩支端點的重點不是「搬」，是**數得出來**： + * GET /storage/audit 兩邊各有幾筆、差在哪幾筆（唯讀，先看再決定要不要搬） + * POST /storage/migrate-to-kbdb 搬，回傳搬之前的數、逐筆結果、搬之後的數 + * + * 三個刻意的設計： + * 1. **只增不刪**：搬完不動 KV 的原始資料。搬錯了、想反悔，原地還在； + * KV 本來就要退成快取，留著它一份沒有壞處（真的要清是另一個決定，不在這支裡順手做）。 + * 2. **冪等**：KBDB 端用固定的 entry id（asset-keys.ts），同一筆搬幾次都只有一列。 + * 可以放心重跑到 missing 歸零為止。 + * 3. **誠實**：每一筆的成敗逐筆列出來，失敗有原因；不吞錯、不四捨五入回一句「成功」 + * （禁假綠，mindset §7）。搬完 after 對不上 before 就是沒搬乾淨，數字自己會講。 + * + * flag 安全：兩支都是人／AI 主動呼叫一次的操作，**不掛 cron、不輪詢** + *（KV list 免費額度 1000/日，這裡會 list，所以更不能自動化重複打）。 + */ + +import { Hono } from 'hono'; +import type { Bindings } from '../types'; +import { unwrapKv } from '../lib/durable-store'; +import { classifyAssetKey, type AssetEntryType } from '../lib/asset-keys'; + +export const storageRouter = new Hono<{ Bindings: Bindings }>(); + +/** 本卷管的四類資產，以及它們住在哪顆 KV。 */ +const ASSET_SOURCES: Array<{ binding: 'WEBHOOKS' | 'RECIPES' }> = [ + { binding: 'WEBHOOKS' }, + { binding: 'RECIPES' }, +]; + +interface ScannedKey { + key: string; + binding: 'WEBHOOKS' | 'RECIPES'; + entry_type: AssetEntryType; + entry_id: string; +} + +/** 掃一顆 KV 的全部 key（跟著 cursor 走完，不只第一頁），挑出屬於資產的。 */ +async function scanAssetKeys(kv: KVNamespace, binding: 'WEBHOOKS' | 'RECIPES'): Promise { + const found: ScannedKey[] = []; + let cursor: string | undefined; + do { + const page = await kv.list(cursor ? { cursor } : {}); + for (const k of page.keys) { + const ref = classifyAssetKey(k.name); + if (ref) found.push({ key: k.name, binding, entry_type: ref.entry_type, entry_id: ref.entry_id }); + } + cursor = page.list_complete ? undefined : page.cursor; + } while (cursor); + return found; +} + +function kbdb(env: Bindings): { base: string; headers: Record } { + const base = (env.KBDB_BASE_URL ?? 'https://arcrun-kbdb.uncle6-me.workers.dev').replace(/\/$/, ''); + const headers: Record = { 'Content-Type': 'application/json' }; + if (env.KBDB_INTERNAL_TOKEN) headers['Authorization'] = `Bearer ${env.KBDB_INTERNAL_TOKEN}`; + return { base, headers }; +} + +/** KBDB 端某型別現有的 entry id 集合（用來算「哪幾筆還沒搬過去」）。 */ +async function kbdbExistingIds(env: Bindings, entryType: AssetEntryType): Promise> { + const { base, headers } = kbdb(env); + const res = await fetch(`${base}/entries?entry_type=${encodeURIComponent(entryType)}&limit=1000`, { headers }); + if (!res.ok) throw new Error(`KBDB 讀取失敗（${entryType}）：HTTP ${res.status}`); + const json = (await res.json().catch(() => null)) as { entries?: Array<{ id: string }> } | null; + return new Set((json?.entries ?? []).map((e) => e.id)); +} + +const ASSET_TYPES: AssetEntryType[] = ['workflow_def', 'api_recipe', 'auth_recipe', 'prompt_recipe']; + +interface Tally { + kv: Record; + kbdb: Record; + missing_in_kbdb: string[]; +} + +/** 兩邊各數一次，並列出「KV 有、KBDB 沒有」的那幾筆。 */ +async function tally(env: Bindings): Promise { + const scanned: ScannedKey[] = []; + for (const src of ASSET_SOURCES) { + const binding = env[src.binding]; + if (!binding) continue; + scanned.push(...(await scanAssetKeys(unwrapKv(binding), src.binding))); + } + + const kvCounts: Record = {}; + for (const t of ASSET_TYPES) kvCounts[t] = 0; + for (const s of scanned) kvCounts[s.entry_type] += 1; + + const kbdbCounts: Record = {}; + const existing = new Map>(); + for (const t of ASSET_TYPES) { + const ids = await kbdbExistingIds(env, t); + existing.set(t, ids); + kbdbCounts[t] = ids.size; + } + + const missing = scanned + .filter((s) => !existing.get(s.entry_type)!.has(s.entry_id)) + .map((s) => s.key); + + return { kv: kvCounts, kbdb: kbdbCounts, missing_in_kbdb: missing }; +} + +// GET /storage/audit — 唯讀盤點。搬之前先看、搬之後再看，兩次數字自己會說話。 +storageRouter.get('/storage/audit', async (c) => { + try { + const t = await tally(c.env); + return c.json({ + success: true, + kv_asset_counts: t.kv, + kbdb_asset_counts: t.kbdb, + missing_in_kbdb: t.missing_in_kbdb, + missing_count: t.missing_in_kbdb.length, + verdict: + t.missing_in_kbdb.length === 0 + ? 'KV 裡的資產在 KBDB 都有一份——換掉 KV 不會弄丟東西。' + : `還有 ${t.missing_in_kbdb.length} 筆只存在於 KV。跑 POST /storage/migrate-to-kbdb 把它們搬過去。`, + }); + } catch (e) { + // 誠實：盤點本身失敗就說失敗，不回一個看起來很乾淨的 0（那會被讀成「沒東西要搬」）。 + return c.json({ success: false, error: e instanceof Error ? e.message : String(e) }, 502); + } +}); + +// POST /storage/migrate-to-kbdb — 把 KV 裡的資產補進 KBDB。只增不刪、冪等、可重跑。 +// body（都可省略）：{ dry_run?: boolean } +storageRouter.post('/storage/migrate-to-kbdb', async (c) => { + const body = (await c.req.json().catch(() => ({}))) as { dry_run?: boolean }; + const dryRun = body.dry_run === true; + + let before: Tally; + try { + before = await tally(c.env); + } catch (e) { + return c.json({ success: false, error: `搬遷前盤點失敗，未動任何資料：${e instanceof Error ? e.message : String(e)}` }, 502); + } + + if (dryRun) { + return c.json({ + success: true, + dry_run: true, + before: { kv: before.kv, kbdb: before.kbdb }, + would_migrate: before.missing_in_kbdb, + would_migrate_count: before.missing_in_kbdb.length, + }); + } + + const migrated: string[] = []; + const errors: Array<{ key: string; error: string }> = []; + + for (const src of ASSET_SOURCES) { + const wrapped = c.env[src.binding]; + if (!wrapped) continue; + const raw = unwrapKv(wrapped); + for (const s of await scanAssetKeys(raw, src.binding)) { + try { + // 從**真** KV 讀原值，再用包裝過的 binding 寫回去——寫入路徑就是平常那條 + //（先 KBDB 後快取），所以搬遷用的是跟日常寫入完全同一段程式碼，不另開一條會漂移的路。 + const value = await raw.get(s.key, 'text'); + if (value === null) continue; // 掃到當下剛好被刪：不是錯，跳過 + await wrapped.put(s.key, value); + migrated.push(s.key); + } catch (e) { + errors.push({ key: s.key, error: e instanceof Error ? e.message : String(e) }); + } + } + } + + let after: Tally | null = null; + let afterError: string | null = null; + try { + after = await tally(c.env); + } catch (e) { + afterError = e instanceof Error ? e.message : String(e); + } + + const clean = errors.length === 0 && after !== null && after.missing_in_kbdb.length === 0; + return c.json( + { + success: clean, + before: { kv: before.kv, kbdb: before.kbdb, missing_in_kbdb: before.missing_in_kbdb.length }, + migrated, + migrated_count: migrated.length, + errors, + after: after + ? { kv: after.kv, kbdb: after.kbdb, missing_in_kbdb: after.missing_in_kbdb } + : { error: afterError }, + verdict: clean + ? '搬完了：KV 裡的每一筆資產在 KBDB 都有對應的一列（missing 歸零）。KV 原始資料原封不動保留。' + : '**沒有搬乾淨**——看 errors 與 after.missing_in_kbdb。修掉原因後可以直接重跑（冪等，不會產生重複）。', + }, + clean ? 200 : 500, + ); +}); diff --git a/cypher-executor/tests/asset-keys.test.ts b/cypher-executor/tests/asset-keys.test.ts new file mode 100644 index 0000000..d95ecc9 --- /dev/null +++ b/cypher-executor/tests/asset-keys.test.ts @@ -0,0 +1,91 @@ +/** + * asset-keys 單元測試 — 「哪些 KV key 是使用者的資產」這張表本身 + * + * KV 退休（Leo/Arcrun#16 + #17）。這支測的是純函式，沒有 KV / KBDB / 網路， + * 因為它要守的東西也很單純：**分類錯了，資產就會被留在會被換掉的那一層。** + * + * 兩個方向都要測，而且分量一樣重： + * - 資產不可以被誤判成衍生（漏收＝下次更新就不見了，這是 2026-08-12 的病） + * - 衍生不可以被誤判成資產（多收＝KBDB 長出算得出來的垃圾列） + */ +import { describe, it, expect } from 'vitest'; +import { classifyAssetKey, classifyListPrefix, assetKvKey } from '../src/lib/asset-keys'; +import { CRON_INDEX_KEY } from '../src/lib/cron-index'; + +describe('classifyAssetKey — 資產', () => { + it('具名工作流 {api_key}:wf:{name} → workflow_def，租戶與名字都拆得出來', () => { + const ref = classifyAssetKey('leo:wf:rag_chat'); + expect(ref).not.toBeNull(); + expect(ref!.entry_type).toBe('workflow_def'); + expect(ref!.owner_id).toBe('leo'); + expect(ref!.page_name).toBe('rag_chat'); + expect(ref!.kv_key).toBe('leo:wf:rag_chat'); + }); + + it('api recipe（uuid key 與 migration 前的 canonical key 都算資產）', () => { + expect(classifyAssetKey('recipe:8f3b-uuid')!.entry_type).toBe('api_recipe'); + expect(classifyAssetKey('recipe:telegram_send')!.entry_type).toBe('api_recipe'); + }); + + it('auth recipe / prompt recipe', () => { + expect(classifyAssetKey('auth_recipe:notion')!.entry_type).toBe('auth_recipe'); + expect(classifyAssetKey('auth_recipe:notion')!.page_name).toBe('notion'); + expect(classifyAssetKey('prompt_recipe:wiki_synthesis')!.entry_type).toBe('prompt_recipe'); + }); + + it('同一個 key 永遠對到同一個 entry_id（冪等的根據——重跑遷移不會長出重複列）', () => { + expect(classifyAssetKey('leo:wf:a')!.entry_id).toBe(classifyAssetKey('leo:wf:a')!.entry_id); + expect(classifyAssetKey('leo:wf:a')!.entry_id).not.toBe(classifyAssetKey('evan:wf:a')!.entry_id); + }); +}); + +describe('classifyAssetKey — 衍生資料不可誤收', () => { + it('recipe 反查索引 idx:* 全部不是資產（算得回來，見 rehydrateRecipeIndices）', () => { + expect(classifyAssetKey('idx:rec_f7e2a1b3')).toBeNull(); + expect(classifyAssetKey('idx:canonical:telegram_send')).toBeNull(); + expect(classifyAssetKey('idx:installed:telegram_send')).toBeNull(); + }); + + it('cron 索引不是資產', () => { + expect(classifyAssetKey(CRON_INDEX_KEY)).toBeNull(); + expect(classifyAssetKey('cron-idx:leo:daily')).toBeNull(); + }); + + it('匿名 webhook token / 其他暫存 key 不是資產（本卷不碰，非漏收）', () => { + expect(classifyAssetKey('a1b2c3d4e5f6')).toBeNull(); + expect(classifyAssetKey('daemon-active:leo')).toBeNull(); + }); + + it('殘缺的 key 不當資產（寧可退回原本的 KV 行為，也不要建出半截的列）', () => { + expect(classifyAssetKey('recipe:')).toBeNull(); + expect(classifyAssetKey('auth_recipe:')).toBeNull(); + expect(classifyAssetKey(':wf:orphan')).toBeNull(); + expect(classifyAssetKey('leo:wf:')).toBeNull(); + }); +}); + +describe('assetKvKey — 從 KBDB 反推回 KV key（回填快取與 list 都靠它）', () => { + it('四型都能原路折返', () => { + for (const key of ['leo:wf:rag_chat', 'recipe:telegram_send', 'auth_recipe:notion', 'prompt_recipe:x']) { + const ref = classifyAssetKey(key)!; + expect(assetKvKey(ref.entry_type, ref.owner_id, ref.page_name)).toBe(key); + } + }); +}); + +describe('classifyListPrefix — 列舉走哪一邊', () => { + it('租戶工作流列舉 → 走 KBDB（帶 owner）', () => { + expect(classifyListPrefix('leo:wf:')).toEqual({ entry_type: 'workflow_def', owner_id: 'leo' }); + }); + + it('recipe / auth_recipe 列舉 → 走 KBDB', () => { + expect(classifyListPrefix('recipe:')).toEqual({ entry_type: 'api_recipe' }); + expect(classifyListPrefix('auth_recipe:')).toEqual({ entry_type: 'auth_recipe' }); + }); + + it('索引與無 prefix 的全域列舉 → 維持原本的 KV 行為', () => { + expect(classifyListPrefix('idx:')).toBeNull(); + expect(classifyListPrefix('cron-idx:')).toBeNull(); + expect(classifyListPrefix(undefined)).toBeNull(); + }); +}); diff --git a/kbdb/migrations/0005_arcrun_asset_templates.sql b/kbdb/migrations/0005_arcrun_asset_templates.sql new file mode 100644 index 0000000..c13c553 --- /dev/null +++ b/kbdb/migrations/0005_arcrun_asset_templates.sql @@ -0,0 +1,45 @@ +-- arcrun 資產型別 template seed — KV 退休（Leo/Arcrun#16 + #17） +-- +-- 為什麼有這一檔（leo 2026-08-12 原話）： +-- 「我要的是寫進 KBDB，不是 KV，他的 Recipes、Cypher 是一段話，文字，數據，一個 entry」 +-- 「如果零件和工作流的 recipe 不見了，是很可怕的事情」 +-- 同一天（2026-08-12）真的發作過：一次例行更新讓使用者的九支工作流在畫面上全部消失 +-- （根因見 cli/src/lib/resource-resolver.ts 檔頭 Arcrun#97——舊 deploy 會照名字新建一顆空 KV +-- 再綁上去）。#97 修的是「不要再把 worker 綁到空的資源上」；本卷修的是更根本的一句： +-- **使用者的資產本來就不該只存在於一個會被換掉的暫存層裡。** +-- +-- KBDB 鐵律（leo 2026-06-14，D38）：三張表打天下，永遠不加新 table，新資料類型一律用 template。 +-- 本檔**零 schema 異動**——只 INSERT OR IGNORE 四列 template 定義，手法與同目錄 +-- 0003_library_map.sql / 0004_execution_log_template.sql 完全相同。 +-- +-- 儲存精神比照 0004（execution_log）與 recipe-stat：template 只負責「schema 文件化 + +-- GET /templates 可發現」，實際一筆資產是 entries 表的**一列**—— +-- entry_type = 'workflow_def' | 'api_recipe' | 'auth_recipe' | 'prompt_recipe' +-- owner_id = 租戶（workflow 才有；recipe 是整台實例共用的庫，故為 NULL） +-- page_name = 該型別的自然鍵（workflow 名 / recipe uuid / service 名） +-- content = 給人看也給語意搜尋看的一句描述 +-- metadata_json = 定義本體（graph / endpoint / inject … 原樣 JSON） +-- ——不走 entry_values 全展開的多列 record：一支 workflow 的 graph 是一整包巢狀 JSON， +-- 拆成 slot 多列既不會變得比較好查，反而讓「一筆資產＝一列」這件事不再成立 +-- （recipe_stat 與 execution_log 早已示範「template 存在 ＋ entries 直接存」這個模式合法）。 +-- +-- 讀寫一律走 HTTP API（/entries、/entries/:id），呼叫端是 cypher-executor 的 +-- src/lib/durable-store.ts。牆外沒有任何一行 SQL。 + +INSERT OR IGNORE INTO templates (id, name, description, slots_json, created_by) +VALUES + ('tpl-workflow-def', 'workflow_def', + '工作流定義本體（KV 退休 #17）。一支工作流＝entries 一列；graph/config/cron_expr 打包進 metadata_json，WEBHOOKS KV 降為可丟棄的快取', + '["name","description","graph","config","cron_expr","created_at"]', 'system'), + + ('tpl-api-recipe', 'api_recipe', + 'API recipe 定義本體（KV 退休 #16）。一份 recipe＝entries 一列；endpoint/headers/body/auth 等打包進 metadata_json，RECIPES KV 降為快取。idx:* 反查索引屬衍生資料，不進 KBDB，由 durable-store 從本型別重建', + '["uuid","canonical_id","hash_id","author","endpoint","method","auth_service","installed"]', 'system'), + + ('tpl-auth-recipe', 'auth_recipe', + 'Auth recipe 定義本體（KV 退休 #16）。一個服務一列；primitive/base_url/required_secrets/inject 打包進 metadata_json。只存「怎麼認證」，不存任何密文（憑證明文在 CF Workers Secrets，見 .claude/rules/01-tech-stack.md）', + '["service","primitive","base_url","version","required_secrets","inject"]', 'system'), + + ('tpl-prompt-recipe', 'prompt_recipe', + 'Prompt recipe 定義本體（KV 退休 #16）。一份 prompt recipe＝entries 一列，定義打包進 metadata_json', + '["name","definition"]', 'system'); diff --git a/kbdb/src/actions/entry-crud.ts b/kbdb/src/actions/entry-crud.ts index d7bf4d6..01bf459 100644 --- a/kbdb/src/actions/entry-crud.ts +++ b/kbdb/src/actions/entry-crud.ts @@ -134,6 +134,51 @@ export async function deleteEntry(db: D1Database, id: string): Promise { await db.prepare('DELETE FROM entries WHERE id = ?').bind(id).run(); } +/** + * 以「呼叫端指定的 id」寫一列——已存在就整列覆蓋，不存在就新建（KV 退休 #16/#17）。 + * + * 為什麼 base 需要這一支：`createEntry` 的 id 預設是隨機的，同一份資產每存一次就多一列； + * 呼叫端若想要「同一份資產永遠是同一列」，只能自己先 GET 再決定 POST 還是 PATCH—— + * 兩趟往返、而且中間有競態。把它收成一個原語，語意才單一（冪等：同樣的輸入跑幾次結果都一樣）。 + * + * 這是 base 的通用能力，不是替某個 entry_type 開的特例——任何有天然鍵的資料型別 + * （arcrun 的 workflow_def / api_recipe / auth_recipe，或未來別的）都用得上。 + * **零 schema 異動**：仍然只寫 entries 這一張既有的表。 + * + * 覆蓋語意刻意是「整列取代」而非 PATCH 式合併：呼叫端手上是一份完整的資產定義， + * 合併語意會讓「刪掉某個欄位」變成做不到的事（舊值會留下來）。 + * created_at 保留原值（資產的誕生時間不因為改一次內容就被重寫），updated_at 更新。 + */ +export async function upsertEntry(db: D1Database, id: string, input: Omit): Promise { + const existing = await getEntry(db, id); + if (!existing) return createEntry(db, { ...input, id }); + await db + .prepare( + `UPDATE entries + SET content = ?, entry_type = ?, owner_id = ?, parent_id = ?, page_name = ?, + refs_json = ?, tags_json = ?, task_status = ?, confidence = ?, metadata_json = ?, + updated_at = unixepoch() + WHERE id = ?`, + ) + .bind( + input.content ?? null, + input.entry_type, + input.owner_id ?? null, + input.parent_id ?? null, + input.page_name ?? null, + input.refs_json ?? '[]', + input.tags_json ?? '[]', + input.task_status ?? null, + input.confidence ?? null, + input.metadata_json ?? null, + id, + ) + .run(); + const row = await getEntry(db, id); + if (!row) throw new Error('upsertEntry: update succeeded but row not found'); + return row; +} + /** * 把某 owner 下某庫的所有 entries 標 deprecated（t135 by-name 移除語意）。 * 沿用既有 deprecated 機制：metadata_json.status='deprecated' → 搜尋端過濾、庫列表排除。 diff --git a/kbdb/src/routes/entries.ts b/kbdb/src/routes/entries.ts index f553ef5..4125782 100644 --- a/kbdb/src/routes/entries.ts +++ b/kbdb/src/routes/entries.ts @@ -9,6 +9,7 @@ import { getEntry, listEntries, updateEntry, + upsertEntry, deleteEntry, searchEntries, isDeprecatedEntry, @@ -426,6 +427,21 @@ entryRoutes.get('/backfill-library/status', async (c) => { return c.json({ success: true, ...status }); }); +// PUT /entries/:id — 以呼叫端指定的 id 整列覆寫（不存在就新建）。KV 退休 #16/#17。 +// +// 與 POST / 的差別：POST 的 id 是隨機的，同一份資產每存一次多一列；PUT 讓「同一份資產永遠 +// 是同一列」，所以重跑遷移、重複部署同一支工作流都不會長出重複資料（冪等）。 +// 與 PATCH /:id 的差別：PATCH 是部分更新（沒帶的欄位留著），PUT 是整列取代 +// ——呼叫端手上是完整定義時要的是後者，否則「刪掉一個欄位」永遠做不到。 +entryRoutes.put('/:id', async (c) => { + const body = await c.req.json().catch(() => null); + if (!body || !body.entry_type) return c.json({ success: false, error: 'entry_type required' }, 400); + const entry = await upsertEntry(c.env.DB, c.req.param('id'), body); + // 與 POST / 同款：標了 embed:true 的才進 Vectorize，fire-and-forget、失敗不致命。 + if (embedEnabled(c.env)) c.executionCtx.waitUntil(embedOnWrite(c.env, entry).catch(() => {})); + return c.json({ success: true, entry }); +}); + // PATCH /entries/:id entryRoutes.patch('/:id', async (c) => { const body = await c.req.json().catch(() => ({})); diff --git a/scripts/verify-kv-retirement.sh b/scripts/verify-kv-retirement.sh new file mode 100644 index 0000000..8db3c6b --- /dev/null +++ b/scripts/verify-kv-retirement.sh @@ -0,0 +1,146 @@ +#!/usr/bin/env bash +# +# verify-kv-retirement.sh — 把「換掉 KV，資產還在」真的做一次 +# +# KV 退休（Leo/Arcrun#16 + #17）。交辦的驗收條件逐字是： +# 「證明『換掉／重建那個暫存層，資產還在』——不是說明它會在，是**真的弄一次給我看**」 +# 「既有的東西要能搬過去，而且搬的過程不能弄丟任何一筆（搬之前先數，搬之後再數）」 +# 這支腳本就是那一次。它做的事，照順序： +# +# 1. 開一台**全新的空**本機實例（local D1 + local KV，跑真的 migrations） +# 2. 用平常那條路（POST /webhooks/named、POST /recipes、POST /auth-recipes） +# 放進 9 支工作流 + 3 份 recipe——9 是照 2026-08-12 那天真的消失的數量 +# 3. 數一次（KV 幾筆、KBDB 幾筆） +# 4. **把整個 KV 層砍掉重建**（rm -rf 那顆 KV 的本機儲存 → 重開 worker） +# ＝模擬 Arcrun#97 那天發生的事：worker 被綁到一顆全新的空 KV +# 5. 再數一次，並且**真的觸發一支工作流**確認它還跑得動 +# +# 通過的定義（不通就 exit 1，不留模稜兩可）： +# 砍掉 KV 之後，列出來仍然是 9 支、recipe 仍在、工作流仍然跑得出結果。 +# +# ⚠️ 全程只碰本機（--local + --persist-to 到暫存目錄），**不碰任何線上實例**。 +# 腳本裡沒有任何 --remote、沒有任何真實帳號憑證。 +# +# 用法： bash scripts/verify-kv-retirement.sh +# 需要： node 22+、pnpm、可執行 npx wrangler / curl 的 shell + +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +WORK="$(mktemp -d)" +KBDB_PORT=8801 +CYPHER_PORT=8802 +TOKEN="e2e-local-token" +TENANT="leo-e2e" +WF_COUNT=9 + +KBDB="http://127.0.0.1:${KBDB_PORT}" +CYPHER="http://127.0.0.1:${CYPHER_PORT}" + +kbdb_pid=""; cypher_pid="" +cleanup() { + [ -n "$kbdb_pid" ] && kill "$kbdb_pid" 2>/dev/null || true + [ -n "$cypher_pid" ] && kill "$cypher_pid" 2>/dev/null || true + rm -rf "$WORK" +} +trap cleanup EXIT + +say() { printf '\n\033[1m== %s\033[0m\n' "$*"; } +fail() { printf '\n\033[31m❌ %s\033[0m\n' "$*"; exit 1; } + +wait_for() { # wait_for