fix(realtime): terminate failed sockets

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" }));
   });
 

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.
+        }
       }
     }
   }