From ab4ce5b06dcc2583ab49d5c25d71fbcadb4e0f06 Mon Sep 17 00:00:00 2001 From: Carl Gabel Date: Sat, 29 Aug 2026 22:21:10 +1000 Subject: [PATCH] perf(server): coalesce thread WS events into batched frames --- apps/server/src/server.test.ts | 8 +++- apps/server/src/ws.ts | 23 ++++++++++- ...nce-pull-v0.0.34-nightly-20260819-STATE.md | 41 +++++++++++++------ 3 files changed, 57 insertions(+), 15 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 29d1ce41e415..186e94b4256b 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -6949,6 +6949,9 @@ it.layer(NodeServices.layer)("server router seam", (it) => { // attach EAGERLY (subscribeDomainEvents) before the snapshot read, or an event // committed during the snapshot window lands on neither leg. The shell path was // hardened for this; the thread path was not. Deferred-gated for determinism. + // Runs under TestClock.withLive because the live leg is now coalesced into + // batched frames (see coalesceThreadStream): the group's flush window is a + // real sleep, which virtual time never advances past. it.effect("delivers a thread event published during snapshot load after the snapshot", () => Effect.gen(function* () { const eventPubSub = yield* PubSub.unbounded(); @@ -6998,7 +7001,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { assert.equal(items[0]?.kind, "snapshot"); assert.equal(items[1]?.kind, "event"); assert.equal(items[1]?.kind === "event" ? items[1].event.sequence : null, 500); - }).pipe(Effect.provide(NodeHttpServer.layerTest)), + }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); // loom: the thread completion marker, re-homed from upstream 8e3467fe6 (its @@ -7007,6 +7010,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { // same queue as the live events, so anything buffered during the snapshot/ // catch-up window is emitted BEFORE it. A marker merely concatenated ahead of // the live leg would overtake those events and claim synchronisation too early. + // Runs under TestClock.withLive for the coalescing window (see above). it.effect("emits the thread completion marker after events buffered during catch-up", () => Effect.gen(function* () { const eventPubSub = yield* PubSub.unbounded(); @@ -7064,7 +7068,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { items.map((item) => (item.kind === "event" ? `event:${item.event.sequence}` : item.kind)), ["event:11", "event:60", "synchronized"], ); - }).pipe(Effect.provide(NodeHttpServer.layerTest)), + }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); it.effect("omits the thread completion marker when it was not requested", () => diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 5c2f911ce697..d7700bcdacca 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -189,6 +189,27 @@ const SHELL_CATCHUP_MAX_EVENTS = 500; // exists. See plans/2026-07-28-thread-catchup-silent-truncation.md. const THREAD_CATCHUP_MAX_EVENTS = 500; +// loom: batch the thread subscription's live leg into multi-value RPC frames. +// One turn emits ~50 thread detail events a few milliseconds apart, and one +// frame each spends ~50 WebSocket envelopes (plus their per-frame overhead) on +// data that fits in a dozen. Re-homed from upstream's pull-5 burst coalescing, +// which applies exactly this `groupedWithin` shape on its shell leg +// (SHELL_COALESCE_*). The thread leg only *batches*: unlike the shell leg it +// must never collapse two events into one, because the client applies every +// activity item, so each group is re-emitted whole and in order as one chunk. +// The window bounds the worst-case added latency for an activity item to reach +// the UI (imperceptible next to a turn) and is what makes the batching +// deterministic rather than a function of how fast the server happens to be. +const THREAD_COALESCE_WINDOW = Duration.millis(50); +const THREAD_COALESCE_MAX_CHUNK = 512; +const coalesceThreadStream = ( + stream: Stream.Stream, +): Stream.Stream => + stream.pipe( + Stream.groupedWithin(THREAD_COALESCE_MAX_CHUNK, THREAD_COALESCE_WINDOW), + Stream.flatMap((items) => Stream.fromIterable(items)), + ); + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const EDITOR_DISCOVERY_TIMEOUT = Duration.seconds(5); @@ -1529,7 +1550,7 @@ const makeWsRpcLayer = ( yield* Effect.forkScoped( reasoningStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), ); - const bufferedLiveStream = Stream.fromQueue(liveBuffer); + const bufferedLiveStream = coalesceThreadStream(Stream.fromQueue(liveBuffer)); // loom: re-homed from upstream 8e3467fe6, whose server-side emission // was lost in the fork's upstream-rehome (777bd20f8) while the diff --git a/docs/upstream-sync/24-cadence-pull-v0.0.34-nightly-20260819-STATE.md b/docs/upstream-sync/24-cadence-pull-v0.0.34-nightly-20260819-STATE.md index 114b6dacb98b..78d13407c136 100644 --- a/docs/upstream-sync/24-cadence-pull-v0.0.34-nightly-20260819-STATE.md +++ b/docs/upstream-sync/24-cadence-pull-v0.0.34-nightly-20260819-STATE.md @@ -411,16 +411,16 @@ lands upstream migrations. `VACUUM INTO` copy of `~/.t3/cockpit/userdata/state.s None of these is a typecheck or gate failure; each is recorded so a reviewer can tell them apart from new breakage. -| file | n | shape | -| ----------------------------------------------------------- | --- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `src/git/GitManager.test.ts` | 11 | PR-worktree materialisation against real git fixtures: the fakes have no `refs/pull/N/head` and one repo has no remote, so upstream's newer fetch path (plus the `remoteExists` guard port) fails to materialise. Fixture work, not a code fix. | -| `src/server.test.ts` | 1 | **performance budget**: measured-turn WebSocket messages 52 vs max 21, wire bytes 8046 vs 8000, for both codex and claudeAgent. Diagnosed at the review gate (see "Review round 1" below): **not** a double-publish. Awaiting a human re-baseline decision. | -| `test/ActivityPayloadProjection.test.ts` | 1 | asserts the mobile lazy `getFullDetail`/`getCopyText` API, which is on the ratified **deferred** ledger; an orphan of that wave. | -| `src/vcs/GitVcsDriverCore.test.ts` | 2 | ref-snapshot fixtures. | -| `src/vcs/VcsStatusBroadcaster.test.ts` | 1 | foreground-demand gating. | -| `src/terminal/Manager.test.ts` | 1 | polling-vs-registration ordering in upstream's adopted rewrite. | -| `src/orchestration/Layers/ProviderRuntimeIngestion.test.ts` | 1 | in-flight tool checkpoint cadence. | -| `src/provider/Layers/ClaudeAdapter.test.ts` | 1 | settle-on-stop pending user-input wait. | +| file | n | shape | +| ----------------------------------------------------------- | --- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `src/git/GitManager.test.ts` | 11 | PR-worktree materialisation against real git fixtures: the fakes have no `refs/pull/N/head` and one repo has no remote, so upstream's newer fetch path (plus the `remoteExists` guard port) fails to materialise. Fixture work, not a code fix. | +| `src/server.test.ts` | 0 | ~~**performance budget**: measured-turn WebSocket messages 52 vs max 21~~ — **RESOLVED** by re-homing burst coalescing onto the thread live leg; see "WS transfer budget" below. Passes at upstream's budget. | +| `test/ActivityPayloadProjection.test.ts` | 1 | asserts the mobile lazy `getFullDetail`/`getCopyText` API, which is on the ratified **deferred** ledger; an orphan of that wave. | +| `src/vcs/GitVcsDriverCore.test.ts` | 2 | ref-snapshot fixtures. | +| `src/vcs/VcsStatusBroadcaster.test.ts` | 1 | foreground-demand gating. | +| `src/terminal/Manager.test.ts` | 1 | polling-vs-registration ordering in upstream's adopted rewrite. | +| `src/orchestration/Layers/ProviderRuntimeIngestion.test.ts` | 1 | in-flight tool checkpoint cadence. | +| `src/provider/Layers/ClaudeAdapter.test.ts` | 1 | settle-on-stop pending user-input wait. | `apps/web` additionally has 7 failures in 3 files, all pre-existing merge fallout in areas this session did not touch: `MessagesTimeline.test.tsx` ×4 @@ -461,7 +461,23 @@ Typechecks clean, invisible to every gate; exactly the marker-less conflict clas this doctrine warns about. Loom's import and call are restored verbatim at their original position. -### WS transfer budget — diagnosed, still a human decision +### WS transfer budget — RESOLVED (coalescing re-homed) + +**Resolution (follow-up branch `t3code/ws-burst-coalescing`):** the human chose +coalescing over re-baselining, and `subscribeThread`'s live leg is now grouped +into batched RPC frames (`coalesceThreadStream` in `apps/server/src/ws.ts`: +`Stream.groupedWithin(512, 50 millis)` re-emitted whole as one chunk). Same +events, same order, ~4 per frame instead of 1: **52 → 18 (codex) / 17 +(claudeAgent) messages, 8024/8045 → 6898/6849 wire bytes**, both inside +upstream's 21/8000 budget. Unlike upstream's shell-leg coalescer this one never +collapses events — the thread client applies every activity item. The #115 +fail-loud mapper is untouched (it lives on the shell leg) and the #4079 marker +ordering is preserved because the marker still rides the same FIFO queue. Two +thread-subscription tests moved to `TestClock.withLive`, matching upstream's own +coalescing tests: the flush window is a real sleep that virtual time never +reaches. The original diagnosis below is kept for the record. + +### The original diagnosis — how it was measured The review gate instrumented the recorder and ran the budget test against both this branch and upstream tip: @@ -482,7 +498,8 @@ re-baseline loom's budget and log a coalescing follow-up (e.g. `Stream.groupedWithin` on the thread live leg - upstream applies exactly that on its shell leg, which loom's #115 fail-loud shell leg deliberately does not share), or fund the coalescing work now. The test keeps failing honestly until -then; it must not be skipped. +then; it must not be skipped. _(The human funded the work; see the resolution +above.)_ ### Rejected, with evidence