diff --git a/src/server/realtime/sessionEventHub.test.ts b/src/server/realtime/sessionEventHub.test.ts index 08641b9..c397dec 100644 --- a/src/server/realtime/sessionEventHub.test.ts +++ b/src/server/realtime/sessionEventHub.test.ts @@ -6,6 +6,7 @@ class FakeSocket extends EventEmitter implements RealtimeSocket { readonly OPEN = 1; readyState = this.OPEN; send = vi.fn(); + terminate = vi.fn(); } describe("SessionEventHub", () => { @@ -54,7 +55,7 @@ describe("SessionEventHub", () => { expect(removed.send).not.toHaveBeenCalled(); }); - it("continues publishing session events when one socket send fails", () => { + it("terminates a failed session socket without disrupting healthy delivery or sequence watermarks", () => { const hub = new SessionEventHub(); const failed = new FakeSocket(); const healthy = new FakeSocket(); @@ -65,6 +66,7 @@ describe("SessionEventHub", () => { hub.publish("s1", { type: "assistant.delta", text: "hello" }); expect(failed.send).toHaveBeenCalledOnce(); + expect(failed.terminate).toHaveBeenCalledOnce(); expect(healthy.send).toHaveBeenCalledWith(JSON.stringify({ type: "assistant.delta", text: "hello", seq: 1 })); expect(hub.currentSeq("s1")).toBe(1); @@ -72,6 +74,7 @@ describe("SessionEventHub", () => { hub.publish("s1", { type: "assistant.delta", text: "again" }); expect(failed.send).not.toHaveBeenCalled(); + expect(failed.terminate).toHaveBeenCalledOnce(); expect(healthy.send).toHaveBeenLastCalledWith(JSON.stringify({ type: "assistant.delta", text: "again", seq: 2 })); expect(hub.currentSeq("s1")).toBe(2); }); @@ -100,23 +103,26 @@ describe("SessionEventHub", () => { expect(sessionSocket.send).not.toHaveBeenCalled(); }); - it("continues publishing unstamped global events when one socket send fails", () => { + it("contains termination failures while publishing unstamped global events", () => { const hub = new SessionEventHub(); const failed = new FakeSocket(); const healthy = new FakeSocket(); failed.send.mockImplementation(() => { throw new Error("socket closed"); }); + failed.terminate.mockImplementation(() => { throw new Error("termination failed"); }); hub.addGlobal(failed); hub.addGlobal(healthy); hub.publishGlobal({ type: "session.name", sessionId: "s1", name: "Renamed" }); expect(failed.send).toHaveBeenCalledOnce(); + expect(failed.terminate).toHaveBeenCalledOnce(); expect(healthy.send).toHaveBeenCalledWith(JSON.stringify({ type: "session.name", sessionId: "s1", name: "Renamed" })); failed.send.mockClear(); hub.publishGlobal({ type: "session.name", sessionId: "s1", name: "Renamed again" }); expect(failed.send).not.toHaveBeenCalled(); + expect(failed.terminate).toHaveBeenCalledOnce(); expect(healthy.send).toHaveBeenLastCalledWith(JSON.stringify({ type: "session.name", sessionId: "s1", name: "Renamed again" })); }); diff --git a/src/server/realtime/sessionEventHub.ts b/src/server/realtime/sessionEventHub.ts index 9df8a1a..2cbb39f 100644 --- a/src/server/realtime/sessionEventHub.ts +++ b/src/server/realtime/sessionEventHub.ts @@ -5,6 +5,7 @@ export interface RealtimeSocket { readonly OPEN: number; readyState: number; send(payload: string): void; + terminate(): void; on(event: "close", listener: () => void): unknown; } @@ -64,6 +65,11 @@ export class SessionEventHub { socket.send(payload); } catch { sockets.delete(socket); + try { + socket.terminate(); + } catch { + // Removal is authoritative; cleanup failure must not block healthy sockets. + } } } }