From 8995244104348c20a88c964ae89d19459c82c4ee Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 00:29:02 +0200 Subject: [PATCH 1/3] perf(server): coalesce live tool updates --- .../ActivityPayloadProjection.ts | 53 +++++++++++++- apps/server/src/server.test.ts | 70 +++++++++++++++++++ apps/server/src/ws.ts | 70 +++++++++++++++++-- .../test/ActivityPayloadProjection.test.ts | 69 ++++++++++++++++++ 4 files changed, 253 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration/ActivityPayloadProjection.ts b/apps/server/src/orchestration/ActivityPayloadProjection.ts index 32f249c251d5..8354791ac944 100644 --- a/apps/server/src/orchestration/ActivityPayloadProjection.ts +++ b/apps/server/src/orchestration/ActivityPayloadProjection.ts @@ -488,8 +488,8 @@ function toolLifecycleIdentity(activity: OrchestrationThreadActivity): string | * update within the turn — a later update belongs to a subsequent call that * reuses the same identity and is still in flight. Rows without a lifecycle * identity pass through, matching the clients, which never collapse them. - * Live `thread.activity-appended` events are untouched: updates still stream - * in real time and the completion supersedes them on the client as before. + * Live `thread.activity-appended` events use the same identity in + * `coalesceLiveToolUpdatedEvents`, but only within a short bounded window. * * Deliberate divergence from client collapse: clients fold only *adjacent* * lifecycle rows, so a superseded update separated from its completion by an @@ -542,6 +542,55 @@ function dropSupersededToolUpdatedActivities( }); } +/** + * Retain only the latest in-flight update for each tool call in a live batch. + * A later completion also supersedes preceding updates for the same call. All + * survivors stay in sequence order so client-side sequence dedup remains safe. + */ +export function coalesceLiveToolUpdatedEvents( + events: ReadonlyArray, +): ReadonlyArray { + const seenUpdates = new Set(); + const seenCompletions = new Set(); + const survivors: Array = []; + + for (let index = events.length - 1; index >= 0; index -= 1) { + const event = events[index]!; + if (event.type !== "thread.activity-appended") { + survivors.push(event); + continue; + } + + const activity = event.payload.activity; + if (activity.kind !== "tool.updated" && activity.kind !== "tool.completed") { + survivors.push(event); + continue; + } + + const identity = toolLifecycleIdentity(activity); + if (!identity) { + survivors.push(event); + continue; + } + + const key = `${activity.turnId ?? ""}\u0000${identity}`; + if (activity.kind === "tool.completed") { + seenCompletions.add(key); + survivors.push(event); + continue; + } + + if (seenUpdates.has(key) || seenCompletions.has(key)) { + continue; + } + seenUpdates.add(key); + survivors.push(event); + } + + survivors.reverse(); + return survivors; +} + export function projectThreadDetailSnapshot( snapshot: OrchestrationThreadDetailSnapshot, ): OrchestrationThreadDetailSnapshot { diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index a9a2c3fa10d6..a8713ddab145 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -18,6 +18,7 @@ import { ExternalLauncherCommandNotFoundError, OrchestrationThreadDetailSnapshot, type OrchestrationThreadStreamItem, + type OrchestrationThreadActivity, type OrchestrationThreadShell, TerminalNotRunningError, type OrchestrationCommand, @@ -29,6 +30,7 @@ import { ProviderInstanceId, ResolvedKeybindingRule, ThreadId, + TurnId, WS_METHODS, WsRpcGroup, EditorId, @@ -6421,6 +6423,74 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); + it.effect("coalesces buffered live tool updates to the latest state", () => + Effect.gen(function* () { + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + const liveEvents = yield* PubSub.unbounded(); + const makeToolUpdate = (sequence: number): OrchestrationEvent => { + const activity: OrchestrationThreadActivity = { + id: EventId.make(`activity-${sequence}`), + tone: "tool", + kind: "tool.updated", + summary: "Editing app.ts", + payload: { + itemType: "file_change", + title: "Editing app.ts", + data: { toolCallId: "call-edit" }, + }, + turnId: TurnId.make("turn-edit"), + createdAt: "2026-01-01T00:00:01.000Z", + }; + return { + sequence, + eventId: EventId.make(`event-tool-${sequence}`), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt: "2026-01-01T00:00:01.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.activity-appended", + payload: { threadId: defaultThreadId, activity }, + }; + }; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + streamDomainEvents: Stream.fromPubSub(liveEvents), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.gen(function* () { + yield* Effect.sleep("25 millis"); + yield* PubSub.publishAll(liveEvents, [ + makeToolUpdate(2), + makeToolUpdate(3), + makeToolUpdate(4), + ]); + return Option.some({ snapshotSequence: 1, thread }); + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ).pipe(Effect.timeout("2 seconds")); + + assert.equal(items[0]?.kind, "snapshot"); + assert.equal(items[1]?.kind, "event"); + assert.equal(items[1]?.kind === "event" ? items[1].event.sequence : null, 4); + }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), + ); + it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () => Effect.gen(function* () { let readEventsCalls = 0; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 226c82cdb1ac..487c995bd242 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -72,6 +72,7 @@ import * as ServerConfig from "./config.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import { + coalesceLiveToolUpdatedEvents, projectActivityEvent, projectThreadDetailSnapshot, } from "./orchestration/ActivityPayloadProjection.ts"; @@ -856,6 +857,55 @@ const makeWsRpcLayer = ( Stream.flatMap((items) => Stream.fromIterable(items)), ); + const THREAD_TOOL_UPDATE_COALESCE_WINDOW = Duration.millis(50); + const THREAD_TOOL_UPDATE_COALESCE_MAX_CHUNK = 512; + type ThreadLiveInput = + | { readonly kind: "event"; readonly event: OrchestrationEvent } + | { readonly kind: "synchronized" }; + + // Queue markers with raw events so a marker cannot overtake an update + // waiting in the coalescing window. Only event segments between markers + // are coalesced, preserving synchronization and sequence order. + const coalesceThreadLiveInputs = ( + inputs: ReadonlyArray, + ): ReadonlyArray => { + const output: Array = []; + let pendingEvents: Array = []; + + const flushEvents = () => { + output.push( + ...coalesceLiveToolUpdatedEvents(pendingEvents).map((event) => ({ + kind: "event" as const, + event: projectActivityEvent(event), + })), + ); + pendingEvents = []; + }; + + for (const input of inputs) { + if (input.kind === "event") { + pendingEvents.push(input.event); + continue; + } + flushEvents(); + output.push({ kind: "synchronized" }); + } + flushEvents(); + return output; + }; + + const coalesceThreadLiveStream = ( + stream: Stream.Stream, + ): Stream.Stream => + stream.pipe( + Stream.groupedWithin( + THREAD_TOOL_UPDATE_COALESCE_MAX_CHUNK, + THREAD_TOOL_UPDATE_COALESCE_WINDOW, + ), + Stream.map(coalesceThreadLiveInputs), + Stream.flatMap((items) => Stream.fromIterable(items)), + ); + const dispatchBootstrapTurnStart = ( command: Extract, ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => @@ -1456,17 +1506,17 @@ const makeWsRpcLayer = ( Stream.filter(isThisThreadDetailEvent), Stream.map((event) => ({ kind: "event" as const, - event: projectActivityEvent(event), + event, })), ); // Attach live delivery before reading either replay or snapshot state. // Otherwise an event published while the snapshot is loading is lost. - const liveBuffer = yield* Queue.unbounded(); + const liveBuffer = yield* Queue.unbounded(); yield* Effect.forkScoped( liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), ); - const bufferedLiveStream = Stream.fromQueue(liveBuffer); + const bufferedLiveStream = coalesceThreadLiveStream(Stream.fromQueue(liveBuffer)); // When the client already loaded the snapshot over HTTP it passes // that snapshot's sequence, and we resume the live subscription by @@ -1515,8 +1565,11 @@ const makeWsRpcLayer = ( input.requestCompletionMarker === true ? Stream.concat( Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), + Queue.offer(liveBuffer, { kind: "synchronized" as const }).pipe( + Effect.andThen(Queue.takeAll(liveBuffer)), + Effect.map(coalesceThreadLiveInputs), + ), + ).pipe(Stream.flatMap((items) => Stream.fromIterable(items))), bufferedLiveStream, ) : bufferedLiveStream; @@ -1557,8 +1610,11 @@ const makeWsRpcLayer = ( input.requestCompletionMarker === true ? Stream.concat( Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), + Queue.offer(liveBuffer, { kind: "synchronized" as const }).pipe( + Effect.andThen(Queue.takeAll(liveBuffer)), + Effect.map(coalesceThreadLiveInputs), + ), + ).pipe(Stream.flatMap((items) => Stream.fromIterable(items))), bufferedLiveStream, ) : bufferedLiveStream; diff --git a/apps/server/test/ActivityPayloadProjection.test.ts b/apps/server/test/ActivityPayloadProjection.test.ts index 7ac564bb41f3..e546197bb01b 100644 --- a/apps/server/test/ActivityPayloadProjection.test.ts +++ b/apps/server/test/ActivityPayloadProjection.test.ts @@ -14,6 +14,7 @@ import { buildThreadFeed, type ThreadFeedActivity } from "../../mobile/src/lib/t import { deriveLatestContextWindowSnapshot } from "../../web/src/lib/contextWindow.ts"; import { deriveWorkLogEntries } from "../../web/src/session-logic.ts"; import { + coalesceLiveToolUpdatedEvents, projectActivityEvent, projectActivityPayload, projectThreadDetailSnapshot, @@ -330,6 +331,32 @@ describe("superseded tool.updated snapshot dedup", () => { }).thread.activities.map((activity) => activity.id); } + function makeActivityEvent( + sequence: number, + activity: OrchestrationThreadActivity, + ): Extract { + const threadId = ThreadId.make("thread-live-coalescing"); + return { + sequence, + eventId: EventId.make(`event-${sequence}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-07-27T00:00:01.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.activity-appended", + payload: { threadId, activity }, + }; + } + + function coalescedIds(activities: ReadonlyArray) { + return coalesceLiveToolUpdatedEvents( + activities.map((activity, index) => makeActivityEvent(index + 1, activity)), + ).map((event) => event.eventId); + } + it("drops updates a later completion supersedes in the same turn", () => { const update1 = makeToolLifecycleActivity("upd-1", "tool.updated"); const update2 = makeToolLifecycleActivity("upd-2", "tool.updated"); @@ -421,6 +448,48 @@ describe("superseded tool.updated snapshot dedup", () => { expect(projectedIds([anonymous, completed])).toEqual([anonymous.id, completed.id]); }); + it("coalesces a live batch to the latest update per tool call", () => { + const firstA = makeToolLifecycleActivity("upd-a-1", "tool.updated", { + toolCallId: "call-a", + }); + const updateB = makeToolLifecycleActivity("upd-b", "tool.updated", { + toolCallId: "call-b", + }); + const latestA = makeToolLifecycleActivity("upd-a-2", "tool.updated", { + toolCallId: "call-a", + }); + + expect(coalescedIds([firstA, updateB, latestA])).toEqual([ + EventId.make("event-2"), + EventId.make("event-3"), + ]); + }); + + it("lets a live completion supersede earlier updates without hiding a later call", () => { + const update = makeToolLifecycleActivity("upd-first", "tool.updated"); + const completed = makeToolLifecycleActivity("done-first", "tool.completed"); + const nextCall = makeToolLifecycleActivity("upd-next", "tool.updated"); + + expect(coalescedIds([update, completed, nextCall])).toEqual([ + EventId.make("event-2"), + EventId.make("event-3"), + ]); + }); + + it("does not coalesce live updates across turns", () => { + const oldTurn = makeToolLifecycleActivity("upd-old", "tool.updated", { + turn: "turn-old", + }); + const newTurn = makeToolLifecycleActivity("upd-new", "tool.updated", { + turn: "turn-new", + }); + + expect(coalescedIds([oldTurn, newTurn])).toEqual([ + EventId.make("event-1"), + EventId.make("event-2"), + ]); + }); + it("leaves the collapsed work log identical to the full history", () => { const activities = [ makeToolLifecycleActivity("upd-1", "tool.updated", { detail: "writing" }), From 161f0b038968e66be8c7db065a7da4f724cb54b9 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 14:26:10 +0200 Subject: [PATCH 2/3] test(server): prove tool update coalescing safety --- apps/server/src/server.test.ts | 181 +++++++++++++++--- .../test/ActivityPayloadProjection.test.ts | 115 +++++++++++ 2 files changed, 265 insertions(+), 31 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index a8713ddab145..902cab73dab5 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -195,6 +195,38 @@ const defaultModelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", } as const; + +const makeLiveToolActivityEvent = ( + sequence: number, + kind: "tool.updated" | "tool.completed" = "tool.updated", +): Extract => { + const activity: OrchestrationThreadActivity = { + id: EventId.make(`activity-${sequence}`), + tone: "tool", + kind, + summary: "Editing app.ts", + payload: { + itemType: "file_change", + title: "Editing app.ts", + data: { toolCallId: "call-edit" }, + }, + turnId: TurnId.make("turn-edit"), + createdAt: "2026-01-01T00:00:01.000Z", + }; + return { + sequence, + eventId: EventId.make(`event-tool-${sequence}`), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt: "2026-01-01T00:00:01.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.activity-appended", + payload: { threadId: defaultThreadId, activity }, + }; +}; const testEnvironmentDescriptor = { environmentId: EnvironmentId.make("environment-test"), label: "Test environment", @@ -6427,34 +6459,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => { Effect.gen(function* () { const thread = makeDefaultOrchestrationReadModel().threads[0]!; const liveEvents = yield* PubSub.unbounded(); - const makeToolUpdate = (sequence: number): OrchestrationEvent => { - const activity: OrchestrationThreadActivity = { - id: EventId.make(`activity-${sequence}`), - tone: "tool", - kind: "tool.updated", - summary: "Editing app.ts", - payload: { - itemType: "file_change", - title: "Editing app.ts", - data: { toolCallId: "call-edit" }, - }, - turnId: TurnId.make("turn-edit"), - createdAt: "2026-01-01T00:00:01.000Z", - }; - return { - sequence, - eventId: EventId.make(`event-tool-${sequence}`), - aggregateKind: "thread", - aggregateId: defaultThreadId, - occurredAt: "2026-01-01T00:00:01.000Z", - commandId: null, - causationEventId: null, - correlationId: null, - metadata: {}, - type: "thread.activity-appended", - payload: { threadId: defaultThreadId, activity }, - }; - }; yield* buildAppUnderTest({ layers: { @@ -6466,9 +6470,9 @@ it.layer(NodeServices.layer)("server router seam", (it) => { Effect.gen(function* () { yield* Effect.sleep("25 millis"); yield* PubSub.publishAll(liveEvents, [ - makeToolUpdate(2), - makeToolUpdate(3), - makeToolUpdate(4), + makeLiveToolActivityEvent(2), + makeLiveToolActivityEvent(3), + makeLiveToolActivityEvent(4), ]); return Option.some({ snapshotSequence: 1, thread }); }), @@ -6491,6 +6495,121 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); + it.effect("flushes more than one tool chunk before the synchronization marker", () => + Effect.gen(function* () { + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + const liveEvents = yield* PubSub.unbounded(); + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + streamDomainEvents: Stream.fromPubSub(liveEvents), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.gen(function* () { + yield* Effect.sleep("25 millis"); + yield* PubSub.publishAll( + liveEvents, + Array.from({ length: 513 }, (_, index) => makeLiveToolActivityEvent(index + 2)), + ); + return Option.some({ snapshotSequence: 1, thread }); + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + requestCompletionMarker: true, + }).pipe(Stream.take(3), Stream.runCollect), + ), + ).pipe(Effect.timeout("2 seconds")); + + assert.equal(items[0]?.kind, "snapshot"); + assert.equal(items[1]?.kind === "event" ? items[1].event.sequence : null, 514); + assert.deepEqual(items[2], { kind: "synchronized" }); + }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), + ); + + it.effect("preserves an interleaved message while coalescing a tool lifecycle", () => + Effect.gen(function* () { + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + const liveEvents = yield* PubSub.unbounded(); + const messageEvent = { + sequence: 3, + eventId: EventId.make("event-interleaved-message"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt: "2026-01-01T00:00:02.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: { + threadId: defaultThreadId, + messageId: MessageId.make("message-interleaved"), + role: "assistant", + text: "Still working", + turnId: TurnId.make("turn-edit"), + streaming: false, + createdAt: "2026-01-01T00:00:02.000Z", + updatedAt: "2026-01-01T00:00:02.000Z", + }, + } satisfies Extract; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + streamDomainEvents: Stream.fromPubSub(liveEvents), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.gen(function* () { + yield* Effect.sleep("25 millis"); + yield* PubSub.publishAll(liveEvents, [ + makeLiveToolActivityEvent(2), + messageEvent, + makeLiveToolActivityEvent(4, "tool.completed"), + ]); + return Option.some({ snapshotSequence: 1, thread }); + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + }).pipe(Stream.take(3), Stream.runCollect), + ), + ).pipe(Effect.timeout("2 seconds")); + + assert.equal(items[0]?.kind, "snapshot"); + assert.deepEqual( + items + .slice(1) + .map((item) => (item.kind === "event" ? [item.event.sequence, item.event.type] : null)), + [ + [3, "thread.message-sent"], + [4, "thread.activity-appended"], + ], + ); + assert.equal( + items[2]?.kind === "event" && items[2].event.type === "thread.activity-appended" + ? items[2].event.payload.activity.kind + : null, + "tool.completed", + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), + ); + it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () => Effect.gen(function* () { let readEventsCalls = 0; diff --git a/apps/server/test/ActivityPayloadProjection.test.ts b/apps/server/test/ActivityPayloadProjection.test.ts index e546197bb01b..bc719458b241 100644 --- a/apps/server/test/ActivityPayloadProjection.test.ts +++ b/apps/server/test/ActivityPayloadProjection.test.ts @@ -1,5 +1,7 @@ import { + CheckpointRef, EventId, + MessageId, ProjectId, ProviderInstanceId, ThreadId, @@ -13,6 +15,7 @@ import { describe, expect, it } from "vite-plus/test"; import { buildThreadFeed, type ThreadFeedActivity } from "../../mobile/src/lib/threadActivity.ts"; import { deriveLatestContextWindowSnapshot } from "../../web/src/lib/contextWindow.ts"; import { deriveWorkLogEntries } from "../../web/src/session-logic.ts"; +import { applyThreadDetailEvent } from "../../../packages/client-runtime/src/state/threadReducer.ts"; import { coalesceLiveToolUpdatedEvents, projectActivityEvent, @@ -490,6 +493,118 @@ describe("superseded tool.updated snapshot dedup", () => { ]); }); + it("keeps client state equivalent after coalescing a mixed event sequence", () => { + const threadId = ThreadId.make("thread-live-coalescing"); + const turnId = TurnId.make("turn-a"); + const update1 = makeActivityEvent( + 1, + makeToolLifecycleActivity("upd-1", "tool.updated", { toolCallId: "call-a" }), + ); + const sessionSet = { + sequence: 2, + eventId: EventId.make("event-session"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-07-27T00:00:02.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.session-set", + payload: { + threadId, + session: { + threadId, + status: "running", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "full-access", + activeTurnId: turnId, + lastError: null, + updatedAt: "2026-07-27T00:00:02.000Z", + }, + }, + } satisfies Extract; + const update2 = makeActivityEvent( + 3, + makeToolLifecycleActivity("upd-2", "tool.updated", { toolCallId: "call-a" }), + ); + const messageSent = { + sequence: 4, + eventId: EventId.make("event-message"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-07-27T00:00:04.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: { + threadId, + messageId: MessageId.make("message-1"), + role: "assistant", + text: "Still working", + turnId, + streaming: false, + createdAt: "2026-07-27T00:00:04.000Z", + updatedAt: "2026-07-27T00:00:04.000Z", + }, + } satisfies Extract; + const completed = makeActivityEvent( + 5, + makeToolLifecycleActivity("done-1", "tool.completed", { toolCallId: "call-a" }), + ); + const diffCompleted = { + sequence: 6, + eventId: EventId.make("event-diff"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-07-27T00:00:06.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.turn-diff-completed", + payload: { + threadId, + turnId, + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("checkpoint-1"), + status: "ready", + files: [{ path: "src/app.ts", kind: "modified", additions: 2, deletions: 1 }], + assistantMessageId: MessageId.make("message-1"), + completedAt: "2026-07-27T00:00:06.000Z", + }, + } satisfies Extract; + const events = [update1, sessionSet, update2, messageSent, completed, diffCompleted]; + const coalesced = coalesceLiveToolUpdatedEvents(events); + + const applyEvents = (input: ReadonlyArray) => { + let thread = makeThread([]); + for (const event of input.map(projectActivityEvent)) { + const result = applyThreadDetailEvent(thread, event); + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + thread = result.thread; + } + } + return thread; + }; + + const originalState = applyEvents(events); + const coalescedState = applyEvents(coalesced); + + expect(coalesced.map((event) => event.sequence)).toEqual([2, 4, 5, 6]); + expect(coalescedState.messages).toEqual(originalState.messages); + expect(coalescedState.session).toEqual(originalState.session); + expect(coalescedState.checkpoints).toEqual(originalState.checkpoints); + expect(coalescedState.latestTurn).toEqual(originalState.latestTurn); + expect(deriveWorkLogEntries(coalescedState.activities)).toEqual( + deriveWorkLogEntries(originalState.activities), + ); + }); + it("leaves the collapsed work log identical to the full history", () => { const activities = [ makeToolLifecycleActivity("upd-1", "tool.updated", { detail: "writing" }), From 53bc703373072c598a5e9a4d59319deb782956fa Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 14:52:55 +0200 Subject: [PATCH 3/3] test(server): verify multi-tool marker flush --- apps/server/src/server.test.ts | 70 +++++++++++++++++++++++++++++----- 1 file changed, 60 insertions(+), 10 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 902cab73dab5..2003374423a6 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -199,16 +199,22 @@ const defaultModelSelection = { const makeLiveToolActivityEvent = ( sequence: number, kind: "tool.updated" | "tool.completed" = "tool.updated", + options: { + readonly toolCallId?: string; + readonly title?: string; + readonly path?: string; + } = {}, ): Extract => { + const { toolCallId = "call-edit", title = "Editing app.ts", path = "src/app.ts" } = options; const activity: OrchestrationThreadActivity = { id: EventId.make(`activity-${sequence}`), tone: "tool", kind, - summary: "Editing app.ts", + summary: title, payload: { itemType: "file_change", - title: "Editing app.ts", - data: { toolCallId: "call-edit" }, + title, + data: { toolCallId, path }, }, turnId: TurnId.make("turn-edit"), createdAt: "2026-01-01T00:00:01.000Z", @@ -6509,10 +6515,16 @@ it.layer(NodeServices.layer)("server router seam", (it) => { getThreadDetailSnapshot: () => Effect.gen(function* () { yield* Effect.sleep("25 millis"); - yield* PubSub.publishAll( - liveEvents, - Array.from({ length: 513 }, (_, index) => makeLiveToolActivityEvent(index + 2)), - ); + yield* PubSub.publishAll(liveEvents, [ + ...Array.from({ length: 512 }, (_, index) => + makeLiveToolActivityEvent(index + 2), + ), + makeLiveToolActivityEvent(514, "tool.updated", { + toolCallId: "call-read", + title: "Reading server.test.ts", + path: "apps/server/src/server.test.ts", + }), + ]); return Option.some({ snapshotSequence: 1, thread }); }), }, @@ -6525,13 +6537,51 @@ it.layer(NodeServices.layer)("server router seam", (it) => { client[ORCHESTRATION_WS_METHODS.subscribeThread]({ threadId: defaultThreadId, requestCompletionMarker: true, - }).pipe(Stream.take(3), Stream.runCollect), + }).pipe(Stream.take(4), Stream.runCollect), ), ).pipe(Effect.timeout("2 seconds")); assert.equal(items[0]?.kind, "snapshot"); - assert.equal(items[1]?.kind === "event" ? items[1].event.sequence : null, 514); - assert.deepEqual(items[2], { kind: "synchronized" }); + assert.deepEqual( + items.slice(1, 3).map((item) => { + assert.equal(item?.kind, "event"); + if (item?.kind !== "event" || item.event.type !== "thread.activity-appended") { + return null; + } + return { + sequence: item.event.sequence, + summary: item.event.payload.activity.summary, + payload: item.event.payload.activity.payload, + }; + }), + [ + { + sequence: 513, + summary: "Editing app.ts", + payload: { + itemType: "file_change", + title: "Editing app.ts", + data: { + files: [{ path: "src/app.ts" }], + toolCallId: "call-edit", + }, + }, + }, + { + sequence: 514, + summary: "Reading server.test.ts", + payload: { + itemType: "file_change", + title: "Reading server.test.ts", + data: { + files: [{ path: "apps/server/src/server.test.ts" }], + toolCallId: "call-read", + }, + }, + }, + ], + ); + assert.deepEqual(items[3], { kind: "synchronized" }); }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), );