From 8d596aa6511bec3c238647c71491412b89204df3 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 19 Jun 2026 04:43:12 +0200 Subject: [PATCH] fix(sdk): reject work after client close --- packages/sdk/src/client.ts | 45 ++++++++++++++--- packages/sdk/src/index.test.ts | 81 ++++++++++++++++++++++++++++++ packages/sdk/src/transport.test.ts | 9 ++++ packages/sdk/src/transport.ts | 8 +++ 4 files changed, 137 insertions(+), 6 deletions(-) diff --git a/packages/sdk/src/client.ts b/packages/sdk/src/client.ts index 5ee6aa35167..5476ace76fb 100644 --- a/packages/sdk/src/client.ts +++ b/packages/sdk/src/client.ts @@ -327,8 +327,10 @@ export class OpenClaw { }); private readonly replayByRunId = new Map(); private connected = false; + private closed = false; private eventPumpPromise: Promise | null = null; private eventPumpReady: Promise | null = null; + private closePromise: Promise | null = null; constructor(options: OpenClawOptions = {}) { this.transport = @@ -351,24 +353,45 @@ export class OpenClaw { } async connect(): Promise { + this.assertOpen(); if (this.connected) { await this.startEventPump(); + this.assertOpen(); return; } if (isConnectableTransport(this.transport)) { await this.transport.connect(); } + this.assertOpen(); this.connected = true; await this.startEventPump(); + this.assertOpen(); } async close(): Promise { - await this.transport.close?.(); - await this.eventPumpPromise?.catch(() => {}); - this.normalizedEvents.close(); - this.eventPumpPromise = null; - this.eventPumpReady = null; - this.connected = false; + if (this.closePromise) { + return await this.closePromise; + } + if (this.closed) { + return; + } + this.closed = true; + this.closePromise = (async () => { + try { + await this.transport.close?.(); + await this.eventPumpPromise?.catch(() => {}); + } finally { + this.normalizedEvents.close(); + this.eventPumpPromise = null; + this.eventPumpReady = null; + this.connected = false; + } + })(); + try { + await this.closePromise; + } finally { + this.closePromise = null; + } } async request( @@ -377,6 +400,7 @@ export class OpenClaw { options?: GatewayRequestOptions, ): Promise { await this.connect(); + this.assertOpen(); return await this.transport.request(method, params, options); } @@ -392,13 +416,21 @@ export class OpenClaw { } rawEvents(filter?: (event: GatewayEvent) => boolean): AsyncIterable { + this.assertOpen(); return this.transport.events(filter); } + private assertOpen(): void { + if (this.closed) { + throw new Error("OpenClaw SDK client is closed"); + } + } + private async *iterateEvents( filter?: (event: OpenClawEvent) => boolean, ): AsyncIterable { await this.connect(); + this.assertOpen(); for await (const event of this.normalizedEvents.stream(filter)) { yield event; } @@ -409,6 +441,7 @@ export class OpenClaw { filter?: (event: OpenClawEvent) => boolean, ): AsyncIterable { await this.connect(); + this.assertOpen(); const replayEvents = this.replaySnapshot(runId); let hasCanonicalAssistantRunEvent = replayEvents.some(isAssistantRunEvent); let hasTerminalRunEvent = replayEvents.some(isTerminalRunEvent); diff --git a/packages/sdk/src/index.test.ts b/packages/sdk/src/index.test.ts index 4712de12772..89e838d45ce 100644 --- a/packages/sdk/src/index.test.ts +++ b/packages/sdk/src/index.test.ts @@ -54,6 +54,51 @@ class FakeTransport implements OpenClawTransport { } } +class DelayedConnectTransport extends FakeTransport { + connectCalls = 0; + private finishConnectCurrent: (() => void) | null = null; + + async connect(): Promise { + this.connectCalls += 1; + await new Promise((resolve) => { + this.finishConnectCurrent = resolve; + }); + } + + finishConnect(): void { + const finish = this.finishConnectCurrent; + if (!finish) { + throw new Error("expected pending connect"); + } + this.finishConnectCurrent = null; + finish(); + } +} + +class ClosingEventPumpTransport extends FakeTransport { + onFirstEventPoll?: () => void; + + override events(): AsyncIterable { + return { + [Symbol.asyncIterator]: (): AsyncIterator => { + let firstPoll = true; + return { + next: async (): Promise> => { + if (firstPoll) { + firstPoll = false; + this.onFirstEventPoll?.(); + await new Promise((resolve) => { + setTimeout(resolve, 0); + }); + } + return { done: true, value: undefined as never }; + }, + }; + }, + }; + } +} + class EventsOnlyTransport implements OpenClawTransport { constructor(private readonly eventSource: AsyncIterable) {} @@ -631,6 +676,42 @@ describe("OpenClaw SDK", () => { ]); }); + it("keeps close terminal when it races a pending connect", async () => { + const transport = new DelayedConnectTransport({ + "agents.list": { agents: [] }, + }); + const oc = new OpenClaw({ transport }); + + const connect = oc.connect(); + const close = oc.close(); + transport.finishConnect(); + + await expect(connect).rejects.toThrow("OpenClaw SDK client is closed"); + await close; + await expect(oc.agents.list()).rejects.toThrow("OpenClaw SDK client is closed"); + await expect(oc.events()[Symbol.asyncIterator]().next()).rejects.toThrow( + "OpenClaw SDK client is closed", + ); + expect(() => oc.rawEvents()).toThrow("OpenClaw SDK client is closed"); + expect(transport.connectCalls).toBe(1); + expect(transport.calls).toEqual([]); + }); + + it("does not request after close races event pump startup", async () => { + const transport = new ClosingEventPumpTransport({ + "agents.list": { agents: [] }, + }); + const oc = new OpenClaw({ transport }); + let closePromise: Promise | undefined; + transport.onFirstEventPoll = () => { + closePromise = oc.close(); + }; + + await expect(oc.agents.list()).rejects.toThrow("OpenClaw SDK client is closed"); + await closePromise; + expect(transport.calls).toEqual([]); + }); + it("cancels runs and checks model auth status through current Gateway methods", async () => { const transport = new FakeTransport({ agent: { status: "accepted", runId: "run_without_session" }, diff --git a/packages/sdk/src/transport.test.ts b/packages/sdk/src/transport.test.ts index 312bc36cfb8..55f98c33302 100644 --- a/packages/sdk/src/transport.test.ts +++ b/packages/sdk/src/transport.test.ts @@ -50,4 +50,13 @@ describe("GatewayClientTransport", () => { await connectExpectation; expect(client?.stopAndWait).toHaveBeenCalledTimes(1); }); + + it("rejects reconnect attempts after close", async () => { + const transport = new GatewayClientTransport(); + + await transport.close(); + + await expect(transport.connect()).rejects.toThrow("gateway transport is closed"); + expect(gatewayClientMocks.instances).toHaveLength(0); + }); }); diff --git a/packages/sdk/src/transport.ts b/packages/sdk/src/transport.ts index 075a0c0b908..a92c40d24e9 100644 --- a/packages/sdk/src/transport.ts +++ b/packages/sdk/src/transport.ts @@ -80,12 +80,16 @@ export class GatewayClientTransport implements ConnectableOpenClawTransport { private connectPromise: Promise | null = null; private rejectPendingConnect: ((error: Error) => void) | null = null; private closePromise: Promise | null = null; + private closed = false; constructor(options: GatewayClientTransportOptions = {}) { this.options = options; } connect(): Promise { + if (this.closed) { + return Promise.reject(new Error("gateway transport is closed")); + } if (this.connectPromise) { return this.connectPromise; } @@ -146,6 +150,10 @@ export class GatewayClientTransport implements ConnectableOpenClawTransport { if (this.closePromise) { return await this.closePromise; } + if (this.closed) { + return; + } + this.closed = true; this.eventsHub.close(); const client = this.client; this.client = null;