fix(surface): keep waitForSignal's transport out of the authored gate on Node 22

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

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<never>((_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');

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,14 +256,27 @@ 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();
     } catch (cause) {
       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<unknown> {
   const runId = options.runId ?? process.env[HOSTED_RUN_ID_ENV];

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<string>();
+    const seen: Array<string | undefined> = [];
+    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<string>();
+    vi.resetModules();
+    const late = await context.run('run-A', () => import('../src/signals.js'));
+    const seen: Array<string | undefined> = [];
+    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<string>();
+    const cancels: Array<{ deadlineReachesRequest: boolean; store: string | undefined }> = [];
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async (_input, init) => new Response(new ReadableStream({
+      pull: () => new Promise<void>(() => 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<Response>((_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' } }),