From 5eee931d268b168d2defd88a37d95eeb605d5ccd Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 21 Oct 2026 07:15:00 +0200 Subject: [PATCH 1/2] fix(surface): keep waitForSignal's transport out of the authored gate on Node 22 A hosted waitForSignal that saw one held 204 failed the run after the body finished: "derived handler for run-1 (f.run) rejected: The operation was aborted due to timeout", or unsettled_derived_work when the request deadline had not fired yet. Hosted sandboxes run the authored body on Node 22, whose bundled undici builds an unreachable body stream even for a 204: its internal promise stays pending, and the request deadline aborting after the response rejects it. The authored completion gate counts every promise created in the body's async context after an f.run as that step's derived work. Run each held request in the module's load context (an AsyncResource), so the transport's internal promises are not attributed to the author. The promise the body awaits is still created and tracked in the authored context and carries every request outcome. Each request's signal is also detached from the deadline and the caller's signal once it settles. No SDK gate rule changes; Node 24 and 26 never created the stream. Co-Authored-By: Claude Opus 5.5 --- .../authored-flow-lifecycle-executor.test.ts | 48 +++++++++++++++ packages/surface/src/signals.ts | 57 ++++++++++++++++-- packages/surface/tests/signals.test.ts | 58 +++++++++++++++++++ 3 files changed, 159 insertions(+), 4 deletions(-) diff --git a/packages/sdk/tests/authored-flow-lifecycle-executor.test.ts b/packages/sdk/tests/authored-flow-lifecycle-executor.test.ts index 71524c50..79eade13 100644 --- a/packages/sdk/tests/authored-flow-lifecycle-executor.test.ts +++ b/packages/sdk/tests/authored-flow-lifecycle-executor.test.ts @@ -1,6 +1,7 @@ import { rmSync } from 'node:fs'; import type { Server } from 'node:net'; import { flow, type Ctx, type FlowHandle, type Step } from '@relayflows/surface'; +import { waitForSignal } from '@relayflows/surface/signals'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { executeAuthoredFlow, type ExecuteAuthoredFlowOptions } from '../src/authored-flow-executor.js'; import { AuthoredFlowLifecycle, parallelPromiseIntrinsicsIntact } from '../src/authored-flow-lifecycle.js'; @@ -716,6 +717,53 @@ describe('authored flow lifecycle through the journal executor', () => { expect(result.journalSteps.map((step) => step.id)).toEqual(['run-1', 'run-2', 'complete-3']); }); + // Hosted run 4e320b7a (Node 22): a signal posted 11 minutes after run-1 + // resumed the body, run-2 completed, and the gate then failed the run with + // "derived handler for run-1 (f.run) rejected: The operation was aborted due + // to timeout". Node 22's undici creates an unreachable body stream for a 204: + // its internal promise stays pending, and rejects when the request signal + // aborts after the response. This fetch does the same, so the test does not + // depend on the runtime's undici. Each case is one face of that failure: + // the deadline fires before the body returns (rejected), it has not fired + // yet (pending), and several held rounds before the signal lands. + it.each([ + { rounds: 1, requestTimeoutMs: 5 }, + { rounds: 1, requestTimeoutMs: 60_000 }, + { rounds: 3, requestTimeoutMs: 5 }, + ])('completes a signal wait after $rounds 204 round(s) with a $requestTimeoutMs ms request deadline', + async ({ rounds, requestTimeoutMs }) => { + const realFetch = globalThis.fetch; + let reads = 0; + let payload: unknown; + globalThis.fetch = (async (_input: unknown, init?: RequestInit) => { + reads += 1; + if (reads > rounds) return Response.json({ payload: { ok: true }, acceptedAt: 1 }); + const orphanedBodyStream = new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => reject(init.signal!.reason), { once: true }); + }); + orphanedBodyStream.catch(() => undefined); + return new Response(null, { status: 204 }); + }) as typeof fetch; + try { + const result = await execute(flow('long-signal-wait', async (f) => { + await f.run('true'); + payload = await waitForSignal('native_done', { + runId: 'run_1', apiUrl: 'https://cloud.example', token: 'read-token', + waitMs: 0, requestTimeoutMs, + }); + // Outlive the short request deadlines before the body finishes. + await new Promise((resolve) => setTimeout(resolve, 50)); + await f.run('printf resumed'); + f.done('success'); + })); + expect(reads).toBe(rounds + 1); + expect(payload).toEqual({ ok: true }); + expect(result.completionReason).toBe('success'); + } finally { + globalThis.fetch = realFetch; + } + }); + it('preserves direct await, Promise.resolve, and Promise.all authoring', async () => { const result = await execute(flow('supported-awaits', async (f) => { await f.run('true'); diff --git a/packages/surface/src/signals.ts b/packages/surface/src/signals.ts index 9938f464..39511b3e 100644 --- a/packages/surface/src/signals.ts +++ b/packages/surface/src/signals.ts @@ -11,8 +11,29 @@ * credential lives only in that process; `f.run` children and agents never * receive it. */ +import { AsyncResource } from "node:async_hooks"; import { setTimeout as delay } from "node:timers/promises"; +/** + * The async context every held request runs in: this module's load context, + * which is outside any authored flow body. + * + * The authored completion gate treats every promise created in the body's + * async context after an `f.run` as work derived from that step, and fails the + * run if one rejected or is still pending when the body returns. Node 22's + * bundled undici creates a body stream even for a 204, which has no body: the + * stream is unreachable, so its internal promise stays pending, and an abort + * after the response (the request deadline) rejects it. Run in the body's + * context, one 204 fails the flow with `unsettled_derived_work` or "derived + * handler for run-1 (f.run) rejected: The operation was aborted due to + * timeout". Node 24 creates no such stream. + * + * Those promises are the transport's, not the author's. The promise the body + * awaits — this helper's — is still created and tracked in the authored + * context, and every request outcome still reaches it. + */ +const requestContext = new AsyncResource("RelayflowsSignalRequest"); + /** The longest hold Cloud's signal GET accepts, and the default. */ export const SIGNAL_MAX_WAIT_MS = 25_000; /** Cloud's run ID for this hosted run (not the engine's `RELAYFLOW_RUN_ID`). */ @@ -130,6 +151,26 @@ function transportError(cause: unknown, error: SignalErrorFactory): Error { : error("transport_error", "Cloud transport failed; check TLS and the configured URL (redirects are refused)."); } +/** + * The abort signal for one held request, detached from the deadline and the + * caller's signal once that request has settled. Both outlive the request; an + * abort reaching a settled `fetch` makes Node 22's undici reject an internal + * promise of the answered response. + */ +function requestScope(deadline: AbortSignal, caller?: AbortSignal): { signal: AbortSignal; release(): void } { + const controller = new AbortController(); + const sources = caller === undefined ? [deadline] : [caller, deadline]; + const forwards = sources.map(source => () => controller.abort(source.reason)); + sources.forEach((source, index) => { + if (source.aborted) controller.abort(source.reason); + else source.addEventListener("abort", forwards[index]!, { once: true }); + }); + return { + signal: controller.signal, + release: () => sources.forEach((source, index) => source.removeEventListener("abort", forwards[index]!)), + }; +} + function normalizeCloudBaseUrl(raw: string): string | undefined { try { const url = new URL(raw); @@ -172,21 +213,24 @@ export async function holdSignalWait( options.signal?.throwIfAborted(); const { baseUrl, token } = await hooks.connection(); const deadline = AbortSignal.timeout(requestTimeout); + const request = requestScope(deadline, options.signal); const requestStartedAt = performance.now(); let response: Response; try { - response = await fetch(`${baseUrl}${path}?waitMs=${waitMs}`, { + response = await requestContext.runInAsyncScope(() => fetch(`${baseUrl}${path}?waitMs=${waitMs}`, { method: "GET", headers: { authorization: `Bearer ${token}` }, - signal: options.signal ? AbortSignal.any([options.signal, deadline]) : deadline, + signal: request.signal, // A redirect must never carry the credential to a different origin. redirect: "error", - }); + })); } catch (cause) { + request.release(); options.signal?.throwIfAborted(); throw transportError(deadline.aborted ? deadline.reason : cause, error); } if (response.status === 204) { + request.release(); if (performance.now() - requestStartedAt < FAST_204_MS) { try { await sleep(fast204BackoffMs, options.signal); @@ -201,7 +245,10 @@ export async function holdSignalWait( continue; } // Do not echo server response bodies: they may contain credentials or source. - if (!response.ok) throw error("http_error", `Cloud signal wait failed with HTTP ${response.status}.`, response.status); + if (!response.ok) { + request.release(); + throw error("http_error", `Cloud signal wait failed with HTTP ${response.status}.`, response.status); + } let body: unknown; try { body = await response.json(); @@ -209,6 +256,8 @@ export async function holdSignalWait( options.signal?.throwIfAborted(); if (!(cause instanceof SyntaxError)) throw transportError(deadline.aborted ? deadline.reason : cause, error); throw error("invalid_response", "Cloud returned an invalid signal."); + } finally { + request.release(); } if (!isRecord(body) || !Object.hasOwn(body, "payload") || typeof body.acceptedAt !== "number") { throw error("invalid_response", "Cloud returned an invalid signal."); diff --git a/packages/surface/tests/signals.test.ts b/packages/surface/tests/signals.test.ts index 264a8304..7a875e2a 100644 --- a/packages/surface/tests/signals.test.ts +++ b/packages/surface/tests/signals.test.ts @@ -1,3 +1,4 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { holdSignalWait, SignalError, waitForSignal } from '../src/signals.js'; @@ -207,6 +208,63 @@ describe('@relayflows/surface/signals waitForSignal', () => { await result; }); + // Node 22's undici rejects a (handled) promise of an answered 204's body + // stream when that request's signal aborts later, and the authored + // completion gate fails the run on it. A settled request's signal must + // therefore never abort: not from its deadline, not from the caller. + it('detaches each settled request from its deadline and the caller signal', async () => { + const caller = new AbortController(); + const signals: AbortSignal[] = []; + vi.spyOn(globalThis, 'fetch').mockImplementation(async (_input, init) => { + signals.push(init!.signal!); + return signals.length === 1 ? new Response(null, { status: 204 }) + : Response.json({ payload: 'done', acceptedAt: 1 }); + }); + expect(await waitForSignal('render_done', { + ...options, waitMs: 0, requestTimeoutMs: 5, signal: caller.signal, + })).toBe('done'); + await new Promise(resolve => setTimeout(resolve, 30)); + caller.abort(new Error('caller gave up later')); + expect(signals).toHaveLength(2); + expect(signals.map(signal => signal.aborted)).toEqual([false, false]); + }); + + // The authored completion gate tracks promises by async context. Requests + // run outside the caller's, so transport-internal promises (Node 22 leaves + // one pending per 204) are never counted as the flow author's work. + it('runs each held request outside the caller async context', async () => { + const context = new AsyncLocalStorage(); + const seen: Array = []; + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { + seen.push(context.getStore()); + return seen.length <= 3 ? new Response(null, { status: 204 }) + : Response.json({ payload: 'done', acceptedAt: 1 }); + }); + clock.sleep = async () => undefined; + const result = await context.run('authored-body', async () => { + const payload = await waitForSignal('render_done', options); + return { payload, store: context.getStore() }; + }); + expect(result).toEqual({ payload: 'done', store: 'authored-body' }); + expect(seen).toEqual([undefined, undefined, undefined, undefined]); + }); + + it('still aborts an in-flight request on its deadline or the caller signal', async () => { + const hang = (_input: unknown, init?: RequestInit) => new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => reject(init.signal!.reason), { once: true }); + }); + vi.spyOn(globalThis, 'fetch').mockImplementation(hang); + await expect(waitForSignal('render_done', { ...options, waitMs: 0, requestTimeoutMs: 5 })) + .rejects.toMatchObject({ code: 'transient_error' }); + const caller = new AbortController(); + const reason = new Error('stop'); + const waiting = waitForSignal('render_done', { ...options, signal: caller.signal }); + const result = expect(waiting).rejects.toBe(reason); + await vi.waitFor(() => expect(globalThis.fetch).toHaveBeenCalledTimes(2)); + caller.abort(reason); + await result; + }); + it('classifies a reset connection during fetch as transient', async () => { vi.spyOn(globalThis, 'fetch').mockRejectedValue( new TypeError('fetch failed', { cause: { code: 'ECONNRESET' } }), From a097d757775637b343d6bea695a738946a1461f6 Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 21 Oct 2026 08:22:45 +0200 Subject: [PATCH 2/2] fix(surface): cancel unread signal refusal bodies under the deadline; correct timeout docs Review on #682: - A non-2xx answer released the request before its body was read, so a server that sent error headers and stalled the body held the connection with no deadline (seen on real undici: the socket stayed open past the 2 s deadline; it now closes on refusal). The unread body is cancelled in the request context, while the deadline still reaches the request, and only then released. - The request context pins the AsyncLocalStorage stores active when the module is first evaluated. Documented as a known limit, with a test that pins it; a static import (the documented usage) captures none. - docs/SURFACE.md and the waitForSignal JSDoc recommended bounding the wait with AbortSignal.timeout. After an f.run that fails the run whether the abort is caught or not (and racing a timer leaves the wait in flight), so the docs now say signal stops a run and there is no supported time-out-and-continue yet. Co-Authored-By: Claude Opus 5.5 --- docs/SURFACE.md | 11 +++++++- packages/surface/src/signals.ts | 28 ++++++++++++++++--- packages/surface/tests/signals.test.ts | 37 ++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 4 deletions(-) diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 0dbd303a..9e0b65d4 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -2100,7 +2100,7 @@ has no credential. Defaults, each overridable through `options`: | `apiUrl` | `FLOWS_CLOUD_URL`, else `https://agentrelay.com/cloud` | | `waitMs` | `25000` (per held request; 0–25000) | | `requestTimeoutMs` | `30000` (must exceed `waitMs`) | -| `signal` | none: the wait has no overall deadline; pass `AbortSignal.timeout(ms)` to bound it | +| `signal` | none: the wait has no overall deadline (see below before passing one) | The wire behaviour is the one `@relayflows/sdk`'s `waitForCloudSignal` uses, which now delegates here: names match `^[A-Za-z0-9_-]{1,80}$` and run IDs @@ -2114,6 +2114,15 @@ are never echoed); malformed JSON or a body without `payload` and a numeric `acceptedAt` is `invalid_response`; a missing run ID, token, or bad URL is `configuration`. +An abort through `signal` ends the run, whether or not the body catches it. +After an `f.run`, the authored completion gate fails the run on any rejected +promise derived from that step, so `try { await waitForSignal(name, { signal: +AbortSignal.timeout(ms) }) } catch {}` fails with `operation_callback_failed`, +and uncaught it fails with the abort reason. Racing the wait against a timer +fails too: the losing wait is still in flight when the body returns +(`unsettled_derived_work`). Pass `signal` only to stop a run that has waited +too long; there is no supported way yet to time a wait out and continue. + The wait is not a journaled step. On resume the body calls it again; a read does not consume the signal, so Cloud answers with the same stored payload until the signal expires. A module diff --git a/packages/surface/src/signals.ts b/packages/surface/src/signals.ts index 39511b3e..dfdc12f2 100644 --- a/packages/surface/src/signals.ts +++ b/packages/surface/src/signals.ts @@ -31,6 +31,14 @@ import { setTimeout as delay } from "node:timers/promises"; * Those promises are the transport's, not the author's. The promise the body * awaits — this helper's — is still created and tracked in the authored * context, and every request outcome still reaches it. + * + * Known limit: the resource captures the AsyncLocalStorage stores active when + * this module is first evaluated. A static import (the documented usage) loads + * it before any flow body runs, so they are empty. A first dynamic import from + * inside a body, or any other `als.run(...)`, pins that store for every later + * request in the process, including other runs'. Creating the resource lazily + * does not help: its first use is always inside a body. Node offers no way to + * name an empty context without the storage instances themselves. */ const requestContext = new AsyncResource("RelayflowsSignalRequest"); @@ -83,7 +91,10 @@ export class SignalError extends Error { export interface WaitForSignalOptions { /** Hold per request, 0–25000 ms. Defaults to 25000. */ waitMs?: number; - /** Cancels the wait; the caller's abort reason is rethrown unchanged. */ + /** + * Cancels the wait; the caller's abort reason is rethrown unchanged. In an + * authored body after an `f.run`, an abort fails the run even if caught. + */ signal?: AbortSignal; /** Defaults to `process.env.RELAYFLOW_V2_CLOUD_RUN_ID`. */ runId?: string; @@ -246,7 +257,15 @@ export async function holdSignalWait( } // Do not echo server response bodies: they may contain credentials or source. if (!response.ok) { - request.release(); + // The refusal body is never read. Cancel it while the deadline still + // bounds this request, so a stalled body cannot hold the connection. + try { + await requestContext.runInAsyncScope(() => response.body?.cancel()); + } catch { + // The refusal below is the error this wait reports. + } finally { + request.release(); + } throw error("http_error", `Cloud signal wait failed with HTTP ${response.status}.`, response.status); } let body: unknown; @@ -295,7 +314,10 @@ function hostedConnection(options: WaitForSignalOptions): { baseUrl: string; tok * * Call it in the authored flow body, not inside an `f.run` command: only the * authored process holds the run-bound read token. The wait has no overall - * deadline; pass `signal` (e.g. `AbortSignal.timeout(ms)`) to bound it. + * deadline. Aborting it through `signal` fails the run even when the body + * catches the abort: after an `f.run`, the authored completion gate fails on + * any rejected promise derived from that step. So `signal` can stop a run + * that waited too long, but cannot time a wait out and continue. */ export async function waitForSignal(name: string, options: WaitForSignalOptions = {}): Promise { const runId = options.runId ?? process.env[HOSTED_RUN_ID_ENV]; diff --git a/packages/surface/tests/signals.test.ts b/packages/surface/tests/signals.test.ts index 7a875e2a..06bcd544 100644 --- a/packages/surface/tests/signals.test.ts +++ b/packages/surface/tests/signals.test.ts @@ -249,6 +249,43 @@ describe('@relayflows/surface/signals waitForSignal', () => { expect(seen).toEqual([undefined, undefined, undefined, undefined]); }); + // Pins a documented limit: the request context is captured when the module + // is first evaluated, so a first import inside a store pins that store. + it('pins the AsyncLocalStorage store active when the module is first evaluated', async () => { + const context = new AsyncLocalStorage(); + vi.resetModules(); + const late = await context.run('run-A', () => import('../src/signals.js')); + const seen: Array = []; + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { + seen.push(context.getStore()); + return Response.json({ payload: 'done', acceptedAt: 1 }); + }); + await context.run('run-B', () => late.waitForSignal('render_done', options)); + await context.run('run-B', () => waitForSignal('render_done', options)); + expect(seen).toEqual(['run-A', undefined]); + }); + + // Greptile on #682: releasing first left a stalled refusal body holding the + // connection with no deadline. The body is cancelled outside the caller's + // async context while the deadline still reaches the request. + it('cancels a stalled refusal body before detaching the request deadline', async () => { + const deadline = new AbortController(); + vi.spyOn(AbortSignal, 'timeout').mockReturnValue(deadline.signal); + const context = new AsyncLocalStorage(); + const cancels: Array<{ deadlineReachesRequest: boolean; store: string | undefined }> = []; + vi.spyOn(globalThis, 'fetch').mockImplementation(async (_input, init) => new Response(new ReadableStream({ + pull: () => new Promise(() => undefined), + cancel: () => { + deadline.abort(new DOMException('deadline', 'TimeoutError')); + cancels.push({ deadlineReachesRequest: init!.signal!.aborted, store: context.getStore() }); + }, + }), { status: 503 })); + const refusal = await context.run('authored-body', () => waitForSignal('render_done', options)) + .catch((error: unknown) => error); + expect(refusal).toMatchObject({ code: 'http_error', status: 503 }); + expect(cancels).toEqual([{ deadlineReachesRequest: true, store: undefined }]); + }); + it('still aborts an in-flight request on its deadline or the caller signal', async () => { const hang = (_input: unknown, init?: RequestInit) => new Promise((_resolve, reject) => { init?.signal?.addEventListener('abort', () => reject(init.signal!.reason), { once: true });