feat(workflow-store): workflow 真相源 KV → KBDB API（entry_type=workflow）

SDD kbdb-base §8.3/§8.4 P2+P3。總管裁定 2026-07-06：
① upsert 走 KBDB base PUT /entries（by owner_id+page_name+entry_type）
② cron 掃 KBDB、不保留 KV cron-idx
③ 一 workflow=一筆 entry_type=workflow；content=description（保 embed）、
   graph+config+cron_expr 進 metadata_json

KBDB：
- entry-crud.ts 加 upsertEntry（read-then-write，無 UNIQUE、表不變）
- entries.ts 加 PUT /entries（upsert，回 created 旗標，content 變動時 embed-on-write）

cypher-executor：
- 新增 lib/workflow-store.ts（走 kbdbBase，禁直連 D1）
- webhooks-named register/trigger/list/delete 切雙軌（讀先 KBDB miss fallback KV、
  寫 KBDB 為主+暫雙寫 KV）；writeWorkflowSearchEntry 上收進 putWorkflow（修重複 entry bug）；
  backfill 升級為 KV→KBDB 遷移入口；移除 migrate-cron-index 端點
- scheduled 改掃 KBDB（listCronWorkflows）；刪除 lib/cron-index.ts
- component-loader（trigger_workflow）、executions 擁有權檢查切雙軌讀

tsc 0 error；kbdb vitest 6/6；cypher vitest 41/42（1 失敗為 pre-existing，與本 PR 無關）。
不 merge、不部署、不動 production KV workflow。搬遷步驟見 MIGRATION.md。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HJiLCRUU2o3aSpPEzVCt2o

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<string, unknown> };
-    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<string, unknown> } | 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');

cypher-executor/src/lib/cron-index.ts

@@ -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<string, string>;
-
-/** 組出索引 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<CronIndex> {
-  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<void> {
-  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));
-}

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<string, unknown>;
+  config?: Record<string, unknown>;
+  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<string, unknown>;
+  config?: Record<string, unknown>;
+  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<KbdbEntry | null> {
+  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<WorkflowRecord | null> {
+  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<WorkflowRecord[]> {
+  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<void> {
+  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<boolean> {
+  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<string, unknown> };
+
+/**
+ * 掃出所有帶 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<CronWorkflow[]> {
+  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 [];
+  }
+}

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',

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<void> {
-  const base = (env.KBDB_BASE_URL ?? 'https://arcrun-kbdb.uncle6-me.workers.dev').replace(/\/$/, '');
-  const headers: Record<string, string> = { '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<string, unknown> = {};
@@ -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<string, { name: string; description: string; created_at: string; cron_expr?: string; webhook_url: string }>();
+  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 });
 });

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<string, unknown>;
-  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`);
 }