diff --git a/MIGRATION.md b/MIGRATION.md new file mode 100644 index 0000000..2f29a6a --- /dev/null +++ b/MIGRATION.md @@ -0,0 +1,103 @@ +# MIGRATION — workflow 真相源 WEBHOOKS KV → KBDB API(entry_type=workflow) + +> 分支:`feat/workflow-store-kbdb` SDD:`system-dev/docs/3-specs/arcrun/kbdb-base/design.md` §8.3/§8.4 +> 總管裁定(2026-07-06):① upsert 走 KBDB base `PUT /entries`;② cron 掃 KBDB、不保留 KV cron-idx; +> ③ 一 workflow = 一筆 `entry_type=workflow`,`content=description`(保 embed),graph+config+cron_expr 進 `metadata_json`。 +> +> **本 PR 不 merge、不部署、不動 production 那 7 筆 KV workflow。** data 搬遷是下一階段(leo 寫入閘)。 + +--- + +## 1. 這個 PR 改了什麼 + +### KBDB(新增能力) +- `kbdb/src/actions/entry-crud.ts`:新增 `upsertEntry(db, input)` — by `(owner_id, page_name, entry_type)` + read-then-write(無 UNIQUE 約束、表不變鐵律;命中最舊一筆更新,重跑收斂)。 +- `kbdb/src/routes/entries.ts`:新增 `PUT /entries`(upsert,回 `{ success, entry, created }`;content 變動時 embed-on-write)。 + +### cypher-executor(workflow store + 六路 + 兩路切雙軌) +- **新檔** `cypher-executor/src/lib/workflow-store.ts`:`getWorkflow / listWorkflows / putWorkflow / + deleteWorkflow / listCronWorkflows`,全走 KBDB base API(`kbdbBase` = KBDB_BASE_URL + 選用 + KBDB_INTERNAL_TOKEN),owner_id=namespace 租戶隔離。禁直連 D1/SQL。 +- `routes/webhooks-named.ts`: + - register:寫 `putWorkflow`(KBDB 為主)+ 暫雙寫 KV;移除 `writeWorkflowSearchEntry`(上收進 putWorkflow, + 修掉每次 POST 堆重複 entry 的 bug);移除 KV cron-idx 維護。 + - trigger / list / delete:讀先 KBDB、miss fallback KV;list = KBDB ∪ KV(同名 KBDB 為準);delete 兩邊都刪。 + - `backfill-search-entries`:升級為 **KV→KBDB 完整遷移入口**(putWorkflow upsert,冪等可重跑)。 + - 移除 `migrate-cron-index` 端點(KV cron-idx 已退場)。 +- `scheduled.ts`:cron 掃描改 `listCronWorkflows`(掃 KBDB、記憶體濾 cron_expr);不再讀 KV cron-idx。 +- `lib/component-loader.ts`(trigger_workflow 子流程)、`routes/executions.ts`(擁有權檢查):切雙軌讀。 +- **刪除** `lib/cron-index.ts`(KV cron-idx 模型退場)。 + +### 綠燈 +- `kbdb` tsc 0 error、vitest 6/6 綠。 +- `cypher-executor` tsc 0 error、vitest 41/42 綠(1 個 `executor.test.ts > 不存在的零件` 失敗為 + **pre-existing**,main 上同樣失敗,與本 PR 無關——component-not-found 錯誤訊息措辭斷言,非 workflow 路徑)。 + +--- + +## 2. 部署後驗證計畫(對真端點 curl,禁 miniflare 假綠) + +> 前提:部署到 **leo21c**(auto 模式本輪不做;下述為部署後的驗收步驟)。 +> 端點:cypher `https://arcrun-cypher-executor.leo21c.workers.dev`、KBDB `https://arcrun-kbdb.leo21c.workers.dev`。 +> **全程只用測試 namespace(如 `ak_wfmigtest`),不碰現有 7 筆 production workflow。** + +設 `NS=ak_wfmigtest`、`CY=https://arcrun-cypher-executor.leo21c.workers.dev`、`KB=https://arcrun-kbdb.leo21c.workers.dev`。 + +1. **KBDB upsert 端點本身** + ``` + curl -XPUT $KB/entries -H 'content-type: application/json' \ + -d '{"entry_type":"workflow","owner_id":"'$NS'","page_name":"t1","content":"desc A","metadata_json":"{\"graph\":{\"id\":\"g\"}}"}' + # → { success:true, created:true } + # 再打一次(改 content:"desc B")→ { success:true, created:false }(同一 entry 更新,非新增) + curl "$KB/entries?entry_type=workflow&owner_id=$NS&page_name=t1" # → 恰 1 筆、content="desc B" + ``` +2. **register(部署)→ 恰一筆、redeploy 不重複** + ``` + curl -XPOST $CY/webhooks/named -H "X-Arcrun-API-Key: $NS" -H 'content-type: application/json' \ + -d '{"name":"wf_a","description":"測試工作流","graph":{"id":"wf_a","nodes":[]}}' + curl "$KB/entries?entry_type=workflow&owner_id=$NS&page_name=wf_a" # → 恰 1 筆 + # 同名再 POST 一次(改 description)→ 仍恰 1 筆、content 更新(驗 upsert 冪等,無重複 entry bug) + ``` +3. **list / trigger / delete 走 KBDB** + ``` + curl $CY/webhooks/named -H "X-Arcrun-API-Key: $NS" # → 含 wf_a,欄位齊 + curl -XPOST $CY/webhooks/named/wf_a/trigger -H "X-Arcrun-API-Key: $NS" -d '{}' # → 執行(讀到 KBDB graph) + curl -XDELETE $CY/webhooks/named/wf_a -H "X-Arcrun-API-Key: $NS" # → {deleted:true} + curl "$KB/entries?entry_type=workflow&owner_id=$NS&page_name=wf_a" # → 0 筆 + ``` +4. **cron 掃 KBDB**:部署一個首節點為 cron 的 workflow(cron_expr 每分鐘),等 1–2 分鐘看 `wrangler tail` + `[scheduled] scanned N KBDB cron workflows, k triggered` 有觸發。 +5. **雙軌 fallback(不寫 KBDB 的舊 KV workflow 仍可 trigger)**:手動塞一筆 `{NS}:wf:legacy` 進 WEBHOOKS KV + (只 KV、不進 KBDB),打 `/webhooks/named/legacy/trigger` → 應 fallback KV 成功執行;`GET /webhooks/named` + 應同時列出 KBDB 的與這筆 KV-only 的(union)。 + +**驗收標準**:上述每步 HTTP status + 回傳體符合預期;特別是「redeploy 同名恆為 1 筆」(upsert 生效) +與「cron 從 KBDB 觸發」。缺任一即不算綠,回報而非假綠(mindset §7)。 + +--- + +## 3. 剩餘 data 搬遷步驟(下一階段,leo 寫入閘,本輪別做) + +現有 7 筆 production workflow 仍只在 WEBHOOKS KV。雙軌讀讓它們仍可 trigger,但: +- **cron 掃描只看 KBDB** → 若這 7 筆裡有 cron workflow,未搬遷前它們的排程會**停止觸發**(⚠️ 高優先)。 +- semantic 搜尋也只看 KBDB entry。 + +**搬遷程序(冪等、可重跑)**: +1. 逐 namespace 呼叫 `POST /workflows/backfill-search-entries`(帶該租戶 `X-Arcrun-API-Key`)。 + 它 list KV 的 `{ns}:wf:*`、對每筆 `putWorkflow`(upsert)遷進 KBDB。回傳 `backfilled` / `needs_description`。 +2. 驗證:`GET $KB/entries?entry_type=workflow&owner_id={ns}` 每筆都有對應 entry(graph 在 metadata)。 +3. `needs_description` 清單交操盤 CC re-deploy 時據實補描述(不自動編造)。 +4. 全部驗穩後,才由 leo 決定**拆掉 register 的 KV 雙寫 + trigger/list/delete 的 KV fallback**(另一次閘)。 + 拆除前 KV 是安全網,不要急著拿掉。 + +> 誰有幾筆、哪些是 cron,需先盤點(`WEBHOOKS.list` 或 console)再排搬遷順序(cron 的先搬)。 + +--- + +## 4. 回滾 + +本 PR 不動 production 資料,且 register 仍雙寫 KV、讀有 KV fallback → 直接 revert 分支即可回到「KV 為真相源」。 +KBDB 端只**新增** `PUT /entries`(不改既有端點行為),revert cypher 側即停用;KBDB 側可留(無副作用)。 +唯一不可逆的取捨:本 PR 刪了 `cron-index.ts` 與 `migrate-cron-index` 端點(KV cron-idx 退場); +如需回滾 cron 到 KV,從 git 歷史取回即可。 diff --git a/cypher-executor/src/lib/component-loader.ts b/cypher-executor/src/lib/component-loader.ts index 5eec1a7..b59c1c8 100644 --- a/cypher-executor/src/lib/component-loader.ts +++ b/cypher-executor/src/lib/component-loader.ts @@ -20,6 +20,7 @@ import { isComponentHash, isRecipeHash } from './hash'; import { resolveRecipe, resolveAuthRecipe } from '../routes/recipes'; import type { AuthRecipeDefinition } from '../routes/recipes'; import type { Bindings, ComponentRunner, ServiceBinding } from '../types'; +import { getWorkflow } from './workflow-store'; /** * WASM HTTP runner:canonical_id → 對應獨立 Worker URL。 @@ -184,15 +185,16 @@ function makeTriggerWorkflowRunner(env: Bindings): ComponentRunner { if (!workflowName) return { success: false, error: 'trigger_workflow 缺 workflow_name' }; if (!apiKey) return { success: false, error: 'trigger_workflow 缺 api_key' }; - // 從 WEBHOOKS KV 撈目標 workflow 的 graph - const wfKey = `${apiKey}:wf:${workflowName}`; - const wfRaw = await env.WEBHOOKS.get(wfKey, 'text'); - if (!wfRaw) return { success: false, error: `找不到 workflow "${workflowName}" (key=${wfKey})` }; - - let record: { graph?: Record }; - try { record = JSON.parse(wfRaw); } - catch { return { success: false, error: `workflow "${workflowName}" KV 內容非 JSON` }; } - if (!record.graph) return { success: false, error: `workflow "${workflowName}" 缺 graph 欄位` }; + // 雙軌讀目標 workflow 的 graph:先 KBDB(真相源),miss 才 fallback WEBHOOKS KV(尚未搬遷的舊 workflow)。 + let record: { graph?: Record } | null = await getWorkflow(env, apiKey, workflowName); + if (!record) { + const wfKey = `${apiKey}:wf:${workflowName}`; + const wfRaw = await env.WEBHOOKS.get(wfKey, 'text'); + if (!wfRaw) return { success: false, error: `找不到 workflow "${workflowName}" (key=${wfKey})` }; + try { record = JSON.parse(wfRaw); } + catch { return { success: false, error: `workflow "${workflowName}" KV 內容非 JSON` }; } + } + if (!record?.graph) return { success: false, error: `workflow "${workflowName}" 缺 graph 欄位` }; // 動態 import 避循環依賴 const { executeWebhookGraph } = await import('../actions/webhook-handlers'); diff --git a/cypher-executor/src/lib/cron-index.ts b/cypher-executor/src/lib/cron-index.ts deleted file mode 100644 index 928df58..0000000 --- a/cypher-executor/src/lib/cron-index.ts +++ /dev/null @@ -1,70 +0,0 @@ -/** - * Cron index — 單一固定 key 模型(kbdb-base 8.P0 止血)。 - * - * 背景:原本每個 cron workflow 寫一筆 `cron-idx:{apiKey}:{name}`,scheduled() 每分鐘 - * `WEBHOOKS.list({prefix:'cron-idx:'})` 一次 = 1440 list/日,單獨就爆 CF KV 免費 list 上限(1000/日)。 - * - * 解法(SDD §8.2):所有 cron workflow 的 cron_expr 集中存進**單一固定 key** `cron-idx:_all`。 - * scheduled() 每分鐘只 `get` 一次(KV get 免費額度 100K/日,遠夠)→ list 次數歸零。 - * acr push(webhooks-named POST)/ delete 時對這個 key 做 read-modify-write 維護。 - * - * 結構:{ [ "{apiKey}:{name}" ]: cron_expr } - * key 用 `{apiKey}:{name}` 維持多租戶隔離(scheduled 觸發時拆回 apiKey/name 去讀完整 record)。 - */ - -// KVNamespace 用全域 ambient 型別(與 types.ts 一致,不從 @cloudflare/workers-types import -// 以免產生第二個不相容的 KVNamespace 型別)。 - -/** 單一固定索引 key — 全租戶共用一筆,scheduled() 只 get 這個 */ -export const CRON_INDEX_KEY = 'cron-idx:_all'; - -/** 索引內容:entryKey("{apiKey}:{name}")→ cron_expr */ -export type CronIndex = Record; - -/** 組出索引 entry 的 key(apiKey + name),含 ':' 也安全:split 時 name 用 slice 還原 */ -export function cronEntryKey(apiKey: string, name: string): string { - return `${apiKey}:${name}`; -} - -/** 從 entryKey 拆回 { apiKey, name }(name 可能含 ':',取第一個 ':' 後全部為 name) */ -export function parseCronEntryKey(entryKey: string): { apiKey: string; name: string } | null { - const idx = entryKey.indexOf(':'); - if (idx <= 0) return null; - return { apiKey: entryKey.slice(0, idx), name: entryKey.slice(idx + 1) }; -} - -/** 讀整個 cron index(單次 get,不 list) */ -export async function readCronIndex(kv: KVNamespace): Promise { - const raw = await kv.get(CRON_INDEX_KEY, 'text'); - if (!raw) return {}; - try { - const parsed = JSON.parse(raw); - return parsed && typeof parsed === 'object' ? (parsed as CronIndex) : {}; - } catch { - return {}; - } -} - -/** - * upsert / 移除單筆 cron entry(read-modify-write 單一 key)。 - * @param cronExpr - 有值=upsert;null/undefined=移除(push 改掉 cron 後清乾淨) - */ -export async function updateCronIndexEntry( - kv: KVNamespace, - apiKey: string, - name: string, - cronExpr: string | null | undefined, -): Promise { - const index = await readCronIndex(kv); - const entryKey = cronEntryKey(apiKey, name); - - if (cronExpr) { - if (index[entryKey] === cronExpr) return; // 無變化,不浪費一次 put - index[entryKey] = cronExpr; - } else { - if (!(entryKey in index)) return; // 本來就沒有,不浪費一次 put - delete index[entryKey]; - } - - await kv.put(CRON_INDEX_KEY, JSON.stringify(index)); -} diff --git a/cypher-executor/src/lib/workflow-store.ts b/cypher-executor/src/lib/workflow-store.ts new file mode 100644 index 0000000..e95ad78 --- /dev/null +++ b/cypher-executor/src/lib/workflow-store.ts @@ -0,0 +1,203 @@ +/** + * Workflow store — workflow 真相源從 WEBHOOKS KV 遷到 KBDB API(entry_type=workflow entry)。 + * + * SDD: kbdb-base design.md §8.3/§8.4(workflow record 從 WEBHOOKS KV 遷 D1)。 + * 總管裁定(2026-07-06): + * ① upsert:KBDB base 加 PUT /entries(by owner_id+page_name+entry_type);本檔 putWorkflow 走它。 + * ② cron 掃描:listCronWorkflows 撈全部 entry_type=workflow、記憶體濾 metadata.cron_expr; + * **不保留 KV cron-idx**(目標退 KV)。TODO(規模):workflow 量大再加 KBDB「只回有 cron 的」filter 端點。 + * ③ 形態:一 workflow = 一筆 entry_type=workflow。**content=description**(保 embed 語意搜尋, + * 對齊 workflow-discovery),完整 graph+config+cron_expr 放 **metadata_json**(TEXT,base 視為不透明)。 + * + * 身份/連法:走 KBDB base API(kbdb-proxy 的 kbdbBase:KBDB_BASE_URL + 選用 KBDB_INTERNAL_TOKEN), + * owner_id = namespace(api_key)=租戶隔離。不新增 service binding(rule 02 §3.1)。禁直連 D1 / SQL(D6)。 + * + * 雙軌過渡(安全可回滾):呼叫端「讀先 KBDB、miss fallback KV;寫 KBDB 為主 + 暫雙寫 KV」。 + * 本檔只負責 KBDB 這一側;KV 雙寫/fallback 由呼叫端(webhooks-named / scheduled / …)銜接。 + */ + +import type { Bindings } from '../types'; +import { kbdbBase } from '../routes/kbdb-proxy'; + +/** workflow 記錄(與舊 KV NamedWorkflowRecord 對齊,供呼叫端無痛替換)。 */ +export type WorkflowRecord = { + name: string; + graph: Record; + config?: Record; + description: string; + created_at: string; + cron_expr?: string; +}; + +/** KBDB entry 形狀(本檔只用到的欄位)。 */ +type KbdbEntry = { + id: string; + content: string | null; + page_name: string | null; + owner_id: string | null; + metadata_json: string | null; +}; + +/** metadata_json 裡承載 workflow 本體的形狀(③:graph/config/cron_expr 都在這)。 */ +type WorkflowMeta = { + embed?: boolean; + workflow_name?: string; + workflow_id?: string; + graph?: Record; + config?: Record; + cron_expr?: string; + created_at?: string; +}; + +function parseMeta(raw: string | null): WorkflowMeta | null { + if (!raw) return null; + try { + const m = JSON.parse(raw); + return m && typeof m === 'object' ? (m as WorkflowMeta) : null; + } catch { + return null; + } +} + +/** + * entry → WorkflowRecord。只認「帶完整 graph 的新形態 entry」; + * 舊的 discovery-only entry(content=description、metadata 無 graph)→ 回 null, + * 讓呼叫端判為 miss 去 fallback KV(雙軌過渡正確行為)。 + */ +function entryToRecord(entry: KbdbEntry): WorkflowRecord | null { + const meta = parseMeta(entry.metadata_json); + const graph = meta?.graph; + if (!graph || typeof graph !== 'object') return null; + return { + name: entry.page_name ?? meta?.workflow_name ?? '', + graph, + config: meta?.config, + description: entry.content ?? '', + created_at: meta?.created_at ?? '', + cron_expr: meta?.cron_expr, + }; +} + +/** WorkflowRecord → KBDB upsert body(③ 的欄位分配)。 */ +function recordToEntryBody(owner: string, name: string, record: WorkflowRecord) { + const graphId = (record.graph as { id?: string })?.id ?? name; + return { + entry_type: 'workflow', + owner_id: owner, // 租戶隔離(與 kbdb-proxy 同身份模型) + page_name: name, // 唯一鍵(owner+type 內) + content: record.description, // 被 embed / LIKE 命中的主體(③:保語意搜尋) + metadata_json: JSON.stringify({ + embed: true, // #7 精耕開關:標 true 才進 Vectorize(embed 讀 content=description) + workflow_name: name, + workflow_id: graphId, + graph: record.graph, // ③:完整 graph 放 metadata(base 不透明) + config: record.config, + cron_expr: record.cron_expr, + created_at: record.created_at, + } satisfies WorkflowMeta), + }; +} + +/** 取單筆 workflow 的 raw entry(含 id,供 delete 用)。miss/error 回 null。 */ +async function getWorkflowEntry(env: Bindings, owner: string, name: string): Promise { + try { + const { base, headers } = kbdbBase(env); + const params = new URLSearchParams({ + entry_type: 'workflow', + owner_id: owner, + page_name: name, + limit: '1', + }); + const res = await fetch(`${base}/entries?${params.toString()}`, { headers }); + if (!res.ok) return null; + const data = await res.json() as { entries?: KbdbEntry[] }; + return data.entries?.[0] ?? null; + } catch { + return null; + } +} + +/** 讀單筆 workflow(KBDB)。miss/error/舊形態 → null(呼叫端據此 fallback KV)。 */ +export async function getWorkflow(env: Bindings, owner: string, name: string): Promise { + const entry = await getWorkflowEntry(env, owner, name); + return entry ? entryToRecord(entry) : null; +} + +/** 列本租戶所有 workflow(KBDB)。error 回 [](呼叫端仍可 union KV)。 */ +export async function listWorkflows(env: Bindings, owner: string): Promise { + try { + const { base, headers } = kbdbBase(env); + const params = new URLSearchParams({ + entry_type: 'workflow', + owner_id: owner, + limit: '1000', + }); + const res = await fetch(`${base}/entries?${params.toString()}`, { headers }); + if (!res.ok) return []; + const data = await res.json() as { entries?: KbdbEntry[] }; + return (data.entries ?? []) + .map(entryToRecord) + .filter((r): r is WorkflowRecord => r !== null); + } catch { + return []; + } +} + +/** upsert 一筆 workflow(KBDB PUT /entries;redeploy 同名不堆重複)。失敗會 throw(呼叫端決定是否致命)。 */ +export async function putWorkflow(env: Bindings, owner: string, name: string, record: WorkflowRecord): Promise { + const { base, headers } = kbdbBase(env); + const res = await fetch(`${base}/entries`, { + method: 'PUT', + headers, + body: JSON.stringify(recordToEntryBody(owner, name, record)), + }); + if (!res.ok) { + const text = await res.text().catch(() => ''); + throw new Error(`putWorkflow KBDB PUT /entries ${res.status}: ${text.slice(0, 200)}`); + } +} + +/** 刪一筆 workflow(KBDB)。回是否真的刪到(供呼叫端與 KV 結果合併判 404)。 */ +export async function deleteWorkflow(env: Bindings, owner: string, name: string): Promise { + const entry = await getWorkflowEntry(env, owner, name); + if (!entry) return false; + try { + const { base, headers } = kbdbBase(env); + const res = await fetch(`${base}/entries/${encodeURIComponent(entry.id)}`, { method: 'DELETE', headers }); + return res.ok; + } catch { + return false; + } +} + +/** cron workflow(跨租戶)— scheduled() 每分鐘用。撈全部 workflow entry、記憶體濾出有 cron_expr 的。 */ +export type CronWorkflow = { owner: string; name: string; cron_expr: string; graph: Record }; + +/** + * 掃出所有帶 cron_expr 的 workflow(跨租戶,不帶 owner_id filter)。 + * 成本(總管裁 (a)):每分鐘 1 次 KBDB /entries D1 query(1440/日 << D1 免費 5M 讀/日,§8.1), + * graph 已在 metadata 一併帶回 → scheduled 不用第二次 fetch。 + * TODO(規模):workflow 量很大時,base 加「只回 metadata.cron_expr 非空」的 filter 端點, + * 避免每分鐘撈回全部 workflow entry(今天量小無感;此為已知擴充點,非本輪範圍)。 + */ +export async function listCronWorkflows(env: Bindings): Promise { + try { + const { base, headers } = kbdbBase(env); + const params = new URLSearchParams({ entry_type: 'workflow', limit: '1000' }); + const res = await fetch(`${base}/entries?${params.toString()}`, { headers }); + if (!res.ok) return []; + const data = await res.json() as { entries?: KbdbEntry[] }; + const out: CronWorkflow[] = []; + for (const e of data.entries ?? []) { + const meta = parseMeta(e.metadata_json); + const cron = meta?.cron_expr; + const graph = meta?.graph; + if (!cron || !graph || typeof graph !== 'object') continue; + if (!e.owner_id || !e.page_name) continue; + out.push({ owner: e.owner_id, name: e.page_name, cron_expr: cron, graph }); + } + return out; + } catch { + return []; + } +} diff --git a/cypher-executor/src/routes/executions.ts b/cypher-executor/src/routes/executions.ts index e357585..942d396 100644 --- a/cypher-executor/src/routes/executions.ts +++ b/cypher-executor/src/routes/executions.ts @@ -13,6 +13,7 @@ import { Hono } from 'hono'; import type { Bindings } from '../types'; import { listPausedRunsByApiKey } from '../lib/paused-runs'; +import { getWorkflow } from '../lib/workflow-store'; export const executionsRouter = new Hono<{ Bindings: Bindings }>(); @@ -153,9 +154,10 @@ executionsRouter.get('/workflows/:name/executions', async (c) => { const limitParam = c.req.query('limit'); const limit = Math.min(Math.max(parseInt(limitParam || '10', 10), 1), 100); - // 確認 workflow 是該 api_key 的(防偷看他人) - const wfRaw = await c.env.WEBHOOKS.get(`${apiKey}:wf:${name}`, 'text'); - if (!wfRaw) { + // 確認 workflow 是該 api_key 的(防偷看他人)。雙軌:先 KBDB(真相源),miss 才 fallback KV。 + const owned = (await getWorkflow(c.env, apiKey, name)) !== null + || (await c.env.WEBHOOKS.get(`${apiKey}:wf:${name}`, 'text')) !== null; + if (!owned) { return c.json({ ok: false, error_code: 'not_found', diff --git a/cypher-executor/src/routes/webhooks-named.ts b/cypher-executor/src/routes/webhooks-named.ts index 3537ddc..eb4cdd9 100644 --- a/cypher-executor/src/routes/webhooks-named.ts +++ b/cypher-executor/src/routes/webhooks-named.ts @@ -28,8 +28,8 @@ import { executeWebhookGraph } from '../actions/webhook-handlers'; import { writeExecutionVerdict } from '../actions/execution-logger'; import type { GraphNode } from '../types'; import { extractCronExpr } from '../lib/cron-match'; -import { updateCronIndexEntry, CRON_INDEX_KEY } from '../lib/cron-index'; import { recordTelemetry } from '../lib/telemetry'; +import { getWorkflow, listWorkflows, putWorkflow, deleteWorkflow } from '../lib/workflow-store'; export const webhooksNamedRouter = new Hono<{ Bindings: Bindings }>(); @@ -50,43 +50,10 @@ function kvKey(apiKey: string, name: string): string { return `${apiKey}:wf:${name}`; } -/** - * workflow-discovery R2/Phase 2.1:部署時雙寫一個 embeddable entry 到 KBDB,讓 workflow 可被語意搜尋。 - * - * 雙寫(design 方案 C):WEBHOOKS KV record 照舊(list/get/trigger 不動),另寫 entry_type=workflow 的 - * entry 供 search。owner_id = api_key(租戶隔離,與 kbdb-proxy 同身份模型)。 - * content = description(被 embed 的主體);metadata.embed:true → 命中 #7 精耕條件進 Vectorize(模組開時)。 - * - * 非阻塞 + 失敗不致命(waitUntil + catch):search 可發現性是加值,不該擋部署成功(對齊 #7 embedOnWrite 慣例)。 - * KBDB 連法沿用既有慣例(KBDB_BASE_URL fetch + 選用 token),不新增 service binding(rule 02 §3.1)。 - */ -async function writeWorkflowSearchEntry( - env: Bindings, - apiKey: string, - name: string, - description: string, - workflowId?: string, -): Promise { - 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}`; - await fetch(`${base}/entries`, { - method: 'POST', - headers, - body: JSON.stringify({ - entry_type: 'workflow', - owner_id: apiKey, // 租戶隔離(與 kbdb-proxy 同身份) - page_name: name, - content: description, // 被 embed / LIKE 命中的主體 - // KBDB createEntry 吃 metadata_json(TEXT),embed.ts isEmbeddable 讀 metadata_json.embed === true。 - metadata_json: JSON.stringify({ - embed: true, // #7 精耕開關:標 true 才進 Vectorize - workflow_name: name, - workflow_id: workflowId ?? name, - }), - }), - }); -} +// 註(workflow-store 遷移,2026-07-06):原 writeWorkflowSearchEntry(雙寫一筆 content=description 的 +// search-entry)已被 lib/workflow-store.ts 的 putWorkflow 取代並上收——putWorkflow 寫的就是那筆 +// entry_type=workflow(content=description 保 embed + 完整 graph 進 metadata),且走 KBDB PUT upsert, +// 故 redeploy 同名不再堆重複 entry(修掉舊 POST 每次新增的重複 bug)。search 可發現性沿用不變。 // POST /webhooks/named — 部署(acr push 呼叫) webhooksNamedRouter.post('/webhooks/named', async (c) => { @@ -136,18 +103,20 @@ webhooksNamedRouter.post('/webhooks/named', async (c) => { }; const start = Date.now(); + + // 寫 KBDB 為主(真相源,upsert:redeploy 同名不堆重複)。cron_expr 存在 entry metadata, + // scheduled() 改掃 KBDB(不再維護 KV cron-idx;總管裁 ②「不保留 KV cron-idx」)。 + // 失敗不致命:仍有下面 KV 雙寫當安全網 → 讀路徑 fallback KV 仍可觸發(雙軌過渡,可回滾)。 + try { + await putWorkflow(c.env, apiKey, name, record); + } catch (e) { + console.warn('[register] KBDB putWorkflow 失敗,暫靠 KV 雙寫(雙軌過渡)', name, e instanceof Error ? e.message : String(e)); + } + + // 暫時雙寫 WEBHOOKS KV(保險 + 可回滾):讀路徑 miss KBDB 時 fallback 到這裡。 + // data 搬遷完成 + 驗穩後,另一次 leo 寫入閘再拆 KV 雙寫(見 MIGRATION.md)。 await c.env.WEBHOOKS.put(kvKey(apiKey, name), JSON.stringify(record)); - // 維護單一 cron index key(8.P0):有 cron_expr 就 upsert / 沒有就移除 - // (避免 push 改 yaml 拿掉 cron 後殘留)。scheduled() 每分鐘只 get 這一個 key。 - await updateCronIndexEntry(c.env.WEBHOOKS, apiKey, name, cronExpr); - - // workflow-discovery Phase 2.1:雙寫 embeddable search-entry(讓此 workflow 可被語意搜尋)。 - // 非阻塞(waitUntil)+ 失敗不致命(catch):可發現性是加值,不擋部署成功(對齊 #7 embedOnWrite 慣例)。 - c.executionCtx.waitUntil( - writeWorkflowSearchEntry(c.env, apiKey, name, record.description).catch(() => {}), - ); - // Implicit telemetry (LI M1.2) recordTelemetry(c.env, apiKey, { event_type: 'deploy_success', @@ -190,9 +159,11 @@ webhooksNamedRouter.get('/workflows/search', async (c) => { return new Response(res.body, { status: res.status, headers: { 'Content-Type': 'application/json' } }); }); -// POST /workflows/backfill-search-entries — workflow-discovery R3:把既有 workflow 補成可搜的 search-entry。 -// 有 description 的 → 補寫 entry(讓它們可被 u6u_search_workflows 搜到);無 description 的 → 列出待 re-deploy。 -// 誠實:不自動編造 description(無 desc 的只列出、不假裝)。flag 安全:人/AI 主動呼叫一次,非 cron/輪詢。 +// POST /workflows/backfill-search-entries — 既有 KV workflow 一次性遷進 KBDB(entry_type=workflow)。 +// workflow-store 遷移後上收:從純寫 search-entry(只 description)升級為「遷完整 record(graph+config+ +// cron_expr 進 metadata、description 進 content 保 embed)」,用 putWorkflow upsert(冪等、可重跑)。 +// 這就是 MIGRATION.md 的 KV→KBDB 資料搬遷入口(人/AI 主動呼叫一次,非 cron/輪詢;flag 安全)。 +// 誠實:無 description 的仍遷(graph 才是本體、trigger 要用),但另列出來提醒它們搜不到、請補描述。 webhooksNamedRouter.post('/workflows/backfill-search-entries', async (c) => { const apiKey = c.req.header('X-Arcrun-API-Key'); if (!apiKey) return c.json({ error: '缺少 X-Arcrun-API-Key header' }, 401); @@ -208,15 +179,20 @@ webhooksNamedRouter.post('/workflows/backfill-search-entries', async (c) => { const raw = await c.env.WEBHOOKS.get(k.name, 'text'); if (!raw) continue; const rec = JSON.parse(raw) as NamedWorkflowRecord; - const desc = rec.description?.trim(); - if (!desc) { - // 不自動編造:無 description 的列出來,請操盤 CC re-deploy 時據實補(誠實,mindset §7)。 - needsDescription.push(name); - continue; - } + const desc = rec.description?.trim() ?? ''; try { - await writeWorkflowSearchEntry(c.env, apiKey, name, desc); + // 遷完整 record(graph 是本體,即使無 description 也要能被 trigger 讀到)。 + await putWorkflow(c.env, apiKey, name, { + name, + graph: rec.graph, + config: rec.config, + description: desc, + created_at: rec.created_at ?? new Date().toISOString(), + cron_expr: rec.cron_expr, + }); backfilled.push(name); + // 不自動編造:無 description 的仍遷,但列出來(搜不到),請操盤 CC re-deploy 時據實補(誠實,mindset §7)。 + if (!desc) needsDescription.push(name); } catch (e) { errors.push(`${name}: ${e instanceof Error ? e.message : String(e)}`); } @@ -229,39 +205,11 @@ webhooksNamedRouter.post('/workflows/backfill-search-entries', async (c) => { needs_description_count: needsDescription.length, errors, hint: needsDescription.length > 0 - ? `${needsDescription.length} 個工作流缺 description 無法被搜尋。請操盤的 AI re-deploy 它們時據實補一句「能做什麼」(不自動編造)。` + ? `${needsDescription.length} 個工作流缺 description 無法被搜尋(已遷 KBDB、trigger 可用)。請操盤的 AI re-deploy 它們時據實補一句「能做什麼」(不自動編造)。` : undefined, }); }); -// POST /webhooks/named/migrate-cron-index — 一次性 migration(8.P0):把舊的 per-key -// cron-idx:{apiKey}:{name} 折進單一 cron-idx:_all(這裡才 list 一次,非每分鐘 tick)。 -// 增量寫、不刪舊 key(重跑安全、冪等)。部署 8.P0 後跑一次,讓既有 cron workflow 不漏掉。 -// 必須在 /:name/trigger 之前註冊,否則 :name 會攔截 "migrate-cron-index"。 -webhooksNamedRouter.post('/webhooks/named/migrate-cron-index', async (c) => { - const list = await c.env.WEBHOOKS.list({ prefix: 'cron-idx:' }); - let migrated = 0, skipped = 0; - const errors: string[] = []; - for (const k of list.keys) { - if (k.name === CRON_INDEX_KEY) { skipped++; continue; } // 跳過新的集中 key 自己 - const parts = k.name.split(':'); // cron-idx:{apiKey}:{name} - if (parts.length < 3) { skipped++; continue; } - const apiKey = parts[1]; - const name = parts.slice(2).join(':'); - try { - const raw = await c.env.WEBHOOKS.get(k.name, 'text'); - if (!raw) { skipped++; continue; } - const idx = JSON.parse(raw) as { cron_expr?: string }; - if (!idx.cron_expr) { skipped++; continue; } - await updateCronIndexEntry(c.env.WEBHOOKS, apiKey, name, idx.cron_expr); - migrated++; - } catch (e) { - errors.push(`${k.name}: ${e instanceof Error ? e.message : String(e)}`); - } - } - return c.json({ success: errors.length === 0, migrated, skipped, errors }); -}); - // POST /webhooks/named/:name/trigger — 觸發執行(api_key 走 header;標準/向後相容) webhooksNamedRouter.post('/webhooks/named/:name/trigger', async (c) => { const apiKey = c.req.header('X-Arcrun-API-Key'); @@ -285,16 +233,18 @@ async function triggerNamed( apiKey: string, name: string, ) { - const raw = await c.env.WEBHOOKS.get(kvKey(apiKey, name), 'text'); - if (!raw) { - return c.json({ error: `找不到 workflow "${name}",請先執行 acr push` }, 404); - } - - let record: NamedWorkflowRecord; - try { - record = JSON.parse(raw) as NamedWorkflowRecord; - } catch { - return c.json({ error: 'workflow 定義損毀' }, 500); + // 雙軌讀:先 KBDB(真相源),miss 才 fallback WEBHOOKS KV(相容尚未搬遷的舊 workflow)。 + let record: NamedWorkflowRecord | null = await getWorkflow(c.env, apiKey, name); + if (!record) { + const raw = await c.env.WEBHOOKS.get(kvKey(apiKey, name), 'text'); + if (!raw) { + return c.json({ error: `找不到 workflow "${name}",請先執行 acr push` }, 404); + } + try { + record = JSON.parse(raw) as NamedWorkflowRecord; + } catch { + return c.json({ error: 'workflow 定義損毀' }, 500); + } } let triggerContext: Record = {}; @@ -348,27 +298,41 @@ webhooksNamedRouter.get('/webhooks/named', async (c) => { return c.json({ error: '缺少 X-Arcrun-API-Key header' }, 401); } + const baseUrl = new URL(c.req.url).origin; + + // 雙軌列舉:KBDB(真相源)優先,再 union 尚未搬遷的 KV-only workflow(同名以 KBDB 為準)。 + const kbdbList = await listWorkflows(c.env, apiKey); + const byName = new Map(); + for (const w of kbdbList) { + byName.set(w.name, { + name: w.name, + description: w.description ?? '', + created_at: w.created_at ?? '', + cron_expr: w.cron_expr, + webhook_url: `${baseUrl}/webhooks/named/${w.name}/trigger`, + }); + } + + // fallback:補上只在 KV 的舊 workflow(KBDB 已有的不覆蓋)。 const prefix = `${apiKey}:wf:`; const list = await c.env.WEBHOOKS.list({ prefix }); - - // workflow-discovery 方向①:list 回完整欄位(description/created_at),讓 MCP u6u_list_workflows - // 改讀本端點時欄位齊(取代舊的讀 workflow_metadata record)。需 get 每個 record 取 description。 - const baseUrl = new URL(c.req.url).origin; - const result = await Promise.all( + await Promise.all( list.keys.map(async (k) => { const name = k.name.slice(prefix.length); + if (byName.has(name)) return; // KBDB 為準 const raw = await c.env.WEBHOOKS.get(k.name, 'text'); const rec = raw ? (JSON.parse(raw) as NamedWorkflowRecord) : null; - return { + byName.set(name, { name, description: rec?.description ?? '', created_at: rec?.created_at ?? '', cron_expr: rec?.cron_expr, webhook_url: `${baseUrl}/webhooks/named/${name}/trigger`, - }; + }); }), ); + const result = [...byName.values()]; return c.json({ workflows: result, total: result.length }); }); @@ -380,12 +344,14 @@ webhooksNamedRouter.delete('/webhooks/named/:name', async (c) => { } const name = c.req.param('name'); - const existing = await c.env.WEBHOOKS.get(kvKey(apiKey, name), 'text'); - if (!existing) { + + // 雙軌刪:KBDB(真相源)+ KV(雙寫期間的鏡像)。任一有刪到即算成功;兩邊都沒有才 404。 + const kbdbDeleted = await deleteWorkflow(c.env, apiKey, name); + const kvExisting = await c.env.WEBHOOKS.get(kvKey(apiKey, name), 'text'); + if (kvExisting) await c.env.WEBHOOKS.delete(kvKey(apiKey, name)); + + if (!kbdbDeleted && !kvExisting) { return c.json({ error: `找不到 workflow "${name}"` }, 404); } - - await c.env.WEBHOOKS.delete(kvKey(apiKey, name)); - await updateCronIndexEntry(c.env.WEBHOOKS, apiKey, name, null); return c.json({ deleted: true, name }); }); diff --git a/cypher-executor/src/scheduled.ts b/cypher-executor/src/scheduled.ts index ffc5c2b..0b6c113 100644 --- a/cypher-executor/src/scheduled.ts +++ b/cypher-executor/src/scheduled.ts @@ -1,30 +1,29 @@ /** * scheduled() handler — 對應 wrangler.toml [triggers].crons 觸發。 * - * 流程: - * 1. 單次 get cron index(cron-idx:_all,集中存所有 cron workflow 的 cron_expr) - * 2. 在記憶體比對每筆 cron_expr 跟 event.scheduledTime(UTC 分鐘精度) - * 3. 匹配才去讀完整 workflow record({apiKey}:wf:{name}) - * 4. 匹配 → executeWebhookGraph 跑(waitUntil 背景,不擋) + * 流程(workflow-store 遷移後,總管裁 ②): + * 1. 單次掃 KBDB:listCronWorkflows(撈全部 entry_type=workflow、記憶體濾 metadata.cron_expr)。 + * graph 已在 metadata 一併帶回 → 不用第二次讀。 + * 2. 記憶體比對每筆 cron_expr 跟 event.scheduledTime(UTC 分鐘精度)。 + * 3. 匹配 → executeWebhookGraph 跑(waitUntil 背景,不擋)。 * - * 8.P0 止血(SDD §8.2):原本每分鐘 WEBHOOKS.list('cron-idx:') = 1440 list/日 爆 KV 上限, - * 改成單一固定 key 只 get 一次 → list 歸零。 + * 為何不再讀 KV cron-idx(cron-idx:_all):workflow 真相源已遷 KBDB(§8.3),cron_expr 住在 entry + * metadata。總管裁 ②「不保留 KV cron-idx」(目標退 KV,不回頭掛 KV 快取)。成本:每分鐘 1 次 + * KBDB /entries D1 query = 1440/日 << D1 免費 5M 讀/日(§8.1),遠夠。 + * TODO(規模):workflow 量很大時 base 加「只回有 cron 的」filter 端點,見 workflow-store.listCronWorkflows。 * - * SDD: arcrun.md 三-A P1 #3 / kbdb-base §8.2 + * ⚠️ 過渡注意:尚未搬遷、只在 KV 的舊 cron workflow 不會被掃到(本掃描只看 KBDB)。部署前/時須先跑 + * 一次 KV→KBDB 搬遷(見 MIGRATION.md),否則既有 cron workflow 會停止觸發。 + * + * SDD: arcrun.md 三-A P1 #3 / kbdb-base §8.3 */ import type { ExecutionContext, ScheduledController } from '@cloudflare/workers-types'; import type { Bindings } from './types'; import { cronMatch } from './lib/cron-match'; -import { readCronIndex, parseCronEntryKey } from './lib/cron-index'; +import { listCronWorkflows } from './lib/workflow-store'; import { executeWebhookGraph } from './actions/webhook-handlers'; -type StoredWorkflowRecord = { - graph: Record; - cron_expr?: string; - // 其他欄位(id, name, created_at 等)忽略 -}; - export async function handleScheduled( controller: ScheduledController, env: Bindings, @@ -33,44 +32,29 @@ export async function handleScheduled( const now = new Date(controller.scheduledTime); console.log('[scheduled] tick', now.toISOString(), 'controller.cron=', controller.cron); - // 8.P0:單次 get 集中索引(取代每分鐘 list),主 workflow record 仍在 {apiKey}:wf:{name} - const index = await readCronIndex(env.WEBHOOKS); - const entries = Object.entries(index); + // 單次掃 KBDB 撈所有 cron workflow(graph 一併帶回,不用第二次讀)。 + const crons = await listCronWorkflows(env); let triggered = 0; - for (const [entryKey, cronExpr] of entries) { - const parsed = parseCronEntryKey(entryKey); - if (!parsed) continue; - const { apiKey, name } = parsed; - - if (!cronExpr) continue; - if (!cronMatch(cronExpr, now)) continue; - - // 匹配才去讀完整 workflow record - const wfKey = `${apiKey}:wf:${name}`; - const wfRaw = await env.WEBHOOKS.get(wfKey, 'text'); - if (!wfRaw) { - console.warn('[scheduled] cron-idx 對應 workflow 不存在', wfKey); - continue; - } - let record: StoredWorkflowRecord; - try { record = JSON.parse(wfRaw) as StoredWorkflowRecord; } catch { continue; } + for (const wf of crons) { + if (!wf.cron_expr) continue; + if (!cronMatch(wf.cron_expr, now)) continue; triggered++; - console.log('[scheduled] trigger', name, 'apiKey=', apiKey.slice(0, 12) + '...', 'cron=', cronExpr); + console.log('[scheduled] trigger', wf.name, 'apiKey=', wf.owner.slice(0, 12) + '...', 'cron=', wf.cron_expr); // 把 apiKey 也放進 triggerContext,讓 workflow 內節點能用 {{api_key}}(跟 webhook trigger 慣例一致) const triggerContext = { - api_key: apiKey, + api_key: wf.owner, _triggered_by: 'cron' as const, _scheduled_at: now.toISOString(), }; ctx.waitUntil( - executeWebhookGraph(env, record.graph, triggerContext, name, apiKey) + executeWebhookGraph(env, wf.graph, triggerContext, wf.name, wf.owner) .then( - (r) => console.log('[scheduled] done', name, r.success, r.duration_ms + 'ms'), - (e) => console.error('[scheduled] fail', name, e), + (r) => console.log('[scheduled] done', wf.name, r.success, r.duration_ms + 'ms'), + (e) => console.error('[scheduled] fail', wf.name, e), ), ); } - console.log(`[scheduled] scanned ${entries.length} cron-idx entries, ${triggered} triggered`); + console.log(`[scheduled] scanned ${crons.length} KBDB cron workflows, ${triggered} triggered`); } diff --git a/kbdb/src/actions/entry-crud.ts b/kbdb/src/actions/entry-crud.ts index 192826e..b05a103 100644 --- a/kbdb/src/actions/entry-crud.ts +++ b/kbdb/src/actions/entry-crud.ts @@ -123,6 +123,47 @@ export async function deleteEntry(db: D1Database, id: string): Promise { await db.prepare('DELETE FROM entries WHERE id = ?').bind(id).run(); } +// Upsert-by-(owner_id, page_name, entry_type) — capability lives in API (thin-shell §1 正例: +// 「API 提供 upsert 端點(內部 GET 找→有則 update 無則 insert)」)。base 沒有 UNIQUE 約束 +// (表不變鐵律 + 既有可能有重複 page_name),故用 read-then-write,不靠 ON CONFLICT。 +// 用途源: kbdb-base SDD §8.3 workflow 遷 D1 —— page_name 當唯一鍵(owner+type 內), +// register/redeploy 同名不再堆重複 entry(修掉 writeWorkflowSearchEntry 每次 POST 的重複 bug)。 +// key tuple 三者皆為 idempotency key;命中最舊一筆(created_at ASC)更新,讓重跑收斂到同一 canonical。 +export async function upsertEntry( + db: D1Database, + input: CreateEntryInput, +): Promise<{ entry: Entry; created: boolean }> { + if (!input.entry_type || !input.page_name) { + throw new Error('upsertEntry: entry_type 與 page_name 為 upsert key,必填'); + } + const owner = input.owner_id ?? null; + const existing = owner === null + ? await db + .prepare('SELECT * FROM entries WHERE entry_type = ? AND page_name = ? AND owner_id IS NULL ORDER BY created_at ASC LIMIT 1') + .bind(input.entry_type, input.page_name) + .first() + : await db + .prepare('SELECT * FROM entries WHERE entry_type = ? AND page_name = ? AND owner_id = ? ORDER BY created_at ASC LIMIT 1') + .bind(input.entry_type, input.page_name, owner) + .first(); + + if (existing) { + const entry = await updateEntry(db, existing.id, { + content: input.content ?? null, + parent_id: input.parent_id ?? null, + refs_json: input.refs_json, + tags_json: input.tags_json, + task_status: input.task_status ?? null, + confidence: input.confidence ?? null, + metadata_json: input.metadata_json ?? null, + }); + if (!entry) throw new Error('upsertEntry: update 後找不到 row'); + return { entry, created: false }; + } + const entry = await createEntry(db, input); + return { entry, created: true }; +} + // D1 LIKE keyword search (base; semantic search is the optional embed module). // entry_type: optional base filter (generic — caller passes any type, base stays type-agnostic). export async function searchEntries( diff --git a/kbdb/src/routes/entries.ts b/kbdb/src/routes/entries.ts index db139cb..abde084 100644 --- a/kbdb/src/routes/entries.ts +++ b/kbdb/src/routes/entries.ts @@ -8,6 +8,7 @@ import { updateEntry, deleteEntry, searchEntries, + upsertEntry, } from '../actions/entry-crud'; import { embedEnabled, embedOnWrite, semanticSearch } from '../embed'; @@ -23,6 +24,21 @@ entryRoutes.post('/', async (c) => { return c.json({ success: true, entry }); }); +// PUT /entries — upsert by (owner_id, page_name, entry_type)(能力長在 API,thin-shell §1 正例)。 +// 有則 update、無則 insert,回 created 旗標。用途源: kbdb-base SDD §8.3 workflow 遷 D1 +// —— cypher workflow-store 靠這個「redeploy 同名不堆重複 entry」。 +// 三者為 upsert key(缺 page_name/entry_type → 400);content/metadata_json 等隨之覆寫。 +entryRoutes.put('/', async (c) => { + const body = await c.req.json().catch(() => null); + if (!body || !body.entry_type || !body.page_name) { + return c.json({ success: false, error: 'entry_type 與 page_name required(upsert key)' }, 400); + } + const { entry, created } = await upsertEntry(c.env.DB, body); + // 內容可能新增或改寫 → 重 embed(保持向量新鮮)。embedOnWrite 內部自檢模組開 + embeddable。 + if (embedEnabled(c.env)) c.executionCtx.waitUntil(embedOnWrite(c.env, entry).catch(() => {})); + return c.json({ success: true, entry, created }); +}); + // GET /entries — list with filters (entry_type, owner_id, parent_id, page_name, source, q/search) // e.g. list workflows under a project: ?parent_id=PROJECT&entry_type=workflow // e.g. get one by idempotency key: ?page_name=skill-rag_with_arcrun