mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor(gateway): consolidate TypeScript WebSocket clients (#106113)
* refactor(gateway): unify websocket clients * fix(gateway): preserve adapter lifecycle behavior * test(gateway): keep reconnect proof black-box * chore(gateway): satisfy shared client lint * refactor(gateway): keep protocol decisions internal * refactor(gateway): preserve shared reconnect supervision * test(gateway): use canonical browser barrel * fix(gateway): resolve browser retry workspace source * refactor(gateway): tighten shared protocol types
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
581cf6601a
commit
7c9b8baa2b
@@ -14,6 +14,11 @@
|
||||
"import": "./dist/index.mjs",
|
||||
"default": "./dist/index.mjs"
|
||||
},
|
||||
"./browser": {
|
||||
"types": "./dist/browser.d.mts",
|
||||
"import": "./dist/browser.mjs",
|
||||
"default": "./dist/browser.mjs"
|
||||
},
|
||||
"./readiness": {
|
||||
"types": "./dist/readiness.d.mts",
|
||||
"import": "./dist/readiness.mjs",
|
||||
@@ -26,12 +31,12 @@
|
||||
}
|
||||
},
|
||||
"scripts": {
|
||||
"build": "tsdown src/index.ts src/readiness.ts src/timeouts.ts --no-config --platform node --format esm --dts --out-dir dist --clean"
|
||||
"build": "tsdown src/index.ts src/browser.ts src/readiness.ts src/timeouts.ts --no-config --platform node --format esm --dts --out-dir dist --clean"
|
||||
},
|
||||
"dependencies": {
|
||||
"@openclaw/gateway-protocol": "workspace:*",
|
||||
"@openclaw/net-policy": "workspace:*",
|
||||
"@openclaw/retry": "workspace:*",
|
||||
"ipaddr.js": "2.4.0",
|
||||
"ws": "8.21.0"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
// Browser-safe gateway client surface. Keep Node transport/TLS dependencies out
|
||||
// of this entry so browser consumers share the wire engine without polyfills.
|
||||
export * from "./device-auth.js";
|
||||
export * from "./protocol-client.js";
|
||||
export * from "./reconnect-policy.js";
|
||||
export * from "@openclaw/gateway-protocol/client-info";
|
||||
export * from "@openclaw/gateway-protocol/connect-error-details";
|
||||
export * from "@openclaw/gateway-protocol/startup-unavailable";
|
||||
export * from "@openclaw/gateway-protocol/version";
|
||||
export type { ConnectParams, ErrorShape, EventFrame, HelloOk } from "@openclaw/gateway-protocol";
|
||||
File diff suppressed because it is too large
Load Diff
@@ -4,6 +4,7 @@ import { createServer } from "node:net";
|
||||
import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
import { WebSocket, WebSocketServer } from "ws";
|
||||
import { GatewayClient, resolveGatewayClientConnectChallengeTimeoutMs } from "./client.js";
|
||||
import type { GatewayProtocolSocket } from "./protocol-client.js";
|
||||
import {
|
||||
DEFAULT_PREAUTH_HANDSHAKE_TIMEOUT_MS,
|
||||
MAX_SAFE_TIMEOUT_DELAY_MS,
|
||||
@@ -45,20 +46,44 @@ function createOpenGatewayClient(requestTimeoutMs: number): {
|
||||
requestTimeoutMs,
|
||||
});
|
||||
const send = vi.fn();
|
||||
(
|
||||
client as unknown as {
|
||||
ws: WebSocket | { readyState: number; send: () => void; close: () => void };
|
||||
}
|
||||
).ws = {
|
||||
readyState: WebSocket.OPEN,
|
||||
send,
|
||||
close: vi.fn(),
|
||||
};
|
||||
installSyntheticSocket(client, send, vi.fn());
|
||||
return { client, send };
|
||||
}
|
||||
|
||||
function getPendingCount(client: GatewayClient): number {
|
||||
return (client as unknown as { pending: Map<string, unknown> }).pending.size;
|
||||
return protocolHarness(client).requests.pending.size;
|
||||
}
|
||||
|
||||
type ProtocolHarness = {
|
||||
socket: GatewayProtocolSocket | null;
|
||||
stopped: boolean;
|
||||
generation: number;
|
||||
reconnectSupervisor: { reset(initialMs?: number): void };
|
||||
requests: { pending: Map<string, unknown> };
|
||||
handleMessage: (socket: GatewayProtocolSocket, generation: number, raw: string) => void;
|
||||
};
|
||||
|
||||
function protocolHarness(client: GatewayClient): ProtocolHarness {
|
||||
return (client as unknown as { protocol: ProtocolHarness }).protocol;
|
||||
}
|
||||
|
||||
function installSyntheticSocket(
|
||||
client: GatewayClient,
|
||||
send: (data: string) => unknown,
|
||||
close: (code?: number, reason?: string) => unknown,
|
||||
): void {
|
||||
const socket: GatewayProtocolSocket = {
|
||||
isOpen: () => true,
|
||||
send: (data) => send(data),
|
||||
close: (code, reason) => close(code, reason),
|
||||
};
|
||||
Object.assign(protocolHarness(client), { socket, stopped: false, generation: 1 });
|
||||
(client as unknown as { ws: unknown }).ws = {
|
||||
readyState: WebSocket.OPEN,
|
||||
send,
|
||||
close,
|
||||
terminate: vi.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
function trackSettlement(promise: Promise<unknown>): () => boolean {
|
||||
@@ -86,13 +111,8 @@ function createWatchedGatewayClient(): {
|
||||
});
|
||||
const close = vi.fn();
|
||||
const send = vi.fn();
|
||||
Object.assign(client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number }, {
|
||||
ws: {
|
||||
readyState: WebSocket.OPEN,
|
||||
send,
|
||||
close,
|
||||
terminate: vi.fn(),
|
||||
},
|
||||
installSyntheticSocket(client, send, close);
|
||||
Object.assign(client as unknown as { tickIntervalMs: number; lastTick: number }, {
|
||||
tickIntervalMs: 5,
|
||||
lastTick: Date.now(),
|
||||
});
|
||||
@@ -101,9 +121,11 @@ function createWatchedGatewayClient(): {
|
||||
}
|
||||
|
||||
function handleGatewayMessage(client: GatewayClient, payload: Record<string, unknown>): void {
|
||||
(client as unknown as { handleMessage: (raw: string) => void }).handleMessage(
|
||||
JSON.stringify(payload),
|
||||
);
|
||||
const protocol = protocolHarness(client);
|
||||
if (!protocol.socket) {
|
||||
throw new Error("synthetic protocol socket missing");
|
||||
}
|
||||
protocol.handleMessage(protocol.socket, protocol.generation, JSON.stringify(payload));
|
||||
}
|
||||
|
||||
async function stopSyntheticClient(client: GatewayClient): Promise<void> {
|
||||
@@ -116,6 +138,8 @@ describe("GatewayClient", () => {
|
||||
let httpsServer: ReturnType<typeof createHttpsServer> | null = null;
|
||||
|
||||
afterEach(async () => {
|
||||
vi.useRealTimers();
|
||||
vi.restoreAllMocks();
|
||||
if (wss) {
|
||||
for (const client of wss.clients) {
|
||||
client.terminate();
|
||||
@@ -296,11 +320,7 @@ describe("GatewayClient", () => {
|
||||
helloCount += 1;
|
||||
if (helloCount === 1) {
|
||||
// Keep the real reconnect lifecycle fast without changing production defaults.
|
||||
(
|
||||
client as unknown as {
|
||||
reconnectSupervisor: { reset(initialMs?: number): void };
|
||||
}
|
||||
).reconnectSupervisor.reset(10);
|
||||
protocolHarness(client).reconnectSupervisor.reset(10);
|
||||
resolveFirstHello();
|
||||
return;
|
||||
}
|
||||
@@ -339,25 +359,21 @@ describe("GatewayClient", () => {
|
||||
|
||||
test("lets finite pending requests own their timeout when ticks are missing", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client, close } = createWatchedGatewayClient();
|
||||
const request = client.request("status", undefined, { timeoutMs: 100 });
|
||||
const requestExpectation = expect(request).rejects.toThrow(
|
||||
"gateway request timeout for status",
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
const { client, close } = createWatchedGatewayClient();
|
||||
const request = client.request("status", undefined, { timeoutMs: 100 });
|
||||
const requestExpectation = expect(request).rejects.toThrow(
|
||||
"gateway request timeout for status",
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(80);
|
||||
await requestExpectation;
|
||||
await vi.advanceTimersByTimeAsync(5);
|
||||
await vi.advanceTimersByTimeAsync(80);
|
||||
await requestExpectation;
|
||||
await vi.advanceTimersByTimeAsync(5);
|
||||
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
await stopSyntheticClient(client);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
await stopSyntheticClient(client);
|
||||
});
|
||||
|
||||
test.each([
|
||||
@@ -373,218 +389,123 @@ describe("GatewayClient", () => {
|
||||
},
|
||||
])("keeps the watchdog active for $label request", async ({ method, options }) => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client, close } = createWatchedGatewayClient();
|
||||
const request = client.request(method, undefined, options);
|
||||
const requestExpectation = expect(request).rejects.toThrow("gateway client stopped");
|
||||
const { client, close } = createWatchedGatewayClient();
|
||||
const request = client.request(method, undefined, options);
|
||||
const requestExpectation = expect(request).rejects.toThrow("gateway client stopped");
|
||||
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
await stopSyntheticClient(client);
|
||||
await requestExpectation;
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
await stopSyntheticClient(client);
|
||||
await requestExpectation;
|
||||
});
|
||||
|
||||
test("keeps the watchdog active for mixed finite and unbounded requests", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client, close } = createWatchedGatewayClient();
|
||||
const requests = [
|
||||
client.request("status", undefined, { timeoutMs: 100 }),
|
||||
client.request("chat.send", undefined, { expectFinal: true }),
|
||||
];
|
||||
const settlements = Promise.allSettled(requests);
|
||||
const { client, close } = createWatchedGatewayClient();
|
||||
const requests = [
|
||||
client.request("status", undefined, { timeoutMs: 100 }),
|
||||
client.request("chat.send", undefined, { expectFinal: true }),
|
||||
];
|
||||
const settlements = Promise.allSettled(requests);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
await stopSyntheticClient(client);
|
||||
await expect(settlements).resolves.toEqual([
|
||||
expect.objectContaining({ status: "rejected" }),
|
||||
expect.objectContaining({ status: "rejected" }),
|
||||
]);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
await stopSyntheticClient(client);
|
||||
await expect(settlements).resolves.toEqual([
|
||||
expect.objectContaining({ status: "rejected" }),
|
||||
expect.objectContaining({ status: "rejected" }),
|
||||
]);
|
||||
});
|
||||
|
||||
test("keeps an unbounded request alive while inbound ticks continue", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client, close, send } = createWatchedGatewayClient();
|
||||
const request = client.request<{ status: string }>("chat.send", undefined, {
|
||||
expectFinal: true,
|
||||
});
|
||||
const requestFrame = JSON.parse(String(send.mock.calls[0]?.[0])) as { id: string };
|
||||
const { client, close, send } = createWatchedGatewayClient();
|
||||
const request = client.request<{ status: string }>("chat.send", undefined, {
|
||||
expectFinal: true,
|
||||
});
|
||||
const requestFrame = JSON.parse(String(send.mock.calls[0]?.[0])) as { id: string };
|
||||
|
||||
for (let seq = 1; seq <= 4; seq += 1) {
|
||||
await vi.advanceTimersByTimeAsync(5);
|
||||
handleGatewayMessage(client, { type: "event", event: "tick", seq, payload: {} });
|
||||
}
|
||||
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
handleGatewayMessage(client, {
|
||||
type: "res",
|
||||
id: requestFrame.id,
|
||||
ok: true,
|
||||
payload: { status: "ok" },
|
||||
});
|
||||
await expect(request).resolves.toEqual({ status: "ok" });
|
||||
await stopSyntheticClient(client);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
for (let seq = 1; seq <= 4; seq += 1) {
|
||||
await vi.advanceTimersByTimeAsync(5);
|
||||
handleGatewayMessage(client, { type: "event", event: "tick", seq, payload: {} });
|
||||
}
|
||||
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
handleGatewayMessage(client, {
|
||||
type: "res",
|
||||
id: requestFrame.id,
|
||||
ok: true,
|
||||
payload: { status: "ok" },
|
||||
});
|
||||
await expect(request).resolves.toEqual({ status: "ok" });
|
||||
await stopSyntheticClient(client);
|
||||
});
|
||||
|
||||
test("honors explicit tick watchdog timeout threshold", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const client = new GatewayClient({
|
||||
tickWatchMinIntervalMs: 5,
|
||||
tickWatchTimeoutMs: 50,
|
||||
});
|
||||
const close = vi.fn();
|
||||
Object.assign(
|
||||
client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number },
|
||||
{
|
||||
ws: {
|
||||
readyState: WebSocket.OPEN,
|
||||
send: vi.fn(),
|
||||
close,
|
||||
},
|
||||
tickIntervalMs: 5,
|
||||
lastTick: Date.now(),
|
||||
},
|
||||
);
|
||||
const client = new GatewayClient({
|
||||
tickWatchMinIntervalMs: 5,
|
||||
tickWatchTimeoutMs: 50,
|
||||
});
|
||||
const close = vi.fn();
|
||||
installSyntheticSocket(client, vi.fn(), close);
|
||||
Object.assign(client as unknown as { tickIntervalMs: number; lastTick: number }, {
|
||||
tickIntervalMs: 5,
|
||||
lastTick: Date.now(),
|
||||
});
|
||||
|
||||
(
|
||||
client as unknown as {
|
||||
startTickWatch: () => void;
|
||||
}
|
||||
).startTickWatch();
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
(
|
||||
client as unknown as {
|
||||
startTickWatch: () => void;
|
||||
}
|
||||
).startTickWatch();
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(35);
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(35);
|
||||
expect(close).toHaveBeenCalledWith(4000, "tick timeout");
|
||||
});
|
||||
|
||||
test("clamps oversized tick watchdog intervals before scheduling", () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const setIntervalSpy = vi.spyOn(globalThis, "setInterval");
|
||||
const client = new GatewayClient({
|
||||
tickWatchMinIntervalMs: 5,
|
||||
});
|
||||
Object.assign(
|
||||
client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number },
|
||||
{
|
||||
ws: {
|
||||
readyState: WebSocket.OPEN,
|
||||
send: vi.fn(),
|
||||
close: vi.fn(),
|
||||
},
|
||||
tickIntervalMs: Number.MAX_SAFE_INTEGER,
|
||||
lastTick: Date.now(),
|
||||
},
|
||||
);
|
||||
const setIntervalSpy = vi.spyOn(globalThis, "setInterval");
|
||||
const client = new GatewayClient({
|
||||
tickWatchMinIntervalMs: 5,
|
||||
});
|
||||
Object.assign(client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number }, {
|
||||
ws: {
|
||||
readyState: WebSocket.OPEN,
|
||||
send: vi.fn(),
|
||||
close: vi.fn(),
|
||||
},
|
||||
tickIntervalMs: Number.MAX_SAFE_INTEGER,
|
||||
lastTick: Date.now(),
|
||||
});
|
||||
|
||||
(
|
||||
client as unknown as {
|
||||
startTickWatch: () => void;
|
||||
}
|
||||
).startTickWatch();
|
||||
(
|
||||
client as unknown as {
|
||||
startTickWatch: () => void;
|
||||
}
|
||||
).startTickWatch();
|
||||
|
||||
expect(setIntervalSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS);
|
||||
client.stop();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("times out unresolved requests and clears pending state", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client, send } = createOpenGatewayClient(25);
|
||||
|
||||
const requestPromise = client.request("status");
|
||||
const requestExpectation = expect(requestPromise).rejects.toThrow(
|
||||
"gateway request timeout for status",
|
||||
);
|
||||
expect(send).toHaveBeenCalledTimes(1);
|
||||
expect(getPendingCount(client)).toBe(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(25);
|
||||
|
||||
await requestExpectation;
|
||||
expect(getPendingCount(client)).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(setIntervalSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS);
|
||||
client.stop();
|
||||
});
|
||||
|
||||
test("cleans pending request state when websocket send throws", async () => {
|
||||
const client = new GatewayClient({
|
||||
requestTimeoutMs: 25,
|
||||
const { client, send } = createOpenGatewayClient(25);
|
||||
send.mockImplementationOnce(() => {
|
||||
throw new Error("synthetic send failure");
|
||||
});
|
||||
const sendError = new Error("synthetic send failure");
|
||||
(
|
||||
client as unknown as {
|
||||
ws: WebSocket | { readyState: number; send: () => void; close: () => void };
|
||||
}
|
||||
).ws = {
|
||||
readyState: WebSocket.OPEN,
|
||||
send: vi.fn(() => {
|
||||
throw sendError;
|
||||
}),
|
||||
close: vi.fn(),
|
||||
};
|
||||
|
||||
await expect(client.request("status")).rejects.toThrow("synthetic send failure");
|
||||
expect(getPendingCount(client)).toBe(0);
|
||||
});
|
||||
|
||||
test("does not auto-timeout expectFinal requests", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client, send } = createOpenGatewayClient(25);
|
||||
|
||||
const requestPromise = client.request("chat.send", undefined, { expectFinal: true });
|
||||
const isSettled = trackSettlement(requestPromise);
|
||||
expect(send).toHaveBeenCalledTimes(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(25);
|
||||
|
||||
expect(isSettled()).toBe(false);
|
||||
expect(getPendingCount(client)).toBe(1);
|
||||
|
||||
client.stop();
|
||||
await expect(requestPromise).rejects.toThrow("gateway client stopped");
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("notifies accepted expectFinal requests while continuing to wait for final", async () => {
|
||||
const client = new GatewayClient({
|
||||
requestTimeoutMs: 25,
|
||||
});
|
||||
const send = vi.fn();
|
||||
(
|
||||
client as unknown as {
|
||||
ws: WebSocket | { readyState: number; send: (data: string) => void; close: () => void };
|
||||
}
|
||||
).ws = {
|
||||
readyState: WebSocket.OPEN,
|
||||
send,
|
||||
close: vi.fn(),
|
||||
};
|
||||
const { client, send } = createOpenGatewayClient(25);
|
||||
|
||||
const onAccepted = vi.fn();
|
||||
const requestPromise = client.request<{ status: string }>("agent", undefined, {
|
||||
@@ -593,53 +514,29 @@ describe("GatewayClient", () => {
|
||||
});
|
||||
const frame = JSON.parse(String(send.mock.calls[0]?.[0])) as { id: string };
|
||||
|
||||
(
|
||||
client as unknown as {
|
||||
handleMessage: (raw: string) => void;
|
||||
}
|
||||
).handleMessage(
|
||||
JSON.stringify({
|
||||
type: "res",
|
||||
id: frame.id,
|
||||
ok: true,
|
||||
payload: { status: "accepted", runId: "run-1" },
|
||||
}),
|
||||
);
|
||||
handleGatewayMessage(client, {
|
||||
type: "res",
|
||||
id: frame.id,
|
||||
ok: true,
|
||||
payload: { status: "accepted", runId: "run-1" },
|
||||
});
|
||||
|
||||
expect(onAccepted).toHaveBeenCalledWith({ status: "accepted", runId: "run-1" });
|
||||
expect((client as unknown as { pending: Map<string, unknown> }).pending.size).toBe(1);
|
||||
expect(getPendingCount(client)).toBe(1);
|
||||
|
||||
(
|
||||
client as unknown as {
|
||||
handleMessage: (raw: string) => void;
|
||||
}
|
||||
).handleMessage(
|
||||
JSON.stringify({
|
||||
type: "res",
|
||||
id: frame.id,
|
||||
ok: true,
|
||||
payload: { status: "ok" },
|
||||
}),
|
||||
);
|
||||
handleGatewayMessage(client, {
|
||||
type: "res",
|
||||
id: frame.id,
|
||||
ok: true,
|
||||
payload: { status: "ok" },
|
||||
});
|
||||
|
||||
await expect(requestPromise).resolves.toEqual({ status: "ok" });
|
||||
expect((client as unknown as { pending: Map<string, unknown> }).pending.size).toBe(0);
|
||||
expect(getPendingCount(client)).toBe(0);
|
||||
});
|
||||
|
||||
test("aborts in-flight requests from caller AbortSignal", async () => {
|
||||
const client = new GatewayClient({
|
||||
requestTimeoutMs: 25,
|
||||
});
|
||||
const send = vi.fn();
|
||||
(
|
||||
client as unknown as {
|
||||
ws: WebSocket | { readyState: number; send: () => void; close: () => void };
|
||||
}
|
||||
).ws = {
|
||||
readyState: WebSocket.OPEN,
|
||||
send,
|
||||
close: vi.fn(),
|
||||
};
|
||||
const { client, send } = createOpenGatewayClient(25);
|
||||
|
||||
const controller = new AbortController();
|
||||
const requestPromise = client.request("status", undefined, {
|
||||
@@ -647,20 +544,24 @@ describe("GatewayClient", () => {
|
||||
timeoutMs: null,
|
||||
});
|
||||
expect(send).toHaveBeenCalledTimes(1);
|
||||
expect((client as unknown as { pending: Map<string, unknown> }).pending.size).toBe(1);
|
||||
expect(getPendingCount(client)).toBe(1);
|
||||
|
||||
controller.abort();
|
||||
|
||||
await expect(requestPromise).rejects.toThrow("gateway request aborted for status");
|
||||
expect((client as unknown as { pending: Map<string, unknown> }).pending.size).toBe(0);
|
||||
expect(getPendingCount(client)).toBe(0);
|
||||
});
|
||||
|
||||
test("clamps oversized explicit request timeouts before scheduling", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client } = createOpenGatewayClient(25);
|
||||
test.each([
|
||||
{ defaultTimeoutMs: 25, options: { timeoutMs: 2_592_010_000 } },
|
||||
{ defaultTimeoutMs: 2_592_010_000, options: undefined },
|
||||
])(
|
||||
"clamps oversized request timeouts before scheduling",
|
||||
async ({ defaultTimeoutMs, options }) => {
|
||||
vi.useFakeTimers();
|
||||
const { client } = createOpenGatewayClient(defaultTimeoutMs);
|
||||
|
||||
const requestPromise = client.request("status", undefined, { timeoutMs: 2_592_010_000 });
|
||||
const requestPromise = client.request("status", undefined, options);
|
||||
const isSettled = trackSettlement(requestPromise);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
@@ -670,56 +571,29 @@ describe("GatewayClient", () => {
|
||||
|
||||
client.stop();
|
||||
await expect(requestPromise).rejects.toThrow("gateway client stopped");
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("clamps oversized default request timeouts before scheduling", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const { client } = createOpenGatewayClient(2_592_010_000);
|
||||
|
||||
const requestPromise = client.request("status");
|
||||
const isSettled = trackSettlement(requestPromise);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
|
||||
expect(isSettled()).toBe(false);
|
||||
expect(getPendingCount(client)).toBe(1);
|
||||
|
||||
client.stop();
|
||||
await expect(requestPromise).rejects.toThrow("gateway client stopped");
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
test("clamps oversized stopAndWait timeouts before scheduling", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const client = new GatewayClient({});
|
||||
const ws = {
|
||||
readyState: WebSocket.OPEN,
|
||||
close: vi.fn(),
|
||||
terminate: vi.fn(),
|
||||
};
|
||||
(client as unknown as { ws: unknown }).ws = ws;
|
||||
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
|
||||
const client = new GatewayClient({});
|
||||
const ws = {
|
||||
readyState: WebSocket.OPEN,
|
||||
close: vi.fn(),
|
||||
terminate: vi.fn(),
|
||||
};
|
||||
(client as unknown as { ws: unknown }).ws = ws;
|
||||
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
|
||||
|
||||
const stopPromise = client.stopAndWait({ timeoutMs: Number.MAX_SAFE_INTEGER });
|
||||
const stopPromise = client.stopAndWait({ timeoutMs: Number.MAX_SAFE_INTEGER });
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(ws.terminate).not.toHaveBeenCalled();
|
||||
expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(ws.terminate).not.toHaveBeenCalled();
|
||||
expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(249);
|
||||
await expect(stopPromise).resolves.toBeUndefined();
|
||||
expect(ws.terminate).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(249);
|
||||
await expect(stopPromise).resolves.toBeUndefined();
|
||||
expect(ws.terminate).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test("rejects mismatched tls fingerprint", async () => {
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
import type { ErrorShape, EventFrame, HelloOk } from "@openclaw/gateway-protocol";
|
||||
export type GatewayProtocolSocket = {
|
||||
isOpen: () => boolean;
|
||||
send: (data: string) => void;
|
||||
close: (code?: number, reason?: string) => void;
|
||||
};
|
||||
export type GatewayProtocolSocketHandlers = {
|
||||
open: () => void;
|
||||
message: (data: string) => void;
|
||||
close: (code: number, reason: string) => void;
|
||||
error: (error: Error) => void;
|
||||
};
|
||||
export type GatewayProtocolRequestOptions = {
|
||||
timeoutMs?: number | null;
|
||||
expectFinal?: boolean;
|
||||
onAccepted?: (payload: unknown) => void;
|
||||
signal?: AbortSignal;
|
||||
};
|
||||
type GatewayProtocolConnectContext<TPlan> = {
|
||||
generation: number;
|
||||
nonce: string | null;
|
||||
plan: TPlan;
|
||||
};
|
||||
export type GatewayProtocolCloseContext = {
|
||||
code: number;
|
||||
reason: string;
|
||||
generation: number;
|
||||
socketOpened: boolean;
|
||||
helloReceived: boolean;
|
||||
connectRequestSent: boolean;
|
||||
connectFailure?: { error: Error; reconnectDelayMs?: number };
|
||||
};
|
||||
type GatewayProtocolConnectDecision = {
|
||||
closeCode: number;
|
||||
closeReason: string;
|
||||
reconnectDelayMs?: number;
|
||||
stop?: boolean;
|
||||
error?: Error;
|
||||
};
|
||||
type GatewayProtocolCloseDecision = {
|
||||
retry: boolean;
|
||||
notify: boolean;
|
||||
reconnectDelayMs?: number;
|
||||
pendingError?: Error;
|
||||
};
|
||||
export type GatewayProtocolTiming<TPlan> = {
|
||||
phase:
|
||||
| "socket-open"
|
||||
| "challenge"
|
||||
| "fallback"
|
||||
| "device-identity-ready"
|
||||
| "connect-plan-ready"
|
||||
| "request-sent"
|
||||
| "hello"
|
||||
| "failed";
|
||||
generation: number;
|
||||
durationMs: number;
|
||||
phaseDurationMs: number;
|
||||
hasChallenge: boolean;
|
||||
usedFallback: boolean;
|
||||
plan?: TPlan;
|
||||
detail?: unknown;
|
||||
};
|
||||
export type GatewayProtocolRequestTiming = {
|
||||
id: string;
|
||||
method: string;
|
||||
ok: boolean;
|
||||
durationMs: number;
|
||||
startedAtMs: number;
|
||||
endedAtMs: number;
|
||||
errorCode?: string;
|
||||
};
|
||||
export type GatewayProtocolClientOptions<TPlan> = {
|
||||
createSocket: (handlers: GatewayProtocolSocketHandlers) => GatewayProtocolSocket;
|
||||
createRequestId: () => string;
|
||||
createRequestError?: (error: Partial<ErrorShape>) => GatewayProtocolRequestError;
|
||||
createRequestTimeoutError?: (method: string, timeoutMs: number) => Error;
|
||||
createRequestAbortError?: (method: string) => Error;
|
||||
buildConnectPlan: (params: {
|
||||
nonce: string | null;
|
||||
generation: number;
|
||||
}) => TPlan | Promise<TPlan>;
|
||||
buildConnectParams: (plan: TPlan) => unknown;
|
||||
onConnectPlanError?: (error: Error) => GatewayProtocolConnectDecision;
|
||||
onConnectHello?: (hello: HelloOk, context: GatewayProtocolConnectContext<TPlan>) => void;
|
||||
onHello?: (hello: HelloOk) => void;
|
||||
onConnectFailure?: (
|
||||
error: GatewayProtocolRequestError,
|
||||
context: GatewayProtocolConnectContext<TPlan>,
|
||||
) => GatewayProtocolConnectDecision;
|
||||
resolveClose: (context: GatewayProtocolCloseContext) => GatewayProtocolCloseDecision;
|
||||
onClose?: (context: GatewayProtocolCloseContext, decision: GatewayProtocolCloseDecision) => void;
|
||||
notifyStoppedClose?: boolean;
|
||||
onConnectError?: (error: Error) => void;
|
||||
onSocketFactoryError?: (error: Error) => void;
|
||||
onParseError?: (error: unknown) => void;
|
||||
onEvent?: (event: EventFrame) => void;
|
||||
onGap?: (info: { expected: number; received: number }) => void;
|
||||
onActivity?: () => void;
|
||||
onTiming?: (timing: GatewayProtocolTiming<TPlan>) => void;
|
||||
onRequestTiming?: (timing: GatewayProtocolRequestTiming) => void;
|
||||
onCallbackError?: (label: string, error: unknown) => void;
|
||||
handshake:
|
||||
| { mode: "fallback"; timeoutMs: number }
|
||||
| {
|
||||
mode: "require-challenge";
|
||||
timeoutMs: number;
|
||||
timeoutMessage?: (elapsedMs: number) => string;
|
||||
};
|
||||
reconnect: { initialMs: number; multiplier: number; maxMs: number };
|
||||
requestTimeoutMs?: number;
|
||||
nowMs?: () => number;
|
||||
rethrowSocketFactoryError?: (error: Error) => boolean;
|
||||
};
|
||||
export class GatewayProtocolRequestError extends Error {
|
||||
readonly code: string;
|
||||
readonly details?: unknown;
|
||||
readonly retryable?: boolean;
|
||||
readonly retryAfterMs?: number;
|
||||
|
||||
constructor(error: Partial<ErrorShape>) {
|
||||
super(error.message ?? "request failed");
|
||||
this.name = "GatewayProtocolRequestError";
|
||||
this.code = error.code ?? "UNAVAILABLE";
|
||||
this.details = error.details;
|
||||
this.retryable = error.retryable;
|
||||
this.retryAfterMs = error.retryAfterMs;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,462 @@
|
||||
import type { EventFrame, HelloOk } from "@openclaw/gateway-protocol";
|
||||
import {
|
||||
isGatewayEventFrame,
|
||||
isGatewayResponseFrame,
|
||||
} from "@openclaw/gateway-protocol/frame-guards";
|
||||
import { RetrySupervisor, sleepWithAbort } from "@openclaw/retry";
|
||||
import {
|
||||
GatewayProtocolRequestError,
|
||||
type GatewayProtocolClientOptions,
|
||||
type GatewayProtocolCloseContext,
|
||||
type GatewayProtocolRequestOptions,
|
||||
type GatewayProtocolSocket,
|
||||
type GatewayProtocolTiming,
|
||||
} from "./protocol-client-types.js";
|
||||
import { GatewayProtocolRequests } from "./protocol-requests.js";
|
||||
|
||||
export * from "./protocol-client-types.js";
|
||||
|
||||
type ConnectTimingState = {
|
||||
generation: number;
|
||||
startedAtMs: number;
|
||||
lastAtMs: number;
|
||||
hasChallenge: boolean;
|
||||
usedFallback: boolean;
|
||||
};
|
||||
type CloseSnapshot = Omit<GatewayProtocolCloseContext, "code" | "reason">;
|
||||
|
||||
/**
|
||||
* Browser-safe gateway wire client. Environment adapters own transport and auth
|
||||
* policy; this class owns the single socket/handshake/reconnect/frame state machine.
|
||||
*/
|
||||
export class GatewayProtocolClient<TPlan> {
|
||||
private socket: GatewayProtocolSocket | null = null;
|
||||
private readonly requests: GatewayProtocolRequests<TPlan>;
|
||||
private listeners = new Set<(event: EventFrame) => void>();
|
||||
private stopped = true;
|
||||
private generation = 0;
|
||||
private lastSeq: number | null = null;
|
||||
private connectNonce: string | null = null;
|
||||
private connectSent = false;
|
||||
private connectRequestSent = false;
|
||||
private handshakeTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private readonly reconnectSupervisor: RetrySupervisor;
|
||||
private socketOpened = false;
|
||||
private helloReceived = false;
|
||||
private connectFailure: GatewayProtocolCloseContext["connectFailure"];
|
||||
private connectTiming: ConnectTimingState | null = null;
|
||||
private stoppedSocket?: { socket: GatewayProtocolSocket; context: CloseSnapshot };
|
||||
|
||||
constructor(private readonly opts: GatewayProtocolClientOptions<TPlan>) {
|
||||
this.reconnectSupervisor = new RetrySupervisor({
|
||||
initialMs: opts.reconnect.initialMs,
|
||||
maxMs: opts.reconnect.maxMs,
|
||||
factor: opts.reconnect.multiplier,
|
||||
jitter: 0,
|
||||
});
|
||||
this.requests = new GatewayProtocolRequests(opts);
|
||||
}
|
||||
|
||||
get connected(): boolean {
|
||||
return this.socket?.isOpen() ?? false;
|
||||
}
|
||||
|
||||
get hasPendingRequests(): boolean {
|
||||
return this.requests.hasPending;
|
||||
}
|
||||
|
||||
get connecting(): boolean {
|
||||
return this.connectSent && !this.helloReceived;
|
||||
}
|
||||
|
||||
get hasUnboundedPendingRequests(): boolean {
|
||||
return this.requests.hasUnboundedPending;
|
||||
}
|
||||
|
||||
start(): void {
|
||||
this.stopped = false;
|
||||
this.reconnectSupervisor.cancel();
|
||||
this.connect();
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
this.stopped = true;
|
||||
this.clearHandshakeTimer();
|
||||
this.reconnectSupervisor.reset();
|
||||
const socket = this.socket;
|
||||
if (socket && this.opts.notifyStoppedClose) {
|
||||
// Node callers observe the transport's final close during explicit stop;
|
||||
// browser callers intentionally suppress it.
|
||||
this.stoppedSocket = { socket, context: this.closeContext() };
|
||||
}
|
||||
this.socket = null;
|
||||
this.connectFailure = undefined;
|
||||
this.connectTiming = null;
|
||||
this.requests.flush(new Error("gateway client stopped"));
|
||||
if (!socket) {
|
||||
return;
|
||||
}
|
||||
socket.close();
|
||||
}
|
||||
|
||||
request<T = unknown>(
|
||||
method: string,
|
||||
params?: unknown,
|
||||
options?: GatewayProtocolRequestOptions,
|
||||
): Promise<T> {
|
||||
const socket = this.socket;
|
||||
if (!socket?.isOpen()) {
|
||||
return Promise.reject(new Error("gateway not connected"));
|
||||
}
|
||||
if (typeof method !== "string" || method.length === 0) {
|
||||
return Promise.reject(new Error("invalid request frame: method must be a non-empty string"));
|
||||
}
|
||||
return this.requests.request<T>(socket, method, params, options);
|
||||
}
|
||||
|
||||
addEventListener(listener: (event: EventFrame) => void): () => void {
|
||||
this.listeners.add(listener);
|
||||
return () => this.listeners.delete(listener);
|
||||
}
|
||||
|
||||
closeSocket(code?: number, reason?: string): void {
|
||||
this.socket?.close(code, reason);
|
||||
}
|
||||
|
||||
resetReconnectBackoff(initialMs: number): void {
|
||||
this.reconnectSupervisor.reset(initialMs);
|
||||
}
|
||||
|
||||
recordTiming(
|
||||
phase: GatewayProtocolTiming<TPlan>["phase"],
|
||||
generation: number,
|
||||
plan?: TPlan,
|
||||
detail?: unknown,
|
||||
): void {
|
||||
const now = this.nowMs();
|
||||
const state = this.connectTiming;
|
||||
if (!state || state.generation !== generation) {
|
||||
return;
|
||||
}
|
||||
state.hasChallenge ||= phase === "challenge";
|
||||
state.usedFallback ||= phase === "fallback";
|
||||
this.invoke("connect timing", () =>
|
||||
this.opts.onTiming?.({
|
||||
phase,
|
||||
generation,
|
||||
durationMs: Math.max(0, now - state.startedAtMs),
|
||||
phaseDurationMs: Math.max(0, now - state.lastAtMs),
|
||||
hasChallenge: state.hasChallenge,
|
||||
usedFallback: state.usedFallback,
|
||||
plan,
|
||||
detail,
|
||||
}),
|
||||
);
|
||||
state.lastAtMs = now;
|
||||
if (phase === "hello" || phase === "failed") {
|
||||
this.connectTiming = null;
|
||||
}
|
||||
}
|
||||
|
||||
private connect(): void {
|
||||
if (this.stopped) {
|
||||
return;
|
||||
}
|
||||
const generation = this.generation + 1;
|
||||
this.connectNonce = null;
|
||||
this.connectSent = false;
|
||||
this.connectRequestSent = false;
|
||||
this.socketOpened = false;
|
||||
this.helloReceived = false;
|
||||
this.connectFailure = undefined;
|
||||
let socket: GatewayProtocolSocket;
|
||||
try {
|
||||
socket = this.opts.createSocket({
|
||||
open: () => this.handleOpen(socket, generation),
|
||||
message: (data) => this.handleMessage(socket, generation, data),
|
||||
close: (code, reason) => this.handleClose(socket, generation, code, reason),
|
||||
error: (error) => this.handleSocketError(socket, generation, error),
|
||||
});
|
||||
} catch (error) {
|
||||
const normalized = error instanceof Error ? error : new Error(String(error));
|
||||
this.opts.onSocketFactoryError?.(normalized);
|
||||
this.opts.onConnectError?.(normalized);
|
||||
if (this.opts.rethrowSocketFactoryError?.(normalized)) {
|
||||
throw normalized;
|
||||
}
|
||||
return;
|
||||
}
|
||||
this.generation = generation;
|
||||
this.socket = socket;
|
||||
const now = this.nowMs();
|
||||
this.connectTiming = {
|
||||
generation,
|
||||
startedAtMs: now,
|
||||
lastAtMs: now,
|
||||
hasChallenge: false,
|
||||
usedFallback: false,
|
||||
};
|
||||
}
|
||||
|
||||
private handleOpen(socket: GatewayProtocolSocket, generation: number): void {
|
||||
if (!this.isActive(socket, generation)) {
|
||||
return;
|
||||
}
|
||||
this.socketOpened = true;
|
||||
this.recordTiming("socket-open", generation);
|
||||
if (this.connectNonce) {
|
||||
this.sendConnect(socket, generation);
|
||||
return;
|
||||
}
|
||||
this.armHandshakeTimer(socket, generation);
|
||||
}
|
||||
|
||||
private armHandshakeTimer(socket: GatewayProtocolSocket, generation: number): void {
|
||||
this.clearHandshakeTimer();
|
||||
const armedAt = Date.now();
|
||||
this.handshakeTimer = setTimeout(() => {
|
||||
this.handshakeTimer = null;
|
||||
if (!this.isActive(socket, generation) || this.connectSent || !socket.isOpen()) {
|
||||
return;
|
||||
}
|
||||
if (this.opts.handshake.mode === "fallback") {
|
||||
this.recordTiming("fallback", generation);
|
||||
this.sendConnect(socket, generation);
|
||||
return;
|
||||
}
|
||||
const elapsedMs = Date.now() - armedAt;
|
||||
const error = new Error(
|
||||
this.opts.handshake.timeoutMessage?.(elapsedMs) ??
|
||||
`gateway connect challenge timeout after ${elapsedMs}ms`,
|
||||
);
|
||||
this.opts.onConnectError?.(error);
|
||||
socket.close(1008, "connect challenge timeout");
|
||||
}, this.opts.handshake.timeoutMs);
|
||||
this.handshakeTimer.unref?.();
|
||||
}
|
||||
|
||||
private sendConnect(socket: GatewayProtocolSocket, generation: number): void {
|
||||
if (!this.isActive(socket, generation) || !socket.isOpen() || this.connectSent) {
|
||||
return;
|
||||
}
|
||||
this.connectSent = true;
|
||||
this.clearHandshakeTimer();
|
||||
let planOrPromise: TPlan | Promise<TPlan>;
|
||||
try {
|
||||
planOrPromise = this.opts.buildConnectPlan({ nonce: this.connectNonce, generation });
|
||||
} catch (error) {
|
||||
this.handleConnectPlanError(socket, generation, error);
|
||||
return;
|
||||
}
|
||||
if (planOrPromise instanceof Promise) {
|
||||
void planOrPromise
|
||||
.then((plan) => this.sendConnectPlan(socket, generation, plan))
|
||||
.catch((error: unknown) => this.handleConnectPlanError(socket, generation, error));
|
||||
return;
|
||||
}
|
||||
this.sendConnectPlan(socket, generation, planOrPromise);
|
||||
}
|
||||
|
||||
private handleConnectPlanError(
|
||||
socket: GatewayProtocolSocket,
|
||||
generation: number,
|
||||
error: unknown,
|
||||
): void {
|
||||
if (!this.isActive(socket, generation)) {
|
||||
return;
|
||||
}
|
||||
const normalized = error instanceof Error ? error : new Error(String(error));
|
||||
const outcome = this.opts.onConnectPlanError?.(normalized) ?? {
|
||||
closeCode: 1008,
|
||||
closeReason: "connect failed",
|
||||
};
|
||||
this.opts.onConnectError?.(outcome.error ?? normalized);
|
||||
if (outcome.stop) {
|
||||
this.stopped = true;
|
||||
}
|
||||
socket.close(outcome.closeCode, outcome.closeReason);
|
||||
}
|
||||
|
||||
private sendConnectPlan(socket: GatewayProtocolSocket, generation: number, plan: TPlan): void {
|
||||
if (!this.isActive(socket, generation) || !socket.isOpen()) {
|
||||
return;
|
||||
}
|
||||
const context = { generation, nonce: this.connectNonce, plan };
|
||||
this.recordTiming("connect-plan-ready", generation, plan);
|
||||
this.recordTiming("request-sent", generation, plan);
|
||||
this.connectRequestSent = true;
|
||||
void this.requests
|
||||
.request<HelloOk>(socket, "connect", this.opts.buildConnectParams(plan))
|
||||
.then((hello) => {
|
||||
if (!this.isActive(socket, generation)) {
|
||||
return;
|
||||
}
|
||||
this.helloReceived = true;
|
||||
this.connectFailure = undefined;
|
||||
this.reconnectSupervisor.reset();
|
||||
this.recordTiming("hello", generation, plan);
|
||||
this.opts.onConnectHello?.(hello, context);
|
||||
this.invoke("hello", () => this.opts.onHello?.(hello));
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
if (!this.isActive(socket, generation)) {
|
||||
return;
|
||||
}
|
||||
const requestError =
|
||||
error instanceof GatewayProtocolRequestError
|
||||
? error
|
||||
: new GatewayProtocolRequestError({ message: String(error) });
|
||||
const outcome = this.opts.onConnectFailure?.(requestError, context) ?? {
|
||||
closeCode: 1008,
|
||||
closeReason: "connect failed",
|
||||
};
|
||||
this.connectFailure = { error: requestError, reconnectDelayMs: outcome.reconnectDelayMs };
|
||||
if (outcome.stop) {
|
||||
this.stopped = true;
|
||||
}
|
||||
socket.close(outcome.closeCode, outcome.closeReason);
|
||||
});
|
||||
}
|
||||
|
||||
private handleMessage(socket: GatewayProtocolSocket, generation: number, raw: string): void {
|
||||
if (!this.isActive(socket, generation)) {
|
||||
return;
|
||||
}
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(raw);
|
||||
} catch (error) {
|
||||
this.opts.onParseError?.(error);
|
||||
return;
|
||||
}
|
||||
if (isGatewayEventFrame(parsed)) {
|
||||
this.opts.onActivity?.();
|
||||
if (parsed.event === "connect.challenge") {
|
||||
const payload = parsed.payload as { nonce?: unknown } | undefined;
|
||||
const nonce = typeof payload?.nonce === "string" ? payload.nonce.trim() : "";
|
||||
if (!nonce) {
|
||||
if (this.opts.handshake.mode === "require-challenge") {
|
||||
const error = new Error("gateway connect challenge missing nonce");
|
||||
this.opts.onConnectError?.(error);
|
||||
socket.close(1008, "connect challenge missing nonce");
|
||||
}
|
||||
return;
|
||||
}
|
||||
this.connectNonce = nonce;
|
||||
this.recordTiming("challenge", generation);
|
||||
this.sendConnect(socket, generation);
|
||||
return;
|
||||
}
|
||||
const seq = typeof parsed.seq === "number" ? parsed.seq : null;
|
||||
if (seq !== null) {
|
||||
if (this.lastSeq !== null && seq > this.lastSeq + 1) {
|
||||
const expected = this.lastSeq + 1;
|
||||
this.invoke("gap", () => this.opts.onGap?.({ expected, received: seq }));
|
||||
}
|
||||
this.lastSeq = seq;
|
||||
}
|
||||
this.invoke("event", () => this.opts.onEvent?.(parsed));
|
||||
for (const listener of this.listeners) {
|
||||
this.invoke("event listener", () => listener(parsed));
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (!isGatewayResponseFrame(parsed)) {
|
||||
return;
|
||||
}
|
||||
this.opts.onActivity?.();
|
||||
this.requests.handleResponse(parsed);
|
||||
}
|
||||
|
||||
private handleClose(
|
||||
socket: GatewayProtocolSocket,
|
||||
generation: number,
|
||||
code: number,
|
||||
reason: string,
|
||||
): void {
|
||||
if (this.socket !== socket) {
|
||||
if (this.stoppedSocket?.socket === socket) {
|
||||
const context = { ...this.stoppedSocket.context, code, reason };
|
||||
this.stoppedSocket = undefined;
|
||||
this.invoke("close", () => this.opts.onClose?.(context, { retry: false, notify: true }));
|
||||
}
|
||||
return;
|
||||
}
|
||||
this.socket = null;
|
||||
this.clearHandshakeTimer();
|
||||
const context: GatewayProtocolCloseContext = {
|
||||
...this.closeContext(),
|
||||
code,
|
||||
reason,
|
||||
generation,
|
||||
};
|
||||
this.connectFailure = undefined;
|
||||
const decision = this.opts.resolveClose(context);
|
||||
this.requests.flush(
|
||||
decision.pendingError ??
|
||||
context.connectFailure?.error ??
|
||||
new Error(`gateway closed (${code}): ${reason}`),
|
||||
);
|
||||
this.invoke("close", () => this.opts.onClose?.(context, decision));
|
||||
if (decision.retry && !this.stopped) {
|
||||
this.scheduleReconnect(decision.reconnectDelayMs ?? context.connectFailure?.reconnectDelayMs);
|
||||
}
|
||||
}
|
||||
|
||||
private handleSocketError(socket: GatewayProtocolSocket, generation: number, error: Error): void {
|
||||
if (!this.isActive(socket, generation) || this.connectSent) {
|
||||
return;
|
||||
}
|
||||
this.opts.onConnectError?.(error);
|
||||
}
|
||||
|
||||
private scheduleReconnect(overrideMs?: number): void {
|
||||
if (overrideMs !== undefined) {
|
||||
// Retry-After is a floor for this wait, not a failed attempt. Preserve
|
||||
// the exponential sequence for the next transport failure.
|
||||
this.reconnectSupervisor.nextDelayOverrideMs = overrideMs;
|
||||
}
|
||||
const retry = this.reconnectSupervisor.next();
|
||||
if (!retry) {
|
||||
return;
|
||||
}
|
||||
// Ignore cancelled sleeps only; reconnect start failures stay observable.
|
||||
void sleepWithAbort(retry.delayMs, retry.signal).then(
|
||||
() => this.connect(),
|
||||
() => {},
|
||||
);
|
||||
}
|
||||
|
||||
private closeContext(): CloseSnapshot {
|
||||
return {
|
||||
generation: this.generation,
|
||||
socketOpened: this.socketOpened,
|
||||
helloReceived: this.helloReceived,
|
||||
connectRequestSent: this.connectRequestSent,
|
||||
connectFailure: this.connectFailure,
|
||||
};
|
||||
}
|
||||
|
||||
private isActive(socket: GatewayProtocolSocket, generation: number): boolean {
|
||||
return !this.stopped && this.socket === socket && this.generation === generation;
|
||||
}
|
||||
|
||||
private nowMs(): number {
|
||||
return this.opts.nowMs?.() ?? Date.now();
|
||||
}
|
||||
|
||||
private clearHandshakeTimer(): void {
|
||||
if (this.handshakeTimer) {
|
||||
clearTimeout(this.handshakeTimer);
|
||||
this.handshakeTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private invoke(label: string, callback: () => void): void {
|
||||
try {
|
||||
callback();
|
||||
} catch (error) {
|
||||
this.opts.onCallbackError?.(label, error);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
import type { ResponseFrame } from "@openclaw/gateway-protocol";
|
||||
import {
|
||||
GatewayProtocolRequestError,
|
||||
type GatewayProtocolClientOptions,
|
||||
type GatewayProtocolRequestOptions,
|
||||
type GatewayProtocolSocket,
|
||||
} from "./protocol-client-types.js";
|
||||
|
||||
type Pending = {
|
||||
resolve: (value: unknown) => void;
|
||||
reject: (error: Error) => void;
|
||||
expectFinal: boolean;
|
||||
acceptedNotified: boolean;
|
||||
onAccepted?: (payload: unknown) => void;
|
||||
cleanup?: () => void;
|
||||
unbounded: boolean;
|
||||
method: string;
|
||||
startedAtMs: number;
|
||||
};
|
||||
|
||||
export class GatewayProtocolRequests<TPlan> {
|
||||
private readonly pending = new Map<string, Pending>();
|
||||
|
||||
constructor(private readonly opts: GatewayProtocolClientOptions<TPlan>) {}
|
||||
|
||||
get hasPending(): boolean {
|
||||
return this.pending.size > 0;
|
||||
}
|
||||
|
||||
get hasUnboundedPending(): boolean {
|
||||
return [...this.pending.values()].some((pending) => pending.unbounded);
|
||||
}
|
||||
|
||||
request<T>(
|
||||
socket: GatewayProtocolSocket,
|
||||
method: string,
|
||||
params?: unknown,
|
||||
options?: GatewayProtocolRequestOptions,
|
||||
): Promise<T> {
|
||||
const id = this.opts.createRequestId();
|
||||
const frame = { type: "req", id, method, params };
|
||||
const timeoutMs =
|
||||
options?.timeoutMs === null ? undefined : (options?.timeoutMs ?? this.opts.requestTimeoutMs);
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
let timeout: ReturnType<typeof setTimeout> | undefined;
|
||||
const pending: Pending = {
|
||||
resolve: (value) => resolve(value as T),
|
||||
reject,
|
||||
expectFinal: options?.expectFinal === true,
|
||||
acceptedNotified: false,
|
||||
onAccepted: options?.onAccepted,
|
||||
unbounded: timeoutMs === undefined,
|
||||
method,
|
||||
startedAtMs: this.opts.nowMs?.() ?? Date.now(),
|
||||
};
|
||||
const onAbort = () => {
|
||||
this.pending.delete(id);
|
||||
if (timeout) {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
this.finishTiming(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(() => {
|
||||
this.pending.delete(id);
|
||||
options?.signal?.removeEventListener("abort", onAbort);
|
||||
this.finishTiming(id, pending, false, "CLIENT_TIMEOUT");
|
||||
reject(
|
||||
this.opts.createRequestTimeoutError?.(method, timeoutMs) ??
|
||||
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(frame));
|
||||
} catch (error) {
|
||||
this.pending.delete(id);
|
||||
cleanup();
|
||||
this.finishTiming(id, pending, false, "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();
|
||||
}
|
||||
|
||||
private finishTiming(id: string, pending: Pending, ok: boolean, errorCode?: string): void {
|
||||
const endedAtMs = this.opts.nowMs?.() ?? Date.now();
|
||||
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 invoke(label: string, callback: () => void): void {
|
||||
try {
|
||||
callback();
|
||||
} catch (error) {
|
||||
this.opts.onCallbackError?.(label, error);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
import {
|
||||
ConnectErrorDetailCodes,
|
||||
readConnectErrorDetailCode,
|
||||
readPairingConnectErrorDetails,
|
||||
} from "@openclaw/gateway-protocol/connect-error-details";
|
||||
|
||||
const NON_RECOVERABLE_AUTH_ERRORS = new Set<string>([
|
||||
ConnectErrorDetailCodes.AUTH_TOKEN_MISSING,
|
||||
ConnectErrorDetailCodes.AUTH_BOOTSTRAP_TOKEN_INVALID,
|
||||
ConnectErrorDetailCodes.AUTH_PASSWORD_MISSING,
|
||||
ConnectErrorDetailCodes.AUTH_PASSWORD_MISMATCH,
|
||||
ConnectErrorDetailCodes.AUTH_RATE_LIMITED,
|
||||
ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH,
|
||||
ConnectErrorDetailCodes.AUTH_SCOPE_MISMATCH,
|
||||
ConnectErrorDetailCodes.PAIRING_REQUIRED,
|
||||
ConnectErrorDetailCodes.CONTROL_UI_DEVICE_IDENTITY_REQUIRED,
|
||||
ConnectErrorDetailCodes.DEVICE_IDENTITY_REQUIRED,
|
||||
]);
|
||||
|
||||
export function shouldPauseGatewayReconnect(params: {
|
||||
details?: unknown;
|
||||
deviceTokenRetryPending?: boolean;
|
||||
tokenMismatchIsTerminal?: boolean;
|
||||
protocolMismatchIsTerminal?: boolean;
|
||||
clientVersionMismatchIsTerminal?: boolean;
|
||||
}): boolean {
|
||||
const code = readConnectErrorDetailCode(params.details);
|
||||
if (!code) {
|
||||
return false;
|
||||
}
|
||||
const pairing = readPairingConnectErrorDetails(params.details);
|
||||
if (
|
||||
code === ConnectErrorDetailCodes.PAIRING_REQUIRED &&
|
||||
(pairing?.pauseReconnect === false || pairing?.recommendedNextStep === "wait_then_retry")
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (code === ConnectErrorDetailCodes.AUTH_TOKEN_MISMATCH) {
|
||||
return params.tokenMismatchIsTerminal === true && !params.deviceTokenRetryPending;
|
||||
}
|
||||
return (
|
||||
NON_RECOVERABLE_AUTH_ERRORS.has(code) ||
|
||||
(params.protocolMismatchIsTerminal === true &&
|
||||
code === ConnectErrorDetailCodes.PROTOCOL_MISMATCH) ||
|
||||
(params.clientVersionMismatchIsTerminal === true &&
|
||||
code === ConnectErrorDetailCodes.CLIENT_VERSION_MISMATCH)
|
||||
);
|
||||
}
|
||||
Generated
+6
-3
@@ -1991,12 +1991,12 @@ importers:
|
||||
'@openclaw/gateway-protocol':
|
||||
specifier: workspace:*
|
||||
version: link:../gateway-protocol
|
||||
'@openclaw/net-policy':
|
||||
specifier: workspace:*
|
||||
version: link:../net-policy
|
||||
'@openclaw/retry':
|
||||
specifier: workspace:*
|
||||
version: link:../retry
|
||||
ipaddr.js:
|
||||
specifier: 2.4.0
|
||||
version: 2.4.0
|
||||
ws:
|
||||
specifier: 8.21.0
|
||||
version: 8.21.0
|
||||
@@ -2097,6 +2097,9 @@ importers:
|
||||
|
||||
ui:
|
||||
dependencies:
|
||||
'@openclaw/gateway-client':
|
||||
specifier: workspace:*
|
||||
version: link:../packages/gateway-client
|
||||
'@codemirror/commands':
|
||||
specifier: 6.10.4
|
||||
version: 6.10.4
|
||||
|
||||
+84
-78
@@ -276,6 +276,10 @@ beforeAll(async () => {
|
||||
await loadGatewayClientModule();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
describe("GatewayClient security checks", () => {
|
||||
const envSnapshot = captureEnv([
|
||||
"OPENCLAW_ALLOW_INSECURE_PRIVATE_WS",
|
||||
@@ -822,25 +826,21 @@ describe("GatewayClient close handling", () => {
|
||||
|
||||
it("keeps a managed reconnect timer after gateway restart closes", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
|
||||
client.start();
|
||||
getLatestWs().emitClose(1012, "service restart");
|
||||
client.start();
|
||||
getLatestWs().emitClose(1012, "service restart");
|
||||
|
||||
expect(wsInstances).toHaveLength(1);
|
||||
await vi.advanceTimersByTimeAsync(999);
|
||||
expect(wsInstances).toHaveLength(1);
|
||||
expect(wsInstances).toHaveLength(1);
|
||||
await vi.advanceTimersByTimeAsync(999);
|
||||
expect(wsInstances).toHaveLength(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
|
||||
expect(wsInstances).toHaveLength(2);
|
||||
client.stop();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(wsInstances).toHaveLength(2);
|
||||
client.stop();
|
||||
});
|
||||
|
||||
it("reconnects quietly after one clean pre-hello close with a pending connect", async () => {
|
||||
@@ -987,99 +987,86 @@ describe("GatewayClient close handling", () => {
|
||||
|
||||
it("clears pending reconnect timers on stop", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
|
||||
client.start();
|
||||
getLatestWs().emitClose(1012, "service restart");
|
||||
client.stop();
|
||||
client.start();
|
||||
getLatestWs().emitClose(1012, "service restart");
|
||||
client.stop();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
|
||||
expect(wsInstances).toHaveLength(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(wsInstances).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("force-terminates a lingering socket after stop", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
|
||||
client.start();
|
||||
const ws = getLatestWs();
|
||||
ws.autoCloseOnClose = false;
|
||||
client.start();
|
||||
const ws = getLatestWs();
|
||||
ws.autoCloseOnClose = false;
|
||||
|
||||
client.stop();
|
||||
client.stop();
|
||||
|
||||
expect(ws.closeCalls).toBe(1);
|
||||
expect(ws.terminateCalls).toBe(0);
|
||||
expect(ws.closeCalls).toBe(1);
|
||||
expect(ws.terminateCalls).toBe(0);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
expect(ws.terminateCalls).toBe(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(ws.terminateCalls).toBe(1);
|
||||
});
|
||||
|
||||
it("does not force-terminate a socket that closes during stop", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
const onClose = vi.fn();
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
onClose,
|
||||
});
|
||||
|
||||
client.start();
|
||||
const ws = getLatestWs();
|
||||
client.start();
|
||||
const ws = getLatestWs();
|
||||
|
||||
client.stop();
|
||||
client.stop();
|
||||
|
||||
expect(ws.closeCalls).toBe(1);
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(ws.closeCalls).toBe(1);
|
||||
expect(onClose).toHaveBeenCalledOnce();
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
expect(ws.terminateCalls).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(ws.terminateCalls).toBe(0);
|
||||
});
|
||||
|
||||
it("waits for a lingering socket to terminate in stopAndWait", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
});
|
||||
|
||||
client.start();
|
||||
const ws = getLatestWs();
|
||||
ws.autoCloseOnClose = false;
|
||||
client.start();
|
||||
const ws = getLatestWs();
|
||||
ws.autoCloseOnClose = false;
|
||||
|
||||
let settled = false;
|
||||
const stopPromise = client.stopAndWait().then(() => {
|
||||
settled = true;
|
||||
});
|
||||
let settled = false;
|
||||
const stopPromise = client.stopAndWait().then(() => {
|
||||
settled = true;
|
||||
});
|
||||
|
||||
expect(ws.closeCalls).toBe(1);
|
||||
expect(settled).toBe(false);
|
||||
expect(ws.closeCalls).toBe(1);
|
||||
expect(settled).toBe(false);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(249);
|
||||
expect(ws.terminateCalls).toBe(0);
|
||||
expect(settled).toBe(false);
|
||||
await vi.advanceTimersByTimeAsync(249);
|
||||
expect(ws.terminateCalls).toBe(0);
|
||||
expect(settled).toBe(false);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await stopPromise;
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await stopPromise;
|
||||
|
||||
expect(ws.terminateCalls).toBe(1);
|
||||
expect(settled).toBe(true);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(ws.terminateCalls).toBe(1);
|
||||
expect(settled).toBe(true);
|
||||
});
|
||||
|
||||
it("does not clear persisted device auth when explicit shared token is provided", () => {
|
||||
@@ -1347,6 +1334,8 @@ describe("GatewayClient connect auth payload", () => {
|
||||
expect(logDebugMock).toHaveBeenCalledWith(
|
||||
"gateway client hello-ok handler error: Error: hello callback failed",
|
||||
);
|
||||
ws.emitClose(1012, "service restart");
|
||||
expect(onConnectError).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
client.stop();
|
||||
}
|
||||
@@ -1536,6 +1525,23 @@ describe("GatewayClient connect auth payload", () => {
|
||||
client.stop();
|
||||
});
|
||||
|
||||
it("reports a transport close while the connect request is pending", () => {
|
||||
const onConnectError = vi.fn();
|
||||
const client = new GatewayClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
token: "shared-token",
|
||||
onConnectError,
|
||||
});
|
||||
|
||||
const { ws } = startClientAndConnect({ client });
|
||||
ws.emitClose(1006, "socket lost");
|
||||
|
||||
expect(firstMockArg(onConnectError, "connect error")).toMatchObject({
|
||||
message: "gateway closed (1006): socket lost",
|
||||
});
|
||||
client.stop();
|
||||
});
|
||||
|
||||
it("logs stopped connect handshakes at debug level during teardown", async () => {
|
||||
const onConnectError = vi.fn();
|
||||
const client = new GatewayClient({
|
||||
|
||||
@@ -193,6 +193,10 @@ export const sharedVitestConfig = {
|
||||
find: "@openclaw/whatsapp/api.js",
|
||||
replacement: path.join(repoRoot, "extensions", "whatsapp", "api.ts"),
|
||||
},
|
||||
{
|
||||
find: "@openclaw/gateway-client/browser",
|
||||
replacement: path.join(repoRoot, "packages", "gateway-client", "src", "browser.ts"),
|
||||
},
|
||||
{
|
||||
find: "@openclaw/gateway-client/readiness",
|
||||
replacement: path.join(repoRoot, "packages", "gateway-client", "src", "readiness.ts"),
|
||||
|
||||
@@ -66,6 +66,7 @@
|
||||
],
|
||||
"@openclaw/model-catalog-core/*": ["./packages/model-catalog-core/src/*"],
|
||||
"@openclaw/gateway-client": ["./packages/gateway-client/src/index.ts"],
|
||||
"@openclaw/gateway-client/browser": ["./packages/gateway-client/src/browser.ts"],
|
||||
"@openclaw/gateway-client/*": ["./packages/gateway-client/src/*"],
|
||||
"@openclaw/gateway-protocol": ["./packages/gateway-protocol/src/index.ts"],
|
||||
"@openclaw/gateway-protocol/client-info": [
|
||||
|
||||
+3
-5
@@ -383,6 +383,7 @@ function buildGatewayClientDistEntries(): Record<string, string> {
|
||||
// Keep package entrypoints explicit so package.json exports and root build
|
||||
// config cannot drift when client internals are split again.
|
||||
index: "packages/gateway-client/src/index.ts",
|
||||
browser: "packages/gateway-client/src/browser.ts",
|
||||
readiness: "packages/gateway-client/src/readiness.ts",
|
||||
timeouts: "packages/gateway-client/src/timeouts.ts",
|
||||
};
|
||||
@@ -566,11 +567,8 @@ function shouldExternalizeGatewayProtocolDependency(id: string): boolean {
|
||||
}
|
||||
|
||||
function shouldExternalizeGatewayClientDependency(id: string): boolean {
|
||||
return (
|
||||
id === "ws" ||
|
||||
id.startsWith("ws/") ||
|
||||
id === "@openclaw/gateway-protocol" ||
|
||||
id.startsWith("@openclaw/gateway-protocol/")
|
||||
return ["ws", "@openclaw/net-policy", "@openclaw/gateway-protocol"].some(
|
||||
(dependency) => id === dependency || id.startsWith(`${dependency}/`),
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
"@modelcontextprotocol/ext-apps": "1.7.4",
|
||||
"@modelcontextprotocol/sdk": "1.29.0",
|
||||
"@noble/ed25519": "3.1.0",
|
||||
"@openclaw/gateway-client": "workspace:*",
|
||||
"@openclaw/libterminal": "0.3.1",
|
||||
"@openclaw/media-core": "workspace:*",
|
||||
"@openclaw/normalization-core": "workspace:*",
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
// @vitest-environment node
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { GATEWAY_CLIENT_CAPS } from "../../../packages/gateway-protocol/src/client-info.js";
|
||||
import { ConnectErrorDetailCodes } from "../../../packages/gateway-protocol/src/connect-error-details.js";
|
||||
/** @vitest-environment node */
|
||||
import {
|
||||
ConnectErrorDetailCodes,
|
||||
GATEWAY_CLIENT_CAPS,
|
||||
MIN_CLIENT_PROTOCOL_VERSION,
|
||||
PROTOCOL_VERSION,
|
||||
} from "../../../packages/gateway-protocol/src/version.js";
|
||||
} from "@openclaw/gateway-client/browser";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { DeviceIdentity } from "../lib/nodes/index.ts";
|
||||
import {
|
||||
loadDeviceAuthToken as loadScopedDeviceAuthToken,
|
||||
@@ -646,6 +646,9 @@ describe("GatewayBrowserClient", () => {
|
||||
|
||||
it("reports connect phase timing without credentials or nonce values", async () => {
|
||||
const onConnectTiming = vi.fn();
|
||||
vi.stubGlobal("performance", {
|
||||
now: vi.fn().mockReturnValueOnce(10).mockReturnValueOnce(35).mockReturnValue(40),
|
||||
});
|
||||
const client = new GatewayBrowserClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
token: "shared-auth-token",
|
||||
@@ -661,6 +664,7 @@ describe("GatewayBrowserClient", () => {
|
||||
"connect-plan-ready",
|
||||
"request-sent",
|
||||
]);
|
||||
expect([sentPayloads[0]?.durationMs, sentPayloads[0]?.phaseDurationMs]).toEqual([25, 25]);
|
||||
for (const payload of sentPayloads) {
|
||||
expect(payload.generation).toBe(1);
|
||||
expect(payload.durationMs).toBeTypeOf("number");
|
||||
@@ -1566,11 +1570,13 @@ describe("GatewayBrowserClient", () => {
|
||||
it("reports willRetry=false on credential rejections so the UI can fall back to the login gate", async () => {
|
||||
useNodeFakeTimers();
|
||||
const onClose = vi.fn();
|
||||
const onConnectTiming = vi.fn();
|
||||
|
||||
const client = new GatewayBrowserClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
password: "wrong-password",
|
||||
onClose,
|
||||
onConnectTiming,
|
||||
});
|
||||
|
||||
const { ws, connectFrame } = await startConnect(client);
|
||||
@@ -1589,6 +1595,12 @@ describe("GatewayBrowserClient", () => {
|
||||
|
||||
const close = requireFirstMockArg(onClose, "close");
|
||||
expect(close.willRetry).toBe(false);
|
||||
expect(connectTimingPayloads(onConnectTiming).at(-1)).toMatchObject({
|
||||
phase: "failed",
|
||||
errorCode: "INVALID_REQUEST",
|
||||
hasDeviceIdentity: true,
|
||||
hasPassword: true,
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(wsInstances).toHaveLength(1);
|
||||
|
||||
|
||||
+189
-661
File diff suppressed because it is too large
Load Diff
@@ -14,6 +14,18 @@ import {
|
||||
const here = path.dirname(fileURLToPath(import.meta.url));
|
||||
const repoRoot = path.resolve(here, "..");
|
||||
const workspaceSourceAliases = [
|
||||
{
|
||||
find: "@openclaw/gateway-client/browser",
|
||||
replacement: path.resolve(repoRoot, "packages/gateway-client/src/browser.ts"),
|
||||
},
|
||||
{
|
||||
find: /^@openclaw\/gateway-protocol\/(.+)$/u,
|
||||
replacement: path.resolve(repoRoot, "packages/gateway-protocol/src/$1.ts"),
|
||||
},
|
||||
{
|
||||
find: /^@openclaw\/(gateway-protocol|retry)$/u,
|
||||
replacement: path.resolve(repoRoot, "packages/$1/src/index.ts"),
|
||||
},
|
||||
{
|
||||
find: "../logging/redact.js",
|
||||
replacement: path.resolve(here, "src/lib/browser-redact.ts"),
|
||||
|
||||
Reference in New Issue
Block a user