From b52b1e984a31def0c5989eed8348fa74ec031202 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Mon, 9 Nov 2026 08:26:07 +0000 Subject: [PATCH 1/3] fix(client-runtime): recover thread list updates after stream defects Co-Authored-By: Claude Opus 5.5 --- .../src/state/shell-sync.test.ts | 53 +++++++++++++++++++ packages/client-runtime/src/state/shell.ts | 1 + 2 files changed, 54 insertions(+) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 96a5b59c5d743..3cb1195117e51 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/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,58 @@ 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"); + + yield* Queue.failCause(first.events, Cause.die(new Error("shell invariant failed"))); + // Let the defect settle before the next session arrives, as it does on a real device. + yield* TestClock.adjust("1 second"); + expect((yield* SubscriptionRef.get(h.shellState)).error).toEqual( + Option.some("Could not synchronize environment data."), + ); + + 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()); + }), + ); + it.effect("reloads over HTTP when the server answers a resume with snapshot-required", () => Effect.gen(function* () { const h = yield* makeColdStart(RECENT_ORIGIN); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 86584c4a03faa..c923ba4496cd0 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -702,6 +702,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }; }), { + onDefect: (cause) => setStreamError(Cause.squash(cause)), onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), retryExpectedFailureAfter: "250 millis", resubscribe: resubscriptions, From 0446e4fa468ed0269c1b6ad8b3d72532cc1d2ba9 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:20:37 +0000 Subject: [PATCH 2/3] test(client-runtime): await the stream error instead of advancing the clock Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/state/shell-sync.test.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 3cb1195117e51..3e80c0d37a3d3 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1616,10 +1616,14 @@ describe("environment shell cold start", () => { yield* liveRow(CACHED_SEQUENCE + 1, "completed"); yield* Queue.failCause(first.events, Cause.die(new Error("shell invariant failed"))); - // Let the defect settle before the next session arrives, as it does on a real device. - yield* TestClock.adjust("1 second"); - expect((yield* SubscriptionRef.get(h.shellState)).error).toEqual( - Option.some("Could not synchronize environment data."), + yield* SubscriptionRef.changes(h.shellState).pipe( + Stream.filter((state) => + Option.exists( + state.error, + (error) => error === "Could not synchronize environment data.", + ), + ), + Stream.runHead, ); yield* h.reconnect; From bc398e6309a9d2d695033490628e6a936d68c6cf Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 06:00:08 +0000 Subject: [PATCH 3/3] fix(client-runtime): retry the shell stream after a defect Handling defects turns off the automatic same-session retry for chunk decode defects, so back off and resubscribe instead of waiting for a reconnect or wakeup. Co-Authored-By: Claude Opus 5.5 --- .../src/state/shell-sync.test.ts | 29 +++++++++++++------ packages/client-runtime/src/state/shell.ts | 22 ++++++++++++-- 2 files changed, 39 insertions(+), 12 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 3e80c0d37a3d3..04c165ee72fed 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1615,17 +1615,22 @@ describe("environment shell cold start", () => { ]); yield* liveRow(CACHED_SEQUENCE + 1, "completed"); - yield* Queue.failCause(first.events, Cause.die(new Error("shell invariant failed"))); - yield* SubscriptionRef.changes(h.shellState).pipe( - Stream.filter((state) => - Option.exists( - state.error, - (error) => error === "Could not synchronize environment data.", + 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, + ), ), - ), - Stream.runHead, - ); + ); + yield* failWithDefect(first.events); yield* h.reconnect; const resumed = yield* Queue.take(h.subscriptions); expect(resumed.input).toMatchObject({ afterSequence: CACHED_SEQUENCE + 1 }); @@ -1634,6 +1639,12 @@ describe("environment shell cold start", () => { { 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 }); }), ); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index c923ba4496cd0..cf5462d3f3e2f 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/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(); + let defectRetries = 0; const latestLive = yield* Ref.make>(Option.none()); const persistence = yield* Queue.sliding(1); @@ -702,12 +703,27 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }; }), { - onDefect: (cause) => setStreamError(Cause.squash(cause)), + // 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) => {