feat(cypher-executor): 同步查詢 trigger + graph_neighbors 查詢面 workflow 示範
Part 1（框架，cypher-executor webhooks-named.ts）：加同步查詢 trigger。
現有 named webhook /trigger 回 {success,data,trace,duration_ms} 信封、只有 POST；
查詢面（console/MCP 打 graph neighbors/traverse）要 GET + 直接拿最終節點輸出當 response。
新端點同步 await 執行 workflow graph → 回 result.data（最終節點輸出）本身當 body（非 202）：
- GET /q/:ns/:name （namespace 走 path，input 走 query string）
- GET /webhooks/named/:name/query （X-Arcrun-API-Key header，input 走 query string）
- POST /webhooks/named/:name/query （header，input 走 body）
- POST /webhooks/named/:ns/:name/query （namespace 走 path，input 走 body）
認證沿用 X-Arcrun-API-Key。誠實（mindset §7）：節點失敗回 error+trace（500，非假綠）；
paused 工作流無法同步回答 → 409 明講；輸出 5 MiB 硬上限（超過 413）；duration 走 header 不污染 body。
Part 2（A 類 workflow.yaml）：registry/examples/graph-neighbors/。
http_request 打 base custom domain kbdb.finally.click（避 CF 1042）撈 triplet records
→ code 零件記憶體 BFS（對照 kbdb-graph-plugin graph-traverse.ts）→ 同步回鄰居。
把 graph plugin 內建 GET /graph/neighbors 泛化成查詢面 workflow 的示範。
測試：cypher-executor/tests/query-trigger.test.ts（7 測，全綠）——同步回輸出（非 202）、
GET/POST × header/path 四端點、節點失敗回錯+trace、缺 key 401、不存在 404。
用內建 comp_uppercase（純記憶體）證明 Part 1 同步 trigger 機制本身可用。
待驗：graph_neighbors 需 code 零件部署 leo21c 後才能 live 端到端（另線處理）；
triplet template id 上線前對一次（workflow 已參數化未寫死）。
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015d5jDbuqT5Htwv3Q88XXKk
This commit is contained in:
@@ -341,6 +341,136 @@ async function triggerNamed(
|
||||
return c.json(result, result.success ? 200 : 500);
|
||||
}
|
||||
|
||||
// ── 同步查詢 trigger（sync query）─────────────────────────────────────────────
|
||||
//
|
||||
// 動機：named webhook 的 /trigger 預設路徑雖已同步（await），但它回傳的是
|
||||
// `{ success, data, trace, duration_ms }` **信封**、且只有 POST 形態。
|
||||
// 查詢面（console / MCP 打 graph neighbors / traverse 之類）需要 request→response
|
||||
// **直接拿「工作流最終節點輸出」** 當 HTTP response body，且常是一個 GET。
|
||||
// 這組端點補上這個泛化：同步 await 執行 workflow graph → 把 **result.data（最終節點輸出）本身**
|
||||
// 當 response body 回（非 202、非信封）。補上它，任何查詢端點都能是一個 workflow。
|
||||
//
|
||||
// 認證：沿用 X-Arcrun-API-Key（header 形態）或 namespace 走 path（公開形態，與 /trigger path 版對稱；
|
||||
// self-hosted namespace 是明碼分區標籤非密碼，故可放 path — mindset §3 arcrun 不做授權判斷）。
|
||||
// 誠實（mindset §7）：節點失敗回錯誤 + trace 摘要（非把錯誤當輸出假綠）；paused 工作流無法同步
|
||||
// 給答案 → 明講（409），不假裝成功。
|
||||
|
||||
// 同步查詢輸出上限（防超大 response body 撐爆 Worker / 呼叫端）。
|
||||
// 超過 → 回 413 + 誠實錯誤（請在 workflow 內先聚合/分頁），不截斷假裝成功。
|
||||
const MAX_QUERY_OUTPUT_BYTES = 5 * 1024 * 1024; // 5 MiB
|
||||
|
||||
// GET：把 query string 全部欄位當 triggerContext（值皆 string）。
|
||||
function queryStringContext(c: Context<{ Bindings: Bindings }>): Record<string, unknown> {
|
||||
return { ...c.req.query() };
|
||||
}
|
||||
|
||||
// POST：body（JSON object）當 triggerContext；非物件 / 無 body → 空 context。
|
||||
async function bodyContext(c: Context<{ Bindings: Bindings }>): Promise<Record<string, unknown>> {
|
||||
const body = await c.req.json().catch(() => null);
|
||||
return body && typeof body === 'object' ? (body as Record<string, unknown>) : {};
|
||||
}
|
||||
|
||||
// 共用同步查詢邏輯（header 路徑與 path 路徑、GET 與 POST 都用，避免分叉）。
|
||||
async function queryNamed(
|
||||
c: Context<{ Bindings: Bindings }>,
|
||||
apiKey: string,
|
||||
name: string,
|
||||
triggerContext: Record<string, unknown>,
|
||||
) {
|
||||
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);
|
||||
}
|
||||
|
||||
const graph = record.graph as { id?: string; nodes?: unknown[] };
|
||||
const workflowId = graph.id ?? name;
|
||||
const nodes = Array.isArray(graph.nodes) ? (graph.nodes as GraphNode[]) : [];
|
||||
const userAgent = c.req.header('User-Agent') ?? undefined;
|
||||
|
||||
// 同步執行（await，非 waitUntil）：查詢端點必須 request→response 拿到結果。
|
||||
const result = await executeWebhookGraph(
|
||||
c.env,
|
||||
record.graph,
|
||||
triggerContext,
|
||||
name,
|
||||
apiKey,
|
||||
c.executionCtx,
|
||||
userAgent,
|
||||
);
|
||||
|
||||
// 執行判決寫入不阻塞回應（waitUntil，與 /trigger 一致）。
|
||||
c.executionCtx.waitUntil(
|
||||
writeExecutionVerdict(c.env, workflowId, nodes, result.success ? 'success' : 'failed', result.duration_ms, result.error ?? ''),
|
||||
);
|
||||
|
||||
if (!result.success) {
|
||||
// paused（如 claude_api 等外部 callback resume）無法同步給答案 → 明講，不假裝成功。
|
||||
const paused = typeof result.error === 'string' && /workflow paused/i.test(result.error);
|
||||
return c.json(
|
||||
{
|
||||
success: false,
|
||||
error: result.error ?? '工作流執行失敗',
|
||||
trace: result.trace,
|
||||
...(paused
|
||||
? { paused: true, hint: '此工作流會暫停等待非同步 callback，無法當同步查詢端點；改用 /webhooks/named/:name/trigger?async=1 + /workflows/resume。' }
|
||||
: {}),
|
||||
},
|
||||
paused ? 409 : 500,
|
||||
);
|
||||
}
|
||||
|
||||
// 成功 → 回「最終節點輸出」本身當 response body（非 202、非信封）。
|
||||
const serialized = JSON.stringify(result.data ?? null);
|
||||
const byteLen = new TextEncoder().encode(serialized).byteLength;
|
||||
if (byteLen > MAX_QUERY_OUTPUT_BYTES) {
|
||||
return c.json(
|
||||
{
|
||||
success: false,
|
||||
error: `查詢輸出過大（${byteLen} bytes > 上限 ${MAX_QUERY_OUTPUT_BYTES}）。請在 workflow 內先聚合 / 分頁再回。`,
|
||||
},
|
||||
413,
|
||||
);
|
||||
}
|
||||
return new Response(serialized, {
|
||||
status: 200,
|
||||
headers: {
|
||||
'Content-Type': 'application/json; charset=UTF-8',
|
||||
'X-Arcrun-Duration-Ms': String(result.duration_ms),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// GET /webhooks/named/:name/query — header 認證，input 走 query string（console/MCP 主用）
|
||||
webhooksNamedRouter.get('/webhooks/named/:name/query', async (c) => {
|
||||
const apiKey = c.req.header('X-Arcrun-API-Key');
|
||||
if (!apiKey) return c.json({ error: '缺少 X-Arcrun-API-Key header' }, 401);
|
||||
return queryNamed(c, apiKey, c.req.param('name'), queryStringContext(c));
|
||||
});
|
||||
|
||||
// POST /webhooks/named/:name/query — header 認證，input 走 body
|
||||
webhooksNamedRouter.post('/webhooks/named/:name/query', async (c) => {
|
||||
const apiKey = c.req.header('X-Arcrun-API-Key');
|
||||
if (!apiKey) return c.json({ error: '缺少 X-Arcrun-API-Key header' }, 401);
|
||||
return queryNamed(c, apiKey, c.req.param('name'), await bodyContext(c));
|
||||
});
|
||||
|
||||
// POST /webhooks/named/:ns/:name/query — namespace 走 path（公開查詢，與 /trigger path 版對稱）
|
||||
webhooksNamedRouter.post('/webhooks/named/:ns/:name/query', async (c) => {
|
||||
return queryNamed(c, c.req.param('ns'), c.req.param('name'), await bodyContext(c));
|
||||
});
|
||||
|
||||
// GET /q/:ns/:name — 簡短查詢入口（namespace 走 path，input 走 query string）
|
||||
webhooksNamedRouter.get('/q/:ns/:name', async (c) => {
|
||||
return queryNamed(c, c.req.param('ns'), c.req.param('name'), queryStringContext(c));
|
||||
});
|
||||
|
||||
// GET /webhooks/named — 列出當前 api_key 下所有 workflow
|
||||
webhooksNamedRouter.get('/webhooks/named', async (c) => {
|
||||
const apiKey = c.req.header('X-Arcrun-API-Key');
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
/**
|
||||
* 同步查詢 trigger（sync query）測試 — webhooks-named.ts 的 /query 與 /q/* 端點。
|
||||
*
|
||||
* 驗證重點（對應交辦 Part 1）：
|
||||
* 1. 同步回「最終節點輸出」本身（非 202、非 {success,data,...} 信封）— GET(query string) 與 POST(body) 皆是。
|
||||
* 2. namespace 走 path 的公開形態（/q/:ns/:name、POST /webhooks/named/:ns/:name/query）。
|
||||
* 3. 節點失敗 → 誠實回錯誤 + trace（500），非假綠。
|
||||
* 4. 認證：缺 X-Arcrun-API-Key → 401；workflow 不存在 → 404。
|
||||
*
|
||||
* 用內建零件（comp_uppercase / comp_passthrough，純記憶體、無外部 fetch）當「極簡 workflow」，
|
||||
* 因此本測試就是「Part 1 同步 trigger 機制本身可用」的證據（確實同步回值而非 202）。
|
||||
*/
|
||||
import { describe, it, expect, beforeAll } from 'vitest';
|
||||
import { env, SELF } from 'cloudflare:test';
|
||||
|
||||
const API_KEY = 'test-tenant-query';
|
||||
|
||||
function kvKey(name: string, apiKey = API_KEY): string {
|
||||
return `${apiKey}:wf:${name}`;
|
||||
}
|
||||
|
||||
// 極簡 workflow：單一 comp_uppercase 節點。caller input（text）→ 大寫 → 當最終輸出回。
|
||||
// 無 Input/Output 節點：最終節點輸出即 comp_uppercase 的回傳，乾淨可斷言。
|
||||
const UPPER_WF = {
|
||||
name: 'q_upper',
|
||||
graph: {
|
||||
id: 'q_upper',
|
||||
name: 'sync query upper',
|
||||
nodes: [{ id: 'upper', type: 'Component', componentId: 'comp_uppercase' }],
|
||||
edges: [],
|
||||
},
|
||||
description: '同步查詢：把 text 轉大寫回傳（測試用）',
|
||||
created_at: new Date().toISOString(),
|
||||
};
|
||||
|
||||
// 會失敗的 workflow：引用不存在的零件 → 節點執行失敗。
|
||||
const FAIL_WF = {
|
||||
name: 'q_fail',
|
||||
graph: {
|
||||
id: 'q_fail',
|
||||
name: 'sync query fail',
|
||||
nodes: [{ id: 'boom', type: 'Component', componentId: 'comp_does_not_exist' }],
|
||||
edges: [],
|
||||
},
|
||||
description: '同步查詢：故意引用不存在零件（測試失敗路徑）',
|
||||
created_at: new Date().toISOString(),
|
||||
};
|
||||
|
||||
beforeAll(async () => {
|
||||
await env.WEBHOOKS.put(kvKey(UPPER_WF.name), JSON.stringify(UPPER_WF));
|
||||
await env.WEBHOOKS.put(kvKey(FAIL_WF.name), JSON.stringify(FAIL_WF));
|
||||
});
|
||||
|
||||
describe('同步查詢 trigger — 成功回最終節點輸出（非 202、非信封）', () => {
|
||||
it('POST /webhooks/named/:name/query（header 認證，body input）同步回輸出', async () => {
|
||||
const res = await SELF.fetch('http://localhost/webhooks/named/q_upper/query', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', 'X-Arcrun-API-Key': API_KEY },
|
||||
body: JSON.stringify({ text: 'hello' }),
|
||||
});
|
||||
expect(res.status).toBe(200); // 關鍵：非 202
|
||||
const data = await res.json() as Record<string, unknown>;
|
||||
// 回的是「最終節點輸出本身」：頂層直接有 text=HELLO，而非包在 { data: ... } 信封裡
|
||||
expect(data.text).toBe('HELLO');
|
||||
expect(data).not.toHaveProperty('duration_ms'); // 信封欄位不該出現在 body
|
||||
// duration 走 header，不污染輸出
|
||||
expect(res.headers.get('X-Arcrun-Duration-Ms')).not.toBeNull();
|
||||
});
|
||||
|
||||
it('GET /webhooks/named/:name/query（header 認證，query string input）同步回輸出', async () => {
|
||||
const res = await SELF.fetch('http://localhost/webhooks/named/q_upper/query?text=world', {
|
||||
headers: { 'X-Arcrun-API-Key': API_KEY },
|
||||
});
|
||||
expect(res.status).toBe(200);
|
||||
const data = await res.json() as Record<string, unknown>;
|
||||
expect(data.text).toBe('WORLD');
|
||||
});
|
||||
|
||||
it('GET /q/:ns/:name（namespace 走 path，query string input）同步回輸出', async () => {
|
||||
const res = await SELF.fetch(`http://localhost/q/${API_KEY}/q_upper?text=abc`);
|
||||
expect(res.status).toBe(200);
|
||||
const data = await res.json() as Record<string, unknown>;
|
||||
expect(data.text).toBe('ABC');
|
||||
});
|
||||
|
||||
it('POST /webhooks/named/:ns/:name/query（namespace 走 path，body input）同步回輸出', async () => {
|
||||
const res = await SELF.fetch(`http://localhost/webhooks/named/${API_KEY}/q_upper/query`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ text: 'path' }),
|
||||
});
|
||||
expect(res.status).toBe(200);
|
||||
const data = await res.json() as Record<string, unknown>;
|
||||
expect(data.text).toBe('PATH');
|
||||
});
|
||||
});
|
||||
|
||||
describe('同步查詢 trigger — 誠實錯誤（不假綠）', () => {
|
||||
it('節點失敗 → 500 + error + trace（非把錯誤當輸出）', async () => {
|
||||
const res = await SELF.fetch('http://localhost/webhooks/named/q_fail/query', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', 'X-Arcrun-API-Key': API_KEY },
|
||||
body: JSON.stringify({}),
|
||||
});
|
||||
expect(res.status).toBe(500);
|
||||
const data = await res.json() as { success: boolean; error: string; trace: unknown };
|
||||
expect(data.success).toBe(false);
|
||||
expect(typeof data.error).toBe('string');
|
||||
expect(data.error.length).toBeGreaterThan(0);
|
||||
expect(data.trace).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('同步查詢 trigger — 認證與存在性', () => {
|
||||
it('缺 X-Arcrun-API-Key（header 形態）→ 401', async () => {
|
||||
const res = await SELF.fetch('http://localhost/webhooks/named/q_upper/query', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ text: 'x' }),
|
||||
});
|
||||
expect(res.status).toBe(401);
|
||||
});
|
||||
|
||||
it('workflow 不存在 → 404', async () => {
|
||||
const res = await SELF.fetch('http://localhost/webhooks/named/no_such_wf/query', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', 'X-Arcrun-API-Key': API_KEY },
|
||||
body: JSON.stringify({}),
|
||||
});
|
||||
expect(res.status).toBe(404);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user