authorgravatar for adam.bogucki2018@gmail.comAdamulek123 <adam.bogucki2018@gmail.com> 2026-09-01 12:11:41+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 03:11:41-07:00
log7e4ce3bbb16c3cfa1ea756de917d8f9398e5999e
treef58821736bfdfc82fbfb3244a55649195f1b723e
parent8f1ef8b9eb72ac6cd8510364608e01dc7febf9c1
signature Signed by PGP key B5690EEEBB952194

perf(server): cut chatty tool-update frames by 90% (#8368)

5 files changed, 633 insertions(+), 15 deletions(-)

apps/server/src/orchestration/ActivityPayloadProjection.ts+2-5
......@@ -495,9 +495,6 @@ function toolLifecycleIdentity(activity: OrchestrationThreadActivity): string |
495495 * update within the turn — a later update belongs to a subsequent call that
496496 * reuses the same identity and is still in flight. Rows without a lifecycle
497497 * identity pass through, matching the clients, which never collapse them.
498 * Live `thread.activity-appended` events are untouched: updates still stream
499 * in real time and the completion supersedes them on the client as before.
500 *
501498 * Deliberate divergence from client collapse: clients fold only *adjacent*
502499 * lifecycle rows, so a superseded update separated from its completion by an
503500 * interleaved parallel call renders as its own row today, and this drop
......@@ -524,7 +521,7 @@ function dropSupersededToolUpdatedActivities(
524521 if (!identity) {
525522 continue;
526523 }
527 const key = `${activity.turnId ?? ""}${identity}`;
524 const key = `${activity.turnId ?? ""}\u0000${identity}`;
528525 const indices = completionIndicesByKey.get(key);
529526 if (indices) {
530527 indices.push(index);
......@@ -544,7 +541,7 @@ function dropSupersededToolUpdatedActivities(
544541 if (!identity) {
545542 return true;
546543 }
547 const indices = completionIndicesByKey.get(`${activity.turnId ?? ""}${identity}`);
544 const indices = completionIndicesByKey.get(`${activity.turnId ?? ""}\u0000${identity}`);
548545 return !indices?.some((completionIndex) => completionIndex > index);
549546 });
550547}
apps/server/src/orchestration/ThreadLiveEventCoalescer.test.ts created+171
......@@ -0,0 +1,171 @@
1import {
2 EventId,
3 MessageId,
4 ThreadId,
5 TurnId,
6 type OrchestrationEvent,
7 type OrchestrationThreadActivity,
8} from "@t3tools/contracts";
9import { it } from "@effect/vitest";
10import * as Clock from "effect/Clock";
11import * as Effect from "effect/Effect";
12import * as TestClock from "effect/testing/TestClock";
13import { describe, expect } from "vite-plus/test";
14
15import {
16 coalesceLiveToolUpdatedEvents,
17 makeThreadLiveEventCoalescer,
18} from "./ThreadLiveEventCoalescer.ts";
19
20const threadId = ThreadId.make("thread-coalescer-test");
21const turnId = TurnId.make("turn-coalescer-test");
22
23function makeToolActivity(
24 sequence: number,
25 options: {
26 readonly kind?: "tool.updated" | "tool.completed";
27 readonly toolCallId?: string;
28 readonly turnId?: TurnId;
29 } = {},
30): OrchestrationEvent {
31 const {
32 kind = "tool.updated",
33 toolCallId = "call-edit",
34 turnId: activityTurnId = turnId,
35 } = options;
36 const activity: OrchestrationThreadActivity = {
37 id: EventId.make(`activity-${sequence}`),
38 tone: "tool",
39 kind,
40 summary: "Editing app.ts",
41 payload: {
42 itemType: "file_change",
43 title: "Editing app.ts",
44 data: toolCallId ? { toolCallId } : {},
45 },
46 turnId: activityTurnId,
47 createdAt: "2026-01-01T00:00:01.000Z",
48 };
49 return {
50 sequence,
51 eventId: EventId.make(`event-${sequence}`),
52 aggregateKind: "thread",
53 aggregateId: threadId,
54 occurredAt: "2026-01-01T00:00:01.000Z",
55 commandId: null,
56 causationEventId: null,
57 correlationId: null,
58 metadata: {},
59 type: "thread.activity-appended",
60 payload: { threadId, activity },
61 };
62}
63
64function makeMessage(sequence: number): OrchestrationEvent {
65 return {
66 sequence,
67 eventId: EventId.make(`event-${sequence}`),
68 aggregateKind: "thread",
69 aggregateId: threadId,
70 occurredAt: "2026-01-01T00:00:02.000Z",
71 commandId: null,
72 causationEventId: null,
73 correlationId: null,
74 metadata: {},
75 type: "thread.message-sent",
76 payload: {
77 threadId,
78 messageId: MessageId.make(`message-${sequence}`),
79 role: "assistant",
80 text: "Still working",
81 turnId,
82 streaming: false,
83 createdAt: "2026-01-01T00:00:02.000Z",
84 updatedAt: "2026-01-01T00:00:02.000Z",
85 },
86 };
87}
88
89describe("ThreadLiveEventCoalescer", () => {
90 it("coalesces only calls with a stable toolCallId", () => {
91 const events = [
92 makeToolActivity(1, { toolCallId: "call-a" }),
93 makeToolActivity(2, { toolCallId: "call-b" }),
94 makeToolActivity(3, { toolCallId: "call-a" }),
95 ];
96
97 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3]);
98 });
99
100 it("preserves parallel same-label calls without a stable toolCallId", () => {
101 const events = [
102 makeToolActivity(1, { toolCallId: "" }),
103 makeToolActivity(2, { toolCallId: "" }),
104 makeToolActivity(3, { kind: "tool.completed", toolCallId: "" }),
105 ];
106
107 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2, 3]);
108 });
109
110 it("does not coalesce stable tool calls across turns", () => {
111 const events = [
112 makeToolActivity(1, { turnId: TurnId.make("turn-old") }),
113 makeToolActivity(2, { turnId: TurnId.make("turn-new") }),
114 ];
115
116 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2]);
117 });
118
119 it("flushes a stable update run before a completion boundary", () => {
120 const events = [
121 makeToolActivity(1),
122 makeToolActivity(2),
123 makeToolActivity(3, { kind: "tool.completed" }),
124 makeToolActivity(4),
125 ];
126
127 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3, 4]);
128 });
129
130 it.effect("flushes pending tool updates as soon as an unrelated event arrives", () =>
131 Effect.scoped(
132 Effect.gen(function* () {
133 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "500 millis" });
134 const startedAt = yield* Clock.currentTimeMillis;
135 yield* Effect.forEach(
136 Array.from({ length: 10 }, (_, index) => index + 2),
137 (sequence) =>
138 coalescer.offerAndWait({ kind: "event", event: makeToolActivity(sequence) }),
139 { discard: true },
140 );
141 yield* coalescer.offerAndWait({ kind: "event", event: makeMessage(12) });
142
143 expect(yield* Clock.currentTimeMillis).toBe(startedAt);
144 expect(
145 Array.from(yield* coalescer.takeAll).map((item) =>
146 item.kind === "event" ? item.event.sequence : item.kind,
147 ),
148 ).toEqual([11, 12]);
149 }),
150 ).pipe(Effect.provide(TestClock.layer())),
151 );
152
153 it.effect("flushes pending tool updates as soon as a synchronization marker arrives", () =>
154 Effect.scoped(
155 Effect.gen(function* () {
156 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "500 millis" });
157 const startedAt = yield* Clock.currentTimeMillis;
158 yield* coalescer.offerAndWait({ kind: "event", event: makeToolActivity(2) });
159 yield* coalescer.offerAndWait({ kind: "event", event: makeToolActivity(3) });
160 yield* coalescer.offerAndWait({ kind: "synchronized" });
161
162 expect(yield* Clock.currentTimeMillis).toBe(startedAt);
163 expect(
164 Array.from(yield* coalescer.takeAll).map((item) =>
165 item.kind === "event" ? item.event.sequence : item.kind,
166 ),
167 ).toEqual([3, "synchronized"]);
168 }),
169 ).pipe(Effect.provide(TestClock.layer())),
170 );
171});
apps/server/src/orchestration/ThreadLiveEventCoalescer.ts created+207
......@@ -0,0 +1,207 @@
1import type { OrchestrationEvent, OrchestrationThreadStreamItem } from "@t3tools/contracts";
2import * as Deferred from "effect/Deferred";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";
6import * as Predicate from "effect/Predicate";
7import * as Queue from "effect/Queue";
8import * as Semaphore from "effect/Semaphore";
9import * as Stream from "effect/Stream";
10
11import { projectActivityEvent } from "./ActivityPayloadProjection.ts";
12
13const COALESCE_WINDOW = Duration.millis(50);
14const MAX_PENDING_UPDATES = 512;
15
16export type ThreadLiveInput =
17 | { readonly kind: "event"; readonly event: OrchestrationEvent }
18 | { readonly kind: "synchronized" };
19
20function isToolUpdated(event: OrchestrationEvent): boolean {
21 return (
22 event.type === "thread.activity-appended" && event.payload.activity.kind === "tool.updated"
23 );
24}
25
26function asTrimmedString(value: unknown): string | null {
27 if (!Predicate.isString(value)) {
28 return null;
29 }
30 const trimmed = value.trim();
31 return trimmed.length > 0 ? trimmed : null;
32}
33
34function stableToolCallIdentity(event: OrchestrationEvent): string | null {
35 if (event.type !== "thread.activity-appended") {
36 return null;
37 }
38 const payload = event.payload.activity.payload;
39 if (!Predicate.isObject(payload)) {
40 return null;
41 }
42 const data = Predicate.isObject(payload.data) ? payload.data : null;
43 return asTrimmedString(payload.toolCallId) ?? asTrimmedString(data?.toolCallId);
44}
45
46/**
47 * Retain only the latest in-flight update for each stable tool-call id in a
48 * live run. Anonymous calls pass through because labels are not unique when
49 * tools execute in parallel. Survivors remain in sequence order.
50 */
51export function coalesceLiveToolUpdatedEvents(
52 events: ReadonlyArray<OrchestrationEvent>,
53): ReadonlyArray<OrchestrationEvent> {
54 const survivors: Array<OrchestrationEvent> = [];
55 let pendingUpdates: Array<OrchestrationEvent> = [];
56
57 const flushUpdates = () => {
58 const seen = new Set<string>();
59 const latestUpdates: Array<OrchestrationEvent> = [];
60 for (let index = pendingUpdates.length - 1; index >= 0; index -= 1) {
61 const event = pendingUpdates[index]!;
62 const identity = stableToolCallIdentity(event);
63 const activity =
64 event.type === "thread.activity-appended" ? event.payload.activity : undefined;
65 const key = identity ? `${activity?.turnId ?? ""}\u0000${identity}` : null;
66 if (key && seen.has(key)) {
67 continue;
68 }
69 if (key) {
70 seen.add(key);
71 }
72 latestUpdates.push(event);
73 }
74 latestUpdates.reverse();
75 survivors.push(...latestUpdates);
76 pendingUpdates = [];
77 };
78
79 for (const event of events) {
80 if (isToolUpdated(event)) {
81 pendingUpdates.push(event);
82 continue;
83 }
84 flushUpdates();
85 survivors.push(event);
86 }
87 flushUpdates();
88 return survivors;
89}
90
91export const makeThreadLiveEventCoalescer = Effect.fn("makeThreadLiveEventCoalescer")(
92 function* (options?: { readonly coalesceWindow?: Duration.Input }) {
93 const output = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
94 const input = yield* Queue.unbounded<{
95 readonly value: ThreadLiveInput;
96 readonly processed?: Deferred.Deferred<void>;
97 }>();
98 const mutex = yield* Semaphore.make(1);
99 const coalesceWindow = options?.coalesceWindow ?? COALESCE_WINDOW;
100 let pendingUpdates: Array<OrchestrationEvent> = [];
101 let windowGeneration = 0;
102 let windowFiber: Fiber.Fiber<void, never> | null = null;
103
104 const cancelWindow = Effect.fn("ThreadLiveEventCoalescer.cancelWindow")(function* () {
105 const fiber = windowFiber;
106 if (!fiber) {
107 return;
108 }
109 windowFiber = null;
110 yield* Fiber.interrupt(fiber);
111 });
112
113 const flushPending = Effect.fn("ThreadLiveEventCoalescer.flushPending")(function* (
114 boundary?: OrchestrationEvent,
115 ) {
116 const events = boundary ? [...pendingUpdates, boundary] : pendingUpdates;
117 pendingUpdates = [];
118 if (events.length === 0) {
119 return;
120 }
121 yield* Queue.offerAll(
122 output,
123 coalesceLiveToolUpdatedEvents(events).map((event) => ({
124 kind: "event" as const,
125 event: projectActivityEvent(event),
126 })),
127 );
128 });
129
130 const flushWindow = (generation: number) =>
131 Effect.sleep(coalesceWindow).pipe(
132 Effect.andThen(
133 mutex.withPermits(1)(
134 Effect.suspend(() => (generation === windowGeneration ? flushPending() : Effect.void)),
135 ),
136 ),
137 Effect.ensuring(
138 Effect.sync(() => {
139 if (generation === windowGeneration) {
140 windowFiber = null;
141 }
142 }),
143 ),
144 );
145
146 const process = Effect.fn("ThreadLiveEventCoalescer.process")(function* (
147 input: ThreadLiveInput,
148 ) {
149 yield* mutex.withPermits(1)(
150 Effect.gen(function* () {
151 if (input.kind === "event" && isToolUpdated(input.event)) {
152 pendingUpdates.push(input.event);
153 if (pendingUpdates.length === 1) {
154 const generation = ++windowGeneration;
155 windowFiber = yield* Effect.forkScoped(flushWindow(generation));
156 }
157 if (pendingUpdates.length >= MAX_PENDING_UPDATES) {
158 yield* cancelWindow();
159 windowGeneration += 1;
160 yield* flushPending();
161 }
162 return;
163 }
164
165 yield* cancelWindow();
166 windowGeneration += 1;
167 // A non-update event closes the run immediately. The coalescer keeps
168 // that boundary after the final update from the run.
169 if (input.kind === "event") {
170 yield* flushPending(input.event);
171 } else {
172 yield* flushPending();
173 yield* Queue.offer(output, { kind: "synchronized" });
174 }
175 }),
176 );
177 });
178
179 yield* Stream.fromQueue(input).pipe(
180 Stream.runForEach(({ value, processed }) =>
181 process(value).pipe(
182 Effect.andThen(processed ? Deferred.succeed(processed, undefined) : Effect.void),
183 ),
184 ),
185 Effect.forkScoped,
186 );
187
188 const offer = (value: ThreadLiveInput) => Queue.offer(input, { value }).pipe(Effect.asVoid);
189
190 // Synchronization callers wait for their marker to pass through the same
191 // ordered input queue before draining output produced ahead of it.
192 const offerAndWait = Effect.fn("ThreadLiveEventCoalescer.offerAndWait")(function* (
193 value: ThreadLiveInput,
194 ) {
195 const processed = yield* Deferred.make<void>();
196 yield* Queue.offer(input, { value, processed });
197 yield* Deferred.await(processed);
198 });
199
200 return {
201 offer,
202 offerAndWait,
203 stream: Stream.fromQueue(output),
204 takeAll: Queue.takeAll(output),
205 } as const;
206 },
207);
apps/server/src/server.test.ts+240
......@@ -19,6 +19,7 @@ import {
1919 ExternalLauncherCommandNotFoundError,
2020 OrchestrationThreadDetailSnapshot,
2121 type OrchestrationThreadStreamItem,
22 type OrchestrationThreadActivity,
2223 type OrchestrationThreadShell,
2324 TerminalNotRunningError,
2425 type OrchestrationCommand,
......@@ -30,6 +31,7 @@ import {
3031 ProviderInstanceId,
3132 ResolvedKeybindingRule,
3233 ThreadId,
34 TurnId,
3335 WS_METHODS,
3436 WsRpcGroup,
3537 EditorId,
......@@ -197,6 +199,44 @@ const defaultModelSelection = {
197199 instanceId: ProviderInstanceId.make("codex"),
198200 model: "gpt-5-codex",
199201} as const;
202
203const makeLiveToolActivityEvent = (
204 sequence: number,
205 kind: "tool.updated" | "tool.completed" = "tool.updated",
206 options: {
207 readonly toolCallId?: string;
208 readonly title?: string;
209 readonly path?: string;
210 } = {},
211): Extract<OrchestrationEvent, { type: "thread.activity-appended" }> => {
212 const { toolCallId = "call-edit", title = "Editing app.ts", path = "src/app.ts" } = options;
213 const activity: OrchestrationThreadActivity = {
214 id: EventId.make(`activity-${sequence}`),
215 tone: "tool",
216 kind,
217 summary: title,
218 payload: {
219 itemType: "file_change",
220 title,
221 data: { toolCallId, path },
222 },
223 turnId: TurnId.make("turn-edit"),
224 createdAt: "2026-01-01T00:00:01.000Z",
225 };
226 return {
227 sequence,
228 eventId: EventId.make(`event-tool-${sequence}`),
229 aggregateKind: "thread",
230 aggregateId: defaultThreadId,
231 occurredAt: "2026-01-01T00:00:01.000Z",
232 commandId: null,
233 causationEventId: null,
234 correlationId: null,
235 metadata: {},
236 type: "thread.activity-appended",
237 payload: { threadId: defaultThreadId, activity },
238 };
239};
200240const testEnvironmentDescriptor = {
201241 environmentId: EnvironmentId.make("environment-test"),
202242 label: "Test environment",
......@@ -6918,6 +6958,206 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
69186958 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
69196959 );
69206960
6961 it.effect("coalesces buffered live tool updates to the latest state", () =>
6962 Effect.gen(function* () {
6963 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
6964 const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>();
6965
6966 yield* buildAppUnderTest({
6967 layers: {
6968 orchestrationEngine: {
6969 streamDomainEvents: Stream.fromPubSub(liveEvents),
6970 },
6971 projectionSnapshotQuery: {
6972 getThreadDetailSnapshot: () =>
6973 Effect.gen(function* () {
6974 yield* Effect.sleep("25 millis");
6975 yield* PubSub.publishAll(liveEvents, [
6976 makeLiveToolActivityEvent(2),
6977 makeLiveToolActivityEvent(3),
6978 makeLiveToolActivityEvent(4),
6979 ]);
6980 return Option.some({ snapshotSequence: 1, thread });
6981 }),
6982 },
6983 },
6984 });
6985
6986 const wsUrl = yield* getWsServerUrl("/ws");
6987 const items = yield* Effect.scoped(
6988 withWsRpcClient(wsUrl, (client) =>
6989 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
6990 threadId: defaultThreadId,
6991 }).pipe(Stream.take(2), Stream.runCollect),
6992 ),
6993 ).pipe(Effect.timeout("2 seconds"));
6994
6995 assert.equal(items[0]?.kind, "snapshot");
6996 assert.equal(items[1]?.kind, "event");
6997 assert.equal(items[1]?.kind === "event" ? items[1].event.sequence : null, 4);
6998 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
6999 );
7000
7001 it.effect("flushes more than one tool chunk before the synchronization marker", () =>
7002 Effect.gen(function* () {
7003 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
7004 const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>();
7005
7006 yield* buildAppUnderTest({
7007 layers: {
7008 orchestrationEngine: {
7009 streamDomainEvents: Stream.fromPubSub(liveEvents),
7010 },
7011 projectionSnapshotQuery: {
7012 getThreadDetailSnapshot: () =>
7013 Effect.gen(function* () {
7014 yield* Effect.sleep("25 millis");
7015 yield* PubSub.publishAll(liveEvents, [
7016 ...Array.from({ length: 512 }, (_, index) =>
7017 makeLiveToolActivityEvent(index + 2),
7018 ),
7019 makeLiveToolActivityEvent(514, "tool.updated", {
7020 toolCallId: "call-read",
7021 title: "Reading server.test.ts",
7022 path: "apps/server/src/server.test.ts",
7023 }),
7024 ]);
7025 return Option.some({ snapshotSequence: 1, thread });
7026 }),
7027 },
7028 },
7029 });
7030
7031 const wsUrl = yield* getWsServerUrl("/ws");
7032 const items = yield* Effect.scoped(
7033 withWsRpcClient(wsUrl, (client) =>
7034 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
7035 threadId: defaultThreadId,
7036 requestCompletionMarker: true,
7037 }).pipe(Stream.take(4), Stream.runCollect),
7038 ),
7039 ).pipe(Effect.timeout("2 seconds"));
7040
7041 assert.equal(items[0]?.kind, "snapshot");
7042 assert.deepEqual(
7043 items.slice(1, 3).map((item) => {
7044 assert.equal(item?.kind, "event");
7045 if (item?.kind !== "event" || item.event.type !== "thread.activity-appended") {
7046 return null;
7047 }
7048 return {
7049 sequence: item.event.sequence,
7050 summary: item.event.payload.activity.summary,
7051 payload: item.event.payload.activity.payload,
7052 };
7053 }),
7054 [
7055 {
7056 sequence: 513,
7057 summary: "Editing app.ts",
7058 payload: {
7059 itemType: "file_change",
7060 title: "Editing app.ts",
7061 data: {
7062 files: [{ path: "src/app.ts" }],
7063 toolCallId: "call-edit",
7064 },
7065 },
7066 },
7067 {
7068 sequence: 514,
7069 summary: "Reading server.test.ts",
7070 payload: {
7071 itemType: "file_change",
7072 title: "Reading server.test.ts",
7073 data: {
7074 files: [{ path: "apps/server/src/server.test.ts" }],
7075 toolCallId: "call-read",
7076 },
7077 },
7078 },
7079 ],
7080 );
7081 assert.deepEqual(items[3], { kind: "synchronized" });
7082 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
7083 );
7084
7085 it.effect("flushes a tool update before an interleaved message", () =>
7086 Effect.gen(function* () {
7087 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
7088 const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>();
7089 const messageEvent = {
7090 sequence: 3,
7091 eventId: EventId.make("event-interleaved-message"),
7092 aggregateKind: "thread",
7093 aggregateId: defaultThreadId,
7094 occurredAt: "2026-01-01T00:00:02.000Z",
7095 commandId: null,
7096 causationEventId: null,
7097 correlationId: null,
7098 metadata: {},
7099 type: "thread.message-sent",
7100 payload: {
7101 threadId: defaultThreadId,
7102 messageId: MessageId.make("message-interleaved"),
7103 role: "assistant",
7104 text: "Still working",
7105 turnId: TurnId.make("turn-edit"),
7106 streaming: false,
7107 createdAt: "2026-01-01T00:00:02.000Z",
7108 updatedAt: "2026-01-01T00:00:02.000Z",
7109 },
7110 } satisfies Extract<OrchestrationEvent, { type: "thread.message-sent" }>;
7111
7112 yield* buildAppUnderTest({
7113 layers: {
7114 orchestrationEngine: {
7115 streamDomainEvents: Stream.fromPubSub(liveEvents),
7116 },
7117 projectionSnapshotQuery: {
7118 getThreadDetailSnapshot: () =>
7119 Effect.gen(function* () {
7120 yield* Effect.sleep("25 millis");
7121 yield* PubSub.publishAll(liveEvents, [
7122 makeLiveToolActivityEvent(2),
7123 messageEvent,
7124 makeLiveToolActivityEvent(4, "tool.completed"),
7125 ]);
7126 return Option.some({ snapshotSequence: 1, thread });
7127 }),
7128 },
7129 },
7130 });
7131
7132 const wsUrl = yield* getWsServerUrl("/ws");
7133 const items = yield* Effect.scoped(
7134 withWsRpcClient(wsUrl, (client) =>
7135 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
7136 threadId: defaultThreadId,
7137 }).pipe(Stream.take(4), Stream.runCollect),
7138 ),
7139 ).pipe(Effect.timeout("2 seconds"));
7140
7141 assert.equal(items[0]?.kind, "snapshot");
7142 assert.deepEqual(
7143 items
7144 .slice(1)
7145 .map((item) => (item.kind === "event" ? [item.event.sequence, item.event.type] : null)),
7146 [
7147 [2, "thread.activity-appended"],
7148 [3, "thread.message-sent"],
7149 [4, "thread.activity-appended"],
7150 ],
7151 );
7152 assert.equal(
7153 items[3]?.kind === "event" && items[3].event.type === "thread.activity-appended"
7154 ? items[3].event.payload.activity.kind
7155 : null,
7156 "tool.completed",
7157 );
7158 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
7159 );
7160
69217161 it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () =>
69227162 Effect.gen(function* () {
69237163 let readEventsCalls = 0;
apps/server/src/ws.ts+13-10
......@@ -80,6 +80,7 @@ import {
8080 projectActivityEvent,
8181 projectThreadDetailSnapshot,
8282} from "./orchestration/ActivityPayloadProjection.ts";
83import { makeThreadLiveEventCoalescer } from "./orchestration/ThreadLiveEventCoalescer.ts";
8384import {
8485 cleanupFailedUploadedAttachments,
8586 normalizeDispatchCommand,
......@@ -1501,17 +1502,15 @@ const makeWsRpcLayer = (
15011502 Stream.filter(isThisThreadDetailEvent),
15021503 Stream.map((event) => ({
15031504 kind: "event" as const,
1504 event: projectActivityEvent(event),
1505 event,
15051506 })),
15061507 );
15071508
15081509 // Attach live delivery before reading either replay or snapshot state.
15091510 // Otherwise an event published while the snapshot is loading is lost.
1510 const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
1511 yield* Effect.forkScoped(
1512 liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
1513 );
1514 const bufferedLiveStream = Stream.fromQueue(liveBuffer);
1511 const liveBuffer = yield* makeThreadLiveEventCoalescer();
1512 yield* Effect.forkScoped(liveStream.pipe(Stream.runForEach(liveBuffer.offer)));
1513 const bufferedLiveStream = liveBuffer.stream;
15151514
15161515 // When the client already loaded the snapshot over HTTP it passes
15171516 // that snapshot's sequence, and we resume the live subscription by
......@@ -1560,8 +1559,10 @@ const makeWsRpcLayer = (
15601559 input.requestCompletionMarker === true
15611560 ? Stream.concat(
15621561 Stream.fromEffect(
1563 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1564 ).pipe(Stream.drain),
1562 liveBuffer
1563 .offerAndWait({ kind: "synchronized" as const })
1564 .pipe(Effect.andThen(liveBuffer.takeAll)),
1565 ).pipe(Stream.flatMap((items) => Stream.fromIterable(items))),
15651566 bufferedLiveStream,
15661567 )
15671568 : bufferedLiveStream;
......@@ -1602,8 +1603,10 @@ const makeWsRpcLayer = (
16021603 input.requestCompletionMarker === true
16031604 ? Stream.concat(
16041605 Stream.fromEffect(
1605 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1606 ).pipe(Stream.drain),
1606 liveBuffer
1607 .offerAndWait({ kind: "synchronized" as const })
1608 .pipe(Effect.andThen(liveBuffer.takeAll)),
1609 ).pipe(Stream.flatMap((items) => Stream.fromIterable(items))),
16071610 bufferedLiveStream,
16081611 )
16091612 : bufferedLiveStream;