vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-10-03 09:39:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 30

fix(gateway): await and fence device token storage (#150349)

Browse source 
* refactor(storage): extract device token SQL kernels

* fix(gateway): await and fence device token storage
This commit is contained in:
Peter Steinberger 2026-09-17 01:09:10 -07:00 • committed by GitHub
parent 798c2974bd
commit 74d61d4855
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 1337 additions and 239 deletions
Download patch file Download diff file Expand all files Collapse all files

2
config/assertion-safety-baseline.txt
View file

@ -1461,7 +1461,7 @@ packages/ai/src/utils/credential-redaction.ts 1
packages/ai/src/utils/oauth/openai-chatgpt-jwt.ts 1
packages/ai/src/utils/stream-first-event-timeout.ts 2
packages/ai/src/utils/tls-certificate-errors.ts 1
packages/gateway-client/src/client.ts 5
packages/gateway-client/src/client.ts 4
packages/gateway-client/src/gateway-origin-scope.ts 1
packages/gateway-client/src/pending-request.ts 2
packages/gateway-client/src/protocol-client.ts 1

31
packages/gateway-client/README.md
View file

@ -46,6 +46,37 @@ For device-authenticated Node connections, supply `deviceIdentity` (or
storage through `GatewayClientHostDeps`; the package does not load OpenClaw's
local identity or credentials automatically.
Token storage callbacks may return their existing synchronous result or a
`Promise`. The client waits for token loading before sending `connect`, and for
issued-token persistence before calling `onHelloOk`. An accepted hello creates a
persistence obligation that survives disconnect or stop; bootstrap credentials
retire after that persistence succeeds. Readiness still belongs to the current
connection. `stop()` requests shutdown synchronously; `stopAndWait()` also waits
for accepted token operations to settle, even when transport closure times out.
A reconnect waits for earlier token operations before reading the token again.
Accepted asynchronous persistence failures reach `onConnectError` even after the
connection retires. If that callback is absent or throws, `stopAndWait()` rejects
with the first undelivered persistence error after draining accepted work. A later
connection can still load credentials; a reported failure does not poison its
storage queue. Synchronous callback exceptions retain their existing connect-error
behavior.
Hosts should honor the optional `expectedToken` storage condition. A string
compares the existing row before writing or clearing; `null` on a store means
insert only when no row exists. Omission preserves unconditional storage behavior.
This prevents an older receipt or cleanup from replacing another client's newer
token. Close cleanup waits for pending persistence and, if its result is uncertain,
conditionally clears only the sampled and received tokens. Cleanup before any token
observation retains its existing unconditional behavior. An observed empty cache
does not permit unconditional cleanup.
When storage callbacks receive `signal` or `assertCurrent`, check them immediately
before admission and before committing a write. These callbacks stay local to the
host; do not send them to a worker. Loads use the current connection lifetime;
cleanup uses the client lifetime. Accepted hello persistence is independent of
transport lifetime and retains the host's normal storage admission checks.
## Entry points
- `@openclaw/gateway-client` exports the Node `GatewayClient`, device-auth

17
packages/gateway-client/src/client-address-utils.ts
View file

@ -8,10 +8,25 @@ export function normalizeGatewayErrorText(value: unknown): string {
return typeof value === "string" ? value.trim().toLowerCase() : "";
}
export function isSensitiveUrlQueryParamName(key: string): boolean {
function isSensitiveUrlQueryParamName(key: string): boolean {
return /(?:token|password|secret|key|auth|credential)/iu.test(key);
}
export function isGatewayClientStoppedError(err: unknown): boolean {
const message = err instanceof Error ? err.message : String(err);
return message === "gateway client stopped" || message === "Error: gateway client stopped";
}
export function formatGatewayClientErrorForLog(err: unknown): string {
const redactedUrlLikeString = String(err)
.replace(/\/\/([^@/?#\s]+)@/g, "//***:***@")
.replace(/(Authorization:\s*Bearer\s+)[^\s]+/giu, "$1***")
.replace(/([?&])([^=&\s]+)=([^&#\s"'<>)]*)/g, (match, prefix: string, key: string) =>
isSensitiveUrlQueryParamName(key) ? `${prefix}${key}=***` : match,
);
return redactedUrlLikeString;
}
const SHA256_HEX_FINGERPRINT = /^[a-fA-F0-9]{64}$/u;
const SHA256_COLON_FINGERPRINT = /^(?:[a-fA-F0-9]{2}:){31}[a-fA-F0-9]{2}$/u;

154
packages/gateway-client/src/client-device-auth.ts Normal file
View file

@ -0,0 +1,154 @@
export type DeviceAuthTokenRecord = {
token?: string;
scopes?: string[];
};
export type DeviceAuthTokenObservation = {
// Null records a completed load without a token; undefined means no load completed.
token: string | null | undefined;
receiptToken?: string;
persistence?: Promise<void>;
};
export type MaybePromise<T> = T | Promise<T>;
type DeviceAuthOperation = {
signal?: AbortSignal;
assertCurrent?: () => void;
};
export type GatewayClientDeviceAuthStorage = {
loadDeviceAuthToken?: (
params: DeviceAuthOperation & {
deviceId: string;
role: string;
env?: NodeJS.ProcessEnv;
},
) => MaybePromise<DeviceAuthTokenRecord | null>;
storeDeviceAuthToken?: (
params: DeviceAuthOperation & {
deviceId: string;
role: string;
token: string;
scopes: string[];
env?: NodeJS.ProcessEnv;
expectedToken?: string | null;
},
) => unknown;
clearDeviceAuthToken?: (
params: DeviceAuthOperation & {
deviceId: string;
role: string;
env?: NodeJS.ProcessEnv;
expectedToken?: string;
},
) => unknown;
};
/** Owns pending host storage work for one client across connection generations. */
export class GatewayClientDeviceAuth {
private readonly operations = new Set<Promise<unknown>>();
private unreportedPersistenceFailure: Error | undefined;
constructor(private readonly storage: Required<GatewayClientDeviceAuthStorage>) {}
load(params: Parameters<NonNullable<GatewayClientDeviceAuthStorage["loadDeviceAuthToken"]>>[0]) {
const load = () => {
params.signal?.throwIfAborted();
params.assertCurrent?.();
return this.track(this.storage.loadDeviceAuthToken(params));
};
// A retired connection's accepted store/clear must settle before the next read.
return this.operations.size > 0 ? this.settle().then(load) : load();
}
store(
params: Parameters<NonNullable<GatewayClientDeviceAuthStorage["storeDeviceAuthToken"]>>[0],
onStored: () => void,
onFailed: (error: Error) => boolean,
) {
const failed = (error: unknown): never => {
const failure = error instanceof Error ? error : new Error(String(error));
let reported = false;
try {
reported = onFailed(failure);
} catch {
// A broken reporter must not replace or discard the original storage error.
}
if (!reported) {
this.unreportedPersistenceFailure ??= failure;
}
throw failure;
};
params.signal?.throwIfAborted();
params.assertCurrent?.();
const stored = this.storage.storeDeviceAuthToken(params);
return stored instanceof Promise ? this.track(stored.then(onStored).catch(failed)) : onStored();
}
clear(
params: Parameters<NonNullable<GatewayClientDeviceAuthStorage["clearDeviceAuthToken"]>>[0],
observation: DeviceAuthTokenObservation,
canClear: () => boolean,
): MaybePromise<void> {
const clear = () => {
if (!canClear()) {
return undefined;
}
// A rejected result may follow a committed write. Reconcile only the sampled
// and received token identities; never clear an unrelated replacement.
const knownTokens = [observation.token, observation.receiptToken].filter(
(token): token is string => typeof token === "string",
);
const expectedTokens: Array<string | undefined> = [...new Set(knownTokens)];
if (observation.token === undefined && expectedTokens.length === 0) {
expectedTokens.push(undefined);
}
const clearToken = (expectedToken: string | undefined) => {
params.signal?.throwIfAborted();
params.assertCurrent?.();
const cleared = this.storage.clearDeviceAuthToken({
...params,
...(expectedToken === undefined ? {} : { expectedToken }),
});
return cleared instanceof Promise ? cleared.then(() => {}) : undefined;
};
let pending: Promise<void> | undefined;
for (const expectedToken of expectedTokens) {
pending = pending
? pending.then(() => clearToken(expectedToken))
: clearToken(expectedToken);
}
return pending;
};
// Register the entire cleanup now so reconnect and shutdown also join deferred clearing.
return this.track(
observation.persistence ? observation.persistence.then(clear, clear) : clear(),
);
}
async settle(): Promise<void> {
await Promise.allSettled(this.operations);
}
async drain(): Promise<void> {
await this.settle();
const failure = this.unreportedPersistenceFailure;
this.unreportedPersistenceFailure = undefined;
if (failure) {
throw failure;
}
}
private track<T>(operation: MaybePromise<T>): MaybePromise<T> {
if (!(operation instanceof Promise)) {
return operation;
}
this.operations.add(operation);
void operation.then(
() => this.operations.delete(operation),
() => this.operations.delete(operation),
);
return operation;
}
}

46
packages/gateway-client/src/client-upgrade-error.ts Normal file
View file

@ -0,0 +1,46 @@
import type { IncomingMessage } from "node:http";
const MAX_UPGRADE_ERROR_BODY_BYTES = 2 * 1024;
const UPGRADE_ERROR_BODY_TIMEOUT_MS = 1_000;
export async function readUpgradeErrorBody(response: IncomingMessage): Promise<string> {
return await new Promise<string>((resolve) => {
const chunks: Buffer[] = [];
let totalBytes = 0;
let settled = false;
const finish = () => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
response.off("data", onData);
response.off("end", finish);
response.off("error", finish);
response.off("aborted", finish);
resolve(Buffer.concat(chunks, totalBytes).toString("utf8").replace(/\s+/gu, " ").trim());
};
const stop = () => {
finish();
response.destroy();
};
const onData = (chunk: Buffer | string) => {
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
const remaining = MAX_UPGRADE_ERROR_BODY_BYTES - totalBytes;
if (remaining > 0) {
const prefix = buffer.subarray(0, remaining);
chunks.push(prefix);
totalBytes += prefix.byteLength;
}
if (buffer.byteLength >= remaining) {
stop();
}
};
const timer = setTimeout(stop, UPGRADE_ERROR_BODY_TIMEOUT_MS);
timer.unref?.();
response.on("data", onData);
response.once("end", finish);
response.once("error", finish);
response.once("aborted", finish);
});
}

501
packages/gateway-client/src/client.device-auth.test.ts Normal file
View file

@ -0,0 +1,501 @@
import assert from "node:assert/strict";
import { EventEmitter } from "node:events";
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { createDeferred } from "../../../test/helpers/promise.js";
import type {
DeviceAuthTokenRecord,
GatewayClient as GatewayClientInstance,
GatewayClientHostDeps,
GatewayClientOptions,
} from "./client.js";
class MockWebSocket extends EventEmitter {
static readonly OPEN = 1;
static instances: MockWebSocket[] = [];
readyState = 0;
readonly send = vi.fn<(data: string) => void>();
constructor() {
super();
MockWebSocket.instances.push(this);
}
open() {
this.readyState = MockWebSocket.OPEN;
this.emit("open");
this.receive({
type: "event",
event: "connect.challenge",
payload: { nonce: "synthetic-nonce", ts: 1_800_000_000_000 },
});
}
receive(frame: unknown) {
this.emit("message", JSON.stringify(frame));
}
respond(payload: unknown, error?: unknown) {
const sent = this.send.mock.calls[0];
assert(sent);
const { id } = JSON.parse(sent[0]) as { id: string };
this.receive({ type: "res", id, ok: !error, payload, error });
}
close(code = 1000, reason = "") {
this.readyState = 3;
this.emit("close", code, Buffer.from(reason));
}
terminate() {
this.close();
}
}
vi.mock("./websocket.js", () => ({ WebSocket: MockWebSocket }));
let GatewayClient: typeof GatewayClientInstance;
const clients: GatewayClientInstance[] = [];
beforeAll(async () => {
({ GatewayClient } = await import("./client.js"));
});
beforeEach(() => {
vi.useFakeTimers();
MockWebSocket.instances = [];
});
afterEach(async () => {
for (const client of clients.splice(0)) {
await client.stopAndWait();
}
vi.useRealTimers();
});
function connect(
hostDeps: GatewayClientHostDeps,
options: Pick<
GatewayClientOptions,
"preferBootstrapToken" | "bootstrapToken" | "token" | "password"
> = {},
reportError?: ((error: Error) => void) | false,
) {
const onHelloOk = vi.fn();
const onConnectError = vi.fn((error: Error) => {
if (reportError) {
reportError(error);
}
});
const onClose = vi.fn();
const onReconnectPaused = vi.fn();
const client = new GatewayClient({
...options,
url: "ws://127.0.0.1:18789",
deviceIdentity: {
deviceId: "synthetic-device",
privateKeyPem: "synthetic-private-key",
publicKeyPem: "synthetic-public-key",
},
hostDeps: {
signDevicePayload: () => "synthetic-signature",
publicKeyRawBase64UrlFromPem: () => "synthetic-public-key",
...hostDeps,
},
onHelloOk,
onConnectError: reportError === false ? undefined : onConnectError,
onClose,
onReconnectPaused,
});
clients.push(client);
client.start();
const socket = MockWebSocket.instances[0];
assert(socket);
socket.open();
return { client, socket, onHelloOk, onConnectError, onClose, onReconnectPaused };
}
const storedToken = { token: "synthetic-stored-token", scopes: ["operator.read"] };
const hello = {
type: "hello-ok",
protocol: 4,
auth: { deviceToken: "synthetic-issued-token", role: "operator", scopes: ["operator.read"] },
};
describe("GatewayClient host token storage", () => {
it("awaits a stored token before sending its connect request", async () => {
const loaded = createDeferred<DeviceAuthTokenRecord | null>();
const { socket } = connect({ loadDeviceAuthToken: () => loaded.promise });
expect(socket.send).not.toHaveBeenCalled();
loaded.resolve(storedToken);
await vi.advanceTimersByTimeAsync(0);
const sent = socket.send.mock.calls[0];
assert(sent);
expect(JSON.parse(sent[0])).toMatchObject({
method: "connect",
params: { auth: { deviceToken: storedToken.token }, scopes: storedToken.scopes },
});
});
it("keeps synchronous storage callbacks and their ignored return values compatible", async () => {
const { socket, onHelloOk } = connect({
loadDeviceAuthToken: () => storedToken,
storeDeviceAuthToken: () => storedToken,
clearDeviceAuthToken: () => true,
});
expect(socket.send).toHaveBeenCalledOnce();
socket.respond(hello);
await vi.advanceTimersByTimeAsync(0);
expect(onHelloOk).toHaveBeenCalledOnce();
});
it.each(["ready", "closing", "stop"] as const)(
"settles token persistence before completing %s",
async (completion) => {
const stored = createDeferred();
let persistedToken: string | undefined;
const storeDeviceAuthToken = vi.fn<
NonNullable<GatewayClientHostDeps["storeDeviceAuthToken"]>
>(async (params) => {
await stored.promise;
params.signal?.throwIfAborted();
params.assertCurrent?.();
persistedToken = params.token;
});
const { client, socket, onHelloOk, onConnectError } = connect({
loadDeviceAuthToken: () => storedToken,
storeDeviceAuthToken,
});
socket.respond(hello);
await vi.advanceTimersByTimeAsync(0);
expect(onHelloOk).not.toHaveBeenCalled();
expect(storeDeviceAuthToken).toHaveBeenCalledOnce();
const operation = storeDeviceAuthToken.mock.calls[0]?.[0];
assert(operation);
expect(operation).toMatchObject({ token: hello.auth.deviceToken, scopes: hello.auth.scopes });
expect(() => operation.assertCurrent?.()).not.toThrow();
if (completion === "closing") {
socket.readyState = 2;
}
const stopped = vi.fn();
const stop = completion === "stop" ? client.stopAndWait().then(stopped) : undefined;
if (completion === "stop") {
expect(operation.signal).toBeUndefined();
expect(operation.assertCurrent).toBeUndefined();
await vi.advanceTimersByTimeAsync(0);
expect(stopped).not.toHaveBeenCalled();
}
stored.resolve();
await vi.advanceTimersByTimeAsync(0);
await stop;
expect(persistedToken).toBe(hello.auth.deviceToken);
expect(onConnectError).not.toHaveBeenCalled();
expect(onHelloOk).toHaveBeenCalledTimes(completion === "ready" ? 1 : 0);
},
);
it("reconnects with the durable receipt after bootstrap hello's transport retires", async () => {
const permitStore = createDeferred();
let cached: DeviceAuthTokenRecord | null = null;
const { socket, onHelloOk } = connect(
{
loadDeviceAuthToken: () => cached,
storeDeviceAuthToken: async (params) => {
await permitStore.promise;
params.signal?.throwIfAborted();
params.assertCurrent?.();
cached = { token: params.token, scopes: params.scopes };
},
},
{ bootstrapToken: "synthetic-bootstrap", preferBootstrapToken: true },
);
try {
socket.respond(hello);
await vi.advanceTimersByTimeAsync(0);
socket.close(1006, "transport retired");
await vi.advanceTimersByTimeAsync(1_000);
const replacement = MockWebSocket.instances[1];
assert(replacement);
replacement.open();
expect(replacement.send).not.toHaveBeenCalled();
permitStore.resolve();
await vi.advanceTimersByTimeAsync(0);
const sent = replacement.send.mock.calls[0];
assert(sent);
expect(JSON.parse(sent[0])).toMatchObject({
method: "connect",
params: { auth: { deviceToken: hello.auth.deviceToken } },
});
expect(JSON.parse(sent[0]).params.auth.bootstrapToken).toBeUndefined();
expect(onHelloOk).not.toHaveBeenCalled();
} finally {
permitStore.resolve();
}
});
it.each([false, true])(
"keeps a newer token after observing an empty cache (async load: %s)",
async (asyncLoad) => {
let cached: DeviceAuthTokenRecord | null = null;
const newer = { token: "synthetic-newer-token", scopes: ["operator.read"] };
const clearDeviceAuthToken = vi.fn<
NonNullable<GatewayClientHostDeps["clearDeviceAuthToken"]>
>((params) => {
if (params.expectedToken === undefined || params.expectedToken === cached?.token) {
cached = null;
}
});
const { client, socket } = connect({
loadDeviceAuthToken: () => (asyncLoad ? Promise.resolve(cached) : cached),
clearDeviceAuthToken,
});
await vi.advanceTimersByTimeAsync(0);
expect(socket.send).toHaveBeenCalledOnce();
cached = newer;
socket.close(1008, "device token mismatch");
await client.stopAndWait();
expect(clearDeviceAuthToken).not.toHaveBeenCalled();
expect(cached).toEqual(newer);
},
);
it.each([
{ initial: storedToken, result: "committed", replaceWithNewer: false, bootstrap: true },
{ initial: storedToken, result: "committed", replaceWithNewer: true },
{ initial: null, result: "committed", replaceWithNewer: false },
{ initial: null, result: "committed", replaceWithNewer: true },
{ initial: storedToken, result: "uncertain", replaceWithNewer: false },
{ initial: null, result: "uncertain", replaceWithNewer: false },
{ initial: null, result: "failed", replaceWithNewer: true },
] as const)("reconciles mismatch cleanup against the accepted receipt: %j", async (entry) => {
const { initial, result, replaceWithNewer } = entry;
const permitStore = createDeferred();
let cached: DeviceAuthTokenRecord | null = initial;
const newer = { token: "synthetic-newer-token", scopes: ["operator.read"] };
const clearDeviceAuthToken = vi.fn<NonNullable<GatewayClientHostDeps["clearDeviceAuthToken"]>>(
(params) => {
params.assertCurrent?.();
if (params.expectedToken === undefined || params.expectedToken === cached?.token) {
cached = null;
}
},
);
const { socket, onHelloOk } = connect(
{
loadDeviceAuthToken: () => cached,
storeDeviceAuthToken: async (params) => {
await permitStore.promise;
params.signal?.throwIfAborted();
params.assertCurrent?.();
if (result === "failed") {
throw new Error("synthetic persistence rejected");
}
if (
params.expectedToken === undefined ||
(params.expectedToken === null
? cached === null
: params.expectedToken === cached?.token)
) {
cached = { token: params.token, scopes: params.scopes };
}
if (result === "uncertain") {
throw new Error("synthetic result unavailable");
}
},
clearDeviceAuthToken,
},
"bootstrap" in entry
? {
preferBootstrapToken: true,
bootstrapToken: "synthetic-bootstrap",
token: "synthetic-shared-token",
password: "synthetic-shared-password",
}
: {},
);
try {
socket.respond(hello);
await vi.advanceTimersByTimeAsync(0);
socket.close(1008, "device token mismatch");
expect(clearDeviceAuthToken).not.toHaveBeenCalled();
if (replaceWithNewer) {
cached = newer;
}
permitStore.resolve();
await vi.advanceTimersByTimeAsync(0);
expect(clearDeviceAuthToken).toHaveBeenCalled();
expect(
clearDeviceAuthToken.mock.calls.every(([params]) => params.expectedToken !== undefined),
).toBe(true);
expect(cached).toEqual(replaceWithNewer ? newer : null);
expect(onHelloOk).not.toHaveBeenCalled();
} finally {
permitStore.resolve();
}
});
it.each(["active", "disconnected", "stopped"] as const)(
"reports an accepted persistence rejection exactly once when %s",
async (lifetime) => {
const permitStore = createDeferred();
const failure = new Error("synthetic async persistence failure");
const { client, socket, onConnectError } = connect({
loadDeviceAuthToken: () => storedToken,
storeDeviceAuthToken: async () => {
await permitStore.promise;
throw failure;
},
});
try {
socket.respond(hello);
await vi.advanceTimersByTimeAsync(0);
if (lifetime === "disconnected") {
socket.close(1006, "transport retired");
}
const stopped = lifetime === "stopped" ? client.stopAndWait() : undefined;
permitStore.resolve();
await vi.advanceTimersByTimeAsync(0);
await stopped;
expect(onConnectError).toHaveBeenCalledExactlyOnceWith(failure);
} finally {
permitStore.resolve();
}
},
);
it.each(["missing", "throwing"] as const)(
"drains the original undelivered async persistence error with a %s reporter",
async (reporter) => {
const permitStore = createDeferred();
const failure = new Error("synthetic undelivered persistence failure");
const { client, socket } = connect(
{
loadDeviceAuthToken: () => storedToken,
storeDeviceAuthToken: async () => {
await permitStore.promise;
throw failure;
},
},
{},
reporter === "missing"
? false
: () => {
throw new Error("synthetic reporter failure");
},
);
try {
socket.respond(hello);
await vi.advanceTimersByTimeAsync(0);
if (reporter === "throwing") {
const stopped = expect(client.stopAndWait()).rejects.toBe(failure);
permitStore.resolve();
await vi.advanceTimersByTimeAsync(0);
await stopped;
} else {
socket.close(1006, "transport retired");
permitStore.resolve();
await vi.advanceTimersByTimeAsync(1_000);
const replacement = MockWebSocket.instances[1];
assert(replacement);
replacement.open();
expect(replacement.send).toHaveBeenCalledOnce();
await expect(client.stopAndWait()).rejects.toBe(failure);
}
} finally {
permitStore.resolve();
}
},
);
it("keeps synchronous store exceptions on the existing connect-error path", async () => {
const { client, socket, onConnectError } = connect({
loadDeviceAuthToken: () => storedToken,
storeDeviceAuthToken: () => {
throw new Error("synthetic synchronous persistence failure");
},
});
socket.respond(hello);
await vi.advanceTimersByTimeAsync(0);
expect(onConnectError).toHaveBeenCalledOnce();
await expect(client.stopAndWait()).resolves.toBeUndefined();
});
it("retires pending token loading when stopped", async () => {
const loaded = createDeferred<DeviceAuthTokenRecord | null>();
const { client, socket, onHelloOk } = connect({ loadDeviceAuthToken: () => loaded.promise });
const stopped = client.stopAndWait();
loaded.resolve(storedToken);
await stopped;
await vi.advanceTimersByTimeAsync(0);
expect(socket.send).not.toHaveBeenCalled();
expect(onHelloOk).not.toHaveBeenCalled();
});
it.each([false, true])(
"preserves rejection and cleanup when the peer closes (already closing: %s)",
async (alreadyClosing) => {
const cleared = createDeferred();
const clearDeviceAuthToken = vi.fn<
NonNullable<GatewayClientHostDeps["clearDeviceAuthToken"]>
>(() => cleared.promise);
const { socket, onConnectError, onClose, onReconnectPaused } = connect({
loadDeviceAuthToken: () => storedToken,
clearDeviceAuthToken,
});
try {
socket.respond(undefined, {
code: "INVALID_REQUEST",
message: "synthetic token rejected",
details: { code: "AUTH_DEVICE_TOKEN_MISMATCH" },
});
if (alreadyClosing) {
socket.readyState = 2;
}
await vi.advanceTimersByTimeAsync(0);
expect(clearDeviceAuthToken).toHaveBeenCalledWith(
expect.objectContaining({ expectedToken: storedToken.token }),
);
expect(onConnectError).toHaveBeenCalledOnce();
const error = onConnectError.mock.calls[0]?.[0];
expect(error).toMatchObject({ message: "synthetic token rejected" });
socket.close(1008, "connect failed");
const cleanup = clearDeviceAuthToken.mock.calls[0]?.[0];
assert(cleanup);
expect(() => cleanup.assertCurrent?.()).not.toThrow();
expect(onClose).toHaveBeenCalledWith(
1008,
"connect failed",
expect.objectContaining({ connectError: error }),
);
expect(onReconnectPaused).toHaveBeenCalledWith(
expect.objectContaining({ detailCode: "AUTH_DEVICE_TOKEN_MISMATCH" }),
);
cleared.resolve();
await vi.advanceTimersByTimeAsync(1_000);
expect(onConnectError).toHaveBeenCalledOnce();
expect(MockWebSocket.instances).toHaveLength(1);
} finally {
cleared.resolve();
}
},
);
it("finishes close cleanup before loading credentials for the replacement connection", async () => {
const cleared = createDeferred();
const loadDeviceAuthToken = vi.fn(() => storedToken);
const clearDeviceAuthToken = vi.fn<NonNullable<GatewayClientHostDeps["clearDeviceAuthToken"]>>(
() => cleared.promise,
);
const { socket } = connect({ loadDeviceAuthToken, clearDeviceAuthToken });
socket.close(1008, "device token mismatch");
const cleanup = clearDeviceAuthToken.mock.calls[0]?.[0];
assert(cleanup);
expect(cleanup.expectedToken).toBe(storedToken.token);
expect(() => cleanup.assertCurrent?.()).not.toThrow();
await vi.advanceTimersByTimeAsync(1_000);
const replacement = MockWebSocket.instances[1];
assert(replacement);
replacement.open();
expect(loadDeviceAuthToken).toHaveBeenCalledOnce();
cleared.resolve();
await vi.advanceTimersByTimeAsync(0);
expect(loadDeviceAuthToken).toHaveBeenCalledTimes(2);
expect(replacement.send).toHaveBeenCalledOnce();
});
});

370
packages/gateway-client/src/client.ts
View file

@ -21,10 +21,19 @@ import {
import { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensitive-url";
import { isRecord } from "@openclaw/normalization-core/record-coerce";
import {
isSensitiveUrlQueryParamName,
formatGatewayClientErrorForLog,
isGatewayClientStoppedError,
normalizeTlsFingerprint,
normalizeGatewayErrorText,
} from "./client-address-utils.js";
import {
GatewayClientDeviceAuth,
type DeviceAuthTokenRecord,
type DeviceAuthTokenObservation,
type GatewayClientDeviceAuthStorage,
type MaybePromise,
} from "./client-device-auth.js";
import { readUpgradeErrorBody } from "./client-upgrade-error.js";
import {
buildGatewayConnectAuth,
type GatewayConnectAuthSelection,
@ -34,6 +43,7 @@ import {
} from "./connect-auth.js";
import { buildDeviceAuthPayloadV3 } from "./device-auth.js";
import { resolveModelCatalogConnect } from "./model-catalog-connect.js";
import type { GatewayProtocolConnectAuthority } from "./protocol-client-contract.js";
import {
GatewayProtocolClient,
type GatewayProtocolCloseContext,
@ -46,7 +56,7 @@ import {
GatewayProtocolRequestTimeoutError,
} from "./protocol-request.js";
import { shouldPauseGatewayReconnect } from "./reconnect-policy.js";
import { GatewayClientRequestError } from "./request-error.js";
import { GatewayClientRequestError, markGatewayConnectAssemblyError } from "./request-error.js";
import {
DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS,
resolveConnectChallengeTimeoutMs,
@ -67,34 +77,14 @@ export type DeviceIdentity = {
publicKeyPem: string;
};
export type DeviceAuthTokenRecord = {
token?: string;
scopes?: string[];
};
export type { DeviceAuthTokenRecord } from "./client-device-auth.js";
// The package stays reusable by depending on host callbacks for OpenClaw-owned
// state: device keys, token storage, proxy routing, logging, and TLS formatting.
export type GatewayClientHostDeps = {
export type GatewayClientHostDeps = GatewayClientDeviceAuthStorage & {
loadOrCreateDeviceIdentity?: () => DeviceIdentity | undefined;
signDevicePayload?: (privateKeyPem: string, payload: string) => string;
publicKeyRawBase64UrlFromPem?: (publicKeyPem: string) => string;
loadDeviceAuthToken?: (params: {
deviceId: string;
role: string;
env?: NodeJS.ProcessEnv;
}) => DeviceAuthTokenRecord | null;
storeDeviceAuthToken?: (params: {
deviceId: string;
role: string;
token: string;
scopes: string[];
env?: NodeJS.ProcessEnv;
}) => void;
clearDeviceAuthToken?: (params: {
deviceId: string;
role: string;
env?: NodeJS.ProcessEnv;
}) => void;
beforeConnect?: () => void;
registerGatewayLoopbackBypass?: (url: string) => (() => void) | undefined;
logDebug?: (message: string) => void;
@ -141,55 +131,11 @@ type AssembledConnect = {
storedScopes: string[] | undefined;
storedToken: string | undefined;
usingStoredDeviceToken: boolean | undefined;
persistenceFailed?: boolean;
};
const DEFAULT_GATEWAY_CLIENT_URL = "ws://127.0.0.1:18789";
const DEFAULT_CLIENT_VERSION = "0.0.0";
const MAX_UPGRADE_ERROR_BODY_BYTES = 2 * 1024;
const UPGRADE_ERROR_BODY_TIMEOUT_MS = 1_000;
async function readUpgradeErrorBody(response: IncomingMessage): Promise<string> {
return await new Promise<string>((resolve) => {
const chunks: Buffer[] = [];
let totalBytes = 0;
let settled = false;
const finish = () => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
response.off("data", onData);
response.off("end", finish);
response.off("error", finish);
response.off("aborted", finish);
resolve(Buffer.concat(chunks, totalBytes).toString("utf8").replace(/\s+/gu, " ").trim());
};
const stop = () => {
finish();
response.destroy();
};
const onData = (chunk: Buffer | string) => {
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
const remaining = MAX_UPGRADE_ERROR_BODY_BYTES - totalBytes;
if (remaining > 0) {
const prefix = buffer.subarray(0, remaining);
chunks.push(prefix);
totalBytes += prefix.byteLength;
}
if (buffer.byteLength >= remaining) {
stop();
}
};
const timer = setTimeout(stop, UPGRADE_ERROR_BODY_TIMEOUT_MS);
timer.unref?.();
response.on("data", onData);
response.once("end", finish);
response.once("error", finish);
response.once("aborted", finish);
});
}
export type GatewayReconnectPausedInfo = {
code: number;
reason: string;
@ -205,7 +151,7 @@ export type GatewayClientCloseInfo = {
connectError?: Error;
};
export { GatewayClientRequestError } from "./request-error.js";
export { GatewayClientRequestError, isGatewayConnectAssemblyError } from "./request-error.js";
export { isGatewayProtocolResponseError } from "./protocol-request.js";
export class GatewayClientRequestTimeoutError extends GatewayProtocolRequestTimeoutError {
@ -217,27 +163,6 @@ export class GatewayClientRequestTimeoutError extends GatewayProtocolRequestTime
class GatewayClientTransportPolicyError extends GatewayWebSocketTransportConfigurationError {}
const GATEWAY_CONNECT_ASSEMBLY_ERROR = Symbol("gateway.connectAssemblyError");
type GatewayConnectAssemblyError = Error & {
[GATEWAY_CONNECT_ASSEMBLY_ERROR]?: true;
};
function markGatewayConnectAssemblyError(error: Error): Error {
Object.defineProperty(error, GATEWAY_CONNECT_ASSEMBLY_ERROR, {
configurable: true,
value: true,
});
return error;
}
export function isGatewayConnectAssemblyError(value: unknown): value is Error {
return (
value instanceof Error &&
(value as GatewayConnectAssemblyError)[GATEWAY_CONNECT_ASSEMBLY_ERROR] === true
);
}
export type GatewayClientOptions = {
url?: string; // ws://127.0.0.1:18789
origin?: string;
@ -302,21 +227,6 @@ export type GatewayClientConnectionMetadata = {
preauthHandshakeTimeoutMs?: number;
};
function isGatewayClientStoppedError(err: unknown): boolean {
const message = err instanceof Error ? err.message : String(err);
return message === "gateway client stopped" || message === "Error: gateway client stopped";
}
function formatGatewayClientErrorForLog(err: unknown): string {
const redactedUrlLikeString = String(err)
.replace(/\/\/([^@/?#\s]+)@/g, "//***:***@")
.replace(/(Authorization:\s*Bearer\s+)[^\s]+/giu, "$1***")
.replace(/([?&])([^=&\s]+)=([^&#\s"'<>)]*)/g, (match, prefix: string, key: string) =>
isSensitiveUrlQueryParamName(key) ? `${prefix}${key}=***` : match,
);
return redactedUrlLikeString;
}
const FORCE_STOP_TERMINATE_GRACE_MS = 250;
const STOP_AND_WAIT_TIMEOUT_MS = 1_000;
const MAX_SUPPRESSED_TRANSIENT_PRE_HELLO_CLEAN_CLOSES = 1;
@ -347,7 +257,6 @@ export class GatewayClient {
private stopped = false;
private useLegacyNodeProtocolEnvelope = false;
private nodeProtocolTransitionPending = false;
private suppressNextHelloCallback = false;
private pendingDeviceTokenRetry = false;
private deviceTokenRetryBudgetUsed = false;
private approvalRuntimeTokenCompatibilityDisabled = false;
@ -358,12 +267,15 @@ export class GatewayClient {
private tickTimer: NodeJS.Timeout | null = null;
private readonly requestTimeoutMs: number;
private pendingStop: PendingStop | null = null;
private readonly deviceAuth: GatewayClientDeviceAuth;
private connectionStoredToken?: DeviceAuthTokenObservation & { generation: number };
private transportValidated = false;
private suppressedTransientPreHelloCleanCloses = 0;
constructor(opts: GatewayClientOptions) {
// Defaults keep the package inert until device identity support is used.
this.deps = resolveHostDeps(opts.hostDeps);
this.deviceAuth = new GatewayClientDeviceAuth(this.deps);
this.opts = {
...opts,
deviceIdentity:
@ -389,19 +301,35 @@ export class GatewayClient {
createRequestTimeoutError: (method, timeoutMs, requestSent) =>
new GatewayClientRequestTimeoutError({ method, timeoutMs, requestSent }),
createRequestAbortError: createGatewayRequestAbortError,
buildConnectPlan: ({ nonce, challengeTs, serverCapabilities }) => {
buildConnectPlan: ({ nonce, challengeTs, serverCapabilities, generation, ...authority }) => {
if (!nonce) {
throw new Error("gateway connect challenge missing nonce");
}
if (this.opts.deviceIdentity && challengeTs == null) {
throw new Error("gateway connect challenge timestamp invalid");
}
return this.assembleConnectParams({
role: this.opts.role ?? "operator",
nonce,
signedAtMs: challengeTs ?? Date.now(),
serverCapabilities,
});
const role = this.opts.role ?? "operator";
const assemble = (storedAuth: DeviceAuthTokenRecord | null) => {
authority.assertCurrent();
const assembled = this.assembleConnectParams({
role,
nonce,
signedAtMs: challengeTs ?? Date.now(),
serverCapabilities,
storedAuth,
});
this.connectionStoredToken = { generation, token: assembled.storedToken ?? null };
return assembled;
};
const storedAuth = this.opts.deviceIdentity
? this.deviceAuth.load({
deviceId: this.opts.deviceIdentity.deviceId,
role,
env: this.opts.env,
...authority,
})
: null;
return storedAuth instanceof Promise ? storedAuth.then(assemble) : assemble(storedAuth);
},
buildConnectParams: (assembled) => assembled.params,
onConnectPlanError: (error) => {
@ -415,12 +343,8 @@ export class GatewayClient {
}
return { closeCode: 1008, closeReason: "connect failed", stop: true, error: marked };
},
onConnectHello: (hello, context) => this.handleConnectHello(hello, context.plan),
onConnectHello: (hello, context) => this.handleConnectHello(hello, context.plan, context),
onHello: (hello) => {
if (this.suppressNextHelloCallback) {
this.suppressNextHelloCallback = false;
return;
}
this.opts.onHelloOk?.(hello);
},
onConnectFailure: (error, context) => this.handleConnectRequestFailure(error, context.plan),
@ -648,12 +572,8 @@ export class GatewayClient {
}
async stopAndWait(opts?: { timeoutMs?: number }): Promise<void> {
// Some callers need teardown ordering, not just "close requested". Wait for
// the socket to close or the terminate fallback to fire.
// Teardown includes accepted storage work as well as transport closure.
const stopPromise = this.beginStop();
if (!stopPromise) {
return;
}
const timeoutMs =
opts?.timeoutMs === undefined
? STOP_AND_WAIT_TIMEOUT_MS
@ -673,6 +593,8 @@ export class GatewayClient {
if (timeout) {
clearTimeout(timeout);
}
// The transport deadline must never abandon accepted durable operations.
await this.deviceAuth.drain();
}
}
@ -749,11 +671,12 @@ export class GatewayClient {
nonce: string;
signedAtMs: number;
serverCapabilities: readonly string[];
storedAuth: DeviceAuthTokenRecord | null;
}): AssembledConnect {
const { role, nonce, signedAtMs } = params;
// Auth selection is intentionally centralized: retry decisions depend on
// whether a token was explicit, cached, or compatibility-derived.
const selectedAuth = this.selectConnectAuth(role);
const selectedAuth = this.selectConnectAuth(params.storedAuth);
const {
authDeviceToken,
authApprovalRuntimeToken,
@ -953,7 +876,73 @@ export class GatewayClient {
};
}
private handleConnectHello(helloOk: HelloOk, assembled: AssembledConnect): void {
private handleConnectHello(
helloOk: HelloOk,
assembled: AssembledConnect,
authority: GatewayProtocolConnectAuthority,
): MaybePromise<void> {
authority.assertCurrent();
const role = this.opts.role ?? "operator";
const authInfo = helloOk.auth;
const observation = this.connectionStoredToken;
const persisted = () => {
if (authInfo?.deviceToken && (authInfo.role ?? role) === role && observation) {
observation.token = authInfo.deviceToken;
}
if (this.opts.preferBootstrapToken) {
// This accepted receipt redeems setup auth even when its transport has retired.
this.opts.token = undefined;
this.opts.bootstrapToken = undefined;
this.opts.password = undefined;
this.opts.preferBootstrapToken = false;
}
};
let stored: MaybePromise<void> = undefined;
if (authInfo?.deviceToken && this.opts.deviceIdentity) {
const tokenRole = authInfo.role ?? role;
const scopes =
tokenRole === role && authInfo.deviceToken === assembled.storedToken
? (assembled.storedScopes ?? authInfo.scopes ?? [])
: (authInfo.scopes ?? []);
if (tokenRole === role && observation) {
observation.receiptToken = authInfo.deviceToken;
}
stored = this.deviceAuth.store(
{
deviceId: this.opts.deviceIdentity.deviceId,
role: tokenRole,
token: authInfo.deviceToken,
scopes,
env: this.opts.env,
...(tokenRole === role ? { expectedToken: assembled.storedToken ?? null } : {}),
},
persisted,
(error) => {
assembled.persistenceFailed = true;
const reported = this.notifyConnectError(error);
this.logError(
`gateway device token persistence failed: ${formatGatewayClientErrorForLog(error)}`,
);
return reported;
},
);
if (observation && stored instanceof Promise) {
observation.persistence = stored;
}
} else {
persisted();
}
const complete = () => {
if (authority.signal.aborted || !this.protocol.connected) {
return;
}
authority.assertCurrent();
this.completeConnectHello(helloOk);
};
return stored instanceof Promise ? stored.then(complete) : complete();
}
private completeConnectHello(helloOk: HelloOk): void {
const reconnectWithCurrentNodeProtocol =
this.useLegacyNodeProtocolEnvelope &&
this.shouldNegotiateLegacyNodeProtocol() &&
@ -965,47 +954,41 @@ export class GatewayClient {
this.pendingDeviceTokenRetry = false;
this.deviceTokenRetryBudgetUsed = false;
this.suppressedTransientPreHelloCleanCloses = 0;
const role = this.opts.role ?? "operator";
const authInfo = helloOk.auth;
if (authInfo?.deviceToken && this.opts.deviceIdentity) {
const tokenRole = authInfo.role ?? role;
const scopes =
tokenRole === role && authInfo.deviceToken === assembled.storedToken
? (assembled.storedScopes ?? authInfo.scopes ?? [])
: (authInfo.scopes ?? []);
this.deps.storeDeviceAuthToken({
deviceId: this.opts.deviceIdentity.deviceId,
role: tokenRole,
token: authInfo.deviceToken,
scopes,
env: this.opts.env,
});
}
if (this.opts.preferBootstrapToken) {
// The setup credential is single-use; reconnects must use the stored device token.
this.opts.token = undefined;
this.opts.bootstrapToken = undefined;
this.opts.password = undefined;
this.opts.preferBootstrapToken = false;
}
this.tickIntervalMs =
typeof helloOk.policy?.tickIntervalMs === "number" ? helloOk.policy.tickIntervalMs : 30_000;
if (reconnectWithCurrentNodeProtocol) {
// A v4 Gateway accepted the exact-v3 probe as a legacy session. Reconnect
// before reporting readiness so node capabilities are not silently filtered.
this.suppressNextHelloCallback = true;
this.protocol.resetReconnectBackoff(250);
this.protocol.closeSocket(1012, "gateway protocol upgraded");
return;
}
this.lastTick = Date.now();
this.startTickWatch();
void assembled;
}
private handleConnectRequestFailure(
error: GatewayProtocolRequestError,
assembled: AssembledConnect,
) {
if (assembled.persistenceFailed) {
return { closeCode: 1008, closeReason: "connect failed" };
}
const detailCode =
error instanceof GatewayClientRequestError ? readConnectErrorDetailCode(error.details) : null;
const cleared =
this.opts.deviceIdentity &&
assembled.usingStoredDeviceToken &&
detailCode === ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH
? this.clearDeviceToken(this.opts.deviceIdentity.deviceId, { token: assembled.storedToken })
: undefined;
const decision = this.resolveConnectRequestFailure(error, assembled);
return cleared instanceof Promise ? cleared.then(() => decision) : decision;
}
private resolveConnectRequestFailure(
error: GatewayProtocolRequestError,
assembled: AssembledConnect,
) {
if (this.shouldRetryWithCurrentNodeProtocol(error)) {
const resetBackoff = !this.nodeProtocolTransitionPending;
@ -1028,7 +1011,6 @@ export class GatewayClient {
return { closeCode: 1008, closeReason: "connect retry" };
}
this.nodeProtocolTransitionPending = false;
const role = this.opts.role ?? "operator";
const detailCode =
error instanceof GatewayClientRequestError ? readConnectErrorDetailCode(error.details) : null;
const shouldRetryWithDeviceToken = shouldRetryGatewayWithDeviceToken({
@ -1039,21 +1021,6 @@ export class GatewayClient {
trustedEndpoint: this.isTrustedDeviceRetryEndpoint(),
errorDetails: error instanceof GatewayClientRequestError ? error.details : undefined,
});
if (
this.opts.deviceIdentity &&
assembled.usingStoredDeviceToken &&
detailCode === ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH
) {
const deviceId = this.opts.deviceIdentity.deviceId;
try {
this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env });
this.logDebug(`cleared stale device-auth token for device ${deviceId}`);
} catch (clearError) {
this.logDebug(
`failed clearing stale device-auth token for device ${deviceId}: ${String(clearError)}`,
);
}
}
if (shouldRetryWithDeviceToken) {
this.pendingDeviceTokenRetry = true;
this.deviceTokenRetryBudgetUsed = true;
@ -1147,7 +1114,7 @@ export class GatewayClient {
this.notifyConnectError(error);
this.logError(`gateway connect failed: ${formatGatewayClientErrorForLog(error)}`);
}
this.clearStaleDeviceTokenForClose(context.code, context.reason);
this.clearStaleDeviceTokenForClose(context);
if (
shouldPauseGatewayReconnect({
details,
@ -1179,35 +1146,71 @@ export class GatewayClient {
};
}
private clearStaleDeviceTokenForClose(code: number, reason: string): void {
private clearStaleDeviceTokenForClose(context: GatewayProtocolCloseContext): void {
const { code, reason, generation } = context;
if (
code !== 1008 ||
!normalizeGatewayErrorText(reason).includes("device token mismatch") ||
this.opts.token ||
this.opts.password ||
!this.opts.deviceIdentity
) {
return;
}
const deviceId = this.opts.deviceIdentity.deviceId;
const role = this.opts.role ?? "operator";
try {
this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env });
this.logDebug(`cleared stale device-auth token for device ${deviceId}`);
} catch (error) {
const observation =
this.connectionStoredToken?.generation === generation
? this.connectionStoredToken
: { token: undefined };
void this.clearDeviceToken(
this.opts.deviceIdentity.deviceId,
observation,
() => !this.opts.token && !this.opts.password,
);
}
private clearDeviceToken(
deviceId: string,
observation: DeviceAuthTokenObservation,
canClear = () => true,
) {
const failed = (error: unknown) => {
this.logDebug(
`failed clearing stale device-auth token for device ${deviceId}: ${String(error)}`,
);
};
const cleared = () => this.logDebug(`cleared stale device-auth token for device ${deviceId}`);
try {
const result = this.deviceAuth.clear(
{
deviceId,
role: this.opts.role ?? "operator",
env: this.opts.env,
// Cleanup follows a received rejection even after its transport retires.
assertCurrent: () => {
if (this.stopped) {
throw new Error("gateway client stopped");
}
},
},
observation,
canClear,
);
return result instanceof Promise ? result.then(cleared, failed) : cleared();
} catch (error) {
failed(error);
}
}
private notifyConnectError(error: Error) {
private notifyConnectError(error: Error): boolean {
if (!this.opts.onConnectError) {
return false;
}
try {
this.opts.onConnectError?.(error);
this.opts.onConnectError(error);
return true;
} catch (err) {
this.logDebug(
`gateway client connect error handler error: ${formatGatewayClientErrorForLog(err)}`,
);
return false;
}
}
@ -1279,14 +1282,7 @@ export class GatewayClient {
}
}
private selectConnectAuth(role: string): GatewayConnectAuthSelection {
const storedAuth = this.opts.deviceIdentity
? this.deps.loadDeviceAuthToken({
deviceId: this.opts.deviceIdentity.deviceId,
role,
env: this.opts.env,
})
: null;
private selectConnectAuth(storedAuth: DeviceAuthTokenRecord | null): GatewayConnectAuthSelection {
return selectGatewayConnectAuth({
token: this.opts.token,
bootstrapToken: this.opts.bootstrapToken,

24
packages/gateway-client/src/protocol-client-contract.ts
View file

@ -14,7 +14,11 @@ export type GatewayProtocolSocketHandlers = {
close: (code: number, reason: string) => void;
error: (error: Error) => void;
};
type GatewayProtocolConnectContext<TPlan> = {
export type GatewayProtocolConnectAuthority = {
signal: AbortSignal;
assertCurrent: () => void;
};
type GatewayProtocolConnectContext<TPlan> = GatewayProtocolConnectAuthority & {
generation: number;
nonce: string | null;
challengeTs: number | null | undefined;
@ -66,20 +70,22 @@ export type GatewayProtocolClientOptions<TPlan> = {
createRequestError?: (error: Partial<ErrorShape>) => GatewayProtocolRequestError;
createRequestTimeoutError?: (method: string, timeoutMs: number, requestSent: boolean) => Error;
createRequestAbortError?: (method: string) => Error;
buildConnectPlan: (params: {
nonce: string | null;
challengeTs: number | null | undefined;
serverCapabilities: readonly string[];
generation: number;
}) => TPlan | Promise<TPlan>;
buildConnectPlan: (
params: GatewayProtocolConnectAuthority & {
nonce: string | null;
challengeTs: number | null | undefined;
serverCapabilities: readonly string[];
generation: number;
},
) => TPlan | Promise<TPlan>;
buildConnectParams: (plan: TPlan) => unknown;
onConnectPlanError?: (error: Error) => GatewayProtocolConnectDecision;
onConnectHello?: (hello: HelloOk, context: GatewayProtocolConnectContext<TPlan>) => void;
onConnectHello?: (hello: HelloOk, context: GatewayProtocolConnectContext<TPlan>) => unknown;
onHello?: (hello: HelloOk) => void;
onConnectFailure?: (
error: GatewayProtocolRequestError,
context: GatewayProtocolConnectContext<TPlan>,
) => GatewayProtocolConnectDecision;
) => GatewayProtocolConnectDecision | Promise<GatewayProtocolConnectDecision>;
resolveClose: (context: GatewayProtocolCloseContext) => GatewayProtocolCloseDecision;
onClose?: (context: GatewayProtocolCloseContext, decision: GatewayProtocolCloseDecision) => void;
notifyStoppedClose?: boolean;

12
packages/gateway-client/src/protocol-client.handshake.test.ts
View file

@ -15,7 +15,7 @@ function createHandshakeClient(
) {
const connections: HandshakeConnection[] = [];
const onHello = vi.fn();
const onConnectHello = vi.fn();
const onConnectHello = vi.fn(() => ({ ignored: true }));
const onClose = vi.fn();
const onTiming = vi.fn();
let nextRequestId = 0;
@ -261,6 +261,8 @@ describe("GatewayProtocolClient connect handshake", () => {
challengeTs: 1_700_000_000_123,
serverCapabilities: [],
generation: 1,
signal: expect.any(AbortSignal),
assertCurrent: expect.any(Function),
});
client.stop();
});
@ -286,6 +288,8 @@ describe("GatewayProtocolClient connect handshake", () => {
challengeTs: 1,
serverCapabilities: ["model-catalog-snapshot"],
generation: 1,
signal: expect.any(AbortSignal),
assertCurrent: expect.any(Function),
});
first.handlers.close(1006, "reconnect");
await vi.advanceTimersByTimeAsync(10);
@ -297,6 +301,8 @@ describe("GatewayProtocolClient connect handshake", () => {
challengeTs: 1_800_000_000_000,
serverCapabilities: [],
generation: 2,
signal: expect.any(AbortSignal),
assertCurrent: expect.any(Function),
});
} finally {
client.stop();
@ -325,6 +331,8 @@ describe("GatewayProtocolClient connect handshake", () => {
challengeTs: null,
serverCapabilities: [],
generation: 1,
signal: expect.any(AbortSignal),
assertCurrent: expect.any(Function),
});
client.stop();
@ -348,6 +356,8 @@ describe("GatewayProtocolClient connect handshake", () => {
challengeTs: null,
serverCapabilities: [],
generation: 1,
signal: expect.any(AbortSignal),
assertCurrent: expect.any(Function),
});
secondClient.client.stop();
});

84
packages/gateway-client/src/protocol-client.ts
View file

@ -11,6 +11,7 @@ import type {
ConnectTimingState,
GatewayProtocolClientOptions,
GatewayProtocolCloseContext,
GatewayProtocolConnectAuthority,
GatewayProtocolSocket,
GatewayProtocolTiming,
} from "./protocol-client-contract.js";
@ -30,6 +31,7 @@ export {
export type {
GatewayProtocolCloseContext,
GatewayProtocolConnectAuthority,
GatewayProtocolSocket,
GatewayProtocolSocketHandlers,
GatewayProtocolTiming,
@ -45,6 +47,7 @@ export class GatewayProtocolClient<TPlan> {
private readonly listeners = new GatewayEventListeners<EventFrame>();
private stopped = true;
private generation = 0;
private connectionAbort: AbortController | null = null;
private lastSeq: number | null = null;
private connectNonce: string | null = null;
private connectChallengeTs: number | null | undefined;
@ -106,6 +109,7 @@ export class GatewayProtocolClient<TPlan> {
stop(): void {
this.stopped = true;
this.connectionAbort?.abort();
this.clearHandshakeTimer();
this.reconnectSignal = null;
this.reconnectSupervisor.reset();
@ -142,6 +146,7 @@ export class GatewayProtocolClient<TPlan> {
}
closeSocket(code?: number, reason?: string): void {
this.connectionAbort?.abort();
this.socket?.close(code, reason);
}
@ -226,6 +231,7 @@ export class GatewayProtocolClient<TPlan> {
return;
}
this.generation = generation;
this.connectionAbort = new AbortController();
this.socket = socket;
const now = this.nowMs();
this.connectTiming = {
@ -294,6 +300,7 @@ export class GatewayProtocolClient<TPlan> {
challengeTs: this.connectChallengeTs,
serverCapabilities: this.serverCapabilities,
generation,
...this.connectAuthority(socket, generation),
});
} catch (error) {
this.handleConnectPlanError(socket, generation, error);
@ -313,7 +320,7 @@ export class GatewayProtocolClient<TPlan> {
generation: number,
error: unknown,
): void {
if (!this.isActive(socket, generation)) {
if (!this.isConnectCurrent(socket, generation)) {
return;
}
const normalized = error instanceof Error ? error : new Error(String(error));
@ -329,10 +336,11 @@ export class GatewayProtocolClient<TPlan> {
}
private sendConnectPlan(socket: GatewayProtocolSocket, generation: number, plan: TPlan): void {
if (!this.isActive(socket, generation) || !socket.isOpen()) {
if (!this.isConnectCurrent(socket, generation)) {
return;
}
const context = {
...this.connectAuthority(socket, generation),
generation,
nonce: this.connectNonce,
challengeTs: this.connectChallengeTs,
@ -345,7 +353,7 @@ export class GatewayProtocolClient<TPlan> {
.then((hello) => {
// Closing transports remain current until their close callback runs;
// a late response must not publish readiness or reset reconnect backoff.
if (!this.isActive(socket, generation) || !socket.isOpen()) {
if (!this.isConnectCurrent(socket, generation)) {
return;
}
this.helloReceived = true;
@ -353,8 +361,17 @@ export class GatewayProtocolClient<TPlan> {
this.connectFailure = undefined;
this.reconnectSupervisor.reset();
this.recordTiming("hello", generation, plan);
this.opts.onConnectHello?.(hello, context);
this.invoke("hello", () => this.opts.onHello?.(hello));
const publishHello = () => {
if (!this.isConnectCurrent(socket, generation)) {
return;
}
this.invoke("hello", () => this.opts.onHello?.(hello));
};
const accepted = this.opts.onConnectHello?.(hello, context);
if (accepted instanceof Promise) {
return accepted.then(publishHello);
}
return publishHello();
})
.catch((error: unknown) => {
if (!this.isActive(socket, generation)) {
@ -364,18 +381,37 @@ export class GatewayProtocolClient<TPlan> {
error instanceof GatewayProtocolRequestError
? error
: new GatewayProtocolRequestError({ message: String(error) });
// Close can arrive while adapter cleanup is pending; retain the received error now.
this.connectFailure = { error: requestError };
const outcome = this.opts.onConnectFailure?.(requestError, context) ?? {
closeCode: 1008,
closeReason: "connect failed",
};
this.connectFailure = {
error: requestError,
reconnectDelayMs: outcome.reconnectDelayMs,
const applyFailure = (decision: Awaited<typeof outcome>) => {
if (!this.isActive(socket, generation)) {
return;
}
this.connectFailure = {
error: requestError,
reconnectDelayMs: decision.reconnectDelayMs,
};
if (decision.stop) {
this.stopped = true;
}
this.connectionAbort?.abort();
socket.close(decision.closeCode, decision.closeReason);
};
if (outcome.stop) {
this.stopped = true;
if (outcome instanceof Promise) {
return outcome.then(applyFailure);
}
socket.close(outcome.closeCode, outcome.closeReason);
return applyFailure(outcome);
})
.catch((error: unknown) => {
if (!this.isConnectCurrent(socket, generation)) {
return;
}
this.opts.onConnectError?.(error instanceof Error ? error : new Error(String(error)));
this.closeSocket(1008, "connect failed");
});
}
@ -467,6 +503,7 @@ export class GatewayProtocolClient<TPlan> {
return;
}
this.socket = null;
this.connectionAbort?.abort();
this.clearHandshakeTimer();
const context: GatewayProtocolCloseContext = {
...this.closeContext(),
@ -551,6 +588,31 @@ export class GatewayProtocolClient<TPlan> {
};
}
private isConnectCurrent(socket: GatewayProtocolSocket, generation: number): boolean {
return (
this.isActive(socket, generation) && socket.isOpen() && !this.connectionAbort?.signal.aborted
);
}
private connectAuthority(
socket: GatewayProtocolSocket,
generation: number,
): GatewayProtocolConnectAuthority {
const signal = this.connectionAbort?.signal;
if (!signal) {
throw new Error("gateway connection authority is unavailable");
}
return {
signal,
assertCurrent: () => {
signal.throwIfAborted();
if (!this.isConnectCurrent(socket, generation)) {
throw new Error("gateway connection retired");
}
},
};
}
private isActive(socket: GatewayProtocolSocket, generation: number): boolean {
return !this.stopped && this.socket === socket && this.generation === generation;
}

22
packages/gateway-client/src/request-error.ts
View file

@ -11,3 +11,25 @@ export class GatewayClientRequestError extends GatewayProtocolRequestError {
this.name = "GatewayClientRequestError";
}
}
const GATEWAY_CONNECT_ASSEMBLY_ERROR = Symbol("gateway.connectAssemblyError");
type GatewayConnectAssemblyError = Error & {
[GATEWAY_CONNECT_ASSEMBLY_ERROR]?: true;
};
export function markGatewayConnectAssemblyError(error: Error): Error {
Object.defineProperty(error, GATEWAY_CONNECT_ASSEMBLY_ERROR, {
configurable: true,
value: true,
});
return error;
}
export function isGatewayConnectAssemblyError(value: unknown): value is Error {
return (
value instanceof Error &&
// SAFETY: The Error may carry this module's optional private-symbol marker; absence reads undefined.
(value as GatewayConnectAssemblyError)[GATEWAY_CONNECT_ASSEMBLY_ERROR] === true
);
}

206
src/gateway/client.device-auth.test.ts Normal file
View file

@ -0,0 +1,206 @@
import assert from "node:assert/strict";
import { beforeEach, expect, it, vi } from "vitest";
import type { GatewayClientHostDeps } from "../../packages/gateway-client/src/index.js";
import type { loadDeviceAuthToken } from "../infra/device-auth-store.js";
import { GatewayClient } from "./client.js";
const fixture = vi.hoisted(() => ({
hosts: [] as GatewayClientHostDeps[],
token: "fixture-existing" as string | undefined,
malformed: false,
load: vi.fn(),
store: vi.fn(),
clear: vi.fn(),
readOnlyLoad: vi.fn(),
}));
vi.mock("../../packages/gateway-client/src/index.js", async (importOriginal) => ({
...(await importOriginal<typeof import("../../packages/gateway-client/src/index.js")>()),
GatewayClient: class {
constructor(private readonly options: { hostDeps: GatewayClientHostDeps }) {}
start() {
fixture.hosts.push(this.options.hostDeps);
}
},
}));
vi.mock("../infra/device-auth-store.js", async (importOriginal) => ({
...(await importOriginal<typeof import("../infra/device-auth-store.js")>()),
loadDeviceAuthToken: fixture.load,
loadOriginDeviceToken: fixture.load,
storeDeviceAuthToken: fixture.store,
storeOriginDeviceToken: fixture.store,
clearDeviceAuthToken: fixture.clear,
clearOriginDeviceToken: fixture.clear,
loadDeviceAuthTokenReadOnly: fixture.readOnlyLoad,
loadOriginDeviceTokenReadOnly: fixture.readOnlyLoad,
}));
beforeEach(() => {
fixture.hosts.length = 0;
fixture.token = "fixture-existing";
fixture.malformed = false;
fixture.load
.mockReset()
.mockImplementation((params: Parameters<typeof loadDeviceAuthToken>[0]) => {
const entry =
fixture.token === undefined || fixture.malformed
? null
: {
token: fixture.token,
role: "operator",
scopes: [],
updatedAtMs: 1,
};
params.onSnapshot?.({ entry, expectedToken: fixture.token ?? null });
return entry;
});
fixture.store
.mockReset()
.mockImplementation((params: { token: string; expectedToken?: string | null }) => {
if (
params.expectedToken === undefined ||
(params.expectedToken === null
? fixture.token === undefined
: params.expectedToken === fixture.token)
) {
fixture.token = params.token;
return { token: params.token, role: "operator", scopes: [], updatedAtMs: 1 };
}
return null;
});
fixture.clear
.mockReset()
.mockImplementation((params: { expectedToken?: string; observedToken?: string }) => {
if (
params.expectedToken === fixture.token ||
(params.observedToken?.trim() === params.expectedToken &&
params.observedToken === fixture.token)
) {
fixture.token = undefined;
return true;
}
return false;
});
fixture.readOnlyLoad.mockReset();
});
function host(origin: boolean, readOnly = false, explicit = true) {
const client = new GatewayClient({
url: "wss://gateway.example.test",
...(origin ? { deviceAuthScope: "wss://gateway.example.test" } : {}),
...(explicit ? { token: "fixture-shared-auth" } : {}),
...(readOnly ? { sharedStateMode: "read-only" } : {}),
});
client.start();
const deps = fixture.hosts[0];
assert(deps?.loadDeviceAuthToken && deps.storeDeviceAuthToken && deps.clearDeviceAuthToken);
return {
load: deps.loadDeviceAuthToken,
store: deps.storeDeviceAuthToken,
clear: deps.clearDeviceAuthToken,
};
}
const observations = [
{ name: "valid", token: "fixture-existing", malformed: false },
{ name: "malformed scopes", token: "fixture-existing", malformed: true },
{ name: "absent", token: undefined, malformed: false },
{ name: "raw whitespace", token: " fixture-existing ", malformed: false },
];
it.each(
observations.flatMap(({ name, token, malformed }) =>
[false, true].flatMap((origin) =>
[false, true].map((rotated) => ({ name, token, malformed, origin, rotated })),
),
),
)(
"uses exact $name comparison privately (origin: $origin, rotated: $rotated)",
({ token, malformed, origin, rotated }) => {
fixture.token = token;
fixture.malformed = malformed;
const deps = host(origin);
const scope = { deviceId: "fixture-device", role: "operator" };
const loaded = deps.load(scope);
assert(
!(loaded instanceof Promise),
"the synchronous facade must return its load result inline",
);
if (origin || malformed || token === undefined) {
expect(loaded).toBeNull();
} else {
expect(loaded?.token).toBe(token);
}
if (rotated) {
fixture.token = "fixture-newer";
}
const stored = deps.store({
...scope,
token: "fixture-issued",
scopes: [],
expectedToken: loaded?.token?.trim() ?? null,
});
expect(stored).toEqual(
rotated
? null
: {
token: "fixture-issued",
role: "operator",
scopes: [],
updatedAtMs: 1,
},
);
expect(fixture.store).toHaveBeenCalledWith({
...scope,
...(origin ? { gatewayScope: "wss://gateway.example.test" } : {}),
token: "fixture-issued",
scopes: [],
expectedToken: token ?? null,
});
expect(fixture.token).toBe(rotated ? "fixture-newer" : "fixture-issued");
},
);
it.each([false, true])(
"cleans exact raw legacy bytes without redirecting receipt cleanup (origin: %s)",
(origin) => {
fixture.token = " fixture-existing ";
const deps = host(origin, false, false);
const scope = { deviceId: "fixture-device", role: "operator" };
const loaded = deps.load(scope);
assert(
!(loaded instanceof Promise),
"the synchronous facade must return its load result inline",
);
expect(deps.clear({ ...scope, expectedToken: "fixture-existing" })).toBe(true);
expect(fixture.token).toBeUndefined();
expect(fixture.clear).toHaveBeenLastCalledWith({
...scope,
...(origin ? { gatewayScope: "wss://gateway.example.test" } : {}),
expectedToken: "fixture-existing",
observedToken: " fixture-existing ",
});
fixture.token = "fixture-newer";
expect(deps.clear({ ...scope, expectedToken: "fixture-issued" })).toBe(false);
expect(fixture.token).toBe("fixture-newer");
expect(fixture.clear).toHaveBeenLastCalledWith({
...scope,
...(origin ? { gatewayScope: "wss://gateway.example.test" } : {}),
expectedToken: "fixture-issued",
});
},
);
it("keeps explicit read-only origin auth off storage reads and writes", () => {
const deps = host(true, true);
const scope = { deviceId: "fixture-device", role: "operator" };
expect(deps.load(scope)).toBeNull();
expect(
deps.store({ ...scope, token: "fixture-issued", scopes: [], expectedToken: null }),
).toBeUndefined();
expect(fixture.load).not.toHaveBeenCalled();
expect(fixture.readOnlyLoad).not.toHaveBeenCalled();
expect(fixture.store).not.toHaveBeenCalled();
});

16
src/gateway/client.test-support.ts Normal file
View file

@ -0,0 +1,16 @@
import { vi } from "vitest";
export function waitForFast<T>(
callback: () => T | Promise<T>,
options: { timeout?: number; interval?: number } = {},
) {
return vi.waitFor(callback, { interval: 1, ...options });
}
export function firstMockArg(mock: ReturnType<typeof vi.fn>, label: string): unknown {
const [arg] = mock.mock.calls[0] ?? [];
if (arg === undefined) {
throw new Error(`expected ${label}`);
}
return arg;
}

23
src/gateway/client.test.ts
View file

@ -20,13 +20,7 @@ import {
} from "../infra/device-identity.js";
import { captureEnv } from "../test-utils/env.js";
import type { GatewayClientOptions } from "./client.js";
function waitForFast<T>(
callback: () => T | Promise<T>,
options: { timeout?: number; interval?: number } = {},
) {
return vi.waitFor(callback, { interval: 1, ...options });
}
import { firstMockArg, waitForFast } from "./client.test-support.js";
type MockLoggingConfig = {
redactPatterns?: string[];
@ -190,8 +184,7 @@ vi.mock("../infra/device-auth-store.js", async () => {
loadDeviceAuthToken: (...args: unknown[]) => loadDeviceAuthTokenMock(...args),
loadDeviceAuthTokenReadOnly: (...args: unknown[]) => loadDeviceAuthTokenReadOnlyMock(...args),
loadOriginDeviceToken: (...args: unknown[]) => loadOriginDeviceTokenMock(...args),
loadOriginDeviceTokenReadOnly: (...args: unknown[]) =>
loadOriginDeviceTokenReadOnlyMock(...args),
loadOriginDeviceTokenReadOnly: loadOriginDeviceTokenReadOnlyMock,
storeDeviceAuthToken: (...args: unknown[]) => storeDeviceAuthTokenMock(...args),
storeOriginDeviceToken: (...args: unknown[]) => storeOriginDeviceTokenMock(...args),
clearDeviceAuthToken: (...args: unknown[]) => clearDeviceAuthTokenMock(...args),
@ -250,13 +243,6 @@ function expectRecordFields(
return record;
}
function firstMockArg(mock: ReturnType<typeof vi.fn>, label: string): unknown {
const [arg] = mock.mock.calls[0] ?? [];
if (arg === undefined) {
throw new Error(`expected ${label}`);
}
return arg;
}
function createClientWithIdentity(
deviceId: string,
onClose: (code: number, reason: string) => void,
@ -738,6 +724,7 @@ describe("GatewayClient close handling", () => {
deviceId: "dev-1",
role: "operator",
env,
assertCurrent: expect.any(Function),
});
expect(logDebugMock).toHaveBeenCalledWith("cleared stale device-auth token for device dev-1");
expect(onClose).toHaveBeenCalledWith(
@ -2321,7 +2308,7 @@ describe("GatewayClient connect auth payload", () => {
connectId: connect.id,
failureDetails: { code: "AUTH_TOKEN_MISMATCH", canRetryWithDeviceToken: true },
});
expect(loadOriginDeviceTokenMock).not.toHaveBeenCalled();
expect(loadOriginDeviceTokenMock).toHaveBeenCalledOnce();
expect(onReconnectPaused).toHaveBeenCalledWith({
code: 1008,
reason: "connect failed",
@ -2363,6 +2350,7 @@ describe("GatewayClient connect auth payload", () => {
token: "stored-origin-token",
scopes: ["operator.admin", "operator.read"],
env: undefined,
expectedToken: "stored-origin-token",
});
});
client.stop();
@ -2398,6 +2386,7 @@ describe("GatewayClient connect auth payload", () => {
token: "issued-origin-token",
scopes: ["operator.read"],
env: undefined,
expectedToken: null,
});
});
expect(storeDeviceAuthTokenMock).not.toHaveBeenCalled();

67
src/gateway/client.ts
View file

@ -30,7 +30,7 @@ import {
import { logDebug, logError } from "../logger.js";
import { redactToolPayloadText } from "../logging/redact.js";
import { registerSecretValueForRedaction } from "../logging/secret-redaction-registry.js";
import type { DeviceAuthEntry } from "../shared/device-auth.js";
import { type DeviceAuthEntry, normalizeDeviceAuthRole } from "../shared/device-auth.js";
import { resolveGatewayClientPlatformIdentity } from "../shared/gateway-client-platform.js";
import { VERSION } from "../version.js";
@ -67,24 +67,64 @@ function createOpenClawGatewayClientHostDeps(
const rotationFence = preparedDeviceAuth
? { expectedToken: preparedDeviceAuth.token }
: undefined;
let tokenObservation:
| { deviceId: string; role: string; expectedToken: string | null }
| undefined;
const observe = (params: { deviceId: string; role: string }) => {
const deviceId = params.deviceId;
const role = normalizeDeviceAuthRole(params.role);
return (snapshot: { expectedToken: string | null }) => {
tokenObservation = { deviceId, role, expectedToken: snapshot.expectedToken };
};
};
const observedFor = (params: { deviceId: string; role: string }) =>
tokenObservation?.deviceId === params.deviceId &&
tokenObservation.role === normalizeDeviceAuthRole(params.role)
? tokenObservation
: undefined;
const writeFence = (params: { deviceId: string; role: string }) => {
if (rotationFence) {
return rotationFence;
}
// Each connection's accepted writes settle before its successor loads another observation.
const observed = observedFor(params);
return observed ? { expectedToken: observed.expectedToken } : undefined;
};
const clearFence = (params: { deviceId: string; role: string; expectedToken?: string }) => {
const expectedToken = rotationFence?.expectedToken ?? params.expectedToken;
const raw = observedFor(params)?.expectedToken;
return {
...rotationFence,
...(typeof raw === "string" && raw !== expectedToken && raw.trim() === expectedToken
? { observedToken: raw }
: {}),
};
};
const deviceAuthDeps: Pick<
GatewayClientHostDeps,
"loadDeviceAuthToken" | "storeDeviceAuthToken" | "clearDeviceAuthToken"
> = deviceAuthScope
? {
loadDeviceAuthToken: (params) =>
suppressStoredDeviceAuth
? null
: readOnly
? loadOriginDeviceTokenReadOnly({ ...params, gatewayScope: deviceAuthScope })
: loadOriginDeviceToken({ ...params, gatewayScope: deviceAuthScope }),
loadDeviceAuthToken: (params) => {
if (readOnly) {
return suppressStoredDeviceAuth
? null
: loadOriginDeviceTokenReadOnly({ ...params, gatewayScope: deviceAuthScope });
}
const load = loadOriginDeviceToken({
...params,
gatewayScope: deviceAuthScope,
onSnapshot: observe(params),
});
return suppressStoredDeviceAuth ? null : load;
},
storeDeviceAuthToken: readOnly
? () => {}
: (params) =>
storeOriginDeviceToken({
...params,
gatewayScope: deviceAuthScope,
...rotationFence,
...writeFence(params),
}),
clearDeviceAuthToken: readOnly
? () => {}
@ -92,7 +132,7 @@ function createOpenClawGatewayClientHostDeps(
clearOriginDeviceToken({
...params,
gatewayScope: deviceAuthScope,
...rotationFence,
...clearFence(params),
}),
}
: readOnly
@ -102,9 +142,12 @@ function createOpenClawGatewayClientHostDeps(
clearDeviceAuthToken: () => {},
}
: {
loadDeviceAuthToken,
storeDeviceAuthToken: (params) => storeDeviceAuthToken({ ...params, ...rotationFence }),
clearDeviceAuthToken: (params) => clearDeviceAuthToken({ ...params, ...rotationFence }),
loadDeviceAuthToken: (params) =>
loadDeviceAuthToken({ ...params, onSnapshot: observe(params) }),
storeDeviceAuthToken: (params) =>
storeDeviceAuthToken({ ...params, ...writeFence(params) }),
clearDeviceAuthToken: (params) =>
clearDeviceAuthToken({ ...params, ...clearFence(params) }),
};
const preparedDeviceAuthDeps = preparedDeviceAuth
? { ...deviceAuthDeps, loadDeviceAuthToken: () => preparedDeviceAuth }

1
src/tui/gateway-chat.scopes.test.ts
View file

@ -298,6 +298,7 @@ describe("GatewayChatClient operator scopes", () => {
token: "approved-tui-device-token",
scopes: requestedScopes,
env: undefined,
expectedToken: "cached-read-only-device-token",
});
const history = approved.client.loadHistory({ sessionKey: "main", limit: 20 });