From d0246a8290f189695347612105ac41249cc2e850 Mon Sep 17 00:00:00 2001 From: "wizzoapp[bot]" <254688279+wizzoapp[bot]@users.noreply.github.com> Date: Mon, 6 Jul 2026 11:43:03 +0100 Subject: [PATCH 1/2] fix(stability): keep remote websocket sessions alive --- apps/server/src/ws.ts | 5 +- apps/server/src/wsKeepalive.test.ts | 30 +++++ apps/server/src/wsKeepalive.ts | 19 +++ .../web/src/components/ChatView.logic.test.ts | 96 +++++++++++++++ apps/web/src/components/ChatView.logic.ts | 52 ++++++++ apps/web/src/components/ChatView.tsx | 112 ++++++++++++++++-- .../src/components/chat/ThreadErrorBanner.tsx | 12 +- .../src/errors/transport.test.ts | 37 +++++- .../client-runtime/src/errors/transport.ts | 39 +++++- .../client-runtime/src/state/server.test.ts | 15 +++ packages/client-runtime/src/state/server.ts | 2 + packages/contracts/src/server.ts | 7 ++ 12 files changed, 409 insertions(+), 17 deletions(-) create mode 100644 apps/server/src/wsKeepalive.test.ts create mode 100644 apps/server/src/wsKeepalive.ts diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 346a271be2f..0bcaec274a8 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -114,6 +114,7 @@ import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; import * as SessionStore from "./auth/SessionStore.ts"; import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; +import { makeServerConfigHeartbeatStream } from "./wsKeepalive.ts"; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); const isOrchestrationScheduledTaskMutationError = Schema.is( OrchestrationScheduledTaskMutationError, @@ -1569,8 +1570,8 @@ const makeWsRpcLayer = ( .pipe(Effect.ignoreCause({ log: true }), Effect.forkScoped); const liveUpdates = Stream.merge( - keybindingsUpdates, - Stream.merge(providerStatuses, settingsUpdates), + makeServerConfigHeartbeatStream(), + Stream.merge(keybindingsUpdates, Stream.merge(providerStatuses, settingsUpdates)), ); return Stream.concat( diff --git a/apps/server/src/wsKeepalive.test.ts b/apps/server/src/wsKeepalive.test.ts new file mode 100644 index 00000000000..7fb39b53bde --- /dev/null +++ b/apps/server/src/wsKeepalive.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import { makeServerConfigHeartbeatStream, WS_KEEPALIVE_INTERVAL_MS } from "./wsKeepalive.ts"; + +describe("websocket keepalive", () => { + it.effect("emits application frames well before Cloudflare's idle websocket timeout", () => + Effect.gen(function* () { + expect(WS_KEEPALIVE_INTERVAL_MS).toBeLessThan(100_000); + + const eventsFiber = yield* makeServerConfigHeartbeatStream().pipe( + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + + yield* TestClock.adjust(Duration.millis(WS_KEEPALIVE_INTERVAL_MS)); + yield* TestClock.adjust(Duration.millis(WS_KEEPALIVE_INTERVAL_MS)); + + expect(Array.from(yield* Fiber.join(eventsFiber))).toEqual([ + { version: 1, type: "heartbeat" }, + { version: 1, type: "heartbeat" }, + ]); + }).pipe(Effect.provide(TestClock.layer())), + ); +}); diff --git a/apps/server/src/wsKeepalive.ts b/apps/server/src/wsKeepalive.ts new file mode 100644 index 00000000000..94a51415e8f --- /dev/null +++ b/apps/server/src/wsKeepalive.ts @@ -0,0 +1,19 @@ +import type { ServerConfigStreamEvent } from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; +import * as Stream from "effect/Stream"; + +export const WS_KEEPALIVE_INTERVAL_MS = 30_000; + +export function makeServerConfigHeartbeatEvent(): ServerConfigStreamEvent { + return { + version: 1, + type: "heartbeat", + }; +} + +export function makeServerConfigHeartbeatStream(): Stream.Stream { + return Stream.tick(Duration.millis(WS_KEEPALIVE_INTERVAL_MS)).pipe( + Stream.drop(1), + Stream.map(() => makeServerConfigHeartbeatEvent()), + ); +} diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index c507ea6f2d8..f4b918698fd 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -3,8 +3,10 @@ import { describe, expect, it } from "vite-plus/test"; import type { Thread } from "../types"; import { + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS, MAX_HIDDEN_MOUNTED_PREVIEW_THREADS, MAX_HIDDEN_MOUNTED_TERMINAL_THREADS, + buildThreadErrorDismissKey, buildExpiredTerminalContextToastCopy, buildThreadTurnInterruptInput, createLocalDispatchSnapshot, @@ -15,7 +17,9 @@ import { hasServerAcknowledgedLocalDispatch, reconcileMountedTerminalThreadIds, reconcileRetainedMountedThreadIds, + resolveVisibleServerThreadError, resolveSendEnvMode, + shouldShowEnvironmentUnavailableBanner, shouldWriteThreadErrorToCurrentServerThread, threadHasEstablishedProviderBinding, } from "./ChatView.logic"; @@ -209,6 +213,98 @@ describe("buildThreadTurnInterruptInput", () => { }); }); +describe("shouldShowEnvironmentUnavailableBanner", () => { + it("suppresses short reconnect blips", () => { + expect( + shouldShowEnvironmentUnavailableBanner({ + connectionPhase: "reconnecting", + unavailableSinceMs: 1_000, + nowMs: 1_000 + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS - 1, + }), + ).toBe(false); + }); + + it("shows the banner once the connection has stayed unavailable", () => { + expect( + shouldShowEnvironmentUnavailableBanner({ + connectionPhase: "reconnecting", + unavailableSinceMs: 1_000, + nowMs: 1_000 + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS, + }), + ).toBe(true); + }); + + it("clears immediately when the connection is back", () => { + expect( + shouldShowEnvironmentUnavailableBanner({ + connectionPhase: "connected", + unavailableSinceMs: 1_000, + nowMs: 1_000 + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS, + }), + ).toBe(false); + }); +}); + +describe("resolveVisibleServerThreadError", () => { + it("uses a dismissed key to hide a persisted server error for that turn", () => { + const turnId = TurnId.make("turn-dismissed"); + const dismissKey = buildThreadErrorDismissKey({ + threadKey: "environment-local:thread-1", + turnId, + error: "The turn was interrupted. Send your message again to retry.", + }); + + expect(dismissKey).not.toBeNull(); + expect( + resolveVisibleServerThreadError({ + localError: null, + sessionError: "The turn was interrupted. Send your message again to retry.", + dismissedSessionErrorKeys: { [dismissKey!]: true }, + threadKey: "environment-local:thread-1", + turnId, + }), + ).toBeNull(); + }); + + it("allows the same friendly interruption message to appear on a later turn", () => { + const dismissedTurnId = TurnId.make("turn-dismissed"); + const laterTurnId = TurnId.make("turn-later"); + const dismissKey = buildThreadErrorDismissKey({ + threadKey: "environment-local:thread-1", + turnId: dismissedTurnId, + error: "The turn was interrupted. Send your message again to retry.", + }); + + expect( + resolveVisibleServerThreadError({ + localError: null, + sessionError: "The turn was interrupted. Send your message again to retry.", + dismissedSessionErrorKeys: { [dismissKey!]: true }, + threadKey: "environment-local:thread-1", + turnId: laterTurnId, + }), + ).toBe("The turn was interrupted. Send your message again to retry."); + }); + + it("keeps local errors visible even when a persisted session error was dismissed", () => { + const dismissKey = buildThreadErrorDismissKey({ + threadKey: "environment-local:thread-1", + turnId: null, + error: "Persisted error", + }); + + expect( + resolveVisibleServerThreadError({ + localError: "Select a base branch before sending.", + sessionError: "Persisted error", + dismissedSessionErrorKeys: { [dismissKey!]: true }, + threadKey: "environment-local:thread-1", + turnId: null, + }), + ).toBe("Select a base branch before sending."); + }); +}); + describe("deriveComposerSendState", () => { it("treats expired terminal pills as non-sendable content", () => { const state = deriveComposerSendState({ diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index d56043d3c22..3577d998367 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -9,6 +9,7 @@ import { type ThreadId, type TurnId, } from "@t3tools/contracts"; +import type { EnvironmentConnectionPresentation } from "@t3tools/client-runtime/connection"; import { type ChatMessage, type SessionPhase, type Thread } from "../types"; import { type ComposerImageAttachment, type DraftThreadState } from "../composerDraftStore"; import * as Schema from "effect/Schema"; @@ -24,6 +25,7 @@ import type { DraftThreadEnvMode } from "../composerDraftStore"; export const LAST_INVOKED_SCRIPT_BY_PROJECT_KEY = "t3code:last-invoked-script-by-project"; export const MAX_HIDDEN_MOUNTED_TERMINAL_THREADS = 10; export const MAX_HIDDEN_MOUNTED_PREVIEW_THREADS = 3; +export const ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS = 2_000; export const LastInvokedScriptByProjectSchema = Schema.Record(ProjectId, Schema.String); @@ -74,6 +76,56 @@ export function shouldWriteThreadErrorToCurrentServerThread(input: { ); } +export function shouldShowEnvironmentUnavailableBanner(input: { + connectionPhase: EnvironmentConnectionPresentation["phase"]; + unavailableSinceMs: number | null; + nowMs: number; + debounceMs?: number; +}): boolean { + if (input.connectionPhase === "connected" || input.unavailableSinceMs === null) { + return false; + } + + const debounceMs = input.debounceMs ?? ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS; + return input.nowMs - input.unavailableSinceMs >= debounceMs; +} + +export function buildThreadErrorDismissKey(input: { + threadKey: string; + turnId: TurnId | null | undefined; + error: string | null | undefined; +}): string | null { + const error = input.error?.trim(); + if (!error) { + return null; + } + + return `${input.threadKey}:${input.turnId ?? "session"}:${error}`; +} + +export function resolveVisibleServerThreadError(input: { + localError: string | null; + sessionError: string | null; + dismissedSessionErrorKeys: Readonly>; + threadKey: string; + turnId: TurnId | null | undefined; +}): string | null { + if (input.localError !== null) { + return input.localError; + } + + const dismissKey = buildThreadErrorDismissKey({ + threadKey: input.threadKey, + turnId: input.turnId, + error: input.sessionError, + }); + if (dismissKey !== null && input.dismissedSessionErrorKeys[dismissKey]) { + return null; + } + + return input.sessionError; +} + export function buildThreadTurnInterruptInput(thread: Pick): { threadId: ThreadId; turnId?: TurnId; diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 754a5323ef4..181dffe1276 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -220,6 +220,8 @@ import { ThreadErrorBanner } from "./chat/ThreadErrorBanner"; import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack"; import { MAX_HIDDEN_MOUNTED_TERMINAL_THREADS, + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS, + buildThreadErrorDismissKey, buildExpiredTerminalContextToastCopy, buildLocalDraftThread, buildThreadTurnInterruptInput, @@ -237,7 +239,9 @@ import { hasQueuedSubmissionBeenObservedByShell, readFileAsDataUrl, reconcileMountedTerminalThreadIds, + resolveVisibleServerThreadError, resolveSendEnvMode, + shouldShowEnvironmentUnavailableBanner, threadHasEstablishedProviderBinding, revokeBlobPreviewUrl, revokeUserMessagePreviewUrls, @@ -1393,6 +1397,9 @@ function ChatViewContent(props: ChatViewProps) { const [localServerErrorsByThreadKey, setLocalServerErrorsByThreadKey] = useState< Record >({}); + const [dismissedServerThreadErrorKeys, setDismissedServerThreadErrorKeys] = useState< + Record + >({}); const [isConnecting, _setIsConnecting] = useState(false); const [isRevertingCheckpoint, setIsRevertingCheckpoint] = useState(false); const [maximizedRightPanelThreadKey, setMaximizedRightPanelThreadKey] = useState( @@ -1575,8 +1582,16 @@ function ChatViewContent(props: ChatViewProps) { ); const isServerThread = routeKind === "server" && serverThread !== null; const activeThread = isServerThread ? serverThread : localDraftThread; + const serverThreadSessionError = sanitizeThreadErrorMessage(serverThread?.session?.lastError); + const serverThreadSessionErrorTurnId = serverThread?.latestTurn?.turnId ?? null; const threadError = isServerThread - ? (localServerError ?? serverThread?.session?.lastError ?? null) + ? resolveVisibleServerThreadError({ + localError: localServerError, + sessionError: serverThreadSessionError, + dismissedSessionErrorKeys: dismissedServerThreadErrorKeys, + threadKey: routeThreadKey, + turnId: serverThreadSessionErrorTurnId, + }) : localDraftError; const runtimeMode = composerRuntimeMode ?? activeThread?.runtimeMode ?? DEFAULT_RUNTIME_MODE; const interactionMode = @@ -1793,6 +1808,71 @@ function ChatViewContent(props: ChatViewProps) { connection: activeEnvironment.connection, }; }, [activeEnvironment, activeEnvironmentUnavailable, activeEnvironmentUnavailableLabel]); + const [environmentUnavailableSince, setEnvironmentUnavailableSince] = useState<{ + readonly environmentId: EnvironmentId; + readonly sinceMs: number; + } | null>(null); + const [environmentUnavailableDebounceClock, setEnvironmentUnavailableDebounceClock] = useState(0); + useEffect(() => { + if (activeEnvironmentUnavailableState === null) { + setEnvironmentUnavailableSince(null); + return; + } + + setEnvironmentUnavailableSince((existing) => + existing?.environmentId === activeEnvironmentUnavailableState.environmentId + ? existing + : { + environmentId: activeEnvironmentUnavailableState.environmentId, + sinceMs: Date.now(), + }, + ); + }, [activeEnvironmentUnavailableState]); + useEffect(() => { + if ( + activeEnvironmentUnavailableState === null || + environmentUnavailableSince === null || + environmentUnavailableSince.environmentId !== activeEnvironmentUnavailableState.environmentId + ) { + return; + } + + const remainingMs = + ENVIRONMENT_UNAVAILABLE_BANNER_DEBOUNCE_MS - + (Date.now() - environmentUnavailableSince.sinceMs); + if (remainingMs <= 0) { + return; + } + + const timeoutId = window.setTimeout(() => { + setEnvironmentUnavailableDebounceClock((clock) => clock + 1); + }, remainingMs); + + return () => window.clearTimeout(timeoutId); + }, [activeEnvironmentUnavailableState, environmentUnavailableSince]); + const debouncedActiveEnvironmentUnavailableState = + useMemo(() => { + if ( + activeEnvironmentUnavailableState === null || + environmentUnavailableSince === null || + environmentUnavailableSince.environmentId !== + activeEnvironmentUnavailableState.environmentId + ) { + return null; + } + + return shouldShowEnvironmentUnavailableBanner({ + connectionPhase: activeEnvironmentUnavailableState.connection.phase, + unavailableSinceMs: environmentUnavailableSince.sinceMs, + nowMs: Date.now(), + }) + ? activeEnvironmentUnavailableState + : null; + }, [ + activeEnvironmentUnavailableState, + environmentUnavailableDebounceClock, + environmentUnavailableSince, + ]); const handleReconnectActiveEnvironment = useCallback( async (environmentId: EnvironmentId) => { const result = await retryEnvironment(environmentId); @@ -2019,15 +2099,15 @@ function ChatViewContent(props: ChatViewProps) { dismissedScheduleBannerKey === scheduleBanner.dismissKey); const composerBannerItems = useMemo(() => { const items: ComposerBannerStackItem[] = []; - if (activeEnvironmentUnavailableState) { - const connection = activeEnvironmentUnavailableState.connection; + if (debouncedActiveEnvironmentUnavailableState) { + const connection = debouncedActiveEnvironmentUnavailableState.connection; const isReconnecting = connection.phase === "connecting" || connection.phase === "reconnecting"; items.push({ - id: `environment-unavailable:${activeEnvironmentUnavailableState.environmentId}`, + id: `environment-unavailable:${debouncedActiveEnvironmentUnavailableState.environmentId}`, variant: connection.phase === "error" ? "error" : "warning", icon: , - title: `${activeEnvironmentUnavailableState.label}: ${connectionStatusText(connection)}`, + title: `${debouncedActiveEnvironmentUnavailableState.label}: ${connectionStatusText(connection)}`, description: connection.error ?? "Plain text messages queue through reloads. Thread setup and attached images retry only while this app stays open.", @@ -2038,7 +2118,7 @@ function ChatViewContent(props: ChatViewProps) { disabled={isReconnecting} onClick={() => void handleReconnectActiveEnvironment( - activeEnvironmentUnavailableState.environmentId, + debouncedActiveEnvironmentUnavailableState.environmentId, ) } > @@ -2092,7 +2172,7 @@ function ChatViewContent(props: ChatViewProps) { } return items; }, [ - activeEnvironmentUnavailableState, + debouncedActiveEnvironmentUnavailableState, handleReconnectActiveEnvironment, navigate, scheduleBanner, @@ -6102,7 +6182,21 @@ function ChatViewContent(props: ChatViewProps) { setThreadError(activeThread.id, null)} + onDismiss={() => { + if (isServerThread) { + const dismissKey = buildThreadErrorDismissKey({ + threadKey: routeThreadKey, + turnId: serverThreadSessionErrorTurnId, + error: threadError, + }); + if (dismissKey !== null) { + setDismissedServerThreadErrorKeys((existing) => + existing[dismissKey] ? existing : { ...existing, [dismissKey]: true }, + ); + } + } + setThreadError(activeThread.id, null); + }} /> {/* Main content area with optional plan sidebar */}
@@ -6201,7 +6295,7 @@ function ChatViewContent(props: ChatViewProps) { isSendBusy={isSendBusy} isPreparingWorktree={isPreparingWorktree} sessionOnlyDisconnectedSendReason={sessionOnlyDisconnectedSendReason} - environmentUnavailable={activeEnvironmentUnavailableState} + environmentUnavailable={debouncedActiveEnvironmentUnavailableState} activePendingApproval={activePendingApproval} pendingApprovals={pendingApprovals} pendingUserInputs={pendingUserInputs} diff --git a/apps/web/src/components/chat/ThreadErrorBanner.tsx b/apps/web/src/components/chat/ThreadErrorBanner.tsx index d0b3ac80b8a..6742fa1b332 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.tsx @@ -13,14 +13,18 @@ export const ThreadErrorBanner = memo(function ThreadErrorBanner({ }) { if (!error) return null; return ( -
- +
+ - }> + + } + > {error} - + {error} diff --git a/packages/client-runtime/src/errors/transport.test.ts b/packages/client-runtime/src/errors/transport.test.ts index 692b3af4a51..83abb5c36e2 100644 --- a/packages/client-runtime/src/errors/transport.test.ts +++ b/packages/client-runtime/src/errors/transport.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it } from "vite-plus/test"; -import { isTransportConnectionErrorMessage, sanitizeThreadErrorMessage } from "./transport.ts"; +import { + INTERRUPTED_TURN_ERROR_MESSAGE, + isClaudeInterruptedTurnDiagnosticMessage, + isTransportConnectionErrorMessage, + sanitizeThreadErrorMessage, +} from "./transport.ts"; describe("isTransportConnectionErrorMessage", () => { it("returns true for SocketCloseError", () => { @@ -53,11 +58,41 @@ describe("isTransportConnectionErrorMessage", () => { }); }); +describe("isClaudeInterruptedTurnDiagnosticMessage", () => { + it("recognizes Claude SDK abort diagnostics", () => { + expect( + isClaudeInterruptedTurnDiagnosticMessage( + "[ede_diagnostic] result_type=user last_content_type=n/a stop_reason=tool_use", + ), + ).toBe(true); + expect( + isClaudeInterruptedTurnDiagnosticMessage( + "provider result terminal_reason=aborted_streaming stop_reason=tool_use", + ), + ).toBe(true); + }); + + it("does not match unrelated provider errors", () => { + expect(isClaudeInterruptedTurnDiagnosticMessage("Claude API key missing")).toBe(false); + }); +}); + describe("sanitizeThreadErrorMessage", () => { it("strips transport errors", () => { expect(sanitizeThreadErrorMessage("SocketCloseError: oops")).toBeNull(); }); + it("replaces raw Claude interrupted-turn diagnostics with friendly copy", () => { + expect( + sanitizeThreadErrorMessage( + "[ede_diagnostic] result_type=user last_content_type=n/a stop_reason=tool_use", + ), + ).toBe(INTERRUPTED_TURN_ERROR_MESSAGE); + expect(sanitizeThreadErrorMessage("Error: Request was aborted.")).toBe( + INTERRUPTED_TURN_ERROR_MESSAGE, + ); + }); + it("preserves non-transport errors", () => { expect(sanitizeThreadErrorMessage("Thread not found")).toBe("Thread not found"); expect(sanitizeThreadErrorMessage("Select a base branch before sending.")).toBe( diff --git a/packages/client-runtime/src/errors/transport.ts b/packages/client-runtime/src/errors/transport.ts index e21c5d4ecf5..fba03f57ec4 100644 --- a/packages/client-runtime/src/errors/transport.ts +++ b/packages/client-runtime/src/errors/transport.ts @@ -11,6 +11,15 @@ const TRANSPORT_ERROR_PATTERNS = [ /\bping timeout\b/i, ] as const; +const CLAUDE_INTERRUPTED_TURN_PATTERNS = [ + /\[ede_diagnostic\][^\n]*(?:stop_reason=tool_use|terminal_reason=aborted_streaming)/i, + /\bterminal_reason=aborted_streaming\b/i, + /^Error:\s*Request was aborted\.?$/i, +] as const; + +export const INTERRUPTED_TURN_ERROR_MESSAGE = + "The turn was interrupted. Send your message again to retry."; + /** * Check whether an error message originates from a transport-level connection * failure (socket close, socket open, ping timeout, etc.) rather than a @@ -29,11 +38,39 @@ export function isTransportConnectionErrorMessage(message: string | null | undef return TRANSPORT_ERROR_PATTERNS.some((pattern) => pattern.test(normalizedMessage)); } +export function isClaudeInterruptedTurnDiagnosticMessage( + message: string | null | undefined, +): boolean { + if (typeof message !== "string") { + return false; + } + + const normalizedMessage = message.trim(); + if (normalizedMessage.length === 0) { + return false; + } + + return CLAUDE_INTERRUPTED_TURN_PATTERNS.some((pattern) => pattern.test(normalizedMessage)); +} + /** * Strip transport connection errors from user-facing error messages. * Returns `null` for transport errors so the UI can distinguish between * real errors and transient connection issues. */ export function sanitizeThreadErrorMessage(message: string | null | undefined): string | null { - return isTransportConnectionErrorMessage(message) ? null : (message ?? null); + if (typeof message !== "string") { + return null; + } + + const normalizedMessage = message.trim(); + if (normalizedMessage.length === 0 || isTransportConnectionErrorMessage(normalizedMessage)) { + return null; + } + + if (isClaudeInterruptedTurnDiagnosticMessage(normalizedMessage)) { + return INTERRUPTED_TURN_ERROR_MESSAGE; + } + + return normalizedMessage; } diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index 4b9564e031c..4df3d16176b 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -33,6 +33,21 @@ describe("server state projection", () => { expect(result.latestEvent.type).toBe("settingsUpdated"); }); + it("keeps websocket heartbeat frames out of the projected server config", () => { + const snapshot = applyServerConfigProjection(Option.none(), { + version: 1, + type: "snapshot", + config: CONFIG, + }); + + expect( + applyServerConfigProjection(snapshot, { + version: 1, + type: "heartbeat", + }), + ).toBe(snapshot); + }); + it("retains welcome when a ready event follows in the same stream chunk", () => { const welcome = { environment: {} as ServerLifecycleWelcomePayload["environment"], diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index fb004efab88..bb1ab2f9730 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -57,6 +57,8 @@ export function applyServerConfigProjection( }, latestEvent: event, })); + case "heartbeat": + return current; } } diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index e56972a2ed5..84d32cec8c9 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -609,11 +609,18 @@ export const ServerConfigStreamSettingsUpdatedEvent = Schema.Struct({ export type ServerConfigStreamSettingsUpdatedEvent = typeof ServerConfigStreamSettingsUpdatedEvent.Type; +export const ServerConfigStreamHeartbeatEvent = Schema.Struct({ + version: Schema.Literal(1), + type: Schema.Literal("heartbeat"), +}); +export type ServerConfigStreamHeartbeatEvent = typeof ServerConfigStreamHeartbeatEvent.Type; + export const ServerConfigStreamEvent = Schema.Union([ ServerConfigStreamSnapshotEvent, ServerConfigStreamKeybindingsUpdatedEvent, ServerConfigStreamProviderStatusesEvent, ServerConfigStreamSettingsUpdatedEvent, + ServerConfigStreamHeartbeatEvent, ]); export type ServerConfigStreamEvent = typeof ServerConfigStreamEvent.Type; From dc2a36fa8695afa989f584e638d675dd82ea579f Mon Sep 17 00:00:00 2001 From: "wizzoapp[bot]" <254688279+wizzoapp[bot]@users.noreply.github.com> Date: Mon, 6 Jul 2026 13:24:15 +0100 Subject: [PATCH 2/2] fix(stability): negotiate ws keepalive heartbeat support (ADA-79) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address Codex review on #87. The keepalive introduced a new `heartbeat` variant of the `ServerConfigStreamEvent` union. Older / version-skewed clients (the mobile app and stale cached web bundles both bundle the shared client-runtime and contracts) decode this union with a closed schema, so an unknown `heartbeat` frame fails decoding ~30s into the connection and tears down the whole `subscribeServerConfig` subscription — exactly the version-skew scenario the version-mismatch banner in this PR is meant to tolerate. Negotiate support instead of emitting unconditionally: - add an optional, additive `supportsHeartbeat` flag to the `subscribeServerConfig` payload (older clients omit it; older servers ignore it on decode — backward compatible both directions); - the server only merges heartbeat frames when the client opts in (`shouldSendServerConfigHeartbeat`), so clients that cannot decode the variant never receive it; - both client-runtime consumers opt in via a single shared `SERVER_CONFIG_SUBSCRIPTION_INPUT` constant, so the subscription stays a single shared stream per environment (no split subscription). Also stop heartbeats from re-emitting the unchanged projection: a quiet session previously pushed the current `ServerConfigProjection` downstream on every 30s tick. `projectServerConfig` now emits no downstream value for heartbeat frames (transport keepalive only). Co-Authored-By: Claude Opus 4.8 --- apps/server/src/ws.ts | 16 +++++++++---- apps/server/src/wsKeepalive.test.ts | 15 +++++++++++- apps/server/src/wsKeepalive.ts | 13 ++++++++++ apps/web/src/state/server.ts | 15 ++++++++---- .../client-runtime/src/state/server.test.ts | 24 ++++++++++++++++++- packages/client-runtime/src/state/server.ts | 20 +++++++++++++++- packages/contracts/src/rpc.ts | 10 +++++++- 7 files changed, 100 insertions(+), 13 deletions(-) diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0bcaec274a8..594776c4ff5 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -114,7 +114,7 @@ import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; import * as SessionStore from "./auth/SessionStore.ts"; import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; -import { makeServerConfigHeartbeatStream } from "./wsKeepalive.ts"; +import { makeServerConfigHeartbeatStream, shouldSendServerConfigHeartbeat } from "./wsKeepalive.ts"; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); const isOrchestrationScheduledTaskMutationError = Schema.is( OrchestrationScheduledTaskMutationError, @@ -1534,7 +1534,7 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "preview" }, ), - [WS_METHODS.subscribeServerConfig]: (_input) => + [WS_METHODS.subscribeServerConfig]: (input) => observeRpcStreamEffect( WS_METHODS.subscribeServerConfig, Effect.gen(function* () { @@ -1569,10 +1569,16 @@ const makeWsRpcLayer = ( .refresh() .pipe(Effect.ignoreCause({ log: true }), Effect.forkScoped); - const liveUpdates = Stream.merge( - makeServerConfigHeartbeatStream(), - Stream.merge(keybindingsUpdates, Stream.merge(providerStatuses, settingsUpdates)), + const configUpdates = Stream.merge( + keybindingsUpdates, + Stream.merge(providerStatuses, settingsUpdates), ); + // Only send keepalive heartbeats to clients that declared support + // for the `heartbeat` union variant; older/version-skewed clients + // would otherwise fail to decode it and lose the subscription. + const liveUpdates = shouldSendServerConfigHeartbeat(input) + ? Stream.merge(makeServerConfigHeartbeatStream(), configUpdates) + : configUpdates; return Stream.concat( Stream.make({ diff --git a/apps/server/src/wsKeepalive.test.ts b/apps/server/src/wsKeepalive.test.ts index 7fb39b53bde..e1854a47797 100644 --- a/apps/server/src/wsKeepalive.test.ts +++ b/apps/server/src/wsKeepalive.test.ts @@ -5,9 +5,22 @@ import * as Fiber from "effect/Fiber"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; -import { makeServerConfigHeartbeatStream, WS_KEEPALIVE_INTERVAL_MS } from "./wsKeepalive.ts"; +import { + makeServerConfigHeartbeatStream, + shouldSendServerConfigHeartbeat, + WS_KEEPALIVE_INTERVAL_MS, +} from "./wsKeepalive.ts"; describe("websocket keepalive", () => { + it("only sends heartbeats to clients that opted in", () => { + expect(shouldSendServerConfigHeartbeat({ supportsHeartbeat: true })).toBe(true); + // Older/version-skewed clients omit the flag and must not receive the new + // `heartbeat` union variant they cannot decode. + expect(shouldSendServerConfigHeartbeat({ supportsHeartbeat: false })).toBe(false); + expect(shouldSendServerConfigHeartbeat({})).toBe(false); + expect(shouldSendServerConfigHeartbeat({ supportsHeartbeat: undefined })).toBe(false); + }); + it.effect("emits application frames well before Cloudflare's idle websocket timeout", () => Effect.gen(function* () { expect(WS_KEEPALIVE_INTERVAL_MS).toBeLessThan(100_000); diff --git a/apps/server/src/wsKeepalive.ts b/apps/server/src/wsKeepalive.ts index 94a51415e8f..7e1acc2c109 100644 --- a/apps/server/src/wsKeepalive.ts +++ b/apps/server/src/wsKeepalive.ts @@ -4,6 +4,19 @@ import * as Stream from "effect/Stream"; export const WS_KEEPALIVE_INTERVAL_MS = 30_000; +/** + * Whether to emit websocket keepalive `heartbeat` frames on this + * `subscribeServerConfig` subscription. Heartbeats are a new + * `ServerConfigStreamEvent` union variant, so they must only be sent to clients + * that declared support — otherwise an older/version-skewed client's schema + * decoder fails on the unknown variant and tears down the config subscription. + */ +export function shouldSendServerConfigHeartbeat(input: { + readonly supportsHeartbeat?: boolean | undefined; +}): boolean { + return input.supportsHeartbeat === true; +} + export function makeServerConfigHeartbeatEvent(): ServerConfigStreamEvent { return { version: 1, diff --git a/apps/web/src/state/server.ts b/apps/web/src/state/server.ts index 3271eefd1e1..0356938f5f1 100644 --- a/apps/web/src/state/server.ts +++ b/apps/web/src/state/server.ts @@ -7,7 +7,10 @@ import { type ServerProvider, type ServerSettings, } from "@t3tools/contracts"; -import { createServerEnvironmentAtoms } from "@t3tools/client-runtime/state/server"; +import { + createServerEnvironmentAtoms, + SERVER_CONFIG_SUBSCRIPTION_INPUT, +} from "@t3tools/client-runtime/state/server"; import { createEnvironmentServerConfigsAtom } from "@t3tools/client-runtime/state/shell"; import { DEFAULT_RESOLVED_KEYBINDINGS } from "@t3tools/shared/keybindings"; import * as Option from "effect/Option"; @@ -46,11 +49,15 @@ export const primaryServerStateAtom = Atom.make((get): PrimaryServerState => { return EMPTY_PRIMARY_SERVER_STATE; } - const target = { environmentId, input: {} }; + // Must match the input used by `configValueAtom` in client-runtime so both + // consumers share a single `subscribeServerConfig` stream per environment. + const configTarget = { environmentId, input: SERVER_CONFIG_SUBSCRIPTION_INPUT }; const configProjection = Option.getOrNull( - AsyncResult.value(get(serverEnvironment.configProjection(target))), + AsyncResult.value(get(serverEnvironment.configProjection(configTarget))), + ); + const welcome = Option.getOrNull( + AsyncResult.value(get(serverEnvironment.welcome({ environmentId, input: {} }))), ); - const welcome = Option.getOrNull(AsyncResult.value(get(serverEnvironment.welcome(target)))); return { config: get(serverEnvironment.configValueAtom(environmentId)), diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index 4df3d16176b..8947a9ae950 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -2,7 +2,11 @@ import { type ServerConfig, type ServerLifecycleWelcomePayload } from "@t3tools/ import { describe, expect, it } from "@effect/vitest"; import * as Option from "effect/Option"; -import { applyServerConfigProjection, projectServerWelcome } from "./server.ts"; +import { + applyServerConfigProjection, + projectServerConfig, + projectServerWelcome, +} from "./server.ts"; const CONFIG = { availableEditors: [], @@ -48,6 +52,24 @@ describe("server state projection", () => { ).toBe(snapshot); }); + it("emits no downstream projection for websocket heartbeat frames", () => { + const [snapshotState] = projectServerConfig(Option.none(), { + version: 1, + type: "snapshot", + config: CONFIG, + }); + + const [nextState, emitted] = projectServerConfig(snapshotState, { + version: 1, + type: "heartbeat", + }); + + // State is preserved but nothing is pushed to subscribers, so a quiet + // session does not re-emit the unchanged config every keepalive tick. + expect(nextState).toBe(snapshotState); + expect(emitted).toEqual([]); + }); + it("retains welcome when a ready event follows in the same stream chunk", () => { const welcome = { environment: {} as ServerLifecycleWelcomePayload["environment"], diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index bb1ab2f9730..cc080d16fd7 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -67,6 +67,12 @@ export function projectServerConfig( event: ServerConfigStreamEvent, ): readonly [Option.Option, ReadonlyArray] { const next = applyServerConfigProjection(current, event); + if (event.type === "heartbeat") { + // Heartbeats are transport keepalive only. Never surface a downstream + // projection for them — otherwise a quiet session re-emits the unchanged + // config every keepalive tick and can wake server-config consumers. + return [next, []]; + } return [next, Option.toArray(next)]; } @@ -87,6 +93,16 @@ export function projectServerWelcome( return [Option.some(welcome), [welcome]]; } +/** + * Shared payload for every `subscribeServerConfig` consumer. The subscription + * atom family is keyed on `JSON.stringify([environmentId, input])`, so ALL + * consumers must pass this identical value — otherwise the environment ends up + * with two independent websocket streams (each running its own + * `providerRegistry.refresh()`). `supportsHeartbeat` opts this heartbeat-aware + * client build into the server's keepalive frames. + */ +export const SERVER_CONFIG_SUBSCRIPTION_INPUT = { supportsHeartbeat: true } as const; + export function createServerEnvironmentAtoms( runtime: Atom.AtomRuntime, options: { @@ -115,7 +131,9 @@ export function createServerEnvironmentAtoms( } return Atom.make((get): ServerConfig | null => { const projection = Option.getOrNull( - AsyncResult.value(get(configProjection({ environmentId, input: {} }))), + AsyncResult.value( + get(configProjection({ environmentId, input: SERVER_CONFIG_SUBSCRIPTION_INPUT })), + ), ); return projection?.config ?? get(options.initialConfigValueAtom(environmentId)); }).pipe(Atom.withLabel(`environment-data:server:config:${environmentId}`)); diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 4aa1aae1ed6..70e69417364 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -727,7 +727,15 @@ export const WsSubscribeTerminalMetadataRpc = Rpc.make(WS_METHODS.subscribeTermi }); export const WsSubscribeServerConfigRpc = Rpc.make(WS_METHODS.subscribeServerConfig, { - payload: Schema.Struct({}), + // `supportsHeartbeat` is an additive, backward-compatible capability flag. + // Clients built before the websocket keepalive omit it (decoded as + // `undefined`), and the server only merges `heartbeat` frames into the stream + // when a client opts in — so an older client's `ServerConfigStreamEvent` union + // decoder never receives a `heartbeat` variant it cannot parse. A newer client + // talking to an older server is also safe: the extra field is ignored on decode. + payload: Schema.Struct({ + supportsHeartbeat: Schema.optional(Schema.Boolean), + }), success: ServerConfigStreamEvent, error: Schema.Union([KeybindingsConfigError, ServerSettingsError, EnvironmentAuthorizationError]), stream: true,