Fix thread subscription event loss and snapshot sequence race

- Attach the domain-event PubSub subscription before reading the
  catch-up replay or snapshot baseline in subscribeThread, so events
  published while the baseline loads are buffered instead of dropped
  (new OrchestrationEngine.subscribeDomainEvents, scoped to the RPC
  stream's lifetime via observeRpcStreamEffect).
- Make the afterSequence catch-up replay exhaustive instead of
  truncating at the event store's default 1,000-event cap, which could
  silently drop thread events under multi-thread activity.
- Read the thread detail and projection snapshot sequence inside one
  transaction (ProjectionSnapshotQuery.getThreadDetailSnapshot), shared
  by the HTTP threadSnapshot endpoint and the WS snapshot path, so the
  reported sequence can never run ahead of the embedded thread.

Author: cursoragent

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

@@ -107,6 +107,7 @@ describe("CheckpointDiffQuery.layer", () => {
               }),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -199,6 +200,7 @@ describe("CheckpointDiffQuery.layer", () => {
             getFullThreadDiffContext: () => Effect.die("unused"),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -281,6 +283,7 @@ describe("CheckpointDiffQuery.layer", () => {
             getFullThreadDiffContext: () => Effect.die("unused"),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -348,6 +351,7 @@ describe("CheckpointDiffQuery.layer", () => {
             getFullThreadDiffContext: () => Effect.die("unused"),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -400,6 +404,7 @@ describe("CheckpointDiffQuery.layer", () => {
             getFullThreadDiffContext: () => Effect.succeed(Option.none()),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );

apps/server/src/observability/RpcInstrumentation.ts

@@ -5,6 +5,7 @@ import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Metric from "effect/Metric";
 import * as References from "effect/References";
+import type * as Scope from "effect/Scope";
 import * as Stream from "effect/Stream";
 
 import { outcomeFromExit } from "./Attributes.ts";
@@ -123,7 +124,14 @@ export const observeRpcStreamEffect = <A, StreamError, StreamContext, EffectErro
   method: string,
   effect: Effect.Effect<Stream.Stream<A, StreamError, StreamContext>, EffectError, EffectContext>,
   traceAttributes?: Readonly<Record<string, unknown>>,
-): Stream.Stream<A, StreamError | EffectError, StreamContext | EffectContext> => {
+  // `Stream.unwrap` scopes the setup effect to the stream's lifetime, so a
+  // `Scope` requirement (e.g. PubSub subscriptions attached before a snapshot
+  // is loaded) is satisfied by the stream itself rather than the caller.
+): Stream.Stream<
+  A,
+  StreamError | EffectError,
+  StreamContext | Exclude<EffectContext, Scope.Scope>
+> => {
   const instrumented = Stream.unwrap(
     Effect.gen(function* () {
       const startedAt = yield* Clock.currentTimeNanos;

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

@@ -200,6 +200,7 @@ describe("OrchestrationEngine", () => {
           getFullThreadDiffContext: () => Effect.succeed(Option.none()),
           getThreadShellById: () => Effect.succeed(Option.none()),
           getThreadDetailById: () => Effect.succeed(Option.none()),
+          getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
         }),
       ),
       Layer.provide(

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

@@ -306,8 +306,8 @@ const makeOrchestrationEngine = Effect.gen(function* () {
     Effect.annotateLogs({ sequence: commandReadModel.snapshotSequence }),
   );
 
-  const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive) =>
-    eventStore.readFromSequence(fromSequenceExclusive);
+  const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
+    eventStore.readFromSequence(fromSequenceExclusive, limit);
 
   const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
     Effect.gen(function* () {
@@ -329,6 +329,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
     get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] {
       return Stream.fromPubSub(eventPubSub);
     },
+    subscribeDomainEvents: Effect.map(PubSub.subscribe(eventPubSub), Stream.fromSubscription),
   } satisfies OrchestrationEngineShape;
 });
 

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

@@ -2033,6 +2033,23 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
       );
     });
 
+  const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = (
+    threadId,
+  ) =>
+    sql.withTransaction(Effect.all([getThreadDetailById(threadId), getSnapshotSequence()])).pipe(
+      Effect.map(([threadDetail, { snapshotSequence }]) =>
+        Option.map(threadDetail, (thread) => ({ snapshotSequence, thread })),
+      ),
+      Effect.mapError((error) => {
+        if (isPersistenceError(error)) {
+          return error;
+        }
+        return toPersistenceSqlError("ProjectionSnapshotQuery.getThreadDetailSnapshot:query")(
+          error,
+        );
+      }),
+    );
+
   return {
     getCommandReadModel,
     getSnapshot,
@@ -2047,6 +2064,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
     getFullThreadDiffContext,
     getThreadShellById,
     getThreadDetailById,
+    getThreadDetailSnapshot,
   } satisfies ProjectionSnapshotQueryShape;
 });
 

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

@@ -13,6 +13,7 @@
 import type { OrchestrationCommand, OrchestrationEvent } from "@t3tools/contracts";
 import * as Context from "effect/Context";
 import type * as Effect from "effect/Effect";
+import type * as Scope from "effect/Scope";
 import type * as Stream from "effect/Stream";
 
 import type { OrchestrationDispatchError } from "../Errors.ts";
@@ -26,10 +27,14 @@ export interface OrchestrationEngineShape {
    * Replay persisted orchestration events from an exclusive sequence cursor.
    *
    * @param fromSequenceExclusive - Sequence cursor (exclusive).
+   * @param limit - Optional maximum number of events to replay. Defaults to
+   *   the event store's replay cap; pass `Number.MAX_SAFE_INTEGER` for an
+   *   exhaustive catch-up replay.
    * @returns Stream containing ordered events.
    */
   readonly readEvents: (
     fromSequenceExclusive: number,
+    limit?: number,
   ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
 
   /**
@@ -49,8 +54,26 @@ export interface OrchestrationEngineShape {
    * Stream persisted domain events in dispatch order.
    *
    * This is a hot runtime stream (new events only), not a historical replay.
+   * The underlying PubSub subscription is only attached once the stream is
+   * pulled; use `subscribeDomainEvents` when the subscription must be
+   * attached before other work (e.g. loading a snapshot baseline).
    */
   readonly streamDomainEvents: Stream.Stream<OrchestrationEvent>;
+
+  /**
+   * Attach a domain event subscription immediately and return the stream of
+   * events it receives.
+   *
+   * Unlike `streamDomainEvents`, events published between running this effect
+   * and pulling the returned stream are buffered by the subscription instead
+   * of dropped, which is required for snapshot/catch-up + live combinations.
+   * The subscription is released when the surrounding scope closes.
+   */
+  readonly subscribeDomainEvents: Effect.Effect<
+    Stream.Stream<OrchestrationEvent>,
+    never,
+    Scope.Scope
+  >;
 }
 
 /**

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

@@ -14,6 +14,7 @@ import type {
   OrchestrationReadModel,
   OrchestrationShellSnapshot,
   OrchestrationThread,
+  OrchestrationThreadDetailSnapshot,
   OrchestrationThreadShell,
   ProjectId,
   ThreadId,
@@ -157,6 +158,15 @@ export interface ProjectionSnapshotQueryShape {
   readonly getThreadDetailById: (
     threadId: ThreadId,
   ) => Effect.Effect<Option.Option<OrchestrationThread>, ProjectionRepositoryError>;
+
+  /**
+   * Read a single active thread detail together with the projection snapshot
+   * sequence, both observed inside one transaction so the sequence never runs
+   * ahead of the thread rows it is paired with.
+   */
+  readonly getThreadDetailSnapshot: (
+    threadId: ThreadId,
+  ) => Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>, ProjectionRepositoryError>;
 }
 
 /**

apps/server/src/orchestration/http.ts

@@ -45,18 +45,17 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group(
         Effect.fn("environment.orchestration.threadSnapshot")(function* (args) {
           yield* annotateEnvironmentRequest(args.endpoint.name);
           yield* requireEnvironmentScope(AuthOrchestrationReadScope);
-          const [threadDetail, { snapshotSequence }] = yield* Effect.all([
-            projectionSnapshotQuery.getThreadDetailById(args.params.threadId),
-            projectionSnapshotQuery.getSnapshotSequence(),
-          ]).pipe(
-            Effect.catch((cause) =>
-              failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
-            ),
-          );
-          if (Option.isNone(threadDetail)) {
+          const snapshot = yield* projectionSnapshotQuery
+            .getThreadDetailSnapshot(args.params.threadId)
+            .pipe(
+              Effect.catch((cause) =>
+                failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
+              ),
+            );
+          if (Option.isNone(snapshot)) {
             return yield* failEnvironmentNotFound("thread_not_found");
           }
-          return { snapshotSequence, thread: threadDetail.value };
+          return snapshot.value;
         }),
       )
       .handle(

apps/server/src/project/ProjectSetupScriptRunner.test.ts

@@ -43,6 +43,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
     getFullThreadDiffContext: () => Effect.die("unused"),
     getThreadShellById: () => Effect.die("unused"),
     getThreadDetailById: () => Effect.die("unused"),
+    getThreadDetailSnapshot: () => Effect.die("unused"),
   });
 
 const makeTerminalManagerLayer = (

apps/server/src/provider/Layers/ProviderSessionReaper.test.ts

@@ -209,6 +209,7 @@ describe("ProviderSessionReaper", () => {
                 : Option.none(),
             ),
           getThreadDetailById: () => Effect.die("unused"),
+          getThreadDetailSnapshot: () => Effect.die("unused"),
         }),
       ),
       Layer.provideMerge(NodeServices.layer),