perf(gateway): per-transcript broadcast lanes (#121104)

* perf(gateway): per-transcript broadcast lanes

Transcript-update broadcasts were serialized behind one global promise queue,
so one session's pending async seq read head-of-line blocked every other
session's session.message and sessions.changed delivery. Ordering is a
per-transcript contract, so key the queue by transcript identity and drop
drained lanes.

Per-message transcript usage stays derived: session.message payloads carry
live totalTokens/estimatedCostUsd as a dashboard API contract (#50101).

* fix(ui): target the Cloud Worker Desktop labs row in its toggle test

The row added in #120727 reused index 6, which belongs to Message audit
metadata, so the case toggled the audit row and asserted its patch. Labs
features render in registry order, and workerDesktop is the eighth entry.

src/gateway/server-session-events.test.ts

@@ -14,6 +14,7 @@ const sessionRow = vi.hoisted(() => ({
   agentRuntime: { id: "openclaw", source: "model" },
 }));
 const isEmbeddedAgentRunInProgressMock = vi.hoisted(() => vi.fn());
+const loadGatewaySessionRowMock = vi.hoisted(() => vi.fn());
 const projectChatDisplayMessageMock = vi.hoisted(() => vi.fn((message: unknown) => message));
 const loadAccessorSessionEntryReadOnlyMock = vi.hoisted(() => vi.fn());
 const loadGatewaySessionEntryReadOnlyMock = vi.hoisted(() => vi.fn());
@@ -32,7 +33,7 @@ vi.mock("./chat-display-projection.js", () => ({
 }));
 vi.mock("./session-utils.js", () => ({
   attachOpenClawTranscriptMeta: (message: unknown) => message,
-  loadGatewaySessionRow: () => sessionRow,
+  loadGatewaySessionRow: loadGatewaySessionRowMock,
   loadSessionEntry: () => ({ entry: undefined, storePath: "" }),
   loadSessionEntryReadOnly: loadGatewaySessionEntryReadOnlyMock,
 }));
@@ -102,6 +103,7 @@ describe("createTranscriptUpdateBroadcastHandler", () => {
     isEmbeddedAgentRunInProgressMock.mockReturnValue(false);
     loadAccessorSessionEntryReadOnlyMock.mockReturnValue(undefined);
     loadGatewaySessionEntryReadOnlyMock.mockReturnValue({ entry: undefined, storePath: "" });
+    loadGatewaySessionRowMock.mockReturnValue(sessionRow);
     readSessionMessageCountAsyncMock.mockResolvedValue(undefined);
     sessionRow.thinkingLevel = "ultra";
   });
@@ -550,9 +552,95 @@ describe("createTranscriptUpdateBroadcastHandler", () => {
     });
     expect(broadcastToConnIds.mock.calls[0]?.[1]).toMatchObject({ messageSeq: 7 });
   });
