arcrun — AI workflow execution engine (clean history)
Self-hosted 開源：WASM 零件 + recipe + cypher-executor，跑在你自己的 Cloudflare。 此為重建的乾淨歷史起點（移除曾誤 commit 的 GCP SA 金鑰，舊歷史保留在 richblack/arcrun 與本地 backup 分支）。含： - acr init --self-hosted installer（建 KV/R2 + codeload 拉預編譯 wasm + wrangler deploy + seed recipe） - recipe push 把關（資料外流提醒 + 打通檢查） - 19 個正當零件預編譯 wasm（claude_api/km_writer/kbdb_upsert_block 排除：違反 DECISIONS §1） - CLI / cypher-executor / registry / 完整 SDD Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
uncle6me-web
@@ -0,0 +1,107 @@
|
||||
/**
|
||||
* Auth Dispatcher
|
||||
*
|
||||
* 對需要認證的零件,在執行前 HTTP POST 到對應的 auth primitive Worker,
|
||||
* 取回 auth_headers / auth_query / auth_body 合併進節點 context。
|
||||
*
|
||||
* 嚴格邊界(rule 02 §2.2):
|
||||
* - 本檔**不做**任何 credential 解密 / template 展開 / JWT 簽章
|
||||
* - 那些全部在 auth primitive WASM 零件內執行(透過 host function `crypto_decrypt` 等)
|
||||
* - 本檔只做「查 recipe 決定走哪個 primitive Worker」+「HTTP fetch 取回注入結果」
|
||||
*
|
||||
* 目前階段接上 `auth_static_key` + `auth_service_account` + `auth_oauth2`,
|
||||
* Phase 4 剩 `auth_mtls`（mTLS handshake 在 Worker runtime 層）。
|
||||
*
|
||||
* 執行時機:graph-executor 在節點 runner 執行前呼叫,取回的 ctx 會:
|
||||
* 1. 先試本 dispatcher(命中才 return enriched ctx)
|
||||
* 2. 沒命中 fallback 到 `injectCredentials`(Phase 1.9 才刪除)
|
||||
*/
|
||||
|
||||
import type { Bindings } from '../types';
|
||||
import { resolveAuthRecipe, resolveRecipe } from '../routes/recipes';
|
||||
import { wasmWorkerUrl } from '../lib/component-loader';
|
||||
|
||||
/** 對應 Phase 1-4 會部署的 auth primitive Worker */
|
||||
const SUPPORTED_PRIMITIVES = new Set(['static_key', 'service_account', 'oauth2']);
|
||||
|
||||
/** auth primitive 本身的 componentId(避免自引用) */
|
||||
const AUTH_PRIMITIVE_IDS = new Set([
|
||||
'auth_static_key',
|
||||
'auth_service_account',
|
||||
'auth_oauth2',
|
||||
'auth_mtls',
|
||||
]);
|
||||
|
||||
/**
|
||||
* 試著對零件做 auth 注入。
|
||||
* - 命中(有對應 auth recipe 且 primitive 已支援)→ 回傳注入後的 ctx
|
||||
* - 未命中 → 回傳 null(呼叫端繼續跑舊路徑)
|
||||
*/
|
||||
export async function tryAuthDispatch(
|
||||
componentId: string,
|
||||
input: Record<string, unknown>,
|
||||
env: Bindings,
|
||||
apiKey: string,
|
||||
): Promise<Record<string, unknown> | null> {
|
||||
if (AUTH_PRIMITIVE_IDS.has(componentId)) {
|
||||
// auth primitive 本身不需要再做 auth
|
||||
return null;
|
||||
}
|
||||
|
||||
// 決定 auth service name：
|
||||
// 1. 若 API recipe 宣告了 auth_service（例 recipe:kbdb_get → "kbdb"）→ 用它，
|
||||
// 讓多個 recipe 共用同一把 auth_recipe（不必每個 action 複製 auth recipe）。
|
||||
// 2. 否則 fallback 到把 componentId 當 service name（向後相容舊行為）。
|
||||
let service = componentId;
|
||||
const apiRecipe = await resolveRecipe(componentId, env.RECIPES);
|
||||
if (apiRecipe?.auth_service) {
|
||||
service = apiRecipe.auth_service;
|
||||
}
|
||||
|
||||
const recipe = await resolveAuthRecipe(service, env.RECIPES);
|
||||
if (!recipe) return null;
|
||||
if (!SUPPORTED_PRIMITIVES.has(recipe.primitive)) return null;
|
||||
|
||||
// 走新路徑:HTTP POST 到對應 auth primitive Worker
|
||||
// 走 workers.dev 避開同 zone 死鎖（P0 #9）
|
||||
const primitiveUrl = wasmWorkerUrl(`auth_${recipe.primitive}`, env.WORKER_SUBDOMAIN);
|
||||
const res = await fetch(primitiveUrl, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
action: 'authenticate',
|
||||
api_key: apiKey,
|
||||
service,
|
||||
}),
|
||||
});
|
||||
|
||||
if (!res.ok) {
|
||||
const text = await res.text().catch(() => '');
|
||||
throw new Error(
|
||||
`auth primitive "${recipe.primitive}" 回傳 ${res.status}: ${text.slice(0, 200)}`,
|
||||
);
|
||||
}
|
||||
|
||||
const result = await res.json().catch(() => null) as {
|
||||
success?: boolean;
|
||||
error?: string;
|
||||
auth_headers?: Record<string, string>;
|
||||
auth_query?: Record<string, string>;
|
||||
auth_body?: Record<string, string>;
|
||||
auth_path?: Record<string, string>;
|
||||
} | null;
|
||||
|
||||
if (!result || result.success === false) {
|
||||
throw new Error(
|
||||
`auth primitive 失敗: ${result?.error ?? '未知錯誤'}`,
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
...input,
|
||||
_auth_headers: result.auth_headers ?? {},
|
||||
_auth_query: result.auth_query ?? {},
|
||||
_auth_body: result.auth_body ?? {},
|
||||
_auth_path: result.auth_path ?? {},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,235 @@
|
||||
/**
|
||||
* Credential Injector
|
||||
*
|
||||
* 執行順序：
|
||||
* 1. 檢查是否有對應的 auth recipe（auth_recipe:{componentId} in RECIPES KV）
|
||||
* → 有：走 auth recipe 路徑（支援 static_key, service_account）
|
||||
* → 無：走舊有 flat injection 路徑（向後相容）
|
||||
*
|
||||
* Auth Recipe 路徑：
|
||||
* - static_key：展開 inject.header/query/body 的 {{secret.KEY}} 模板
|
||||
* - service_account：JWT signing → token exchange → 展開 {{runtime.access_token}}
|
||||
* - 注入結果以 _auth_headers / _auth_query / _auth_body 攜帶，不污染業務欄位
|
||||
*
|
||||
* 舊有路徑（向後相容）：
|
||||
* - 從 RECIPES KV 讀取 credentials_required（動態 recipe）
|
||||
* - 或從 BUILTIN_CREDENTIALS_MAP（內建清單）
|
||||
* - 解密後以 inject_as 欄位名稱直接注入 context
|
||||
*/
|
||||
|
||||
import type { Bindings } from '../types';
|
||||
import { resolveRecipe, resolveAuthRecipe } from '../routes/recipes';
|
||||
import type { AuthRecipeDefinition } from '../routes/recipes';
|
||||
|
||||
export interface CredentialRequirement {
|
||||
key: string; // CREDENTIALS_KV 的 credential 名稱（如 gmail_token）
|
||||
inject_as: string; // 注入到 input 的欄位名稱（如 access_token）
|
||||
}
|
||||
|
||||
/** 內建 API recipe 的 credentials_required（對應 component-loader 的 BUILTIN_API_RECIPES）*/
|
||||
const BUILTIN_CREDENTIALS_MAP: Record<string, CredentialRequirement[]> = {
|
||||
gmail: [{ key: 'gmail_token', inject_as: 'access_token' }],
|
||||
google_sheets: [{ key: 'google_oauth', inject_as: 'access_token' }],
|
||||
telegram: [{ key: 'telegram_bot_token', inject_as: 'bot_token' }],
|
||||
line_notify: [{ key: 'line_token', inject_as: 'token' }],
|
||||
};
|
||||
|
||||
// ── AES-GCM 解密 ──────────────────────────────────────────────────────────────
|
||||
|
||||
async function decryptCredential(encryptedJson: string, encryptionKey: string): Promise<string> {
|
||||
const { encrypted, iv } = JSON.parse(encryptedJson) as { encrypted: string; iv: string };
|
||||
|
||||
const keyBytes = hexToUint8Array(encryptionKey);
|
||||
const cryptoKey = await crypto.subtle.importKey(
|
||||
'raw', keyBytes, { name: 'AES-GCM' }, false, ['decrypt'],
|
||||
);
|
||||
|
||||
const decrypted = await crypto.subtle.decrypt(
|
||||
{ name: 'AES-GCM', iv: base64ToUint8Array(iv) },
|
||||
cryptoKey,
|
||||
base64ToUint8Array(encrypted),
|
||||
);
|
||||
|
||||
return new TextDecoder().decode(decrypted);
|
||||
}
|
||||
|
||||
function hexToUint8Array(hex: string): Uint8Array {
|
||||
const bytes = new Uint8Array(hex.length / 2);
|
||||
for (let i = 0; i < hex.length; i += 2) bytes[i / 2] = parseInt(hex.slice(i, i + 2), 16);
|
||||
return bytes;
|
||||
}
|
||||
|
||||
function base64ToUint8Array(b64: string): Uint8Array {
|
||||
const binary = atob(b64);
|
||||
const bytes = new Uint8Array(binary.length);
|
||||
for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i);
|
||||
return bytes;
|
||||
}
|
||||
|
||||
// ── 解密所有 required_secrets → { key: decryptedValue } ──────────────────────
|
||||
|
||||
async function decryptSecrets(
|
||||
recipe: AuthRecipeDefinition,
|
||||
apiKey: string,
|
||||
env: Bindings,
|
||||
): Promise<Record<string, string>> {
|
||||
const result: Record<string, string> = {};
|
||||
|
||||
for (const req of recipe.required_secrets) {
|
||||
if (req.optional) continue;
|
||||
|
||||
const kvKey = `${apiKey}:cred:${req.key}`;
|
||||
const record = await env.CREDENTIALS_KV.get(kvKey);
|
||||
|
||||
if (!record) {
|
||||
throw new Error(
|
||||
`缺少 credential：${req.key}（${req.label}）\n` +
|
||||
`修復步驟：\n` +
|
||||
` 1. 在 credentials.yaml 加入 ${req.key}: "your-value"\n` +
|
||||
` 2. 執行：acr creds push`,
|
||||
);
|
||||
}
|
||||
|
||||
result[req.key] = await decryptCredential(record, env.ENCRYPTION_KEY);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
// ── Template 展開：{{secret.KEY}} 和 {{runtime.KEY}} ─────────────────────────
|
||||
|
||||
function interpolateTemplate(
|
||||
template: string,
|
||||
secrets: Record<string, string>,
|
||||
runtime: Record<string, string>,
|
||||
): string {
|
||||
return template.replace(/\{\{(secret|runtime)\.(\w+)\}\}/g, (_, ns, key) => {
|
||||
if (ns === 'secret') return secrets[key] ?? '';
|
||||
if (ns === 'runtime') return runtime[key] ?? '';
|
||||
return '';
|
||||
});
|
||||
}
|
||||
|
||||
function interpolateRecord(
|
||||
record: Record<string, string>,
|
||||
secrets: Record<string, string>,
|
||||
runtime: Record<string, string>,
|
||||
): Record<string, string> {
|
||||
const result: Record<string, string> = {};
|
||||
for (const [k, v] of Object.entries(record)) {
|
||||
result[k] = interpolateTemplate(v, secrets, runtime);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
// ── Auth Recipe 注入（新路徑）────────────────────────────────────────────────
|
||||
|
||||
async function injectFromAuthRecipe(
|
||||
recipe: AuthRecipeDefinition,
|
||||
input: Record<string, unknown>,
|
||||
env: Bindings,
|
||||
apiKey: string,
|
||||
): Promise<Record<string, unknown>> {
|
||||
// 解密所有 required_secrets
|
||||
const secrets = await decryptSecrets(recipe, apiKey, env);
|
||||
|
||||
// runtime token：service_account 路徑已改走 auth-dispatcher → auth_service_account WASM;
|
||||
// 這條 TS fallback 只處理 static_key (runtime 為空即可),service_account 永遠不會走到這裡
|
||||
const runtime: Record<string, string> = {};
|
||||
|
||||
if (recipe.primitive === 'service_account') {
|
||||
throw new Error(
|
||||
`service_account primitive 應由 auth-dispatcher → auth_service_account WASM 處理,` +
|
||||
`不應進到 credential-injector TS fallback (service=${recipe.service})`,
|
||||
);
|
||||
}
|
||||
|
||||
// 展開 inject 模板
|
||||
const authHeaders = recipe.inject.header
|
||||
? interpolateRecord(recipe.inject.header, secrets, runtime)
|
||||
: {};
|
||||
const authQuery = recipe.inject.query
|
||||
? interpolateRecord(recipe.inject.query, secrets, runtime)
|
||||
: {};
|
||||
const authBody = recipe.inject.body
|
||||
? interpolateRecord(recipe.inject.body, secrets, runtime)
|
||||
: {};
|
||||
|
||||
return {
|
||||
...input,
|
||||
_auth_headers: authHeaders,
|
||||
_auth_query: authQuery,
|
||||
_auth_body: authBody,
|
||||
};
|
||||
}
|
||||
|
||||
// ── 舊有路徑：flat injection（向後相容）──────────────────────────────────────
|
||||
|
||||
async function loadCredentialsRequired(
|
||||
componentId: string,
|
||||
env: Bindings,
|
||||
): Promise<CredentialRequirement[]> {
|
||||
const recipe = await resolveRecipe(componentId, env.RECIPES);
|
||||
if (recipe?.credentials_required?.length) {
|
||||
return recipe.credentials_required;
|
||||
}
|
||||
return BUILTIN_CREDENTIALS_MAP[componentId] ?? [];
|
||||
}
|
||||
|
||||
// ── 主入口 ────────────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* 執行 credential 注入。
|
||||
*
|
||||
* @param componentId - 零件 canonical_id 或 hash
|
||||
* @param input - 節點的 merged context
|
||||
* @param env - Cloudflare Worker Bindings
|
||||
* @param apiKey - 用戶的 API Key（ak_前綴），作為 KV namespace
|
||||
*/
|
||||
export async function injectCredentials(
|
||||
componentId: string,
|
||||
input: Record<string, unknown>,
|
||||
env: Bindings,
|
||||
apiKey?: string,
|
||||
): Promise<Record<string, unknown>> {
|
||||
// 沒有 api_key → local 模式，略過
|
||||
if (!apiKey) return input;
|
||||
|
||||
// ── 新路徑：auth recipe ──
|
||||
const authRecipe = await resolveAuthRecipe(componentId, env.RECIPES);
|
||||
if (authRecipe) {
|
||||
return injectFromAuthRecipe(authRecipe, input, env, apiKey);
|
||||
}
|
||||
|
||||
// ── 舊路徑：flat injection（向後相容）──
|
||||
const required = await loadCredentialsRequired(componentId, env);
|
||||
if (required.length === 0) return input;
|
||||
|
||||
const enriched = { ...input };
|
||||
|
||||
for (const cred of required) {
|
||||
const kvKey = `${apiKey}:cred:${cred.key}`;
|
||||
const record = await env.CREDENTIALS_KV.get(kvKey);
|
||||
|
||||
if (!record) {
|
||||
throw new Error(
|
||||
`缺少 credential：${cred.key}\n` +
|
||||
`修復步驟：\n` +
|
||||
` 1. 在 credentials.yaml 中加入 ${cred.key}: "your-token"\n` +
|
||||
` 2. 執行：acr creds push`,
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
const decrypted = await decryptCredential(record, env.ENCRYPTION_KEY);
|
||||
enriched[cred.inject_as] = decrypted;
|
||||
} catch (e) {
|
||||
throw new Error(
|
||||
`credential "${cred.key}" 解密失敗：${e instanceof Error ? e.message : String(e)}\n` +
|
||||
`修復步驟：重新執行 acr creds push。`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return enriched;
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
import type { Bindings, ExecutionGraph } from '../types';
|
||||
import { ExecutionError, WorkflowPaused } from '../types';
|
||||
import { GraphExecutor } from '../graph-executor';
|
||||
import { graphSchema } from '../lib/schemas';
|
||||
import { createComponentLoader } from '../lib/component-loader';
|
||||
import { writeEvaluation, updateComponentStats } from './execution-evaluator';
|
||||
import { parseTriplets } from './triplet-parser';
|
||||
import { searchNodes } from './search-nodes';
|
||||
import { buildExecutionGraph } from './graph-builder';
|
||||
|
||||
export async function handleCypherSearch(
|
||||
triplets: unknown[],
|
||||
env: Bindings,
|
||||
): Promise<{ nodes: Record<string, unknown>; cypher: unknown; missing: string[] }> {
|
||||
const parsed = parseTriplets(triplets);
|
||||
if (!parsed) {
|
||||
throw new Error('無法解析任何節點');
|
||||
}
|
||||
|
||||
const { nodeResults } = searchNodes(parsed);
|
||||
|
||||
const graph = buildExecutionGraph(parsed, nodeResults, 'cypher-search-result', 'Cypher Search Result');
|
||||
return { nodes: nodeResults, cypher: { nodes: graph.nodes, edges: graph.edges }, missing: [] };
|
||||
}
|
||||
|
||||
export async function handleCypherExecute(
|
||||
triplets: unknown[],
|
||||
context: Record<string, unknown> | undefined,
|
||||
graphId: string,
|
||||
graphName: string,
|
||||
config: Record<string, Record<string, unknown>> | undefined,
|
||||
env: Bindings,
|
||||
waitUntil: (promise: Promise<void>) => void,
|
||||
apiKey?: string,
|
||||
): Promise<{
|
||||
success: boolean;
|
||||
data?: unknown;
|
||||
error?: string;
|
||||
trace?: unknown;
|
||||
duration_ms: number;
|
||||
graph?: ExecutionGraph;
|
||||
// resumable workflow: 節點 pending 時回 paused（不算 success 也不算 fail）
|
||||
paused?: boolean;
|
||||
task_id?: string;
|
||||
run_id?: string;
|
||||
paused_node_id?: string;
|
||||
}> {
|
||||
const parsed = parseTriplets(triplets as unknown[]);
|
||||
if (!parsed) {
|
||||
throw new Error('無法解析任何節點');
|
||||
}
|
||||
|
||||
const { nodeResults } = searchNodes(parsed, config);
|
||||
|
||||
const graph = buildExecutionGraph(parsed, nodeResults, graphId, graphName, config);
|
||||
const parseResult = graphSchema.safeParse(graph);
|
||||
if (!parseResult.success) {
|
||||
throw new Error('圖定義產生失敗');
|
||||
}
|
||||
|
||||
const loader = createComponentLoader(env);
|
||||
const executor = new GraphExecutor(loader, undefined, env, apiKey);
|
||||
const start = Date.now();
|
||||
|
||||
try {
|
||||
const result = await executor.execute(parseResult.data as ExecutionGraph, context ?? {}, env.EXEC_CONTEXT);
|
||||
const duration_ms = Date.now() - start;
|
||||
|
||||
// 非同步記錄統計（Phase 7 補充 analytics，目前為 no-op）
|
||||
const componentId = graph.nodes.find(n => n.componentId)?.componentId ?? graphId;
|
||||
const runId = `${graphId}-${Date.now()}`;
|
||||
waitUntil(writeEvaluation(env, {
|
||||
run_id: runId,
|
||||
workflow_id: graphId,
|
||||
component_id: componentId,
|
||||
verdict: 'success',
|
||||
duration_ms,
|
||||
evaluated_at: Date.now(),
|
||||
}));
|
||||
waitUntil(updateComponentStats(env, componentId, 'success', duration_ms));
|
||||
|
||||
return { success: true, data: result.data, trace: result.trace, duration_ms, graph };
|
||||
} catch (err) {
|
||||
const duration_ms = Date.now() - start;
|
||||
|
||||
// Resumable workflow: 節點回 pending → 回 paused 結構，不算成功也不算失敗
|
||||
// SDD: resumable-workflow/design.md
|
||||
if (err instanceof WorkflowPaused) {
|
||||
return {
|
||||
success: true,
|
||||
paused: true,
|
||||
task_id: err.task_id,
|
||||
run_id: err.run_id,
|
||||
paused_node_id: err.paused_node_id,
|
||||
trace: err.trace_so_far,
|
||||
duration_ms,
|
||||
graph,
|
||||
};
|
||||
}
|
||||
|
||||
const errMsg = err instanceof Error ? err.message : String(err);
|
||||
const componentId = graph.nodes.find(n => n.componentId)?.componentId ?? graphId;
|
||||
const runId = `${graphId}-${Date.now()}`;
|
||||
waitUntil(writeEvaluation(env, {
|
||||
run_id: runId,
|
||||
workflow_id: graphId,
|
||||
component_id: componentId,
|
||||
verdict: 'failed',
|
||||
duration_ms,
|
||||
error_message: errMsg.slice(0, 200),
|
||||
evaluated_at: Date.now(),
|
||||
}));
|
||||
waitUntil(updateComponentStats(env, componentId, 'failed', duration_ms));
|
||||
if (err instanceof ExecutionError) {
|
||||
const traceFormatted = err.trace.map(s => ({
|
||||
node: s.nodeId,
|
||||
status: s.error ? 'failed' : 'success',
|
||||
...(s.error ? { error: s.error } : {}),
|
||||
}));
|
||||
throw new Error(JSON.stringify({
|
||||
success: false,
|
||||
error: errMsg,
|
||||
failed_node: err.failed_node,
|
||||
failed_input: err.failed_input,
|
||||
trace: traceFormatted,
|
||||
duration_ms,
|
||||
}));
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
/**
|
||||
* Execution Analytics — 零件執行後的統計記錄
|
||||
*
|
||||
* Phase 1 MVP：stub（不寫入任何外部服務）
|
||||
* Phase 7 補充：fire-and-forget POST 至 registry.arcrun.dev/analytics/record
|
||||
*/
|
||||
|
||||
import type { Bindings } from '../types';
|
||||
|
||||
export interface EvaluationRecord {
|
||||
run_id: string;
|
||||
workflow_id: string;
|
||||
component_id: string;
|
||||
verdict: 'success' | 'failed' | 'timeout';
|
||||
duration_ms: number;
|
||||
error_message?: string;
|
||||
evaluated_at: number;
|
||||
}
|
||||
|
||||
/** 記錄執行結果（MVP：no-op，Phase 7 補充 analytics）*/
|
||||
export async function writeEvaluation(
|
||||
_env: Bindings,
|
||||
_record: EvaluationRecord,
|
||||
): Promise<void> {
|
||||
// Phase 7: POST to registry.arcrun.dev/analytics/record
|
||||
}
|
||||
|
||||
/** 更新零件統計（MVP：no-op，Phase 7 補充）*/
|
||||
export async function updateComponentStats(
|
||||
_env: Bindings,
|
||||
_componentId: string,
|
||||
_verdict: 'success' | 'failed' | 'timeout',
|
||||
_durationMs: number,
|
||||
): Promise<void> {
|
||||
// Phase 7: update ANALYTICS_KV via registry worker
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
/**
|
||||
* Execution Logger — 執行結果寫入 ANALYTICS_KV（fire-and-forget）
|
||||
*
|
||||
* 設計：每次 workflow 執行後，將統計數據寫入 ANALYTICS_KV（key = stats:{workflowId}）。
|
||||
* Phase 7 可升級為 POST 至 registry.arcrun.dev/analytics/record。
|
||||
*/
|
||||
|
||||
import type { Bindings, GraphNode } from '../types';
|
||||
|
||||
export interface ExecutionVerdict {
|
||||
workflow_id: string;
|
||||
component_ids: string[];
|
||||
verdict: 'success' | 'failed';
|
||||
duration_ms: number;
|
||||
message: string;
|
||||
recorded_at: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* 寫入執行結果至 ANALYTICS_KV（fire-and-forget，不阻擋主流程）
|
||||
* 由 c.executionCtx.waitUntil() 包裹呼叫
|
||||
*/
|
||||
export async function writeExecutionVerdict(
|
||||
env: Bindings,
|
||||
workflowId: string,
|
||||
nodes: GraphNode[],
|
||||
verdict: 'success' | 'failed',
|
||||
durationMs: number,
|
||||
message: string,
|
||||
): Promise<void> {
|
||||
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 天
|
||||
});
|
||||
} catch {
|
||||
// fire-and-forget：不拋錯，不影響主流程
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
import type { ParsedTriplets } from './triplet-parser';
|
||||
import { toEdgeType } from './triplet-parser';
|
||||
import type { SearchResult } from './search-nodes';
|
||||
|
||||
/** 從 nodeResults + parsed 組成可直接送入 /execute 的 ExecutionGraph
|
||||
*
|
||||
* config 格式（來自 workflow YAML 的 config 欄位）：
|
||||
* { node_name: { component: "cmp_xxxxxxxx" | "rec_xxxxxxxx" | canonical_id, ...params } }
|
||||
*
|
||||
* 若 config[node].component 存在，以它覆蓋 searchNodes 偵測到的 componentId。
|
||||
* config[node] 的其他欄位作為節點靜態參數（node.data），合併進執行 context。
|
||||
*/
|
||||
export function buildExecutionGraph(
|
||||
parsed: ParsedTriplets,
|
||||
nodeResults: SearchResult['nodeResults'],
|
||||
graphId: string,
|
||||
graphName: string,
|
||||
config?: Record<string, Record<string, unknown>>,
|
||||
) {
|
||||
const nodes = [...parsed.nodeNames].map(name => {
|
||||
const nr = nodeResults[name]!;
|
||||
const id = name.toLowerCase().replace(/\s+/g, '-');
|
||||
const nodeConfig = config?.[name] ?? {};
|
||||
|
||||
// config[name].component 可以是 hash 或 canonical_id，覆蓋自動偵測的 componentId
|
||||
const componentId = (nodeConfig.component as string | undefined) ?? nr.componentId;
|
||||
|
||||
// 其他 config 欄位作為 node.data（靜態參數）
|
||||
const { component: _component, ...staticParams } = nodeConfig;
|
||||
const data = Object.keys(staticParams).length > 0 ? staticParams : undefined;
|
||||
|
||||
return { id, type: nr.type, componentId, label: name, data };
|
||||
});
|
||||
|
||||
const edges = parsed.edges.map(e => {
|
||||
// 「對每個 X」label 抽 iterator：cypher binding 表達 FOREACH 的迭代變數
|
||||
// 例：'A >> 對每個 paragraph >> B' → type=FOREACH, iterator='paragraph'
|
||||
// getIterableFromContext 會找 ctx.paragraphs（複數）或 ctx.paragraph
|
||||
let iterator: string | undefined;
|
||||
let label = e.label;
|
||||
const foreachMatch = label.match(/^(?:對每個|FOREACH)\s+(\w+)$/i);
|
||||
if (foreachMatch) {
|
||||
iterator = foreachMatch[1];
|
||||
label = '對每個'; // 改回標準 label 走 SEMANTIC_EDGE_MAP
|
||||
}
|
||||
const edge: { from: string; to: string; type: ReturnType<typeof toEdgeType>; iterator?: string } = {
|
||||
from: e.from.toLowerCase().replace(/\s+/g, '-'),
|
||||
to: e.to.toLowerCase().replace(/\s+/g, '-'),
|
||||
type: toEdgeType(label),
|
||||
};
|
||||
if (iterator) edge.iterator = iterator;
|
||||
return edge;
|
||||
});
|
||||
|
||||
return { id: graphId, name: graphName, nodes, edges };
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
import type { ParsedTriplets, NodeRole } from './triplet-parser';
|
||||
import { resolveNodeRole } from './triplet-parser';
|
||||
|
||||
export type SearchResult = {
|
||||
nodeResults: Record<string, { status: 'found' | 'missing'; componentId?: string; type: NodeRole }>;
|
||||
missingNodes: string[];
|
||||
};
|
||||
|
||||
/**
|
||||
* 對所有節點進行解析，確認每個節點對應的零件 ID。
|
||||
*
|
||||
* 注意：此步驟只做靜態解析，不做遠端查找。
|
||||
* 零件是否真的存在由 component-loader 在執行時決定（Service Binding / KV / URL）。
|
||||
*
|
||||
* 優先序：
|
||||
* 1. Input/Output 角色：自動標記，componentId = 小寫節點名稱
|
||||
* 2. config[nodeName].component 已指定：使用 config 提供的 componentId
|
||||
* 3. 其他：componentId = 節點名稱（交給 component-loader 在執行時解析）
|
||||
*/
|
||||
export function searchNodes(
|
||||
parsed: ParsedTriplets,
|
||||
config?: Record<string, Record<string, unknown>>,
|
||||
): SearchResult {
|
||||
const nodeResults: Record<string, { status: 'found' | 'missing'; componentId?: string; type: NodeRole }> = {};
|
||||
|
||||
for (const nodeName of parsed.nodeNames) {
|
||||
const role = resolveNodeRole(nodeName, parsed);
|
||||
|
||||
if (role === 'Input' || role === 'Output') {
|
||||
nodeResults[nodeName] = { status: 'found', componentId: nodeName.toLowerCase(), type: role };
|
||||
continue;
|
||||
}
|
||||
|
||||
const configComponent = config?.[nodeName]?.component as string | undefined;
|
||||
const componentId = configComponent ?? nodeName;
|
||||
nodeResults[nodeName] = { status: 'found', componentId, type: role };
|
||||
}
|
||||
|
||||
return { nodeResults, missingNodes: [] };
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
import { SEMANTIC_EDGE_MAP, VALID_EDGE_TYPES } from '../lib/constants';
|
||||
import type { EdgeType } from '../types';
|
||||
|
||||
export type ParsedTriplets = {
|
||||
edges: Array<{ from: string; to: string; label: string }>;
|
||||
nodeNames: Set<string>;
|
||||
/** 出現在 from 但不出現在任何 to 的節點（事件源 / 起始點） */
|
||||
sourceNodes: Set<string>;
|
||||
/** 出現在 to 但不出現在任何 from 的節點（終點）*/
|
||||
sinkNodes: Set<string>;
|
||||
};
|
||||
|
||||
export type NodeRole = 'Input' | 'Component' | 'Output';
|
||||
|
||||
/**
|
||||
* 解析後的零件 URI
|
||||
* 支援格式：
|
||||
* component://validate_json
|
||||
* component://validate_json@stable
|
||||
* component://validate_json@pinned:v1
|
||||
* workflow://wf_save_to_db
|
||||
* ui://u6u-btn
|
||||
* style://glow-effect
|
||||
*/
|
||||
export interface ResolvedComponentId {
|
||||
type: 'component' | 'workflow' | 'ui' | 'style';
|
||||
canonicalId: string;
|
||||
stability: 'floating' | 'stable' | 'pinned';
|
||||
pinnedVersion?: string;
|
||||
/** 原始 URI 字串 */
|
||||
raw: string;
|
||||
}
|
||||
|
||||
/** 解析零件 URI 協議 */
|
||||
export function resolveComponentId(uri: string): ResolvedComponentId {
|
||||
const raw = uri.trim();
|
||||
|
||||
// 解析協議前綴
|
||||
let type: ResolvedComponentId['type'] = 'component';
|
||||
let rest = raw;
|
||||
|
||||
if (raw.startsWith('component://')) {
|
||||
type = 'component';
|
||||
rest = raw.slice('component://'.length);
|
||||
} else if (raw.startsWith('workflow://')) {
|
||||
type = 'workflow';
|
||||
rest = raw.slice('workflow://'.length);
|
||||
} else if (raw.startsWith('ui://')) {
|
||||
type = 'ui';
|
||||
rest = raw.slice('ui://'.length);
|
||||
} else if (raw.startsWith('style://')) {
|
||||
type = 'style';
|
||||
rest = raw.slice('style://'.length);
|
||||
}
|
||||
|
||||
// 解析穩定性標籤
|
||||
// component://id@stable
|
||||
// component://id@pinned:v1
|
||||
let canonicalId = rest;
|
||||
let stability: ResolvedComponentId['stability'] = 'floating';
|
||||
let pinnedVersion: string | undefined;
|
||||
|
||||
const atIdx = rest.indexOf('@');
|
||||
if (atIdx > 0) {
|
||||
canonicalId = rest.slice(0, atIdx);
|
||||
const tag = rest.slice(atIdx + 1);
|
||||
if (tag === 'stable') {
|
||||
stability = 'stable';
|
||||
} else if (tag.startsWith('pinned:')) {
|
||||
stability = 'pinned';
|
||||
pinnedVersion = tag.slice('pinned:'.length);
|
||||
}
|
||||
}
|
||||
|
||||
return { type, canonicalId, stability, pinnedVersion, raw };
|
||||
}
|
||||
|
||||
/** 解析 triplets 字串陣列，回傳節點與邊的結構 */
|
||||
export function parseTriplets(rawTriplets: unknown[]): ParsedTriplets | null {
|
||||
const edges: Array<{ from: string; to: string; label: string }> = [];
|
||||
const nodeNames = new Set<string>();
|
||||
const fromSet = new Set<string>();
|
||||
const toSet = new Set<string>();
|
||||
|
||||
for (const line of rawTriplets) {
|
||||
if (typeof line !== 'string') continue;
|
||||
const parts = line.split('>>').map((s: string) => s.trim());
|
||||
if (parts.length !== 3) continue;
|
||||
const [from, action, to] = parts;
|
||||
edges.push({ from, to, label: action });
|
||||
nodeNames.add(from);
|
||||
nodeNames.add(to);
|
||||
fromSet.add(from);
|
||||
toSet.add(to);
|
||||
}
|
||||
|
||||
if (nodeNames.size === 0) return null;
|
||||
|
||||
const sourceNodes = new Set([...fromSet].filter(n => !toSet.has(n)));
|
||||
const sinkNodes = new Set([...toSet].filter(n => !fromSet.has(n)));
|
||||
return { edges, nodeNames, sourceNodes, sinkNodes };
|
||||
}
|
||||
|
||||
/** 保留字節點名稱 — 明確宣告為 Input 或 Output 端點 */
|
||||
const INPUT_NAMES = new Set(['input', 'trigger', 'webhook', 'start']);
|
||||
const OUTPUT_NAMES = new Set(['output', 'result', 'end', 'done']);
|
||||
|
||||
/** 根據節點在圖中的位置決定其 type
|
||||
*
|
||||
* 規則：
|
||||
* - 名稱在 INPUT_NAMES → Input（無論位置）
|
||||
* - 名稱在 OUTPUT_NAMES → Output（無論位置）
|
||||
* - sourceNode（只出現在 from）且名稱不在 INPUT_NAMES → Component（例如 cron 作為觸發源）
|
||||
* - sinkNode（只出現在 to）且名稱不在 OUTPUT_NAMES → Component（最常見情況：最後一個實際零件）
|
||||
* - 其他中間節點 → Component
|
||||
*/
|
||||
export function resolveNodeRole(name: string, parsed: ParsedTriplets): NodeRole {
|
||||
if (INPUT_NAMES.has(name.toLowerCase())) return 'Input';
|
||||
if (OUTPUT_NAMES.has(name.toLowerCase())) return 'Output';
|
||||
if (parsed.sourceNodes.has(name)) return 'Input';
|
||||
return 'Component';
|
||||
}
|
||||
|
||||
/** 將 edge label 轉換為合法 EdgeType
|
||||
* 優先序：VALID_EDGE_TYPES（完整匹配）→ SEMANTIC_EDGE_MAP（語意別名）→ 預設 PIPE */
|
||||
export function toEdgeType(label: string): EdgeType {
|
||||
const upper = label.toUpperCase();
|
||||
if (VALID_EDGE_TYPES.has(upper)) return upper as EdgeType;
|
||||
return (SEMANTIC_EDGE_MAP[label] ?? SEMANTIC_EDGE_MAP[upper] ?? 'PIPE') as EdgeType;
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import type { Bindings } from '../types';
|
||||
import { graphSchema } from '../lib/schemas';
|
||||
import { parseTriplets } from './triplet-parser';
|
||||
import { searchNodes } from './search-nodes';
|
||||
import { buildExecutionGraph } from './graph-builder';
|
||||
|
||||
export async function resolveWebhookGraph(
|
||||
body: Record<string, unknown>,
|
||||
description: string,
|
||||
env: Bindings,
|
||||
): Promise<{ resolvedGraph: Record<string, unknown>; error?: string }> {
|
||||
// 路徑 A：triplets 格式
|
||||
if (Array.isArray(body.triplets) && body.triplets.length > 0) {
|
||||
const parsed = parseTriplets(body.triplets as unknown[]);
|
||||
if (!parsed) return { resolvedGraph: {}, error: '無法解析 triplets' };
|
||||
|
||||
const { nodeResults } = searchNodes(parsed);
|
||||
|
||||
const graphId = `webhook-${Date.now()}`;
|
||||
const graphName = description || `Webhook ${new Date().toISOString()}`;
|
||||
const graph = buildExecutionGraph(parsed, nodeResults, graphId, graphName) as Record<string, unknown>;
|
||||
|
||||
const parseResult = graphSchema.safeParse(graph);
|
||||
if (!parseResult.success) {
|
||||
return { resolvedGraph: {}, error: '圖定義產生失敗' };
|
||||
}
|
||||
|
||||
return { resolvedGraph: graph };
|
||||
}
|
||||
|
||||
// 路徑 B：graph 格式
|
||||
if (body.graph && typeof body.graph === 'object') {
|
||||
const graphWithDefaults = {
|
||||
id: `webhook-${Date.now()}`,
|
||||
name: description || `Webhook ${new Date().toISOString()}`,
|
||||
...(body.graph as Record<string, unknown>),
|
||||
};
|
||||
const parsed = graphSchema.safeParse(graphWithDefaults);
|
||||
if (!parsed.success) {
|
||||
return { resolvedGraph: {}, error: '圖定義驗證失敗' };
|
||||
}
|
||||
return { resolvedGraph: graphWithDefaults };
|
||||
}
|
||||
|
||||
// 路徑 C：body 直接就是 graph
|
||||
if (body.nodes && body.edges) {
|
||||
const graphWithDefaults = {
|
||||
id: `webhook-${Date.now()}`,
|
||||
name: description || `Webhook ${new Date().toISOString()}`,
|
||||
...body,
|
||||
};
|
||||
const parsed = graphSchema.safeParse(graphWithDefaults);
|
||||
if (!parsed.success) {
|
||||
return { resolvedGraph: {}, error: '圖定義驗證失敗' };
|
||||
}
|
||||
return { resolvedGraph: graphWithDefaults };
|
||||
}
|
||||
|
||||
return { resolvedGraph: {}, error: '需提供 graph 物件或 triplets 陣列' };
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
import type { Bindings, ExecutionGraph, ExecutionContext } from '../types';
|
||||
import { ExecutionError } from '../types';
|
||||
import { GraphExecutor } from '../graph-executor';
|
||||
import { graphSchema } from '../lib/schemas';
|
||||
import { createComponentLoader } from '../lib/component-loader';
|
||||
import { recordTelemetry } from '../lib/telemetry';
|
||||
|
||||
type WebhookRecord = {
|
||||
graph: Record<string, unknown>;
|
||||
description: string;
|
||||
created_at: string;
|
||||
};
|
||||
|
||||
export function generateToken(): string {
|
||||
const tokenBytes = crypto.getRandomValues(new Uint8Array(16));
|
||||
return Array.from(tokenBytes).map(b => b.toString(16).padStart(2, '0')).join('');
|
||||
}
|
||||
|
||||
export async function validateAndParseWebhook(raw: string): Promise<WebhookRecord | null> {
|
||||
try {
|
||||
return JSON.parse(raw) as WebhookRecord;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export async function executeWebhookGraph(
|
||||
env: Bindings,
|
||||
graph: Record<string, unknown>,
|
||||
triggerContext: Record<string, unknown>,
|
||||
token: string,
|
||||
apiKey?: string,
|
||||
ctx?: ExecutionContext, // 可選 — 用 waitUntil 把 telemetry 推到背景
|
||||
userAgent?: string, // MCP / SDK client 帶過來
|
||||
): Promise<{ success: boolean; data?: unknown; error?: string; trace?: unknown; duration_ms: number }> {
|
||||
const parsed = graphSchema.safeParse(graph);
|
||||
if (!parsed.success) {
|
||||
return { success: false, error: '圖定義已失效', duration_ms: 0 };
|
||||
}
|
||||
|
||||
const loader = createComponentLoader(env);
|
||||
const executor = new GraphExecutor(loader, undefined, env, apiKey);
|
||||
const start = Date.now();
|
||||
|
||||
try {
|
||||
const result = await executor.execute(
|
||||
parsed.data as ExecutionGraph,
|
||||
{ ...triggerContext, _webhook_token: token },
|
||||
env.EXEC_CONTEXT,
|
||||
);
|
||||
const duration_ms = Date.now() - start;
|
||||
|
||||
// Implicit telemetry：成功 run（含 paused 也算「成功啟動」由 trigger_workflow 那層分類）
|
||||
recordTelemetry(env, apiKey, {
|
||||
event_type: 'run_success',
|
||||
workflow_name: token,
|
||||
duration_ms,
|
||||
agent_user_agent: userAgent,
|
||||
}, ctx);
|
||||
|
||||
return { success: true, data: result.data, duration_ms };
|
||||
} catch (err) {
|
||||
const duration_ms = Date.now() - start;
|
||||
const errMsg = err instanceof Error ? err.message : String(err);
|
||||
const isPaused = /workflow paused/i.test(errMsg);
|
||||
|
||||
// Implicit telemetry：paused 算 run_success；真錯才 run_fail
|
||||
recordTelemetry(env, apiKey, {
|
||||
event_type: isPaused ? 'run_success' : 'run_fail',
|
||||
workflow_name: token,
|
||||
error_code: isPaused ? 'paused_awaiting_resume' : 'execution_error',
|
||||
duration_ms,
|
||||
agent_user_agent: userAgent,
|
||||
}, ctx);
|
||||
|
||||
if (err instanceof ExecutionError) {
|
||||
const traceFormatted = err.trace.map(s => ({
|
||||
node: s.nodeId,
|
||||
status: s.error ? 'failed' : 'success',
|
||||
...(s.error ? { error: s.error } : {}),
|
||||
}));
|
||||
return {
|
||||
success: false,
|
||||
error: errMsg,
|
||||
trace: traceFormatted,
|
||||
duration_ms,
|
||||
};
|
||||
}
|
||||
return { success: false, error: errMsg, duration_ms };
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user