diff --git a/.agents/upstream-review.md b/.agents/upstream-review.md index b8e54ee74..58cfa6c1d 100644 --- a/.agents/upstream-review.md +++ b/.agents/upstream-review.md @@ -15,6 +15,14 @@ Two standing sections outlive any single batch and must be read on every review: ## Review batches +## 2026-09-06 — project live updates before applying buffer limits (partial source) + +Adopted the live-stream portion of upstream #9799, `ce4712d5b04fb998f79fe132245289191147e5d5`, under the maintainer's standing approval for compatible fixes. The complete upstream patch, PR description, and review context were read. This port changes only the live budget/coalescer, shell stream metadata, and focused WebSocket regression tests. The upstream HTML, mobile outbox, highlighting, animation, marketing, and CI changes are separate concerns and are not claimed as adopted here. The full range cursor remains unchanged. + +A raw 9 MiB tool result reproduced `OrchestrationGetSnapshotError: The live event buffer is full` in both thread and shell subscriptions on the Pylon baseline, despite a small client projection. Thread queues now charge the already-projected activity payload; shell queues retain event identity and refetch the current aggregate. Existing item/byte bounds, sequence ordering, completion markers, replay, and actual overflow recovery remain intact. Provider/runtime state and persisted tool output are unchanged; this is a provider-independent client stream fix. + +Verification: both new regression cases failed before the port; all 183 tests across server routing, live budget, event coalescing, and activity projection passed afterward. The scoped server typecheck, targeted lint, and formatting passed. No UI layout or native dependency changes. Implementation branch: `upstream/2026-09-06-projected-live-stream-budget` from `origin/pylon` at `71b11a43bef90229c255c704dbbd8b368080b97c`. + ## 2026-09-06 — approved Pylon cleanup adaptations (partial range) The maintainer approved broad compatible adoption and routine Pylon adaptations. This batch adopts 36 source commits after reading their complete patches, PR descriptions, review context, and Pylon callers. The bounded upstream head is `223ff4490f764a74ff911589e97b9bbcd595fee8`; the full review cursor remains unchanged because the complete 614-commit range is not yet classified. diff --git a/apps/server/src/orchestration/LiveStreamBudget.ts b/apps/server/src/orchestration/LiveStreamBudget.ts index 3e3d31128..e86127b2b 100644 --- a/apps/server/src/orchestration/LiveStreamBudget.ts +++ b/apps/server/src/orchestration/LiveStreamBudget.ts @@ -93,7 +93,7 @@ export const makeLiveStreamBudget = Effect.fn("makeLiveStreamBudget")(function* return Effect.succeed(item); }); - // Replace one coalescing batch atomically. Both raw and projected payloads + // Replace one coalescing batch atomically. Queued and coalesced payloads // count against the same budget, and discarded updates release their charge. const replace = ( previous: ReadonlyArray>, diff --git a/apps/server/src/orchestration/ThreadLiveEventCoalescer.ts b/apps/server/src/orchestration/ThreadLiveEventCoalescer.ts index c7f2155b4..2831746d3 100644 --- a/apps/server/src/orchestration/ThreadLiveEventCoalescer.ts +++ b/apps/server/src/orchestration/ThreadLiveEventCoalescer.ts @@ -130,7 +130,7 @@ export const makeThreadLiveEventCoalescer = Effect.fn("makeThreadLiveEventCoales pendingUpdates, coalesceLiveToolUpdatedEvents(pendingUpdates.map((item) => item.value)).map((event) => ({ kind: "event" as const, - event: projectActivityEvent(event), + event, })), (item) => item.event, ); @@ -167,7 +167,8 @@ export const makeThreadLiveEventCoalescer = Effect.fn("makeThreadLiveEventCoales Effect.gen(function* () { yield* budget.check; if (input.kind === "event") { - yield* budget.retain(input.event).pipe( + // Retain only the client payload, not full persisted tool output. + yield* budget.retain(projectActivityEvent(input.event)).pipe( Effect.tap((item) => Effect.sync(() => pendingUpdates.push(item))), Effect.uninterruptible, ); diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 77c3ddc2e..3a14b421f 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7162,6 +7162,104 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + for (const subscription of ["thread", "shell"] as const) { + it.effect("delivers large raw tool results through the " + subscription + " stream", () => + Effect.gen(function* () { + const eventThreadId = + subscription === "thread" ? defaultThreadId : ThreadId.make("other-live-thread"); + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + const baseEvent = makeLiveToolActivityEvent(2, "tool.completed"); + const event: OrchestrationEvent = { + ...baseEvent, + aggregateId: eventThreadId, + payload: { + threadId: eventThreadId, + activity: { + ...baseEvent.payload.activity, + summary: "Build complete", + payload: { + itemType: "command_execution", + toolCallId: "call-build", + status: "completed", + title: "Build complete", + data: { + item: { + command: "build", + aggregatedOutput: "Build complete\n" + "x".repeat(9 * 1024 * 1024), + }, + }, + }, + }, + }, + }; + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + streamDomainEvents: Stream.concat(Stream.make(event), Stream.never), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.succeed(Option.some({ snapshotSequence: 1, thread })), + getThreadShellById: (threadId) => + Effect.succeed( + Option.some({ + ...makeDefaultOrchestrationThreadShell(), + id: threadId, + title: "Build complete", + }), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = + subscription === "thread" + ? yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + requestCompletionMarker: true, + }).pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.runCollect, + ), + ), + ) + : yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + requestCompletionMarker: true, + }).pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.runCollect, + ), + ), + ); + + assert.equal(items[0]?.kind, "snapshot"); + const update = items[1]; + if (subscription === "thread") { + assertTrue(update?.kind === "event" && update.event.type === "thread.activity-appended"); + assert.equal(update.event.sequence, 2); + assert.deepEqual(update.event.payload.activity.payload, { + itemType: "command_execution", + toolCallId: "call-build", + status: "completed", + title: "Build complete", + data: { item: { command: "build", aggregatedOutput: "Build complete" } }, + }); + } else { + assertTrue(update?.kind === "thread-upserted"); + assert.equal(update.sequence, 2); + assert.equal(update.thread.id, eventThreadId); + assert.equal(update.thread.title, "Build complete"); + } + assert.deepEqual(items[2], { kind: "synchronized" }); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + } + it.effect("coalesces buffered live tool updates to the latest state", () => Effect.gen(function* () { const thread = makeDefaultOrchestrationReadModel().threads[0]!; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 66e512d16..12ff001c9 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -36,7 +36,7 @@ import { OrchestrationSearchThreadsError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, - type ProjectId, + ProjectId, type ProviderInstanceId, type ServerSettings as ContractServerSettings, type ServerSettingsPatch, @@ -741,19 +741,33 @@ const makeWsRpcLayer = ( }); }; + // Shell updates refetch the aggregate. Message and tool bodies are not needed. + const toShellEvent = ({ + type, + aggregateKind, + aggregateId, + sequence, + }: OrchestrationEvent) => ({ + type, + aggregateKind, + aggregateId, + sequence, + }); + type ShellEvent = ReturnType; + const toShellStreamEvent = ( - event: OrchestrationEvent, + event: ShellEvent, ): Effect.Effect, never, never> => { switch (event.type) { case "project.created": case "project.meta-updated": - return projectUpsertOrRemove(event.payload.projectId, event.sequence); + return projectUpsertOrRemove(ProjectId.make(event.aggregateId), event.sequence); case "project.deleted": return Effect.succeed( Option.some({ kind: "project-removed" as const, sequence: event.sequence, - projectId: event.payload.projectId, + projectId: ProjectId.make(event.aggregateId), }), ); case "thread.deleted": @@ -762,11 +776,11 @@ const makeWsRpcLayer = ( Option.some({ kind: "thread-removed" as const, sequence: event.sequence, - threadId: event.payload.threadId, + threadId: ThreadId.make(event.aggregateId), }), ); case "thread.unarchived": - return threadUpsertOrRemove(event.payload.threadId, event.sequence); + return threadUpsertOrRemove(ThreadId.make(event.aggregateId), event.sequence); default: if (event.aggregateKind !== "thread") { return Effect.succeed(Option.none()); @@ -880,13 +894,13 @@ const makeWsRpcLayer = ( // item. The refetch runs with bounded concurrency (order-preserving). const SHELL_REFETCH_CONCURRENCY = 8; const coalesceShellEvents = ( - events: ReadonlyArray, + events: ReadonlyArray, ): Effect.Effect, never, never> => Effect.gen(function* () { if (events.length === 0) { return []; } - const latestByAggregate = new Map(); + const latestByAggregate = new Map(); for (const event of events) { latestByAggregate.set(`${event.aggregateKind}:${event.aggregateId}`, event); } @@ -909,16 +923,17 @@ const makeWsRpcLayer = ( stream: Stream.Stream, ): Stream.Stream => stream.pipe( + Stream.map(toShellEvent), Stream.groupedWithin(SHELL_COALESCE_MAX_CHUNK, SHELL_COALESCE_WINDOW), Stream.mapEffect(coalesceShellEvents), Stream.flatMap((items) => Stream.fromIterable(items)), ); type ShellLiveInput = - | { readonly kind: "event"; readonly event: OrchestrationEvent } + | { readonly kind: "event"; readonly event: ShellEvent } | { readonly kind: "synchronized" }; - // A completion marker is queued alongside raw live events so it cannot + // A completion marker is queued alongside live event metadata so it cannot // overtake an event still waiting in the coalescing window. Split each // batch at markers and coalesce only the event segments on either side. const coalesceShellLiveInputs = ( @@ -926,7 +941,7 @@ const makeWsRpcLayer = ( ): Effect.Effect, never, never> => Effect.gen(function* () { const output: Array = []; - let pendingEvents: Array = []; + let pendingEvents: Array = []; for (const input of inputs) { if (input.kind === "event") { @@ -1448,6 +1463,7 @@ const makeWsRpcLayer = ( ); yield* Effect.forkScoped( orchestrationEngine.streamDomainEvents.pipe( + Stream.map(toShellEvent), Stream.runForEach((event) => liveBudget.retain({ kind: "event" as const, event }, event).pipe( Effect.flatMap((item) => Queue.offer(liveBuffer, item)), diff --git a/docs/internals/live-updates.md b/docs/internals/live-updates.md new file mode 100644 index 000000000..1fa4c3b31 --- /dev/null +++ b/docs/internals/live-updates.md @@ -0,0 +1,9 @@ +# Live update buffers + +Thread subscriptions apply activity payload projection before retaining events in the live buffer. The buffer therefore measures the client payload, rather than the full persisted tool result. A large command log that projects to a short summary does not consume the raw log size from the live-update budget. + +Shell subscriptions retain only event type, aggregate kind, aggregate ID, and sequence. Coalescing keeps the latest event per aggregate, then refetches its current shell projection. Message bodies and tool results are not needed to update the sidebar. + +Item and byte limits still apply to queued and in-flight projected data. Completion markers remain ordered behind earlier events. A real overflow asks the client to resume from its last received sequence; bounded replay and snapshot recovery remain the same. + +The implementation is shared by providers and clients. See `apps/server/src/orchestration/LiveStreamBudget.ts`, `ThreadLiveEventCoalescer.ts` in the same directory, and the subscription handlers in `apps/server/src/ws.ts`. Persisted events retain their original payloads.