fix(kv-quota): workflow 執行紀錄搬離 KV，改走 KBDB template 機制（A1/A2/A7）

事故：cypher-executor/src/actions/execution-logger.ts 舊版每跑完一次 workflow 就 ANALYTICS_KV.put() 一筆新 key（註解寫「避免覆蓋」）= 只增不減，封測者 Evan 處理約 690 個檔案就把 KV 免費層 1,000 write/日打爆（實測 1,070 write），整個實例 429。 A1 少記：workflow 執行紀錄改走 KBDB template 機制（entries 表 entry_type='execution_log'， kbdb/migrations/0004_execution_log_template.sql 只 seed 一列 template 定義，零建表/改表）。儲存精神比照既有 recipe_stat（kbdb/src/actions/recipe-stat.ts）：template 只負責文件化，實際一筆執行是 entries 表一列（1 次執行=1 次 D1 寫入，不走 entry_values 全展開）。欄位收斂：時間/workflow/verdict/duration/錯誤訊息/(可得的)目標；成功記最少，失敗多記（訊息截斷長度不對稱：200 vs 2000 字）。target 只認 trigger context 的 page_name/path，不整包存 input。 A2 自我降級：D1 額度仍與知識卡共用同一顆 100,000 rows/日，本模組自設 20% 軟上限（可用 EXECUTION_LOG_DAILY_WRITE_LIMIT 覆寫），超過 80% 降成只記失敗、超過 100% 完全停止記錄，但 workflow 執行永遠照跑（cypher-executor 端 fire-and-forget 永不 throw）。 A7 讀取端：/workflows/:name/executions、/portal/data/workflows 的 last_execution、MCP list_recent_executions 全部改打 KBDB HTTP API（GET /execution-log、/execution-log/latest），取代原本的 ANALYTICS_KV list/get（免費層 list 也是 1,000/日）。架構鐵律修正（本次施工中兩度被抓到走偏，過程留痕於 commit 訊息供後續參考）： - KBDB 三張表打天下（entries/templates/entry_values），永遠不加新 table——新資料類型一律用 template + entries，不建表、不 ALTER TABLE。 - KBDB = API-as-Wall，零 SQL：cypher-executor 端一律走 KBDB 的 HTTP API（連法比照既有 recordRecipeStats/kbdbFetch 慣例），不直連任何 D1、不對 arcrun-kbdb 下任何原生 SQL。順帶修復：kbdb/src/actions/entry-crud.ts listEntries 的 ORDER BY 補 `, rowid DESC` 二級排序——entries.created_at 是 unixepoch() 秒級解析度，高頻寫入（execution_log 一秒內多筆）常同秒，單靠 created_at DESC 不保證「最新一筆」正確，此為本次測試（latestExecutionLog）發現的既有潛在缺陷，順手補上決定性排序，不改變任何既有查詢在 created_at 不同時的行為。隔離：portal-data.ts INTERNAL_ENTRY_TYPES 加入 execution_log/execution_log_usage（與既有 value/workflow 同層級排除），避免用戶知識搜尋混進執行 log；本模組從不設 metadata_json.embed，故永不進 Vectorize 語意搜尋索引。不動：registry/src/actions/recordAnalytics.ts（零件市場統計，獨立 Worker、獨立 KV 命名空間、不同資料模型，非本次事故根因所指範圍）；cypher-executor/{wrangler.toml,kbdb/wrangler.toml} 未變動（repo 層級 deny 規則保護這兩個生產設定檔不被 AI 編輯）——ANALYTICS_KV binding 因此仍留在 wrangler.toml 宣告中但程式碼零讀寫點（見 PR 說明的完整 grep 佐證）。 KV 裡既有的 stats:* 舊資料不搬移（是統計不是真相源，維持原樣任其依 90 天 TTL 自然過期）。測試：kbdb/tests/execution-log.test.ts（13 個，含零建表證明/少記/A2 降級/route）、 cypher-executor/tests/execution-logger.test.ts（payload 正確性/永不 throw）、 cypher-executor/tests/executions-route.test.ts（讀取端轉發）、portal-data.test.ts 對應區塊改寫。kbdb 全測試 104/104 通過；cypher-executor 320 個測試中 9 個失敗為 main 既有（與本次改動無關，改動前後 stash 對照確認）。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-07 16:13:00 +08:00
parent 36a5630c63
commit 60688c3108
18 changed files with 818 additions and 84 deletions
@@ -93,7 +93,12 @@ export async function listEntries(db: D1Database, f: ListEntriesFilter = {}): Pr
  const offset = f.offset ?? 0;
  const [rowsRes, countRow] = await Promise.all([
    db
-      .prepare(`SELECT * FROM entries ${where} ORDER BY created_at DESC LIMIT ? OFFSET ?`)
+      // `, rowid DESC` 二級排序（KV 額度事故修復，2026-08-07 發現）：created_at 是
+      // unixepoch()＝秒級解析度，高頻寫入（例如 execution_log 一秒內多筆執行）常同秒，
+      // 單靠 created_at DESC 的同分排序不保證插入序，「最新一筆」可能取到錯的一列。
+      // rowid 是 SQLite/D1 一般表的隱含遞增欄，同分時退回插入序，不改變既有排序結果
+      // （created_at 不同時完全一字不變），純粹補上同分時的決定性。
+      .prepare(`SELECT * FROM entries ${where} ORDER BY created_at DESC, rowid DESC LIMIT ? OFFSET ?`)
      .bind(...params, limit, offset)
      .all<Entry>(),
    db.prepare(`SELECT COUNT(*) as total FROM entries ${where}`).bind(...params).first<{ total: number }>(),
@@ -0,0 +1,201 @@
+// Execution log — workflow 執行紀錄（KV 額度事故修復，總管交辦，2026-08-07）
+//
+// SDD：無專屬 SDD（事故修復任務）。root cause 見 kbdb/migrations/0004_execution_log_template.sql
+// 開頭註解：cypher-executor 舊版每跑完一次 workflow 就 ANALYTICS_KV.put() 一筆新 key（永不覆蓋）
+// ⇒ 封測者 690 個檔案就把 KV 免費層 1,000 write/日打爆（實測 1,070 write）。
+//
+// KBDB 鐵律（leo 2026-06-14）：三張表打天下，永遠不加新 table；新資料類型一律用 template。
+// 本模組 schema 走 template 機制（tpl-execution-log，見上述 migration），但**儲存精神比照既有
+// recipe-stat.ts**：template 只負責文件化（GET /templates 可發現欄位定義），實際一筆執行紀錄
+// 是 entries 表的**一列**（entry_type='execution_log'，結構化欄位打包進 metadata_json），
+// 不走 entry_values 全展開的多列 record——那樣一筆執行要拆 5+ 列，1 次執行變 6+ 次 D1 寫入，
+// 直接違反「少記」精神；recipe_stat 早已示範「template 存在＋entries 直接存」這個模式合法。
+//
+// leo 兩條判準：
+//   ① 執行紀錄是稽核資料 → 搬 D1（entries 表，rows written 100,000/日，額度是 KV 的 100 倍）。
+//   ② 不是 n8n、不靠 Execution 計費 → 少記：不留每節點輸入輸出，只留時間/workflow/verdict/
+//      duration/錯誤訊息/(可得的)目標；成功記最少，失敗多記一點（見 SUCCESS/FAILED_MESSAGE_MAX）。
+//
+// A2 自我降級：執行紀錄與知識卡（一般 entries）共用同一顆 D1 100,000 rows/日，搬 D1 只是油箱
+// 大了 100 倍，不是解掉共用額度本身。本模組自設更低的「軟上限」（DEFAULT_DAILY_LIMIT），
+// 用量超過 80% → 降成只記失敗；超過 100% → 完全停止記錄，但呼叫端（cypher-executor）的
+// workflow 執行永遠照跑——寫入永不 throw（recordExecutionLog 本身 catch 見呼叫端 route）。
+//
+// 隔離（不污染知識搜尋）：entry_type='execution_log'/'execution_log_usage' 是內部型別，與既有
+// 'value'/'workflow' 同層級。cypher-executor 端（portal-data.ts INTERNAL_ENTRY_TYPES）比照這兩者
+// 一併排除；本模組也從不設 metadata_json.embed=true，故永不進 Vectorize 語意搜尋索引。
+import type { Bindings } from '../types';
+import { createEntry, listEntries } from './entry-crud';
+
+export interface ExecutionLogInput {
+  workflow_id: string;
+  owner_id?: string | null;
+  verdict: 'success' | 'failed';
+  duration_ms: number;
+  message?: string;
+  target?: string | null;
+}
+
+export interface ExecutionLogRow {
+  workflow_id: string;
+  verdict: string;
+  duration_ms: number;
+  message: string;
+  target?: string;
+  recorded_at: number; // unix seconds（entries.created_at 既有慣例，非毫秒）
+}
+
+/** 成功訊息截斷長度（少記：夠看一眼結果就好，不留診斷用的長上下文）。 */
+const SUCCESS_MESSAGE_MAX = 200;
+/** 失敗訊息截斷長度（不對稱：失敗要留夠診斷用的上下文，比成功多 10 倍）。 */
+const FAILED_MESSAGE_MAX = 2000;
+/** target 欄位截斷長度（page_name / path 通常是檔名或路徑，不會太長；異常長輸入也不整包吞）。 */
+const TARGET_MAX = 300;
+
+/**
+ * 每日軟上限預設值：D1 免費層 100,000 rows written/日與知識卡（一般 entries）共用，
+ * 本模組自設 20%（20,000）——不是 Cloudflare 硬限制，是「執行紀錄不該把知識卡的額度吃光」的
+ * 自我節制門檻，可用 env.EXECUTION_LOG_DAILY_WRITE_LIMIT 覆寫。
+ */
+const DEFAULT_DAILY_LIMIT = 20000;
+/** 用量超過門檻比例 → 降成只記失敗（寫死比例＋可測試，不靠感覺調參）。 */
+const DEGRADE_RATIO = 0.8;
+
+export type UsageMode = 'log' | 'log_failure_only' | 'skip';
+
+function dailyLimit(env: Pick<Bindings, 'EXECUTION_LOG_DAILY_WRITE_LIMIT'>): number {
+  const raw = env.EXECUTION_LOG_DAILY_WRITE_LIMIT;
+  const n = raw ? parseInt(raw, 10) : NaN;
+  return Number.isFinite(n) && n > 0 ? n : DEFAULT_DAILY_LIMIT;
+}
+
+function utcDay(): string {
+  return new Date().toISOString().slice(0, 10);
+}
+
+function truncate(s: string, max: number): string {
+  if (s.length <= max) return s;
+  return s.slice(0, Math.max(0, max - 1)) + '…';
+}
+
+/**
+ * A2 用量計數＋降級判斷。單一 entries 列/日（id=`exlog-usage:{day}`，entry_type=
+ * 'execution_log_usage'，計數包進 metadata_json）——精神完全比照 recipe-stat.ts 的
+ * upsert 慣例（讀現有列 → +1 → UPDATE，不存在則 INSERT）。
+ *
+ * 刻意計「每次呼叫嘗試次數」而非「實際寫入 execution_log 的列數」——即使已降級到
+ * 「只記失敗」或「完全停止」，仍要繼續計數，不然額度耗盡後下一次呼叫又會誤判成
+ * 「還沒超過」而重新開始寫爆（等於沒有降級機制）。day 用 UTC 日期字串，換日自然歸零。
+ */
+export async function checkUsage(db: D1Database, limit: number): Promise<UsageMode> {
+  const id = `exlog-usage:${utcDay()}`;
+  const existing = await db
+    .prepare('SELECT metadata_json FROM entries WHERE id = ?')
+    .bind(id)
+    .first<{ metadata_json: string | null }>();
+
+  let count: number;
+  if (existing) {
+    let prevWrites = 0;
+    try {
+      const prev = existing.metadata_json ? (JSON.parse(existing.metadata_json) as { writes?: number }) : {};
+      prevWrites = Number(prev.writes) || 0;
+    } catch {
+      prevWrites = 0; // 壞資料誠實視為 0，不讓損毀的計數器卡死降級機制
+    }
+    count = prevWrites + 1;
+    await db
+      .prepare('UPDATE entries SET metadata_json = ?, updated_at = unixepoch() WHERE id = ?')
+      .bind(JSON.stringify({ day: utcDay(), writes: count }), id)
+      .run();
+  } else {
+    count = 1;
+    await db
+      .prepare(`INSERT INTO entries (id, entry_type, metadata_json) VALUES (?, 'execution_log_usage', ?)`)
+      .bind(id, JSON.stringify({ day: utcDay(), writes: count }))
+      .run();
+  }
+
+  if (count > limit) return 'skip';
+  if (count > limit * DEGRADE_RATIO) return 'log_failure_only';
+  return 'log';
+}
+
+/**
+ * 寫入一筆執行紀錄（fire-and-forget 語意由呼叫端 route 的 try/catch 保證，本函式本身
+ * 不主動吞錯——route 層統一吞，保持單一吞錯點，避免兩層都吞導致除錯時看不到真因）。
+ */
+export async function recordExecutionLog(
+  db: D1Database,
+  env: Pick<Bindings, 'EXECUTION_LOG_DAILY_WRITE_LIMIT'>,
+  input: ExecutionLogInput,
+): Promise<{ written: boolean; mode: UsageMode }> {
+  const limit = dailyLimit(env);
+  let mode: UsageMode;
+  try {
+    mode = await checkUsage(db, limit);
+  } catch {
+    // fail-open：計數機制本身故障（含 D1 額度打滿）不該連執行紀錄都不寫，
+    // 寧可暫時失去降級能力也不要靜默漏記——這一步的失敗仍不影響下面的實際寫入。
+    mode = 'log';
+  }
+  if (mode === 'skip') return { written: false, mode };
+  if (mode === 'log_failure_only' && input.verdict !== 'failed') return { written: false, mode };
+
+  const maxLen = input.verdict === 'failed' ? FAILED_MESSAGE_MAX : SUCCESS_MESSAGE_MAX;
+  const target = input.target ? truncate(String(input.target), TARGET_MAX) : null;
+
+  await createEntry(db, {
+    entry_type: 'execution_log',
+    owner_id: input.owner_id ?? null,
+    page_name: input.workflow_id, // 索引欄位（idx_entries_page）＝查詢鍵，讀取端靠它篩單一 workflow
+    content: truncate(input.message ?? '', maxLen),
+    metadata_json: JSON.stringify({
+      verdict: input.verdict,
+      duration_ms: Math.max(0, Math.round(input.duration_ms)),
+      target,
+    }),
+  });
+  return { written: true, mode };
+}
+
+/** 讀某 workflow 最近 N 次執行紀錄（降冪）。owner_id 給了才過濾（租戶隔離，caller 決定）。 */
+export async function listExecutionLog(
+  db: D1Database,
+  workflowId: string,
+  ownerId: string | undefined,
+  limit: number,
+): Promise<ExecutionLogRow[]> {
+  const { entries } = await listEntries(db, {
+    entry_type: 'execution_log',
+    page_name: workflowId,
+    owner_id: ownerId,
+    limit,
+  });
+  return entries.map((e) => {
+    let meta: { verdict?: string; duration_ms?: number; target?: string | null } = {};
+    try {
+      meta = e.metadata_json ? (JSON.parse(e.metadata_json) as typeof meta) : {};
+    } catch {
+      /* 壞資料誠實留空，不整筆丟掉（still 回傳 verdict='unknown' 好過整筆消失） */
+    }
+    return {
+      workflow_id: workflowId,
+      verdict: meta.verdict ?? 'unknown',
+      duration_ms: meta.duration_ms ?? 0,
+      message: e.content ?? '',
+      ...(meta.target ? { target: meta.target } : {}),
+      recorded_at: e.created_at,
+    };
+  });
+}
+
+/** 讀某 workflow 最新一次執行紀錄（portal-data.ts last_execution 用）。 */
+export async function latestExecutionLog(
+  db: D1Database,
+  workflowId: string,
+  ownerId: string | undefined,
+): Promise<ExecutionLogRow | null> {
+  const rows = await listExecutionLog(db, workflowId, ownerId, 1);
+  return rows[0] ?? null;
+}