mirror of
https://github.com/hangwin/mcp-chrome.git
synced 2026-10-31 03:02:28 +08:00
chore: 重构回放引擎
This commit is contained in:
hangerye
@@ -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<RunResult> {
|
||||
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<string, any> = 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<string, any> = 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<RunResult> {
|
||||
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<string, any> | 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<string, any> | 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> | void,
|
||||
appendOverlayFail: (s: Step, e: any) => Promise<void> | 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<Run
|
||||
await ensureReadPageIfWeb();
|
||||
}
|
||||
if (!result?.alreadyLogged)
|
||||
logger.push({ stepId: step.id, status: 'success', tookMs: Date.now() - t0 });
|
||||
await pluginManager.afterStep({ runId, flow, vars, step, result });
|
||||
this.logger.push({ stepId: step.id, status: 'success', tookMs: Date.now() - t0 });
|
||||
await this.pluginManager.afterStep({
|
||||
runId: this.runId,
|
||||
flow: this.flow,
|
||||
vars: this.vars,
|
||||
step,
|
||||
result,
|
||||
});
|
||||
await appendOverlayOk(step);
|
||||
if (result?.deferAfterScript) pendingAfterScripts.push(result.deferAfterScript);
|
||||
await flushAfterScripts(ctx, pendingAfterScripts, vars, logger);
|
||||
if (result?.nextLabel) stepNextLabel = String(result.nextLabel);
|
||||
if (result?.control) controlOut = result.control;
|
||||
if (result?.deferAfterScript) this.pendingAfterScripts.push(result.deferAfterScript);
|
||||
await flushAfterScripts(ctx, this.pendingAfterScripts, this.vars, this.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 });
|
||||
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<Run
|
||||
backoff: (step as any).retry?.backoff || 'none',
|
||||
},
|
||||
);
|
||||
if (paused) break;
|
||||
} catch (e: any) {
|
||||
this.failed++;
|
||||
this.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 this.logger.screenshotOnFailure();
|
||||
const hook = await this.pluginManager.onError({
|
||||
runId: this.runId,
|
||||
flow: this.flow,
|
||||
vars: this.vars,
|
||||
step,
|
||||
error: e,
|
||||
});
|
||||
if (hook?.pause) return { status: 'paused' };
|
||||
return { status: 'failed' };
|
||||
}
|
||||
await pluginManager.subflowEnd({ runId, flow, vars, subflowId });
|
||||
};
|
||||
// Propagate control (foreach/while) to caller
|
||||
return { status: 'success', nextLabel: stepNextLabel, control: controlOut };
|
||||
}
|
||||
|
||||
try {
|
||||
// DAG traversal (single-branch by nextLabel)
|
||||
private async runSubflowById(subflowId: string, ctx: ExecCtx) {
|
||||
const sub = (this.flow.subflows || {})[subflowId];
|
||||
if (!sub || !Array.isArray(sub.nodes) || sub.nodes.length === 0) return;
|
||||
await this.pluginManager.subflowStart({
|
||||
runId: this.runId,
|
||||
flow: this.flow,
|
||||
vars: this.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;
|
||||
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<RunResult> {
|
||||
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<string, Array<any>>();
|
||||
for (const e of edges) {
|
||||
@@ -341,158 +387,167 @@ export async function runFlow(flow: Flow, options: RunOptions = {}): Promise<Run
|
||||
const indeg = new Map<string, number>(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<string, any> = {};
|
||||
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<string, any> = {};
|
||||
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<RunResult> {
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -53,12 +53,12 @@ const registry = new Map<string, NodeRuntime<any>>([
|
||||
]);
|
||||
|
||||
export async function executeStep(ctx: ExecCtx, step: Step): Promise<ExecResult> {
|
||||
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';
|
||||
|
||||
Reference in New Issue
Block a user