fix(server): replay command events across persistence pages

apps/server/src/orchestration-v2/FoundationPersistence.test.ts

@@ -1149,6 +1149,166 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => {
       }),
   );
 
+  it.effect("replays every command event across bounded persistence pages", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      const eventStore = yield* EventStoreV2;
+      const eventSink = yield* EventSinkV2;
+      const now = yield* DateTime.now;
+      const occurredAt = DateTime.formatIso(now);
+      const commandId = CommandId.make("command:paged-command-replay");
+      const exactPageCommandId = CommandId.make("command:paged-command-replay:exact");
+      const emptyCommandId = CommandId.make("command:paged-command-replay:empty");
+      const threadId = "thread:paged-command-replay";
+
+      const eventRow = (
+        ordinal: number,
+        input: {
+          readonly commandId?: string | null;
+          readonly aggregateKind?: "project" | "thread";
+          readonly streamId?: string;
+          readonly version?: number;
+        } = {},
+      ) => ({
+        event_id: `event:paged-command-replay:${ordinal}`,
+        aggregate_kind: input.aggregateKind ?? "thread",
+        stream_id: input.streamId ?? threadId,
+        stream_version: ordinal,
+        event_type: "provider-session.detached",
+        occurred_at: occurredAt,
+        command_id: input.commandId ?? null,
+        causation_event_id: null,
+        correlation_id: null,
+        actor_kind: "server",
+        payload_json: JSON.stringify({
+          providerSessionId: `session:paged-command-replay:${ordinal}`,
+          detachedAt: occurredAt,
+        }),
+        metadata_json: "{}",
+        application_event_version: input.version ?? 2,
+      });
+
+      const matchingCount = 1_001;
+      const rows: Array<ReturnType<typeof eventRow>> = [];
+      let ordinal = 0;
+      for (let index = 0; index < matchingCount; index += 1) {
+        rows.push(eventRow(++ordinal, { commandId }));
+        if (index % 2 === 0) {
+          rows.push(eventRow(++ordinal, { commandId: "command:unrelated" }));
+        }
+        if (index % 5 === 0) {
+          rows.push(eventRow(++ordinal));
+          rows.push(eventRow(++ordinal, { commandId, version: 1 }));
+          rows.push(
+            eventRow(++ordinal, {
+              commandId,
+              aggregateKind: "project",
+              streamId: `project:paged-command-replay:${index}`,
+            }),
+          );
+        }
+      }
+      for (let index = 0; index < 500; index += 1) {
+        rows.push(eventRow(++ordinal, { commandId: exactPageCommandId }));
+      }
+      const inserted = yield* Effect.forEach(
+        Array.from({ length: Math.ceil(rows.length / 400) }, (_, chunk) =>
+          rows.slice(chunk * 400, (chunk + 1) * 400),
+        ),
+        (chunk) =>
+          sql<{
+            readonly sequence: number;
+            readonly command_id: string | null;
+            readonly aggregate_kind: string;
+            readonly application_event_version: number;
+          }>`
+            INSERT INTO orchestration_events ${sql.insert(chunk)}
+            RETURNING sequence, command_id, aggregate_kind, application_event_version
+          `,
+        { concurrency: 1 },
+      ).pipe(Effect.map((chunks) => chunks.flat()));
+      const expectedSequences = inserted
+        .filter(
+          (row) =>
+            row.command_id === commandId &&
+            row.aggregate_kind === "thread" &&
+            row.application_event_version === 2,
+        )
+        .map((row) => row.sequence);
+      assert.lengthOf(expectedSequences, matchingCount);
+
+      yield* sql`
+        INSERT INTO orchestration_command_receipts (
+          command_id, aggregate_kind, aggregate_id, accepted_at, result_sequence, status, command_type
+        )
+        VALUES (
+          ${commandId}, 'thread', ${threadId}, ${occurredAt},
+          ${expectedSequences.at(-1)!}, 'accepted', 'thread.create'
+        )
+      `;
+
+      const pageLimits: Array<number> = [];
+      const recordReads: Statement.Transformer = (statement) => {
+        const [query, params] = statement.compile();
+        if (query.includes("FROM orchestration_events") && query.includes("command_id")) {
+          const limit = params.at(-1);
+          if (typeof limit === "number") {
+            pageLimits.push(limit);
+          }
+        }
+        return Effect.succeed(statement);
+      };
+      const collectByCommandId = (id: CommandId) =>
+        eventStore.readByCommandId({ commandId: id }).pipe(
+          Stream.provideService(Statement.CurrentTransformer, recordReads),
+          Stream.runCollect,
+          Effect.map((events) => Array.from(events)),
+        );
+
+      const replayed = yield* collectByCommandId(commandId);
+      assert.deepEqual(
+        replayed.map((stored) => stored.sequence),
+        expectedSequences,
+      );
+      assert.deepEqual(pageLimits, [500, 500, 500]);
+
+      pageLimits.length = 0;
+      const exactPage = yield* collectByCommandId(exactPageCommandId);
+      assert.lengthOf(exactPage, 500);
+      assert.deepEqual(pageLimits, [500, 500]);
+
+      pageLimits.length = 0;
+      const empty = yield* collectByCommandId(emptyCommandId);
+      assert.lengthOf(empty, 0);
+      assert.deepEqual(pageLimits, [500]);
+
+      pageLimits.length = 0;
+      const retried = yield* eventSink
+        .commitCommand({
+          commandId,
+          threadId: ThreadId.make(threadId),
+          commandType: "thread.create",
+          acceptedAt: now,
+          events: [
+            threadCreatedEvent({
+              id: "event:paged-command-replay:retry",
+              thread: makeThread(ThreadId.make(threadId), now),
+              now,
+            }),
+          ],
+          effects: [],
+        })
+        .pipe(Effect.provideService(Statement.CurrentTransformer, recordReads));
+      assert.isFalse(retried.committed);
+      assert.equal(retried.receipt.resultSequence, expectedSequences.at(-1));
+      assert.deepEqual(
+        retried.storedEvents.map((stored) => stored.sequence),
+        expectedSequences,
+      );
+      assert.deepEqual(pageLimits, [500, 500, 500]);
+    }),
+  );
+
   it.effect("keeps one durable effect across command retries and executes it after recovery", () =>
     Effect.gen(function* () {
       const eventSink = yield* EventSinkV2;

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

@@ -470,20 +470,43 @@ const makeEventStore = Effect.gen(function* () {
       ),
     );
 
-  const readAgentEvents: OrchestrationEventStoreShape["readAgentEvents"] = (input) =>
-    Stream.fromEffect(
-      readApplicationRows({
-        afterSequence: input?.afterSequence ?? 0,
-        ...(input?.throughSequence === undefined ? {} : { throughSequence: input.throughSequence }),
-        ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
-        ...(input?.commandId === undefined ? {} : { commandId: input.commandId }),
-        onlyAgentEvents: true,
-        limit: input?.limit ?? DEFAULT_READ_FROM_SEQUENCE_LIMIT,
-      }).pipe(
-        Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")),
-      ),
+  const readAgentEvents: OrchestrationEventStoreShape["readAgentEvents"] = (input) => {
+    const totalLimit =
+      input?.limit === undefined ? Number.MAX_SAFE_INTEGER : Math.max(0, Math.floor(input.limit));
+    if (totalLimit === 0) {
+      return Stream.empty;
+    }
+    return Stream.paginate(
+      { cursor: input?.afterSequence ?? 0, remaining: totalLimit },
+      ({ cursor, remaining }) => {
+        const pageLimit = Math.min(remaining, READ_PAGE_SIZE);
+        return readApplicationRows({
+          afterSequence: cursor,
+          ...(input?.throughSequence === undefined
+            ? {}
+            : { throughSequence: input.throughSequence }),
+          ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
+          ...(input?.commandId === undefined ? {} : { commandId: input.commandId }),
+          onlyAgentEvents: true,
+          limit: pageLimit,
+        }).pipe(
+          Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")),
+          Effect.map((rows) => {
+            const last = rows.at(-1);
+            const nextRemaining = remaining - rows.length;
+            return [
+              rows,
+              last === undefined ||
+              rows.length < pageLimit ||
+              nextRemaining <= 0 ||
+              (input?.throughSequence !== undefined && last.sequence >= input.throughSequence)
+                ? Option.none()
+                : Option.some({ cursor: last.sequence, remaining: nextRemaining }),
+            ] as const;
+          }),
+        );
+      },
     ).pipe(
-      Stream.flatMap(Stream.fromIterable),
       Stream.mapEffect((row) =>
         rowToV2StoredEvent(row).pipe(
           Effect.mapError(
@@ -492,6 +515,7 @@ const makeEventStore = Effect.gen(function* () {
         ),
       ),
     );
+  };
 
   const getAgentReplayStats: OrchestrationEventStoreShape["getAgentReplayStats"] = (input) =>
     sql<{

apps/server/src/persistence/Services/OrchestrationEventStore.ts

@@ -67,7 +67,12 @@ export interface OrchestrationEventStoreShape {
     readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
   }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, OrchestrationEventStoreError>;
 
-  /** Read only V2 thread events from the application log. */
+  /**
+   * Read only V2 thread events from the application log.
+   *
+   * Reads in fixed-size sequence pages until the filtered range is exhausted;
+   * `limit` caps the total emitted events across pages.
+   */
   readonly readAgentEvents: (input?: {
     readonly afterSequence?: number;
     readonly throughSequence?: number;