/** * Paused workflow runs:節點回 pending 時把 run state 持久化進 KV, * webhook callback 進來時撿回繼續執行 * * SDD: matrix/arcrun/.agents/specs/resumable-workflow/design.md §2.1 * * KV key: paused_run:{task_id} * TTL: 24h(避免 KV 累積,超過就 GC) * * 設計筆記: * - 用 task_id 當 key(daemon 派的 unique id),不用 run_id(同 run 可能多 paused 節點 v2) * - consume = load + delete 原子操作(避免重複 callback 重複執行) */ import type { ExecutionGraph, TraceStep } from '../types'; export interface PausedRunState { run_id: string; graph: ExecutionGraph; paused_node_id: string; paused_context: Record; paused_pending_result: Record; // 節點回的 {pending, task_id, ...} trace_so_far: TraceStep[]; api_key?: string; expires_at: number; // unix ms // resume 時用來 parse callback result 的 recipe output 規格(resumable + recipe 整合) recipe_output_format?: 'text' | 'json'; recipe_output_required_fields?: string[]; } const KEY_PREFIX = 'paused_run:'; /** * Per-user paused index:列出某 api_key 當前 paused tasks 不依賴 CF KV list(強 eventual * consistent,30-60s 延遲)。改維護一個 user-keyed JSON list,list 操作改 single KV.get。 * * Key: `paused_idx:{api_key}` * Value: JSON Array<{task_id, paused_node_id, run_id, workflow_name?, expires_at, persisted_at}> * * 對應 LI SDD M2.1 — /executions/paused endpoint 即時性。 */ const IDX_PREFIX = 'paused_idx:'; const TTL_SECONDS = 24 * 60 * 60; export type PausedIndexEntry = { task_id: string; run_id: string; paused_node_id: string; workflow_name?: string; expires_at: number; persisted_at: number; }; type KvBinding = { get: (key: string) => Promise; put: (key: string, value: string, options?: { expirationTtl?: number }) => Promise; delete: (key: string) => Promise; }; async function readIndex(kv: KvBinding, apiKey: string): Promise { const raw = await kv.get(`${IDX_PREFIX}${apiKey}`); if (!raw) return []; try { const arr = JSON.parse(raw); return Array.isArray(arr) ? arr : []; } catch { return []; } } async function writeIndex(kv: KvBinding, apiKey: string, entries: PausedIndexEntry[]): Promise { // 過濾過期項目(避免 index 爆量) const now = Date.now(); const fresh = entries.filter((e) => e.expires_at > now); await kv.put(`${IDX_PREFIX}${apiKey}`, JSON.stringify(fresh), { expirationTtl: TTL_SECONDS }); } export async function persistPausedRun( kv: KvBinding, taskId: string, state: PausedRunState, ): Promise { // 1) 寫單一 task state await kv.put(`${KEY_PREFIX}${taskId}`, JSON.stringify(state), { expirationTtl: TTL_SECONDS }); // 2) 維護 per-user index(讓 /executions/paused list 不靠 KV list 即時拿到) if (state.api_key) { const idx = await readIndex(kv, state.api_key); // 去重(重複 paused 同 task_id 時覆蓋) const filtered = idx.filter((e) => e.task_id !== taskId); filtered.unshift({ task_id: taskId, run_id: state.run_id, paused_node_id: state.paused_node_id, workflow_name: state.graph.name, expires_at: state.expires_at, persisted_at: Date.now(), }); // 限 100 筆避免 index 無限長(超過捨棄最舊) await writeIndex(kv, state.api_key, filtered.slice(0, 100)); } } export async function loadPausedRun( kv: KvBinding, taskId: string, ): Promise { const raw = await kv.get(`${KEY_PREFIX}${taskId}`); if (!raw) return null; try { return JSON.parse(raw) as PausedRunState; } catch { return null; } } /** * 列某 api_key 當前 paused tasks。走 per-user index(強 consistent,無 KV list 延遲) */ export async function listPausedRunsByApiKey( kv: KvBinding, apiKey: string, limit = 20, ): Promise { const idx = await readIndex(kv, apiKey); const now = Date.now(); return idx.filter((e) => e.expires_at > now).slice(0, limit); } /** * 原子讀+刪:避免同 task_id 重複 callback 重複執行下游 * (CF KV 沒真原子操作,但 delete 失敗不影響 load 已成功) */ export async function consumePausedRun( kv: KvBinding, taskId: string, ): Promise { const state = await loadPausedRun(kv, taskId); if (!state) return null; await kv.delete(`${KEY_PREFIX}${taskId}`).catch(() => { // delete 失敗不擋,最多就重複執行一次(接受) }); // 同步從 per-user index 移除 if (state.api_key) { const idx = await readIndex(kv, state.api_key); const filtered = idx.filter((e) => e.task_id !== taskId); await writeIndex(kv, state.api_key, filtered).catch(() => {}); } return state; } /** 偵測 component result 是否為「需要 resume」的 pending pattern */ export function isResumablePending(result: unknown): { task_id: string } | null { if (!result || typeof result !== 'object') return null; const r = result as Record; if (r.pending !== true) return null; if (typeof r.task_id !== 'string' || !r.task_id) return null; return { task_id: r.task_id }; } /** * Parse claude_api result with recipe output format. * 同步路徑跟 resume 路徑都用同一個解析器,避免邏輯歪掉。 * * 輸入:result(可能是 {data:{text:"..."}} 或 {text:"..."}) * 輸出:parsed object 或 fallback 結構 */ export function parseRecipeOutput( result: unknown, format: 'text' | 'json' | undefined, requiredFields: string[] | undefined, ): unknown { if (format !== 'json' || !result || typeof result !== 'object') return result; const r = result as Record; const text = (r.data as Record | undefined)?.text ?? r.text; if (typeof text !== 'string') return result; // 剝除 ```json ... ``` markdown fence(Claude 常這樣包) let jsonText = String(text).trim(); const fenceMatch = jsonText.match(/^```(?:json)?\s*\n([\s\S]*?)\n```$/); if (fenceMatch) jsonText = fenceMatch[1].trim(); try { const parsed = JSON.parse(jsonText); if (requiredFields && parsed && typeof parsed === 'object') { const missing = requiredFields.filter((f) => !(f in (parsed as Record))); if (missing.length > 0) { return { success: false, error: `recipe output 缺欄位: ${missing.join(', ')}`, raw: parsed }; } } // 把 parsed 的欄位 spread 到 top-level,FOREACH / 下游 {{var}} 都好取 return { success: true, data: parsed, ...(parsed && typeof parsed === 'object' ? parsed as Record : {}) }; } catch (e) { return { success: false, error: `recipe output JSON parse 失敗: ${e instanceof Error ? e.message : String(e)}`, raw_text: text }; } }