From 68abe7c37f2fa13d0a8112dcfd689b7ebad007c8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 25 Oct 2026 05:59:05 +0000 Subject: [PATCH 1/6] test(client-runtime): reproduce issue 24659 A message.dispatch whose socket closes before acknowledgment is dropped instead of retried with the same commandId after reconnect. Co-Authored-By: Claude Opus 5.5 --- .../src/operations/commands.test.ts | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index bcfae19403483..a762b3e6fc6c8 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -21,10 +21,14 @@ import { } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Crypto from "effect/Crypto"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import { RpcClientError } from "effect/rpc"; import { AVAILABLE_CONNECTION_STATE, @@ -299,6 +303,59 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestCrypto)), ); + it.effect("retries a message dispatch lost to a socket close with the same command id", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ commands, projects: [] }); + const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session)); + const firstAttemptLost = yield* Deferred.make(); + const lostClient = { + ...healthySession.client, + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => + Effect.gen(function* () { + commands.push(command); + yield* SubscriptionRef.set(supervisor.session, Option.none()); + yield* Deferred.succeed(firstAttemptLost, undefined); + return yield* new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }); + }), + } as unknown as WsRpcProtocolClient; + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: lostClient }), + ); + + const sendFiber = yield* startThreadTurn({ + commandId: CommandId.make("command-lost-ack"), + threadId: v2ThreadId, + message: { + messageId: MessageId.make("message-lost-ack"), + role: "user", + text: "hello", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + yield* Deferred.await(firstAttemptLost); + yield* SubscriptionRef.set(supervisor.session, Option.some(healthySession)); + const exit = yield* Fiber.await(sendFiber); + + expect(commands.map((command) => [command.type, command.commandId])).toEqual([ + ["message.dispatch", "command-lost-ack"], + ["message.dispatch", "command-lost-ack"], + ]); + expect(Exit.isSuccess(exit)).toBe(true); + }).pipe(Effect.provide(layerTestCrypto)), + ); + it.effect("retains projection-shaped delivery for servers without command context support", () => Effect.gen(function* () { const activeRunId = RunId.make("legacy-active-run"); From 27039417731e85cc54026b1b960df4e9a08e0ad4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 25 Oct 2026 06:02:33 +0000 Subject: [PATCH 2/6] fix(client-runtime): a message sent as the socket drops is resent after reconnect A message.dispatch whose socket closed before the server answered failed outright, so web removed the optimistic message and restored the composer even when the server had already saved it. Transport failures now wait for the next session and resend with the same commandId, which the server's command receipt dedups. Resends are bounded so a frame that itself kills the socket cannot loop across reconnects. Co-Authored-By: Claude Opus 5.5 --- .../src/operations/commands.test.ts | 52 +++++++++++++++++++ .../client-runtime/src/operations/commands.ts | 39 ++++++++++++-- 2 files changed, 88 insertions(+), 3 deletions(-) diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index a762b3e6fc6c8..443c54097019a 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -356,6 +356,58 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestCrypto)), ); + it.effect("stops resending a message whose dispatch keeps losing the socket", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ commands, projects: [] }); + const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session)); + const dyingClient: WsRpcProtocolClient = { + ...healthySession.client, + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => + Effect.gen(function* () { + commands.push(command); + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: dyingClient }), + ); + return yield* new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }); + }), + } as unknown as WsRpcProtocolClient; + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: dyingClient }), + ); + + const error = yield* startThreadTurn({ + commandId: CommandId.make("command-socket-killer"), + threadId: v2ThreadId, + message: { + messageId: MessageId.make("message-socket-killer"), + role: "user", + text: "hello", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.flip, + ); + + expect(commands.map((command) => command.commandId)).toEqual([ + "command-socket-killer", + "command-socket-killer", + "command-socket-killer", + ]); + expect(error._tag).toBe("RpcClientError"); + }).pipe(Effect.provide(layerTestCrypto)), + ); + it.effect("retains projection-shaped delivery for servers without command context support", () => Effect.gen(function* () { const activeRunId = RunId.make("legacy-active-run"); diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 1f96499470712..c3944dcca1262 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -33,8 +33,13 @@ import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2Pend import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Result from "effect/Result"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; -import { getInitialServerConfig, request } from "../rpc/client.ts"; +import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import { getInitialServerConfig, isRpcClientError, request } from "../rpc/client.ts"; interface CommandMetadata { readonly commandId?: CommandId; @@ -272,6 +277,34 @@ const allocateCommandId = Effect.fn("EnvironmentCommands.allocateCommandId")(fun const dispatch = (command: OrchestrationV2Command) => request(ORCHESTRATION_V2_WS_METHODS.dispatchCommand, command); +// A socket can close after the server accepted the frame but before it answered. +// The server replays an accepted commandId from its receipt, so resending on the +// next session cannot duplicate the message. The bound stops a frame that itself +// kills the socket from looping across reconnects. +const MAX_LOST_TRANSPORT_RESENDS = 2; + +const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(function* ( + command: Extract, +) { + const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; + for (let resends = 0; ; resends += 1) { + const attemptSession = yield* SubscriptionRef.get(supervisor.session); + const result = yield* dispatch(command).pipe(Effect.result); + if (Result.isSuccess(result)) return result.success; + if (!isRpcClientError(result.failure) || resends >= MAX_LOST_TRANSPORT_RESENDS) { + return yield* result.failure; + } + yield* SubscriptionRef.changes(supervisor.session).pipe( + Stream.filter( + (session) => + Option.isSome(session) && + !(Option.isSome(attemptSession) && session.value === attemptSession.value), + ), + Stream.runHead, + ); + } +}); + const getProjection = (threadId: ThreadId) => request(ORCHESTRATION_V2_WS_METHODS.getThreadProjection, { threadId }); @@ -734,7 +767,7 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( const requestedMode = input.dispatchMode ?? "auto"; if (requestedMode === "start") { - return yield* dispatch({ + return yield* dispatchMessage({ type: "message.dispatch", commandId, createdBy: "user", @@ -801,7 +834,7 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( const shouldSendTitleSeed = input.titleSeed !== undefined && (serverResolvesCommandContext || projection?.messages.length === 0); - return yield* dispatch({ + return yield* dispatchMessage({ type: "message.dispatch", commandId, createdBy: "user", From ca08e1a09859a6d9309395a9e0ec29000f836829 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 01:22:40 +0000 Subject: [PATCH 3/6] fix(client-runtime): a lost message is given back when its environment stays down Bound the wait for a replacement session so a long outage fails the send with the original error, restoring the draft as on main. The mobile outbox already retries on its own and holds one slot across environments, so its sends opt out of the resend. AI-assisted; verified with the tests and tools listed above. --- apps/mobile/src/state/threadOutboxDelivery.ts | 2 + .../src/operations/commands.test.ts | 58 +++++++++++++ .../client-runtime/src/operations/commands.ts | 86 +++++++++++-------- 3 files changed, 111 insertions(+), 35 deletions(-) diff --git a/apps/mobile/src/state/threadOutboxDelivery.ts b/apps/mobile/src/state/threadOutboxDelivery.ts index 9b2da676eee44..23bae8d90ac42 100644 --- a/apps/mobile/src/state/threadOutboxDelivery.ts +++ b/apps/mobile/src/state/threadOutboxDelivery.ts @@ -895,6 +895,7 @@ export function createThreadOutboxDelivery(): { interactionMode: sendSettings.interactionMode, createdAt: queuedMessage.createdAt, dispatchMode: "queue", + resendAfterSocketLoss: false, }, }); if (AsyncResult.isFailure(compactionResult) && !hasAccess()) return true; @@ -935,6 +936,7 @@ export function createThreadOutboxDelivery(): { createdAt: queuedMessage.createdAt, dispatchMode: queuedMessage.dispatchMode ?? "start", ...(queuedMessage.freshContext ? { freshContext: true } : {}), + resendAfterSocketLoss: false, }, }); if (AsyncResult.isFailure(deliveryResult) && !hasAccess()) return true; diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 288e47f819199..383110d5369a5 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -28,6 +28,7 @@ import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; import { RpcClientError } from "effect/rpc"; import { @@ -402,6 +403,63 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); + it.effect("gives a lost message back when no replacement session arrives", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ commands, projects: [] }); + const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session)); + const lostClient = { + ...healthySession.client, + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => + Effect.gen(function* () { + commands.push(command); + yield* SubscriptionRef.set(supervisor.session, Option.none()); + return yield* new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }); + }), + } as unknown as WsRpcProtocolClient; + const send = (resendAfterSocketLoss: boolean) => + Effect.gen(function* () { + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: lostClient }), + ); + return yield* startThreadTurn({ + commandId: CommandId.make(`command-host-gone-${resendAfterSocketLoss}`), + threadId: v2ThreadId, + message: { + messageId: MessageId.make(`message-host-gone-${resendAfterSocketLoss}`), + role: "user", + text: "hello", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + resendAfterSocketLoss, + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.flip, + Effect.forkChild, + ); + }); + + const optedOut = yield* send(false); + expect((yield* Fiber.join(optedOut))._tag).toBe("RpcClientError"); + + const waiting = yield* send(true); + yield* TestClock.adjust("15 seconds"); + expect((yield* Fiber.join(waiting))._tag).toBe("RpcClientError"); + expect(commands.map((command) => command.commandId)).toEqual([ + "command-host-gone-false", + "command-host-gone-true", + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); + it.effect("Stop with no run left ends the thread's pull request watches", () => Effect.gen(function* () { const link = { diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 8a9a711f5f26a..5823fbb765450 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -190,6 +190,8 @@ export interface StartThreadTurnInput extends ThreadCommandInput { readonly dispatchMode?: "auto" | "queue" | "steer" | "restart" | "start"; /** Start a new provider session; needs the server's `freshContext` capability. */ readonly freshContext?: boolean; + /** Defaults to true. Callers with their own retry queue opt out so a dead socket fails fast. */ + readonly resendAfterSocketLoss?: boolean; } export interface InterruptThreadTurnInput extends ThreadCommandInput { @@ -286,13 +288,16 @@ const dispatch = (command: OrchestrationV2Command) => // A socket can close after the server accepted the frame but before it answered. // The server replays an accepted commandId from its receipt, so resending on the -// next session cannot duplicate the message. The bound stops a frame that itself -// kills the socket from looping across reconnects. +// next session cannot duplicate the message. The resend cap stops a frame that +// kills the socket from looping; the wait cap hands long outages back to the caller. const MAX_LOST_TRANSPORT_RESENDS = 2; +const REPLACEMENT_SESSION_WAIT = "15 seconds"; const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(function* ( command: Extract, + resendAfterSocketLoss: boolean, ) { + if (!resendAfterSocketLoss) return yield* dispatch(command); const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; for (let resends = 0; ; resends += 1) { const attemptSession = yield* SubscriptionRef.get(supervisor.session); @@ -301,14 +306,16 @@ const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(functio if (!isRpcClientError(result.failure) || resends >= MAX_LOST_TRANSPORT_RESENDS) { return yield* result.failure; } - yield* SubscriptionRef.changes(supervisor.session).pipe( + const replacement = yield* SubscriptionRef.changes(supervisor.session).pipe( Stream.filter( (session) => Option.isSome(session) && !(Option.isSome(attemptSession) && session.value === attemptSession.value), ), Stream.runHead, + Effect.timeoutOption(REPLACEMENT_SESSION_WAIT), ); + if (Option.isNone(replacement)) return yield* result.failure; } }); @@ -766,56 +773,65 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( } const requestedMode = input.dispatchMode ?? "auto"; + const resendAfterSocketLoss = input.resendAfterSocketLoss ?? true; if (requestedMode === "start") { - return yield* dispatchMessage({ + return yield* dispatchMessage( + { + type: "message.dispatch", + commandId, + createdBy: "user", + creationSource: input.creationSource ?? "web", + threadId: input.threadId, + messageId: input.message.messageId, + ...(input.manualContinuationOfRunId === undefined + ? {} + : { manualContinuationOfRunId: input.manualContinuationOfRunId }), + text: input.message.text, + ...(context ? { context } : {}), + ...(input.message.idleCompaction === undefined + ? {} + : { idleCompaction: input.message.idleCompaction }), + attachments, + ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), + ...(input.sourceProposedPlan === undefined + ? {} + : { sourcePlanRef: input.sourceProposedPlan }), + ...(input.freshContext ? { freshContext: true } : {}), + dispatchMode: { type: "start_immediately" }, + }, + resendAfterSocketLoss, + ); + } + + const dispatchMode = + requestedMode === "queue" + ? ({ type: "queue_after_active" } as const) + : ({ type: "start_immediately" } as const); + return yield* dispatchMessage( + { type: "message.dispatch", commandId, createdBy: "user", creationSource: input.creationSource ?? "web", threadId: input.threadId, messageId: input.message.messageId, - ...(input.manualContinuationOfRunId === undefined - ? {} - : { manualContinuationOfRunId: input.manualContinuationOfRunId }), text: input.message.text, ...(context ? { context } : {}), ...(input.message.idleCompaction === undefined ? {} : { idleCompaction: input.message.idleCompaction }), attachments, + ...(input.titleSeed === undefined ? {} : { titleSeed: input.titleSeed }), ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), ...(input.sourceProposedPlan === undefined ? {} : { sourcePlanRef: input.sourceProposedPlan }), ...(input.freshContext ? { freshContext: true } : {}), - dispatchMode: { type: "start_immediately" }, - }); - } - - const dispatchMode = - requestedMode === "queue" - ? ({ type: "queue_after_active" } as const) - : ({ type: "start_immediately" } as const); - return yield* dispatchMessage({ - type: "message.dispatch", - commandId, - createdBy: "user", - creationSource: input.creationSource ?? "web", - threadId: input.threadId, - messageId: input.message.messageId, - text: input.message.text, - ...(context ? { context } : {}), - ...(input.message.idleCompaction === undefined - ? {} - : { idleCompaction: input.message.idleCompaction }), - attachments, - ...(input.titleSeed === undefined ? {} : { titleSeed: input.titleSeed }), - ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), - ...(input.sourceProposedPlan === undefined ? {} : { sourcePlanRef: input.sourceProposedPlan }), - ...(input.freshContext ? { freshContext: true } : {}), - ...(requestedMode !== "queue" ? { deliveryIntent: requestedMode } : {}), - dispatchMode, - }); + ...(requestedMode !== "queue" ? { deliveryIntent: requestedMode } : {}), + dispatchMode, + }, + resendAfterSocketLoss, + ); }); export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThreadTurn")(function* ( From 59b674a6fc4d3b950c11a94f45c694bf62763637 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 01:37:50 +0000 Subject: [PATCH 4/6] fix(client-runtime): a lost message waits out reconnect backoff instead of a timer A 15 second deadline gave the message back while the supervisor was still reconnecting, so resubmitting the restored draft could duplicate a message the server already committed. Keep the command pending until a replacement session arrives, and give it back only when the supervisor stops trying: the user disconnected or the connection is blocked. AI-assisted; verified with the tests and tools listed above. --- .../src/operations/commands.test.ts | 15 ++++++++-- .../client-runtime/src/operations/commands.ts | 29 +++++++++++-------- 2 files changed, 29 insertions(+), 15 deletions(-) diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 383110d5369a5..21a475fc52b8b 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -60,6 +60,12 @@ const TARGET = new PrimaryConnectionTarget({ wsBaseUrl: "wss://environment.example.test", }); +const CONNECTED_STATE = { + ...AVAILABLE_CONNECTION_STATE, + desired: true, + phase: "connected", +} as const; + const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(function* (input: { readonly commands: OrchestrationV2Command[]; readonly projects: ProjectMutation[]; @@ -126,7 +132,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct }; return EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, - state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + state: yield* SubscriptionRef.make(CONNECTED_STATE), session: yield* SubscriptionRef.make(Option.some(session)), prepared: yield* SubscriptionRef.make(Option.none()), connect: Effect.void, @@ -403,7 +409,7 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); - it.effect("gives a lost message back when no replacement session arrives", () => + it.effect("holds a lost message through reconnect backoff until disconnect", () => Effect.gen(function* () { const commands: OrchestrationV2Command[] = []; const supervisor = yield* makeSupervisor({ commands, projects: [] }); @@ -451,7 +457,10 @@ describe("V2 environment commands", () => { expect((yield* Fiber.join(optedOut))._tag).toBe("RpcClientError"); const waiting = yield* send(true); - yield* TestClock.adjust("15 seconds"); + yield* SubscriptionRef.set(supervisor.state, { ...CONNECTED_STATE, phase: "backoff" }); + yield* TestClock.adjust("2 minutes"); + expect(waiting.pollUnsafe()).toBeUndefined(); + yield* SubscriptionRef.set(supervisor.state, AVAILABLE_CONNECTION_STATE); expect((yield* Fiber.join(waiting))._tag).toBe("RpcClientError"); expect(commands.map((command) => command.commandId)).toEqual([ "command-host-gone-false", diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 5823fbb765450..ee268fa23b4b7 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -286,12 +286,11 @@ const allocateCommandId = Effect.fn("EnvironmentCommands.allocateCommandId")(fun const dispatch = (command: OrchestrationV2Command) => request(ORCHESTRATION_V2_WS_METHODS.dispatchCommand, command); -// A socket can close after the server accepted the frame but before it answered. -// The server replays an accepted commandId from its receipt, so resending on the -// next session cannot duplicate the message. The resend cap stops a frame that -// kills the socket from looping; the wait cap hands long outages back to the caller. +// The socket can close after the server accepted the frame but before it answered. +// The server replays an accepted commandId from its receipt, so the same command +// waits out reconnect backoff until the supervisor stops trying. The resend cap +// stops a frame that kills the socket from looping. const MAX_LOST_TRANSPORT_RESENDS = 2; -const REPLACEMENT_SESSION_WAIT = "15 seconds"; const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(function* ( command: Extract, @@ -306,14 +305,20 @@ const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(functio if (!isRpcClientError(result.failure) || resends >= MAX_LOST_TRANSPORT_RESENDS) { return yield* result.failure; } - const replacement = yield* SubscriptionRef.changes(supervisor.session).pipe( - Stream.filter( - (session) => - Option.isSome(session) && - !(Option.isSome(attemptSession) && session.value === attemptSession.value), + const replacement = yield* Effect.raceFirst( + SubscriptionRef.changes(supervisor.session).pipe( + Stream.filter( + (session) => + Option.isSome(session) && + !(Option.isSome(attemptSession) && session.value === attemptSession.value), + ), + Stream.runHead, + ), + SubscriptionRef.changes(supervisor.state).pipe( + Stream.filter((state) => state.phase === "available" || state.phase === "blocked"), + Stream.runHead, + Effect.as(Option.none()), ), - Stream.runHead, - Effect.timeoutOption(REPLACEMENT_SESSION_WAIT), ); if (Option.isNone(replacement)) return yield* result.failure; } From b45d877d183ad9fbbd074f3e698c5417f500a4cb Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 01:39:02 +0000 Subject: [PATCH 5/6] test(client-runtime): type the connected supervisor state fixture AI-assisted; verified with the tests and tools listed above. --- .../client-runtime/src/operations/commands.test.ts | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 21a475fc52b8b..5818d9e9a04b6 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -35,6 +35,7 @@ import { AVAILABLE_CONNECTION_STATE, PrimaryConnectionTarget, type PreparedConnection, + type SupervisorConnectionState, } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as RpcSession from "../rpc/session.ts"; @@ -60,11 +61,11 @@ const TARGET = new PrimaryConnectionTarget({ wsBaseUrl: "wss://environment.example.test", }); -const CONNECTED_STATE = { +const CONNECTED_STATE: SupervisorConnectionState = { ...AVAILABLE_CONNECTION_STATE, desired: true, phase: "connected", -} as const; +}; const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(function* (input: { readonly commands: OrchestrationV2Command[]; @@ -457,7 +458,10 @@ describe("V2 environment commands", () => { expect((yield* Fiber.join(optedOut))._tag).toBe("RpcClientError"); const waiting = yield* send(true); - yield* SubscriptionRef.set(supervisor.state, { ...CONNECTED_STATE, phase: "backoff" }); + yield* SubscriptionRef.set(supervisor.state, { + ...CONNECTED_STATE, + phase: "backoff" as const, + }); yield* TestClock.adjust("2 minutes"); expect(waiting.pollUnsafe()).toBeUndefined(); yield* SubscriptionRef.set(supervisor.state, AVAILABLE_CONNECTION_STATE); From 8ce300064f41c9704e5afcf21d8c1a1a6f1eb667 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 09:46:13 +0000 Subject: [PATCH 6/6] fix(client-runtime): a message lost when its session closes is resent too The supervisor can close a dead session before the RPC protocol reports the socket error, and closing the session interrupts the in-flight dispatch instead of failing it. Treat an interrupt-only request exit as transport loss so the same command is resent on the replacement session. AI-assisted; verified with the tests and tools listed above. --- .../src/operations/commands.test.ts | 23 +++++++++++++------ .../client-runtime/src/operations/commands.ts | 17 ++++++++------ 2 files changed, 26 insertions(+), 14 deletions(-) diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 5818d9e9a04b6..5d1d59ac9cd4e 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -20,6 +20,7 @@ import { type ProjectMutation, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -305,7 +306,20 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); - it.effect("retries a message dispatch lost to a socket close with the same command id", () => + // A dead socket reaches the caller as a protocol error, or as an interrupt when + // the supervisor closes the session before the protocol reports the error. + it.effect.each([ + [ + "protocol error", + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }), + ], + ["session close", Effect.failCause(Cause.interrupt())], + ] as const)("retries a message dispatch lost to a %s with the same command id", ([, loss]) => Effect.gen(function* () { const commands: OrchestrationV2Command[] = []; const supervisor = yield* makeSupervisor({ commands, projects: [] }); @@ -318,12 +332,7 @@ describe("V2 environment commands", () => { commands.push(command); yield* SubscriptionRef.set(supervisor.session, Option.none()); yield* Deferred.succeed(firstAttemptLost, undefined); - return yield* new RpcClientError.RpcClientError({ - reason: new RpcClientError.RpcClientDefect({ - message: "socket closed", - cause: new Error("socket closed"), - }), - }); + return yield* loss; }), } as unknown as WsRpcProtocolClient; yield* SubscriptionRef.set( diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index ee268fa23b4b7..5a08879cf7cc3 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -31,9 +31,10 @@ import { import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Option from "effect/Option"; -import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -300,11 +301,13 @@ const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(functio const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; for (let resends = 0; ; resends += 1) { const attemptSession = yield* SubscriptionRef.get(supervisor.session); - const result = yield* dispatch(command).pipe(Effect.result); - if (Result.isSuccess(result)) return result.success; - if (!isRpcClientError(result.failure) || resends >= MAX_LOST_TRANSPORT_RESENDS) { - return yield* result.failure; - } + const exit = yield* Effect.exit(dispatch(command)); + if (Exit.isSuccess(exit)) return exit.value; + // Closing the session interrupts the request, which can beat the protocol error. + const lostTransport = + Cause.hasInterruptsOnly(exit.cause) || + exit.cause.reasons.some((reason) => reason._tag === "Fail" && isRpcClientError(reason.error)); + if (!lostTransport || resends >= MAX_LOST_TRANSPORT_RESENDS) return yield* exit; const replacement = yield* Effect.raceFirst( SubscriptionRef.changes(supervisor.session).pipe( Stream.filter( @@ -320,7 +323,7 @@ const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(functio Effect.as(Option.none()), ), ); - if (Option.isNone(replacement)) return yield* result.failure; + if (Option.isNone(replacement)) return yield* exit; } });