perf(server): replay only the selected thread

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

@@ -325,6 +325,8 @@ describe("OrchestrationEngine", () => {
           }),
         ),
       hasEventAfter: () => Effect.succeed(false),
+      readAggregateRange: () => Stream.die("unused aggregate replay"),
+      getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"),
     };
 
     const projectionSnapshot = {
@@ -1235,6 +1237,8 @@ describe("OrchestrationEngine", () => {
         return Stream.fromIterable(events);
       },
       hasEventAfter: () => Effect.succeed(false),
+      readAggregateRange: () => Stream.die("unused aggregate replay"),
+      getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"),
     };
 
     const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
@@ -1473,6 +1477,8 @@ describe("OrchestrationEngine", () => {
         return Stream.fromIterable(events);
       },
       hasEventAfter: () => Effect.succeed(false),
+      readAggregateRange: () => Stream.die("unused aggregate replay"),
+      getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"),
     };
 
     let shouldFailProjection = true;

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

@@ -381,6 +381,19 @@ const makeOrchestrationEngine = Effect.gen(function* () {
   const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
     eventStore.readFromSequence(fromSequenceExclusive, limit);
 
+  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* () {
       const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
@@ -395,6 +408,8 @@ const makeOrchestrationEngine = Effect.gen(function* () {
 
   return {
     readEvents,
+    readThreadEvents,
+    getThreadReplayStats,
     dispatch,
     subscribeDomainEvents: PubSub.subscribe(eventPubSub).pipe(Effect.map(Stream.fromSubscription)),
     // Each access creates a fresh PubSub subscription so that multiple

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

@@ -418,6 +418,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/Services/OrchestrationEngine.ts

@@ -14,6 +14,7 @@ import type {
   OrchestrationClientOrigin,
   OrchestrationCommand,
   OrchestrationEvent,
+  ThreadId,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
 import type * as Effect from "effect/Effect";
@@ -22,6 +23,13 @@ import type * as Stream from "effect/Stream";
 
 import type { OrchestrationDispatchError } from "../Errors.ts";
 import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts";
+import type { OrchestrationAggregateReplayStats } from "../../persistence/Services/OrchestrationEventStore.ts";
+
+export interface OrchestrationThreadReplayRange {
+  readonly threadId: ThreadId;
+  readonly fromSequenceExclusive: number;
+  readonly toSequenceInclusive: number;
+}
 
 /**
  * OrchestrationEngineShape - Service API for orchestration command and event flow.
@@ -33,15 +41,24 @@ export interface OrchestrationEngineShape {
    * @param fromSequenceExclusive - Sequence cursor (exclusive).
    * @param limit - Maximum number of events to read. Defaults to the event
    *   store's page-bounded default; pass a higher value when the caller must
-   *   read every event after the cursor (e.g. per-thread catch-up that filters
-   *   a small subset out of a potentially larger global range).
+   *   read a wider global range. Thread subscriptions use readThreadEvents.
    * @returns Stream containing ordered events.
    */
   readonly readEvents: (
     fromSequenceExclusive: number,
     limit?: number,
   ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
 
+  /** Read only this thread's events through a captured authoritative head. */
+  readonly readThreadEvents: (
+    input: OrchestrationThreadReplayRange & { readonly limit?: number },
+  ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;
+
+  /** Measure a bounded thread replay without decoding its event bodies. */
+  readonly getThreadReplayStats: (
+    input: OrchestrationThreadReplayRange & { readonly maxEvents: number },
+  ) => Effect.Effect<OrchestrationAggregateReplayStats, OrchestrationEventStoreError>;
+
   /**
    * Dispatch a validated orchestration command.
    *

apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts

@@ -1,4 +1,11 @@
-import { CommandId, EventId, ProjectId } from "@t3tools/contracts";
+import {
+  CommandId,
+  EventId,
+  MessageId,
+  ProjectId,
+  ThreadId,
+  type OrchestrationEvent,
+} from "@t3tools/contracts";
 import { assert, it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
@@ -12,6 +19,31 @@ import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts";
 import { SqlitePersistenceMemory } from "./Sqlite.ts";
 const isPersistenceDecodeError = Schema.is(PersistenceDecodeError);
 
+function messageEvent(threadId: ThreadId, id: string): Omit<OrchestrationEvent, "sequence"> {
+  const now = "2026-01-01T00:00:00.000Z";
+  return {
+    type: "thread.message-sent",
+    eventId: EventId.make(id),
+    aggregateKind: "thread",
+    aggregateId: threadId,
+    occurredAt: now,
+    commandId: null,
+    causationEventId: null,
+    correlationId: null,
+    metadata: {},
+    payload: {
+      threadId,
+      messageId: MessageId.make(id),
+      role: "assistant",
+      text: id,
+      turnId: null,
+      streaming: false,
+      createdAt: now,
+      updatedAt: now,
+    },
+  };
+}
+
 const layer = it.layer(
   OrchestrationEventStoreLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
 );
@@ -79,7 +111,7 @@ layer("OrchestrationEventStore", (it) => {
       const sql = yield* SqlClient.SqlClient;
       const now = "2026-01-01T00:00:00.000Z";
 
-      yield* sql`
+      const invalidRows = yield* sql<{ readonly sequence: number }>`
         INSERT INTO orchestration_events (
           event_id,
           aggregate_kind,
@@ -108,6 +140,7 @@ layer("OrchestrationEventStore", (it) => {
           ${"{"},
           ${"{}"}
         )
+        RETURNING sequence
       `;
 
       const replayResult = yield* Effect.result(
@@ -122,6 +155,154 @@ layer("OrchestrationEventStore", (it) => {
           ),
         );
       }
+      const scopedResult = yield* eventStore
+        .readAggregateRange({
+          aggregateKind: "project",
+          aggregateId: "project-invalid-json",
+          fromSequenceExclusive: 0,
+          toSequenceInclusive: invalidRows[0]!.sequence,
+        })
+        .pipe(Stream.runCollect, Effect.result);
+      assert.equal(scopedResult._tag, "Failure");
+      if (scopedResult._tag === "Failure") {
+        assert.ok(isPersistenceDecodeError(scopedResult.failure));
+        assert.ok(
+          scopedResult.failure.operation.includes(
+            "OrchestrationEventStore.readAggregateRange:decodeRows",
+          ),
+        );
+      }
+    }),
+  );
+
+  it.effect("reads one aggregate through the captured head across pruned global gaps", () =>
+    Effect.gen(function* () {
+      const store = yield* OrchestrationEventStore;
+      const sql = yield* SqlClient.SqlClient;
+      const threadId = ThreadId.make("shared-stream-id");
+      const first = yield* store.append(messageEvent(threadId, "scoped-first"));
+      const pruned = yield* store.append(
+        messageEvent(ThreadId.make("pruned-thread"), "pruned-event"),
+      );
+      const second = yield* store.append(messageEvent(threadId, "scoped-second"));
+      // The same stream ID in a different aggregate is not part of this thread.
+      // Its invalid JSON must never reach the event decoder.
+      yield* sql`
+        INSERT INTO orchestration_events (
+          event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at,
+          actor_kind, payload_json, metadata_json
+        ) VALUES (
+          'same-id-project', 'project', ${threadId}, 0, 'project.created',
+          '2026-01-01T00:00:00.000Z', 'server', '{', '{'
+        ), (
+          'unrelated-invalid', 'thread', 'unrelated-invalid-thread', 0, 'thread.activity-appended',
+          '2026-01-01T00:00:00.000Z', 'server', '{', '{'
+        )
+      `;
+      const last = yield* store.append(messageEvent(threadId, "scoped-last"));
+      yield* sql`DELETE FROM orchestration_events WHERE sequence = ${pruned.sequence}`;
+      yield* store.append(messageEvent(threadId, "after-captured-head"));
+
+      const events = yield* store
+        .readAggregateRange({
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          fromSequenceExclusive: first.sequence,
+          toSequenceInclusive: last.sequence,
+          limit: 100,
+        })
+        .pipe(Stream.runCollect);
+      assert.deepEqual(
+        events.map((event) => event.sequence),
+        [second.sequence, last.sequence],
+      );
+    }),
+  );
+
+  it.effect("bounds thread replay metadata and counts UTF-8 bytes without decoding payloads", () =>
+    Effect.gen(function* () {
+      const store = yield* OrchestrationEventStore;
+      const sql = yield* SqlClient.SqlClient;
+      const rows = yield* sql<{ readonly sequence: number }>`
+        INSERT INTO orchestration_events (
+          event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at,
+          actor_kind, payload_json, metadata_json
+        ) VALUES
+          ('stats-1', 'thread', 'stats-thread', 0, 'thread.message-sent',
+            '2026-01-01T00:00:00.000Z', 'provider', '{"output":"😀"}', '{}'),
+          ('stats-unrelated', 'thread', 'another-thread', 0, 'thread.created',
+            '2026-01-01T00:00:00.000Z', 'provider', printf('%.*c', 10000, 'x'), '{}'),
+          ('stats-2', 'thread', 'stats-thread', 1, 'thread.activity-appended',
+            '2026-01-01T00:00:00.000Z', 'provider', '{', '{}'),
+          ('stats-other-kind', 'project', 'stats-thread', 0, 'project.deleted',
+            '2026-01-01T00:00:00.000Z', 'provider', printf('%.*c', 20000, 'x'), '{}'),
+          ('stats-3', 'thread', 'stats-thread', 2, 'thread.deleted',
+            '2026-01-01T00:00:00.000Z', 'provider', '{"output":"é"}', '{}'),
+          ('stats-4', 'thread', 'stats-thread', 3, 'thread.created',
+            '2026-01-01T00:00:00.000Z', 'provider', printf('%.*c', 2000, 'x'), '{}')
+        RETURNING sequence
+      `;
+      const range = {
+        aggregateKind: "thread" as const,
+        aggregateId: "stats-thread",
+        fromSequenceExclusive: 0,
+        toSequenceInclusive: rows.at(-1)!.sequence,
+      };
+      assert.deepEqual(yield* store.getAggregateReplayStats({ ...range, maxEvents: 2 }), {
+        eventCount: 3,
+        payloadBytes: 33,
+        hasCreateEvent: false,
+      });
+      assert.deepEqual(yield* store.getAggregateReplayStats({ ...range, maxEvents: 10 }), {
+        eventCount: 4,
+        payloadBytes: 2033,
+        hasCreateEvent: true,
+      });
+      assert.deepEqual(
+        yield* store.getAggregateReplayStats({
+          ...range,
+          toSequenceInclusive: rows[2]!.sequence,
+          maxEvents: 10,
+        }),
+        {
+          eventCount: 2,
+          payloadBytes: 18,
+          hasCreateEvent: false,
+        },
+      );
+    }),
+  );
+
+  it.effect("keeps later pages below the captured head when new events are appended", () =>
+    Effect.gen(function* () {
+      const store = yield* OrchestrationEventStore;
+      const threadId = ThreadId.make("paged-thread");
+      const persisted = yield* Effect.forEach(
+        Array.from({ length: 502 }, (_, index) => index),
+        (index) => store.append(messageEvent(threadId, `paged-${index}`)),
+      );
+      const head = persisted.at(-1)!.sequence;
+      let appendedDuringReplay = false;
+      const replayed = yield* store
+        .readAggregateRange({
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          fromSequenceExclusive: 0,
+          toSequenceInclusive: head,
+          limit: 1_000,
+        })
+        .pipe(
+          Stream.tap(() => {
+            if (appendedDuringReplay) return Effect.void;
+            appendedDuringReplay = true;
+            return store.append(messageEvent(threadId, "appended-during-replay"));
+          }),
+          Stream.runCollect,
+        );
+      assert.deepEqual(
+        replayed.map((event) => event.sequence),
+        persisted.map((event) => event.sequence),
+      );
     }),
   );
 });