Load thread snapshots over HTTP before live sync

[juliusmarminge]

Summary

• Load thread detail snapshots over HTTP first, then resume the WebSocket subscription from the returned snapshot sequence.
• Add a dedicated orchestration HTTP endpoint for thread snapshots, including 404 handling when the thread is missing.
• Update client runtime state to fall back to the socket-embedded snapshot when HTTP loading fails, while deduplicating overlapping replayed events.
• Extend shared contracts and error mapping to support the new snapshot fetch path across web, mobile, and server.
• Add tests covering HTTP snapshot seeding and live event resume behavior.

Testing

• vp check
• vp run typecheck
• vp run lint
• vp test / vp run test for the updated client-runtime state tests
• Not run: full mobile-specific lint (vp run lint:mobile)

---

Note

Medium Risk
Changes the orchestration sync path end-to-end (HTTP + WS replay ordering) and bumps on-disk thread cache schema, so upgrades force a one-time cold cache until snapshots are refetched.

Overview
Shell and thread sync now prefer loading snapshots from local cache or new HTTP endpoints, then opening WebSocket subscriptions with afterSequence so large snapshot frames are not re-sent on the socket. HTTP failures (and thread 404) fall back to the socket-embedded snapshot; overlapping replay is deduped by sequence on the client.

Server: Adds GET /api/orchestration/shell and GET /api/orchestration/threads/:threadId, getThreadDetailSnapshot (thread detail + projection sequence in one transaction), EnvironmentResourceNotFoundError, and optional limit on readEvents. subscribeShell / subscribeThread buffer live events before catch-up replay when afterSequence is set.

Clients (web/mobile): Thread cache schema v2 stores OrchestrationThreadDetailSnapshot (sequence + thread); v1 entries no longer decode. ShellSnapshotLoader / ThreadSnapshotLoader layers and shared environmentHttpAuth (cookies, Bearer, DPoP) wire into connection runtimes.

^{Reviewed by Cursor Bugbot for commit 9254fe5. Bugbot is set up for automated code reviews on this repo. Configure here.}

Note

Load thread and shell snapshots over HTTP before live WebSocket sync

• Adds HTTP endpoints GET /api/orchestration/shell and GET /api/orchestration/threads/:threadId on the server to serve snapshots before a WebSocket connection is established.
• Extends subscribeShell and subscribeThread RPCs to accept an optional afterSequence parameter; when provided, the server replays persisted events after that sequence instead of sending a full initial snapshot.
• On the client, shell and thread state initialization attempts to load a snapshot from cache or HTTP (via new ShellSnapshotLoader and ThreadSnapshotLoader services), then passes the snapshot sequence as afterSequence to the WebSocket subscription, reducing initial socket payload.
• Bumps the thread snapshot cache schema from v1 to v2, replacing the stored thread field with an OrchestrationThreadDetailSnapshot that includes snapshotSequence.
• Risk: existing v1 cache entries will not match the new schema and will be ignored, causing a one-time full sync on first load after upgrade.

^{Macroscope summarized 9254fe5.}

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: 500fc1b7-e1b1-4e76-8444-eeacea5bd39f

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch t3code/http-snapshot-loading

---

_{Comment @coderabbitai help to get the list of available commands.}

[macroscopeapp]

🟡 Medium state/threads.ts:193

makeEnvironmentThreadState now awaits snapshotLoader.load(...) before calling subscribe(...), so when the HTTP snapshot endpoint is slow or times out, live WebSocket updates are blocked for the entire duration even though the socket session is already healthy. The UI stays in stale cached/synchronizing state and misses live updates until the HTTP fallback completes. Consider subscribing to live events immediately and applying the HTTP snapshot in parallel when it arrives — applyItem already deduplicates by sequence, so overlapping events are safe.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @packages/client-runtime/src/state/threads.ts around line 193:

`makeEnvironmentThreadState` now awaits `snapshotLoader.load(...)` before calling `subscribe(...)`, so when the HTTP snapshot endpoint is slow or times out, live WebSocket updates are blocked for the entire duration even though the socket session is already healthy. The UI stays in stale `cached`/`synchronizing` state and misses live updates until the HTTP fallback completes. Consider subscribing to live events immediately and applying the HTTP snapshot in parallel when it arrives — `applyItem` already deduplicates by sequence, so overlapping events are safe.

[juliusmarminge]

