Add multiplexed terminal stream slots

2026-03-23 12:36:35 +07:00
parent ed3ed78ec3
commit ea41dbc8e7
8 changed files with 430 additions and 154 deletions
@@ -1257,6 +1257,7 @@ describe("DaemonClient", () => {
        type: "subscribe_terminal_response",
        payload: {
          terminalId: "term-1",
+          slot: 11,
          error: null,
          requestId: "sub-1",
        },
@@ -1267,6 +1268,7 @@ describe("DaemonClient", () => {
    mock.triggerMessage(
      encodeTerminalStreamFrame({
        opcode: TerminalStreamOpcode.Output,
+        slot: 11,
        payload: new TextEncoder().encode("hello"),
      }),
    );
@@ -1276,7 +1278,7 @@ describe("DaemonClient", () => {
    unsubscribe();
  });

-  test("emits snapshot events for the active terminal stream", async () => {
+  test("emits snapshot events for the subscribed terminal stream", async () => {
    const logger = createMockLogger();
    const mock = createMockTransport();

@@ -1307,6 +1309,7 @@ describe("DaemonClient", () => {
        type: "subscribe_terminal_response",
        payload: {
          terminalId: "term-1",
+          slot: 12,
          error: null,
          requestId: "sub-2",
        },
@@ -1324,6 +1327,7 @@ describe("DaemonClient", () => {
    mock.triggerMessage(
      encodeTerminalStreamFrame({
        opcode: TerminalStreamOpcode.Snapshot,
+        slot: 12,
        payload: encodeTerminalSnapshotPayload(state),
      }),
    );
@@ -1331,7 +1335,7 @@ describe("DaemonClient", () => {
    expect(snapshots).toEqual([state]);
  });

-  test("sends input and resize frames for the active terminal", async () => {
+  test("sends input and resize frames for the subscribed terminal slot", async () => {
    const logger = createMockLogger();
    const mock = createMockTransport();

@@ -1354,6 +1358,7 @@ describe("DaemonClient", () => {
        type: "subscribe_terminal_response",
        payload: {
          terminalId: "term-1",
+          slot: 13,
          error: null,
          requestId: "sub-3",
        },
@@ -1376,14 +1381,105 @@ describe("DaemonClient", () => {
    const resizeFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[1])!);

    expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input);
+    expect(inputFrame?.slot).toBe(13);
    expect(new TextDecoder().decode(inputFrame?.payload ?? new Uint8Array())).toBe("echo hello\r");
    expect(resizeFrame?.opcode).toBe(TerminalStreamOpcode.Resize);
+    expect(resizeFrame?.slot).toBe(13);
    expect(decodeTerminalResizePayload(resizeFrame?.payload ?? new Uint8Array())).toEqual({
      rows: 24,
      cols: 80,
    });
  });

+  test("routes concurrent terminal stream frames by slot", async () => {
+    const logger = createMockLogger();
+    const mock = createMockTransport();
+
+    const client = new DaemonClient({
+      url: "ws://test",
+      clientId: "clsk_unit_test",
+      logger,
+      reconnect: { enabled: false },
+      transportFactory: () => mock.transport,
+    });
+    clients.push(client);
+
+    const connectPromise = client.connect();
+    mock.triggerOpen();
+    await connectPromise;
+
+    const seen: string[] = [];
+    client.onTerminalStreamEvent((event) => {
+      if (event.type !== "output") {
+        return;
+      }
+      seen.push(`${event.terminalId}:${new TextDecoder().decode(event.data)}`);
+    });
+
+    const subscribeFirstPromise = client.subscribeTerminal("term-1", "sub-multi-1");
+    mock.triggerMessage(
+      wrapSessionMessage({
+        type: "subscribe_terminal_response",
+        payload: {
+          terminalId: "term-1",
+          slot: 21,
+          error: null,
+          requestId: "sub-multi-1",
+        },
+      }),
+    );
+    await subscribeFirstPromise;
+
+    const subscribeSecondPromise = client.subscribeTerminal("term-2", "sub-multi-2");
+    mock.triggerMessage(
+      wrapSessionMessage({
+        type: "subscribe_terminal_response",
+        payload: {
+          terminalId: "term-2",
+          slot: 22,
+          error: null,
+          requestId: "sub-multi-2",
+        },
+      }),
+    );
+    await subscribeSecondPromise;
+    mock.sent.length = 0;
+
+    mock.triggerMessage(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        slot: 22,
+        payload: new TextEncoder().encode("beta"),
+      }),
+    );
+    mock.triggerMessage(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        slot: 21,
+        payload: new TextEncoder().encode("alpha"),
+      }),
+    );
+
+    client.sendTerminalInput("term-2", {
+      type: "input",
+      data: "echo beta\r",
+    });
+    client.sendTerminalInput("term-1", {
+      type: "resize",
+      rows: 10,
+      cols: 20,
+    });
+
+    const inputFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[0])!);
+    const resizeFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[1])!);
+
+    expect(seen).toEqual(["term-2:beta", "term-1:alpha"]);
+    expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input);
+    expect(inputFrame?.slot).toBe(22);
+    expect(resizeFrame?.opcode).toBe(TerminalStreamOpcode.Resize);
+    expect(resizeFrame?.slot).toBe(21);
+  });
+
  test("ignores terminal stream frames after terminal_stream_exit", async () => {
    const logger = createMockLogger();
    const mock = createMockTransport();
@@ -1415,6 +1511,7 @@ describe("DaemonClient", () => {
        type: "subscribe_terminal_response",
        payload: {
          terminalId: "term-1",
+          slot: 14,
          error: null,
          requestId: "sub-4",
        },
@@ -1425,6 +1522,7 @@ describe("DaemonClient", () => {
    mock.triggerMessage(
      encodeTerminalStreamFrame({
        opcode: TerminalStreamOpcode.Output,
+        slot: 14,
        payload: new TextEncoder().encode("before-exit"),
      }),
    );
@@ -1442,6 +1540,7 @@ describe("DaemonClient", () => {
    mock.triggerMessage(
      encodeTerminalStreamFrame({
        opcode: TerminalStreamOpcode.Output,
+        slot: 14,
        payload: new TextEncoder().encode("after-exit"),
      }),
    );
@@ -419,7 +419,8 @@ export class DaemonClient {
    { cwd: string; compare: { mode: "uncommitted" | "base"; baseRef?: string } }
  >();
  private terminalDirectorySubscriptions = new Set<string>();
-  private activeTerminalId: string | null = null;
+  private terminalSlots = new Map<string, number>();
+  private slotTerminals = new Map<number, string>();
  private readonly terminalStreamListeners = new Set<(event: TerminalStreamEvent) => void>();
  private logger: Logger;
  private pendingSendQueue: PendingSend[] = [];
@@ -659,7 +660,7 @@ export class DaemonClient {
    this.disposeTransport(1000, "Client closed");
    this.clearWaiters(new Error("Daemon client closed"));
    this.rejectPendingSendQueue(new Error("Daemon client closed"));
-    this.activeTerminalId = null;
+    this.clearTerminalSlots();
    this.lastServerInfoMessage = null;
    this.updateConnectionState(
      { status: "disposed" },
@@ -2613,15 +2614,13 @@ export class DaemonClient {
      options: { skipQueue: true },
    });
    if (!payload.error) {
-      this.activeTerminalId = terminalId;
+      this.setTerminalSlot(terminalId, payload.slot);
    }
    return payload;
  }

  unsubscribeTerminal(terminalId: string): void {
-    if (this.activeTerminalId === terminalId) {
-      this.activeTerminalId = null;
-    }
+    this.removeTerminalSlot(terminalId);
    this.sendSessionMessage({
      type: "unsubscribe_terminal_request",
      terminalId,
@@ -2629,11 +2628,13 @@ export class DaemonClient {
  }

  sendTerminalInput(terminalId: string, message: TerminalInput["message"]): void {
-    if (this.activeTerminalId === terminalId) {
+    const slot = this.terminalSlots.get(terminalId);
+    if (typeof slot === "number") {
      if (message.type === "input") {
        this.sendBinaryFrame(
          encodeTerminalStreamFrame({
            opcode: TerminalStreamOpcode.Input,
+            slot,
            payload: encodeUtf8String(message.data),
          }),
        );
@@ -2643,6 +2644,7 @@ export class DaemonClient {
        this.sendBinaryFrame(
          encodeTerminalStreamFrame({
            opcode: TerminalStreamOpcode.Resize,
+            slot,
            payload: encodeTerminalResizePayload({
              rows: message.rows,
              cols: message.cols,
@@ -2711,6 +2713,37 @@ export class DaemonClient {
    return requestId ?? crypto.randomUUID();
  }

+  private setTerminalSlot(terminalId: string, slot: number): void {
+    const existingTerminalId = this.slotTerminals.get(slot);
+    if (existingTerminalId && existingTerminalId !== terminalId) {
+      this.terminalSlots.delete(existingTerminalId);
+    }
+
+    const existingSlot = this.terminalSlots.get(terminalId);
+    if (typeof existingSlot === "number" && existingSlot !== slot) {
+      this.slotTerminals.delete(existingSlot);
+    }
+
+    this.terminalSlots.set(terminalId, slot);
+    this.slotTerminals.set(slot, terminalId);
+  }
+
+  private removeTerminalSlot(terminalId: string): void {
+    const slot = this.terminalSlots.get(terminalId);
+    if (typeof slot !== "number") {
+      return;
+    }
+    this.terminalSlots.delete(terminalId);
+    if (this.slotTerminals.get(slot) === terminalId) {
+      this.slotTerminals.delete(slot);
+    }
+  }
+
+  private clearTerminalSlots(): void {
+    this.terminalSlots.clear();
+    this.slotTerminals.clear();
+  }
+
  getLastServerInfoMessage(): ServerInfoStatusPayload | null {
    return this.lastServerInfoMessage;
  }
@@ -2840,7 +2873,7 @@ export class DaemonClient {
  }

  private handleBinaryFrame(frame: TerminalStreamFrame): void {
-    const terminalId = this.activeTerminalId;
+    const terminalId = this.slotTerminals.get(frame.slot);
    if (!terminalId) {
      return;
    }
@@ -2934,7 +2967,7 @@ export class DaemonClient {
    // and responses from the previous connection will never arrive.
    this.clearWaiters(new Error(reason ?? "Connection lost"));
    this.rejectPendingSendQueue(new Error(reason ?? "Connection lost"));
-    this.activeTerminalId = null;
+    this.clearTerminalSlots();
    this.lastServerInfoMessage = null;

    if (wasDisposed) {
@@ -2989,9 +3022,7 @@ export class DaemonClient {
    }

    if (msg.type === "terminal_stream_exit") {
-      if (this.activeTerminalId === msg.payload.terminalId) {
-        this.activeTerminalId = null;
-      }
+      this.removeTerminalSlot(msg.payload.terminalId);
    }

    if (this.rawMessageListeners.size > 0) {