fix(server): bound Codex subagent progress events

apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts

@@ -34,6 +34,30 @@ const [CHILD_A, CHILD_B] = wireFixture.childThreadIds as [string, string];
  */
 function buildScript() {
   const captured = wireFixture.notifications;
+  const progressBurst = captured.flatMap((entry) => {
+    const item = (entry.params as { item?: { type?: string } }).item;
+    const threadId = (entry.params as { threadId?: string }).threadId;
+    if (entry.method === "thread/tokenUsage/updated" && threadId === CHILD_A) {
+      return [entry, entry, entry];
+    }
+    if (
+      (entry.method === "item/started" || entry.method === "item/completed") &&
+      item?.type === "collabAgentToolCall"
+    ) {
+      return [
+        entry,
+        {
+          ...entry,
+          params: {
+            ...entry.params,
+            threadId: CHILD_A,
+            turnId: `${CHILD_A}-turn-1`,
+          },
+        },
+      ];
+    }
+    return [entry];
+  });
   const extras = [
     {
       method: "item/completed",
@@ -66,15 +90,18 @@ function buildScript() {
   ];
   return {
     rootThreadId: ROOT,
-    notifications: [...captured.filter((entry) => entry.method !== "turn/completed"), ...extras],
+    notifications: [
+      ...progressBurst.filter((entry) => entry.method !== "turn/completed"),
+      ...extras,
+    ],
   };
 }
 
 const scriptPath = NodePath.join(import.meta.dirname, "../testFixtures/.collab-script.json");
 const peerPath = NodePath.join(import.meta.dirname, "../testFixtures/codexCollabMockPeer.sh");
 
 describe("CodexSessionRuntime collab integration", () => {
-  it.effect("replays the captured fan-out into synthetic agent events without child leaks", () =>
+  it.live("replays bounded child progress before terminal lifecycle without child leaks", () =>
     Effect.gen(function* () {
       // @effect-diagnostics-next-line preferSchemaOverJson:off
       NodeFS.writeFileSync(scriptPath, JSON.stringify(buildScript()), "utf8");
@@ -116,6 +143,47 @@ describe("CodexSessionRuntime collab integration", () => {
       );
       assert.isDefined(childTurnCompleted, "child A's turn completion becomes an agent event");
 
+      const childAProgress = events.filter(
+        (event) =>
+          (event.method === "collabAgent/item" || event.method === "collabAgent/tokenUsage") &&
+          (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_A,
+      );
+      const childAItems = childAProgress.filter((event) => event.method === "collabAgent/item");
+      const childAUsage = childAProgress.filter(
+        (event) => event.method === "collabAgent/tokenUsage",
+      );
+      assert.isAtLeast(childAItems.length, 1, "the child item lane survives coalescing");
+      assert.isBelow(childAItems.length, 2, "the two child item updates coalesce");
+      assert.isAtLeast(childAUsage.length, 1, "the child usage lane survives coalescing");
+      assert.isBelow(childAUsage.length, 6, "the six cumulative child usage updates coalesce");
+      const latestItem = childAItems.findLast(() => true);
+      assert.isDefined(latestItem);
+      assert.equal(
+        (latestItem.payload as { item?: { status?: string } }).item?.status,
+        "completed",
+        "the latest child item survives coalescing",
+      );
+      const latestUsage = childAUsage.findLast(() => true);
+      assert.isDefined(latestUsage);
+      assert.equal(
+        (latestUsage.payload as { tokenUsage?: { total?: { totalTokens?: number } } }).tokenUsage
+          ?.total?.totalTokens,
+        41_529,
+        "the latest cumulative child usage survives coalescing",
+      );
+      const childAIdleIndex = events.findIndex(
+        (event) =>
+          event.method === "collabAgent/statusChanged" &&
+          (event.payload as { agentThreadId?: string; status?: { type?: string } })
+            .agentThreadId === CHILD_A &&
+          (event.payload as { status?: { type?: string } }).status?.type === "idle",
+      );
+      assert.isAtLeast(childAIdleIndex, 0, "child A's terminal idle status becomes an agent event");
+      assert.isTrue(
+        childAProgress.every((event) => events.indexOf(event) < childAIdleIndex),
+        "latest child progress must be emitted before terminal lifecycle",
+      );
+
       const childClosed = events.find(
         (event) =>
           event.method === "collabAgent/closed" &&

apps/server/src/provider/Layers/CodexSessionRuntime.ts

@@ -16,6 +16,7 @@ import {
   ThreadId,
   TurnId,
 } from "@t3tools/contracts";
+import { makeKeyedCoalescingWorker } from "@t3tools/shared/KeyedCoalescingWorker";
 import { resolveSpawnCommand } from "@t3tools/shared/shell";
 import { normalizeModelSlug } from "@t3tools/shared/model";
 import * as Crypto from "effect/Crypto";
@@ -52,6 +53,7 @@ const BENIGN_ERROR_LOG_SNIPPETS = [
   "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back",
 ];
 const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const;
+const CODEX_COLLAB_PROGRESS_DEBOUNCE = "250 millis" as const;
 const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [
   "not found",
   "missing thread",
@@ -949,6 +951,46 @@ export const makeCodexSessionRuntime = (
         message,
       });
 
+    type CollabProgressEvent = Parameters<typeof emitEvent>[0];
+    interface PendingCollabProgress {
+      readonly item?: CollabProgressEvent;
+      readonly tokenUsage?: CollabProgressEvent;
+    }
+    const collabProgressWorker = yield* makeKeyedCoalescingWorker<
+      string,
+      PendingCollabProgress,
+      never,
+      never
+    >({
+      merge: (current, next) => ({
+        ...current,
+        ...(next.item ? { item: next.item } : {}),
+        ...(next.tokenUsage ? { tokenUsage: next.tokenUsage } : {}),
+      }),
+      process: (agentThreadId, progress, context) =>
+        Effect.gen(function* () {
+          // One scoped processor throttles normal progress across this
+          // session while per-child pending lanes merge. Terminal flushes
+          // bypass the delay so lifecycle events are never stuck behind it.
+          if (!context.flush) {
+            yield* Effect.sleep(CODEX_COLLAB_PROGRESS_DEBOUNCE);
+          }
+          if (progress.item) {
+            yield* emitEvent(progress.item);
+          }
+          if (progress.tokenUsage) {
+            yield* emitEvent(progress.tokenUsage);
+          }
+        }).pipe(
+          Effect.catch((cause) =>
+            Effect.logWarning("Failed to emit coalesced Codex subagent progress.", {
+              agentThreadId,
+              cause,
+            }),
+          ),
+        ),
+    }).pipe(Effect.provideService(Scope.Scope, runtimeScope));
+
     const settlePendingApprovals = (decision: ProviderApprovalDecision) =>
       Ref.get(pendingApprovalsRef).pipe(
         Effect.flatMap((pendingApprovals) =>
@@ -1138,6 +1180,7 @@ export const makeCodexSessionRuntime = (
             return true;
           }
           case "turn/completed":
+            yield* collabProgressWorker.flushKey(child.agentThreadId);
             yield* Ref.update(collabChildLiveTurnsRef, (current) => {
               const next = new Map(current);
               next.delete(child.agentThreadId);
@@ -1155,6 +1198,9 @@ export const makeCodexSessionRuntime = (
             });
             return true;
           case "thread/status/changed":
+            if (notification.params.status.type !== "active") {
+              yield* collabProgressWorker.flushKey(child.agentThreadId);
+            }
             yield* emitEvent({
               kind: "notification",
               threadId: options.threadId,
@@ -1167,34 +1213,39 @@ export const makeCodexSessionRuntime = (
             });
             return true;
           case "thread/tokenUsage/updated":
-            yield* emitEvent({
-              kind: "notification",
-              threadId: options.threadId,
-              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
-              method: "collabAgent/tokenUsage",
-              payload: {
-                ...childIdentity,
-                tokenUsage: notification.params.tokenUsage,
+            yield* collabProgressWorker.enqueue(child.agentThreadId, {
+              tokenUsage: {
+                kind: "notification",
+                threadId: options.threadId,
+                ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
+                method: "collabAgent/tokenUsage",
+                payload: {
+                  ...childIdentity,
+                  tokenUsage: notification.params.tokenUsage,
+                },
               },
             });
             return true;
           case "item/started":
           case "item/completed":
-            yield* emitEvent({
-              kind: "notification",
-              threadId: options.threadId,
-              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
-              method: "collabAgent/item",
-              payload: {
-                ...childIdentity,
-                item: notification.params.item,
+            yield* collabProgressWorker.enqueue(child.agentThreadId, {
+              item: {
+                kind: "notification",
+                threadId: options.threadId,
+                ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
+                method: "collabAgent/item",
+                payload: {
+                  ...childIdentity,
+                  item: notification.params.item,
+                },
               },
             });
             return true;
           case "thread/closed":
             // The child is gone: drop its live-turn entry so a later Stop
             // doesn't waste a turn/interrupt RPC on a closed thread before
             // reaching the parent (review finding).
+            yield* collabProgressWorker.flushKey(child.agentThreadId);
             yield* Ref.update(collabChildLiveTurnsRef, (current) => {
               const next = new Map(current);
               next.delete(child.agentThreadId);
@@ -1221,6 +1272,7 @@ export const makeCodexSessionRuntime = (
             if (willRetry) {
               return true;
             }
+            yield* collabProgressWorker.flushKey(child.agentThreadId);
             yield* Ref.update(collabChildLiveTurnsRef, (current) => {
               const next = new Map(current);
               next.delete(child.agentThreadId);
@@ -1733,12 +1785,16 @@ export const makeCodexSessionRuntime = (
         status: "closed",
         activeTurnId: undefined,
       });
+      // Runtime close cancels pending best-effort progress before publishing
+      // the terminal session event. The worker shares this scope with the
+      // notification consumer, so neither can offer after the event queue is
+      // shut down below.
+      yield* Scope.close(runtimeScope, Exit.void);
       yield* emitSessionEvent("session/closed", "Session stopped").pipe(
         Effect.catch((cause) =>
           Effect.logError("Failed to emit Codex session closed event.", { cause }),
         ),
       );
-      yield* Scope.close(runtimeScope, Exit.void);
       yield* Queue.shutdown(serverNotifications);
       yield* Queue.shutdown(events);
     });

packages/shared/src/KeyedCoalescingWorker.test.ts

@@ -2,6 +2,9 @@ import { it } from "@effect/vitest";
 import { describe, expect } from "vite-plus/test";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+import * as Fiber from "effect/Fiber";
+import * as Scope from "effect/Scope";
 
 import { makeKeyedCoalescingWorker } from "./KeyedCoalescingWorker.ts";
 
@@ -94,4 +97,76 @@ describe("makeKeyedCoalescingWorker", () => {
       }),
     ),
   );
+
+  it.live("flushes one key without waiting behind unrelated queued work", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const processed: string[] = [];
+        const slowStarted = yield* Deferred.make<void>();
+        const releaseSlow = yield* Deferred.make<void>();
+        const urgentStarted = yield* Deferred.make<void>();
+
+        const worker = yield* makeKeyedCoalescingWorker<string, string, never, never>({
+          merge: (_current, next) => next,
+          process: (key, value, context) =>
+            Effect.gen(function* () {
+              processed.push(`${key}:${value}:${context.flush ? "flush" : "normal"}`);
+              if (key === "slow") {
+                yield* Deferred.succeed(slowStarted, undefined).pipe(Effect.orDie);
+                yield* Deferred.await(releaseSlow);
+              }
+              if (key === "urgent") {
+                yield* Deferred.succeed(urgentStarted, undefined).pipe(Effect.orDie);
+              }
+            }),
+        });
+
+        yield* worker.enqueue("slow", "first");
+        yield* Deferred.await(slowStarted);
+        yield* worker.enqueue("queued", "second");
+        yield* worker.enqueue("urgent", "terminal");
+
+        const flushed = yield* Effect.forkChild(worker.flushKey("urgent"));
+        yield* Deferred.await(urgentStarted);
+
+        expect(processed).toEqual(["slow:first:normal", "urgent:terminal:flush"]);
+
+        yield* Fiber.join(flushed);
+        yield* Deferred.succeed(releaseSlow, undefined);
+        yield* worker.drainKey("queued");
+
+        expect(processed).toEqual([
+          "slow:first:normal",
+          "urgent:terminal:flush",
+          "queued:second:normal",
+        ]);
+      }),
+    ),
+  );
+
+  it.live("stops active processing when its owning scope closes", () =>
+    Effect.gen(function* () {
+      const workerScope = yield* Scope.make();
+      const started = yield* Deferred.make<void>();
+      const interrupted = yield* Deferred.make<void>();
+      const neverRelease = yield* Deferred.make<void>();
+
+      const worker = yield* makeKeyedCoalescingWorker<string, string, never, never>({
+        merge: (_current, next) => next,
+        process: () =>
+          Effect.gen(function* () {
+            yield* Deferred.succeed(started, undefined).pipe(Effect.orDie);
+            yield* Deferred.await(neverRelease);
+          }).pipe(
+            Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined).pipe(Effect.orDie)),
+          ),
+      }).pipe(Effect.provideService(Scope.Scope, workerScope));
+
+      yield* worker.enqueue("child", "progress");
+      yield* Deferred.await(started);
+      yield* Scope.close(workerScope, Exit.void);
+
+      expect(yield* Deferred.isDone(interrupted)).toBe(true);
+    }),
+  );
 });