From 3e0044983c037fb7282f1a032a4151a16aed591f Mon Sep 17 00:00:00 2001 From: Dwifax Date: Fri, 26 Jun 2026 01:10:38 +0800 Subject: [PATCH 1/3] feat: per-step AbortController timeout & signal propagation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add STEP_TIMEOUT_MS env var (default 30s). Create AbortController per step with setTimeout→abort(). Catch AbortError as timed-out step. Guard vault releasePayment with signal?.aborted to avoid wasting on-chain escrow. Propagate AbortSignal through x402-client, mpp-client, and checkHealth. Document STEP_TIMEOUT_MS in .env.example. --- .env.example | 5 +++ packages/orchestrator/src/executor.ts | 53 ++++++++++++++++++++++-- packages/orchestrator/src/mpp-client.ts | 3 ++ packages/orchestrator/src/x402-client.ts | 6 +++ 4 files changed, 63 insertions(+), 4 deletions(-) diff --git a/.env.example b/.env.example index 75266fd..f3b84b3 100644 --- a/.env.example +++ b/.env.example @@ -118,6 +118,11 @@ DEFAULT_BUDGET=1.0 # the orchestrator auto-approves it and proceeds. Default is 60 seconds. PLAN_APPROVAL_TIMEOUT_MS=60000 +# Per-step execution timeout in ms. When a specialist agent step exceeds this, +# the AbortController fires, the step is marked failed, and parallel steps +# continue normally. Default 30 seconds. +STEP_TIMEOUT_MS=30000 + # URL of the Stellar Sponsored Agent Account service — used by the orchestrator # to provision new agent wallets without exposing secret keys to the service. SPONSOR_SERVICE_URL=https://stellar-sponsored-agent-account.onrender.com diff --git a/packages/orchestrator/src/executor.ts b/packages/orchestrator/src/executor.ts index 2345803..04d0395 100644 --- a/packages/orchestrator/src/executor.ts +++ b/packages/orchestrator/src/executor.ts @@ -53,6 +53,8 @@ export interface ExecutorEvents { task_complete: { task_id: string; status: string; total_cost: number; total_time_ms: number }; } +const STEP_TIMEOUT_MS = parseInt(process.env.STEP_TIMEOUT_MS ?? '30000', 10); + // ── Helpers ────────────────────────────────────────────────────────────────── function buildDependencyLevels(steps: ExecutionStep[]): number[][] { @@ -87,14 +89,16 @@ function normaliseDeps(depends_on: number | number[] | null): number[] { return [depends_on]; } -async function checkHealth(agent: AgentRecord): Promise { +async function checkHealth(agent: AgentRecord, signal?: AbortSignal): Promise { // Render free tier cold-starts: service returns 503 immediately, then takes ~50-60s to wake. // Poll every 10s for up to ~90s total so we catch the service after it finishes starting. const delays = [0, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000]; // 9 attempts, ~80s total wait for (let attempt = 0; attempt < delays.length; attempt++) { + // Bail early if the outer step timeout fired — do not waste time on further polls + if (signal?.aborted) return false; if (delays[attempt] > 0) await new Promise((r) => setTimeout(r, delays[attempt])); try { - const response = await fetch(agent.health_check, { signal: AbortSignal.timeout(15000) }); + const response = await fetch(agent.health_check, { signal: signal ?? AbortSignal.timeout(15000) }); if (response.ok) return true; // 503/502 = still sleeping/starting, keep retrying; any 4xx = genuinely down if (response.status !== 503 && response.status !== 502) return false; @@ -148,7 +152,33 @@ export class PlanExecutor extends EventEmitter { const levelSteps = level.map((id) => stepMap.get(id)!); const results = await Promise.all( - levelSteps.map((step) => this.executeStep(step, task_id, stepResultMap, registryUrl)), + levelSteps.map((step) => { + const stepStart = Date.now(); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), STEP_TIMEOUT_MS); + + return this.executeStep( + step, + task_id, + stepResultMap, + registryUrl, + controller.signal, + ) + .catch((err) => { + if ((err as Error).name === 'AbortError') { + // The signal was aborted either by the outer timer or by a + // propagated AbortError from an internal fetch — in both + // cases the step timed out. + return this.makeFailedResult( + step, + `Step ${step.step_id} timed out after ${STEP_TIMEOUT_MS}ms`, + Date.now() - stepStart, + ); + } + throw err; + }) + .finally(() => clearTimeout(timer)); + }), ); for (const result of results) { @@ -201,6 +231,7 @@ export class PlanExecutor extends EventEmitter { task_id: string, previousResults: Map, _registryUrl: string, + signal?: AbortSignal, ): Promise { const agent = this.agentMap.get(step.agent_id); const stepStart = Date.now(); @@ -235,7 +266,7 @@ export class PlanExecutor extends EventEmitter { return result; } - const healthy = await checkHealth(agent); + const healthy = await checkHealth(agent, signal); if (!healthy) { const latency_ms = Date.now() - stepStart; const result = this.makeFailedResult( @@ -255,6 +286,18 @@ export class PlanExecutor extends EventEmitter { try { const amountUsdc = agent.pricing.price_per_call; + // Bail before vault release if this step was already aborted — + // releasing on-chain escrow for a dead step wastes gas and USDC + if (signal?.aborted) { + const latency_ms = Date.now() - stepStart; + const result = this.makeFailedResult( + step, + `Step ${step.step_id} aborted before payment`, + latency_ms, + ); + return result; + } + // ── Vault release: contract → orchestrator (serialized to avoid sequence conflicts) let releaseHash: string | null = null; if (VAULT_ACTIVE && this.orchestratorKeypair && this.vaultTaskId !== null) { @@ -306,6 +349,7 @@ export class PlanExecutor extends EventEmitter { step.action, context || undefined, orchestratorSecret, + signal, ); output = x402Result.output; tx_hash = x402Result.tx_hash; @@ -316,6 +360,7 @@ export class PlanExecutor extends EventEmitter { { data: context || '' }, step.action, orchestratorSecret, + signal, ); output = mppResult.output; tx_hash = mppResult.tx_hash; diff --git a/packages/orchestrator/src/mpp-client.ts b/packages/orchestrator/src/mpp-client.ts index 8e9e2c0..e1558c8 100644 --- a/packages/orchestrator/src/mpp-client.ts +++ b/packages/orchestrator/src/mpp-client.ts @@ -34,12 +34,14 @@ export interface MPPResult { * @param data Arbitrary payload object to POST * @param instruction Instruction/query for the agent * @param secretKey Secret key to sign payments (defaults to ORCHESTRATOR_SECRET_KEY) + * @param signal Optional AbortSignal — when aborted, the call is cancelled immediately */ export async function makeMPPPayment( endpoint: string, data: Record, instruction: string, secretKey?: string, + signal?: AbortSignal, ): Promise { const key = secretKey ?? process.env.ORCHESTRATOR_SECRET_KEY!; const mppFetch = buildMPPFetch(key); @@ -48,6 +50,7 @@ export async function makeMPPPayment( method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ data, instruction }), + signal, }); if (!response.ok) { diff --git a/packages/orchestrator/src/x402-client.ts b/packages/orchestrator/src/x402-client.ts index 395a1f1..ae12cb5 100644 --- a/packages/orchestrator/src/x402-client.ts +++ b/packages/orchestrator/src/x402-client.ts @@ -37,17 +37,22 @@ export interface X402Result { * @param action Instruction string for the agent * @param context Output from previous steps to pass as context * @param secretKey Secret key to sign payments (defaults to ORCHESTRATOR_SECRET_KEY) + * @param signal Optional AbortSignal — when aborted, the call is cancelled immediately */ export async function makeX402Payment( endpoint: string, action: string, context?: string, secretKey?: string, + signal?: AbortSignal, ): Promise { const key = secretKey ?? process.env.ORCHESTRATOR_SECRET_KEY!; let lastError: Error | null = null; for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + // Bail early if the outer step timeout has already fired + if (signal?.aborted) throw new DOMException('The operation was aborted', 'AbortError'); + // Fresh signer + scheme on every attempt — avoids stale sequence numbers const payingFetch = buildPayingFetch(key); @@ -56,6 +61,7 @@ export async function makeX402Payment( method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ query: action, context: context ?? '' }), + signal, }); if (!response.ok) { From 17a74700d4daaf2bfc4d092433e9c30595475609 Mon Sep 17 00:00:00 2001 From: Dwifax Date: Fri, 26 Jun 2026 16:53:31 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix:=20address=20CodeRabbit=20review=20?= =?UTF-8?q?=E2=80=94=20NaN=20guard,=20abort-aware=20retry,=20release=20loc?= =?UTF-8?q?k=20abort=20check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/orchestrator/src/executor.ts | 6 +++++- packages/orchestrator/src/x402-client.ts | 3 +++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/packages/orchestrator/src/executor.ts b/packages/orchestrator/src/executor.ts index 04d0395..9051bf1 100644 --- a/packages/orchestrator/src/executor.ts +++ b/packages/orchestrator/src/executor.ts @@ -53,7 +53,8 @@ export interface ExecutorEvents { task_complete: { task_id: string; status: string; total_cost: number; total_time_ms: number }; } -const STEP_TIMEOUT_MS = parseInt(process.env.STEP_TIMEOUT_MS ?? '30000', 10); +const _parsed = parseInt(process.env.STEP_TIMEOUT_MS ?? '30000', 10); +const STEP_TIMEOUT_MS = Number.isFinite(_parsed) && _parsed > 0 ? _parsed : 30000; // ── Helpers ────────────────────────────────────────────────────────────────── @@ -302,6 +303,9 @@ export class PlanExecutor extends EventEmitter { let releaseHash: string | null = null; if (VAULT_ACTIVE && this.orchestratorKeypair && this.vaultTaskId !== null) { const released = await this.releaseSequential(async () => { + // Re-check after acquiring the serialization lock — the step may + // have timed out while we were waiting for a prior release. + if (signal?.aborted) return null; return releasePayment(this.orchestratorKeypair!, this.vaultTaskId!, amountUsdc); }); diff --git a/packages/orchestrator/src/x402-client.ts b/packages/orchestrator/src/x402-client.ts index ae12cb5..f3d1d9c 100644 --- a/packages/orchestrator/src/x402-client.ts +++ b/packages/orchestrator/src/x402-client.ts @@ -92,6 +92,9 @@ export async function makeX402Payment( return { output, tx_hash }; } catch (err: any) { lastError = err; + // Short-circuit retries once the step timeout has fired — no point + // backing off when every subsequent fetch will abort immediately. + if (signal?.aborted) break; if (attempt < MAX_ATTEMPTS) { const backoffMs = 2000 * attempt; console.warn( From c822de35176917698b9c7ba89b4cf067336cd04a Mon Sep 17 00:00:00 2001 From: Dwifax Date: Sun, 28 Jun 2026 23:21:49 +0800 Subject: [PATCH 3/3] fix: emit step_failed on timeout + abort-aware health-check sleep --- packages/orchestrator/src/executor.ts | 26 ++++++++++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/packages/orchestrator/src/executor.ts b/packages/orchestrator/src/executor.ts index 9051bf1..2317bc6 100644 --- a/packages/orchestrator/src/executor.ts +++ b/packages/orchestrator/src/executor.ts @@ -97,7 +97,22 @@ async function checkHealth(agent: AgentRecord, signal?: AbortSignal): Promise 0) await new Promise((r) => setTimeout(r, delays[attempt])); + if (delays[attempt] > 0) { + // Abort-aware sleep: resolve as soon as the signal fires instead of + // blocking for the full delay, so a timed-out step stops polling promptly. + const interrupted = await new Promise((resolve) => { + const t = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(false); + }, delays[attempt]); + const onAbort = () => { + clearTimeout(t); + resolve(true); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); + if (interrupted) return false; + } try { const response = await fetch(agent.health_check, { signal: signal ?? AbortSignal.timeout(15000) }); if (response.ok) return true; @@ -170,11 +185,18 @@ export class PlanExecutor extends EventEmitter { // The signal was aborted either by the outer timer or by a // propagated AbortError from an internal fetch — in both // cases the step timed out. - return this.makeFailedResult( + const result = this.makeFailedResult( step, `Step ${step.step_id} timed out after ${STEP_TIMEOUT_MS}ms`, Date.now() - stepStart, ); + this.emit('step_failed', { + task_id, + step_id: step.step_id, + agent_name: step.agent_name, + error: result.error!, + }); + return result; } throw err; })