fix(gateway): bound stalled session subscriptions (#121164)

2026-08-23 19:08:22 -06:00 · 2026-08-11 10:10:20 -07:00
parent 8fa897a3f1
commit 3a55867ea1
11 changed files with 754 additions and 168 deletions
@@ -1 +1 @@
-{"contentHash":"1456c41077c880274a40b67c96c974ec8b22736b77abe8cec9966d0725fe18d3","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"}
+{"contentHash":"737ad952cf6bb0f9d65d97bd17498af686dae166716274017fd26e5f602981d2","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"}
@@ -138,6 +138,9 @@ from `@openclaw/gateway-protocol`, not from bundled implementation paths.
  the socket; `stop()` closes it and rejects pending requests.
 - A request uses `request(method, params)` after `hello-ok`. Passing
  `timeoutMs: null` creates an intentionally unbounded request.
+- Finite request deadlines reject with `GatewayProtocolRequestTimeoutError`,
+  whose `CLIENT_TIMEOUT` code, method, deadline, and send-boundary flag remain
+  distinct from authoritative Gateway response errors.
 - Device identity persistence, signing, proxy routing, TLS formatting, and
  logging stay host-owned through `GatewayClientHostDeps`.
 - Protocol changes are additive first. Incompatible changes require an explicit
@@ -1,6 +1,9 @@
 import { afterEach, expect, test, vi } from "vitest";
 import { GatewayClient, GatewayClientRequestTimeoutError } from "./client.js";
-import type { GatewayProtocolSocket } from "./protocol-client.js";
+import {
+  GatewayProtocolRequestTimeoutError,
+  type GatewayProtocolSocket,
+} from "./protocol-client.js";

 afterEach(() => {
  vi.useRealTimers();
@@ -28,7 +31,9 @@ test("reports that a timed-out request crossed the transport send boundary", asy

  const error = await outcome;
  expect(error).toBeInstanceOf(GatewayClientRequestTimeoutError);
+  expect(error).toBeInstanceOf(GatewayProtocolRequestTimeoutError);
  expect(error).toMatchObject({
+    code: "CLIENT_TIMEOUT",
    method: "node.invoke",
    timeoutMs: 100,
    requestSent: true,
@@ -42,7 +42,10 @@ import {
  type GatewayProtocolSocket,
  type GatewayProtocolSocketHandlers,
 } from "./protocol-client.js";
-import { GatewayProtocolRequestError } from "./protocol-request.js";
+import {
+  GatewayProtocolRequestError,
+  GatewayProtocolRequestTimeoutError,
+} from "./protocol-request.js";
 import { shouldPauseGatewayReconnect } from "./reconnect-policy.js";
 import { GatewayClientRequestError } from "./request-error.js";
 import {
@@ -242,17 +245,10 @@ export type GatewayClientCloseInfo = {

 export { GatewayClientRequestError } from "./request-error.js";

-export class GatewayClientRequestTimeoutError extends Error {
-  readonly method: string;
-  readonly timeoutMs: number;
-  readonly requestSent: boolean;
-
+export class GatewayClientRequestTimeoutError extends GatewayProtocolRequestTimeoutError {
  constructor(params: { method: string; timeoutMs: number; requestSent: boolean }) {
-    super(`gateway request timeout for ${params.method}`);
+    super(params, `gateway request timeout for ${params.method}`);
    this.name = "GatewayClientRequestTimeoutError";
-    this.method = params.method;
-    this.timeoutMs = params.timeoutMs;
-    this.requestSent = params.requestSent;
  }
 }

@@ -404,7 +400,7 @@ export class GatewayClient {
    };
    this.requestTimeoutMs =
      typeof opts.requestTimeoutMs === "number" && Number.isFinite(opts.requestTimeoutMs)
-        ? resolveSafeTimeoutDelayMs(opts.requestTimeoutMs, { minMs: 0 })
+        ? opts.requestTimeoutMs
        : DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS;
    const connectChallengeTimeoutMs = resolveConnectChallengeTimeoutMs(
      this.opts.connectChallengeTimeoutMs,
@@ -1359,7 +1355,7 @@ export class GatewayClient {
      opts?.timeoutMs === null
        ? null
        : typeof opts?.timeoutMs === "number" && Number.isFinite(opts.timeoutMs)
-          ? resolveSafeTimeoutDelayMs(opts.timeoutMs, { minMs: 0 })
+          ? opts.timeoutMs
          : expectFinal
            ? null
            : this.requestTimeoutMs;
@@ -35,8 +35,8 @@ function createOpenGatewayClient(requestTimeoutMs: number): {
  return { client, send };
 }

-function getPendingCount(client: GatewayClient): number {
-  return protocolHarness(client).pending.size;
+function hasPendingRequests(client: GatewayClient): boolean {
+  return protocolHarness(client).hasPendingRequests;
 }

 test("decodes every ws raw-data shape", () => {
@@ -50,8 +50,8 @@ type ProtocolHarness = {
  socket: GatewayProtocolSocket | null;
  stopped: boolean;
  generation: number;
+  hasPendingRequests: boolean;
  reconnectSupervisor: { reset(initialMs?: number): void };
-  pending: Map<string, unknown>;
  handleMessage: (socket: GatewayProtocolSocket, generation: number, raw: string) => void;
 };

@@ -915,7 +915,7 @@ describe("GatewayClient", () => {
      "synthetic send failure",
    );
    expect(onSent).not.toHaveBeenCalled();
-    expect(getPendingCount(client)).toBe(0);
+    expect(hasPendingRequests(client)).toBe(false);
  });

  test("notifies accepted expectFinal requests while continuing to wait for final", async () => {
@@ -939,7 +939,7 @@ describe("GatewayClient", () => {

    expect(onSent).toHaveBeenCalledOnce();
    expect(onAccepted).toHaveBeenCalledWith({ status: "accepted", runId: "run-1" });
-    expect(getPendingCount(client)).toBe(1);
+    expect(hasPendingRequests(client)).toBe(true);

    handleGatewayMessage(client, {
      type: "res",
@@ -949,7 +949,7 @@ describe("GatewayClient", () => {
    });

    await expect(requestPromise).resolves.toEqual({ status: "ok" });
-    expect(getPendingCount(client)).toBe(0);
+    expect(hasPendingRequests(client)).toBe(false);
  });

  test("aborts in-flight requests from caller AbortSignal", async () => {
@@ -961,12 +961,12 @@ describe("GatewayClient", () => {
      timeoutMs: null,
    });
    expect(send).toHaveBeenCalledTimes(1);
-    expect(getPendingCount(client)).toBe(1);
+    expect(hasPendingRequests(client)).toBe(true);

    controller.abort();

    await expect(requestPromise).rejects.toThrow("gateway request aborted for status");
-    expect(getPendingCount(client)).toBe(0);
+    expect(hasPendingRequests(client)).toBe(false);
  });

  test.each([
@@ -984,7 +984,7 @@ describe("GatewayClient", () => {
      await vi.advanceTimersByTimeAsync(1);

      expect(isSettled()).toBe(false);
-      expect(getPendingCount(client)).toBe(1);
+      expect(hasPendingRequests(client)).toBe(true);

      client.stop();
      await expect(requestPromise).rejects.toThrow("gateway client stopped");
@@ -1,5 +1,26 @@
-/** Owned settlement, cleanup, and timing state for one Gateway wire request. */
-export type GatewayPendingRequest = {
+import type { ErrorShape, ResponseFrame } from "@openclaw/gateway-protocol";
+import {
+  GatewayProtocolRequestError,
+  GatewayProtocolRequestTimeoutError,
+  type GatewayProtocolRequestOptions,
+} from "./protocol-request.js";
+import { resolveSafeTimeoutDelayMs } from "./timeouts.js";
+
+export type GatewayProtocolRequestTiming = {
+  id: string;
+  method: string;
+  ok: boolean;
+  durationMs: number;
+  startedAtMs: number;
+  endedAtMs: number;
+  errorCode?: string;
+};
+
+type GatewayRequestSender = {
+  send: (data: string) => void;
+};
+
+type GatewayPendingRequest = {
  resolve: (value: unknown) => void;
  reject: (error: Error) => void;
  expectFinal: boolean;
@@ -10,3 +31,205 @@ export type GatewayPendingRequest = {
  method: string;
  startedAtMs: number;
 };
+
+type GatewayPendingRequestsOptions = {
+  createRequestId: () => string;
+  createRequestError?: (error: Partial<ErrorShape>) => GatewayProtocolRequestError;
+  createRequestTimeoutError?: (method: string, timeoutMs: number, requestSent: boolean) => Error;
+  createRequestAbortError?: (method: string) => Error;
+  requestTimeoutMs?: number;
+  nowMs: () => number;
+  onTiming?: (timing: GatewayProtocolRequestTiming) => void;
+  onCallbackError?: (label: string, error: unknown) => void;
+};
+
+/** Owns request deadlines, correlation, settlement, and generation-scoped IDs. */
+export class GatewayPendingRequests {
+  private readonly pending = new Map<string, GatewayPendingRequest>();
+  private readonly retiredIds = new Set<string>();
+  private collisionSuffix = 0;
+
+  constructor(private readonly opts: GatewayPendingRequestsOptions) {}
+
+  get hasPending(): boolean {
+    return this.pending.size > 0;
+  }
+
+  get hasUnboundedPending(): boolean {
+    return [...this.pending.values()].some((pending) => pending.unbounded);
+  }
+
+  request<T>(
+    sender: GatewayRequestSender,
+    method: string,
+    params?: unknown,
+    options?: GatewayProtocolRequestOptions,
+  ): Promise<T> {
+    let id: string;
+    try {
+      id = this.allocateRequestId();
+    } catch (error) {
+      return Promise.reject(error instanceof Error ? error : new Error(String(error)));
+    }
+    const requestedTimeoutMs =
+      options?.timeoutMs === null ? undefined : (options?.timeoutMs ?? this.opts.requestTimeoutMs);
+    const timeoutMs =
+      typeof requestedTimeoutMs === "number" && Number.isFinite(requestedTimeoutMs)
+        ? resolveSafeTimeoutDelayMs(requestedTimeoutMs, { minMs: 0 })
+        : undefined;
+    return new Promise<T>((resolve, reject) => {
+      let timeout: ReturnType<typeof setTimeout> | undefined;
+      let requestSent = false;
+      const pending: GatewayPendingRequest = {
+        resolve: (value) => resolve(value as T),
+        reject,
+        expectFinal: options?.expectFinal === true,
+        acceptedNotified: false,
+        onAccepted: options?.onAccepted,
+        unbounded: timeoutMs === undefined,
+        method,
+        startedAtMs: this.opts.nowMs(),
+      };
+      const cleanup = () => {
+        if (timeout !== undefined) {
+          clearTimeout(timeout);
+        }
+        options?.signal?.removeEventListener("abort", onAbort);
+      };
+      const retire = (errorCode: string): boolean => {
+        if (this.pending.get(id) !== pending) {
+          return false;
+        }
+        this.pending.delete(id);
+        this.retiredIds.add(id);
+        cleanup();
+        this.finishTiming(id, pending, false, errorCode);
+        return true;
+      };
+      const onAbort = () => {
+        if (!retire("CLIENT_ABORTED")) {
+          return;
+        }
+        reject(
+          this.opts.createRequestAbortError?.(method) ??
+            new Error(`gateway request aborted for ${method}`),
+        );
+      };
+      if (options?.signal?.aborted) {
+        reject(
+          this.opts.createRequestAbortError?.(method) ??
+            new Error(`gateway request aborted for ${method}`),
+        );
+        return;
+      }
+      pending.cleanup = cleanup;
+      if (timeoutMs !== undefined) {
+        timeout = setTimeout(() => {
+          if (!retire("CLIENT_TIMEOUT")) {
+            return;
+          }
+          reject(
+            this.opts.createRequestTimeoutError?.(method, timeoutMs, requestSent) ??
+              new GatewayProtocolRequestTimeoutError({ method, timeoutMs, requestSent }),
+          );
+        }, timeoutMs);
+        timeout.unref?.();
+      }
+      options?.signal?.addEventListener("abort", onAbort, { once: true });
+      this.pending.set(id, pending);
+      try {
+        sender.send(JSON.stringify({ type: "req", id, method, params }));
+        if (this.pending.get(id) !== pending) {
+          return;
+        }
+        requestSent = true;
+        this.invoke("sent", () => options?.onSent?.());
+      } catch (error) {
+        if (retire("CLIENT_SEND_ERROR")) {
+          reject(error instanceof Error ? error : new Error(String(error)));
+        }
+      }
+    });
+  }
+
+  handleResponse(frame: ResponseFrame): void {
+    const pending = this.pending.get(frame.id);
+    if (!pending) {
+      return;
+    }
+    const status = (frame.payload as { status?: unknown } | undefined)?.status;
+    if (pending.expectFinal && status === "accepted") {
+      if (!pending.acceptedNotified) {
+        pending.acceptedNotified = true;
+        this.invoke("accepted", () => pending.onAccepted?.(frame.payload));
+      }
+      return;
+    }
+    this.pending.delete(frame.id);
+    pending.cleanup?.();
+    if (frame.ok) {
+      this.finishTiming(frame.id, pending, true);
+      pending.resolve(frame.payload);
+      return;
+    }
+    this.finishTiming(frame.id, pending, false, frame.error?.code);
+    pending.reject(
+      this.opts.createRequestError?.(frame.error ?? {}) ??
+        new GatewayProtocolRequestError(frame.error ?? {}),
+    );
+  }
+
+  flush(error: Error): void {
+    for (const [id, pending] of this.pending) {
+      this.finishTiming(id, pending, false, "CLIENT_CLOSED");
+      pending.cleanup?.();
+      pending.reject(error);
+    }
+    this.pending.clear();
+    // IDs are tombstoned only for one socket generation. Retired socket frames
+    // are fenced by GatewayProtocolClient before a replacement generation runs.
+    this.retiredIds.clear();
+    this.collisionSuffix = 0;
+  }
+
+  private allocateRequestId(): string {
+    const id = this.opts.createRequestId();
+    if (!this.pending.has(id) && !this.retiredIds.has(id)) {
+      return id;
+    }
+    let uniqueId: string;
+    do {
+      this.collisionSuffix += 1;
+      uniqueId = `${id}:${this.collisionSuffix}`;
+    } while (this.pending.has(uniqueId) || this.retiredIds.has(uniqueId));
+    return uniqueId;
+  }
+
+  private finishTiming(
+    id: string,
+    pending: GatewayPendingRequest,
+    ok: boolean,
+    errorCode?: string,
+  ): void {
+    const endedAtMs = this.opts.nowMs();
+    this.invoke("request timing", () =>
+      this.opts.onTiming?.({
+        id,
+        method: pending.method,
+        ok,
+        durationMs: Math.max(0, endedAtMs - pending.startedAtMs),
+        startedAtMs: pending.startedAtMs,
+        endedAtMs,
+        errorCode,
+      }),
+    );
+  }
+
+  private invoke(label: string, callback: () => void): void {
+    try {
+      callback();
+    } catch (error) {
+      this.opts.onCallbackError?.(label, error);
+    }
+  }
+}
@@ -0,0 +1,350 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import {
+  GatewayProtocolClient,
+  GatewayProtocolRequestError,
+  GatewayProtocolRequestTimeoutError,
+  type GatewayProtocolRequestOptions,
+  type GatewayProtocolRequestTiming,
+  type GatewayProtocolSocketHandlers,
+} from "./protocol-client.js";
+import { MAX_SAFE_TIMEOUT_DELAY_MS } from "./timeouts.js";
+
+type RequestFrame = {
+  id: string;
+  method: string;
+};
+
+type RequestConnection = {
+  handlers: GatewayProtocolSocketHandlers;
+  frames: RequestFrame[];
+  close: (code?: number, reason?: string) => void;
+};
+
+function createRequestHarness(options?: {
+  createRequestId?: () => string;
+  requestTimeoutMs?: number;
+  onRequestTiming?: (timing: GatewayProtocolRequestTiming) => void;
+  onCallbackError?: (label: string, error: unknown) => void;
+  send?: (frame: RequestFrame) => void;
+  nowMs?: () => number;
+}) {
+  const connections: RequestConnection[] = [];
+  let nextRequestId = 0;
+  const client = new GatewayProtocolClient<Record<string, never>>({
+    createSocket: (handlers) => {
+      let open = true;
+      const frames: RequestFrame[] = [];
+      const close = (code = 1000, reason = "") => {
+        open = false;
+        handlers.close(code, reason);
+      };
+      connections.push({ handlers, frames, close });
+      return {
+        isOpen: () => open,
+        send: (data) => {
+          const frame = JSON.parse(data) as RequestFrame;
+          frames.push(frame);
+          options?.send?.(frame);
+        },
+        close,
+      };
+    },
+    createRequestId: options?.createRequestId ?? (() => `request-${++nextRequestId}`),
+    buildConnectPlan: () => ({}),
+    buildConnectParams: (plan) => plan,
+    resolveClose: () => ({ retry: false, notify: false }),
+    handshake: { mode: "require-challenge", timeoutMs: 100 },
+    reconnect: { initialMs: 10, multiplier: 2, maxMs: 100 },
+    requestTimeoutMs: options?.requestTimeoutMs,
+    onRequestTiming: options?.onRequestTiming,
+    onCallbackError: options?.onCallbackError,
+    nowMs: options?.nowMs,
+  });
+  client.start();
+  return { client, connections };
+}
+
+function latestFrame(connection: RequestConnection): RequestFrame {
+  const frame = connection.frames.at(-1);
+  if (!frame) {
+    throw new Error("expected request frame");
+  }
+  return frame;
+}
+
+function respond(connection: RequestConnection, id: string, payload: unknown, ok = true): void {
+  connection.handlers.message(
+    JSON.stringify({
+      type: "res",
+      id,
+      ok,
+      ...(ok ? { payload } : { error: payload }),
+    }),
+  );
+}
+
+afterEach(() => {
+  vi.useRealTimers();
+  vi.restoreAllMocks();
+});
+
+describe("GatewayProtocolClient requests", () => {
+  it.each([
+    {
+      label: "an explicit finite deadline",
+      requestTimeoutMs: undefined,
+      requestOptions: { timeoutMs: 25 } satisfies GatewayProtocolRequestOptions,
+      expectedTimerMs: 25,
+      unbounded: false,
+    },
+    {
+      label: "the client default deadline",
+      requestTimeoutMs: 30,
+      requestOptions: undefined,
+      expectedTimerMs: 30,
+      unbounded: false,
+    },
+    {
+      label: "an oversized finite deadline",
+      requestTimeoutMs: undefined,
+      requestOptions: {
+        timeoutMs: Number.MAX_SAFE_INTEGER,
+      } satisfies GatewayProtocolRequestOptions,
+      expectedTimerMs: MAX_SAFE_TIMEOUT_DELAY_MS,
+      unbounded: false,
+    },
+    {
+      label: "an explicit null deadline",
+      requestTimeoutMs: 30,
+      requestOptions: { timeoutMs: null } satisfies GatewayProtocolRequestOptions,
+      expectedTimerMs: null,
+      unbounded: true,
+    },
+    {
+      label: "the browser default",
+      requestTimeoutMs: undefined,
+      requestOptions: undefined,
+      expectedTimerMs: null,
+      unbounded: true,
+    },
+  ])(
+    "normalizes $label only in the scheduling owner",
+    ({ requestTimeoutMs, requestOptions, expectedTimerMs, unbounded }) => {
+      vi.useFakeTimers();
+      const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
+      const { client } = createRequestHarness({ requestTimeoutMs });
+      const request = client.request("status", {}, requestOptions);
+      void request.catch(() => {});
+
+      expect(client.hasUnboundedPendingRequests).toBe(unbounded);
+      if (expectedTimerMs === null) {
+        expect(setTimeoutSpy).not.toHaveBeenCalled();
+      } else {
+        expect(setTimeoutSpy).toHaveBeenLastCalledWith(expect.any(Function), expectedTimerMs);
+      }
+      client.stop();
+    },
+  );
+
+  it("reports typed deadlines before and after the send boundary", async () => {
+    vi.useFakeTimers();
+    const sentHarness = createRequestHarness();
+    const sentRequest = sentHarness.client.request("sent.request", {}, { timeoutMs: 5 });
+    const sentOutcome = sentRequest.catch((error: unknown) => error);
+
+    await vi.advanceTimersByTimeAsync(5);
+
+    await expect(sentOutcome).resolves.toMatchObject({
+      code: "CLIENT_TIMEOUT",
+      method: "sent.request",
+      timeoutMs: 5,
+      requestSent: true,
+    });
+
+    let deadline: (() => void) | undefined;
+    vi.spyOn(globalThis, "setTimeout").mockImplementation(((
+      callback: Parameters<typeof setTimeout>[0],
+    ) => {
+      deadline = callback as () => void;
+      return { unref: () => undefined } as unknown as ReturnType<typeof setTimeout>;
+    }) as unknown as typeof setTimeout);
+    const onSent = vi.fn();
+    const unsentHarness = createRequestHarness({ send: () => deadline?.() });
+    const unsentRequest = unsentHarness.client.request(
+      "unsent.request",
+      {},
+      { timeoutMs: 5, onSent },
+    );
+
+    await expect(unsentRequest).rejects.toMatchObject({
+      code: "CLIENT_TIMEOUT",
+      method: "unsent.request",
+      timeoutMs: 5,
+      requestSent: false,
+    });
+    expect(onSent).not.toHaveBeenCalled();
+    expect(unsentHarness.client.hasPendingRequests).toBe(false);
+    expect(sentHarness.client.hasPendingRequests).toBe(false);
+    sentHarness.client.stop();
+    unsentHarness.client.stop();
+  });
+
+  it("retires aborted and send-failed IDs before a replacement request", async () => {
+    const controller = new AbortController();
+    let sendCalls = 0;
+    const { client, connections } = createRequestHarness({
+      createRequestId: () => "same-id",
+      send: () => {
+        sendCalls += 1;
+        if (sendCalls === 3) {
+          throw new Error("synthetic send failure");
+        }
+      },
+    });
+    const connection = connections[0];
+    if (!connection) {
+      throw new Error("expected request connection");
+    }
+    const aborted = client.request("aborted", {}, { timeoutMs: null, signal: controller.signal });
+    controller.abort();
+    await expect(aborted).rejects.toThrow("gateway request aborted for aborted");
+
+    const replacement = client.request("replacement", {}, { timeoutMs: null });
+    expect(latestFrame(connection)).toMatchObject({ id: "same-id:1", method: "replacement" });
+    respond(connection, "same-id", { stale: true });
+    expect(client.hasPendingRequests).toBe(true);
+    respond(connection, "same-id:1", { current: true });
+    await expect(replacement).resolves.toEqual({ current: true });
+
+    await expect(client.request("send.failure", {}, { timeoutMs: null })).rejects.toThrow(
+      "synthetic send failure",
+    );
+    expect(latestFrame(connection)).toMatchObject({ id: "same-id:2", method: "send.failure" });
+    expect(client.hasPendingRequests).toBe(false);
+    client.stop();
+  });
+
+  it("ignores late accepted and final replies after a timeout collision", async () => {
+    vi.useFakeTimers();
+    const onAccepted = vi.fn();
+    const { client, connections } = createRequestHarness({ createRequestId: () => "same-id" });
+    const connection = connections[0];
+    if (!connection) {
+      throw new Error("expected request connection");
+    }
+    const retired = client.request("agent", {}, { timeoutMs: 5, expectFinal: true, onAccepted });
+    const retiredOutcome = retired.catch((error: unknown) => error);
+    await vi.advanceTimersByTimeAsync(5);
+    await expect(retiredOutcome).resolves.toBeInstanceOf(GatewayProtocolRequestTimeoutError);
+
+    const replacement = client.request(
+      "agent",
+      {},
+      { timeoutMs: null, expectFinal: true, onAccepted },
+    );
+    expect(latestFrame(connection)).toMatchObject({ id: "same-id:1", method: "agent" });
+    respond(connection, "same-id", { status: "accepted", runId: "old" });
+    respond(connection, "same-id", { status: "ok", runId: "old" });
+    expect(onAccepted).not.toHaveBeenCalled();
+    expect(client.hasPendingRequests).toBe(true);
+
+    respond(connection, "same-id:1", { status: "accepted", runId: "new" });
+    respond(connection, "same-id:1", { status: "ok", runId: "new" });
+    await expect(replacement).resolves.toEqual({ status: "ok", runId: "new" });
+    expect(onAccepted).toHaveBeenCalledExactlyOnceWith({ status: "accepted", runId: "new" });
+    client.stop();
+  });
+
+  it("keeps authoritative Gateway errors distinct from local deadlines", async () => {
+    const { client, connections } = createRequestHarness();
+    const connection = connections[0];
+    if (!connection) {
+      throw new Error("expected request connection");
+    }
+    const request = client.request("sessions.subscribe", {}, { timeoutMs: 25 });
+    const frame = latestFrame(connection);
+    respond(
+      connection,
+      frame.id,
+      { code: "FORBIDDEN", message: "subscription rejected", retryable: false },
+      false,
+    );
+
+    const error = await request.catch((value: unknown) => value);
+    expect(error).toBeInstanceOf(GatewayProtocolRequestError);
+    expect(error).not.toBeInstanceOf(GatewayProtocolRequestTimeoutError);
+    expect(error).toMatchObject({ code: "FORBIDDEN", retryable: false });
+    client.stop();
+  });
+
+  it("isolates callbacks while preserving accepted/final settlement and timing", async () => {
+    let nowMs = 10;
+    const onRequestTiming = vi.fn<(timing: GatewayProtocolRequestTiming) => void>();
+    const onCallbackError = vi.fn<(label: string, error: unknown) => void>();
+    const { client, connections } = createRequestHarness({
+      nowMs: () => nowMs,
+      onRequestTiming,
+      onCallbackError,
+    });
+    const connection = connections[0];
+    if (!connection) {
+      throw new Error("expected request connection");
+    }
+    const request = client.request(
+      "agent",
+      {},
+      {
+        timeoutMs: null,
+        expectFinal: true,
+        onSent: () => {
+          throw new Error("sent callback failed");
+        },
+        onAccepted: () => {
+          throw new Error("accepted callback failed");
+        },
+      },
+    );
+    const frame = latestFrame(connection);
+    respond(connection, frame.id, { status: "accepted" });
+    expect(client.hasPendingRequests).toBe(true);
+    nowMs = 25;
+    respond(connection, frame.id, { status: "ok" });
+
+    await expect(request).resolves.toEqual({ status: "ok" });
+    expect(onCallbackError.mock.calls.map(([label]) => label)).toEqual(["sent", "accepted"]);
+    expect(onRequestTiming).toHaveBeenCalledExactlyOnceWith({
+      id: frame.id,
+      method: "agent",
+      ok: true,
+      durationMs: 15,
+      startedAtMs: 10,
+      endedAtMs: 25,
+    });
+    client.stop();
+  });
+
+  it("clears generation tombstones when the socket flushes", async () => {
+    vi.useFakeTimers();
+    const { client, connections } = createRequestHarness({ createRequestId: () => "same-id" });
+    const firstConnection = connections[0];
+    if (!firstConnection) {
+      throw new Error("expected first request connection");
+    }
+    const retired = client.request("first", {}, { timeoutMs: 5 });
+    const retiredOutcome = retired.catch((error: unknown) => error);
+    await vi.advanceTimersByTimeAsync(5);
+    await expect(retiredOutcome).resolves.toBeInstanceOf(GatewayProtocolRequestTimeoutError);
+
+    firstConnection.close(1000, "socket generation complete");
+    client.start();
+    const secondConnection = connections[1];
+    if (!secondConnection) {
+      throw new Error("expected replacement request connection");
+    }
+    const replacement = client.request("second", {}, { timeoutMs: null });
+    expect(latestFrame(secondConnection)).toMatchObject({ id: "same-id", method: "second" });
+    respond(secondConnection, "same-id", { ok: true });
+    await expect(replacement).resolves.toEqual({ ok: true });
+    client.stop();
+  });
+});
@@ -1,18 +1,24 @@
-import type { ErrorShape, EventFrame, HelloOk, ResponseFrame } from "@openclaw/gateway-protocol";
+import type { ErrorShape, EventFrame, HelloOk } from "@openclaw/gateway-protocol";
 import {
  isGatewayEventFrame,
  isGatewayResponseFrame,
 } from "@openclaw/gateway-protocol/frame-guards";
 import { RetrySupervisor, sleepWithAbort } from "@openclaw/retry";
 import { GatewayEventListeners } from "./event-listeners.js";
-import type { GatewayPendingRequest } from "./pending-request.js";
+import { GatewayPendingRequests, type GatewayProtocolRequestTiming } from "./pending-request.js";
 import {
  GatewayProtocolRequestError,
+  GatewayProtocolRequestTimeoutError,
  type GatewayProtocolRequestOptions,
 } from "./protocol-request.js";
 import { clearGatewayConnectTimeout, startGatewayConnectTimeout } from "./timeouts.js";

-export { GatewayProtocolRequestError, type GatewayProtocolRequestOptions };
+export {
+  GatewayProtocolRequestError,
+  GatewayProtocolRequestTimeoutError,
+  type GatewayProtocolRequestOptions,
+  type GatewayProtocolRequestTiming,
+};

 export type GatewayProtocolSocket = {
  isOpen: () => boolean;
@@ -71,15 +77,6 @@ export type GatewayProtocolTiming<TPlan> = {
  plan?: TPlan;
  detail?: unknown;
 };
-export type GatewayProtocolRequestTiming = {
-  id: string;
-  method: string;
-  ok: boolean;
-  durationMs: number;
-  startedAtMs: number;
-  endedAtMs: number;
-  errorCode?: string;
-};
 type GatewayProtocolClientOptions<TPlan> = {
  createSocket: (handlers: GatewayProtocolSocketHandlers) => GatewayProtocolSocket;
  createRequestId: () => string;
@@ -139,7 +136,7 @@ type CloseSnapshot = Omit<GatewayProtocolCloseContext, "code" | "reason">;
 */
 export class GatewayProtocolClient<TPlan> {
  private socket: GatewayProtocolSocket | null = null;
-  private readonly pending = new Map<string, GatewayPendingRequest>();
+  private readonly requests: GatewayPendingRequests;
  private readonly listeners = new GatewayEventListeners<EventFrame>();
  private stopped = true;
  private generation = 0;
@@ -164,6 +161,16 @@ export class GatewayProtocolClient<TPlan> {
      factor: opts.reconnect.multiplier,
      jitter: 0,
    });
+    this.requests = new GatewayPendingRequests({
+      createRequestId: opts.createRequestId,
+      createRequestError: opts.createRequestError,
+      createRequestTimeoutError: opts.createRequestTimeoutError,
+      createRequestAbortError: opts.createRequestAbortError,
+      requestTimeoutMs: opts.requestTimeoutMs,
+      nowMs: () => this.nowMs(),
+      onTiming: opts.onRequestTiming,
+      onCallbackError: opts.onCallbackError,
+    });
  }

  get connected(): boolean {
@@ -171,7 +178,7 @@ export class GatewayProtocolClient<TPlan> {
  }

  get hasPendingRequests(): boolean {
-    return this.pending.size > 0;
+    return this.requests.hasPending;
  }

  get connecting(): boolean {
@@ -179,7 +186,7 @@ export class GatewayProtocolClient<TPlan> {
  }

  get hasUnboundedPendingRequests(): boolean {
-    return [...this.pending.values()].some((pending) => pending.unbounded);
+    return this.requests.hasUnboundedPending;
  }

  start(): void {
@@ -205,7 +212,7 @@ export class GatewayProtocolClient<TPlan> {
    this.socket = null;
    this.connectFailure = undefined;
    this.connectTiming = null;
-    this.flushRequests(new Error("gateway client stopped"));
+    this.requests.flush(new Error("gateway client stopped"));
    socket?.close();
  }

@@ -221,73 +228,7 @@ export class GatewayProtocolClient<TPlan> {
    if (typeof method !== "string" || method.length === 0) {
      return Promise.reject(new Error("invalid request frame: method must be a non-empty string"));
    }
-    const id = this.opts.createRequestId();
-    const timeoutMs =
-      options?.timeoutMs === null ? undefined : (options?.timeoutMs ?? this.opts.requestTimeoutMs);
-    return new Promise<T>((resolve, reject) => {
-      let timeout: ReturnType<typeof setTimeout> | undefined;
-      let requestSent = false;
-      const pending: GatewayPendingRequest = {
-        resolve: (value) => resolve(value as T),
-        reject,
-        expectFinal: options?.expectFinal === true,
-        acceptedNotified: false,
-        onAccepted: options?.onAccepted,
-        unbounded: timeoutMs === undefined,
-        method,
-        startedAtMs: this.nowMs(),
-      };
-      const onAbort = () => {
-        this.pending.delete(id);
-        pending.cleanup?.();
-        this.finishRequestTiming(id, pending, false, "CLIENT_ABORTED");
-        reject(
-          this.opts.createRequestAbortError?.(method) ??
-            new Error(`gateway request aborted for ${method}`),
-        );
-      };
-      const cleanup = () => {
-        if (timeout) {
-          clearTimeout(timeout);
-        }
-        options?.signal?.removeEventListener("abort", onAbort);
-      };
-      if (options?.signal?.aborted) {
-        reject(
-          this.opts.createRequestAbortError?.(method) ??
-            new Error(`gateway request aborted for ${method}`),
-        );
-        return;
-      }
-      pending.cleanup = cleanup;
-      if (timeoutMs !== undefined && timeoutMs >= 0) {
-        timeout = setTimeout(() => {
-          if (this.pending.get(id) !== pending) {
-            return;
-          }
-          this.pending.delete(id);
-          options?.signal?.removeEventListener("abort", onAbort);
-          this.finishRequestTiming(id, pending, false, "CLIENT_TIMEOUT");
-          reject(
-            this.opts.createRequestTimeoutError?.(method, timeoutMs, requestSent) ??
-              new Error(`gateway request timed out after ${timeoutMs}ms: ${method}`),
-          );
-        }, timeoutMs);
-        timeout.unref?.();
-      }
-      options?.signal?.addEventListener("abort", onAbort, { once: true });
-      this.pending.set(id, pending);
-      try {
-        socket.send(JSON.stringify({ type: "req", id, method, params }));
-        requestSent = true;
-        this.invoke("sent", () => options?.onSent?.());
-      } catch (error) {
-        this.pending.delete(id);
-        cleanup();
-        this.finishRequestTiming(id, pending, false, "CLIENT_SEND_ERROR");
-        reject(error instanceof Error ? error : new Error(String(error)));
-      }
-    });
+    return this.requests.request<T>(socket, method, params, options);
  }

  addEventListener(listener: (event: EventFrame) => void): () => void {
@@ -588,34 +529,7 @@ export class GatewayProtocolClient<TPlan> {
      return;
    }
    this.opts.onActivity?.();
-    this.handleResponse(parsed);
-  }
-
-  private handleResponse(frame: ResponseFrame): void {
-    const pending = this.pending.get(frame.id);
-    if (!pending) {
-      return;
-    }
-    const status = (frame.payload as { status?: unknown } | undefined)?.status;
-    if (pending.expectFinal && status === "accepted") {
-      if (!pending.acceptedNotified) {
-        pending.acceptedNotified = true;
-        this.invoke("accepted", () => pending.onAccepted?.(frame.payload));
-      }
-      return;
-    }
-    this.pending.delete(frame.id);
-    pending.cleanup?.();
-    if (frame.ok) {
-      this.finishRequestTiming(frame.id, pending, true);
-      pending.resolve(frame.payload);
-      return;
-    }
-    this.finishRequestTiming(frame.id, pending, false, frame.error?.code);
-    pending.reject(
-      this.opts.createRequestError?.(frame.error ?? {}) ??
-        new GatewayProtocolRequestError(frame.error ?? {}),
-    );
+    this.requests.handleResponse(parsed);
  }

  private handleClose(
@@ -642,7 +556,7 @@ export class GatewayProtocolClient<TPlan> {
    };
    this.connectFailure = undefined;
    const decision = this.opts.resolveClose(context);
-    this.flushRequests(
+    this.requests.flush(
      decision.pendingError ??
        context.connectFailure?.error ??
        new Error(`gateway closed (${code}): ${reason}`),
@@ -660,35 +574,6 @@ export class GatewayProtocolClient<TPlan> {
    this.opts.onConnectError?.(error);
  }

-  private flushRequests(error: Error): void {
-    for (const [id, pending] of this.pending) {
-      this.finishRequestTiming(id, pending, false, "CLIENT_CLOSED");
-      pending.cleanup?.();
-      pending.reject(error);
-    }
-    this.pending.clear();
-  }
-
-  private finishRequestTiming(
-    id: string,
-    pending: GatewayPendingRequest,
-    ok: boolean,
-    errorCode?: string,
-  ): void {
-    const endedAtMs = this.nowMs();
-    this.invoke("request timing", () =>
-      this.opts.onRequestTiming?.({
-        id,
-        method: pending.method,
-        ok,
-        durationMs: Math.max(0, endedAtMs - pending.startedAtMs),
-        startedAtMs: pending.startedAtMs,
-        endedAtMs,
-        errorCode,
-      }),
-    );
-  }
-
  private scheduleReconnect(overrideMs?: number): void {
    if (overrideMs !== undefined) {
      // Retry-After is a floor for this wait, not a failed attempt. Preserve
@@ -25,3 +25,22 @@ export class GatewayProtocolRequestError extends Error {
    this.retryAfterMs = error.retryAfterMs;
  }
 }
+
+/** A local transport deadline, distinct from a Gateway's authoritative rejection. */
+export class GatewayProtocolRequestTimeoutError extends Error {
+  readonly code = "CLIENT_TIMEOUT";
+  readonly method: string;
+  readonly timeoutMs: number;
+  readonly requestSent: boolean;
+
+  constructor(
+    params: { method: string; timeoutMs: number; requestSent: boolean },
+    message = `gateway request timed out after ${params.timeoutMs}ms: ${params.method}`,
+  ) {
+    super(message);
+    this.name = "GatewayProtocolRequestTimeoutError";
+    this.method = params.method;
+    this.timeoutMs = params.timeoutMs;
+    this.requestSent = params.requestSent;
+  }
+}
@@ -1,4 +1,9 @@
 // @vitest-environment node
+import {
+  DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS,
+  GatewayProtocolClient,
+  type GatewayProtocolSocketHandlers,
+} from "@openclaw/gateway-client/browser";
 import { describe, expect, it, vi } from "vitest";
 import {
  GatewayRequestError,
@@ -336,6 +341,91 @@ describe("session connection hydration", () => {
    }
  });

+  it("recovers the roster after a subscription deadline without admitting late replies", async () => {
+    vi.useFakeTimers();
+    const initialResult = emptySessionsResult();
+    const recoveredResult: SessionsListResult = {
+      ...initialResult,
+      count: 1,
+      sessions: [{ key: "agent:main:recovered", kind: "direct", updatedAt: 2 }],
+    };
+    const sent: Array<{ id: string; method: string }> = [];
+    let handlers: GatewayProtocolSocketHandlers | undefined;
+    const protocol = new GatewayProtocolClient<Record<string, never>>({
+      createSocket: (nextHandlers) => {
+        handlers = nextHandlers;
+        return {
+          isOpen: () => true,
+          send: (data) => {
+            const frame = JSON.parse(data) as { id: string; method: string };
+            sent.push({ id: frame.id, method: frame.method });
+          },
+          close: () => undefined,
+        };
+      },
+      createRequestId: () => "request",
+      buildConnectPlan: () => ({}),
+      buildConnectParams: (plan) => plan,
+      resolveClose: () => ({ retry: false, notify: false }),
+      handshake: { mode: "require-challenge", timeoutMs: 100 },
+      reconnect: { initialMs: 10, multiplier: 2, maxMs: 100 },
+    });
+    protocol.start();
+    const request = protocol.request.bind(protocol) as GatewayBrowserClient["request"];
+    const { sessions, connect } = createSubscriptionHydrationHarness(request);
+    const recoveredStates: SessionsListResult[] = [];
+    const unsubscribe = sessions.subscribe((next) => {
+      if (next.result?.sessions.some((row) => row.key === "agent:main:recovered")) {
+        recoveredStates.push(next.result);
+      }
+    });
+    const respond = (id: string, payload: unknown) => {
+      handlers?.message(JSON.stringify({ type: "res", id, ok: true, payload }));
+    };
+
+    try {
+      connect();
+      expect(sent).toEqual([{ id: "request", method: "sessions.subscribe" }]);
+
+      await vi.advanceTimersByTimeAsync(DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS);
+      expect(sent).toContainEqual({ id: "request:1", method: "sessions.list" });
+      respond("request:1", initialResult);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(sessions.state.result).toEqual(initialResult);
+
+      await vi.advanceTimersByTimeAsync(500);
+      expect(sent).toContainEqual({ id: "request:2", method: "sessions.subscribe" });
+
+      respond("request", { status: "accepted" });
+      respond("request", { subscribed: true });
+      await vi.advanceTimersByTimeAsync(0);
+      expect(sessions.state.error).not.toBeNull();
+      expect(sent.filter(({ method }) => method === "sessions.list")).toHaveLength(1);
+
+      respond("request:2", { subscribed: true });
+      await vi.advanceTimersByTimeAsync(0);
+      expect(sent).toContainEqual({ id: "request:3", method: "sessions.list" });
+      respond("request:3", recoveredResult);
+      await vi.advanceTimersByTimeAsync(0);
+
+      respond("request", { status: "accepted" });
+      respond("request", { subscribed: true });
+      await vi.advanceTimersByTimeAsync(0);
+
+      expect(sessions.state.error).toBeNull();
+      expect(sessions.state.result).toEqual(recoveredResult);
+      expect(recoveredStates).toEqual([recoveredResult]);
+      expect(sent.filter(({ method }) => method === "sessions.subscribe")).toHaveLength(2);
+      expect(sent.filter(({ method }) => method === "sessions.list")).toHaveLength(2);
+      expect(vi.getTimerCount()).toBe(0);
+    } finally {
+      unsubscribe();
+      sessions.dispose();
+      protocol.stop();
+      vi.useRealTimers();
+    }
+  });
+
  it.each([
    { description: "an explicitly declined", response: { subscribed: false } },
    { description: "an unacknowledged", response: {} },
@@ -1,3 +1,7 @@
+import {
+  DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS,
+  GatewayProtocolRequestTimeoutError,
+} from "@openclaw/gateway-client/browser";
 import { GatewayRequestError, type GatewayBrowserClient } from "../../api/gateway.ts";

 type SessionEventSubscriptionScope = {
@@ -46,6 +50,7 @@ export function createSessionEventSubscriptionOwner(params: {
        const response = await scope.client.request<{ subscribed?: boolean }>(
          "sessions.subscribe",
          {},
+          { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS },
        );
        if (!isCurrent(scope, expectedGeneration)) {
          return;
@@ -63,8 +68,18 @@ export function createSessionEventSubscriptionOwner(params: {
        if (!isCurrent(scope, expectedGeneration)) {
          return;
        }
-        params.onError(scope, String(error));
-        const delayMs = params.retryDelayMs(error);
+        // A connected transport can outlive an application acknowledgement.
+        // Only this idempotent observer turns its typed deadline into a retry.
+        const failure =
+          error instanceof GatewayProtocolRequestTimeoutError
+            ? new GatewayRequestError({
+                code: error.code,
+                message: error.message,
+                retryable: true,
+              })
+            : error;
+        params.onError(scope, String(failure));
+        const delayMs = params.retryDelayMs(failure);
        if (delayMs === null || !isCurrent(scope, expectedGeneration)) {
          return;
        }