+
+  it("does not stall one session's broadcasts behind another session's pending seq read", async () => {
+    let releaseSlowCount: (value: number | undefined) => void = () => undefined;
+    readSessionMessageCountAsyncMock.mockImplementation((params: { sessionKey?: string }) =>
+      params.sessionKey === "agent:main:slow"
+        ? new Promise<number | undefined>((resolve) => {
+            releaseSlowCount = resolve;
+          })
+        : Promise.resolve(3),
+    );
+    loadAccessorSessionEntryReadOnlyMock.mockReturnValue({ sessionId: "sess-main" });
+    const { broadcastToConnIds, handler } = createHandler(false);
+
+    // No messageSeq: the slow lane blocks on its async transcript count.
+    const slowTask = handler({
+      message: { role: "assistant", content: [{ type: "text", text: "slow" }] },
+      messageId: "slow-1",
+      target: {
+        agentId: "main",
+        sessionId: "sess-slow",
+        sessionKey: "agent:main:slow",
+        storePath: "/tmp/slow-sessions.json",
+      },
+    });
+
+    await handler({
+      sessionFile: "/tmp/sess-main.jsonl",
+      sessionKey: "agent:main:main",
+      message: { role: "assistant", content: [{ type: "text", text: "fast" }] },
+      messageId: "fast-1",
+      messageSeq: 1,
+    });
+
+    // The independent lane broadcast completed while the slow lane is parked.
+    expect(broadcastToConnIds).toHaveBeenCalledTimes(1);
+    expect(broadcastToConnIds.mock.calls[0]?.[1]).toMatchObject({ messageId: "fast-1" });
+
+    releaseSlowCount(5);
+    await slowTask;
+    expect(broadcastToConnIds).toHaveBeenCalledTimes(2);
+    expect(broadcastToConnIds.mock.calls[1]?.[1]).toMatchObject({ messageId: "slow-1" });
+  });
+
+  it("preserves message order within one session lane", async () => {
+    let releaseFirstCount: (value: number | undefined) => void = () => undefined;
+    readSessionMessageCountAsyncMock.mockImplementationOnce(
+      () =>
+        new Promise<number | undefined>((resolve) => {
+          releaseFirstCount = resolve;
+        }),
+    );
+    loadAccessorSessionEntryReadOnlyMock.mockReturnValue({ sessionId: "sess-main" });
+    const { broadcastToConnIds, handler } = createHandler(false);
+
+    const firstTask = handler({
+      message: { role: "assistant", content: [{ type: "text", text: "first" }] },
+      messageId: "ordered-1",
+      target: {
+        agentId: "main",
+        sessionId: "sess-main",
+        sessionKey: "agent:main:main",
+        storePath: "/tmp/explicit-sessions.json",
+      },
+    });
+    const secondTask = handler({
+      sessionFile: "/tmp/sess-main.jsonl",
+      sessionKey: "agent:main:main",
+      message: { role: "assistant", content: [{ type: "text", text: "second" }] },
+      messageId: "ordered-2",
+      messageSeq: 2,
+    });
+
+    await Promise.resolve();
+    expect(broadcastToConnIds).not.toHaveBeenCalled();
+
+    releaseFirstCount(1);
+    await Promise.all([firstTask, secondTask]);
+    expect(broadcastToConnIds.mock.calls.map((call) => call[1]?.messageId)).toEqual([
+      "ordered-1",
+      "ordered-2",
+    ]);
+  });
 });
 
 describe("createLifecycleEventBroadcastHandler", () => {
+  beforeEach(() => {
+    loadGatewaySessionRowMock.mockReturnValue(sessionRow);
+  });
+
   it("projects swarm phase and log payload fields", () => {
     const broadcastToConnIds = vi.fn();
     const handler = createLifecycleEventBroadcastHandler({

src/gateway/server-session-events.ts

@@ -164,7 +164,10 @@ export function createTranscriptUpdateBroadcastHandler(params: {
   sessionMessageSubscribers: SessionMessageSubscribers;
   chatAbortControllers: Map<string, ChatAbortControllerEntry>;
 }) {
-  let broadcastQueue = Promise.resolve();
+  // Ordering is a per-transcript contract: subscribers merge each session's
+  // updates independently, so lanes keyed by transcript identity keep message
+  // order without one session's async seq reads stalling every other session.
+  const broadcastQueues = new Map<string, Promise<void>>();
   return (update: InternalSessionTranscriptUpdate): Promise<void> => {
     // Capture legacy ownership before the async queue can cross a same-id reset;
     // committed producer ownership always wins over a later session-store read.
@@ -174,10 +177,26 @@ export function createTranscriptUpdateBroadcastHandler(params: {
         ? readTranscriptUpdateLifecycleOwner(update)?.lifecycleRevision
         : undefined);
     const queuedUpdate = lifecycleRevision ? { ...update, lifecycleRevision } : update;
-    // Preserve transcript update order even when counting messages requires an
-    // async read from the session file.
-    const task = broadcastQueue.then(() => handleTranscriptUpdateBroadcast(params, queuedUpdate));
-    broadcastQueue = task.catch(() => undefined);
+    const laneKey =
+      normalizeOptionalString(update.target?.sessionKey) ??
+      normalizeOptionalString(update.sessionKey) ??
+      normalizeOptionalString(update.sessionFile) ??
+      "";
+    // Preserve transcript update order within the lane even when counting
+    // messages requires an async read from the session file.
+    const tail = broadcastQueues.get(laneKey) ?? Promise.resolve();
+    const task = tail.then(() => handleTranscriptUpdateBroadcast(params, queuedUpdate));
+    const settled = task.then(
+      () => undefined,
+      () => undefined,
+    );
+    broadcastQueues.set(laneKey, settled);
+    void settled.then(() => {
+      // Drop drained lanes so idle sessions do not accumulate map entries.
+      if (broadcastQueues.get(laneKey) === settled) {
+        broadcastQueues.delete(laneKey);
+      }
+    });
     return task;
   };
 }
@@ -320,6 +339,8 @@ async function handleTranscriptUpdateBroadcast(
       return;
     }
   }
+  // Message frames must keep transcript-derived live usage (dashboard API
+  // contract from #50101); the 64KB cap bounds the per-message tail read.
   const sessionRow = loadGatewaySessionRow(sessionKey, {
     agentId: routingAgentId,
     transcriptUsageMaxBytes: 64 * 1024,

ui/src/pages/labs/labs-page.test.ts

@@ -201,7 +201,7 @@ describe("LabsPage", () => {
     },
     {
       label: "Cloud Worker Desktop",
-      index: 6,
+      index: 7,
       sourceConfig: { cloudWorkers: { desktop: false } },
       expectedPatch: { cloudWorkers: { desktop: true } },
       note: "labs: update workerDesktop",