fix(sync): stop the shell thread list from wedging on a stale cache

apps/server/src/server.test.ts

@@ -5624,6 +5624,116 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
     }).pipe(Effect.provide(NodeHttpServer.layerTest)),
   );
 
+  // A resuming shell client is only replayed when the replay is both possible
+  // and cheaper than the snapshot it stands in for. See
+  // SHELL_CATCH_UP_REPLAY_LIMIT in ws.ts.
+  const subscribeShellResume = (input: {
+    readonly afterSequence: number;
+    readonly projectedSequence: number;
+    readonly onReadEvents: () => void;
+  }) =>
+    Effect.gen(function* () {
+      yield* buildAppUnderTest({
+        layers: {
+          orchestrationEngine: {
+            readEvents: (_fromSequenceExclusive, _limit) =>
+              Stream.suspend(() => {
+                input.onReadEvents();
+                // `project.deleted` maps to a shell event without any further
+                // projection lookup, so it isolates the replay-vs-snapshot
+                // decision from the rest of the shell mapping.
+                return Stream.make({
+                  sequence: input.afterSequence + 1,
+                  eventId: EventId.make("event-replayed"),
+                  aggregateKind: "project",
+                  aggregateId: defaultProjectId,
+                  occurredAt: "2026-04-05T00:00:00.000Z",
+                  commandId: null,
+                  causationEventId: null,
+                  correlationId: null,
+                  metadata: {},
+                  type: "project.deleted",
+                  payload: {
+                    projectId: defaultProjectId,
+                    deletedAt: "2026-04-05T00:00:00.000Z",
+                  },
+                } satisfies Extract<OrchestrationEvent, { type: "project.deleted" }>);
+              }),
+          },
+          projectionSnapshotQuery: {
+            getSnapshotSequence: () =>
+              Effect.succeed({ snapshotSequence: input.projectedSequence }),
+          },
+        },
+      });
+
+      const wsUrl = yield* getWsServerUrl("/ws");
+      const items = yield* Effect.scoped(
+        withWsRpcClient(wsUrl, (client) =>
+          client[ORCHESTRATION_WS_METHODS.subscribeShell]({
+            afterSequence: input.afterSequence,
+          }).pipe(Stream.take(1), Stream.runCollect),
+        ),
+      );
+      return items[0];
+    });
+
+  it.effect("replays shell events for a client resuming from a recent sequence", () =>
+    Effect.gen(function* () {
+      let readEventsCalls = 0;
+      const first = yield* subscribeShellResume({
+        afterSequence: 10,
+        projectedSequence: 12,
+        onReadEvents: () => {
+          readEventsCalls += 1;
+        },
+      });
+
+      assert.equal(readEventsCalls, 1);
+      assert.equal(first?.kind, "project-removed");
+    }).pipe(Effect.provide(NodeHttpServer.layerTest)),
+  );
+
+  it.effect("sends a shell snapshot instead of replaying when the client is far behind", () =>
+    Effect.gen(function* () {
+      let readEventsCalls = 0;
+      const first = yield* subscribeShellResume({
+        afterSequence: 10,
+        // Far enough behind that replaying the window would cost more than the
+        // snapshot, and slow enough to send that the client may never finish it.
+        projectedSequence: 100_000,
+        onReadEvents: () => {
+          readEventsCalls += 1;
+        },
+      });
+
+      assert.equal(readEventsCalls, 0);
+      assert.equal(first?.kind, "snapshot");
+    }).pipe(Effect.provide(NodeHttpServer.layerTest)),
+  );
+
+  it.effect(
+    "sends a shell snapshot when the client cursor is ahead of the projected sequence",
+    () =>
+      Effect.gen(function* () {
+        let readEventsCalls = 0;
+        // A cursor past our own head means the client cached against an event log
+        // we no longer have (rebuilt, restored, or a different server). Replaying
+        // would send nothing and every live event would be dropped by the client's
+        // sequence check, leaving it stale forever.
+        const first = yield* subscribeShellResume({
+          afterSequence: 500,
+          projectedSequence: 12,
+          onReadEvents: () => {
+            readEventsCalls += 1;
+          },
+        });
+
+        assert.equal(readEventsCalls, 0);
+        assert.equal(first?.kind, "snapshot");
+      }).pipe(Effect.provide(NodeHttpServer.layerTest)),
+  );
+
   it.effect("enriches replayed project events with repository identity metadata", () =>
     Effect.gen(function* () {
       const repositoryIdentity = {

apps/server/src/ws.ts

@@ -282,6 +282,20 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract<
 
 const PROVIDER_STATUS_DEBOUNCE_MS = 200;
 
+/**
+ * How far behind a resuming shell client may be before we send it a fresh
+ * snapshot instead of replaying its catch-up window.
+ *
+ * Replay is only a win while the gap is small. Every thread-aggregate event in
+ * the window costs a `getThreadShellById` query and emits a full thread shell
+ * on the socket, so a client that has been away for a few days can ask for tens
+ * of megabytes to rebuild a list that a single snapshot expresses in a few
+ * hundred kilobytes. Worse, a client that cannot afford the replay never
+ * advances its cursor, so its next attempt is just as expensive — it stays
+ * stale indefinitely.
+ */
+const SHELL_CATCH_UP_REPLAY_LIMIT = 500;
+
 const RPC_REQUIRED_SCOPE = new Map<string, AuthEnvironmentScope>([
   [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope],
   [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope],
@@ -1084,6 +1098,19 @@ const makeWsRpcLayer = (
                 ),
               );
 
+              const loadShellSnapshot = projectionSnapshotQuery.getShellSnapshot().pipe(
+                Effect.tapError((cause) =>
+                  Effect.logError("orchestration shell snapshot load failed", { cause }),
+                ),
+                Effect.mapError(
+                  (cause) =>
+                    new OrchestrationGetSnapshotError({
+                      message: "Failed to load orchestration shell snapshot",
+                      cause,
+                    }),
+                ),
+              );
+
               // When the client already holds a shell snapshot (cached, or loaded
               // over HTTP) it passes that snapshot's sequence, and we resume by
               // replaying shell events after it instead of re-sending the whole
@@ -1101,6 +1128,38 @@ const makeWsRpcLayer = (
                     yield* Effect.forkScoped(
                       liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
                     );
+
+                    // A resume cursor is only usable if we can actually serve it.
+                    // If it sits ahead of our projected sequence the client is
+                    // resuming against an event log that no longer matches its
+                    // cache (rebuilt, restored from backup, or a different
+                    // server), and every live event we send will be discarded by
+                    // its sequence check — it would render its stale cache
+                    // forever. If it sits too far behind, the replay costs far
+                    // more than the snapshot it stands in for. Both cases are
+                    // repaired by sending a snapshot, which the client applies
+                    // wholesale rather than merging by sequence.
+                    const projectedSequence = yield* projectionSnapshotQuery
+                      .getSnapshotSequence()
+                      .pipe(
+                        Effect.map(({ snapshotSequence }) => snapshotSequence),
+                        // If we cannot read our own cursor, prefer the previous
+                        // behaviour (replay) over forcing a full snapshot.
+                        Effect.orElseSucceed(() => afterSequence),
+                      );
+                    const gap = projectedSequence - afterSequence;
+                    if (gap < 0 || gap > SHELL_CATCH_UP_REPLAY_LIMIT) {
+                      yield* Effect.logInfo(
+                        "orchestration shell resume fell back to a full snapshot",
+                        { afterSequence, projectedSequence, gap },
+                      );
+                      const snapshot = yield* loadShellSnapshot;
+                      return Stream.concat(
+                        Stream.make({ kind: "snapshot" as const, snapshot }),
+                        Stream.fromQueue(liveBuffer),
+                      );
+                    }
+
                     const catchUpStream = orchestrationEngine
                       .readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
                       .pipe(
@@ -1121,18 +1180,7 @@ const makeWsRpcLayer = (
                 );
               }
 
-              const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe(
-                Effect.tapError((cause) =>
-                  Effect.logError("orchestration shell snapshot load failed", { cause }),
-                ),
-                Effect.mapError(
-                  (cause) =>
-                    new OrchestrationGetSnapshotError({
-                      message: "Failed to load orchestration shell snapshot",
-                      cause,
-                    }),
-                ),
-              );
+              const snapshot = yield* loadShellSnapshot;
 
               return Stream.concat(
                 Stream.make({

packages/client-runtime/src/rpc/client.ts

@@ -149,7 +149,10 @@ export function runStream<TTag extends EnvironmentStreamCommandRpcTag>(
 
 export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
   tag: TTag,
-  input: EnvironmentRpcInput<TTag>,
+  // Pass a thunk when the input carries a resume cursor: it is re-read on every
+  // (re)subscribe, so a subscription that outlives several sessions resumes
+  // from where it actually got to rather than from where the page booted.
+  input: EnvironmentRpcInput<TTag> | (() => EnvironmentRpcInput<TTag>),
   options?: {
     readonly onExpectedFailure?: (
       cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>,
@@ -161,6 +164,8 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
   EnvironmentRpcStreamFailure<TTag>,
   EnvironmentSupervisor
 > {
+  const resolveInput = (): EnvironmentRpcInput<TTag> =>
+    typeof input === "function" ? (input as () => EnvironmentRpcInput<TTag>)() : input;
   return Stream.unwrap(
     EnvironmentSupervisor.pipe(
       Effect.map((supervisor) =>
@@ -180,7 +185,7 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
                   EnvironmentRpcStreamFailure<TTag>
                 > =>
                   Stream.suspend(() =>
-                    method(input).pipe(
+                    method(resolveInput()).pipe(
                       Stream.catchCause((cause) => {
                         const hasOnlyExpectedFailures =
                           cause.reasons.length > 0 &&

packages/client-runtime/src/state/shell-sync.test.ts

@@ -1,15 +1,18 @@
 import {
   EnvironmentId,
   ORCHESTRATION_WS_METHODS,
+  ThreadId,
   type OrchestrationShellSnapshot,
   type OrchestrationShellStreamItem,
 } from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as Option from "effect/Option";
 import * as Queue from "effect/Queue";
+import * as Ref from "effect/Ref";
 import * as Stream from "effect/Stream";
 import * as SubscriptionRef from "effect/SubscriptionRef";
+import * as TestClock from "effect/testing/TestClock";
 
 import {
   AVAILABLE_CONNECTION_STATE,
@@ -193,4 +196,88 @@ describe("environment shell synchronization", () => {
       expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0);
     }),
   );
+
+  it.effect("retries a failed shell stream and resumes from the latest applied sequence", () =>
+    Effect.gen(function* () {
+      const cachedSnapshot: OrchestrationShellSnapshot = {
+        snapshotSequence: 5,
+        projects: [],
+        threads: [],
+        updatedAt: "2026-06-06T00:00:00.000Z",
+      };
+      // Records the resume cursor the client asks for on each subscribe attempt.
+      const attempts = yield* Ref.make<ReadonlyArray<number | undefined>>([]);
+      const client = {
+        [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) =>
+          Stream.unwrap(
+            Ref.updateAndGet(attempts, (seen) => [...seen, input.afterSequence]).pipe(
+              Effect.map((seen) =>
+                // The first attempt advances the shell to sequence 9 and then
+                // fails with a domain (non-transport) error; without a retry the
+                // shell would stay on that snapshot for the life of the page.
+                seen.length === 1
+                  ? Stream.concat(
+                      Stream.make({
+                        kind: "thread-removed" as const,
+                        sequence: 9,
+                        threadId: ThreadId.make("thread-1"),
+                      }),
+                      Stream.fail(new Error("shell stream failed")),
+                    )
+                  : Stream.never,
+              ),
+            ),
+          ),
+      } as unknown as WsRpcProtocolClient;
+      const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
+      const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
+        Option.some(session(client)),
+      );
+      const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
+        target: TARGET,
+        state: supervisorState,
+        session: activeSession,
+        prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
+        connect: Effect.void,
+        disconnect: Effect.void,
+        retryNow: Effect.void,
+      } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
+      const cache = Persistence.EnvironmentCacheStore.of({
+        loadShell: () => Effect.succeed(Option.some(cachedSnapshot)),
+        saveShell: () => Effect.void,
+        loadThread: () => Effect.succeed(Option.none()),
+        saveThread: () => Effect.void,
+        removeThread: () => Effect.void,
+        clear: () => Effect.void,
+      });
+      const snapshotLoader = ShellSnapshotLoader.of({
+        load: () => Effect.succeed(Option.none()),
+      });
+      yield* makeEnvironmentShellState().pipe(
+        Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
+        Effect.provideService(Persistence.EnvironmentCacheStore, cache),
+        Effect.provideService(ShellSnapshotLoader, snapshotLoader),
+      );
+
+      for (let attempt = 0; attempt < 100; attempt += 1) {
+        if ((yield* Ref.get(attempts)).length >= 1) {
+          break;
+        }
+        yield* Effect.yieldNow;
+      }
+      expect(yield* Ref.get(attempts)).toEqual([5]);
+
+      yield* TestClock.adjust("250 millis");
+      for (let attempt = 0; attempt < 100; attempt += 1) {
+        if ((yield* Ref.get(attempts)).length >= 2) {
+          break;
+        }
+        yield* Effect.yieldNow;
+      }
+
+      // Resubscribed (rather than ending the stream), and asked to resume from
+      // the sequence it actually reached rather than the boot-time cursor.
+      expect(yield* Ref.get(attempts)).toEqual([5, 9]);
+    }),
+  );
 });

packages/client-runtime/src/state/shell.ts

@@ -123,6 +123,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
       ),
     );
 
+  // The sequence the subscription should resume from. Kept as plain closure
+  // state (rather than read back out of `state`) because the subscribe input
+  // thunk below is synchronous. It always tracks the applied snapshot.
+  let resumeSequence: number | null = null;
+
   const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* (
     item: OrchestrationShellStreamItem,
   ) {
@@ -140,6 +145,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
     if (nextSnapshot === null) {
       return;
     }
+    resumeSequence = nextSnapshot.snapshotSequence;
 
     yield* SubscriptionRef.set(state, {
       snapshot: Option.some(nextSnapshot),
@@ -178,13 +184,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
         yield* applyItem({ kind: "snapshot", snapshot: base.value });
       }
 
-      const subscribeInput = Option.match(base, {
-        onNone: () => ({}),
-        onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }),
-      });
+      // Resolved per (re)subscribe so reconnects resume from the latest applied
+      // sequence instead of replaying the boot-time catch-up window again.
+      const subscribeInput = () =>
+        resumeSequence === null ? {} : { afterSequence: resumeSequence };
 
+      // Without a retry an expected (non-transport) failure ends the stream for
+      // the life of the page: the shell keeps rendering its cached snapshot and
+      // silently never updates again, even across reloads, because the cache is
+      // what the next load resumes from. Mirrors the thread subscription.
       yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, {
         onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),
+        retryExpectedFailureAfter: "250 millis",
       }).pipe(Stream.runForEach(applyItem));
     }),
   );