fix(client-runtime): recover thread list updates after stream defects

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

@@ -8,6 +8,7 @@ import {
   type ThreadPullRequestLink,
 } from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
+import * as Cause from "effect/Cause";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
@@ -1580,6 +1581,73 @@ describe("environment shell cold start", () => {
     }),
   );
 
+  it.effect("keeps syncing row status after a shell stream defect", () =>
+    Effect.gen(function* () {
+      const h = yield* makeColdStart(RECENT_ORIGIN);
+      const rowUpdate = (sequence: number, status: "completed" | "running") => ({
+        kind: "thread.updated" as const,
+        sequence,
+        location: "active" as const,
+        thread: { ...v2ThreadShell, status },
+      });
+      const liveRow = (sequence: number, status: "completed" | "running") =>
+        SubscriptionRef.changes(h.shellState).pipe(
+          Stream.filter(
+            (state) =>
+              state.status === "live" &&
+              Option.exists(
+                state.snapshot,
+                (snapshot) =>
+                  snapshot.snapshotSequence === sequence &&
+                  snapshot.threads.some(
+                    (thread) => thread.id === v2ThreadShell.id && thread.status === status,
+                  ),
+              ),
+          ),
+          Stream.runHead,
+          Effect.map(Option.getOrThrow),
+        );
+
+      const first = yield* Queue.take(h.subscriptions);
+      yield* Queue.offerAll(first.events, [
+        rowUpdate(CACHED_SEQUENCE + 1, "completed"),
+        { kind: "synchronized" },
+      ]);
+      yield* liveRow(CACHED_SEQUENCE + 1, "completed");
+
+      const failWithDefect = (events: typeof first.events) =>
+        Queue.failCause(events, Cause.die(new Error("shell invariant failed"))).pipe(
+          Effect.andThen(
+            SubscriptionRef.changes(h.shellState).pipe(
+              Stream.filter((state) =>
+                Option.exists(
+                  state.error,
+                  (error) => error === "Could not synchronize environment data.",
+                ),
+              ),
+              Stream.runHead,
+            ),
+          ),
+        );
+
+      yield* failWithDefect(first.events);
+      yield* h.reconnect;
+      const resumed = yield* Queue.take(h.subscriptions);
+      expect(resumed.input).toMatchObject({ afterSequence: CACHED_SEQUENCE + 1 });
+      yield* Queue.offerAll(resumed.events, [
+        rowUpdate(CACHED_SEQUENCE + 2, "running"),
+        { kind: "synchronized" },
+      ]);
+      expect((yield* liveRow(CACHED_SEQUENCE + 2, "running")).error).toEqual(Option.none());
+
+      // Without a reconnect or wakeup, the stream resubscribes once the backoff passes.
+      yield* failWithDefect(resumed.events);
+      yield* TestClock.adjust("250 millis");
+      const retried = yield* Queue.take(h.subscriptions);
+      expect(retried.input).toMatchObject({ afterSequence: CACHED_SEQUENCE + 2 });
+    }),
+  );
+
   it.effect("reloads over HTTP when the server answers a resume with snapshot-required", () =>
     Effect.gen(function* () {
       const h = yield* makeColdStart(RECENT_ORIGIN);

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

@@ -30,7 +30,7 @@ import * as ConnectionWakeups from "../connection/wakeups.ts";
 import { safeErrorLogAttributes } from "../errors/safeLog.ts";
 import * as Persistence from "../platform/persistence.ts";
 import { runCachePersistence } from "./cachePersistence.ts";
-import { subscribeDynamicWithSession } from "../rpc/client.ts";
+import { subscribeDynamicWithSession, subscriptionRetryDelay } from "../rpc/client.ts";
 import type { RpcSession } from "../rpc/session.ts";
 import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
 import type { DeferredShellSnapshot } from "./shellPullRequests.ts";
@@ -203,6 +203,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
   // Only a server snapshot renews it; `snapshot-required` drops it.
   const origin = yield* Ref.make(cachedOrigin);
   const reloadRequests = yield* Queue.unbounded<void>();
+  let defectRetries = 0;
   const latestLive = yield* Ref.make<Option.Option<LiveShell>>(Option.none());
   const persistence = yield* Queue.sliding<OrchestrationV2ShellSnapshot>(1);
 
@@ -702,11 +703,27 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
         };
       }),
       {
+        // A defect parks the stream, so resubscribe after a backoff rather than wait for a wakeup.
+        onDefect: (cause) =>
+          setStreamError(Cause.squash(cause)).pipe(
+            Effect.andThen(Effect.sleep(subscriptionRetryDelay("250 millis", defectRetries++))),
+            Effect.andThen(Queue.offer(reloadRequests, undefined)),
+          ),
         onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),
         retryExpectedFailureAfter: "250 millis",
         resubscribe: resubscriptions,
       },
-    ).pipe(Stream.runForEachArray((entries) => applyItems(entries))),
+    ).pipe(
+      Stream.runForEachArray((entries) =>
+        applyItems(entries).pipe(
+          Effect.andThen(
+            Effect.sync(() => {
+              defectRetries = 0;
+            }),
+          ),
+        ),
+      ),
+    ),
   );
   yield* SubscriptionRef.changes(supervisor.state).pipe(
     Stream.runForEach((connectionState) => {