From 39c66a2c6befec42fd7563eec3ab62521dba7bcd Mon Sep 17 00:00:00 2001 From: maria-rcks <254055478+maria-rcks@users.noreply.github.com> Date: Fri, 2 Oct 2026 16:14:29 -0300 Subject: [PATCH 1/5] fix(server): allow bounded shell sync buffer override --- apps/server/src/ws.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 58bbcb2dbede..7a976f6df36e 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -4,6 +4,7 @@ import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as NodeCrypto from "node:crypto"; import * as DateTime from "effect/DateTime"; +import * as Config from "effect/Config"; import * as Duration from "effect/Duration"; import * as Encoding from "effect/Encoding"; import * as Effect from "effect/Effect"; @@ -842,6 +843,11 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc readonly afterSequence?: number; readonly requestCompletionMarker?: boolean; }) { + const shellLiveBufferMiB = yield* Config.Int("T3CODE_SHELL_LIVE_BUFFER_MIB").pipe( + Config.withDefault(8), + Config.map((value) => Math.max(1, Math.min(64, value))), + Effect.orDie, + ); const sql = yield* SqlClient.SqlClient; const threadManagement = yield* ThreadManagementService.ThreadManagementService; const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; @@ -943,6 +949,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc project: toShellApplicationEvent, }), ), + { maxSerializedBytes: shellLiveBufferMiB * 1024 * 1024 }, ); const enrichmentRefreshes = Stream.fromSubscription(enrichmentChanges).pipe( From d88a20984998394c29c8c72d9781253c59c60834 Mon Sep 17 00:00:00 2001 From: maria-rcks <254055478+maria-rcks@users.noreply.github.com> Date: Fri, 2 Oct 2026 19:03:12 -0300 Subject: [PATCH 2/5] fix(server): coalesce queued shell updates while awaiting delivery --- .../orchestration-v2/LiveStreamBudget.test.ts | 64 +++++++++++++++ .../src/orchestration-v2/LiveStreamBudget.ts | 79 +++++++++++++++++++ .../src/orchestration-v2/ShellStream.test.ts | 62 +++++++++++++++ .../src/orchestration-v2/ShellStream.ts | 31 +++++++- apps/server/src/ws.ts | 3 +- 5 files changed, 237 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/LiveStreamBudget.test.ts b/apps/server/src/orchestration-v2/LiveStreamBudget.test.ts index f003aea73dbf..1aaafcd0d9a2 100644 --- a/apps/server/src/orchestration-v2/LiveStreamBudget.test.ts +++ b/apps/server/src/orchestration-v2/LiveStreamBudget.test.ts @@ -1,4 +1,5 @@ import { it } from "@effect/vitest"; +import type * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -9,6 +10,7 @@ import { describe, expect } from "vite-plus/test"; import { bufferLiveStream, + bufferLatestLiveStream, makeLiveStreamBudget, replayAndBufferLiveEvents, type RetainedLiveItem, @@ -77,6 +79,68 @@ it.effect("stops draining a slow subscriber when its unacknowledged tail fills", type Event = { readonly sequence: number; readonly text: string; readonly threadId?: string }; +describe("bufferLatestLiveStream", () => { + it.effect( + "replaces queued payloads across batches without replacing the unacknowledged batch", + () => + Effect.scoped( + Effect.gen(function* () { + const input = yield* Queue.unbounded(); + const drained = yield* Deferred.make(); + const first = { sequence: 1, threadId: "a", text: "x".repeat(1000) }; + const pull = yield* Stream.toPull( + bufferLatestLiveStream( + Stream.fromQueue(input).pipe(Stream.ensuring(Deferred.succeed(drained, undefined))), + (event) => event.threadId!, + { maxItems: 3, maxSerializedBytes: 3300 }, + ), + ); + yield* Queue.offer(input, first); + expect(yield* pull).toEqual([first]); + const updates = Array.from({ length: 300 }, (_, index) => ({ + sequence: index + 2, + threadId: index % 2 === 0 ? "a" : "b", + text: "x".repeat(1000), + })); + yield* Queue.offerAll(input, updates); + yield* Queue.end(input); + // No ACK/pull while all 300 updates pass through the producer. + yield* Deferred.await(drained); + expect(yield* pull).toEqual(updates.slice(-2)); + }), + ), + ); + + for (const limit of ["items", "bytes"] as const) { + it.effect( + `keeps an in-flight update charged to the ${limit} budget when the same key updates`, + () => + Effect.scoped( + Effect.gen(function* () { + const input = yield* Queue.unbounded(); + const closed = yield* Deferred.make(); + const first = { sequence: 1, text: "first" }; + const pull = yield* Stream.toPull( + bufferLatestLiveStream( + Stream.fromQueue(input).pipe(Stream.ensuring(Deferred.succeed(closed, undefined))), + () => "same-key", + limit === "items" ? { maxItems: 1 } : { maxSerializedBytes: 32 }, + ), + ); + yield* Queue.offer(input, first); + expect(yield* pull).toEqual([first]); + yield* Queue.offer(input, { sequence: 2, text: "next" }); + yield* Deferred.await(closed); + const result = yield* pull.pipe(Effect.result); + expect(result._tag).toBe("Failure"); + if (result._tag === "Failure") + expect(result.failure._tag).toBe("LiveStreamBufferError"); + }), + ), + ); + } +}); + describe("replayAndBufferLiveEvents", () => { for (const phase of ["high-water", "replay"] as const) { it.effect(`unsubscribes and cancels a blocked ${phase} read on live overflow`, () => diff --git a/apps/server/src/orchestration-v2/LiveStreamBudget.ts b/apps/server/src/orchestration-v2/LiveStreamBudget.ts index 6353b4da9065..7db11573f6c3 100644 --- a/apps/server/src/orchestration-v2/LiveStreamBudget.ts +++ b/apps/server/src/orchestration-v2/LiveStreamBudget.ts @@ -7,6 +7,7 @@ import * as Exit from "effect/Exit"; import * as Queue from "effect/Queue"; import * as PubSub from "effect/PubSub"; import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; export class LiveStreamBufferError extends Schema.TaggedError()( @@ -262,6 +263,84 @@ export const bufferLiveStream = ( }), ); +/** For ordered full-state sources, keep only the latest queued state per aggregate while awaiting ACK. */ +export const bufferLatestLiveStream = ( + source: Stream.Stream, + key: (value: A) => string, + limits?: LiveStreamLimits, +) => + Stream.unwrap( + Effect.gen(function* () { + const budget = yield* makeLiveStreamBudget(limits); + const ready = yield* Queue.unbounded(); + const mutex = yield* Semaphore.make(1); + const pending = new Map>(); + let closed = false; + const close = (error?: LiveStreamBufferError) => + mutex.withPermits(1)( + Effect.gen(function* () { + if (closed) return; + closed = true; + budget.release(pending.values()); + pending.clear(); + if (error) yield* Queue.fail(ready, error); + yield* Queue.shutdown(ready); + }), + ); + yield* Effect.addFinalizer(() => close()); + yield* budget.failed.pipe( + Effect.catchTags({ LiveStreamBufferError: close }), + Effect.forkScoped, + ); + yield* source.pipe( + Stream.runForEach((value) => + mutex.withPermits(1)( + Effect.gen(function* () { + yield* budget.check; + if (closed) return; + const identity = key(value); + const previous = pending.get(identity); + if (previous && previous.value.sequence >= value.sequence) return; + const wasEmpty = pending.size === 0; + const [item] = yield* budget.replace(previous ? [previous] : [], [value]); + pending.set(identity, item!); + // One wakeup per pending batch, not per update. Replacements + // cannot leave an unbounded queue of obsolete keys behind. + if (wasEmpty) yield* Queue.offer(ready, undefined); + }).pipe(Effect.uninterruptible), + ), + ), + Effect.raceFirst(budget.failed), + Effect.exit, + Effect.flatMap((exit) => + Exit.isFailure(exit) ? Queue.failCause(ready, exit.cause) : Queue.end(ready), + ), + Effect.forkScoped({ startImmediately: true }), + ); + return budget.deliver( + Stream.fromPull( + Effect.succeed( + Queue.take(ready).pipe( + Effect.andThen( + mutex.withPermits(1)( + Effect.sync(() => { + const items = Array.from(pending.values()).sort( + (left, right) => left.value.sequence - right.value.sequence, + ); + pending.clear(); + // The wakeup exists only for a nonempty pending batch. + // Delivery now owns these items until the next ACK. + return items as Arr.NonEmptyArray>; + }), + ), + ), + ), + ), + ), + ); + }), + ); + /** * Subscribe and start draining before reading the high-water mark. A blocked * catch-up query or an unacknowledged replay batch must not strand an unbounded diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index 3435468a41d3..8214b184d368 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -2,16 +2,21 @@ import { describe, expect, it } from "@effect/vitest"; import type { ApplicationStoredEvent, OrchestrationV2ShellSnapshot, + OrchestrationV2ShellStreamItem, OrchestrationV2StoredEvent, OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import { ProjectId, ThreadId } from "@t3tools/contracts"; +import type * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import { archivedShellStreamItemFromThreadShell, buildActiveShellSnapshot, + bufferShellLiveStream, coalesceShellApplicationEvents, coalesceStoredThreadEvents, composeShellStreamWithEnrichment, @@ -45,6 +50,63 @@ const emptyShellSnapshot = { archivedThreads: [], } as OrchestrationV2ShellSnapshot; +it.effect("buffers the latest shell state in sequence order across deletion and recreation", () => + Effect.scoped( + Effect.gen(function* () { + const input = yield* Queue.unbounded< + Extract, + Cause.Done + >(); + const drained = yield* Deferred.make(); + const pull = yield* Stream.toPull( + bufferShellLiveStream( + Stream.fromQueue(input).pipe(Stream.ensuring(Deferred.succeed(drained, undefined))), + { maxItems: 5 }, + ), + ); + const initial = { + kind: "project.removed" as const, + sequence: 1, + projectId: ProjectId.make("initial"), + }; + yield* Queue.offer(input, initial); + expect(yield* pull).toEqual([initial]); + const shell = shellFixture({ id: ThreadId.make("same") }); + const updates = [ + { + kind: "thread.updated" as const, + sequence: 2, + location: "active" as const, + thread: shell, + }, + { + kind: "thread.removed" as const, + sequence: 3, + location: "active" as const, + threadId: shell.id, + }, + { kind: "project.removed" as const, sequence: 4, projectId: ProjectId.make("same") }, + { + kind: "thread.updated" as const, + sequence: 5, + location: "active" as const, + thread: shell, + }, + { + kind: "thread.removed" as const, + sequence: 6, + location: "archive" as const, + threadId: shell.id, + }, + ]; + yield* Queue.offerAll(input, updates); + yield* Queue.end(input); + yield* Deferred.await(drained); + expect(yield* pull).toEqual(updates.slice(2)); + }), + ), +); + describe("buildActiveShellSnapshot", () => { it("never duplicates archived rows into the regular shell", () => { const active = shellFixture({ archivedAt: null }); diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 31f556ce6dec..c030e91ee6a9 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -12,6 +12,32 @@ import { OrchestrationProjectShell as ProjectShellSchema } from "@t3tools/contra import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import { bufferLatestLiveStream } from "./LiveStreamBudget.ts"; + +type ShellDelta = Extract; + +/** Full shell deltas replace earlier queued state; snapshots and sync markers stay outside this buffer. */ +export const bufferShellLiveStream = ( + source: Stream.Stream, + limits?: { readonly maxItems?: number; readonly maxSerializedBytes?: number }, +) => + bufferLatestLiveStream( + source, + (item) => { + switch (item.kind) { + case "project.updated": + return `project:${item.project.id}`; + case "project.removed": + return `project:${item.projectId}`; + case "thread.updated": + return `thread:${item.location}:${item.thread.id}`; + case "thread.removed": + return `thread:${item.location}:${item.threadId}`; + } + }, + limits, + ); + /** Build the regular navigation shell without duplicating the archive dataset. */ export function buildActiveShellSnapshot(input: { readonly projects: ReadonlyArray; @@ -208,7 +234,10 @@ export function coalesceStoredThreadEvents( export function shellStreamItemFromThreadShell(input: { readonly stored: Extract; readonly shell: OrchestrationV2ThreadShell | null; -}): Exclude { +}): Extract< + OrchestrationV2ShellStreamItem, + { readonly kind: "thread.updated" | "thread.removed" } +> { if (input.shell !== null) { if (input.shell.archivedAt !== null) { return { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 7a976f6df36e..e2afcb82316a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -121,6 +121,7 @@ import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; import { archivedShellStreamItemFromThreadShell, buildActiveShellSnapshot, + bufferShellLiveStream, coalesceShellApplicationEvents, coalesceStoredThreadEvents, composeShellStreamWithEnrichment, @@ -942,7 +943,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc ); const liveFrom = (afterSequence: number) => - bufferLiveStream( + bufferShellLiveStream( toShellStream( applicationEvents.streamProjectedApplicationEvents({ afterSequence, From ca34bd412b2dae9d77bba14b6e76bbe7fd9f44b6 Mon Sep 17 00:00:00 2001 From: maria-rcks <254055478+maria-rcks@users.noreply.github.com> Date: Fri, 2 Oct 2026 22:44:54 -0300 Subject: [PATCH 3/5] fix(server): drop host metadata from dismissed pull request links --- apps/server/src/orchestration-v2/Orchestrator.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index df9805d504c9..ec0c63a84df8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2775,6 +2775,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : link, ); } + // Tombstones only block re-adding a stack layer, and clients hide them. + // Host metadata on them is dead weight copied into every thread event. + pullRequests = pullRequests.map((link) => + link.source === "stack-dismissed" && (link.snapshot !== null || link.stack !== null) + ? { ...link, snapshot: null, stack: null } + : link, + ); return { ...thread, pullRequests, From 8e8256f066fc1286a9acd899c3c2333b7bd14dd7 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 6 Oct 2026 16:08:01 +0000 Subject: [PATCH 4/5] fix(server): report malformed shell buffer budgets --- apps/server/src/ws.test.ts | 29 +++++++++++++++++++++++++++++ apps/server/src/ws.ts | 8 +++++++- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index ae263ac15fd3..09db3654eba1 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -5,6 +5,12 @@ import { type ServerConfigStreamEvent, } from "@t3tools/contracts"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; +import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; +import * as ThreadManagementService from "./orchestration-v2/ThreadManagementService.ts"; +import * as ProjectService from "./project/ProjectService.ts"; +import * as ProjectEnrichmentService from "./project/ProjectEnrichmentService.ts"; import * as ConfigProvider from "effect/ConfigProvider"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; @@ -23,6 +29,7 @@ import { hasCompatibleOrchestrationProtocol, resolveAvailableEditorsForConfig, shouldUseBoundedThreadSnapshot, + subscribeOrchestrationV2Shell, withLateEditorConfig, } from "./ws.ts"; @@ -234,3 +241,25 @@ it.effect("recovers a reveal kind whose real probe outlasts the config timeout", } }).pipe(Effect.scoped), ); + +it.effect("returns a typed shell error for a malformed live buffer budget", () => + Effect.gen(function* () { + const error = yield* subscribeOrchestrationV2Shell({}).pipe(Effect.flip); + assert.strictEqual(error._tag, "OrchestrationV2GetShellSnapshotError"); + assert.strictEqual(error.message, "Failed to prepare the application shell stream"); + }).pipe( + Effect.provide( + Layer.mergeAll( + ConfigProvider.layer( + ConfigProvider.fromEnv({ env: { T3CODE_SHELL_LIVE_BUFFER_MIB: "not-an-integer" } }), + ), + Layer.mock(SqlClient.SqlClient)({}), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}), + Layer.mock(ProjectStore.ProjectStoreV2)({}), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({}), + ), + ), + ), +); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 7060a8ab8b48..d45bf3c540bc 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -943,7 +943,13 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc const shellLiveBufferMiB = yield* Config.Int("T3CODE_SHELL_LIVE_BUFFER_MIB").pipe( Config.withDefault(8), Config.map((value) => Math.max(1, Math.min(64, value))), - Effect.orDie, + Effect.mapError( + (cause) => + new OrchestrationV2GetShellSnapshotError({ + message: "Failed to prepare the application shell stream", + cause, + }), + ), ); const sql = yield* SqlClient.SqlClient; const threadManagement = yield* ThreadManagementService.ThreadManagementService; From 0127fa9f0187f8d6489f760ac0f40dff12d06b65 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 6 Oct 2026 16:24:11 +0000 Subject: [PATCH 5/5] test(server): keep shell configuration failure typed --- apps/server/src/ws.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 09db3654eba1..7ff31bf57738 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -253,7 +253,7 @@ it.effect("returns a typed shell error for a malformed live buffer budget", () = ConfigProvider.layer( ConfigProvider.fromEnv({ env: { T3CODE_SHELL_LIVE_BUFFER_MIB: "not-an-integer" } }), ), - Layer.mock(SqlClient.SqlClient)({}), + Layer.succeed(SqlClient.SqlClient, {} as SqlClient.SqlClient), Layer.mock(ThreadManagementService.ThreadManagementService)({}), Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}), Layer.mock(ProjectStore.ProjectStoreV2)({}),