Mitigated in c420e9e. The snapshot fetch requires the sequence before we can resume live events over the socket, so some coupling is inherent, but: the cached thread renders while the fetch runs (so first paint is not blocked), the wait is bounded (timeout lowered to 6s with fallback to the socket snapshot), and this only affects the initial mount — reconnects resume via the socket without re-fetching.

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.

Autofix Details

Bugbot Autofix prepared fixes for all 3 issues found in the latest run.

• ✅ Fixed: PubSub attached after replay
  • Added OrchestrationEngine.subscribeDomainEvents, which attaches the PubSub subscription (scoped to the RPC stream's lifetime) before subscribeThread reads its catch-up or snapshot baseline, so events published during the baseline read are buffered instead of dropped.
• ✅ Fixed: Thread catch-up hits replay cap
  • readEvents now accepts an optional limit forwarded to the event store, and the afterSequence catch-up passes Number.MAX_SAFE_INTEGER so the replay is exhaustive (internally paged) instead of truncating at the global 1,000-event default.
• ✅ Fixed: HTTP snapshot reads race
  • Added a transactional ProjectionSnapshotQuery.getThreadDetailSnapshot that reads the thread detail and snapshot sequence in one sql.withTransaction, now used by both the HTTP threadSnapshot endpoint and the WS snapshot path so the sequence never runs ahead of the embedded thread.

Or push these changes by commenting:

@cursor push 29677d1179

Preview (29677d1179)

diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
--- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
+++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
@@ -107,6 +107,7 @@
               }),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -199,6 +200,7 @@
             getFullThreadDiffContext: () => Effect.die("unused"),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -281,6 +283,7 @@
             getFullThreadDiffContext: () => Effect.die("unused"),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -348,6 +351,7 @@
             getFullThreadDiffContext: () => Effect.die("unused"),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );
@@ -400,6 +404,7 @@
             getFullThreadDiffContext: () => Effect.succeed(Option.none()),
             getThreadShellById: () => Effect.succeed(Option.none()),
             getThreadDetailById: () => Effect.succeed(Option.none()),
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
           }),
         ),
       );

diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts
--- a/apps/server/src/observability/RpcInstrumentation.ts
+++ b/apps/server/src/observability/RpcInstrumentation.ts
@@ -5,6 +5,7 @@
 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 @@
   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;

diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
--- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
+++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
@@ -200,6 +200,7 @@
           getFullThreadDiffContext: () => Effect.succeed(Option.none()),
           getThreadShellById: () => Effect.succeed(Option.none()),
           getThreadDetailById: () => Effect.succeed(Option.none()),
+          getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
         }),
       ),
       Layer.provide(

diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts
--- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts
+++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts
@@ -306,8 +306,8 @@
     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 @@
     get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] {
       return Stream.fromPubSub(eventPubSub);
     },
+    subscribeDomainEvents: Effect.map(PubSub.subscribe(eventPubSub), Stream.fromSubscription),
   } satisfies OrchestrationEngineShape;
 });
 

diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
--- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
+++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
@@ -2033,6 +2033,23 @@
       );
     });
 
+  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 @@
     getFullThreadDiffContext,
     getThreadShellById,
     getThreadDetailById,
+    getThreadDetailSnapshot,
   } satisfies ProjectionSnapshotQueryShape;
 });
 

diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts
--- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts
+++ b/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 @@
    * 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 @@
    * 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
