From 76e984101fc85c4cfb60ba1fa7668287c09ef65c Mon Sep 17 00:00:00 2001 From: hangerye Date: Tue, 28 Oct 2025 10:06:04 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E9=87=8D=E6=9E=84=E5=9B=9E=E6=94=BE?= =?UTF-8?q?=E5=BC=95=E6=93=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../record-replay/engine/scheduler.ts | 823 ++++++++++-------- .../background/record-replay/nodes/index.ts | 12 +- 2 files changed, 445 insertions(+), 390 deletions(-) diff --git a/app/chrome-extension/entrypoints/background/record-replay/engine/scheduler.ts b/app/chrome-extension/entrypoints/background/record-replay/engine/scheduler.ts index a44fbf2..2fbfc72 100644 --- a/app/chrome-extension/entrypoints/background/record-replay/engine/scheduler.ts +++ b/app/chrome-extension/entrypoints/background/record-replay/engine/scheduler.ts @@ -1,5 +1,6 @@ // engine/scheduler.ts — DAG-only orchestrator for record-replay -// Note: consolidates wait/retry/logging and delegates node execution to nodes/* registry +// Core responsibilities are split across: Orchestrator (prepare → traverse → cleanup), +// policies (wait/retry), nodes registry (executeStep), logger (overlay/persist), plugins (hooks). import { TOOL_NAMES } from 'chrome-mcp-shared'; import { handleCallTool } from '@/entrypoints/background/tools'; @@ -34,266 +35,235 @@ export interface RunOptions { plugins?: RunPlugin[]; } -export async function runFlow(flow: Flow, options: RunOptions = {}): Promise { - const runId = `run_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; - const startAt = Date.now(); - const logger = new RunLogger(runId); - const pluginManager = new PluginManager( - options.plugins && options.plugins.length ? options.plugins : [breakpointPlugin()], +class ExecutionOrchestrator { + private runId = `run_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; + private startAt = Date.now(); + private logger = new RunLogger(this.runId); + private pluginManager = new PluginManager( + this.options.plugins && this.options.plugins.length + ? this.options.plugins + : [breakpointPlugin()], ); + private vars: Record = Object.create(null); + private deadline = 0; + private networkCaptureStarted = false; + private paused = false; + private failed = 0; + private pendingAfterScripts: StepScript[] = []; + private steps: Step[] = []; + private prepareError: RunResult | null = null; - // Global deadline for run (optional) - const globalTimeout = Math.max(0, Number(options.timeoutMs || 0)); - const deadline = globalTimeout > 0 ? startAt + globalTimeout : 0; - const ensureWithinDeadline = () => { - if (deadline > 0 && Date.now() > deadline) { + constructor( + private flow: Flow, + private options: RunOptions = {}, + ) { + for (const v of flow.variables || []) if (v.default !== undefined) this.vars[v.key] = v.default; + if (options.args) Object.assign(this.vars, options.args); + const globalTimeout = Math.max(0, Number(options.timeoutMs || 0)); + this.deadline = globalTimeout > 0 ? this.startAt + globalTimeout : 0; + } + + private ensureWithinDeadline() { + if (this.deadline > 0 && Date.now() > this.deadline) { const err = new Error('Global timeout reached'); - logger.push({ + this.logger.push({ stepId: 'global-timeout', status: 'failed', message: 'Global timeout reached', }); throw err; } - }; + } - // prepare variables - const vars: Record = Object.create(null); - for (const v of flow.variables || []) if (v.default !== undefined) vars[v.key] = v.default; - if (options.args) Object.assign(vars, options.args); - - // Derive a default startUrl when not provided: prefer first navigate step - let derivedStartUrl: string | undefined = undefined; - try { - const hasDag0 = Array.isArray((flow as any).nodes) && (flow as any).nodes.length > 0; - const nodes0 = hasDag0 ? (((flow as any).nodes || []) as any[]) : []; - const edges0 = hasDag0 ? (((flow as any).edges || []) as any[]) : []; - const defaultEdges0 = hasDag0 ? defaultEdgesOnly(edges0 as any) : []; - const order0 = hasDag0 ? topoOrder(nodes0 as any, defaultEdges0 as any) : []; - const steps0: Step[] = hasDag0 ? order0.map((n) => mapDagNodeToStep(n as any)) : []; - const nav = steps0.find((s: any) => s && (s as any).type === 'navigate') as any; - if (nav && typeof nav.url === 'string') derivedStartUrl = expandTemplatesDeep(nav.url, {}); - } catch {} - - const ensured = await ensureTab({ - tabTarget: options.tabTarget, - startUrl: options.startUrl || derivedStartUrl, - refresh: options.refresh, - }); - - // register run state - try { - await runState.restore(); - await runState.add(runId, { - id: runId, - flowId: flow.id, - name: flow.name, - status: 'running', - startedAt: startAt, - updatedAt: startAt, - }); - } catch {} - - // plugins: run start - await pluginManager.runStart({ runId, flow, vars }); - - // pre-load read_page to init bridges only when on a web page - try { - const u = ensured?.url || ''; - if (/^(https?:|file:)/i.test(u)) { - await handleCallTool({ name: TOOL_NAMES.BROWSER.READ_PAGE, args: {} }); + async run(): Promise { + try { + await this.prepareExecution(); + if (this.prepareError) return this.prepareError; + return await this.traverseDag(); + } finally { + await this.cleanup(); } - } catch {} + } - // collect required variables via overlay prompt - try { - const needed = (flow.variables || []).filter( - (v) => - (options.args?.[v.key] == null || options.args?.[v.key] === '') && - (v.rules?.required || (v.default ?? '') === ''), - ); - if (needed.length) { - const res = await handleCallTool({ - name: TOOL_NAMES.BROWSER.SEND_COMMAND_TO_INJECT_SCRIPT, - args: { - eventName: 'collectVariables', - payload: JSON.stringify({ variables: needed, useOverlay: true }), - }, - }); - let values: Record | null = null; - try { - const t = (res?.content || []).find((c: any) => c.type === 'text')?.text; - const j = t ? JSON.parse(t) : null; - if (j && j.success && j.values) values = j.values; - } catch {} - if (!values) { - const tabs = await chrome.tabs.query({ active: true, currentWindow: true }); - const tabId = tabs?.[0]?.id; - if (typeof tabId === 'number') { - const res2 = await chrome.tabs.sendMessage(tabId, { - action: 'collectVariables', - variables: needed, - useOverlay: true, - } as any); - if (res2 && res2.success && res2.values) values = res2.values; + private async prepareExecution() { + // Derive default startUrl + let derivedStartUrl: string | undefined; + try { + const hasDag0 = Array.isArray(this.flow.nodes) && this.flow.nodes.length > 0; + const nodes0 = hasDag0 ? this.flow.nodes || [] : []; + const edges0 = hasDag0 ? this.flow.edges || [] : []; + const defaultEdges0 = hasDag0 ? defaultEdgesOnly(edges0) : []; + const order0 = hasDag0 ? topoOrder(nodes0, defaultEdges0) : []; + const steps0: Step[] = hasDag0 ? order0.map((n) => mapDagNodeToStep(n)) : []; + const nav = steps0.find((s) => s && s.type === 'navigate'); + if (nav && typeof (nav as any).url === 'string') + derivedStartUrl = expandTemplatesDeep((nav as any).url, {}); + } catch {} + + const ensured = await ensureTab({ + tabTarget: this.options.tabTarget, + startUrl: this.options.startUrl || derivedStartUrl, + refresh: this.options.refresh, + }); + + // register run state + await runState.restore(); + await runState.add(this.runId, { + id: this.runId, + flowId: this.flow.id, + name: this.flow.name, + status: 'running', + startedAt: this.startAt, + updatedAt: this.startAt, + }); + + await this.pluginManager.runStart({ runId: this.runId, flow: this.flow, vars: this.vars }); + + // pre-load read_page when on web + try { + const u = ensured?.url || ''; + if (/^(https?:|file:)/i.test(u)) + await handleCallTool({ name: TOOL_NAMES.BROWSER.READ_PAGE, args: {} }); + } catch {} + + // overlay variable collection + try { + const needed = (this.flow.variables || []).filter( + (v) => + (this.options.args?.[v.key] == null || this.options.args?.[v.key] === '') && + (v.rules?.required || (v.default ?? '') === ''), + ); + if (needed.length) { + const res = await handleCallTool({ + name: TOOL_NAMES.BROWSER.SEND_COMMAND_TO_INJECT_SCRIPT, + args: { + eventName: 'collectVariables', + payload: JSON.stringify({ variables: needed, useOverlay: true }), + }, + }); + let values: Record | null = null; + try { + const t = (res?.content || []).find((c: any) => c.type === 'text')?.text; + const j = t ? JSON.parse(t) : null; + if (j && j.success && j.values) values = j.values; + } catch {} + if (!values) { + const tabs = await chrome.tabs.query({ active: true, currentWindow: true }); + const tabId = tabs?.[0]?.id; + if (typeof tabId === 'number') { + const res2 = await chrome.tabs.sendMessage(tabId, { + action: 'collectVariables', + variables: needed, + useOverlay: true, + } as any); + if (res2 && res2.success && res2.values) values = res2.values; + } + } + if (values) Object.assign(this.vars, values); + } + } catch {} + + await this.logger.overlayInit(); + + // binding enforcement + try { + const tabs = await chrome.tabs.query({ active: true, currentWindow: true }); + const currentUrl = tabs?.[0]?.url || ''; + const bindings = this.flow.meta?.bindings || []; + if (!this.options.startUrl && bindings.length > 0) { + const ok = bindings.some((b) => { + try { + if (b.type === 'domain') return new URL(currentUrl).hostname.includes(b.value); + if (b.type === 'path') return new URL(currentUrl).pathname.startsWith(b.value); + if (b.type === 'url') return currentUrl.startsWith(b.value); + } catch {} + return false; + }); + if (!ok) { + this.prepareError = { + runId: this.runId, + success: false, + summary: { total: 0, success: 0, failed: 0, tookMs: 0 }, + url: currentUrl, + outputs: null, + logs: [ + { + stepId: 'binding-check', + status: 'failed', + message: + 'Flow binding mismatch. Provide startUrl or open a page matching flow.meta.bindings.', + }, + ], + screenshots: { onFailure: null }, + paused: false, + }; + return; } } - if (values) Object.assign(vars, values); - } - } catch {} - - await logger.overlayInit(); - - // binding enforcement - try { - const tabs = await chrome.tabs.query({ active: true, currentWindow: true }); - const currentUrl = tabs?.[0]?.url || ''; - const bindings = flow.meta?.bindings || []; - if (!options.startUrl && bindings.length > 0) { - const ok = bindings.some((b) => { - try { - if (b.type === 'domain') return new URL(currentUrl).hostname.includes(b.value); - if (b.type === 'path') return new URL(currentUrl).pathname.startsWith(b.value); - if (b.type === 'url') return currentUrl.startsWith(b.value); - } catch {} - return false; - }); - if (!ok) { - return { - runId, - success: false, - summary: { total: 0, success: 0, failed: 0, tookMs: 0 }, - url: currentUrl, - outputs: null, - logs: [ - { - stepId: 'binding-check', - status: 'failed', - message: - 'Flow binding mismatch. Provide startUrl or open a page matching flow.meta.bindings.', - }, - ], - screenshots: { onFailure: null }, - } as RunResult; - } - } - } catch {} - - // long-running network capture (debugger) if requested - let networkCaptureStarted = false; - const stopAndSummarizeNetwork = async () => { - try { - const stopRes = await handleCallTool({ - name: TOOL_NAMES.BROWSER.NETWORK_DEBUGGER_STOP, - args: {}, - }); - const text = (stopRes?.content || []).find((c: any) => c.type === 'text')?.text; - if (!text) return; - const data = JSON.parse(text); - const requests: any[] = Array.isArray(data?.requests) ? data.requests : []; - const snippets = requests - .filter((r) => ['XHR', 'Fetch'].includes(String(r.type))) - .slice(0, 10) - .map((r) => ({ - method: String(r.method || 'GET'), - url: String(r.url || ''), - status: r.statusCode || r.status, - ms: Math.max(0, (r.responseTime || 0) - (r.requestTime || 0)), - })); - logger.push({ - stepId: 'network-capture', - status: 'success', - message: `Captured ${Number(data?.requestCount || 0)} requests` as any, - networkSnippets: snippets, - } as any); - } catch {} - }; - if (options.captureNetwork) { - try { - const res = await handleCallTool({ - name: TOOL_NAMES.BROWSER.NETWORK_DEBUGGER_START, - args: { includeStatic: false, maxCaptureTime: 3 * 60_000, inactivityTimeout: 0 }, - }); - if (!(res as any)?.isError) networkCaptureStarted = true; } catch {} + + // network capture start + if (this.options.captureNetwork) { + try { + const res = await handleCallTool({ + name: TOOL_NAMES.BROWSER.NETWORK_DEBUGGER_START, + args: { includeStatic: false, maxCaptureTime: 3 * 60_000, inactivityTimeout: 0 }, + }); + if (!(res as any)?.isError) this.networkCaptureStarted = true; + } catch {} + } + + // build DAG steps + const hasDag = Array.isArray((this.flow as any).nodes) && (this.flow as any).nodes.length > 0; + if (!hasDag) { + this.prepareError = { + runId: this.runId, + success: false, + summary: { total: 0, success: 0, failed: 0, tookMs: 0 }, + url: null, + outputs: null, + logs: [ + { + stepId: 'dag-required', + status: 'failed', + message: + 'Flow has no DAG nodes. Linear steps are no longer supported. Please migrate this flow to nodes/edges.', + }, + ], + screenshots: { onFailure: null }, + paused: false, + }; + return; + } + const nodes = ((this.flow as any).nodes || []) as any[]; + const edges = ((this.flow as any).edges || []) as any[]; + const defaultEdges = defaultEdgesOnly(edges as any); + const order = topoOrder(nodes as any, defaultEdges as any); + this.steps = order.map((n) => mapDagNodeToStep(n as any)); } - // DAG required - const hasDag = Array.isArray((flow as any).nodes) && (flow as any).nodes.length > 0; - if (!hasDag) { - await logger.overlayDone(); - return { - runId, - success: false, - summary: { total: 0, success: 0, failed: 0, tookMs: 0 }, - url: null, - outputs: null, - logs: [ - { - stepId: 'dag-required', - status: 'failed', - message: - 'Flow has no DAG nodes. Linear steps are no longer supported. Please migrate this flow to nodes/edges.', - }, - ], - screenshots: { onFailure: null }, - } as RunResult; - } - - const nodes = ((flow as any).nodes || []) as any[]; - const edges = ((flow as any).edges || []) as any[]; - const defaultEdges = defaultEdgesOnly(edges as any); - const order = topoOrder(nodes as any, defaultEdges as any); - const stepsToRun: Step[] = order.map((n) => mapDagNodeToStep(n as any)); - const steps = stepsToRun; - - let failed = 0; - let paused = false; - const ctx: ExecCtx = { vars, logger: (e: RunLogEntry) => logger.push(e) }; - const pendingAfterScripts: StepScript[] = []; - - const appendOverlayOk = (s: Step) => logger.overlayAppend(`✔ ${s.type} (${s.id})`); - const appendOverlayFail = (s: Step, err: any) => - logger.overlayAppend(`✘ ${s.type} (${s.id}) -> ${err?.message || String(err)}`); - - const evalCondition = (cond: any): boolean => { + private async executeSingleStep( + step: Step, + ctx: ExecCtx, + appendOverlayOk: (s: Step) => Promise | void, + appendOverlayFail: (s: Step, e: any) => Promise | void, + ): Promise<{ status: 'success' | 'failed' | 'paused'; nextLabel?: string; control?: any }> { + const t0 = Date.now(); + this.ensureWithinDeadline(); + const ctrlStart = await this.pluginManager.beforeStep({ + runId: this.runId, + flow: this.flow, + vars: this.vars, + step, + }); + if (ctrlStart?.pause) return { status: 'paused' }; + let stepNextLabel: string | undefined; + let controlOut: any = undefined; + const beforeInfo = await getActiveTabInfo(); try { - if (cond && typeof cond.expression === 'string' && cond.expression.trim()) { - const fn = new Function( - 'vars', - `try { return !!(${cond.expression}); } catch (e) { return false; }`, - ); - return !!fn(vars); - } - if (cond && typeof cond.var === 'string') { - const v = vars[cond.var]; - if ('equals' in cond) return String(v) === String(cond.equals); - return !!v; - } - } catch {} - return false; - }; - - const runSubflowById = async (subflowId: string) => { - const sub = (flow.subflows || {})[subflowId]; - if (!sub || !Array.isArray(sub.nodes) || sub.nodes.length === 0) return; - await pluginManager.subflowStart({ runId, flow, vars, subflowId }); - const sNodes: any[] = sub.nodes; - const sEdges: any[] = defaultEdgesOnly((sub.edges || []) as any) as any[]; - const sOrder = topoOrder(sNodes as any, sEdges as any); - const sSteps: Step[] = sOrder.map((n) => mapDagNodeToStep(n as any)) as any; - for (const step of sSteps) { - const t0 = Date.now(); - ensureWithinDeadline(); - const ctrl = await pluginManager.beforeStep({ runId, flow, vars, step }); - if (ctrl?.pause) { - paused = true; - break; - } await withRetry( async () => { - const beforeInfo = await getActiveTabInfo(); - const result = await executeStep(ctx, step); + const result: any = await executeStep(ctx, step); if (step.type === 'click' || step.type === 'dblclick') { const after = ((step as any).after || {}) as any; if (after.waitForNavigation) @@ -309,15 +279,34 @@ export async function runFlow(flow: Flow, options: RunOptions = {}): Promise { - logger.push({ stepId: step.id, status: 'retrying', message: e?.message || String(e) }); - await pluginManager.onRetry({ runId, flow, vars, step, error: e, attempt }); + this.logger.push({ + stepId: step.id, + status: 'retrying', + message: e?.message || String(e), + }); + await this.pluginManager.onRetry({ + runId: this.runId, + flow: this.flow, + vars: this.vars, + step, + error: e, + attempt, + }); }, { count: Math.max(0, (step as any).retry?.count ?? 0), @@ -325,13 +314,70 @@ export async function runFlow(flow: Flow, options: RunOptions = {}): Promise mapDagNodeToStep(n as any)) as any; + const ok = (s: Step) => this.logger.overlayAppend(`✔ ${s.type} (${s.id})`); + const fail = (s: Step, e: any) => + this.logger.overlayAppend(`✘ ${s.type} (${s.id}) -> ${e?.message || String(e)}`); + for (const step of sSteps) { + this.ensureWithinDeadline(); + const r = await this.executeSingleStep(step, ctx, ok, fail); + if (r.status === 'paused') { + this.paused = true; + break; + } + if (this.paused) break; + } + await this.pluginManager.subflowEnd({ + runId: this.runId, + flow: this.flow, + vars: this.vars, + subflowId, + }); + } + + private async traverseDag(): Promise { + if (!this.steps.length) { + await this.logger.overlayDone(); + return this.prepareError!; + } + const nodes = ((this.flow as any).nodes || []) as any[]; + const edges = ((this.flow as any).edges || []) as any[]; const id2node = new Map(nodes.map((n: any) => [n.id, n] as const)); const outEdges = new Map>(); for (const e of edges) { @@ -341,158 +387,167 @@ export async function runFlow(flow: Flow, options: RunOptions = {}): Promise(nodes.map((n: any) => [n.id, 0] as const)); for (const e of edges) indeg.set(e.to, (indeg.get(e.to) || 0) + 1); let currentId = - options.startNodeId && id2node.has(options.startNodeId) - ? options.startNodeId + this.options.startNodeId && id2node.has(this.options.startNodeId) + ? this.options.startNodeId : nodes.find((n: any) => (indeg.get(n.id) || 0) === 0)?.id || nodes[0]?.id; let guard = 0; + const ctx: ExecCtx = { vars: this.vars, logger: (e: RunLogEntry) => this.logger.push(e) }; while (currentId && guard++ < 10000) { - ensureWithinDeadline(); + this.ensureWithinDeadline(); const node = id2node.get(currentId); if (!node) break; const step: any = mapDagNodeToStep(node as any); - const t0 = Date.now(); - const beforeInfo = await getActiveTabInfo(); - let jumpedOnError = false; - let stepNextLabel: string | undefined; - const ctrlStart = await pluginManager.beforeStep({ runId, flow, vars, step }); - if (ctrlStart?.pause) { - paused = true; + const r = await this.executeSingleStep( + step, + ctx, + (s) => this.logger.overlayAppend(`✔ ${s.type} (${s.id})`), + (s, e) => this.logger.overlayAppend(`✘ ${s.type} (${s.id}) -> ${e?.message || String(e)}`), + ); + if (r.status === 'paused') { + this.paused = true; break; } - try { - await withRetry( - async () => { - const result = await executeStep(ctx, step); - if (step.type === 'click' || step.type === 'dblclick') { - const after = ((step as any).after || {}) as any; - if (after.waitForNavigation) - await waitForNavigationDone(beforeInfo.url, (step as any).timeoutMs); - else if (after.waitForNetworkIdle) - await waitForNetworkIdle(Math.min((step as any).timeoutMs || 5000, 120000), 1200); - else await maybeQuickWaitForNav(beforeInfo.url, (step as any).timeoutMs); - } - if (step.type === 'navigate' || step.type === 'openTab') { - await waitForNavigationDone(beforeInfo.url, (step as any).timeoutMs); - await ensureReadPageIfWeb(); - } else if (step.type === 'switchTab') { - await ensureReadPageIfWeb(); - } - if (!result?.alreadyLogged) - logger.push({ stepId: step.id, status: 'success', tookMs: Date.now() - t0 }); - await pluginManager.afterStep({ runId, flow, vars, step, result }); - await appendOverlayOk(step); - if (result?.nextLabel) stepNextLabel = String(result.nextLabel); - if (result?.control) { - if (result.control.kind === 'foreach') { - const list = Array.isArray(vars[result.control.listVar]) - ? (vars[result.control.listVar] as any[]) - : []; - for (const it of list) { - vars[result.control.itemVar] = it; - await runSubflowById(result.control.subflowId); - } - } else if (result.control.kind === 'while') { - let i = 0; - while ( - i < result.control.maxIterations && - evalCondition(result.control.condition) - ) { - await runSubflowById(result.control.subflowId); - i++; - } - } - } - if (result?.deferAfterScript) pendingAfterScripts.push(result.deferAfterScript); - await flushAfterScripts(ctx, pendingAfterScripts, vars, logger); - }, - async (attempt, e) => { - logger.push({ stepId: step.id, status: 'retrying', message: e?.message || String(e) }); - await pluginManager.onRetry({ runId, flow, vars, step, error: e, attempt }); - }, - { - count: Math.max(0, (step as any).retry?.count ?? 0), - intervalMs: Math.max(0, (step as any).retry?.intervalMs ?? 0), - backoff: (step as any).retry?.backoff || 'none', - }, - ); - } catch (e: any) { - failed++; - logger.push({ - stepId: step.id, - status: 'failed', - message: e?.message || String(e), - tookMs: Date.now() - t0, - }); - await appendOverlayFail(step, e); - if ((step as any).screenshotOnFail !== false) await logger.screenshotOnFailure(); - const hook = await pluginManager.onError({ runId, flow, vars, step, error: e }); - if (hook?.pause) { - paused = true; - break; - } + if (r.status === 'failed') { const oes = (outEdges.get(currentId) || []) as any[]; const errEdge = oes.find((edg) => edg.label === 'onError'); if (errEdge) { currentId = errEdge.to; - jumpedOnError = true; + continue; } else { - throw e; + break; } } - if (paused) break; - if (!jumpedOnError) { - // choose next by label from success path - let nextLabel: string = stepNextLabel ? String(stepNextLabel) : 'default'; - const override = await pluginManager.onChooseNextLabel({ - runId, - flow, - vars, - step, - suggested: nextLabel, - }); - if (override) nextLabel = String(override); - const oes = (outEdges.get(currentId) || []) as any[]; - const edge = - oes.find((e) => String(e.label || 'default') === nextLabel) || - oes.find((e) => !e.label || e.label === 'default'); - currentId = edge ? edge.to : undefined; + if ((r as any).control) { + const control = (r as any).control; + if (control?.kind === 'foreach') { + const list = Array.isArray(this.vars[control.listVar]) + ? (this.vars[control.listVar] as any[]) + : []; + for (const it of list) { + this.vars[control.itemVar] = it; + await this.runSubflowById(control.subflowId, ctx); + if (this.paused) break; + } + } else if (control?.kind === 'while') { + let i = 0; + while (i < control.maxIterations && this.evalCondition(control.condition)) { + await this.runSubflowById(control.subflowId, ctx); + if (this.paused) break; + i++; + } + } + if (this.paused) break; } + // choose next by label + let nextLabel: string = r.nextLabel ? String(r.nextLabel) : 'default'; + const override = await this.pluginManager.onChooseNextLabel({ + runId: this.runId, + flow: this.flow, + vars: this.vars, + step, + suggested: nextLabel, + }); + if (override) nextLabel = String(override); + const oes = (outEdges.get(currentId) || []) as any[]; + const edge = + oes.find((e) => String(e.label || 'default') === nextLabel) || + oes.find((e) => !e.label || e.label === 'default'); + currentId = edge ? edge.to : undefined; } - } finally { - if (networkCaptureStarted) await stopAndSummarizeNetwork(); + const tookMs = Date.now() - this.startAt; + const sensitiveKeys = new Set( + (this.flow.variables || []).filter((v) => v.sensitive).map((v) => v.key), + ); + const outputs: Record = {}; + for (const [k, v] of Object.entries(this.vars)) if (!sensitiveKeys.has(k)) outputs[k] = v; + return { + runId: this.runId, + success: !this.paused && this.failed === 0, + summary: { + total: this.steps.length, + success: this.steps.length - this.failed, + failed: this.failed, + tookMs, + }, + url: null, + outputs, + logs: this.options.returnLogs ? this.logger.getLogs() : undefined, + screenshots: { + onFailure: this.logger.getLogs().find((l) => l.status === 'failed')?.screenshotBase64, + }, + paused: this.paused, + }; } - await logger.overlayDone(); - const tookMs = Date.now() - startAt; + private evalCondition(cond: any): boolean { + try { + if (cond && typeof cond.expression === 'string' && cond.expression.trim()) { + const fn = new Function( + 'vars', + `try { return !!(${cond.expression}); } catch (e) { return false; }`, + ); + return !!fn(this.vars); + } + if (cond && typeof cond.var === 'string') { + const v = this.vars[cond.var]; + if ('equals' in cond) return String(v) === String(cond.equals); + return !!v; + } + } catch {} + return false; + } - // outputs: filter sensitive variables - const sensitiveKeys = new Set( - (flow.variables || []).filter((v) => v.sensitive).map((v) => v.key), - ); - const outputs: Record = {}; - for (const [k, v] of Object.entries(vars)) if (!sensitiveKeys.has(k)) outputs[k] = v; + private async cleanup() { + if (this.networkCaptureStarted) { + try { + const stopRes = await handleCallTool({ + name: TOOL_NAMES.BROWSER.NETWORK_DEBUGGER_STOP, + args: {}, + }); + const text = (stopRes?.content || []).find((c: any) => c.type === 'text')?.text; + if (text) { + const data = JSON.parse(text); + const requests: any[] = Array.isArray(data?.requests) ? data.requests : []; + const snippets = requests + .filter((r) => ['XHR', 'Fetch'].includes(String(r.type))) + .slice(0, 10) + .map((r) => ({ + method: String(r.method || 'GET'), + url: String(r.url || ''), + status: r.statusCode || r.status, + ms: Math.max(0, (r.responseTime || 0) - (r.requestTime || 0)), + })); + this.logger.push({ + stepId: 'network-capture', + status: 'success', + message: `Captured ${Number(data?.requestCount || 0)} requests` as any, + networkSnippets: snippets, + } as any); + } + } catch {} + } + await this.logger.overlayDone(); + try { + await this.pluginManager.runEnd({ + runId: this.runId, + flow: this.flow, + vars: this.vars, + success: this.failed === 0 && !this.paused, + failed: this.failed, + }); + if (!this.paused) await this.logger.persist(this.flow, this.startAt, this.failed === 0); + await runState.update(this.runId, { + status: this.paused ? 'stopped' : this.failed === 0 ? 'completed' : 'failed', + updatedAt: Date.now(), + } as any); + if (!this.paused) await runState.delete(this.runId); + } catch {} + } +} - await pluginManager.runEnd({ runId, flow, vars, success: failed === 0 && !paused, failed }); - if (!paused) await logger.persist(flow, startAt, failed === 0); - try { - await runState.update(runId, { - status: paused ? 'stopped' : failed === 0 ? 'completed' : 'failed', - updatedAt: Date.now(), - } as any); - if (!paused) await runState.delete(runId); - } catch {} - return { - runId, - success: !paused && failed === 0, - summary: { total: steps.length, success: steps.length - failed, failed, tookMs }, - url: null, - outputs, - logs: options.returnLogs ? logger.getLogs() : undefined, - screenshots: { - onFailure: logger.getLogs().find((l) => l.status === 'failed')?.screenshotBase64, - }, - paused, - }; +export async function runFlow(flow: Flow, options: RunOptions = {}): Promise { + const orchestrator = new ExecutionOrchestrator(flow, options); + return await orchestrator.run(); } async function getActiveTabInfo() { @@ -529,7 +584,7 @@ async function flushAfterScripts( args: [code], world: world as any, } as any); - if ((s as any).saveAs) vars[(s as any).saveAs] = result; + if ((s as any).saveAs) (vars as any)[(s as any).saveAs] = result; if ((s as any).assign && typeof (s as any).assign === 'object') applyAssign(vars, result, (s as any).assign); } diff --git a/app/chrome-extension/entrypoints/background/record-replay/nodes/index.ts b/app/chrome-extension/entrypoints/background/record-replay/nodes/index.ts index 9562f2f..c197e72 100644 --- a/app/chrome-extension/entrypoints/background/record-replay/nodes/index.ts +++ b/app/chrome-extension/entrypoints/background/record-replay/nodes/index.ts @@ -53,12 +53,12 @@ const registry = new Map>([ ]); export async function executeStep(ctx: ExecCtx, step: Step): Promise { - const rt = registry.get((step as any).type); - if (!rt) throw new Error(`unsupported step type: ${String((step as any).type)}`); - const v = rt.validate ? rt.validate(step as any) : { ok: true }; - if (!(v as any).ok) throw new Error(((v as any).errors || []).join(', ') || 'validation failed'); - const out = await rt.run(ctx as any, step as any); - return (out || {}) as ExecResult; + const rt = registry.get(step.type); + if (!rt) throw new Error(`unsupported step type: ${String(step.type)}`); + const v = rt.validate ? rt.validate(step) : { ok: true }; + if (!v.ok) throw new Error((v.errors || []).join(', ') || 'validation failed'); + const out = await rt.run(ctx, step); + return out || {}; } export type { ExecCtx, ExecResult, NodeRuntime } from './types';