From ead722a5eafad18196dd9178a2058987b949f351 Mon Sep 17 00:00:00 2001 From: Wills Manley <216224554+willsmanley@users.noreply.github.com> Date: Sat, 24 Oct 2026 05:52:48 -0400 Subject: [PATCH] fix(client): resubscribe the thread list after a defect A defect in the shell subscription ended the only stream feeding the thread list. Nothing restarted it and its status stayed live, so the list stayed frozen on screen until the app restarted. The list now shows the error and resubscribes with a backoff capped at 30 seconds. Co-Authored-By: Claude Opus 5.5 --- .../src/state/shell-sync.test.ts | 64 +++++++++++++++++++ packages/client-runtime/src/state/shell.ts | 21 +++++- 2 files changed, 84 insertions(+), 1 deletion(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 5ce6d7dbaa290..c18a5a94299c2 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -17,6 +17,7 @@ import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; import { AVAILABLE_CONNECTION_STATE, @@ -416,6 +417,69 @@ describe("environment shell synchronization", () => { }), ); + it.effect("resubscribes after a defect ends the shell subscription", () => + Effect.gen(function* () { + const defect = new Error("shell item could not be applied"); + const events = yield* Queue.unbounded(); + let subscriptions = 0; + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => { + subscriptions += 1; + return subscriptions === 1 ? Stream.die(defect) : Stream.fromQueue(events); + }, + } as unknown as WsRpcProtocolClient; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.some(session(client))), + 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.none()), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.some(LIVE_SHELL_SNAPSHOT)), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), + ); + + const failed = yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => Option.isSome(state.error)), + Stream.runHead, + ); + expect(Option.isSome(failed)).toBe(true); + yield* TestClock.adjust("1 minute"); + yield* Queue.offer(events, { + kind: "snapshot", + snapshot: { ...LIVE_SHELL_SNAPSHOT, snapshotSequence: 5 }, + } as OrchestrationV2ShellStreamItem); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 5, + ), + Stream.runHead, + ); + expect(subscriptions).toBe(2); + }), + ); + it.effect("applies authoritative lower-sequence HTTP resets over client-ahead cache", () => Effect.gen(function* () { const cachedSnapshot: OrchestrationV2ShellSnapshot = { diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 0a7e2aa7c6664..423f5eb7b1072 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -9,11 +9,13 @@ import { type ThreadPullRequestLink, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -66,6 +68,11 @@ function withoutDeferredPullRequests(value: DeferredShellSnapshot): Orchestratio } const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data."; +const SHELL_RESUBSCRIBE_SCHEDULE = Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), +); export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () { const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; @@ -391,7 +398,19 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEachArray((items) => applyItems(items))), + ).pipe( + Stream.runForEachArray((items) => applyItems(items)), + // A defect ends the only subscription feeding the thread list, which + // would otherwise stay on screen, frozen, until the app restarts. + Effect.tapCause((cause) => + Cause.hasInterruptsOnly(cause) ? Effect.void : setStreamError(Cause.squash(cause)), + ), + Effect.sandbox, + Effect.retry({ + schedule: SHELL_RESUBSCRIBE_SCHEDULE, + while: (cause) => !Cause.hasInterruptsOnly(cause), + }), + ), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => {