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/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..dfdc12f2 100644 --- a/packages/surface/src/signals.ts +++ b/packages/surface/src/signals.ts @@ -11,8 +11,37 @@ * 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. + * + * 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"); + /** 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`). */ @@ -62,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; @@ -130,6 +162,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 +224,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 +256,18 @@ 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) { + // 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; try { body = await response.json(); @@ -209,6 +275,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."); @@ -246,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 264a8304..06bcd544 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,100 @@ 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]); + }); + + // 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 }); + }); + 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' } }),