+  >;
 }
 
 /**

diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts
--- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts
+++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts
@@ -14,6 +14,7 @@
   OrchestrationReadModel,
   OrchestrationShellSnapshot,
   OrchestrationThread,
+  OrchestrationThreadDetailSnapshot,
   OrchestrationThreadShell,
   ProjectId,
   ThreadId,
@@ -157,6 +158,15 @@
   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>;
 }
 
 /**

diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts
--- a/apps/server/src/orchestration/http.ts
+++ b/apps/server/src/orchestration/http.ts
@@ -45,18 +45,17 @@
         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(

diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts
--- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts
+++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts
@@ -43,6 +43,7 @@
     getFullThreadDiffContext: () => Effect.die("unused"),
     getThreadShellById: () => Effect.die("unused"),
     getThreadDetailById: () => Effect.die("unused"),
+    getThreadDetailSnapshot: () => Effect.die("unused"),
   });
 
 const makeTerminalManagerLayer = (

diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts
--- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts
+++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts
@@ -209,6 +209,7 @@
                 : Option.none(),
             ),
           getThreadDetailById: () => Effect.die("unused"),
+          getThreadDetailSnapshot: () => Effect.die("unused"),
         }),
       ),
       Layer.provideMerge(NodeServices.layer),

diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts
--- a/apps/server/src/relay/AgentAwarenessRelay.test.ts
+++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts
@@ -461,6 +461,7 @@
           readEvents: () => Stream.empty,
           dispatch: () => Effect.succeed({ sequence: 1 }),
           streamDomainEvents: Stream.fromQueue(events),
+          subscribeDomainEvents: Effect.sync(() => Stream.fromQueue(events)),
         } satisfies OrchestrationEngineShape;
 
         const snapshotQuery = {
@@ -650,6 +651,7 @@
             readEvents: () => Stream.empty,
             dispatch: () => Effect.succeed({ sequence: 1 }),
             streamDomainEvents: Stream.fromQueue(events),
+            subscribeDomainEvents: Effect.sync(() => Stream.fromQueue(events)),
           } satisfies OrchestrationEngineShape),
           Layer.succeed(ProjectionSnapshotQuery, {
             getShellSnapshot: () =>

diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts
--- a/apps/server/src/serverRuntimeStartup.test.ts
+++ b/apps/server/src/serverRuntimeStartup.test.ts
@@ -96,6 +96,7 @@
           getFullThreadDiffContext: () => Effect.succeed(Option.none()),
           getThreadShellById: () => Effect.succeed(Option.none()),
           getThreadDetailById: () => Effect.succeed(Option.none()),
+          getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
         }),
         Effect.provideService(AnalyticsService.AnalyticsService, {
           record: () => Effect.void,
@@ -158,6 +159,7 @@
         getFullThreadDiffContext: () => Effect.succeed(Option.none()),
         getThreadShellById: () => Effect.die("unused"),
         getThreadDetailById: () => Effect.die("unused"),
+        getThreadDetailSnapshot: () => Effect.die("unused"),
       }),
       Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
         readEvents: () => Stream.empty,
@@ -166,6 +168,7 @@
             Effect.as({ sequence: 1 }),
           ),
         streamDomainEvents: Stream.empty,
+        subscribeDomainEvents: Effect.succeed(Stream.empty),
       } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
       Effect.provide(NodeServices.layer),
     );
@@ -200,6 +203,7 @@
         getFullThreadDiffContext: () => Effect.succeed(Option.none()),
         getThreadShellById: () => Effect.die("unused"),
         getThreadDetailById: () => Effect.die("unused"),
+        getThreadDetailSnapshot: () => Effect.die("unused"),
       }),
       Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
         readEvents: () => Stream.empty,
@@ -208,6 +212,7 @@
             Effect.as({ sequence: 1 }),
           ),
         streamDomainEvents: Stream.empty,
+        subscribeDomainEvents: Effect.succeed(Stream.empty),
       } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
       Effect.provide(NodeServices.layer),
     );
@@ -248,6 +253,7 @@
         getFullThreadDiffContext: () => Effect.succeed(Option.none()),
         getThreadShellById: () => Effect.die("unused"),
         getThreadDetailById: () => Effect.die("unused"),
+        getThreadDetailSnapshot: () => Effect.die("unused"),
       }),
       Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
         readEvents: () => Stream.empty,
@@ -256,6 +262,7 @@
             Effect.as({ sequence: 1 }),
           ),
         streamDomainEvents: Stream.empty,
+        subscribeDomainEvents: Effect.succeed(Stream.empty),
       } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
       Effect.provideService(Crypto.Crypto, {
         ...crypto,

diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts
--- a/apps/server/src/ws.ts
+++ b/apps/server/src/ws.ts
@@ -1119,7 +1119,12 @@
                 event.aggregateId === input.threadId &&
                 isThreadDetailEvent(event);
 
-              const liveStream = orchestrationEngine.streamDomainEvents.pipe(
+              // Attach the domain event subscription before loading the
+              // snapshot/catch-up baseline: the PubSub does not buffer for
+              // late subscribers, so events published while the baseline is
+              // read would otherwise be dropped. Overlapping events are
+              // deduped by sequence on the client.
+              const liveStream = (yield* orchestrationEngine.subscribeDomainEvents).pipe(
                 Stream.filter(isThisThreadDetailEvent),
                 Stream.map((event) => ({
                   kind: "event" as const,
@@ -1131,26 +1136,29 @@
               // that snapshot's sequence, and we resume the live subscription by
               // replaying persisted events after it instead of re-sending the
               // (potentially multi-KB) snapshot frame over the socket. The
-              // catch-up replay + live stream carry the same ordering guarantees
-              // as the snapshot-then-live path below; overlapping events are
-              // deduped by sequence on the client.
+              // catch-up replay must be exhaustive — a truncated replay would
+              // silently drop thread updates — so it bypasses the default
+              // replay cap.
               if (input.afterSequence !== undefined) {
-                const catchUpStream = orchestrationEngine.readEvents(input.afterSequence).pipe(
-                  Stream.filter(isThisThreadDetailEvent),
-                  Stream.map((event) => ({ kind: "event" as const, event })),
-                  Stream.mapError(
-                    (cause) =>
-                      new OrchestrationGetSnapshotError({
-                        message: `Failed to replay thread ${input.threadId} events`,
-                        cause,
-                      }),
-                  ),
-                );
+                const catchUpStream = orchestrationEngine
+                  .readEvents(input.afterSequence, Number.MAX_SAFE_INTEGER)
+                  .pipe(
+                    Stream.filter(isThisThreadDetailEvent),
+                    Stream.map((event) => ({ kind: "event" as const, event })),
+                    Stream.mapError(
+                      (cause) =>
+                        new OrchestrationGetSnapshotError({
+                          message: `Failed to replay thread ${input.threadId} events`,
+                          cause,
+                        }),
+                    ),
+                  );
                 return Stream.concat(catchUpStream, liveStream);
               }
 
-              const [threadDetail, snapshotSequence] = yield* Effect.all([
-                projectionSnapshotQuery.getThreadDetailById(input.threadId).pipe(
+              const threadSnapshot = yield* projectionSnapshotQuery
+                .getThreadDetailSnapshot(input.threadId)
+                .pipe(
                   Effect.mapError(
                     (cause) =>
                       new OrchestrationGetSnapshotError({
@@ -1158,20 +1166,9 @@
                         cause,
                       }),
                   ),
-                ),
-                projectionSnapshotQuery.getSnapshotSequence().pipe(
-                  Effect.map(({ snapshotSequence }) => snapshotSequence),
-                  Effect.mapError(
-                    (cause) =>
-                      new OrchestrationGetSnapshotError({
-                        message: "Failed to load orchestration snapshot sequence",
-                        cause,
-                      }),
-                  ),
-                ),
-              ]);
+                );
 
-              if (Option.isNone(threadDetail)) {
+              if (Option.isNone(threadSnapshot)) {
                 return yield* new OrchestrationGetSnapshotError({
                   message: `Thread ${input.threadId} was not found`,
                   cause: input.threadId,
@@ -1181,10 +1178,7 @@
               return Stream.concat(
                 Stream.make({
                   kind: "snapshot" as const,
-                  snapshot: {
-                    snapshotSequence,
-                    thread: threadDetail.value,
-                  },
+                  snapshot: threadSnapshot.value,
                 }),
                 liveStream,
               );

_{You can send follow-ups to the cloud agent here.}

[macroscopeapp]

Approvability

Verdict: Needs human review

1 blocking correctness issue found. This PR introduces new HTTP snapshot loading that changes the client-server sync flow. Two unresolved review comments identify potential correctness issues: live updates may be blocked during HTTP fetch, and shell catch-up may use incorrect state. These substantive concerns combined with the new feature scope warrant human review.

^{You can customize Macroscope's approvability policy. Learn more.}

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

^{Bugbot Autofix is ON. A cloud agent has been kicked off to fix the reported issue. You can view the agent here.}

^{Reviewed by Cursor Bugbot for commit 9bc191b. Configure here.}

[cursor]

Shell catch-up uses current DB

Medium Severity

When subscribeShell resumes with afterSequence, catch-up replays persisted events through toShellStreamEvent, which loads project/thread shells from the current projection (getProjectShellById / getThreadShellById). Those reads reflect today’s state, not the state at each replayed sequence, so the client can apply the wrong shell data (or drop upserts for deleted rows) while still advancing snapshotSequence.

Additional Locations (1)

• apps/server/src/ws.ts#L616-L677

 

^{Reviewed by Cursor Bugbot for commit 9bc191b. Configure here.}

[cursor]

Bugbot Autofix determined this is a false positive.

The shell stream replicates current state (whole-row upserts deduped by monotonic sequence), and because event append and projection update commit in one transaction, replayed events always carry data at a version >= their sequence while every deletion/archive emits its own payload-derived removal event later in the same replay, so the client provably converges to the server's current shell with no dropped changes — the same read-current-projection semantics the pre-existing live path already uses.

_{You can send follow-ups to the cloud agent here.}