Merge branch 'main' into feat/deploy-code-component

This commit is contained in:
Leo
2026-07-07 09:39:20 +00:00
8 changed files with 516 additions and 0 deletions
@@ -36,6 +36,11 @@ import type { Bindings, ComponentRunner, ServiceBinding } from '../types';
const WASM_HTTP_RUNNER_IDS: ReadonlySet<string> = new Set([
// 通用 HTTP 零件
'http_request',
// 通用 code 零件(sandbox inline JSArcrun#10 / 07-thin-shell §3.5 code-node):獨立 Worker
// URL 走 wasmWorkerUrl 通用推導(arcrun-code.{WORKER_SUBDOMAIN}.workers.dev
// self-hosted 由 WORKER_SUBDOMAIN var 注入自己的 subdomain,無寫死官方域名)。
// 漏這行 = workflow 寫 `component: code` 落到 step 8 直接「找不到零件」(#29 發現)。
'code',
// gmail / telegram / line_notify / google_sheets 已降級為 recipe2026-05-29 Phase 2):
// recipe:gmail_send / telegram_send / line_notify_send / google_sheets_read|append
// 走 step 6 KV recipe 解析,不再是零件。零件目錄已刪。
@@ -341,6 +341,136 @@ async function triggerNamed(
return c.json(result, result.success ? 200 : 500);
}
// ── 同步查詢 triggersync 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-Keyheader 形態)或 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() };
}
// POSTbodyJSON 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 stringconsole/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 走 pathinput 走 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,56 @@
/**
* component-loader`code` 零件解析測試(#29 發現的白名單缺口)。
*
* 背景:workflow 寫 `component: code` 時,component-loader 解析鏈 step 1-6 都不命中
* (非 builtin / 非 URL / 非 hash / 非邏輯零件 binding / 非 recipe),必須靠 step 7 的
* WASM_HTTP_RUNNER_IDS 白名單 → wasmWorkerUrl 通用推導。白名單漏 `code` 就會落到
* step 8 直接丟「找不到零件」——graph_neighbors 等用 code 節點的 workflow 全滅。
*
* 驗證:
* 1. `code` 能被解析成 runner(不 throw)。
* 2. runner 打的 URL = arcrun-code.{WORKER_SUBDOMAIN}.workers.dev(通用推導、無寫死
* 官方 code.arcrun.dev → self-hosted 換 WORKER_SUBDOMAIN 即落地)。
*/
import { describe, it, expect, vi, afterEach } from 'vitest';
import { env } from 'cloudflare:test';
import { createComponentLoader, wasmWorkerUrl } from '../src/lib/component-loader';
import type { Bindings } from '../src/types';
afterEach(() => {
vi.unstubAllGlobals();
});
describe('component-loadercode 零件解析(WASM HTTP runner 白名單)', () => {
it('wasmWorkerUrl 通用推導:code → arcrun-code.{subdomain}.workers.devself-hosted 換 subdomain 即成立)', () => {
expect(wasmWorkerUrl('code', 'uncle6-me')).toBe('https://arcrun-code.uncle6-me.workers.dev');
expect(wasmWorkerUrl('code', 'my-selfhosted-sub')).toBe('https://arcrun-code.my-selfhosted-sub.workers.dev');
});
it('component: code 能被解析成 runner,且 fetch 打 subdomain 推導的 URL', async () => {
const fakeEnv = {
...env,
WORKER_SUBDOMAIN: 'test-sub',
} as unknown as Bindings;
const loader = createComponentLoader(fakeEnv);
// 白名單漏 `code` 時這裡就 throw「找不到零件 "code"」(#29 的故障型態)
const runner = await loader('code');
expect(typeof runner).toBe('function');
// stub fetch 捕 URL:證明 runner 打的是通用推導 URL,非寫死官方域名
const calledUrls: string[] = [];
const fetchSpy = vi.fn(async (url: unknown) => {
calledUrls.push(String(url));
return new Response(JSON.stringify({ success: true, data: { ok: true } }), {
status: 200,
headers: { 'Content-Type': 'application/json' },
});
});
vi.stubGlobal('fetch', fetchSpy);
const result = await runner({ code: 'return {ok:true}', input: {} }) as Record<string, unknown>;
expect(fetchSpy).toHaveBeenCalledTimes(1);
expect(calledUrls[0]).toBe('https://arcrun-code.test-sub.workers.dev');
expect(result.success).toBe(true);
});
});
+132
View File
@@ -0,0 +1,132 @@
/**
* 同步查詢 triggersync 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 → 401workflow 不存在 → 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 inputtext)→ 大寫 → 當最終輸出回。
// 無 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/queryheader 認證,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/queryheader 認證,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/:namenamespace 走 pathquery 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/querynamespace 走 pathbody 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-Keyheader 形態)→ 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);
});
});
+1
View File
@@ -29,6 +29,7 @@
| `daily-digest` | cron → 多源聚合(KBDB / RSS / 等) → 推送 |
| `parallel-fanout` | 一份輸入分發多 workflow 並行處理 |
| `error-retry` | try_catch + wait + retry 重試外部 API |
| `graph-neighbors` | 同步查詢:撈 KBDB triplet → 記憶體 BFS 找 N 跳鄰居(查詢面工作流,走同步查詢 trigger) |
## 如何用(AI 視角)
@@ -0,0 +1,67 @@
# graph-neighbors
## 解決什麼問題
把 graph plugin 內建的 `GET /graph/neighbors`(同步 request→response 拿鄰居)**改寫成一條 workflow**。
示範「查詢面工作流」:查詢端點不必是框架寫死的 route,可以是一條 `workflow.yaml` —— 撈 triplet
記錄 → 記憶體 BFS → 同步回鄰居。任何唯讀查詢都能這樣泛化成 workflow。
## 依賴的框架能力(本 PR 補上的)
**同步查詢 trigger**cypher-executor `webhooks-named.ts`):現有 named webhook 的 `/trigger`
回的是 `{success,data,trace,duration_ms}` 信封、且只有 POST。查詢面要 **GET + 直接拿最終節點輸出**
新端點同步 `await` 執行 graph → 把 `result.data`(最終節點輸出)本身當 response body 回(非 202):
- `GET /q/:ns/:name`namespace 走 pathinput 走 query string
- `GET /webhooks/named/:name/query`X-Arcrun-API-Key headerinput 走 query string
- `POST /webhooks/named/:name/query`headerinput 走 body
- `POST /webhooks/named/:ns/:name/query`namespace 走 pathinput 走 body
## 怎麼觸發
```bash
# GET(最像原本的 /graph/neighbors)——{你的-cypher-domain}/{你的-kbdb-domain} 換成自己的部署
curl "https://{你的-cypher-domain}/q/{namespace}/graph_neighbors?node=Arcrun&depth=2&template=graph_triplet&namespace={namespace}&kbdb_base=https://{你的-kbdb-domain}"
# POSTheader 認證)
curl -X POST https://{你的-cypher-domain}/webhooks/named/graph_neighbors/query \
-H "X-Arcrun-API-Key: {namespace}" \
-d '{"node":"Arcrun","depth":2,"template":"graph_triplet","namespace":"{namespace}","kbdb_base":"https://{你的-kbdb-domain}"}'
```
> ⚠️ **敏感輸入請用 `POST /query`**`GET /q/` 的參數走 query string,會進 CF / proxy / access log
> 各層日誌;node 名、namespace 等若屬敏感,改走 POST body。
回傳(最終節點輸出本身,非信封):
```json
{ "success": true, "start": "Arcrun", "depth": 2, "directed": false,
"neighbors": [ { "node": "cypher-executor", "predicate": "包含", "from": "Arcrun", "depth": 1 } ],
"count": 1 }
```
## 參數
- `node`(必填):BFS 起點節點名
- `depth`(預設 1):最大跳數
- `template`(必填):triplet 記錄的 base template id(⚠️ 以實際部署的 kbdb-graph-plugin triplet template 為準)
- `namespace`(必填):租戶 owner_idself-hosted 明碼 namespace
- `kbdb_base`(必填):**你自己的 KBDB base URL**(如 `https://kbdb.example.com`)。
workflow 不寫死任何一家的庫——抄示範時帶錯(或照抄別人的值)=查詢與資料流向直接打進別人的庫
KBDB_BASE_URL fallback 同家族坑,勿重蹈)。
- `directed`(預設 false):`true` 只走 subject→object;否則把 triplet 當雙向邊(無向鄰居)
## kbdb_base 該帶哪種 URL1042
cypher-executor 對 kbdb 發 fetch,若打同 zone URL 會踩 CF 1042same-zone self-fetch)。兩條路皆可:
1. 帶 KBDB 的 **custom domain**(跨 zone、走公網前門,天然避開 1042);
2. self-hosted 有 **`global_fetch_strictly_public`** compatibility flagcredential-primitives-wasm
Phase 7cypher wrangler.toml),開了之後 **workers.dev URL 亦可**直接帶。
## ⚠️ 尚未 live 驗(待辦)
- **`code` 零件尚未部署到 leo21c**(另線處理)→ 本工作流無法端到端 live 跑。
workflow.yaml 已寫好放這裡待驗。
- **同步查詢 trigger 本身已驗**`cypher-executor/tests/query-trigger.test.ts`7 測)用內建零件
comp_uppercase,純記憶體、無外部 fetch)證明「確實同步回最終節點輸出而非 202」。
- 上線前另需對一次實際 triplet template id(本檔用 `{{input.template}}` 參數化,未寫死)。
## 對照
記憶體 BFS 對照 `kbdb-graph-plugin``graph-traverse.ts:23-51`triplet 當有向邊
`subject --predicate--> object`,從起點逐跳擴張到 depth 上限,收集首次訪到的節點當鄰居。
## 學到什麼
- 查詢端點可以是 workflow,不必是框架寫死的 route(同步查詢 trigger 讓這件事成立)
- `code` 零件(sandbox inline JS)承載「非 call-api 的純計算」(BFS),不必為此鑄 domain 零件
- 單一 `{{ref}}` pass-through 保留陣列型別 → `{{fetch_triplets.data.records}}` 拿到真陣列餵給 code
@@ -0,0 +1 @@
["graph", "triplet", "bfs", "neighbors", "kbdb", "sync-query", "query-endpoint", "code-node", "common-pattern"]
@@ -0,0 +1,124 @@
name: graph_neighbors
description: >
同步查詢:給一個節點 → 從 KBDB triplet 記錄建鄰接表 → 記憶體 BFS 找 N 跳鄰居 → 同步回鄰居清單。
這是「查詢面工作流」示範:用同步查詢 triggerGET /q/:ns/graph_neighbors 或
POST /webhooks/named/:name/query),把 workflow 最終節點輸出直接當 HTTP response 拿回,
取代 graph plugin 內建的 GET /graph/neighbors。對照 kbdb-graph-plugin graph-traverse.ts 的記憶體 BFS。
# ══════════════════════════════════════════════════════════════════════════════
# ⚠️ 先改成你自己的 KBDB URLself-hosted 必看)
# fetch_triplets 的 base URL 走 {{input.kbdb_base}} 參數(每次呼叫帶),**沒有寫死官方庫**。
# 照抄本示範時,把觸發參數 kbdb_base 換成你自己部署的 KBDB 對外 URL——
# 抄了別人的值=你的查詢與資料流向直接打進別人的庫(KBDB_BASE_URL fallback 同家族坑)。
# ══════════════════════════════════════════════════════════════════════════════
# ── 1042 與 kbdb_base 該帶哪種 URL ──
# cypher-executor 執行 http_request 節點時對 kbdb 發 fetch。若打同 zone URL 會踩 CF 1042
# same-zone self-fetch)。兩條路皆可:
# a) 帶 KBDB 的 custom domain(跨 zone、走公網前門,天然避開 1042);
# b) self-hosted 有 `global_fetch_strictly_public` compatibility flag
# credential-primitives-wasm Phase 7cypher wrangler.toml),開了之後
# workers.dev URL 亦可直接帶。
# ── 觸發(同步查詢 trigger,非 202)──
# GET https://{你的-cypher-domain}/q/{namespace}/graph_neighbors?node=Arcrun&depth=2&template=graph_triplet&namespace={namespace}&kbdb_base=https://{你的-kbdb-domain}
# POST https://{你的-cypher-domain}/webhooks/named/graph_neighbors/query
# -H "X-Arcrun-API-Key: {namespace}"
# -d '{"node":"Arcrun","depth":2,"template":"graph_triplet","namespace":"{namespace}","kbdb_base":"https://{你的-kbdb-domain}"}'
# → 直接回 { success, start, depth, directed, neighbors:[...], count }(最終節點輸出本身)。
# (敏感輸入用 POST /queryGET /q/ 的 query string 會進各層 log,見 description.md。)
flow:
- "input >> ON_SUCCESS >> fetch_triplets"
- "fetch_triplets >> ON_SUCCESS >> bfs_neighbors"
config:
# 1) 撈本租戶的 triplet 記錄。triplet = base 萬用表的一個 templategraph plugin 寫入),
# slots = subject / predicate / object。base 端點:GET /records/by-template/:template?owner_id=
# 回 { success, records:[{ record_id, values:{subject,predicate,object} }], count }。
# ⚠️ template 名(此處 {{input.template}},預設由呼叫者帶 graph_triplet)以實際部署的
# kbdb-graph-plugin triplet template id 為準——上線前對一次。
# ⚠️ base URL 由呼叫者帶({{input.kbdb_base}})——不寫死任何一家的庫(見檔頂警示)。
fetch_triplets:
component: http_request
method: GET
url: "{{input.kbdb_base}}/records/by-template/{{input.template}}?owner_id={{input.namespace}}"
headers:
Accept: "application/json"
# 2) ★ 記憶體 BFS(通用 code 零件,sandbox inline JS,無 LLM、無 fs/網路,stdin→stdout JSON)。
# 對照 kbdb-graph-plugin graph-traverse.ts:23-51 的記憶體 BFS:把 triplet 當有向邊
# subject --predicate--> object 建鄰接表,從 start 逐跳擴張到 depth 上限,收集新訪節點當鄰居。
# directed=false(預設)時把邊當雙向(無向圖鄰居);directed=true 只走 subject→object。
bfs_neighbors:
component: code
code: |
// graph_neighbors — 記憶體 BFS 找 N 跳鄰居(純函式、決定性、零 token)。
// input(由下方 input: 映射解析後注入):
// records[] : triplet 記錄({ values:{subject,predicate,object} } 或扁平 {subject,predicate,object}
// start : 起點節點名(字串)
// depth : 最大跳數(字串或數字,來自 query string 時是字串)
// directed : "true" 只走 subject→object;否則當無向
const records = Array.isArray(input.records) ? input.records : [];
const start = String(input.start == null ? '' : input.start);
const maxDepth = Math.max(1, parseInt(String(input.depth == null ? 1 : input.depth), 10) || 1);
const directed = String(input.directed == null ? '' : input.directed) === 'true';
if (!start) {
return { success: false, error: 'graph_neighbors 缺 startnode)參數' };
}
// 建鄰接表:subject --predicate--> object。無向時同時加反向邊。
const adj = new Map();
function addEdge(from, to, predicate) {
if (!adj.has(from)) adj.set(from, []);
adj.get(from).push({ node: to, predicate: predicate });
}
for (const r of records) {
const v = (r && typeof r === 'object' && r.values && typeof r.values === 'object') ? r.values : r;
if (!v || typeof v !== 'object') continue;
const s = v.subject, p = v.predicate, o = v.object;
if (!s || !o) continue;
addEdge(s, o, p);
if (!directed) addEdge(o, s, p);
}
// BFS:一層一跳,收集首次訪到的節點當鄰居(記 depth / 來源 / 關係)。
const visited = new Set([start]);
let frontier = [start];
const neighbors = [];
for (let d = 1; d <= maxDepth; d++) {
const next = [];
for (const cur of frontier) {
const outs = adj.get(cur) || [];
for (const e of outs) {
if (visited.has(e.node)) continue;
visited.add(e.node);
neighbors.push({ node: e.node, predicate: e.predicate, from: cur, depth: d });
next.push(e.node);
}
}
frontier = next;
if (frontier.length === 0) break;
}
return {
success: true,
start: start,
depth: maxDepth,
directed: directed,
neighbors: neighbors,
count: neighbors.length,
};
# input 映射:{{...}} 對 workflow context 展開後注入 code 沙箱的 `input` 變數。
# {{input.X}} 的 input = 上游 input 節點輸出(=觸發 context);{{fetch_triplets.data.records}}
# = http_request 回應 body 的 records 陣列(單一 ref pass-through 保留陣列型別)。
input:
records: "{{fetch_triplets.data.records}}"
start: "{{input.node}}"
depth: "{{input.depth}}"
directed: "{{input.directed}}"
limits:
timeout_ms: 3000 # 純 CPU BFS,充裕
max_output_bytes: 2097152 # 鄰居清單上限 2 MiB(呼叫端同步查詢輸出也有 5 MiB 硬上限)