fix(stream): complete fanout mutations before callbacks

CHANGELOG.md

@@ -6,6 +6,8 @@ All notable changes to `@openclaw/libterminal` will be documented in this file.
 
 ### Fixed
 
+- Complete fanout delivery and teardown before event callbacks, preserving output order during reentrant publication and closing all subscribers when callbacks throw.
+
 - Preserve leading U+FEFF characters in terminal session IDs instead of routing frames under a different identity.
 
 - Discard queued stdio input when an attachment ends or fails, while still restoring the terminal without waiting for an active write.

src/stream.ts

@@ -113,23 +113,32 @@ export class TerminalFanout {
       return;
     }
     this.replay.append(bytes);
-    // Callbacks can replace subscribers during publication.
+    // Finish queue mutations before callbacks can publish again or replace subscribers.
     const subscribers = [...this.subscribers];
+    const events: FanoutEvent[] = [];
     for (const [subscriberId, subscriber] of subscribers) {
       const droppedBytes = subscriber.enqueue(bytes, this.slowSubscriberPolicy);
       if (droppedBytes === 0) {
         continue;
       }
-      this.onEvent?.({
+      events.push({
         type: "subscriber-overflow",
         subscriberId,
         droppedBytes,
         policy: this.slowSubscriberPolicy,
       });
       if (this.slowSubscriberPolicy === "disconnect") {
-        this.removeSubscriber(subscriberId, subscriber, "subscriber buffer overflow");
+        const closed = this.removeSubscriber(
+          subscriberId,
+          subscriber,
+          "subscriber buffer overflow",
+        );
+        if (closed) {
+          events.push(closed);
+        }
       }
     }
+    this.notify(events);
   }
 
   subscribe(id: string, options?: { replay?: boolean }): TerminalSubscription {
@@ -150,7 +159,12 @@ export class TerminalFanout {
     return {
       id,
       [Symbol.asyncIterator]: () => subscriber,
-      close: (reason?: string) => this.removeSubscriber(id, subscriber, reason),
+      close: (reason?: string) => {
+        const event = this.removeSubscriber(id, subscriber, reason);
+        if (event) {
+          this.notify([event]);
+        }
+      },
     };
   }
 
@@ -159,19 +173,46 @@ export class TerminalFanout {
       return;
     }
     this.closed = true;
+    const events: FanoutEvent[] = [];
     for (const [subscriberId, subscriber] of this.subscribers) {
-      this.removeSubscriber(subscriberId, subscriber, reason);
+      const event = this.removeSubscriber(subscriberId, subscriber, reason);
+      if (event) {
+        events.push(event);
+      }
     }
     this.replay.clear();
+    this.notify(events);
   }
 
-  private removeSubscriber(id: string, subscriber: AsyncByteQueue, reason?: string): void {
+  private removeSubscriber(
+    id: string,
+    subscriber: AsyncByteQueue,
+    reason?: string,
+  ): FanoutEvent | undefined {
     if (this.subscribers.get(id) !== subscriber) {
-      return;
+      return undefined;
     }
     this.subscribers.delete(id);
     subscriber.close();
-    this.onEvent?.({ type: "subscriber-closed", subscriberId: id, reason });
+    return { type: "subscriber-closed", subscriberId: id, reason };
+  }
+
+  private notify(events: FanoutEvent[]): void {
+    let failed = false;
+    let failure: unknown;
+    for (const event of events) {
+      try {
+        this.onEvent?.(event);
+      } catch (error) {
+        if (!failed) {
+          failed = true;
+          failure = error;
+        }
+      }
+    }
+    if (failed) {
+      throw failure;
+    }
   }
 }
 

test/stream.test.ts

@@ -161,6 +161,75 @@ describe("TerminalFanout", () => {
     expect((await iterator.next()).done).toBe(true);
   });
 
+  it("preserves publication order when an overflow callback publishes again", async () => {
+    const fanout = new TerminalFanout({
+      replayBytes: 0,
+      subscriberBufferBytes: 4,
+      onEvent: (event) => {
+        if (event.type === "subscriber-overflow") {
+          slow.close();
+          fanout.publish(bytes("b"));
+        }
+      },
+    });
+    const slow = fanout.subscribe("slow");
+    const fast = fanout.subscribe("fast")[Symbol.asyncIterator]();
+    fanout.publish(bytes("1234"));
+    await fast.next();
+
+    fanout.publish(bytes("a"));
+
+    expect((await fast.next()).value).toEqual(bytes("a"));
+    expect((await fast.next()).value).toEqual(bytes("b"));
+    fanout.close();
+  });
+
+  it("finishes publication when an overflow callback throws", async () => {
+    const failure = new Error("overflow callback failed");
+    const fanout = new TerminalFanout({
+      replayBytes: 0,
+      subscriberBufferBytes: 4,
+      onEvent: (event) => {
+        if (event.type === "subscriber-overflow") {
+          throw failure;
+        }
+      },
+    });
+    const slow = fanout.subscribe("slow")[Symbol.asyncIterator]();
+    const fast = fanout.subscribe("fast")[Symbol.asyncIterator]();
+    fanout.publish(bytes("1234"));
+    await fast.next();
+
+    expect(() => fanout.publish(bytes("5"))).toThrow(failure);
+
+    expect(fanout.subscriberCount).toBe(1);
+    expect((await fast.next()).value).toEqual(bytes("5"));
+    expect((await slow.next()).value).toEqual(bytes("1234"));
+    expect((await slow.next()).done).toBe(true);
+    fanout.close();
+  });
+
+  it("closes every subscriber before propagating a callback failure", async () => {
+    const failure = new Error("first close callback failed");
+    const notified: string[] = [];
+    const fanout = new TerminalFanout({
+      onEvent: (event) => {
+        notified.push(event.subscriberId);
+        throw notified.length === 1 ? failure : new Error("later callback failed");
+      },
+    });
+    const first = fanout.subscribe("first")[Symbol.asyncIterator]();
+    const second = fanout.subscribe("second")[Symbol.asyncIterator]();
+    const pending = [first.next(), second.next()];
+
+    expect(() => fanout.close()).toThrow(failure);
+
+    expect(fanout.subscriberCount).toBe(0);
+    expect(notified).toEqual(["first", "second"]);
+    expect((await Promise.all(pending)).every((result) => result.done)).toBe(true);
+    expect(() => fanout.close()).not.toThrow();
+  });
+
   it("disconnects slow subscribers without affecting other subscribers", async () => {
     const events: string[] = [];
     const fanout = new TerminalFanout({