Skip to content

perf(server): coalesce thread WS events into batched frames #191

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
QuinRiva merged 1 commit into from
Aug 29, 2026
Merged

perf(server): coalesce thread WS events into batched frames #191

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions apps/server/src/server.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrchestrationEvent>();
Expand Down Expand Up @@ -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
Expand All @@ -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<OrchestrationEvent>();
Expand Down Expand Up @@ -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", () =>
Expand Down
23 changes: 22 additions & 1 deletion apps/server/src/ws.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = <E, R>(
stream: Stream.Stream<OrchestrationThreadStreamItem, E, R>,
): Stream.Stream<OrchestrationThreadStreamItem, E, R> =>
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);

Expand Down Expand Up @@ -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
Expand Down
41 changes: 29 additions & 12 deletions docs/upstream-sync/24-cadence-pull-v0.0.34-nightly-20260819-STATE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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

Expand Down
Loading