diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 405103450e8..b00c00e0d3f 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -218,6 +218,7 @@ describe("OrchestrationEngine", () => { const runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); + expect(await runtime.runPromise(engine.latestSequence)).toBe(7); const result = await runtime.runPromise( engine.dispatch({ type: "thread.meta.update", @@ -228,6 +229,7 @@ describe("OrchestrationEngine", () => { ); expect(result.sequence).toBe(8); + expect(await runtime.runPromise(engine.latestSequence)).toBe(8); expect(fullSnapshotReadCount).toBe(0); await runtime.dispose(); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 9598d1c3ef5..19184915ac7 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -329,6 +329,11 @@ const makeOrchestrationEngine = Effect.gen(function* () { get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] { return Stream.fromPubSub(eventPubSub); }, + // The command read model's snapshotSequence tracks the latest committed + // event sequence (updated on the worker fiber). A plain property read is a + // consistent, committed value — reassignment of `commandReadModel` is + // atomic on the single-threaded event loop. + latestSequence: Effect.sync(() => commandReadModel.snapshotSequence), } satisfies OrchestrationEngineShape; }); diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts index dc1b5fd3003..f8bcfd76ac0 100644 --- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -56,6 +56,13 @@ export interface OrchestrationEngineShape { * This is a hot runtime stream (new events only), not a historical replay. */ readonly streamDomainEvents: Stream.Stream; + + /** + * The latest sequence reflected in the engine's authoritative command read + * model (0 if none). Used to gauge how far behind a resuming client is before + * choosing between an incremental replay and a fresh projected snapshot. + */ + readonly latestSequence: Effect.Effect; } /** diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 9ca17dbc2dd..0c261c8f21e 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -470,6 +470,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { readEvents: () => Stream.empty, dispatch: () => Effect.succeed({ sequence: 1 }), streamDomainEvents: Stream.fromQueue(events), + latestSequence: Effect.succeed(0), } satisfies OrchestrationEngineShape; const snapshotQuery = { @@ -659,6 +660,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { readEvents: () => Stream.empty, dispatch: () => Effect.succeed({ sequence: 1 }), streamDomainEvents: Stream.fromQueue(events), + latestSequence: Effect.succeed(0), } satisfies OrchestrationEngineShape), Layer.succeed(ProjectionSnapshotQuery, { getShellSnapshot: () => diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index a0f41cbf0fc..6122c498145 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -46,6 +46,7 @@ import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as ManagedRuntime from "effect/ManagedRuntime"; import * as Option from "effect/Option"; @@ -677,6 +678,7 @@ const buildAppUnderTest = (options?: { readEvents: () => Stream.empty, dispatch: () => Effect.succeed({ sequence: 0 }), streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), ...options?.layers?.orchestrationEngine, }), ), @@ -5688,7 +5690,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => { projectionSnapshotQuery: { getShellSnapshot: () => Effect.gen(function* () { - yield* Effect.sleep("25 millis"); yield* PubSub.publish(liveEvents, deletedEvent); return { snapshotSequence: 1, @@ -5774,6 +5775,473 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); + it.effect("subscribeShell sends a fresh snapshot instead of replaying a large gap", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const snapshotThreadId = ThreadId.make("thread-from-snapshot"); + const now = "2026-01-01T00:00:00.000Z"; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + // Head is far ahead of the client's afterSequence (gap > 1000). + latestSequence: Effect.succeed(100_000), + readEvents: () => + Stream.sync(() => { + readEventsCalls += 1; + return { + sequence: 1, + eventId: EventId.make("event-should-not-be-read"), + aggregateKind: "thread", + aggregateId: snapshotThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.created", + payload: {} as never, + } satisfies OrchestrationEvent; + }), + }, + projectionSnapshotQuery: { + getShellSnapshot: () => + Effect.succeed({ + snapshotSequence: 100_000, + projects: [], + threads: [makeDefaultOrchestrationThreadShell({ id: snapshotThreadId })], + updatedAt: now, + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: 5, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + + const [first, second] = Array.from(items); + // Large gap => fresh snapshot, and the unbounded replay is never started. + assert.equal(first?.kind, "snapshot"); + if (first?.kind === "snapshot") { + assert.equal(first.snapshot.threads[0]?.id, snapshotThreadId); + } + assert.equal(second?.kind, "synchronized"); + assert.equal(readEventsCalls, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeShell replaces a cursor ahead of the authoritative head", () => + Effect.gen(function* () { + let readEventsCalls = 0; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(5), + readEvents: () => + Stream.sync(() => { + readEventsCalls += 1; + return {} as OrchestrationEvent; + }), + }, + projectionSnapshotQuery: { + getShellSnapshot: () => + Effect.succeed({ + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-01-01T00:00:00.000Z", + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const first = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 10 }).pipe( + Stream.runHead, + ), + ), + ); + + assert.equal(Option.getOrThrow(first).kind, "snapshot"); + assert.equal(readEventsCalls, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeShell coalesces a per-thread burst without stalling other threads", () => + Effect.gen(function* () { + const busyThreadId = ThreadId.make("thread-busy"); + const newThreadId = ThreadId.make("thread-new"); + const now = "2026-01-01T00:00:00.000Z"; + const shellFetches: Array = []; + let replayLimit: number | undefined; + + const messageEvent = (sequence: number): OrchestrationEvent => + ({ + sequence, + eventId: EventId.make(`event-${sequence}`), + aggregateKind: "thread", + aggregateId: busyThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: {} as never, + }) satisfies OrchestrationEvent; + + const createdEvent: OrchestrationEvent = { + sequence: 50, + eventId: EventId.make("event-created"), + aggregateKind: "thread", + aggregateId: newThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.created", + payload: {} as never, + }; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(50), + // A burst of message-sent deltas for the busy thread, plus one + // thread.created for a different thread, all within one batch. + readEvents: (_afterSequence, limit) => { + replayLimit = limit; + return Stream.fromIterable([ + ...Array.from({ length: 20 }, (_unused, index) => messageEvent(index + 1)), + createdEvent, + ]); + }, + }, + projectionSnapshotQuery: { + getThreadShellById: (threadId) => + Effect.sync(() => { + shellFetches.push(threadId); + return Option.some(makeDefaultOrchestrationThreadShell({ id: threadId })); + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: 0, + requestCompletionMarker: true, + }).pipe(Stream.take(3), Stream.runCollect), + ), + ); + + const collected = Array.from(items); + const upsertedIds = collected.flatMap((item) => + item.kind === "thread-upserted" ? [item.thread.id] : [], + ); + // Both threads surface, and the busy thread's 20-event burst collapses to + // a single shell refetch (not 20). The new thread is not stuck behind it. + assert.include(upsertedIds, busyThreadId); + assert.include(upsertedIds, newThreadId); + assert.equal(collected[2]?.kind, "synchronized"); + assert.equal(shellFetches.filter((id) => id === busyThreadId).length, 1); + assert.equal(replayLimit, 50); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeShell coalesces live bursts after the synchronization marker", () => + Effect.gen(function* () { + const busyThreadId = ThreadId.make("thread-live-busy"); + const newThreadId = ThreadId.make("thread-live-new"); + const now = "2026-01-01T00:00:00.000Z"; + const liveEvents = yield* PubSub.unbounded(); + const synchronized = yield* Deferred.make(); + const shellFetches: Array = []; + const observedLiveThreadIds = new Set(); + + const messageEvent = (sequence: number): OrchestrationEvent => + ({ + sequence, + eventId: EventId.make(`event-live-${sequence}`), + aggregateKind: "thread", + aggregateId: busyThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: {} as never, + }) satisfies OrchestrationEvent; + + const createdEvent: OrchestrationEvent = { + sequence: 50, + eventId: EventId.make("event-live-created"), + aggregateKind: "thread", + aggregateId: newThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.created", + payload: {} as never, + }; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + streamDomainEvents: Stream.fromPubSub(liveEvents), + }, + projectionSnapshotQuery: { + getThreadShellById: (threadId) => + Effect.sync(() => { + shellFetches.push(threadId); + return Option.some(makeDefaultOrchestrationThreadShell({ id: threadId })); + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + Effect.gen(function* () { + const itemsFiber = yield* withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + requestCompletionMarker: true, + }).pipe( + Stream.tap((item) => + item.kind === "synchronized" + ? Deferred.succeed(synchronized, undefined).pipe(Effect.ignore) + : Effect.void, + ), + Stream.takeUntil((item) => { + if (item.kind === "thread-upserted") { + observedLiveThreadIds.add(item.thread.id); + } + return ( + observedLiveThreadIds.has(busyThreadId) && observedLiveThreadIds.has(newThreadId) + ); + }), + Stream.runCollect, + ), + ).pipe(Effect.forkScoped); + + yield* Deferred.await(synchronized); + for (const event of [ + ...Array.from({ length: 20 }, (_unused, index) => messageEvent(index + 1)), + createdEvent, + ]) { + yield* PubSub.publish(liveEvents, event); + } + + return yield* Fiber.join(itemsFiber); + }), + ).pipe(Effect.timeout("2 seconds")); + + assert.equal(items[0]?.kind, "snapshot"); + assert.equal(items[1]?.kind, "synchronized"); + const liveUpsertedIds = Array.from(items) + .slice(2) + .flatMap((item) => (item.kind === "thread-upserted" ? [item.thread.id] : [])); + assert.include(liveUpsertedIds, busyThreadId); + assert.include(liveUpsertedIds, newThreadId); + assert.isBelow(shellFetches.filter((id) => id === busyThreadId).length, 20); + }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), + ); + + it.effect("subscribeShell coalescing still emits a removal for a deleted thread", () => + Effect.gen(function* () { + const goneThreadId = ThreadId.make("thread-gone"); + const now = "2026-01-01T00:00:00.000Z"; + + const makeThreadEvent = ( + sequence: number, + type: "thread.deleted" | "thread.message-sent", + ): OrchestrationEvent => + ({ + sequence, + eventId: EventId.make(`event-${sequence}`), + aggregateKind: "thread", + aggregateId: goneThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type, + payload: type === "thread.deleted" ? { threadId: goneThreadId, deletedAt: now } : {}, + }) as OrchestrationEvent; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(2), + // A thread.deleted followed, within the same coalescing window, by a + // later refetchable event for the same thread. The later event wins + // coalescing; its shell refetch returns none (the row is gone), which + // must still surface a removal rather than be swallowed. + readEvents: () => + Stream.fromIterable([ + makeThreadEvent(1, "thread.deleted"), + makeThreadEvent(2, "thread.message-sent"), + ]), + }, + projectionSnapshotQuery: { + getThreadShellById: () => Effect.succeed(Option.none()), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + const [first] = Array.from(items); + assert.equal(first?.kind, "thread-removed"); + assert.equal(first?.kind === "thread-removed" ? first.threadId : null, goneThreadId); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeShell retries a transient shell projection refetch failure", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-transient-refetch"); + const now = "2026-01-01T00:00:00.000Z"; + let attempts = 0; + + const event: OrchestrationEvent = { + sequence: 1, + eventId: EventId.make("event-transient-refetch"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: {} as never, + }; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(1), + readEvents: () => Stream.make(event), + }, + projectionSnapshotQuery: { + getThreadShellById: () => + Effect.suspend(() => { + attempts += 1; + return attempts === 1 + ? Effect.fail( + new PersistenceSqlError({ + operation: "test.shell-refetch", + detail: "transient failure", + }), + ) + : Effect.succeed( + Option.some(makeDefaultOrchestrationThreadShell({ id: threadId })), + ); + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + const [first] = Array.from(items); + assert.equal(first?.kind, "thread-upserted"); + assert.equal(first?.kind === "thread-upserted" ? first.thread.id : null, threadId); + assert.equal(attempts, 2); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeShell coalescing still removes a project after a trailing update", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project-gone"); + const now = "2026-01-01T00:00:00.000Z"; + + const makeProjectEvent = ( + sequence: number, + type: "project.deleted" | "project.meta-updated", + ): OrchestrationEvent => + ({ + sequence, + eventId: EventId.make(`event-project-${sequence}`), + aggregateKind: "project", + aggregateId: projectId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type, + payload: + type === "project.deleted" + ? { projectId, deletedAt: now } + : { projectId, title: "Still deleted", updatedAt: now }, + }) as OrchestrationEvent; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(2), + readEvents: () => + Stream.fromIterable([ + makeProjectEvent(1, "project.deleted"), + makeProjectEvent(2, "project.meta-updated"), + ]), + }, + projectionSnapshotQuery: { + getProjectShellById: () => Effect.succeed(Option.none()), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + const [first] = Array.from(items); + assert.equal(first?.kind, "project-removed"); + assert.equal(first?.kind === "project-removed" ? first.projectId : null, projectId); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("enriches replayed project events with repository identity metadata", () => Effect.gen(function* () { const repositoryIdentity = { diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index 1a331bc717f..b8102bda9ad 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -168,6 +168,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa Effect.as({ sequence: 1 }), ), streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]), Effect.provide(NodeServices.layer), ); @@ -211,6 +212,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when Effect.as({ sequence: 1 }), ), streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]), Effect.provide(NodeServices.layer), ); @@ -260,6 +262,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa Effect.as({ sequence: 1 }), ), streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]), Effect.provideService(Crypto.Crypto, { ...crypto, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 41c29fc3388..08b1770a0a2 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -36,6 +36,7 @@ import { OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, + type ProjectId, type ProjectEntriesFailure, type ProjectFileFailure, type ProjectFileOperation, @@ -276,6 +277,13 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< const PROVIDER_STATUS_DEBOUNCE_MS = 200; +// When a resuming client's cursor is more than this many events behind the +// current head, skip the per-event catch-up replay and send a fresh shell +// snapshot instead. Replaying each intervening event costs a shell refetch; +// past this gap a single O(active-threads) snapshot is cheaper and bounded. +// Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT). +const SHELL_RESUME_MAX_GAP = 1_000; + const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], @@ -621,16 +629,7 @@ const makeWsRpcLayer = ( switch (event.type) { case "project.created": case "project.meta-updated": - return projectionSnapshotQuery.getProjectShellById(event.payload.projectId).pipe( - Effect.map((project) => - Option.map(project, (nextProject) => ({ - kind: "project-upserted" as const, - sequence: event.sequence, - project: nextProject, - })), - ), - Effect.orElseSucceed(() => Option.none()), - ); + return projectUpsertOrRemove(event.payload.projectId, event.sequence); case "project.deleted": return Effect.succeed( Option.some({ @@ -649,35 +648,192 @@ const makeWsRpcLayer = ( }), ); case "thread.unarchived": - return projectionSnapshotQuery.getThreadShellById(event.payload.threadId).pipe( - Effect.map((thread) => - Option.map(thread, (nextThread) => ({ - kind: "thread-upserted" as const, - sequence: event.sequence, - thread: nextThread, - })), - ), - Effect.orElseSucceed(() => Option.none()), - ); + return threadUpsertOrRemove(event.payload.threadId, event.sequence); default: if (event.aggregateKind !== "thread") { return Effect.succeed(Option.none()); } - return projectionSnapshotQuery - .getThreadShellById(ThreadId.make(event.aggregateId)) - .pipe( - Effect.map((thread) => - Option.map(thread, (nextThread) => ({ - kind: "thread-upserted" as const, - sequence: event.sequence, - thread: nextThread, - })), - ), - Effect.orElseSucceed(() => Option.none()), - ); + return threadUpsertOrRemove(ThreadId.make(event.aggregateId), event.sequence); } }; + // Coalescing makes each projection read represent every event for that + // aggregate in the current window. Retry a typed persistence failure once + // so a brief read failure cannot strand the shell at its previous state. + // If both attempts fail, log and drop the stream item; treating an error as + // a missing row would incorrectly remove a still-active aggregate. + const retryShellProjectionRead = ( + aggregateKind: "project" | "thread", + aggregateId: string, + read: Effect.Effect, + ): Effect.Effect, never, never> => + read.pipe( + Effect.retry({ times: 1 }), + Effect.map(Option.some), + Effect.tapError((error) => + Effect.logWarning("orchestration shell projection refetch failed", { + aggregateKind, + aggregateId, + error, + }), + ), + Effect.orElseSucceed(() => Option.none()), + ); + + const projectUpsertOrRemove = ( + projectId: ProjectId, + sequence: number, + ): Effect.Effect, never, never> => + retryShellProjectionRead( + "project", + projectId, + projectionSnapshotQuery.getProjectShellById(projectId), + ).pipe( + Effect.map( + Option.flatMap((project) => + Option.match(project, { + onNone: () => + Option.some({ + kind: "project-removed" as const, + sequence, + projectId, + }), + onSome: (nextProject) => + Option.some({ + kind: "project-upserted" as const, + sequence, + project: nextProject, + }), + }), + ), + ), + ); + + // Refetch a thread's shell and emit an upsert if it is still active, or a + // `thread-removed` if the projection has no active row for it. Emitting a + // removal on a `none` (rather than dropping the event) is what keeps + // coalescing correct: when a burst collapses a `thread.deleted`/`archived` + // into a later refetchable event for the same thread, the refetch returns + // `none` for the now-inactive row and this still tells the sidebar to drop + // it. A `thread-removed` the client does not have is a harmless no-op. The + // projection commits in the same transaction before the event publishes, + // so a `none` reliably means the thread is deleted or archived, not + // not-yet-persisted. + const threadUpsertOrRemove = ( + threadId: ThreadId, + sequence: number, + ): Effect.Effect, never, never> => + retryShellProjectionRead( + "thread", + threadId, + projectionSnapshotQuery.getThreadShellById(threadId), + ).pipe( + Effect.map( + Option.flatMap((thread) => + Option.match(thread, { + onNone: () => + Option.some({ + kind: "thread-removed" as const, + sequence, + threadId, + }), + onSome: (nextThread) => + Option.some({ + kind: "thread-upserted" as const, + sequence, + thread: nextThread, + }), + }), + ), + ), + ); + + // Turn a batch of domain events into shell stream items, coalescing by + // aggregate first. `toShellStreamEvent` re-reads the *current* projected + // shell for an aggregate, so within a batch only the latest event per + // aggregate matters: a burst of streaming `thread.message-sent` deltas for + // one thread collapses into a single shell refetch, and an unrelated + // `thread.created` in the same batch is never stuck behind those DB reads. + // + // Input events arrive in ascending sequence; we keep the last (highest + // sequence) event per aggregate, then re-sort ascending before emitting so + // the client — which applies shell items strictly by increasing sequence + // and drops any `sequence <= snapshotSequence` — never skips a coalesced + // item. The refetch runs with bounded concurrency (order-preserving). + const SHELL_REFETCH_CONCURRENCY = 8; + const coalesceShellEvents = ( + events: ReadonlyArray, + ): Effect.Effect, never, never> => + Effect.gen(function* () { + if (events.length === 0) { + return []; + } + const latestByAggregate = new Map(); + for (const event of events) { + latestByAggregate.set(`${event.aggregateKind}:${event.aggregateId}`, event); + } + const survivors = Array.from(latestByAggregate.values()).sort( + (left, right) => left.sequence - right.sequence, + ); + const shellEvents = yield* Effect.forEach(survivors, toShellStreamEvent, { + concurrency: SHELL_REFETCH_CONCURRENCY, + }); + return shellEvents.flatMap((option) => (Option.isSome(option) ? [option.value] : [])); + }); + + // Small time/size window over which to coalesce shell events. The window + // bounds the worst-case added latency for a brand-new thread to appear in + // the sidebar (imperceptible), while collapsing high-frequency streaming + // traffic so it can't serialize the shell stream behind per-event DB reads. + const SHELL_COALESCE_WINDOW = Duration.millis(50); + const SHELL_COALESCE_MAX_CHUNK = 512; + const coalesceShellStream = ( + stream: Stream.Stream, + ): Stream.Stream => + stream.pipe( + 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: "synchronized" }; + + // A completion marker is queued alongside raw live events 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 = ( + inputs: ReadonlyArray, + ): Effect.Effect, never, never> => + Effect.gen(function* () { + const output: Array = []; + let pendingEvents: Array = []; + + for (const input of inputs) { + if (input.kind === "event") { + pendingEvents.push(input.event); + continue; + } + + output.push(...(yield* coalesceShellEvents(pendingEvents))); + pendingEvents = []; + output.push({ kind: "synchronized" }); + } + + output.push(...(yield* coalesceShellEvents(pendingEvents))); + return output; + }); + + const coalesceShellLiveStream = ( + stream: Stream.Stream, + ): Stream.Stream => + stream.pipe( + Stream.groupedWithin(SHELL_COALESCE_MAX_CHUNK, SHELL_COALESCE_WINDOW), + Stream.mapEffect(coalesceShellLiveInputs), + Stream.flatMap((items) => Stream.fromIterable(items)), + ); + const dispatchBootstrapTurnStart = ( command: Extract, ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => @@ -1068,68 +1224,29 @@ const makeWsRpcLayer = ( observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeShell, Effect.gen(function* () { - const liveStream = orchestrationEngine.streamDomainEvents.pipe( - Stream.mapEffect(toShellStreamEvent), - Stream.flatMap((event) => - Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, + // Coalesce the live shell stream per aggregate over a small window + // so bursts of high-frequency events (streaming message deltas, + // activity appends) collapse into a single shell refetch and never + // serialize a brand-new thread's `thread.created` behind hundreds + // of per-event DB reads. See coalesceShellStream. + // Attach live delivery into a scope-bound buffer BEFORE loading any + // snapshot or draining catch-up, otherwise an event published while + // the snapshot query is in flight is lost (it is past the snapshot's + // sequence but the live subscription is not attached yet). Every + // path below emits from this same buffered live tail. Overlapping + // events are deduped by sequence on the client. + const liveBuffer = yield* Queue.unbounded(); + yield* Effect.forkScoped( + orchestrationEngine.streamDomainEvents.pipe( + Stream.runForEach((event) => + Queue.offer(liveBuffer, { kind: "event" as const, event }), + ), ), + { startImmediately: true }, ); + const bufferedLiveStream = coalesceShellLiveStream(Stream.fromQueue(liveBuffer)); - // When the client already holds a shell snapshot (cached, or loaded - // over HTTP) it passes that snapshot's sequence, and we resume by - // replaying shell events after it instead of re-sending the whole - // projects/threads list over the socket. As in the thread path, the - // live subscription is attached (into a scope-bound buffer) before - // draining the catch-up replay so no event published during the - // replay window is lost; overlapping events are deduped by sequence - // on the client. The full range is read (not the store's default - // page limit) since the shell filter runs after reading. - if (input.afterSequence !== undefined) { - const afterSequence = input.afterSequence; - return Stream.unwrap( - Effect.gen(function* () { - const liveBuffer = yield* Queue.unbounded(); - yield* Effect.forkScoped( - liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), - ); - const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) - .pipe( - Stream.mapEffect(toShellStreamEvent), - Stream.flatMap((event) => - Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, - ), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: "Failed to replay orchestration shell events", - cause, - }), - ), - ); - const afterCatchUp = - input.requestCompletionMarker === true - ? Stream.concat( - Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), - Stream.fromQueue(liveBuffer), - ) - : Stream.fromQueue(liveBuffer); - return Stream.concat(catchUpStream, afterCatchUp); - }), - ); - } - - // The full-snapshot fallback needs the same replay-window safety - // as the resume path: subscribe before loading the projection so - // events published while the snapshot is read are buffered. - const liveBuffer = yield* Queue.unbounded(); - yield* Effect.forkScoped( - liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), - ); - const bufferedLiveStream = Stream.fromQueue(liveBuffer); - const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( + const loadSnapshot = projectionSnapshotQuery.getShellSnapshot().pipe( Effect.tapError((cause) => Effect.logError("orchestration shell snapshot load failed", { cause }), ), @@ -1142,21 +1259,70 @@ const makeWsRpcLayer = ( ), ); - const afterSnapshot = + // Offer the completion marker into the same queue as live events. + // Anything buffered while snapshot/replay work was in flight is + // therefore delivered before the client is told it is synchronized. + const synchronizedThenLive = 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.flatMap(coalesceShellLiveInputs), + ), + ).pipe(Stream.flatMap((items) => Stream.fromIterable(items))), bufferedLiveStream, ) : bufferedLiveStream; + + // When the client already holds a shell snapshot (cached, or loaded + // over HTTP) it passes that snapshot's sequence, and we resume by + // replaying shell events after it instead of re-sending the whole + // projects/threads list over the socket. If the client is too far + // behind, we fall back to a fresh snapshot instead of an unbounded + // replay (see below). + if (input.afterSequence !== undefined) { + const afterSequence = input.afterSequence; + const headSequence = yield* orchestrationEngine.latestSequence; + const replayGap = headSequence - afterSequence; + // Gap too large: replaying every intervening event (each a shell + // refetch) is far more expensive than a single O(active-threads) + // snapshot. A cursor ahead of this engine's authoritative state + // is also invalid, so reset it with a snapshot. Send the snapshot + // followed by the buffered live tail, exactly as the + // no-afterSequence path does. + if (replayGap < 0 || replayGap > SHELL_RESUME_MAX_GAP) { + const snapshot = yield* loadSnapshot; + return Stream.concat( + Stream.make({ kind: "snapshot" as const, snapshot }), + synchronizedThenLive, + ); + } + const catchUpStream = coalesceShellStream( + // Replay only through the head captured above. Newer events + // are already covered by the live subscription, so this bound + // cannot chase a moving event-store head or grow the live + // buffer indefinitely while waiting for an empty page. + orchestrationEngine.readEvents(afterSequence, replayGap), + ).pipe( + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to replay orchestration shell events", + cause, + }), + ), + ); + return Stream.concat(catchUpStream, synchronizedThenLive); + } + + const snapshot = yield* loadSnapshot; return Stream.concat( Stream.make({ kind: "snapshot" as const, snapshot, }), - afterSnapshot, + synchronizedThenLive, ); }), { "rpc.aggregate": "orchestration" },