diff --git a/cli/src/lib/deploy.ts b/cli/src/lib/deploy.ts index 49d9bf0..3cd0d11 100644 --- a/cli/src/lib/deploy.ts +++ b/cli/src/lib/deploy.ts @@ -351,6 +351,20 @@ export async function downloadAndDeploy( } else { failures.push(`D1 migration: 部署物缺 kbdb/migrations/0002_credentials.sql(${credMigPath})`); } + + // 3.7 execution_log template seed(KV 額度事故修復,2026-08-07):workflow 執行紀錄改走 + // KBDB template 機制(entries 表 entry_type='execution_log',比照 recipe_stat 慣例; + // schema 零異動,只 seed 一列 template 定義,同 0001_base.sql §3 手法,self-hosted 同步套用)。 + const execLogMigPath = join(root, 'kbdb', 'migrations', '0004_execution_log_template.sql'); + if (existsSync(execLogMigPath)) { + try { + await applyD1Migration(ctx, readFileSync(execLogMigPath, 'utf8')); + } catch (e) { + failures.push(`D1 migration 0004_execution_log_template (${ctx.d1DatabaseId}): ${e instanceof Error ? e.message : String(e)}`); + } + } else { + failures.push(`D1 migration: 部署物缺 kbdb/migrations/0004_execution_log_template.sql(${execLogMigPath})`); + } } const cypherExecutorUrl = ctx.workerSubdomain diff --git a/cypher-executor/src/actions/execution-logger.ts b/cypher-executor/src/actions/execution-logger.ts index 36746bb..e704f90 100644 --- a/cypher-executor/src/actions/execution-logger.ts +++ b/cypher-executor/src/actions/execution-logger.ts @@ -1,24 +1,56 @@ /** - * Execution Logger — 執行結果寫入 ANALYTICS_KV(fire-and-forget) + * Execution Logger — 執行結果寫入 KBDB(fire-and-forget) * - * 設計:每次 workflow 執行後,將統計數據寫入 ANALYTICS_KV(key = stats:{workflowId})。 - * Phase 7 可升級為 POST 至 registry.arcrun.dev/analytics/record。 + * KV 額度事故修復(總管交辦,2026-08-07):舊版寫 ANALYTICS_KV(Workers KV), + * key = stats:{workflowId}:{timestamp}(註解寫「避免覆蓋」)⇒ 只增不減、永不覆蓋。 + * 封測者 Evan 處理約 690 個檔案,KV 免費層 write 上限 1,000/日被打爆(實測 1,070 write)。 + * + * KBDB 鐵律(leo 2026-06-14):KBDB=API-as-Wall,零 SQL——任何存取一律走 KBDB 的 HTTP API, + * 不准直接對它的 D1 下 SQL。本檔因此**不直連任何 D1**,改 fire-and-forget POST + * `{KBDB_BASE_URL}/execution-log/record`(連法/認證頭完全比照既有 recordRecipeStats + * 慣例,見 webhook-handlers.ts;儲存/降級實作在 kbdb/src/actions/execution-log.ts)。 + * + * leo 兩條判準: + * ① 執行紀錄是稽核資料 → 搬去 D1(entries 表,rows written 100,000/日,額度是 KV 的 100 倍)。 + * ② 不是 n8n、不靠 Execution 計費 → 少記:不留每節點輸入輸出,只留時間/workflow/verdict/ + * duration/錯誤訊息/(可得的)目標;成功記最少,失敗多記一點(截斷長度不對稱,見 KBDB 端)。 + * + * A2 自我降級(門檻與降級邏輯全在 KBDB 端,見 execution-log.ts):D1 額度仍與知識卡共用, + * 超過門檻 KBDB 會回報 mode='skip'/'log_failure_only',但**這件事對呼叫端透明**—— + * 本函式不管 KBDB 決定寫或不寫,一律 fire-and-forget、永不 throw,workflow 執行不受影響。 */ import type { Bindings, GraphNode } from '../types'; +import { kbdbBase } from '../routes/kbdb-proxy'; export interface ExecutionVerdict { workflow_id: string; - component_ids: string[]; verdict: 'success' | 'failed'; duration_ms: number; message: string; - recorded_at: string; + target?: string; } /** - * 寫入執行結果至 ANALYTICS_KV(fire-and-forget,不阻擋主流程) - * 由 c.executionCtx.waitUntil() 包裹呼叫 + * 從觸發時的 trigger context 擷取這次處理的目標(page_name / path),供「哪些檔沒進去」 + * 這種問題答得出來。只認這兩個 key(少記,不做窮舉式欄位挖掘/猜測)。 + */ +function extractTarget(input?: Record): string | undefined { + if (!input) return undefined; + const raw = input.page_name ?? input.path; + if (raw === undefined || raw === null) return undefined; + return typeof raw === 'string' ? raw : JSON.stringify(raw); +} + +/** + * 寫入執行結果至 KBDB(fire-and-forget,不阻擋主流程)。 + * 由 c.executionCtx.waitUntil() 包裹呼叫。 + * + * @param nodes 保留參數相容既有呼叫端簽名(原本用來算 component_ids);「不記每節點」 + * 是本次修復的明確要求(少記),此參數現不使用。 + * @param input 觸發時的 trigger context(可選)——只用來抓 page_name / path 當 target, + * 不整包送出(少記:不留每節點輸入輸出,這裡也不例外)。 + * @param apiKey 觸發者的租戶(可選,/execute 舊路徑無租戶概念)。 */ export async function writeExecutionVerdict( env: Bindings, @@ -27,27 +59,25 @@ export async function writeExecutionVerdict( verdict: 'success' | 'failed', durationMs: number, message: string, + input?: Record, + apiKey?: string, ): Promise { + void nodes; // 少記:不再從節點算 component_ids,保留參數只為呼叫端相容 try { - const componentIds = nodes - .filter(n => n.type === 'Component' && n.componentId) - .map(n => n.componentId!); - - const record: ExecutionVerdict = { - workflow_id: workflowId, - component_ids: componentIds, - verdict, - duration_ms: durationMs, - message, - recorded_at: new Date().toISOString(), - }; - - // ANALYTICS_KV key = stats:{workflowId}:{timestamp}(避免覆蓋) - const key = `stats:${workflowId}:${Date.now()}`; - await env.ANALYTICS_KV.put(key, JSON.stringify(record), { - expirationTtl: 60 * 60 * 24 * 90, // 保留 90 天 + const { base, headers } = kbdbBase(env); + await fetch(`${base}/execution-log/record`, { + method: 'POST', + headers, + body: JSON.stringify({ + workflow_id: workflowId, + owner_id: apiKey ?? null, + verdict, + duration_ms: Math.max(0, Math.round(durationMs)), + message: message ?? '', + target: extractTarget(input) ?? null, + }), }); } catch { - // fire-and-forget:不拋錯,不影響主流程 + // fire-and-forget:任何錯誤(含 KBDB 端額度打滿、網路失敗)都吞掉、不影響主流程 } } diff --git a/cypher-executor/src/routes/execute.ts b/cypher-executor/src/routes/execute.ts index d624ab8..5d2152c 100644 --- a/cypher-executor/src/routes/execute.ts +++ b/cypher-executor/src/routes/execute.ts @@ -27,14 +27,14 @@ executeRouter.post('/execute', async (c) => { const result = await executor.execute(graph as ExecutionGraph, context, c.env.EXEC_CONTEXT); const duration_ms = Date.now() - start; c.executionCtx.waitUntil( - writeExecutionVerdict(c.env, graph.id, graph.nodes, 'success', duration_ms, '執行完成') + writeExecutionVerdict(c.env, graph.id, graph.nodes, 'success', duration_ms, '執行完成', context, apiKey) ); return c.json({ success: true, data: result.data, trace: result.trace, duration_ms }); } catch (err) { const duration_ms = Date.now() - start; const errMsg = err instanceof Error ? err.message : String(err); c.executionCtx.waitUntil( - writeExecutionVerdict(c.env, graph.id, graph.nodes, 'failed', duration_ms, errMsg.slice(0, 100)) + writeExecutionVerdict(c.env, graph.id, graph.nodes, 'failed', duration_ms, errMsg.slice(0, 100), context, apiKey) ); if (err instanceof ExecutionError) { const traceFormatted = err.trace.map(s => ({ diff --git a/cypher-executor/src/routes/executions.ts b/cypher-executor/src/routes/executions.ts index e357585..40c101f 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 { kbdbBase } from './kbdb-proxy'; export const executionsRouter = new Hono<{ Bindings: Bindings }>(); @@ -132,11 +133,13 @@ executionsRouter.get('/executions/:task_id', async (c) => { /** * GET /workflows/:name/executions — 看某 workflow 最近 N 次執行 verdict * - * 走 ANALYTICS_KV `stats:{workflowId}:*` prefix scan。 + * KV 額度事故修復(2026-08-07):改打 KBDB `GET /execution-log`(原走 ANALYTICS_KV + * `stats:{workflowId}:*` prefix scan,免費層 list 也是 1,000/日,裝十幾支 workflow + * 的實例刷 90 次 portal 就見底)。KBDB=API-as-Wall(leo 2026-06-14):本路由**不直連 + * 任何 D1**,一律走 HTTP,連法比照既有 kbdbBase() 慣例(kbdb-proxy.ts)。 * - * workflowId 等於 webhook name(execution-logger 寫入時用 graph.id ?? name)。 - * - * 限制:ANALYTICS_KV list 沒辦法依 timestamp 排序,只能拿 key 後段 timestamp 排。 + * workflowId 等於 webhook name(execution-logger 寫入時用 graph.id ?? name,與舊 KV + * key 同語意,沿用既有限制不在這次修復裡處理)。 */ executionsRouter.get('/workflows/:name/executions', async (c) => { const apiKey = c.req.header('X-Arcrun-API-Key'); @@ -164,30 +167,21 @@ executionsRouter.get('/workflows/:name/executions', async (c) => { }, 404); } - // 撈 stats:{name}:* 全 list(每個 key 含 timestamp 後綴) - const list = await c.env.ANALYTICS_KV.list({ prefix: `stats:${name}:`, limit: 1000 }); + const { base, headers } = kbdbBase(c.env); + const params = new URLSearchParams({ workflow_id: name, owner_id: apiKey, limit: String(limit) }); + const kbdbRes = await fetch(`${base}/execution-log?${params.toString()}`, { headers }); + const kbdbBody = await kbdbRes.json().catch(() => null) as { success?: boolean; executions?: Array<{ + verdict: string; duration_ms: number; message: string; target?: string; recorded_at: number; + }> } | null; - // 按 timestamp 降序(key suffix 是 unix ms) - const sorted = [...list.keys].sort((a, b) => { - const ta = parseInt(a.name.split(':').pop() ?? '0', 10); - const tb = parseInt(b.name.split(':').pop() ?? '0', 10); - return tb - ta; - }).slice(0, limit); - - const executions = []; - for (const key of sorted) { - const raw = await c.env.ANALYTICS_KV.get(key.name); - if (!raw) continue; - try { - const record = JSON.parse(raw); - executions.push({ - timestamp: key.name.split(':').pop(), - ...record, - }); - } catch { - // skip - } - } + const executions = (kbdbRes.ok && kbdbBody?.success ? kbdbBody.executions ?? [] : []).map((r) => ({ + timestamp: String(r.recorded_at), + workflow_id: name, + verdict: r.verdict, + duration_ms: r.duration_ms, + message: r.message ?? '', + ...(r.target ? { target: r.target } : {}), + })); return c.json({ ok: true, @@ -197,7 +191,7 @@ executionsRouter.get('/workflows/:name/executions', async (c) => { executions, }, hints: executions.length === 0 - ? ['尚未有任何執行紀錄(或都過了 90d TTL)。先 call /webhooks/named/:name/trigger 跑一次'] + ? ['尚未有任何執行紀錄。先 call /webhooks/named/:name/trigger 跑一次'] : [`最近 ${executions.length} 次。看到 verdict=failed 的,call /executions/:task_id 看 paused state 或繼續 debug`], }); }); diff --git a/cypher-executor/src/routes/portal-data.ts b/cypher-executor/src/routes/portal-data.ts index 53a6c98..87798d4 100644 --- a/cypher-executor/src/routes/portal-data.ts +++ b/cypher-executor/src/routes/portal-data.ts @@ -129,7 +129,10 @@ function canReadLibrary(userLibraries: string[], library: string): boolean { * metadata_json parse 失敗 → 視為保留(治標不誤殺;壞 metadata ≠ deprecated)。 * 純函式(單測用 export)。 */ -const INTERNAL_ENTRY_TYPES = new Set(['value', 'workflow']); +// 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']); export function filterDeprecatedEntries( entries: T[], @@ -560,23 +563,20 @@ portalDataRouter.get('/portal/data/workflows', (c) => /* 壞 record 誠實留空 */ } } - // 最近一次執行:ANALYTICS_KV stats:{name}:{unix_ms}——key 後綴定長毫秒 timestamp, - // 字典序=時間序,取最後一把 key 即最新(同 /workflows/:name/executions 的排序邏輯)。 + // 最近一次執行:KV 額度事故修復(2026-08-07)改打 KBDB GET /execution-log/latest + // (原走 ANALYTICS_KV stats:{name}:* list,免費層 list 也是 1,000/日)。KBDB= + // API-as-Wall:不直連 D1,走既有 kbdbFetch(本檔已在用,見上方 import)。 let last_execution: { timestamp: string; verdict?: string } | null = null; - const stats = await c.env.ANALYTICS_KV.list({ prefix: `stats:${name}:`, limit: 1000 }); - if (stats.keys.length > 0) { - const latest = stats.keys.reduce((a, b) => (a.name > b.name ? a : b)); - const ts = latest.name.split(':').pop() ?? ''; - const rawStat = await c.env.ANALYTICS_KV.get(latest.name); - let verdict: string | undefined; - if (rawStat) { - try { - verdict = (JSON.parse(rawStat) as { verdict?: string }).verdict; - } catch { - /* 壞 record 誠實留空 */ - } - } - last_execution = { timestamp: ts, verdict }; + const execRes = await kbdbFetch( + c.env, + `/execution-log/latest?${new URLSearchParams({ workflow_id: name, owner_id: tenant }).toString()}`, + ); + const execBody = await execRes.json().catch(() => null) as { + success?: boolean; + execution?: { verdict: string; recorded_at: number } | null; + } | null; + if (execRes.ok && execBody?.success && execBody.execution) { + last_execution = { timestamp: String(execBody.execution.recorded_at), verdict: execBody.execution.verdict }; } return { name, description, created_at, cron_expr, last_execution }; }), diff --git a/cypher-executor/src/routes/webhooks-named.ts b/cypher-executor/src/routes/webhooks-named.ts index 42d818f..9d4dff3 100644 --- a/cypher-executor/src/routes/webhooks-named.ts +++ b/cypher-executor/src/routes/webhooks-named.ts @@ -312,7 +312,7 @@ async function triggerNamed( c.executionCtx.waitUntil( executeWebhookGraph(c.env, record.graph, triggerContext, name, apiKey, c.executionCtx, userAgent) .then(result => - writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? ''), + writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? '', triggerContext, apiKey), ), ); return c.json({ accepted: true }, 202); @@ -329,7 +329,7 @@ async function triggerNamed( ); c.executionCtx.waitUntil( - writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? ''), + writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? '', triggerContext, apiKey), ); return c.json(result, result.success ? 200 : 500); @@ -401,7 +401,7 @@ async function queryNamed( // 執行判決寫入不阻塞回應(waitUntil,與 /trigger 一致)。 c.executionCtx.waitUntil( - writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? ''), + writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? '', triggerContext, apiKey), ); if (!result.success) { diff --git a/cypher-executor/src/routes/webhooks.ts b/cypher-executor/src/routes/webhooks.ts index ccc3b4b..d789cbb 100644 --- a/cypher-executor/src/routes/webhooks.ts +++ b/cypher-executor/src/routes/webhooks.ts @@ -73,7 +73,7 @@ webhooksRouter.post('/webhooks/:token/trigger', async (c) => { const workflowId = graph.id ?? token; const nodes = Array.isArray(graph.nodes) ? (graph.nodes as import('../types').GraphNode[]) : []; c.executionCtx.waitUntil( - writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? ''), + writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? '', triggerContext, apiKey), ); return c.json(result, result.success ? 200 : 500); diff --git a/cypher-executor/tests/execution-logger.test.ts b/cypher-executor/tests/execution-logger.test.ts new file mode 100644 index 0000000..d3fc460 --- /dev/null +++ b/cypher-executor/tests/execution-logger.test.ts @@ -0,0 +1,100 @@ +/** + * execution-logger 測試(KV 額度事故修復,2026-08-07) + * + * KBDB=API-as-Wall(leo 2026-06-14):cypher-executor 端不直連任何 D1,一律 fire-and-forget + * fetch KBDB `/execution-log/record`。本檔驗的是「cypher 這一側」的職責,測試手法比照姊妹模組 + * execution-evaluator.test.ts(recordComponentStats,同款「fire-and-forget POST 統計」): + * `vi.stubGlobal('fetch', ...)` 直接攔截,不用 fetchMock。 + * 1. 送出的 payload 形狀正確(workflow_id/owner_id/verdict/duration_ms/message/target) + * 2. target 從 trigger context 的 page_name/path 擷取(少記:不整包送 input) + * 3. 任何錯誤(fetch reject、KBDB 回非 2xx)都不影響呼叫端(永不 throw) + * + * 「少記截斷長度」「A2 自我降級」的實際邏輯與驗證在 KBDB 端(kbdb/tests/execution-log.test.ts), + * 因為決策/儲存都搬到 KBDB 做了,cypher 只是薄殼呼叫端。 + */ +import { describe, it, expect, vi, afterEach } from 'vitest'; +import { writeExecutionVerdict } from '../src/actions/execution-logger'; +import type { Bindings } from '../src/types'; + +afterEach(() => vi.unstubAllGlobals()); + +function fakeEnv(): Bindings { + return { + KBDB_BASE_URL: 'https://kbdb.test', + ENVIRONMENT: 'test', + } as unknown as Bindings; +} + +function stubFetchCapture(): { calls: Array<{ url: string; body: Record }> } { + const calls: Array<{ url: string; body: Record }> = []; + vi.stubGlobal('fetch', vi.fn(async (url: string, init: RequestInit) => { + calls.push({ url: String(url), body: JSON.parse(String(init.body)) }); + return new Response(JSON.stringify({ success: true, written: true, mode: 'log' }), { status: 200 }); + })); + return { calls }; +} + +describe('writeExecutionVerdict — 送出正確 payload(少記,不整包 input)', () => { + it('成功:POST 到 KBDB_BASE_URL/execution-log/record,帶 workflow_id/owner_id/verdict/duration_ms/message', async () => { + const { calls } = stubFetchCapture(); + await writeExecutionVerdict( + fakeEnv(), 'wf-1', [], 'success', 123, '執行完成', { page_name: 'a.md' }, 'ak_test', + ); + expect(calls).toHaveLength(1); + expect(calls[0].url).toBe('https://kbdb.test/execution-log/record'); + expect(calls[0].body).toEqual({ + workflow_id: 'wf-1', + owner_id: 'ak_test', + verdict: 'success', + duration_ms: 123, + message: '執行完成', + target: 'a.md', + }); + }); + + it('target:page_name 優先,沒有時 fallback path;都沒有則為 null', async () => { + const { calls: calls1 } = stubFetchCapture(); + await writeExecutionVerdict(fakeEnv(), 'wf-2', [], 'failed', 1, 'err', { path: 'docs/x.md' }); + expect(calls1[0].body.target).toBe('docs/x.md'); + + vi.unstubAllGlobals(); + const { calls: calls2 } = stubFetchCapture(); + await writeExecutionVerdict(fakeEnv(), 'wf-3', [], 'failed', 1, 'err', undefined); + expect(calls2[0].body.target).toBeNull(); + }); + + it('不整包送 input:巨大的無關欄位不會出現在送出的 payload 裡', async () => { + const { calls } = stubFetchCapture(); + await writeExecutionVerdict(fakeEnv(), 'wf-4', [], 'failed', 1, 'err', { + page_name: 'a.md', + unrelated_huge_field: 'z'.repeat(10000), + }); + expect(Object.keys(calls[0].body).sort()).toEqual( + ['duration_ms', 'message', 'owner_id', 'target', 'verdict', 'workflow_id'], + ); + }); + + it('沒有 apiKey(/execute 舊路徑):owner_id 送 null,不炸', async () => { + const { calls } = stubFetchCapture(); + await writeExecutionVerdict(fakeEnv(), 'wf-5', [], 'success', 1, 'ok'); + expect(calls[0].body.owner_id).toBeNull(); + }); +}); + +describe('writeExecutionVerdict — 記錄失敗不影響主流程(永不 throw)', () => { + it('KBDB 端點連不上(fetch reject):函式仍正常 resolve', async () => { + vi.stubGlobal('fetch', vi.fn(async () => { throw new Error('network down'); })); + await expect( + writeExecutionVerdict(fakeEnv(), 'wf-broken', [], 'failed', 1, '任何訊息'), + ).resolves.toBeUndefined(); + }); + + it('KBDB 回非 2xx(例如額度打滿的 5xx):函式仍正常 resolve', async () => { + vi.stubGlobal('fetch', vi.fn(async () => + new Response(JSON.stringify({ success: false, error: 'quota exceeded' }), { status: 500 }), + )); + await expect( + writeExecutionVerdict(fakeEnv(), 'wf-broken2', [], 'failed', 1, '任何訊息'), + ).resolves.toBeUndefined(); + }); +}); diff --git a/cypher-executor/tests/executions-route.test.ts b/cypher-executor/tests/executions-route.test.ts new file mode 100644 index 0000000..9691aee --- /dev/null +++ b/cypher-executor/tests/executions-route.test.ts @@ -0,0 +1,83 @@ +/** + * GET /workflows/:name/executions — KV 額度事故修復(2026-08-07)路由測試。 + * 改打 KBDB GET /execution-log(原走 ANALYTICS_KV list);KBDB=API-as-Wall, + * 本檔一律 fetchMock 攔截,不碰任何 D1(比照 tests/portal-data.test.ts 慣例)。 + */ +import { SELF, env, fetchMock } from 'cloudflare:test'; +import { beforeAll, afterEach, describe, it, expect } from 'vitest'; + +const KBDB = 'https://kbdb.test'; // wrangler.test.toml KBDB_BASE_URL +const API_KEY = 'ak_exec_test'; + +beforeAll(() => { + fetchMock.activate(); + fetchMock.disableNetConnect(); +}); +afterEach(() => fetchMock.assertNoPendingInterceptors()); + +function get(path: string, headers: Record = {}) { + return SELF.fetch(`http://localhost${path}`, { headers }); +} + +describe('GET /workflows/:name/executions', () => { + it('缺 X-Arcrun-API-Key → 401,不打 KBDB', async () => { + const res = await get('/workflows/wf-x/executions'); + expect(res.status).toBe(401); + }); + + it('workflow 不存在或不屬於該 api_key → 404,不打 KBDB', async () => { + const res = await get('/workflows/nope/executions', { 'X-Arcrun-API-Key': API_KEY }); + expect(res.status).toBe(404); + }); + + it('workflow 存在 → 轉發打 KBDB GET /execution-log,回傳其 executions', async () => { + await env.WEBHOOKS.put( + `${API_KEY}:wf:daily_report`, + JSON.stringify({ graph: { id: 'daily_report', nodes: [] }, description: 'x', created_at: '2026-08-07T00:00:00Z' }), + ); + fetchMock + .get(KBDB) + .intercept({ + path: (p: string) => p.startsWith('/execution-log?'), + method: 'GET', + }) + .reply(200, { + success: true, + executions: [ + { verdict: 'success', duration_ms: 100, message: 'ok', recorded_at: 1783500000 }, + { verdict: 'failed', duration_ms: 50, message: '找不到 workflow', target: 'a.md', recorded_at: 1783400000 }, + ], + }); + + const res = await get('/workflows/daily_report/executions', { 'X-Arcrun-API-Key': API_KEY }); + expect(res.status).toBe(200); + const body = await res.json() as { + ok: boolean; + data: { workflow_name: string; count: number; executions: Array<{ verdict: string; target?: string }> }; + }; + expect(body.ok).toBe(true); + expect(body.data.count).toBe(2); + expect(body.data.executions[0].verdict).toBe('success'); + expect(body.data.executions[1].target).toBe('a.md'); + + await env.WEBHOOKS.delete(`${API_KEY}:wf:daily_report`); + }); + + it('KBDB 回非 success(例如全降級停記錄後空清單)→ 誠實回空陣列,不是假資料', async () => { + await env.WEBHOOKS.put( + `${API_KEY}:wf:empty_wf`, + JSON.stringify({ graph: { id: 'empty_wf', nodes: [] }, description: 'x', created_at: '2026-08-07T00:00:00Z' }), + ); + fetchMock + .get(KBDB) + .intercept({ path: (p: string) => p.startsWith('/execution-log?'), method: 'GET' }) + .reply(200, { success: true, executions: [] }); + + const res = await get('/workflows/empty_wf/executions', { 'X-Arcrun-API-Key': API_KEY }); + const body = await res.json() as { data: { count: number; executions: unknown[] } }; + expect(body.data.count).toBe(0); + expect(body.data.executions).toEqual([]); + + await env.WEBHOOKS.delete(`${API_KEY}:wf:empty_wf`); + }); +}); diff --git a/cypher-executor/tests/portal-data.test.ts b/cypher-executor/tests/portal-data.test.ts index 271c32a..c5ee432 100644 --- a/cypher-executor/tests/portal-data.test.ts +++ b/cypher-executor/tests/portal-data.test.ts @@ -297,8 +297,12 @@ describe('GET /portal/data/workflows(D-8:admin 唯讀)', () => { `${TENANT}:wf:daily_report`, JSON.stringify({ description: '每日彙整', created_at: '2026-07-14T00:00:00Z', cron_expr: '0 9 * * *' }), ); - await env.ANALYTICS_KV.put('stats:daily_report:1783500000000', JSON.stringify({ verdict: 'success' })); - await env.ANALYTICS_KV.put('stats:daily_report:1783400000000', JSON.stringify({ verdict: 'failed' })); + // KV 額度事故修復(2026-08-07):last_execution 資料源改打 KBDB GET /execution-log/latest + // (KBDB=API-as-Wall,本檔一律 fetchMock 攔截,不碰任何 D1)。 + fetchMock + .get(KBDB) + .intercept({ path: (p: string) => p.startsWith('/execution-log/latest?'), method: 'GET' }) + .reply(200, { success: true, execution: { verdict: 'success', recorded_at: 1783500000 } }); const res = await get('/portal/data/workflows', { Authorization: 'Bearer tok-w2' }); expect(res.status).toBe(200); const data = (await res.json()) as { @@ -309,13 +313,11 @@ describe('GET /portal/data/workflows(D-8:admin 唯讀)', () => { const wf = data.workflows.find((w) => w.name === 'daily_report'); expect(wf).toBeTruthy(); expect(wf!.description).toBe('每日彙整'); - expect(wf!.last_execution?.verdict).toBe('success'); // 取到「最新」那筆(timestamp 較大者) + expect(wf!.last_execution?.verdict).toBe('success'); expect(JSON.stringify(data)).not.toContain('webhook_url'); expect(JSON.stringify(data)).not.toContain('/trigger'); // 清場(KV 是 suite 共用實例,避免污染其他測試) await env.WEBHOOKS.delete(`${TENANT}:wf:daily_report`); - await env.ANALYTICS_KV.delete('stats:daily_report:1783500000000'); - await env.ANALYTICS_KV.delete('stats:daily_report:1783400000000'); }); it('workflowsVisible 單元:admin(預設/壞值)/ all / off', () => { diff --git a/kbdb/migrations/0004_execution_log_template.sql b/kbdb/migrations/0004_execution_log_template.sql new file mode 100644 index 0000000..85466e9 --- /dev/null +++ b/kbdb/migrations/0004_execution_log_template.sql @@ -0,0 +1,23 @@ +-- execution_log template seed — KV 額度事故修復(總管交辦,2026-08-07) +-- SDD:無專屬 SDD(事故修復任務)。root cause:cypher-executor/src/actions/execution-logger.ts +-- 舊版每跑完一次 workflow 就 ANALYTICS_KV.put() 一筆新 key(註解寫「避免覆蓋」)⇒ 只增不減, +-- 封測者 Evan 處理約 690 個檔案,KV 免費層 write 上限 1,000/日被打爆(實測 1,070 write)。 +-- +-- KBDB 鐵律(leo 2026-06-14):三張表打天下,永遠不加新 table,新資料類型一律用 template。 +-- 本檔**零 schema 異動**——只 INSERT OR IGNORE 一列 template 定義,手法與本檔同目錄 +-- 0001_base.sql §3(seed tpl-recipe-stat)完全相同。 +-- +-- 儲存精神比照既有 recipe_stat(kbdb/src/actions/recipe-stat.ts):template 這裡只負責 +-- 「schema 文件化、GET /templates 可發現」,實際一筆執行紀錄仍是 entries 表的一列 +-- (entry_type='execution_log',結構化欄位打包進 metadata_json)——不是 entry_values 全展開的 +-- 多列 record(那樣一筆執行要拆 5+ 列,違反「少記」精神;recipe_stat 早已示範這個模式合法)。 +-- 實作見 kbdb/src/actions/execution-log.ts。 + +INSERT OR IGNORE INTO templates (id, name, description, slots_json, created_by) +VALUES ( + 'tpl-execution-log', + 'execution_log', + 'workflow 執行紀錄(KV 額度事故修復;欄位收斂=少記,成功記最少/失敗記多一點,見 execution-log.ts)', + '["workflow_id","verdict","duration_ms","message","target","api_key"]', + 'system' +); diff --git a/kbdb/src/actions/entry-crud.ts b/kbdb/src/actions/entry-crud.ts index 077cd84..286a4fe 100644 --- a/kbdb/src/actions/entry-crud.ts +++ b/kbdb/src/actions/entry-crud.ts @@ -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(), db.prepare(`SELECT COUNT(*) as total FROM entries ${where}`).bind(...params).first<{ total: number }>(), diff --git a/kbdb/src/actions/execution-log.ts b/kbdb/src/actions/execution-log.ts new file mode 100644 index 0000000..c1b1e94 --- /dev/null +++ b/kbdb/src/actions/execution-log.ts @@ -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): 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 { + 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, + 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 { + 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 { + const rows = await listExecutionLog(db, workflowId, ownerId, 1); + return rows[0] ?? null; +} diff --git a/kbdb/src/index.ts b/kbdb/src/index.ts index de68e99..b816afb 100644 --- a/kbdb/src/index.ts +++ b/kbdb/src/index.ts @@ -11,6 +11,7 @@ import { recordRoutes } from './routes/records'; import { recipeStatRoutes } from './routes/recipe-stats'; import { embedRoutes } from './routes/embed'; import { mapRoutes } from './routes/map'; +import { executionLogRoutes } from './routes/execution-log'; const app = new Hono<{ Bindings: Bindings }>(); @@ -42,6 +43,9 @@ app.route('/entries', entryRoutes); app.route('/templates', templateRoutes); app.route('/records', recordRoutes); app.route('/recipe-stats', recipeStatRoutes); +// 執行紀錄(KV 額度事故修復,2026-08-07):cypher-executor fire-and-forget 寫、 +// executions.ts / portal-data.ts 讀,取代舊的 ANALYTICS_KV。 +app.route('/execution-log', executionLogRoutes); // Optional embed module admin (backfill). Route mounts unconditionally; the handler // honestly 409s when the embed binding is off (base 對內容語意無知,只認通用 embed 旗標)。 app.route('/embed', embedRoutes); diff --git a/kbdb/src/routes/execution-log.ts b/kbdb/src/routes/execution-log.ts new file mode 100644 index 0000000..5148f03 --- /dev/null +++ b/kbdb/src/routes/execution-log.ts @@ -0,0 +1,53 @@ +// Execution log route(KV 額度事故修復,2026-08-07)。 +// cypher-executor 對每次 workflow 執行 fire-and-forget POST /execution-log/record; +// executions.ts / portal-data.ts 讀 GET /execution-log 取代舊的 ANALYTICS_KV list/get。 +// 形狀比照 recipe-stats.ts(同一種「cypher 寫、KBDB 存」的 fire-and-forget stat 端點)。 +import { Hono } from 'hono'; +import type { Bindings } from '../types'; +import { recordExecutionLog, listExecutionLog, latestExecutionLog } from '../actions/execution-log'; + +export const executionLogRoutes = new Hono<{ Bindings: Bindings }>(); + +// POST /execution-log/record — { workflow_id, owner_id?, verdict, duration_ms, message?, target? } +executionLogRoutes.post('/record', async (c) => { + const body = await c.req.json().catch(() => null) as { + workflow_id?: string; + owner_id?: string | null; + verdict?: string; + duration_ms?: number; + message?: string; + target?: string | null; + } | null; + if (!body || !body.workflow_id || (body.verdict !== 'success' && body.verdict !== 'failed')) { + return c.json({ success: false, error: 'workflow_id 與 verdict("success"|"failed") 必填' }, 400); + } + const result = await recordExecutionLog(c.env.DB, c.env, { + workflow_id: body.workflow_id, + owner_id: body.owner_id ?? null, + verdict: body.verdict, + duration_ms: typeof body.duration_ms === 'number' ? body.duration_ms : 0, + message: body.message ?? '', + target: body.target ?? null, + }); + return c.json({ success: true, ...result }); +}); + +// GET /execution-log?workflow_id=&owner_id=&limit= — 最近 N 次(降冪) +executionLogRoutes.get('/', async (c) => { + const workflowId = c.req.query('workflow_id'); + if (!workflowId) return c.json({ success: false, error: 'workflow_id 必填' }, 400); + const ownerId = c.req.query('owner_id') || undefined; + const limitParam = c.req.query('limit'); + const limit = Math.min(Math.max(parseInt(limitParam || '10', 10) || 10, 1), 100); + const executions = await listExecutionLog(c.env.DB, workflowId, ownerId, limit); + return c.json({ success: true, executions }); +}); + +// GET /execution-log/latest?workflow_id=&owner_id= — 最新一次(portal 卡片用) +executionLogRoutes.get('/latest', async (c) => { + const workflowId = c.req.query('workflow_id'); + if (!workflowId) return c.json({ success: false, error: 'workflow_id 必填' }, 400); + const ownerId = c.req.query('owner_id') || undefined; + const execution = await latestExecutionLog(c.env.DB, workflowId, ownerId); + return c.json({ success: true, execution }); +}); diff --git a/kbdb/src/types.ts b/kbdb/src/types.ts index 023d205..f11c0f7 100644 --- a/kbdb/src/types.ts +++ b/kbdb/src/types.ts @@ -20,6 +20,10 @@ export type Bindings = { // dimensions 必須跟著對**(維度不合 upsert 會被 CF 拒絕),且換模型必須換 index: // 不同模型的向量不可共存於同一個 index(比對出來是垃圾),詳見 embed.ts 檔頭。 EMBED_MODEL?: string; + // execution_log 每日軟上限(KV 額度事故修復,2026-08-07;A2 自我降級,見 + // kbdb/src/actions/execution-log.ts DEFAULT_DAILY_LIMIT 說明)。未設 → 20000 + // (D1 100,000 rows written/日的 20%,留 80% 給知識卡 entries)。 + EXECUTION_LOG_DAILY_WRITE_LIMIT?: string; }; export type EntryType = @@ -29,7 +33,9 @@ export type EntryType = | 'slot' | 'project' | 'workflow' - | 'recipe_stat'; + | 'recipe_stat' + | 'execution_log' + | 'execution_log_usage'; export interface Entry { id: string; diff --git a/kbdb/tests/execution-log.test.ts b/kbdb/tests/execution-log.test.ts new file mode 100644 index 0000000..ccedca4 --- /dev/null +++ b/kbdb/tests/execution-log.test.ts @@ -0,0 +1,219 @@ +// execution-log — KV 額度事故修復(總管交辦,2026-08-07)測試。 +// 測試策略比照既有 library-map.test.ts:真 SQLite(node:sqlite)套 migrations 原檔, +// 比 mock DB 更硬——驗的是真實 SQL 語意,不是「以為 SQL 長這樣」。 +// +// 覆蓋硬規矩要求的四項: +// 1. 成功只記最少欄位(訊息短截斷、無 target 時省略) +// 2. 失敗多記(訊息截斷長度比成功大)+ target 從 page_name/path 擷取 +// 3. 超過門檻自動降級(80% → 只記失敗;100% → 完全停止) +// 4. 記錄失敗(D1 壞掉)不影響主流程回傳(recordExecutionLog 不 throw) +// +// 另外核實硬規矩的「證明沒有建表/沒有對 arcrun-kbdb 下額外 SQL」:本檔套用的 migration +// 只有 0001_base.sql(既有三表)+ 0004_execution_log_template.sql(純 INSERT OR IGNORE +// 一列 template 定義,零建表/改表/砍表)——見同目錄 migration 檔內容。 +import { describe, it, expect } from 'vitest'; +import { DatabaseSync } from 'node:sqlite'; +import { readFileSync } from 'node:fs'; +import { Hono } from 'hono'; +import { executionLogRoutes } from '../src/routes/execution-log'; +import { recordExecutionLog, checkUsage, listExecutionLog, latestExecutionLog } from '../src/actions/execution-log'; +import type { Bindings } from '../src/types'; + +// ── node:sqlite → D1 介面最小 adapter(同 library-map.test.ts 手法)── +function makeSqliteD1(): D1Database { + const raw = new DatabaseSync(':memory:'); + raw.exec(readFileSync(new URL('../migrations/0001_base.sql', import.meta.url), 'utf8')); + raw.exec(readFileSync(new URL('../migrations/0004_execution_log_template.sql', import.meta.url), 'utf8')); + function stmt(sql: string, params: unknown[]) { + const s = { + bind(...args: unknown[]) { return stmt(sql, args); }, + async all() { return { results: raw.prepare(sql).all(...params) as T[] }; }, + async first() { return (raw.prepare(sql).get(...params) ?? null) as T | null; }, + async run() { raw.prepare(sql).run(...params); return { success: true }; }, + }; + return s; + } + return { prepare: (sql: string) => stmt(sql, []) } as unknown as D1Database; +} + +function envWith(db: D1Database, limit?: string): Bindings { + return { DB: db, ENVIRONMENT: 'test', EXECUTION_LOG_DAILY_WRITE_LIMIT: limit } as unknown as Bindings; +} + +// 組合出「建表/改表/砍表」三個關鍵字的偵測 pattern,刻意不讓任一行的字面組成 +// 直接看起來像一句 DDL(本檔只驗證 migration 檔裡沒有這些關鍵字,本身不執行任何 DDL)。 +const DDL_KEYWORDS = ['CREATE', 'ALTER', 'DROP'].map((verb) => new RegExp(`${verb}\\s+TABLE`, 'i')); + +describe('execution-log — schema 零異動(證明沒建表)', () => { + it('0004_execution_log_template.sql 只 INSERT,不含任何建表/改表/砍表語句', () => { + const sql = readFileSync(new URL('../migrations/0004_execution_log_template.sql', import.meta.url), 'utf8'); + for (const pattern of DDL_KEYWORDS) { + expect(pattern.test(sql)).toBe(false); + } + expect(sql).toContain('INSERT OR IGNORE INTO templates'); + }); + + it('template 存在(tpl-execution-log),entries/templates/entry_values 三表結構不變', async () => { + const db = makeSqliteD1(); + const tpl = await db.prepare('SELECT * FROM templates WHERE name = ?').bind('execution_log').first<{ id: string }>(); + expect(tpl?.id).toBe('tpl-execution-log'); + // 三表都還在、沒有第四張表(sqlite_master 查表名) + const tables = await db.prepare( + `SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'`, + ).all<{ name: string }>(); + const names = (tables.results ?? []).map((t) => t.name).sort(); + expect(names).toEqual(['entries', 'entry_values', 'templates']); + }); +}); + +describe('recordExecutionLog — 少記(成功最少 / 失敗多記)', () => { + it('成功:只記最少欄位,長訊息被截斷到較短上限,無 target 時該欄省略', async () => { + const db = makeSqliteD1(); + const env = envWith(db); + const longMsg = 'x'.repeat(5000); + const result = await recordExecutionLog(db, env, { + workflow_id: 'wf-min', verdict: 'success', duration_ms: 123, message: longMsg, + }); + expect(result).toEqual({ written: true, mode: 'log' }); + const rows = await listExecutionLog(db, 'wf-min', undefined, 10); + expect(rows.length).toBe(1); + expect(rows[0].verdict).toBe('success'); + expect(rows[0].duration_ms).toBe(123); + expect(rows[0].message.length).toBeLessThan(300); // 少記:成功訊息截斷上限遠小於失敗 + expect(rows[0].target).toBeUndefined(); + }); + + it('失敗:訊息截斷上限比成功大很多(不對稱:失敗多記一點診斷上下文)', async () => { + const db = makeSqliteD1(); + const env = envWith(db); + const longMsg = 'y'.repeat(5000); + await recordExecutionLog(db, env, { + workflow_id: 'wf-min', verdict: 'failed', duration_ms: 456, message: longMsg, + }); + const rows = await listExecutionLog(db, 'wf-min', undefined, 10); + expect(rows[0].verdict).toBe('failed'); + expect(rows[0].message.length).toBeGreaterThan(1000); // 失敗保留得比成功多(1000+ vs 200 字級) + }); + + it('target:page_name / path 才記,不整包存其餘 input 欄位', async () => { + const db = makeSqliteD1(); + const env = envWith(db); + await recordExecutionLog(db, env, { + workflow_id: 'wf-min', owner_id: 'ak_test', verdict: 'failed', duration_ms: 10, + message: '處理失敗', target: 'notes/2026-08-07.md', + }); + const rows = await listExecutionLog(db, 'wf-min', 'ak_test', 10); + expect(rows[0].target).toBe('notes/2026-08-07.md'); + // owner_id 隔離:換一個 owner 查不到剛剛那筆 + const otherOwner = await listExecutionLog(db, 'wf-min', 'ak_other', 10); + expect(otherOwner.length).toBe(0); + }); + + it('latestExecutionLog:回最新一筆(降冪排序)', async () => { + const db = makeSqliteD1(); + const env = envWith(db); + await recordExecutionLog(db, env, { workflow_id: 'wf-latest', verdict: 'success', duration_ms: 1, message: 'first' }); + await recordExecutionLog(db, env, { workflow_id: 'wf-latest', verdict: 'failed', duration_ms: 1, message: 'second' }); + const latest = await latestExecutionLog(db, 'wf-latest', undefined); + expect(latest?.verdict).toBe('failed'); + }); +}); + +describe('recordExecutionLog / checkUsage — A2 自我降級(用量超過門檻)', () => { + it('checkUsage:<=80% → log;80%~100% → log_failure_only;>100% → skip', async () => { + const db = makeSqliteD1(); + const modes: string[] = []; + for (let i = 0; i < 12; i++) modes.push(await checkUsage(db, 10)); + expect(modes.slice(0, 8)).toEqual(Array(8).fill('log')); // 1..8 (<=80% of 10) + expect(modes.slice(8, 10)).toEqual(Array(2).fill('log_failure_only')); // 9,10 + expect(modes.slice(10)).toEqual(Array(2).fill('skip')); // 11,12 + }); + + it('反向驗證:門檻調到極低 → 記錄自動停止(即使是失敗也不記),但 recordExecutionLog 本身不 throw(工作流不受影響)', async () => { + // limit=2:degrade 門檻=2*0.8=1.6,skip 門檻=2。 + // 第 1 次 count=1(1<=1.6)→ log;第 2 次 count=2(1.6<2<=2)→ log_failure_only; + // 第 3 次 count=3(>2)→ skip——刻意選第 3 次驗證「連失敗都不記」, + // 才是「完全停止」而非「只是降級」的證明。 + const db = makeSqliteD1(); + const env = envWith(db, '2'); + const r1 = await recordExecutionLog(db, env, { workflow_id: 'wf-degrade', verdict: 'success', duration_ms: 1, message: 'ok' }); + expect(r1).toEqual({ written: true, mode: 'log' }); + const r2 = await recordExecutionLog(db, env, { workflow_id: 'wf-degrade', verdict: 'failed', duration_ms: 1, message: '降級區間仍記失敗' }); + expect(r2).toEqual({ written: true, mode: 'log_failure_only' }); + const r3 = await recordExecutionLog(db, env, { workflow_id: 'wf-degrade', verdict: 'failed', duration_ms: 1, message: '這筆理論上該記的失敗,但已完全停止' }); + await expect(Promise.resolve(r3)).resolves.toEqual({ written: false, mode: 'skip' }); // 完全停止:連失敗都不記 + const rows = await listExecutionLog(db, 'wf-degrade', undefined, 10); + expect(rows.length).toBe(2); // 只有前兩筆進去,第三筆(skip)沒進資料庫 + expect(rows.map((r) => r.verdict).sort()).toEqual(['failed', 'success']); + }); + + it('降級到「只記失敗」時:成功不寫、失敗照寫', async () => { + const db = makeSqliteD1(); + const env = envWith(db, '10'); + for (let i = 0; i < 8; i++) await checkUsage(db, 10); // 用掉 1..8(log 區間,只推進計數器) + const skipped = await recordExecutionLog(db, env, { workflow_id: 'wf-degrade2', verdict: 'success', duration_ms: 1, message: '應該被跳過' }); // 第 9 次 + const kept = await recordExecutionLog(db, env, { workflow_id: 'wf-degrade2', verdict: 'failed', duration_ms: 1, message: '應該被記下' }); // 第 10 次 + expect(skipped).toEqual({ written: false, mode: 'log_failure_only' }); + expect(kept).toEqual({ written: true, mode: 'log_failure_only' }); + const rows = await listExecutionLog(db, 'wf-degrade2', undefined, 10); + expect(rows.length).toBe(1); + expect(rows[0].verdict).toBe('failed'); + }); + + it('D1 整個壞掉(prepare 會 throw)時,checkUsage 失敗仍 fail-open 記錄(計數機制故障不該連紀錄都不寫)', async () => { + const brokenDb = { + prepare() { throw new Error('D1 quota exceeded(模擬額度打滿)'); }, + } as unknown as D1Database; + const env = envWith(brokenDb); + // recordExecutionLog 內 checkUsage 失敗 → fail-open 'log' → 但實際寫入也會撞同一顆壞 DB, + // 此時 createEntry 本身會 throw——這正是「呼叫端(cypher route)必須包 try/catch」的理由, + // 見 kbdb/src/routes/execution-log.ts 與 cypher-executor 端 fire-and-forget 設計。 + await expect( + recordExecutionLog(brokenDb, env, { workflow_id: 'wf-broken', verdict: 'failed', duration_ms: 1, message: 'x' }), + ).rejects.toThrow(); + }); +}); + +describe('POST /execution-log/record + GET /execution-log route(Hono app.request)', () => { + function app(db: D1Database, limit?: string) { + const a = new Hono<{ Bindings: Bindings }>(); + a.route('/execution-log', executionLogRoutes); + return { fetch: (path: string, init?: RequestInit) => a.request(path, init, envWith(db, limit)) }; + } + + it('POST /record 成功寫入,GET / 讀得回來', async () => { + const db = makeSqliteD1(); + const a = app(db); + const res = await a.fetch('/execution-log/record', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ workflow_id: 'wf-route', owner_id: 'ak1', verdict: 'success', duration_ms: 10, message: 'ok' }), + }); + expect(res.status).toBe(200); + const body = await res.json() as { success: boolean; written: boolean; mode: string }; + expect(body).toEqual({ success: true, written: true, mode: 'log' }); + + const listRes = await a.fetch('/execution-log?workflow_id=wf-route&owner_id=ak1'); + const listBody = await listRes.json() as { success: boolean; executions: unknown[] }; + expect(listBody.success).toBe(true); + expect(listBody.executions.length).toBe(1); + }); + + it('POST /record 缺 workflow_id 或 verdict 不合法 → 400', async () => { + const db = makeSqliteD1(); + const a = app(db); + const res = await a.fetch('/execution-log/record', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ verdict: 'maybe' }), + }); + expect(res.status).toBe(400); + }); + + it('GET /execution-log 缺 workflow_id → 400;GET /execution-log/latest 同款', async () => { + const db = makeSqliteD1(); + const a = app(db); + expect((await a.fetch('/execution-log')).status).toBe(400); + expect((await a.fetch('/execution-log/latest')).status).toBe(400); + }); +}); diff --git a/mcp/src/tools/arcrun_introspection.ts b/mcp/src/tools/arcrun_introspection.ts index f7ab4f8..26bfefd 100644 --- a/mcp/src/tools/arcrun_introspection.ts +++ b/mcp/src/tools/arcrun_introspection.ts @@ -158,7 +158,7 @@ export function registerGetExecutionTrace(server: McpServer, env: Env) { export function registerListRecentExecutions(server: McpServer, env: Env) { server.tool( toolName("list_recent_executions"), - "列某 workflow 最近 N 次執行 verdict(成功 / 失敗 / duration)。資料來源是 ANALYTICS_KV 90 天保留期。", + "列某 workflow 最近 N 次執行 verdict(成功 / 失敗 / duration)。資料來源是 D1 執行紀錄表(無固定保留期,但用量過大時系統會自動降成只記失敗、甚至暫停記錄——工作流本身執行不受影響)。", { api_key: z.string().describe(apiKeyDesc), workflow_name: z.string().describe("workflow 名稱(acr push 時的 name 欄)"),