perf(server): bound orchestration replay and slow-client live buffers

apps/server/src/channels/ChannelRuntime.test.ts

@@ -440,6 +440,8 @@ function makeHarness(input: {
     });
   const engine = {
     readEvents: () => Stream.empty,
+    readThreadEvents: () => Stream.empty,
+    getThreadReplayStats: () => Effect.die("unused thread replay stats"),
     dispatch,
     streamDomainEvents: Stream.empty,
     subscribeDomainEvents: Effect.succeed(Stream.empty),

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts

@@ -85,6 +85,7 @@ describe("CheckpointDiffQuery.layer", () => {
               Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
             getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
             getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
+            getEventReplayStats: () => Effect.die("unused"),
             getOriginalProjectIdByWorkspaceRoot: () => Effect.die("unused"),
             getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
             getProjectShellById: () => Effect.succeed(Option.none()),
@@ -197,6 +198,7 @@ describe("CheckpointDiffQuery.layer", () => {
               Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
             getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
             getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
+            getEventReplayStats: () => Effect.die("unused"),
             getOriginalProjectIdByWorkspaceRoot: () => Effect.die("unused"),
             getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
             getProjectShellById: () => Effect.succeed(Option.none()),
@@ -284,6 +286,7 @@ describe("CheckpointDiffQuery.layer", () => {
               Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
             getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
             getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
+            getEventReplayStats: () => Effect.die("unused"),
             getOriginalProjectIdByWorkspaceRoot: () => Effect.die("unused"),
             getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
             getProjectShellById: () => Effect.succeed(Option.none()),
@@ -356,6 +359,7 @@ describe("CheckpointDiffQuery.layer", () => {
               Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
             getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
             getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
+            getEventReplayStats: () => Effect.die("unused"),
             getOriginalProjectIdByWorkspaceRoot: () => Effect.die("unused"),
             getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
             getProjectShellById: () => Effect.succeed(Option.none()),
@@ -413,6 +417,7 @@ describe("CheckpointDiffQuery.layer", () => {
               Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
             getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
             getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
+            getEventReplayStats: () => Effect.die("unused"),
             getOriginalProjectIdByWorkspaceRoot: () => Effect.die("unused"),
             getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
             getProjectShellById: () => Effect.succeed(Option.none()),

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -106,6 +106,8 @@ describe("OrchestrationEngine", () => {
           return savedEvent;
         }),
       readFromSequence: () => Stream.empty,
+      readAggregateRange: () => Stream.die("unused aggregate replay"),
+      getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"),
       readAll: () =>
         Stream.fail(
           new PersistenceSqlError({
@@ -209,6 +211,7 @@ describe("OrchestrationEngine", () => {
           getSnapshotSequence: () =>
             Effect.succeed({ snapshotSequence: projectionSnapshot.snapshotSequence }),
           getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
+          getEventReplayStats: () => Effect.die("unused"),
           getOriginalProjectIdByWorkspaceRoot: () => Effect.die("unused"),
           getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
           getProjectShellById: () => Effect.succeed(Option.none()),
@@ -823,6 +826,12 @@ describe("OrchestrationEngine", () => {
       readFromSequence(sequenceExclusive) {
         return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));
       },
+      readAggregateRange() {
+        return Stream.die("unused aggregate replay");
+      },
+      getAggregateReplayStats() {
+        return Effect.die("unused aggregate replay stats");
+      },
       readAll() {
         return Stream.fromIterable(events);
       },
@@ -1064,6 +1073,12 @@ describe("OrchestrationEngine", () => {
       readFromSequence(sequenceExclusive) {
         return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));
       },
+      readAggregateRange() {
+        return Stream.die("unused aggregate replay");
+      },
+      getAggregateReplayStats() {
+        return Effect.die("unused aggregate replay stats");
+      },
       readAll() {
         return Stream.fromIterable(events);
       },

apps/server/src/orchestration/Layers/OrchestrationEngine.ts

@@ -408,8 +408,24 @@ const makeOrchestrationEngine = Effect.gen(function* () {
     Effect.annotateLogs({ sequence: commandReadModel.snapshotSequence }),
   );
 
-  const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
-    eventStore.readFromSequence(fromSequenceExclusive, limit);
+  const readEvents: OrchestrationEngineShape["readEvents"] = (
+    fromSequenceExclusive,
+    limit,
+    toSequenceInclusive,
+  ) => eventStore.readFromSequence(fromSequenceExclusive, limit, toSequenceInclusive);
+
+  const readThreadEvents: OrchestrationEngineShape["readThreadEvents"] = ({ threadId, ...range }) =>
+    eventStore.readAggregateRange({ ...range, aggregateKind: "thread", aggregateId: threadId });
+
+  const getThreadReplayStats: OrchestrationEngineShape["getThreadReplayStats"] = ({
+    threadId,
+    ...range
+  }) =>
+    eventStore.getAggregateReplayStats({
+      ...range,
+      aggregateKind: "thread",
+      aggregateId: threadId,
+    });
 
   const dispatch: OrchestrationEngineShape["dispatch"] = (command, options) =>
     Effect.gen(function* () {
@@ -426,6 +442,8 @@ const makeOrchestrationEngine = Effect.gen(function* () {
 
   return {
     readEvents,
+    readThreadEvents,
+    getThreadReplayStats,
     dispatch,
     // Each access creates a fresh PubSub subscription so that multiple
     // consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.)

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts

@@ -1008,6 +1008,55 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
       }),
   );
 
+  it.effect("measures replay payload bytes without decoding event bodies", () =>
+    Effect.gen(function* () {
+      const snapshotQuery = yield* ProjectionSnapshotQuery;
+      const sql = yield* SqlClient.SqlClient;
+
+      yield* sql`DELETE FROM orchestration_events`;
+      const rows = yield* sql<{ readonly sequence: number }>`
+        INSERT INTO orchestration_events (
+          event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at,
+          command_id, causation_event_id, correlation_id, actor_kind, payload_json, metadata_json
+        )
+        VALUES
+          (
+            'replay-event-1', 'thread', 'thread-replay', 1, 'thread.activity-appended',
+            '2026-03-01T00:00:00.000Z', NULL, NULL, NULL, 'provider',
+            json_object('output', printf('%.*c', 1000, 'x')), '{}'
+          ),
+          (
+            'replay-event-2', 'thread', 'thread-replay', 2, 'thread.activity-appended',
+            '2026-03-01T00:00:01.000Z', NULL, NULL, NULL, 'provider',
+            json_object('output', printf('%.*c', 2000, 'x')), '{}'
+          ),
+          (
+            'replay-event-3', 'thread', 'thread-replay', 3, 'thread.activity-appended',
+            '2026-03-01T00:00:02.000Z', NULL, NULL, NULL, 'provider',
+            json_object('output', printf('%.*c', 3000, 'x')), '{}'
+          ),
+          (
+            'replay-event-4', 'thread', 'thread-replay', 4, 'thread.activity-appended',
+            '2026-03-01T00:00:03.000Z', NULL, NULL, NULL, 'provider',
+            json_object('output', '😀'), '{}'
+          )
+        RETURNING sequence
+      `;
+
+      // Bytes, not code points: the 4-byte emoji row is {"output":"😀"}, 17 bytes.
+      // Shared-layer AUTOINCREMENT does not reset after DELETE, so bound the
+      // query to the inserted sequences rather than assuming 1..4.
+      const stats = yield* snapshotQuery.getEventReplayStats({
+        fromSequenceExclusive: rows[0]!.sequence,
+        toSequenceInclusive: rows[3]!.sequence,
+      });
+      assert.deepStrictEqual(stats, {
+        eventCount: 3,
+        payloadBytes: 5043,
+      });
+    }),
+  );
+
   it.effect("reads single-thread checkpoint context without hydrating unrelated threads", () =>
     Effect.gen(function* () {
       const snapshotQuery = yield* ProjectionSnapshotQuery;

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts

@@ -81,6 +81,7 @@ import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityRes
 import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
 import {
   ProjectionSnapshotQuery,
+  type ProjectionEventReplayStats,
   type ProjectionFullThreadDiffContext,
   type ProjectionSnapshotCounts,
   type ProjectionThreadCheckpointContext,
@@ -192,6 +193,14 @@ const ProjectionCountsRowSchema = Schema.Struct({
   projectCount: Schema.Number,
   threadCount: Schema.Number,
 });
+const EventReplayStatsInput = Schema.Struct({
+  fromSequenceExclusive: NonNegativeInt,
+  toSequenceInclusive: NonNegativeInt,
+});
+const EventReplayStatsRowSchema = Schema.Struct({
+  eventCount: Schema.Number,
+  payloadBytes: Schema.Number,
+});
 const ProjectionThreadSearchRequest = Schema.Struct({
   pattern: Schema.String,
   limit: Schema.Int,
@@ -1009,6 +1018,20 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
       `,
   });
 
+  const readEventReplayStats = SqlSchema.findOne({
+    Request: EventReplayStatsInput,
+    Result: EventReplayStatsRowSchema,
+    execute: ({ fromSequenceExclusive, toSequenceInclusive }) =>
+      sql`
+        SELECT
+          COUNT(*) AS "eventCount",
+          COALESCE(SUM(octet_length(payload_json)), 0) AS "payloadBytes"
+        FROM orchestration_events
+        WHERE sequence > ${fromSequenceExclusive}
+          AND sequence <= ${toSequenceInclusive}
+      `,
+  });
+
   const searchActiveThreadRows = SqlSchema.findAll({
     Request: ProjectionThreadSearchRequest,
     Result: ProjectionThreadSearchRow,
@@ -2952,6 +2975,22 @@ pending_approval_requests AS (
       ),
     );
 
+  const getEventReplayStats: ProjectionSnapshotQueryShape["getEventReplayStats"] = (input) =>
+    readEventReplayStats(input).pipe(
+      Effect.mapError(
+        toPersistenceSqlOrDecodeError(
+          "ProjectionSnapshotQuery.getEventReplayStats:query",
+          "ProjectionSnapshotQuery.getEventReplayStats:decodeRow",
+        ),
+      ),
+      Effect.map(
+        (row): ProjectionEventReplayStats => ({
+          eventCount: row.eventCount,
+          payloadBytes: row.payloadBytes,
+        }),
+      ),
+    );
+
   const searchThreads: ProjectionSnapshotQueryShape["searchThreads"] = Effect.fn(
     "ProjectionSnapshotQuery.searchThreads",
   )(function* (input) {
@@ -3665,6 +3704,7 @@ pending_approval_requests AS (
     searchThreads,
     getSnapshotSequence,
     getCounts,
+    getEventReplayStats,
     getActiveProjectByWorkspaceRoot,
     getOriginalProjectIdByWorkspaceRoot,
     getProjectShellById,

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts

@@ -473,6 +473,8 @@ describe("ProviderCommandReactor", () => {
         const engine = yield* OrchestrationEngineService;
         return {
           readEvents: engine.readEvents,
+          readThreadEvents: engine.readThreadEvents,
+          getThreadReplayStats: engine.getThreadReplayStats,
           dispatch: (command) => {
             if (command.type === "thread.title.regeneration.complete") {
               titleRegenerationCompletionDispatchAttempts += 1;

apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts

@@ -69,6 +69,8 @@ describe("ThreadDeletionReactor.start", () => {
 
         const engine: OrchestrationEngineService["Service"] = {
           readEvents: () => Stream.die("unused"),
+          readThreadEvents: () => Stream.die("unused"),
+          getThreadReplayStats: () => Effect.die("unused"),
           dispatch: () => Effect.die("unused"),
           get streamDomainEvents() {
             return Stream.die(

apps/server/src/orchestration/LiveStreamBudget.test.ts

@@ -0,0 +1,47 @@
+import { it } from "@effect/vitest";
+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 { describe, expect } from "vite-plus/test";
+
+import { makeLiveStreamBudget, type RetainedLiveItem } from "./LiveStreamBudget.ts";
+
+describe("LiveStreamBudget", () => {
+  it.effect("closes the source without releasing a batch still waiting for an ACK", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const budget = yield* makeLiveStreamBudget({ maxItems: 3 });
+        const queue = yield* Queue.unbounded<RetainedLiveItem<{ text: string }>>();
+        const sourceClosed = yield* Deferred.make<void>();
+        const first = yield* budget.retain({ text: "first" });
+        const second = yield* budget.retain({ text: "second" });
+        const third = yield* budget.retain({ text: "third" });
+        yield* Queue.offerAll(queue, [first, second, third]);
+        yield* Effect.scoped(
+          Effect.gen(function* () {
+            const pull = yield* Stream.toPull(
+              budget.deliver(
+                Stream.fromQueue(queue).pipe(
+                  Stream.rechunk(1),
+                  Stream.ensuring(Deferred.succeed(sourceClosed, undefined)),
+                ),
+              ),
+            );
+            expect(yield* pull).toEqual([{ text: "first" }]);
+            // The other two items are in the source's pull state, not its queue.
+            expect(yield* Queue.size(queue)).toBe(0);
+            expect((yield* budget.usage).retainedItems).toBe(3);
+            const overflow = yield* budget.retain({ text: "fourth" }).pipe(Effect.result);
+            expect(overflow._tag).toBe("Failure");
+            // Do not resume the consumer. Its source scope must close now.
+            yield* Deferred.await(sourceClosed);
+            yield* budget.closed;
+            expect((yield* budget.usage).retainedItems).toBe(1);
+          }),
+        );
+        expect(yield* budget.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
+      }),
+    ),
+  );
+});