/** * arcrun component loader * * 解析優先序: * * 0. trigger_workflow 內建 orchestration 零件(in-process call,繞 CF self-fetch 死鎖) * 1. 內建零件(BUILTIN_COMPONENTS)— 純 JS,最快 * 2. 外部 URL(https://...)— 直接 fetch,n8n/MCP/任何 HTTP 服務 * 3. cmp_xxxxxxxx hash → 查 WEBHOOKS KV idx → canonical_id → 邏輯 Worker * 4. rec_xxxxxxxx hash → 查 RECIPES KV idx → recipe 執行 * 5. 邏輯零件 canonical_id → Service Binding(同帳號不走公網) * 5.5. Auth recipe(平台預建)→ Auth Recipe Runner * 6. KV recipe canonical_id → 從 RECIPES KV 讀取 recipe → fetch 外部 API * 7. WASM HTTP runner(auth primitive / API 零件 → 獨立 Worker URL) * 8. 找不到 → 報錯 */ import { BUILTIN_COMPONENTS } from './constants'; import { isComponentHash, isRecipeHash } from './hash'; import { resolveRecipe, resolveAuthRecipe } from '../routes/recipes'; import type { AuthRecipeDefinition } from '../routes/recipes'; import type { Bindings, ComponentRunner, ServiceBinding } from '../types'; /** * WASM HTTP runner:canonical_id → 對應獨立 Worker URL。 * * 所有 WASM 零件(auth primitive / API 零件 / 未來用戶自製)都是獨立部署的 Worker, * 以 `{canonical-id-kebab}.arcrun.dev` 為 URL 慣例。cypher-executor 不做 WASM * instantiate,只做 HTTP fetch。這層是 API 零件(及 auth primitive)的唯一入口。 * * R2 動態注入 WASM 路徑作廢(CF workerd 不支援以 R2 物件臨時 instantiate)。 */ // TODO(架構債,2026-05-07):白名單寫死違反 arcrun 「新零件無需改 cypher-executor」承諾 // 應改為從 component-registry KV 動態查(registry 已有 backfill index,知道所有 canonical_id) // SDD 待開:cypher-executor-dynamic-component-discovery const WASM_HTTP_RUNNER_IDS: ReadonlySet = new Set([ // 通用 HTTP 零件 'http_request', // 通用 code 零件(sandbox inline JS,Arcrun#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 已降級為 recipe(2026-05-29 Phase 2): // recipe:gmail_send / telegram_send / line_notify_send / google_sheets_read|append // 走 step 6 KV recipe 解析,不再是零件。零件目錄已刪。 'cron', // Auth primitives 'auth_static_key', 'auth_service_account', 'auth_oauth2', 'auth_mtls', ]); /** * canonical_id → component worker URL(走 workers.dev 子域,避開同 zone 自循環死鎖) * * 為何不用 *.arcrun.dev:cypher-executor 本身綁 cypher.arcrun.dev/*, * fetch 同 zone *.arcrun.dev 會撞 CF 的 zone 自循環防護回 522。 * 詳見 arcrun.md P0 #9(2026-05-13)。 * * subdomain 來自 wrangler.toml [vars] WORKER_SUBDOMAIN(預設 uncle6-me,self-hosted fork 改自己的)。 */ export function wasmWorkerUrl(canonicalId: string, subdomain: string): string { const kebab = canonicalId.replace(/_/g, '-'); // 平台慣例:component worker 名稱 = `arcrun-{kebab}`(見 rule 03 / rule 05), // 例如 canonical_id=http_request → worker 名 arcrun-http-request → URL arcrun-http-request.{subdomain}.workers.dev return `https://arcrun-${kebab}.${subdomain}.workers.dev`; } /** 邏輯零件 canonical_id → Service Binding key */ const LOGIC_BINDING_MAP: Record = { if_control: 'SVC_IF_CONTROL', switch: 'SVC_SWITCH', foreach_control: 'SVC_FOREACH_CONTROL', filter: 'SVC_FILTER', merge: 'SVC_MERGE', try_catch: 'SVC_TRY_CATCH', wait: 'SVC_WAIT', set: 'SVC_SET', array_ops: 'SVC_ARRAY_OPS', string_ops: 'SVC_STRING_OPS', number_ops: 'SVC_NUMBER_OPS', date_ops: 'SVC_DATE_OPS', validate_json: 'SVC_VALIDATE_JSON', // ai_transform_compile / ai_transform_run 已刪除(2026-05-29): // Arcrun 是 AI 呼叫的工具,工作流不該內嵌 AI 節點回頭呼叫 AI(n8n 才需要,因它沒大腦)。 }; export function createComponentLoader(env: Bindings) { return async (componentId: string): Promise => { // 0. 平台內建 orchestration 零件(需要 env / 跨 workflow 能力) // 這類零件「是 orchestrator 的職責」(不是業務邏輯),故不違反「業務邏輯走 WASM」規則。 // 目前只有 trigger_workflow:用 in-process call 觸發另一個 named workflow, // 繞掉 CF 同 zone self-fetch 死鎖(避免 cypher-executor 自打 http_request → 1042)。 if (componentId === 'trigger_workflow') { return makeTriggerWorkflowRunner(env); } // 1. 內建零件(純 JS,最優先) const builtin = BUILTIN_COMPONENTS.get(componentId); if (builtin) return builtin; // 2. 外部 URL if (componentId.startsWith('http://') || componentId.startsWith('https://')) { return makeHttpRunner(componentId); } // 3. cmp_hash → 查 WEBHOOKS KV idx → canonical_id → 邏輯 Worker if (isComponentHash(componentId)) { const canonicalId = await env.WEBHOOKS.get(`idx:${componentId}`); if (canonicalId) { const runner = makeLogicRunner(canonicalId, env); if (runner) return runner; } throw new Error(`找不到零件 hash "${componentId}",請確認已透過 acr push 上傳`); } // 4. rec_hash → 查 RECIPES KV idx → recipe 執行 if (isRecipeHash(componentId)) { const recipe = await resolveRecipe(componentId, env.RECIPES); if (recipe) return makeRecipeRunner(recipe); throw new Error(`找不到 recipe hash "${componentId}",請確認已透過 acr push 上傳`); } // 5. 邏輯零件 canonical_id → Service Binding const logicRunner = makeLogicRunner(componentId, env); if (logicRunner) return logicRunner; // 5.5 Auth recipe(平台預建,auth_recipe:{service} in RECIPES KV) const authRecipe = await resolveAuthRecipe(componentId, env.RECIPES); if (authRecipe) return makeAuthRecipeRunner(authRecipe); // 6. KV recipe(動態,用戶 push 的) const kvRecipe = await resolveRecipe(componentId, env.RECIPES); if (kvRecipe) return makeRecipeRunner(kvRecipe); // 7. WASM HTTP runner:auth primitive / API 零件 → 獨立 Worker URL // 白名單見 WASM_HTTP_RUNNER_IDS(http_request、5 個待降級 API 零件、4 個 auth primitive)。 // 對應 Worker 部署於 arcrun-{canonical-id-kebab}.{WORKER_SUBDOMAIN}.workers.dev // (見 P0 #9 / rule 03)。 if (WASM_HTTP_RUNNER_IDS.has(componentId)) { return makeHttpRunner(wasmWorkerUrl(componentId, env.WORKER_SUBDOMAIN)); } // 8. 找不到 throw new Error( `找不到零件 "${componentId}"。\n` + `邏輯零件:${Object.keys(LOGIC_BINDING_MAP).join(', ')}\n` + `或傳入外部 URL(https://...)、recipe hash(rec_xxxxxxxx)、零件 hash(cmp_xxxxxxxx)` ); }; } // ── 執行器工廠 ──────────────────────────────────────────────────────────────── /** * trigger_workflow 內建 orchestration 零件 * * 用途:在 workflow A 內 in-process 觸發 workflow B,繞 CF 同 zone self-fetch 死鎖。 * * 動機:mira_feed_watcher 之前用 http_request 自打 cypher.arcrun.dev → CF 1042。 * 就算改打 arcrun-cypher-executor.{subdomain}.workers.dev,Worker → 自身 URL 仍 * 被 CF 「self subrequest」防護擋(即使 hostname 不同)。 * 改用 in-process call executeWebhookGraph 徹底繞掉外部 HTTP。 * * 不違反「業務邏輯走 WASM」鐵律:trigger_workflow 是 orchestrator 自己的 routing 能力 * (像 CALLS_SUBFLOW),不是業務邏輯(不解密 / 不簽 JWT / 不打外部 API)。 * * Input ctx: * - workflow_name: string (必填,目標 workflow 名稱) * - api_key: string (必填,KV 查 key prefix) * - input: object (可選,傳給子 workflow 當 triggerContext) * - wait: boolean (預設 true,await 完成;false = fire-and-forget 用 waitUntil) * * 動態 import webhook-handlers 避循環依賴(webhook-handlers → component-loader → 自己)。 */ function makeTriggerWorkflowRunner(env: Bindings): ComponentRunner { return async (ctx: unknown) => { const c = (ctx && typeof ctx === 'object') ? ctx as Record : {}; const workflowName = String(c.workflow_name ?? ''); const apiKey = String(c.api_key ?? ''); const input = (c.input && typeof c.input === 'object') ? c.input as Record : {}; const wait = c.wait !== false; // 預設 true 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 }; 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'); const triggerContext = { ...input, _triggered_by: 'trigger_workflow' }; if (wait) { const r = await executeWebhookGraph(env, record.graph, triggerContext, workflowName, apiKey); // paused 是預期狀態(claude_api 等待外部 callback resume),不算失敗 // executeWebhookGraph 內部把 ExecutionError + "paused at node X" 包成 success:false + 含 error 字串 // // 2026-05-16 rename per LI roadmap (block e924c231) 自評建議: // 舊 `paused_awaiting_resume` 容易被誤讀成「掛起出問題」 // 新 `running_async` 強調「已接受,繼續在背景跑」— 行為一致,命名更清楚 const isPaused = !r.success && typeof r.error === 'string' && /workflow paused/i.test(r.error); return { success: r.success || isPaused, triggered_workflow: workflowName, status: r.success ? 'completed' : (isPaused ? 'running_async' : 'failed'), sub_result: r, }; } else { // fire-and-forget — 不 await,但因為沒拿到 ctx.waitUntil,這裡 promise 可能被 cancel // 目前不啟用,留 wait=true 為預設。未來想要 fire-and-forget 需 plumb ExecutionContext void executeWebhookGraph(env, record.graph, triggerContext, workflowName, apiKey) .catch((e) => console.error('[trigger_workflow] fire-and-forget fail', workflowName, e)); return { success: true, triggered_workflow: workflowName, mode: 'fire_and_forget' }; } }; } function makeHttpRunner(url: string): ComponentRunner { return async (ctx: unknown) => { const res = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(ctx), }); if (!res.ok) { const text = await res.text(); return { success: false, status: res.status, error: text.slice(0, 200) }; } try { return await res.json(); } catch { return { success: true, data: await res.text() }; } }; } function makeLogicRunner(canonicalId: string, env: Bindings): ComponentRunner | null { const bindingKey = LOGIC_BINDING_MAP[canonicalId]; if (!bindingKey) return null; const svc = env[bindingKey] as ServiceBinding | undefined; if (svc) { return async (ctx: unknown) => { const res = await svc.fetch(new Request('https://component/', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(ctx), })); if (!res.ok) { const text = await res.text(); return { success: false, error: `${canonicalId} 回傳 ${res.status}: ${text.slice(0, 200)}` }; } try { return await res.json(); } catch { return { success: false, error: `${canonicalId} 回傳非 JSON` }; } }; } // Service Binding 未配置時 fallback 到公網(自製零件 or 開發環境) // 走 workers.dev 子域避開同 zone 死鎖(P0 #9) return makeHttpRunner(wasmWorkerUrl(canonicalId, env.WORKER_SUBDOMAIN)); } function makeRecipeRunner(recipe: import('../routes/recipes').RecipeDefinition): ComponentRunner { return async (ctx: unknown) => { const ctxObj = (ctx && typeof ctx === 'object') ? ctx as Record : {}; // 模板替換:{{key}} 從 ctx 取;{{auth.K}} 從 _auth_path 取 // (_auth_path 由 auth primitive 解密後注入,供 URL path 用,如 telegram /bot{{auth.token}}/) const authPath = (ctxObj._auth_path as Record) ?? {}; const interpolate = (s: string) => s.replace(/\{\{(auth\.)?(\w+)\}\}/g, (_, authPrefix, k) => String(authPrefix ? (authPath[k] ?? '') : (ctxObj[k] ?? '')), ); const method = (recipe.method ?? 'POST').toUpperCase(); const authHeaders = (ctxObj._auth_headers as Record) ?? {}; const headers: Record = { 'Content-Type': 'application/json', ...authHeaders, }; for (const [k, v] of Object.entries(recipe.headers ?? {})) { headers[k] = interpolate(v); } // body:把 recipe.body 裡的 {{key}} 都換掉 let bodyStr: string | undefined; if (recipe.body) { bodyStr = interpolate(JSON.stringify(recipe.body)); } else if (method !== 'GET') { // 沒指定 body template → 用 ctx 當 body,但剔除 _ 前綴的內部欄位 // (_path / _auth_headers / _auth_query / _auth_body 不該漏進下游請求) const bodyObj = Object.fromEntries( Object.entries(ctxObj).filter(([k]) => !k.startsWith('_')), ); bodyStr = JSON.stringify(bodyObj); } const res = await fetch(interpolate(recipe.endpoint), { method, headers, body: bodyStr, }); const data = await readBodyOnce(res); return { success: res.ok, status: res.status, data }; }; } // ── Auth Recipe Runner ──────────────────────────────────────────────────────── // // credential-injector 已先將認證資訊注入為 _auth_headers / _auth_query / _auth_body。 // 這裡只需要讀取這些欄位,合併進 fetch,再清除 _auth_* 不傳給下游。 function makeAuthRecipeRunner(recipe: AuthRecipeDefinition): ComponentRunner { return async (ctx: unknown) => { const ctxObj = (ctx && typeof ctx === 'object') ? ctx as Record : {}; const authHeaders = (ctxObj._auth_headers as Record) ?? {}; const authQuery = (ctxObj._auth_query as Record) ?? {}; // _path 讓呼叫者指定 endpoint 後綴(e.g. /pages, /messages),可選 const path = typeof ctxObj._path === 'string' ? ctxObj._path : ''; const method = ((ctxObj.method as string) ?? 'POST').toUpperCase(); const url = new URL(recipe.base_url.replace(/\/$/, '') + path); for (const [k, v] of Object.entries(authQuery)) { url.searchParams.set(k, v); } const headers: Record = { 'Content-Type': 'application/json', ...authHeaders, }; // body:剔除所有 _ 前綴的內部欄位,以及 method const bodyObj = Object.fromEntries( Object.entries(ctxObj).filter(([k]) => !k.startsWith('_') && k !== 'method'), ); const res = await fetch(url.toString(), { method, headers, body: method !== 'GET' ? JSON.stringify(bodyObj) : undefined, }); const data = await readBodyOnce(res); return { success: res.ok, status: res.status, data }; }; } // 讀 response body 一次:先取 text,再嘗試 parse JSON。 // 不可用 `res.json().catch(() => res.text())` —— res.json() 失敗時 body 已被消費, // 第二次讀會丟 "Body has already been used"。 async function readBodyOnce(res: Response): Promise { const text = await res.text(); try { return JSON.parse(text); } catch { return text; } }