Commit f2cd53f232198f597bf492504cab5d9151a5abca | the-knife

author	Julius Marminge <julius0216@outlook.com> 2026-04-04 17:12:49-07:00
committer	GitHub <noreply@github.com> 2026-04-04 17:12:49-07:00
log	f2cd53f232198f597bf492504cab5d9151a5abca
tree	9e9c1d477d6620aaebff6868ad7db92adc180179
parent	04a1ae77e6b120c6d9b6a51768a1e411afe970be
signature	Signed by PGP key B5690EEEBB952194

Add WebSocket disconnect recovery and slow RPC toast UX (#1730)

Co-authored-by: Hwanseo Choi <hwanseoc@nvidia.com>
Co-authored-by: codex <codex@users.noreply.github.com>

20 files changed, 1836 insertions(+), 103 deletions(-)

apps/web/src/components/ChatView.tsx+9-7

...	...	@@ -197,6 +197,7 @@ import {
197	197	useServerConfig,
198	198	useServerKeybindings,
199	199	} from "~/rpc/serverState";
	200	import { sanitizeThreadErrorMessage } from "~/rpc/transportError";
200	201
201	202	const ATTACHMENT_PREVIEW_HANDOFF_TTL_MS = 5000;
202	203	const IMAGE_SIZE_LIMIT_LABEL = `${Math.round(PROVIDER_SEND_TURN_MAX_IMAGE_BYTES / (1024 * 1024))}MB`;
...	...	@@ -1599,17 +1600,18 @@ export default function ChatView({ threadId }: ChatViewProps) {
1599	1600	const setThreadError = useCallback(
1600	1601	(targetThreadId: ThreadId \| null, error: string \| null) => {
1601	1602	if (!targetThreadId) return;
	1603	const nextError = sanitizeThreadErrorMessage(error);
1602	1604	if (useStore.getState().threads.some((thread) => thread.id === targetThreadId)) {
1603		setStoreThreadError(targetThreadId, error);
	1605	setStoreThreadError(targetThreadId, nextError);
1604	1606	return;
1605	1607	}
1606	1608	setLocalDraftErrorsByThreadId((existing) => {
1607		if ((existing[targetThreadId] ?? null) === error) {
	1609	if ((existing[targetThreadId] ?? null) === nextError) {
1608	1610	return existing;
1609	1611	}
1610	1612	return {
1611	1613	...existing,
1612		[targetThreadId]: error,
	1614	[targetThreadId]: nextError,
1613	1615	};
1614	1616	});
1615	1617	},
...	...	@@ -3153,14 +3155,14 @@ export default function ChatView({ threadId }: ChatViewProps) {
3153	3155	createdAt: new Date().toISOString(),
3154	3156	})
3155	3157	.catch((err: unknown) => {
3156		setStoreThreadError(
	3158	setThreadError(
3157	3159	activeThreadId,
3158	3160	err instanceof Error ? err.message : "Failed to submit approval decision.",
3159	3161	);
3160	3162	});
3161	3163	setRespondingRequestIds((existing) => existing.filter((id) => id !== requestId));
3162	3164	},
3163		[activeThreadId, setStoreThreadError],
	3165	[activeThreadId, setThreadError],
3164	3166	);
3165	3167
3166	3168	const onRespondToUserInput = useCallback(
...	...	@@ -3181,14 +3183,14 @@ export default function ChatView({ threadId }: ChatViewProps) {
3181	3183	createdAt: new Date().toISOString(),
3182	3184	})
3183	3185	.catch((err: unknown) => {
3184		setStoreThreadError(
	3186	setThreadError(
3185	3187	activeThreadId,
3186	3188	err instanceof Error ? err.message : "Failed to submit user input.",
3187	3189	);
3188	3190	});
3189	3191	setRespondingUserInputRequestIds((existing) => existing.filter((id) => id !== requestId));
3190	3192	},
3191		[activeThreadId, setStoreThreadError],
	3193	[activeThreadId, setThreadError],
3192	3194	);
3193	3195
3194	3196	const setActivePendingUserInputQuestionIndex = useCallback(

apps/web/src/components/WebSocketConnectionSurface.logic.test.ts created+83

...	...	@@ -0,0 +1,83 @@
	1	import { describe, expect, it } from "vitest";
	2
	3	import type { WsConnectionStatus } from "../rpc/wsConnectionState";
	4	import { shouldAutoReconnect } from "./WebSocketConnectionSurface";
	5
	6	function makeStatus(overrides: Partial<WsConnectionStatus> = {}): WsConnectionStatus {
	7	return {
	8	attemptCount: 0,
	9	closeCode: null,
	10	closeReason: null,
	11	connectedAt: null,
	12	disconnectedAt: null,
	13	hasConnected: false,
	14	lastError: null,
	15	lastErrorAt: null,
	16	nextRetryAt: null,
	17	online: true,
	18	phase: "idle",
	19	reconnectAttemptCount: 0,
	20	reconnectMaxAttempts: 8,
	21	reconnectPhase: "idle",
	22	socketUrl: null,
	23	...overrides,
	24	};
	25	}
	26
	27	describe("WebSocketConnectionSurface.logic", () => {
	28	it("forces reconnect on online when the app was offline", () => {
	29	expect(
	30	shouldAutoReconnect(
	31	makeStatus({
	32	disconnectedAt: "2026-04-03T20:00:00.000Z",
	33	online: false,
	34	phase: "disconnected",
	35	}),
	36	"online",
	37	),
	38	).toBe(true);
	39	});
	40
	41	it("forces reconnect on focus only for previously connected disconnected states", () => {
	42	expect(
	43	shouldAutoReconnect(
	44	makeStatus({
	45	hasConnected: true,
	46	online: true,
	47	phase: "disconnected",
	48	reconnectAttemptCount: 3,
	49	reconnectPhase: "waiting",
	50	}),
	51	"focus",
	52	),
	53	).toBe(true);
	54
	55	expect(
	56	shouldAutoReconnect(
	57	makeStatus({
	58	hasConnected: false,
	59	online: true,
	60	phase: "disconnected",
	61	reconnectAttemptCount: 1,
	62	reconnectPhase: "waiting",
	63	}),
	64	"focus",
	65	),
	66	).toBe(false);
	67	});
	68
	69	it("forces reconnect on focus for exhausted reconnect loops", () => {
	70	expect(
	71	shouldAutoReconnect(
	72	makeStatus({
	73	hasConnected: true,
	74	online: true,
	75	phase: "disconnected",
	76	reconnectAttemptCount: 8,
	77	reconnectPhase: "exhausted",
	78	}),
	79	"focus",
	80	),
	81	).toBe(true);
	82	});
	83	});

apps/web/src/components/WebSocketConnectionSurface.tsx created+544

...	...	@@ -0,0 +1,544 @@
	1	import { AlertTriangle, CloudOff, LoaderCircle, RotateCw } from "lucide-react";
	2	import { type ReactNode, useEffect, useEffectEvent, useRef, useState } from "react";
	3
	4	import { APP_DISPLAY_NAME } from "../branding";
	5	import { type SlowRpcAckRequest, useSlowRpcAckRequests } from "../rpc/requestLatencyState";
	6	import { useServerConfig } from "../rpc/serverState";
	7	import {
	8	exhaustWsReconnectIfStillWaiting,
	9	getWsConnectionStatus,
	10	getWsConnectionUiState,
	11	setBrowserOnlineStatus,
	12	type WsConnectionStatus,
	13	type WsConnectionUiState,
	14	useWsConnectionStatus,
	15	WS_RECONNECT_MAX_ATTEMPTS,
	16	} from "../rpc/wsConnectionState";
	17	import { Button } from "./ui/button";
	18	import { toastManager } from "./ui/toast";
	19	import { getWsRpcClient } from "~/wsRpcClient";
	20
	21	const FORCED_WS_RECONNECT_DEBOUNCE_MS = 5_000;
	22	type WsAutoReconnectTrigger = "focus" \| "online";
	23
	24	const connectionTimeFormatter = new Intl.DateTimeFormat(undefined, {
	25	day: "numeric",
	26	hour: "numeric",
	27	minute: "2-digit",
	28	month: "short",
	29	second: "2-digit",
	30	});
	31
	32	function formatConnectionMoment(isoDate: string \| null): string \| null {
	33	if (!isoDate) {
	34	return null;
	35	}
	36
	37	return connectionTimeFormatter.format(new Date(isoDate));
	38	}
	39
	40	function formatRetryCountdown(nextRetryAt: string, nowMs: number): string {
	41	const remainingMs = Math.max(0, new Date(nextRetryAt).getTime() - nowMs);
	42	return `${Math.max(1, Math.ceil(remainingMs / 1000))}s`;
	43	}
	44
	45	function describeOfflineToast(): string {
	46	return "WebSocket disconnected. Waiting for network.";
	47	}
	48
	49	function formatReconnectAttemptLabel(status: WsConnectionStatus): string {
	50	const reconnectAttempt = Math.max(
	51	1,
	52	Math.min(status.reconnectAttemptCount, WS_RECONNECT_MAX_ATTEMPTS),
	53	);
	54	return `Attempt ${reconnectAttempt}/${status.reconnectMaxAttempts}`;
	55	}
	56
	57	function describeExhaustedToast(): string {
	58	return "Retries exhausted trying to reconnect";
	59	}
	60
	61	function buildReconnectTitle(status: WsConnectionStatus): string {
	62	if (status.nextRetryAt === null) {
	63	return "Disconnected from T3 Server";
	64	}
	65
	66	return "Disconnected from T3 Server";
	67	}
	68
	69	function describeRecoveredToast(
	70	previousDisconnectedAt: string \| null,
	71	connectedAt: string \| null,
	72	): string {
	73	const reconnectedAtLabel = formatConnectionMoment(connectedAt);
	74	const disconnectedAtLabel = formatConnectionMoment(previousDisconnectedAt);
	75
	76	if (disconnectedAtLabel && reconnectedAtLabel) {
	77	return `Disconnected at ${disconnectedAtLabel} and reconnected at ${reconnectedAtLabel}.`;
	78	}
	79
	80	if (reconnectedAtLabel) {
	81	return `Connection restored at ${reconnectedAtLabel}.`;
	82	}
	83
	84	return "Connection restored.";
	85	}
	86
	87	function describeSlowRpcAckToast(requests: ReadonlyArray<SlowRpcAckRequest>): ReactNode {
	88	const count = requests.length;
	89	const thresholdSeconds = Math.round((requests[0]?.thresholdMs ?? 0) / 1000);
	90
	91	return `${count} request${count === 1 ? "" : "s"} waiting longer than ${thresholdSeconds}s.`;
	92	}
	93
	94	export function shouldAutoReconnect(
	95	status: WsConnectionStatus,
	96	trigger: WsAutoReconnectTrigger,
	97	): boolean {
	98	const uiState = getWsConnectionUiState(status);
	99
	100	if (trigger === "online") {
	101	return (
	102	uiState === "offline" \|\|
	103	uiState === "reconnecting" \|\|
	104	uiState === "error" \|\|
	105	status.reconnectPhase === "exhausted"
	106	);
	107	}
	108
	109	return (
	110	status.online &&
	111	status.hasConnected &&
	112	(uiState === "reconnecting" \|\| status.reconnectPhase === "exhausted")
	113	);
	114	}
	115
	116	function buildBlockingCopy(
	117	uiState: WsConnectionUiState,
	118	status: WsConnectionStatus,
	119	): {
	120	readonly description: string;
	121	readonly eyebrow: string;
	122	readonly title: string;
	123	} {
	124	if (uiState === "connecting") {
	125	return {
	126	description: `Opening the WebSocket connection to the ${APP_DISPLAY_NAME} server and waiting for the initial config snapshot.`,
	127	eyebrow: "Starting Session",
	128	title: `Connecting to ${APP_DISPLAY_NAME}`,
	129	};
	130	}
	131
	132	if (uiState === "offline") {
	133	return {
	134	description:
	135	"Your browser is offline, so the web client cannot reach the T3 server. Reconnect to the network and the app will retry automatically.",
	136	eyebrow: "Offline",
	137	title: "WebSocket connection unavailable",
	138	};
	139	}
	140
	141	if (status.lastError?.trim()) {
	142	return {
	143	description: `${status.lastError} Verify that the T3 server is running and reachable, then reload the app if needed.`,
	144	eyebrow: "Connection Error",
	145	title: "Cannot reach the T3 server",
	146	};
	147	}
	148
	149	return {
	150	description:
	151	"The web client could not complete its initial WebSocket connection to the T3 server. It will keep retrying in the background.",
	152	eyebrow: "Connection Error",
	153	title: "Cannot reach the T3 server",
	154	};
	155	}
	156
	157	function buildConnectionDetails(status: WsConnectionStatus, uiState: WsConnectionUiState): string {
	158	const details = [
	159	`state: ${uiState}`,
	160	`online: ${status.online ? "yes" : "no"}`,
	161	`attempts: ${status.attemptCount}`,
	162	];
	163
	164	if (status.socketUrl) {
	165	details.push(`socket: ${status.socketUrl}`);
	166	}
	167	if (status.connectedAt) {
	168	details.push(`connectedAt: ${status.connectedAt}`);
	169	}
	170	if (status.disconnectedAt) {
	171	details.push(`disconnectedAt: ${status.disconnectedAt}`);
	172	}
	173	if (status.lastErrorAt) {
	174	details.push(`lastErrorAt: ${status.lastErrorAt}`);
	175	}
	176	if (status.lastError) {
	177	details.push(`lastError: ${status.lastError}`);
	178	}
	179	if (status.closeCode !== null) {
	180	details.push(`closeCode: ${status.closeCode}`);
	181	}
	182	if (status.closeReason) {
	183	details.push(`closeReason: ${status.closeReason}`);
	184	}
	185
	186	return details.join("\n");
	187	}
	188
	189	function WebSocketBlockingState({
	190	status,
	191	uiState,
	192	}: {
	193	readonly status: WsConnectionStatus;
	194	readonly uiState: WsConnectionUiState;
	195	}) {
	196	const copy = buildBlockingCopy(uiState, status);
	197	const disconnectedAt = formatConnectionMoment(status.disconnectedAt ?? status.lastErrorAt);
	198	const Icon =
	199	uiState === "connecting" ? LoaderCircle : uiState === "offline" ? CloudOff : AlertTriangle;
	200
	201	return (
	202	<div className="relative flex min-h-screen items-center justify-center overflow-hidden bg-background px-4 py-10 text-foreground sm:px-6">
	203	<div className="pointer-events-none absolute inset-0 opacity-90">
	204	<div className="absolute inset-x-0 top-0 h-56 bg-[radial-gradient(48rem_18rem_at_top,color-mix(in_srgb,var(--color-amber-500)_16%,transparent),transparent)]" />
	205	<div className="absolute inset-0 bg-[linear-gradient(145deg,color-mix(in_srgb,var(--background)_92%,var(--color-black))_0%,var(--background)_56%)]" />
	206	</div>
	207
	208	<section className="relative w-full max-w-xl rounded-[1.75rem] border border-border/80 bg-card/92 p-6 shadow-2xl shadow-black/20 backdrop-blur-md sm:p-8">
	209	<div className="flex items-start justify-between gap-4">
	210	<div>
	211	<p className="text-[11px] font-semibold tracking-[0.18em] text-muted-foreground uppercase">
	212	{copy.eyebrow}
	213	</p>
	214	<h1 className="mt-3 text-2xl font-semibold tracking-tight sm:text-3xl">{copy.title}</h1>
	215	</div>
	216	<div className="rounded-2xl border border-border/70 bg-background/80 p-3 text-foreground shadow-sm">
	217	<Icon className={uiState === "connecting" ? "size-5 animate-spin" : "size-5"} />
	218	</div>
	219	</div>
	220
	221	<p className="mt-3 text-sm leading-relaxed text-muted-foreground">{copy.description}</p>
	222
	223	<div className="mt-5 grid gap-3 rounded-2xl border border-border/70 bg-background/60 p-4 text-sm sm:grid-cols-2">
	224	<div>
	225	<p className="text-[11px] font-semibold tracking-[0.16em] text-muted-foreground uppercase">
	226	Connection
	227	</p>
	228	<p className="mt-1 font-medium text-foreground">
	229	{uiState === "connecting"
	230	? "Opening WebSocket"
	231	: uiState === "offline"
	232	? "Waiting for network"
	233	: "Retrying server connection"}
	234	</p>
	235	</div>
	236	<div>
	237	<p className="text-[11px] font-semibold tracking-[0.16em] text-muted-foreground uppercase">
	238	Latest Event
	239	</p>
	240	<p className="mt-1 font-medium text-foreground">{disconnectedAt ?? "Pending"}</p>
	241	</div>
	242	</div>
	243
	244	<div className="mt-5 flex flex-wrap gap-2">
	245	<Button size="sm" onClick={() => window.location.reload()}>
	246	<RotateCw />
	247	Reload app
	248	</Button>
	249	</div>
	250
	251	<details className="group mt-5 overflow-hidden rounded-lg border border-border/70 bg-background/55">
	252	<summary className="cursor-pointer list-none px-3 py-2 text-xs font-medium text-muted-foreground">
	253	<span className="group-open:hidden">Show connection details</span>
	254	<span className="hidden group-open:inline">Hide connection details</span>
	255	</summary>
	256	<pre className="max-h-56 overflow-auto border-t border-border/70 bg-background/80 px-3 py-2 text-xs text-foreground/85">
	257	{buildConnectionDetails(status, uiState)}
	258	</pre>
	259	</details>
	260	</section>
	261	</div>
	262	);
	263	}
	264
	265	export function WebSocketConnectionCoordinator() {
	266	const status = useWsConnectionStatus();
	267	const [nowMs, setNowMs] = useState(() => Date.now());
	268	const lastForcedReconnectAtRef = useRef(0);
	269	const toastIdRef = useRef<ReturnType<typeof toastManager.add> \| null>(null);
	270	const toastResetTimerRef = useRef<number \| null>(null);
	271	const previousUiStateRef = useRef<WsConnectionUiState>(getWsConnectionUiState(status));
	272	const previousDisconnectedAtRef = useRef<string \| null>(status.disconnectedAt);
	273
	274	const runReconnect = useEffectEvent((showFailureToast: boolean) => {
	275	if (toastResetTimerRef.current !== null) {
	276	window.clearTimeout(toastResetTimerRef.current);
	277	toastResetTimerRef.current = null;
	278	}
	279	lastForcedReconnectAtRef.current = Date.now();
	280	void getWsRpcClient()
	281	.reconnect()
	282	.catch((error) => {
	283	if (!showFailureToast) {
	284	console.warn("Automatic WebSocket reconnect failed", { error });
	285	return;
	286	}
	287	toastManager.add({
	288	type: "error",
	289	title: "Reconnect failed",
	290	description: error instanceof Error ? error.message : "Unable to restart the WebSocket.",
	291	data: {
	292	dismissAfterVisibleMs: 8_000,
	293	hideCopyButton: true,
	294	},
	295	});
	296	});
	297	});
	298	const syncBrowserOnlineStatus = useEffectEvent(() => {
	299	setBrowserOnlineStatus(navigator.onLine !== false);
	300	});
	301	const triggerManualReconnect = useEffectEvent(() => {
	302	runReconnect(true);
	303	});
	304	const triggerAutoReconnect = useEffectEvent((trigger: WsAutoReconnectTrigger) => {
	305	const currentStatus =
	306	trigger === "online" ? setBrowserOnlineStatus(true) : getWsConnectionStatus();
	307
	308	if (!shouldAutoReconnect(currentStatus, trigger)) {
	309	return;
	310	}
	311	if (Date.now() - lastForcedReconnectAtRef.current < FORCED_WS_RECONNECT_DEBOUNCE_MS) {
	312	return;
	313	}
	314
	315	runReconnect(false);
	316	});
	317
	318	useEffect(() => {
	319	const handleOnline = () => {
	320	triggerAutoReconnect("online");
	321	};
	322	const handleFocus = () => {
	323	triggerAutoReconnect("focus");
	324	};
	325
	326	syncBrowserOnlineStatus();
	327	window.addEventListener("online", handleOnline);
	328	window.addEventListener("offline", syncBrowserOnlineStatus);
	329	window.addEventListener("focus", handleFocus);
	330	return () => {
	331	window.removeEventListener("online", handleOnline);
	332	window.removeEventListener("offline", syncBrowserOnlineStatus);
	333	window.removeEventListener("focus", handleFocus);
	334	};
	335	}, []);
	336
	337	useEffect(() => {
	338	if (status.reconnectPhase !== "waiting" \|\| status.nextRetryAt === null) {
	339	return;
	340	}
	341
	342	setNowMs(Date.now());
	343	const intervalId = window.setInterval(() => {
	344	setNowMs(Date.now());
	345	}, 1_000);
	346
	347	return () => {
	348	window.clearInterval(intervalId);
	349	};
	350	}, [status.nextRetryAt, status.reconnectPhase]);
	351
	352	useEffect(() => {
	353	if (
	354	status.reconnectPhase !== "waiting" \|\|
	355	status.nextRetryAt === null \|\|
	356	!status.online \|\|
	357	!status.hasConnected
	358	) {
	359	return;
	360	}
	361
	362	const nextRetryAt = status.nextRetryAt;
	363	const timeoutMs = Math.max(0, new Date(nextRetryAt).getTime() - Date.now()) + 1_500;
	364	const timeoutId = window.setTimeout(() => {
	365	exhaustWsReconnectIfStillWaiting(nextRetryAt);
	366	}, timeoutMs);
	367
	368	return () => {
	369	window.clearTimeout(timeoutId);
	370	};
	371	}, [
	372	status.hasConnected,
	373	status.nextRetryAt,
	374	status.online,
	375	status.reconnectAttemptCount,
	376	status.reconnectPhase,
	377	]);
	378
	379	useEffect(() => {
	380	const uiState = getWsConnectionUiState(status);
	381	const previousUiState = previousUiStateRef.current;
	382	const previousDisconnectedAt = previousDisconnectedAtRef.current;
	383	const shouldShowReconnectToast = status.hasConnected && uiState === "reconnecting";
	384	const shouldShowOfflineToast = uiState === "offline" && status.disconnectedAt !== null;
	385	const shouldShowExhaustedToast = status.hasConnected && status.reconnectPhase === "exhausted";
	386
	387	if (
	388	toastResetTimerRef.current !== null &&
	389	(shouldShowReconnectToast \|\| shouldShowOfflineToast \|\| shouldShowExhaustedToast)
	390	) {
	391	window.clearTimeout(toastResetTimerRef.current);
	392	toastResetTimerRef.current = null;
	393	}
	394
	395	if (shouldShowReconnectToast \|\| shouldShowOfflineToast \|\| shouldShowExhaustedToast) {
	396	const toastPayload = shouldShowOfflineToast
	397	? {
	398	description: describeOfflineToast(),
	399	timeout: 0,
	400	title: "Offline",
	401	type: "warning" as const,
	402	data: {
	403	hideCopyButton: true,
	404	},
	405	}
	406	: shouldShowExhaustedToast
	407	? {
	408	actionProps: {
	409	children: "Retry",
	410	onClick: triggerManualReconnect,
	411	},
	412	description: describeExhaustedToast(),
	413	timeout: 0,
	414	title: "Disconnected from T3 Server",
	415	type: "error" as const,
	416	data: {
	417	hideCopyButton: true,
	418	},
	419	}
	420	: {
	421	actionProps: {
	422	children: "Retry now",
	423	onClick: triggerManualReconnect,
	424	},
	425	description:
	426	status.nextRetryAt === null
	427	? `Reconnecting... ${formatReconnectAttemptLabel(status)}`
	428	: `Reconnecting in ${formatRetryCountdown(status.nextRetryAt, nowMs)}... ${formatReconnectAttemptLabel(status)}`,
	429	timeout: 0,
	430	title: buildReconnectTitle(status),
	431	type: "loading" as const,
	432	data: {
	433	hideCopyButton: true,
	434	},
	435	};
	436
	437	if (toastIdRef.current) {
	438	toastManager.update(toastIdRef.current, toastPayload);
	439	} else {
	440	toastIdRef.current = toastManager.add(toastPayload);
	441	}
	442	} else if (toastIdRef.current) {
	443	toastManager.close(toastIdRef.current);
	444	toastIdRef.current = null;
	445	}
	446
	447	if (
	448	uiState === "connected" &&
	449	(previousUiState === "offline" \|\| previousUiState === "reconnecting") &&
	450	previousDisconnectedAt !== null
	451	) {
	452	const successToast = {
	453	description: describeRecoveredToast(previousDisconnectedAt, status.connectedAt),
	454	title: "Reconnected to T3 Server",
	455	type: "success" as const,
	456	timeout: 0,
	457	data: {
	458	dismissAfterVisibleMs: 8_000,
	459	hideCopyButton: true,
	460	},
	461	};
	462
	463	if (toastIdRef.current) {
	464	toastManager.update(toastIdRef.current, successToast);
	465	} else {
	466	toastIdRef.current = toastManager.add(successToast);
	467	}
	468
	469	toastResetTimerRef.current = window.setTimeout(() => {
	470	toastIdRef.current = null;
	471	toastResetTimerRef.current = null;
	472	}, 8_250);
	473	}
	474
	475	previousUiStateRef.current = uiState;
	476	previousDisconnectedAtRef.current = status.disconnectedAt;
	477	}, [nowMs, status]);
	478
	479	useEffect(() => {
	480	return () => {
	481	if (toastResetTimerRef.current !== null) {
	482	window.clearTimeout(toastResetTimerRef.current);
	483	}
	484	};
	485	}, []);
	486
	487	return null;
	488	}
	489
	490	export function SlowRpcAckToastCoordinator() {
	491	const slowRequests = useSlowRpcAckRequests();
	492	const status = useWsConnectionStatus();
	493	const toastIdRef = useRef<ReturnType<typeof toastManager.add> \| null>(null);
	494
	495	useEffect(() => {
	496	if (getWsConnectionUiState(status) !== "connected") {
	497	if (toastIdRef.current) {
	498	toastManager.close(toastIdRef.current);
	499	toastIdRef.current = null;
	500	}
	501	return;
	502	}
	503
	504	if (slowRequests.length === 0) {
	505	if (toastIdRef.current) {
	506	toastManager.close(toastIdRef.current);
	507	toastIdRef.current = null;
	508	}
	509	return;
	510	}
	511
	512	const nextToast = {
	513	description: describeSlowRpcAckToast(slowRequests),
	514	timeout: 0,
	515	title: "Some requests are slow",
	516	type: "warning" as const,
	517	};
	518
	519	if (toastIdRef.current) {
	520	toastManager.update(toastIdRef.current, nextToast);
	521	} else {
	522	toastIdRef.current = toastManager.add(nextToast);
	523	}
	524	}, [slowRequests, status]);
	525
	526	return null;
	527	}
	528
	529	export function WebSocketConnectionSurface({ children }: { readonly children: ReactNode }) {
	530	const serverConfig = useServerConfig();
	531	const status = useWsConnectionStatus();
	532
	533	if (serverConfig === null) {
	534	const uiState = getWsConnectionUiState(status);
	535	return (
	536	<WebSocketBlockingState
	537	status={status}
	538	uiState={uiState === "connected" ? "connecting" : uiState}
	539	/>
	540	);
	541	}
	542
	543	return children;
	544	}

apps/web/src/components/ui/toast.tsx+13-10

...	...	@@ -23,6 +23,7 @@ export type ThreadToastData = {
23	23	threadId?: ThreadId \| null;
24	24	tooltipStyle?: boolean;
25	25	dismissAfterVisibleMs?: number;
	26	hideCopyButton?: boolean;
26	27	};
27	28
28	29	const toastManager = Toast.createToastManager<ThreadToastData>();
...	...	@@ -308,15 +309,15 @@ function Toasts({ position = "top-right" }: { position: ToastPosition }) {
308	309	<div className="flex min-w-0 flex-1 flex-col gap-0.5">
309	310	<div className="flex items-center justify-between gap-1">
310	311	<Toast.Title
311		className="min-w-0 break-words font-medium"
	312	className="min-w-0 wrap-break-word font-medium"
312	313	data-slot="toast-title"
313	314	/>
314		{toast.type === "error" && typeof toast.description === "string" && (
315		<CopyErrorButton text={toast.description} />
316		)}
	315	{toast.type === "error" &&
	316	typeof toast.description === "string" &&
	317	!toast.data?.hideCopyButton && <CopyErrorButton text={toast.description} />}
317	318	</div>
318	319	<Toast.Description
319		className="min-w-0 select-text break-words text-muted-foreground"
	320	className="min-w-0 select-text wrap-break-word text-muted-foreground"
320	321	data-slot="toast-description"
321	322	/>
322	323	</div>
...	...	@@ -402,15 +403,17 @@ function AnchoredToasts() {
402	403	<div className="flex min-w-0 flex-1 flex-col gap-0.5">
403	404	<div className="flex items-center gap-1">
404	405	<Toast.Title
405		className="min-w-0 break-words font-medium"
	406	className="min-w-0 wrap-break-word font-medium"
406	407	data-slot="toast-title"
407	408	/>
408		{toast.type === "error" && typeof toast.description === "string" && (
409		<CopyErrorButton text={toast.description} />
410		)}
	409	{toast.type === "error" &&
	410	typeof toast.description === "string" &&
	411	!toast.data?.hideCopyButton && (
	412	<CopyErrorButton text={toast.description} />
	413	)}
411	414	</div>
412	415	<Toast.Description
413		className="min-w-0 select-text break-words text-muted-foreground"
	416	className="min-w-0 select-text wrap-break-word text-muted-foreground"
414	417	data-slot="toast-description"
415	418	/>
416	419	</div>

apps/web/src/orchestrationRecovery.ts+5-1

...	...	@@ -1,4 +1,8 @@
1		export type OrchestrationRecoveryReason = "bootstrap" \| "sequence-gap" \| "replay-failed";
	1	export type OrchestrationRecoveryReason =
	2	\| "bootstrap"
	3	\| "sequence-gap"
	4	\| "resubscribe"
	5	\| "replay-failed";
2	6
3	7	export interface OrchestrationRecoveryPhase {
4	8	kind: "snapshot" \| "replay";

apps/web/src/routes/__root.tsx+39-19

...	...	@@ -16,6 +16,11 @@ import { Throttler } from "@tanstack/react-pacer";
16	16
17	17	import { APP_DISPLAY_NAME } from "../branding";
18	18	import { AppSidebarLayout } from "../components/AppSidebarLayout";
	19	import {
	20	SlowRpcAckToastCoordinator,
	21	WebSocketConnectionCoordinator,
	22	WebSocketConnectionSurface,
	23	} from "../components/WebSocketConnectionSurface";
19	24	import { Button } from "../components/ui/button";
20	25	import { AnchoredToastProvider, ToastProvider, toastManager } from "../components/ui/toast";
21	26	import { resolveAndPersistPreferredEditor } from "../editorPreferences";
...	...	@@ -73,9 +78,13 @@ function RootRouteView() {
73	78	<AnchoredToastProvider>
74	79	<ServerStateBootstrap />
75	80	<EventRouter />
76		<AppSidebarLayout>
77		<Outlet />
78		</AppSidebarLayout>
	81	<WebSocketConnectionCoordinator />
	82	<SlowRpcAckToastCoordinator />
	83	<WebSocketConnectionSurface>
	84	<AppSidebarLayout>
	85	<Outlet />
	86	</AppSidebarLayout>
	87	</WebSocketConnectionSurface>
79	88	</AnchoredToastProvider>
80	89	</ToastProvider>
81	90	);
...	...	@@ -428,8 +437,8 @@ function EventRouter() {
428	437	queueMicrotask(flushPendingDomainEvents);
429	438	};
430	439
431		const recoverFromSequenceGap = async (): Promise<void> => {
432		if (!recovery.beginReplayRecovery("sequence-gap")) {
	440	const runReplayRecovery = async (reason: "sequence-gap" \| "resubscribe"): Promise<void> => {
	441	if (!recovery.beginReplayRecovery(reason)) {
433	442	return;
434	443	}
435	444
...	...	@@ -466,7 +475,7 @@ function EventRouter() {
466	475	return;
467	476	}
468	477	}
469		void recoverFromSequenceGap();
	478	void runReplayRecovery(reason);
470	479	} else if (replayCompletion.shouldReplay && import.meta.env.MODE !== "test") {
471	480	console.warn(
472	481	"[orchestration-recovery]",
...	...	@@ -505,7 +514,7 @@ function EventRouter() {
505	514	syncServerReadModel(snapshot);
506	515	reconcileSnapshotDerivedState();
507	516	if (recovery.completeSnapshotRecovery(snapshot.snapshotSequence)) {
508		void recoverFromSequenceGap();
	517	void runReplayRecovery("sequence-gap");
509	518	}
510	519	}
511	520	} catch {
...	...	@@ -522,18 +531,29 @@ function EventRouter() {
522	531	const fallbackToSnapshotRecovery = async (): Promise<void> => {
523	532	await runSnapshotRecovery("replay-failed");
524	533	};
525		const unsubDomainEvent = api.orchestration.onDomainEvent((event) => {
526		const action = recovery.classifyDomainEvent(event.sequence);
527		if (action === "apply") {
528		pendingDomainEvents.push(event);
529		schedulePendingDomainEventFlush();
530		return;
531		}
532		if (action === "recover") {
533		flushPendingDomainEvents();
534		void recoverFromSequenceGap();
535		}
536		});
	534	const unsubDomainEvent = api.orchestration.onDomainEvent(
	535	(event) => {
	536	const action = recovery.classifyDomainEvent(event.sequence);
	537	if (action === "apply") {
	538	pendingDomainEvents.push(event);
	539	schedulePendingDomainEventFlush();
	540	return;
	541	}
	542	if (action === "recover") {
	543	flushPendingDomainEvents();
	544	void runReplayRecovery("sequence-gap");
	545	}
	546	},
	547	{
	548	onResubscribe: () => {
	549	if (disposed) {
	550	return;
	551	}
	552	flushPendingDomainEvents();
	553	void runReplayRecovery("resubscribe");
	554	},
	555	},
	556	);
537	557	const unsubTerminalEvent = api.terminal.onEvent((event) => {
538	558	const thread = useStore.getState().threads.find((entry) => entry.id === event.threadId);
539	559	if (thread && thread.archivedAt !== null) {

apps/web/src/rpc/protocol.ts+82-5

...	...	@@ -1,9 +1,22 @@
1	1	import { WsRpcGroup } from "@t3tools/contracts";
2		import { Effect, Layer } from "effect";
	2	import { Duration, Effect, Layer, Schedule } from "effect";
3	3	import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
4	4	import * as Socket from "effect/unstable/socket/Socket";
5	5
6	6	import { resolveServerUrl } from "../lib/utils";
	7	import {
	8	acknowledgeRpcRequest,
	9	clearAllTrackedRpcRequests,
	10	trackRpcRequestSent,
	11	} from "./requestLatencyState";
	12	import {
	13	getWsReconnectDelayMsForRetry,
	14	recordWsConnectionAttempt,
	15	recordWsConnectionClosed,
	16	recordWsConnectionErrored,
	17	recordWsConnectionOpened,
	18	WS_RECONNECT_MAX_RETRIES,
	19	} from "./wsConnectionState";
7	20
8	21	export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);
9	22
...	...	@@ -17,11 +30,75 @@ export function createWsRpcProtocolLayer(url?: string) {
17	30	protocol: window.location.protocol === "https:" ? "wss" : "ws",
18	31	pathname: "/ws",
19	32	});
	33	const trackingWebSocketConstructorLayer = Layer.succeed(
	34	Socket.WebSocketConstructor,
	35	(socketUrl, protocols) => {
	36	recordWsConnectionAttempt(socketUrl);
	37	const socket = new globalThis.WebSocket(socketUrl, protocols);
	38
	39	socket.addEventListener(
	40	"open",
	41	() => {
	42	recordWsConnectionOpened();
	43	},
	44	{ once: true },
	45	);
	46	socket.addEventListener(
	47	"error",
	48	() => {
	49	clearAllTrackedRpcRequests();
	50	recordWsConnectionErrored("Unable to connect to the T3 server WebSocket.");
	51	},
	52	{ once: true },
	53	);
	54	socket.addEventListener(
	55	"close",
	56	(event) => {
	57	clearAllTrackedRpcRequests();
	58	recordWsConnectionClosed({
	59	code: event.code,
	60	reason: event.reason,
	61	});
	62	},
	63	{ once: true },
	64	);
	65
	66	return socket;
	67	},
	68	);
20	69	const socketLayer = Socket.layerWebSocket(resolvedUrl).pipe(
21		Layer.provide(Socket.layerWebSocketConstructorGlobal),
	70	Layer.provide(trackingWebSocketConstructorLayer),
22	71	);
23
24		return RpcClient.layerProtocolSocket({ retryTransientErrors: true }).pipe(
25		Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)),
	72	const retryPolicy = Schedule.addDelay(Schedule.recurs(WS_RECONNECT_MAX_RETRIES), (retryCount) =>
	73	Effect.succeed(Duration.millis(getWsReconnectDelayMsForRetry(retryCount) ?? 0)),
	74	);
	75	const protocolLayer = Layer.effect(
	76	RpcClient.Protocol,
	77	Effect.map(
	78	RpcClient.makeProtocolSocket({
	79	retryPolicy,
	80	retryTransientErrors: true,
	81	}),
	82	(protocol) => ({
	83	...protocol,
	84	run: (writeResponse) =>
	85	protocol.run((response) => {
	86	if (response._tag === "Chunk" \|\| response._tag === "Exit") {
	87	acknowledgeRpcRequest(response.requestId);
	88	} else if (response._tag === "ClientProtocolError" \|\| response._tag === "Defect") {
	89	clearAllTrackedRpcRequests();
	90	}
	91	return writeResponse(response);
	92	}),
	93	send: (request, transferables) => {
	94	if (request._tag === "Request") {
	95	trackRpcRequestSent(request.id, request.tag);
	96	}
	97	return protocol.send(request, transferables);
	98	},
	99	}),
	100	),
26	101	);
	102
	103	return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));
27	104	}

apps/web/src/rpc/requestLatencyState.test.ts created+65

...	...	@@ -0,0 +1,65 @@
	1	import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
	2
	3	import {
	4	acknowledgeRpcRequest,
	5	getSlowRpcAckRequests,
	6	resetRequestLatencyStateForTests,
	7	trackRpcRequestSent,
	8	SLOW_RPC_ACK_THRESHOLD_MS,
	9	MAX_TRACKED_RPC_ACK_REQUESTS,
	10	} from "./requestLatencyState";
	11
	12	describe("requestLatencyState", () => {
	13	beforeEach(() => {
	14	vi.useFakeTimers();
	15	resetRequestLatencyStateForTests();
	16	});
	17
	18	afterEach(() => {
	19	vi.useRealTimers();
	20	});
	21
	22	it("marks unary requests as slow when the ack threshold is exceeded", () => {
	23	trackRpcRequestSent("1", "server.getConfig");
	24	vi.advanceTimersByTime(SLOW_RPC_ACK_THRESHOLD_MS - 1);
	25	expect(getSlowRpcAckRequests()).toEqual([]);
	26
	27	vi.advanceTimersByTime(1);
	28	expect(getSlowRpcAckRequests()).toMatchObject([
	29	{
	30	requestId: "1",
	31	tag: "server.getConfig",
	32	thresholdMs: SLOW_RPC_ACK_THRESHOLD_MS,
	33	},
	34	]);
	35	});
	36
	37	it("clears the slow request once the server acknowledges it", () => {
	38	trackRpcRequestSent("1", "git.status");
	39	vi.advanceTimersByTime(SLOW_RPC_ACK_THRESHOLD_MS);
	40	expect(getSlowRpcAckRequests()).toHaveLength(1);
	41
	42	acknowledgeRpcRequest("1");
	43	expect(getSlowRpcAckRequests()).toEqual([]);
	44	});
	45
	46	it("ignores long-lived subscribe requests", () => {
	47	trackRpcRequestSent("1", "subscribeServerConfig");
	48	vi.advanceTimersByTime(SLOW_RPC_ACK_THRESHOLD_MS * 2);
	49
	50	expect(getSlowRpcAckRequests()).toEqual([]);
	51	});
	52
	53	it("evicts the oldest pending requests once the tracker reaches capacity", () => {
	54	for (let index = 0; index < MAX_TRACKED_RPC_ACK_REQUESTS + 1; index += 1) {
	55	trackRpcRequestSent(String(index), "server.getConfig");
	56	}
	57
	58	vi.advanceTimersByTime(SLOW_RPC_ACK_THRESHOLD_MS);
	59
	60	const slowRequests = getSlowRpcAckRequests();
	61	expect(slowRequests).toHaveLength(MAX_TRACKED_RPC_ACK_REQUESTS);
	62	expect(slowRequests[0]?.requestId).toBe("1");
	63	expect(slowRequests.at(-1)?.requestId).toBe(String(MAX_TRACKED_RPC_ACK_REQUESTS));
	64	});
	65	});

apps/web/src/rpc/requestLatencyState.ts created+127

...	...	@@ -0,0 +1,127 @@
	1	import { useAtomValue } from "@effect/atom-react";
	2	import { Atom } from "effect/unstable/reactivity";
	3
	4	import { appAtomRegistry } from "./atomRegistry";
	5
	6	export const SLOW_RPC_ACK_THRESHOLD_MS = 2_500;
	7	export const MAX_TRACKED_RPC_ACK_REQUESTS = 256;
	8
	9	export interface SlowRpcAckRequest {
	10	readonly requestId: string;
	11	readonly startedAt: string;
	12	readonly startedAtMs: number;
	13	readonly tag: string;
	14	readonly thresholdMs: number;
	15	}
	16
	17	interface PendingRpcAckRequest {
	18	readonly request: SlowRpcAckRequest;
	19	readonly timeoutId: ReturnType<typeof setTimeout>;
	20	}
	21
	22	const pendingRpcAckRequests = new Map<string, PendingRpcAckRequest>();
	23
	24	const slowRpcAckRequestsAtom = Atom.make<ReadonlyArray<SlowRpcAckRequest>>([]).pipe(
	25	Atom.keepAlive,
	26	Atom.withLabel("slow-rpc-ack-requests"),
	27	);
	28
	29	function setSlowRpcAckRequests(requests: ReadonlyArray<SlowRpcAckRequest>) {
	30	appAtomRegistry.set(slowRpcAckRequestsAtom, [...requests]);
	31	}
	32
	33	function getSlowRpcAckRequestsValue(): ReadonlyArray<SlowRpcAckRequest> {
	34	return appAtomRegistry.get(slowRpcAckRequestsAtom);
	35	}
	36
	37	function shouldTrackRpcAck(tag: string): boolean {
	38	return !tag.startsWith("subscribe");
	39	}
	40
	41	export function getSlowRpcAckRequests(): ReadonlyArray<SlowRpcAckRequest> {
	42	return getSlowRpcAckRequestsValue();
	43	}
	44
	45	export function trackRpcRequestSent(requestId: string, tag: string): void {
	46	if (!shouldTrackRpcAck(tag)) {
	47	return;
	48	}
	49
	50	clearTrackedRpcRequest(requestId);
	51	evictOldestPendingRpcRequestIfNeeded();
	52
	53	const startedAtMs = Date.now();
	54	const request: SlowRpcAckRequest = {
	55	requestId,
	56	startedAt: new Date(startedAtMs).toISOString(),
	57	startedAtMs,
	58	tag,
	59	thresholdMs: SLOW_RPC_ACK_THRESHOLD_MS,
	60	};
	61	const timeoutId = setTimeout(() => {
	62	pendingRpcAckRequests.delete(requestId);
	63	appendSlowRpcAckRequest(request);
	64	}, SLOW_RPC_ACK_THRESHOLD_MS);
	65
	66	pendingRpcAckRequests.set(requestId, {
	67	request,
	68	timeoutId,
	69	});
	70	}
	71
	72	export function acknowledgeRpcRequest(requestId: string): void {
	73	clearTrackedRpcRequest(requestId);
	74	const slowRequests = getSlowRpcAckRequestsValue();
	75	if (!slowRequests.some((request) => request.requestId === requestId)) {
	76	return;
	77	}
	78
	79	setSlowRpcAckRequests(slowRequests.filter((request) => request.requestId !== requestId));
	80	}
	81
	82	export function clearAllTrackedRpcRequests(): void {
	83	for (const pending of pendingRpcAckRequests.values()) {
	84	clearTimeout(pending.timeoutId);
	85	}
	86	pendingRpcAckRequests.clear();
	87	setSlowRpcAckRequests([]);
	88	}
	89
	90	function clearTrackedRpcRequest(requestId: string): void {
	91	const pending = pendingRpcAckRequests.get(requestId);
	92	if (!pending) {
	93	return;
	94	}
	95
	96	clearTimeout(pending.timeoutId);
	97	pendingRpcAckRequests.delete(requestId);
	98	}
	99
	100	function appendSlowRpcAckRequest(request: SlowRpcAckRequest): void {
	101	const requests = [...getSlowRpcAckRequestsValue(), request];
	102	if (requests.length <= MAX_TRACKED_RPC_ACK_REQUESTS) {
	103	setSlowRpcAckRequests(requests);
	104	return;
	105	}
	106
	107	setSlowRpcAckRequests(requests.slice(-MAX_TRACKED_RPC_ACK_REQUESTS));
	108	}
	109
	110	function evictOldestPendingRpcRequestIfNeeded(): void {
	111	while (pendingRpcAckRequests.size >= MAX_TRACKED_RPC_ACK_REQUESTS) {
	112	const oldestRequestId = pendingRpcAckRequests.keys().next().value;
	113	if (oldestRequestId === undefined) {
	114	return;
	115	}
	116
	117	clearTrackedRpcRequest(oldestRequestId);
	118	}
	119	}
	120
	121	export function resetRequestLatencyStateForTests(): void {
	122	clearAllTrackedRpcRequests();
	123	}
	124
	125	export function useSlowRpcAckRequests(): ReadonlyArray<SlowRpcAckRequest> {
	126	return useAtomValue(slowRpcAckRequestsAtom);
	127	}

apps/web/src/rpc/transportError.test.ts created+24

...	...	@@ -0,0 +1,24 @@
	1	import { describe, expect, it } from "vitest";
	2
	3	import { isTransportConnectionErrorMessage, sanitizeThreadErrorMessage } from "./transportError";
	4
	5	describe("transportError", () => {
	6	it("detects websocket transport failures", () => {
	7	expect(isTransportConnectionErrorMessage("SocketCloseError: 1006")).toBe(true);
	8	expect(isTransportConnectionErrorMessage("Unable to connect to the T3 server WebSocket.")).toBe(
	9	true,
	10	);
	11	expect(isTransportConnectionErrorMessage("SocketOpenError: Timeout")).toBe(true);
	12	});
	13
	14	it("preserves non-transport thread errors", () => {
	15	expect(sanitizeThreadErrorMessage("Turn failed")).toBe("Turn failed");
	16	expect(sanitizeThreadErrorMessage("Select a base branch before sending.")).toBe(
	17	"Select a base branch before sending.",
	18	);
	19	});
	20
	21	it("drops transport failures from thread surfaces", () => {
	22	expect(sanitizeThreadErrorMessage("SocketCloseError: 1006")).toBeNull();
	23	});
	24	});

apps/web/src/rpc/transportError.ts created+23

...	...	@@ -0,0 +1,23 @@
	1	const TRANSPORT_ERROR_PATTERNS = [
	2	/\bSocketCloseError\b/i,
	3	/\bSocketOpenError\b/i,
	4	/Unable to connect to the T3 server WebSocket\./i,
	5	/\bping timeout\b/i,
	6	] as const;
	7
	8	export function isTransportConnectionErrorMessage(message: string \| null \| undefined): boolean {
	9	if (typeof message !== "string") {
	10	return false;
	11	}
	12
	13	const normalizedMessage = message.trim();
	14	if (normalizedMessage.length === 0) {
	15	return false;
	16	}
	17
	18	return TRANSPORT_ERROR_PATTERNS.some((pattern) => pattern.test(normalizedMessage));
	19	}
	20
	21	export function sanitizeThreadErrorMessage(message: string \| null \| undefined): string \| null {
	22	return isTransportConnectionErrorMessage(message) ? null : (message ?? null);
	23	}

apps/web/src/rpc/wsConnectionState.test.ts created+112

...	...	@@ -0,0 +1,112 @@
	1	import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
	2
	3	import {
	4	exhaustWsReconnectIfStillWaiting,
	5	getWsConnectionStatus,
	6	getWsReconnectDelayMsForRetry,
	7	getWsConnectionUiState,
	8	recordWsConnectionAttempt,
	9	recordWsConnectionClosed,
	10	recordWsConnectionErrored,
	11	recordWsConnectionOpened,
	12	resetWsConnectionStateForTests,
	13	setBrowserOnlineStatus,
	14	WS_RECONNECT_MAX_ATTEMPTS,
	15	} from "./wsConnectionState";
	16
	17	describe("wsConnectionState", () => {
	18	beforeEach(() => {
	19	vi.useFakeTimers();
	20	vi.setSystemTime(new Date("2026-04-03T20:30:00.000Z"));
	21	resetWsConnectionStateForTests();
	22	});
	23
	24	afterEach(() => {
	25	vi.useRealTimers();
	26	});
	27
	28	it("treats a disconnected browser as offline once the websocket drops", () => {
	29	recordWsConnectionAttempt("ws://localhost:3020/ws");
	30	recordWsConnectionOpened();
	31	recordWsConnectionClosed({ code: 1006, reason: "offline" });
	32	setBrowserOnlineStatus(false);
	33
	34	expect(getWsConnectionUiState(getWsConnectionStatus())).toBe("offline");
	35	});
	36
	37	it("stays in the initial connecting state until the first disconnect", () => {
	38	recordWsConnectionAttempt("ws://localhost:3020/ws");
	39
	40	expect(getWsConnectionStatus()).toMatchObject({
	41	attemptCount: 1,
	42	hasConnected: false,
	43	phase: "connecting",
	44	});
	45	expect(getWsConnectionUiState(getWsConnectionStatus())).toBe("connecting");
	46	});
	47
	48	it("schedules the next retry after a failed websocket attempt", () => {
	49	recordWsConnectionAttempt("ws://localhost:3020/ws");
	50	recordWsConnectionErrored("Unable to connect to the T3 server WebSocket.");
	51
	52	const firstRetryDelayMs = getWsReconnectDelayMsForRetry(0);
	53	if (firstRetryDelayMs === null) {
	54	throw new Error("Expected an initial retry delay.");
	55	}
	56
	57	expect(getWsConnectionStatus()).toMatchObject({
	58	nextRetryAt: new Date(Date.now() + firstRetryDelayMs).toISOString(),
	59	reconnectAttemptCount: 1,
	60	reconnectPhase: "waiting",
	61	});
	62	});
	63
	64	it("marks the reconnect cycle as exhausted after the final attempt fails", () => {
	65	for (let attempt = 0; attempt < WS_RECONNECT_MAX_ATTEMPTS; attempt += 1) {
	66	recordWsConnectionAttempt("ws://localhost:3020/ws");
	67	recordWsConnectionErrored("Unable to connect to the T3 server WebSocket.");
	68	}
	69
	70	expect(getWsConnectionStatus()).toMatchObject({
	71	nextRetryAt: null,
	72	reconnectAttemptCount: WS_RECONNECT_MAX_ATTEMPTS,
	73	reconnectPhase: "exhausted",
	74	});
	75	});
	76
	77	it("can exhaust a stalled final retry window when no new attempt starts", () => {
	78	recordWsConnectionAttempt("ws://localhost:3020/ws");
	79	recordWsConnectionOpened();
	80
	81	for (let attempt = 0; attempt < WS_RECONNECT_MAX_ATTEMPTS - 1; attempt += 1) {
	82	recordWsConnectionAttempt("ws://localhost:3020/ws");
	83	recordWsConnectionErrored("Unable to connect to the T3 server WebSocket.");
	84	}
	85
	86	const finalRetryDelayMs = getWsReconnectDelayMsForRetry(WS_RECONNECT_MAX_ATTEMPTS - 2);
	87	if (finalRetryDelayMs === null) {
	88	throw new Error("Expected a final retry delay.");
	89	}
	90
	91	const statusBeforeExhaust = getWsConnectionStatus();
	92	expect(statusBeforeExhaust).toMatchObject({
	93	nextRetryAt: new Date(Date.now() + finalRetryDelayMs).toISOString(),
	94	reconnectAttemptCount: 7,
	95	reconnectPhase: "waiting",
	96	});
	97
	98	const nextRetryAt = statusBeforeExhaust.nextRetryAt;
	99	if (!nextRetryAt) {
	100	throw new Error("Expected a scheduled retry.");
	101	}
	102
	103	vi.setSystemTime(new Date(Date.now() + finalRetryDelayMs + 1_000));
	104	exhaustWsReconnectIfStillWaiting(nextRetryAt);
	105
	106	expect(getWsConnectionStatus()).toMatchObject({
	107	nextRetryAt: null,
	108	reconnectAttemptCount: WS_RECONNECT_MAX_ATTEMPTS,
	109	reconnectPhase: "exhausted",
	110	});
	111	});
	112	});

apps/web/src/rpc/wsConnectionState.ts created+219

...	...	@@ -0,0 +1,219 @@
	1	import { useAtomValue } from "@effect/atom-react";
	2	import { Atom } from "effect/unstable/reactivity";
	3
	4	import { appAtomRegistry } from "./atomRegistry";
	5
	6	export type WsConnectionUiState = "connected" \| "connecting" \| "error" \| "offline" \| "reconnecting";
	7	export type WsReconnectPhase = "attempting" \| "exhausted" \| "idle" \| "waiting";
	8
	9	export const WS_RECONNECT_INITIAL_DELAY_MS = 1_000;
	10	export const WS_RECONNECT_BACKOFF_FACTOR = 2;
	11	export const WS_RECONNECT_MAX_DELAY_MS = 64_000;
	12	export const WS_RECONNECT_MAX_RETRIES = 7;
	13	export const WS_RECONNECT_MAX_ATTEMPTS = WS_RECONNECT_MAX_RETRIES + 1;
	14
	15	export interface WsConnectionStatus {
	16	readonly attemptCount: number;
	17	readonly closeCode: number \| null;
	18	readonly closeReason: string \| null;
	19	readonly connectedAt: string \| null;
	20	readonly disconnectedAt: string \| null;
	21	readonly hasConnected: boolean;
	22	readonly lastError: string \| null;
	23	readonly lastErrorAt: string \| null;
	24	readonly nextRetryAt: string \| null;
	25	readonly online: boolean;
	26	readonly phase: "idle" \| "connecting" \| "connected" \| "disconnected";
	27	readonly reconnectAttemptCount: number;
	28	readonly reconnectMaxAttempts: number;
	29	readonly reconnectPhase: WsReconnectPhase;
	30	readonly socketUrl: string \| null;
	31	}
	32
	33	const INITIAL_WS_CONNECTION_STATUS = Object.freeze<WsConnectionStatus>({
	34	attemptCount: 0,
	35	closeCode: null,
	36	closeReason: null,
	37	connectedAt: null,
	38	disconnectedAt: null,
	39	hasConnected: false,
	40	lastError: null,
	41	lastErrorAt: null,
	42	nextRetryAt: null,
	43	online: typeof navigator === "undefined" ? true : navigator.onLine !== false,
	44	phase: "idle",
	45	reconnectAttemptCount: 0,
	46	reconnectMaxAttempts: WS_RECONNECT_MAX_ATTEMPTS,
	47	reconnectPhase: "idle",
	48	socketUrl: null,
	49	});
	50
	51	export const wsConnectionStatusAtom = Atom.make(INITIAL_WS_CONNECTION_STATUS).pipe(
	52	Atom.keepAlive,
	53	Atom.withLabel("ws-connection-status"),
	54	);
	55
	56	function isoNow() {
	57	return new Date().toISOString();
	58	}
	59
	60	function updateWsConnectionStatus(
	61	updater: (current: WsConnectionStatus) => WsConnectionStatus,
	62	): WsConnectionStatus {
	63	const nextStatus = updater(getWsConnectionStatus());
	64	appAtomRegistry.set(wsConnectionStatusAtom, nextStatus);
	65	return nextStatus;
	66	}
	67
	68	export function getWsConnectionStatus(): WsConnectionStatus {
	69	return appAtomRegistry.get(wsConnectionStatusAtom);
	70	}
	71
	72	export function getWsConnectionUiState(status: WsConnectionStatus): WsConnectionUiState {
	73	if (status.phase === "connected") {
	74	return "connected";
	75	}
	76
	77	if (!status.online && (status.disconnectedAt !== null \|\| status.phase === "disconnected")) {
	78	return "offline";
	79	}
	80
	81	if (!status.hasConnected) {
	82	return status.phase === "disconnected" ? "error" : "connecting";
	83	}
	84
	85	return "reconnecting";
	86	}
	87
	88	export function recordWsConnectionAttempt(socketUrl: string): WsConnectionStatus {
	89	return updateWsConnectionStatus((current) => ({
	90	...current,
	91	attemptCount: current.attemptCount + 1,
	92	nextRetryAt: null,
	93	phase: "connecting",
	94	reconnectAttemptCount: current.phase === "connected" ? 1 : current.reconnectAttemptCount + 1,
	95	reconnectPhase: "attempting",
	96	socketUrl,
	97	}));
	98	}
	99
	100	export function recordWsConnectionOpened(): WsConnectionStatus {
	101	return updateWsConnectionStatus((current) => ({
	102	...current,
	103	closeCode: null,
	104	closeReason: null,
	105	connectedAt: isoNow(),
	106	disconnectedAt: null,
	107	hasConnected: true,
	108	nextRetryAt: null,
	109	phase: "connected",
	110	reconnectAttemptCount: 0,
	111	reconnectPhase: "idle",
	112	}));
	113	}
	114
	115	export function recordWsConnectionErrored(message?: string \| null): WsConnectionStatus {
	116	return updateWsConnectionStatus((current) =>
	117	applyDisconnectState(current, {
	118	lastError: message?.trim() ? message : current.lastError,
	119	lastErrorAt: isoNow(),
	120	}),
	121	);
	122	}
	123
	124	export function recordWsConnectionClosed(details?: {
	125	readonly code?: number;
	126	readonly reason?: string;
	127	}): WsConnectionStatus {
	128	return updateWsConnectionStatus((current) =>
	129	applyDisconnectState(current, {
	130	closeCode: details?.code ?? current.closeCode,
	131	closeReason: details?.reason?.trim() ? details.reason : current.closeReason,
	132	}),
	133	);
	134	}
	135
	136	export function setBrowserOnlineStatus(online: boolean): WsConnectionStatus {
	137	return updateWsConnectionStatus((current) => ({
	138	...current,
	139	online,
	140	}));
	141	}
	142
	143	export function resetWsReconnectBackoff(): WsConnectionStatus {
	144	return updateWsConnectionStatus((current) => ({
	145	...current,
	146	nextRetryAt: null,
	147	reconnectAttemptCount: 0,
	148	reconnectPhase: "idle",
	149	}));
	150	}
	151
	152	export function exhaustWsReconnectIfStillWaiting(expectedNextRetryAt: string): WsConnectionStatus {
	153	return updateWsConnectionStatus((current) => {
	154	if (
	155	current.reconnectPhase !== "waiting" \|\|
	156	current.nextRetryAt !== expectedNextRetryAt \|\|
	157	!current.online \|\|
	158	!current.hasConnected
	159	) {
	160	return current;
	161	}
	162
	163	return {
	164	...current,
	165	nextRetryAt: null,
	166	reconnectAttemptCount: current.reconnectMaxAttempts,
	167	reconnectPhase: "exhausted",
	168	};
	169	});
	170	}
	171
	172	export function resetWsConnectionStateForTests(): void {
	173	appAtomRegistry.set(wsConnectionStatusAtom, INITIAL_WS_CONNECTION_STATUS);
	174	}
	175
	176	export function useWsConnectionStatus(): WsConnectionStatus {
	177	return useAtomValue(wsConnectionStatusAtom);
	178	}
	179
	180	export function getWsReconnectDelayMsForRetry(retryIndex: number): number \| null {
	181	if (!Number.isInteger(retryIndex) \|\| retryIndex < 0 \|\| retryIndex >= WS_RECONNECT_MAX_RETRIES) {
	182	return null;
	183	}
	184
	185	return Math.min(
	186	Math.round(WS_RECONNECT_INITIAL_DELAY_MS * WS_RECONNECT_BACKOFF_FACTOR ** retryIndex),
	187	WS_RECONNECT_MAX_DELAY_MS,
	188	);
	189	}
	190
	191	function applyDisconnectState(
	192	current: WsConnectionStatus,
	193	updates: Partial<
	194	Pick<WsConnectionStatus, "closeCode" \| "closeReason" \| "lastError" \| "lastErrorAt">
	195	>,
	196	): WsConnectionStatus {
	197	const disconnectedAt = current.disconnectedAt ?? isoNow();
	198	const nextRetryDelayMs =
	199	current.nextRetryAt !== null \|\| current.reconnectPhase === "exhausted"
	200	? null
	201	: getWsReconnectDelayMsForRetry(Math.max(0, current.reconnectAttemptCount - 1));
	202
	203	return {
	204	...current,
	205	...updates,
	206	disconnectedAt,
	207	nextRetryAt:
	208	nextRetryDelayMs === null
	209	? current.nextRetryAt
	210	: new Date(Date.now() + nextRetryDelayMs).toISOString(),
	211	phase: "disconnected",
	212	reconnectPhase:
	213	current.reconnectPhase === "waiting" \|\| current.reconnectPhase === "exhausted"
	214	? current.reconnectPhase
	215	: nextRetryDelayMs === null
	216	? "exhausted"
	217	: "waiting",
	218	};
	219	}

apps/web/src/store.ts+3-2

...	...	@@ -19,6 +19,7 @@ import {
19	19	derivePendingApprovals,
20	20	derivePendingUserInputs,
21	21	} from "./session-logic";
	22	import { sanitizeThreadErrorMessage } from "./rpc/transportError";
22	23	import { type ChatMessage, type Project, type SidebarThreadSummary, type Thread } from "./types";
23	24
24	25	// ── State ────────────────────────────────────────────────────────────
...	...	@@ -165,7 +166,7 @@ function mapThread(thread: OrchestrationThread): Thread {
165	166	session: thread.session ? mapSession(thread.session) : null,
166	167	messages: thread.messages.map(mapMessage),
167	168	proposedPlans: thread.proposedPlans.map(mapProposedPlan),
168		error: thread.session?.lastError ?? null,
	169	error: sanitizeThreadErrorMessage(thread.session?.lastError),
169	170	createdAt: thread.createdAt,
170	171	archivedAt: thread.archivedAt,
171	172	updatedAt: thread.updatedAt,
...	...	@@ -894,7 +895,7 @@ export function applyOrchestrationEvent(state: AppState, event: OrchestrationEve
894	895	return updateThreadState(state, event.payload.threadId, (thread) => ({
895	896	...thread,
896	897	session: mapSession(event.payload.session),
897		error: event.payload.session.lastError ?? null,
	898	error: sanitizeThreadErrorMessage(event.payload.session.lastError),
898	899	latestTurn:
899	900	event.payload.session.status === "running" && event.payload.session.activeTurnId !== null
900	901	? buildLatestTurn({

apps/web/src/wsNativeApi.test.ts+14

...	...	@@ -243,6 +243,20 @@ describe("wsNativeApi", () => {
243	243	expect(onDomainEvent).toHaveBeenCalledWith(orchestrationEvent);
244	244	});
245	245
	246	it("forwards orchestration stream subscription options to the RPC client", async () => {
	247	const { createWsNativeApi } = await import("./wsNativeApi");
	248
	249	const api = createWsNativeApi();
	250	const onDomainEvent = vi.fn();
	251	const onResubscribe = vi.fn();
	252
	253	api.orchestration.onDomainEvent(onDomainEvent, { onResubscribe });
	254
	255	expect(rpcClientMock.orchestration.onDomainEvent).toHaveBeenCalledWith(onDomainEvent, {
	256	onResubscribe,
	257	});
	258	});
	259
246	260	it("sends orchestration dispatch commands as the direct RPC payload", async () => {
247	261	rpcClientMock.orchestration.dispatchCommand.mockResolvedValue({ sequence: 1 });
248	262	const { createWsNativeApi } = await import("./wsNativeApi");

apps/web/src/wsNativeApi.ts+6-1

...	...	@@ -1,7 +1,9 @@
1	1	import { type ContextMenuItem, type NativeApi } from "@t3tools/contracts";
2	2
3	3	import { showContextMenuFallback } from "./contextMenuFallback";
	4	import { resetRequestLatencyStateForTests } from "./rpc/requestLatencyState";
4	5	import { resetServerStateForTests } from "./rpc/serverState";
	6	import { resetWsConnectionStateForTests } from "./rpc/wsConnectionState";
5	7	import { __resetWsRpcClientForTests, getWsRpcClient } from "./wsRpcClient";
6	8
7	9	let instance: { api: NativeApi } \| null = null;
...	...	@@ -9,7 +11,9 @@ let instance: { api: NativeApi } \| null = null;
9	11	export function __resetWsNativeApiForTests() {
10	12	instance = null;
11	13	__resetWsRpcClientForTests();
	14	resetRequestLatencyStateForTests();
12	15	resetServerStateForTests();
	16	resetWsConnectionStateForTests();
13	17	}
14	18
15	19	export function createWsNativeApi(): NativeApi {
...	...	@@ -98,7 +102,8 @@ export function createWsNativeApi(): NativeApi {
98	102	rpcClient.orchestration
99	103	.replayEvents({ fromSequenceExclusive })
100	104	.then((events) => [...events]),
101		onDomainEvent: (callback) => rpcClient.orchestration.onDomainEvent(callback),
	105	onDomainEvent: (callback, options) =>
	106	rpcClient.orchestration.onDomainEvent(callback, options),
102	107	},
103	108	};
104	109

apps/web/src/wsRpcClient.ts+31-8

...	...	@@ -10,12 +10,17 @@ import {
10	10	import { Effect, Stream } from "effect";
11	11
12	12	import { type WsRpcProtocolClient } from "./rpc/protocol";
	13	import { resetWsReconnectBackoff } from "./rpc/wsConnectionState";
13	14	import { WsTransport } from "./wsTransport";
14	15
15	16	type RpcTag = keyof WsRpcProtocolClient & string;
16	17	type RpcMethod<TTag extends RpcTag> = WsRpcProtocolClient[TTag];
17	18	type RpcInput<TTag extends RpcTag> = Parameters<RpcMethod<TTag>>[0];
18	19
	20	interface StreamSubscriptionOptions {
	21	readonly onResubscribe?: () => void;
	22	}
	23
19	24	type RpcUnaryMethod<TTag extends RpcTag> =
20	25	RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<infer TSuccess, any, any>
21	26	? (input: RpcInput<TTag>) => Promise<TSuccess>
...	...	@@ -28,7 +33,7 @@ type RpcUnaryNoArgMethod<TTag extends RpcTag> =
28	33
29	34	type RpcStreamMethod<TTag extends RpcTag> =
30	35	RpcMethod<TTag> extends (input: any, options?: any) => Stream.Stream<infer TEvent, any, any>
31		? (listener: (event: TEvent) => void) => () => void
	36	? (listener: (event: TEvent) => void, options?: StreamSubscriptionOptions) => () => void
32	37	: never;
33	38
34	39	interface GitRunStackedActionOptions {
...	...	@@ -37,6 +42,7 @@ interface GitRunStackedActionOptions {
37	42
38	43	export interface WsRpcClient {
39	44	readonly dispose: () => Promise<void>;
	45	readonly reconnect: () => Promise<void>;
40	46	readonly terminal: {
41	47	readonly open: RpcUnaryMethod<typeof WS_METHODS.terminalOpen>;
42	48	readonly write: RpcUnaryMethod<typeof WS_METHODS.terminalWrite>;
...	...	@@ -113,6 +119,10 @@ export async function __resetWsRpcClientForTests() {
113	119	export function createWsRpcClient(transport = new WsTransport()): WsRpcClient {
114	120	return {
115	121	dispose: () => transport.dispose(),
	122	reconnect: async () => {
	123	resetWsReconnectBackoff();
	124	await transport.reconnect();
	125	},
116	126	terminal: {
117	127	open: (input) => transport.request((client) => client[WS_METHODS.terminalOpen](input)),
118	128	write: (input) => transport.request((client) => client[WS_METHODS.terminalWrite](input)),
...	...	@@ -120,8 +130,12 @@ export function createWsRpcClient(transport = new WsTransport()): WsRpcClient {
120	130	clear: (input) => transport.request((client) => client[WS_METHODS.terminalClear](input)),
121	131	restart: (input) => transport.request((client) => client[WS_METHODS.terminalRestart](input)),
122	132	close: (input) => transport.request((client) => client[WS_METHODS.terminalClose](input)),
123		onEvent: (listener) =>
124		transport.subscribe((client) => client[WS_METHODS.subscribeTerminalEvents]({}), listener),
	133	onEvent: (listener, options) =>
	134	transport.subscribe(
	135	(client) => client[WS_METHODS.subscribeTerminalEvents]({}),
	136	listener,
	137	options,
	138	),
125	139	},
126	140	projects: {
127	141	searchEntries: (input) =>
...	...	@@ -179,10 +193,18 @@ export function createWsRpcClient(transport = new WsTransport()): WsRpcClient {
179	193	getSettings: () => transport.request((client) => client[WS_METHODS.serverGetSettings]({})),
180	194	updateSettings: (patch) =>
181	195	transport.request((client) => client[WS_METHODS.serverUpdateSettings]({ patch })),
182		subscribeConfig: (listener) =>
183		transport.subscribe((client) => client[WS_METHODS.subscribeServerConfig]({}), listener),
184		subscribeLifecycle: (listener) =>
185		transport.subscribe((client) => client[WS_METHODS.subscribeServerLifecycle]({}), listener),
	196	subscribeConfig: (listener, options) =>
	197	transport.subscribe(
	198	(client) => client[WS_METHODS.subscribeServerConfig]({}),
	199	listener,
	200	options,
	201	),
	202	subscribeLifecycle: (listener, options) =>
	203	transport.subscribe(
	204	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
	205	listener,
	206	options,
	207	),
186	208	},
187	209	orchestration: {
188	210	getSnapshot: () =>
...	...	@@ -197,10 +219,11 @@ export function createWsRpcClient(transport = new WsTransport()): WsRpcClient {
197	219	transport
198	220	.request((client) => client[ORCHESTRATION_WS_METHODS.replayEvents](input))
199	221	.then((events) => [...events]),
200		onDomainEvent: (listener) =>
	222	onDomainEvent: (listener, options) =>
201	223	transport.subscribe(
202	224	(client) => client[WS_METHODS.subscribeOrchestrationDomainEvents]({}),
203	225	listener,
	226	options,
204	227	),
205	228	},
206	229	};

apps/web/src/wsTransport.test.ts+261-5

...	...	@@ -5,6 +5,12 @@ import {
5	5	__resetClientTracingForTests,
6	6	configureClientTracing,
7	7	} from "./observability/clientTracing";
	8	import { getSlowRpcAckRequests, resetRequestLatencyStateForTests } from "./rpc/requestLatencyState";
	9	import {
	10	getWsConnectionStatus,
	11	getWsConnectionUiState,
	12	resetWsConnectionStateForTests,
	13	} from "./rpc/wsConnectionState";
8	14	import { WsTransport } from "./wsTransport";
9	15
10	16	type WsEventType = "open" \| "message" \| "close" \| "error";
...	...	@@ -57,6 +63,10 @@ class MockWebSocket {
57	63	this.emit("message", { data, type: "message" });
58	64	}
59	65
	66	error() {
	67	this.emit("error", { type: "error" });
	68	}
	69
60	70	private emit(type: WsEventType, event?: WsEvent) {
61	71	const listeners = this.listeners.get(type);
62	72	if (!listeners) return;
...	...	@@ -93,7 +103,10 @@ async function waitFor(assertion: () => void, timeoutMs = 1_000): Promise<void>
93	103	}
94	104
95	105	beforeEach(() => {
	106	vi.useRealTimers();
96	107	sockets.length = 0;
	108	resetRequestLatencyStateForTests();
	109	resetWsConnectionStateForTests();
97	110
98	111	Object.defineProperty(globalThis, "window", {
99	112	configurable: true,
...	...	@@ -107,14 +120,20 @@ beforeEach(() => {
107	120	desktopBridge: undefined,
108	121	},
109	122	});
	123	Object.defineProperty(globalThis, "navigator", {
	124	configurable: true,
	125	value: { onLine: true },
	126	});
110	127
111	128	globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket;
112	129	});
113	130
114		afterEach(() => {
	131	afterEach(async () => {
115	132	globalThis.WebSocket = originalWebSocket;
116	133	globalThis.fetch = originalFetch;
117		void __resetClientTracingForTests();
	134	resetRequestLatencyStateForTests();
	135	resetWsConnectionStateForTests();
	136	await __resetClientTracingForTests();
118	137	vi.restoreAllMocks();
119	138	});
120	139
...	...	@@ -148,6 +167,184 @@ describe("WsTransport", () => {
148	167	await transport.dispose();
149	168	});
150	169
	170	it("tracks initial connection failures for the app error state", async () => {
	171	const transport = new WsTransport("ws://localhost:3020");
	172
	173	await waitFor(() => {
	174	expect(sockets).toHaveLength(1);
	175	});
	176
	177	const socket = getSocket();
	178	expect(getWsConnectionStatus()).toMatchObject({
	179	attemptCount: 1,
	180	phase: "connecting",
	181	socketUrl: "ws://localhost:3020/ws",
	182	});
	183
	184	socket.error();
	185	socket.close(1006, "server unavailable");
	186
	187	await waitFor(() => {
	188	expect(getWsConnectionStatus()).toMatchObject({
	189	closeCode: 1006,
	190	closeReason: "server unavailable",
	191	hasConnected: false,
	192	lastError: "Unable to connect to the T3 server WebSocket.",
	193	phase: "disconnected",
	194	});
	195	});
	196	expect(getWsConnectionUiState(getWsConnectionStatus())).toBe("error");
	197
	198	await transport.dispose();
	199	});
	200
	201	it("surfaces reconnecting state after a live socket disconnects", async () => {
	202	const transport = new WsTransport("ws://localhost:3020");
	203
	204	await waitFor(() => {
	205	expect(sockets).toHaveLength(1);
	206	});
	207
	208	const socket = getSocket();
	209	socket.open();
	210
	211	await waitFor(() => {
	212	expect(getWsConnectionStatus()).toMatchObject({
	213	hasConnected: true,
	214	phase: "connected",
	215	});
	216	});
	217
	218	socket.close(1013, "try again later");
	219
	220	await waitFor(() => {
	221	expect(getWsConnectionStatus()).toMatchObject({
	222	closeReason: "try again later",
	223	hasConnected: true,
	224	});
	225	});
	226	expect(getWsConnectionUiState(getWsConnectionStatus())).toBe("reconnecting");
	227
	228	await transport.dispose();
	229	});
	230
	231	it("reconnects the websocket session without disposing the transport", async () => {
	232	const transport = new WsTransport("ws://localhost:3020");
	233
	234	await waitFor(() => {
	235	expect(sockets).toHaveLength(1);
	236	});
	237
	238	const firstSocket = getSocket();
	239	firstSocket.open();
	240
	241	await waitFor(() => {
	242	expect(getWsConnectionStatus()).toMatchObject({
	243	hasConnected: true,
	244	phase: "connected",
	245	});
	246	});
	247
	248	await transport.reconnect();
	249
	250	await waitFor(() => {
	251	expect(sockets).toHaveLength(2);
	252	});
	253
	254	const secondSocket = getSocket();
	255	expect(secondSocket).not.toBe(firstSocket);
	256	expect(firstSocket.readyState).toBe(MockWebSocket.CLOSED);
	257
	258	const requestPromise = transport.request((client) =>
	259	client[WS_METHODS.serverUpsertKeybinding]({
	260	command: "terminal.toggle",
	261	key: "ctrl+k",
	262	}),
	263	);
	264
	265	secondSocket.open();
	266
	267	await waitFor(() => {
	268	expect(secondSocket.sent).toHaveLength(1);
	269	});
	270
	271	const requestMessage = JSON.parse(secondSocket.sent[0] ?? "{}") as { id: string };
	272	secondSocket.serverMessage(
	273	JSON.stringify({
	274	_tag: "Exit",
	275	requestId: requestMessage.id,
	276	exit: {
	277	_tag: "Success",
	278	value: {
	279	keybindings: [],
	280	issues: [],
	281	},
	282	},
	283	}),
	284	);
	285
	286	await expect(requestPromise).resolves.toEqual({
	287	keybindings: [],
	288	issues: [],
	289	});
	290
	291	await transport.dispose();
	292	});
	293
	294	it("marks unary requests as slow until the first server ack arrives", async () => {
	295	const transport = new WsTransport("ws://localhost:3020");
	296
	297	const requestPromise = transport.request((client) =>
	298	client[WS_METHODS.serverUpsertKeybinding]({
	299	command: "terminal.toggle",
	300	key: "ctrl+k",
	301	}),
	302	);
	303
	304	await waitFor(() => {
	305	expect(sockets).toHaveLength(1);
	306	});
	307
	308	const socket = getSocket();
	309	socket.open();
	310
	311	await waitFor(() => {
	312	expect(socket.sent).toHaveLength(1);
	313	});
	314
	315	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
	316	await waitFor(() => {
	317	expect(getSlowRpcAckRequests()).toMatchObject([
	318	{
	319	requestId: requestMessage.id,
	320	tag: WS_METHODS.serverUpsertKeybinding,
	321	},
	322	]);
	323	}, 5_000);
	324
	325	socket.serverMessage(
	326	JSON.stringify({
	327	_tag: "Exit",
	328	requestId: requestMessage.id,
	329	exit: {
	330	_tag: "Success",
	331	value: {
	332	keybindings: [],
	333	issues: [],
	334	},
	335	},
	336	}),
	337	);
	338
	339	await expect(requestPromise).resolves.toEqual({
	340	keybindings: [],
	341	issues: [],
	342	});
	343	expect(getSlowRpcAckRequests()).toEqual([]);
	344
	345	await transport.dispose();
	346	}, 10_000);
	347
151	348	it("sends unary RPC requests and resolves successful exits", async () => {
152	349	const transport = new WsTransport("ws://localhost:3020");
153	350
...	...	@@ -257,10 +454,12 @@ describe("WsTransport", () => {
257	454	it("re-subscribes stream listeners after the stream exits", async () => {
258	455	const transport = new WsTransport("ws://localhost:3020");
259	456	const listener = vi.fn();
	457	const onResubscribe = vi.fn();
260	458
261	459	const unsubscribe = transport.subscribe(
262	460	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
263	461	listener,
	462	{ onResubscribe },
264	463	);
265	464	await waitFor(() => {
266	465	expect(sockets).toHaveLength(1);
...	...	@@ -308,6 +507,7 @@ describe("WsTransport", () => {
308	507	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
309	508	expect(nextRequest).toBeDefined();
310	509	});
	510	expect(onResubscribe).toHaveBeenCalledOnce();
311	511
312	512	const secondRequest = socket.sent
313	513	.map((message) => JSON.parse(message) as { _tag?: string; id?: string; tag?: string })
...	...	@@ -346,6 +546,52 @@ describe("WsTransport", () => {
346	546	await transport.dispose();
347	547	});
348	548
	549	it("does not fire onResubscribe when the first stream attempt exits before any value", async () => {
	550	const transport = new WsTransport("ws://localhost:3020");
	551	const listener = vi.fn();
	552	const onResubscribe = vi.fn();
	553
	554	const unsubscribe = transport.subscribe(
	555	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
	556	listener,
	557	{ onResubscribe },
	558	);
	559	await waitFor(() => {
	560	expect(sockets).toHaveLength(1);
	561	});
	562
	563	const socket = getSocket();
	564	socket.open();
	565
	566	await waitFor(() => {
	567	expect(socket.sent).toHaveLength(1);
	568	});
	569
	570	const firstRequest = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
	571	socket.serverMessage(
	572	JSON.stringify({
	573	_tag: "Exit",
	574	requestId: firstRequest.id,
	575	exit: {
	576	_tag: "Success",
	577	value: null,
	578	},
	579	}),
	580	);
	581
	582	await waitFor(() => {
	583	const nextRequest = socket.sent
	584	.map((message) => JSON.parse(message) as { _tag?: string; id?: string })
	585	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
	586	expect(nextRequest).toBeDefined();
	587	});
	588	expect(onResubscribe).not.toHaveBeenCalled();
	589	expect(listener).not.toHaveBeenCalled();
	590
	591	unsubscribe();
	592	await transport.dispose();
	593	});
	594
349	595	it("streams finite request events without re-subscribing", async () => {
350	596	const transport = new WsTransport("ws://localhost:3020");
351	597	const listener = vi.fn();
...	...	@@ -429,11 +675,21 @@ describe("WsTransport", () => {
429	675	};
430	676	const transport = {
431	677	disposed: false,
432		clientScope: {} as never,
433		runtime,
	678	session: {
	679	clientScope: {} as never,
	680	runtime,
	681	},
	682	closeSession: (
	683	WsTransport.prototype as unknown as {
	684	closeSession: (session: {
	685	clientScope: unknown;
	686	runtime: { dispose: () => Promise<void>; runPromise: () => Promise<void> };
	687	}) => Promise<void>;
	688	}
	689	).closeSession,
434	690	} as unknown as WsTransport;
435	691
436		WsTransport.prototype.dispose.call(transport);
	692	void WsTransport.prototype.dispose.call(transport);
437	693
438	694	expect(runtime.runPromise).toHaveBeenCalledTimes(1);
439	695	expect(runtime.dispose).not.toHaveBeenCalled();

apps/web/src/wsTransport.ts+170-44

...	...	@@ -1,15 +1,26 @@
1		import { Duration, Effect, Exit, Layer, ManagedRuntime, Option, Scope, Stream } from "effect";
	1	import {
	2	Cause,
	3	Duration,
	4	Effect,
	5	Exit,
	6	Layer,
	7	ManagedRuntime,
	8	Option,
	9	Scope,
	10	Stream,
	11	} from "effect";
	12	import { RpcClient } from "effect/unstable/rpc";
2	13
	14	import { ClientTracingLive, configureClientTracing } from "./observability/clientTracing";
3	15	import {
4	16	createWsRpcProtocolLayer,
5	17	makeWsRpcProtocolClient,
6	18	type WsRpcProtocolClient,
7	19	} from "./rpc/protocol";
8		import { RpcClient } from "effect/unstable/rpc";
9		import { ClientTracingLive, configureClientTracing } from "./observability/clientTracing";
10	20
11	21	interface SubscribeOptions {
12	22	readonly retryDelay?: Duration.Input;
	23	readonly onResubscribe?: () => void;
13	24	}
14	25
15	26	interface RequestOptions {
...	...	@@ -17,6 +28,13 @@ interface RequestOptions {
17	28	}
18	29
19	30	const DEFAULT_SUBSCRIPTION_RETRY_DELAY_MS = Duration.millis(250);
	31	const NOOP: () => void = () => undefined;
	32
	33	interface TransportSession {
	34	readonly clientPromise: Promise<WsRpcProtocolClient>;
	35	readonly clientScope: Scope.Closeable;
	36	readonly runtime: ManagedRuntime.ManagedRuntime<RpcClient.Protocol, never>;
	37	}
20	38
21	39	function formatErrorMessage(error: unknown): string {
22	40	if (error instanceof Error && error.message.trim().length > 0) {
...	...	@@ -26,21 +44,16 @@ function formatErrorMessage(error: unknown): string {
26	44	}
27	45
28	46	export class WsTransport {
29		private readonly runtime: ManagedRuntime.ManagedRuntime<RpcClient.Protocol, never>;
30		private readonly clientScope: Scope.Closeable;
31		private readonly clientPromise: Promise<WsRpcProtocolClient>;
32	47	private readonly tracingReady: Promise<void>;
	48	private readonly url: string \| undefined;
33	49	private disposed = false;
	50	private reconnectChain: Promise<void> = Promise.resolve();
	51	private session: TransportSession;
34	52
35	53	constructor(url?: string) {
	54	this.url = url;
36	55	this.tracingReady = configureClientTracing();
37		this.runtime = ManagedRuntime.make(
38		Layer.mergeAll(createWsRpcProtocolLayer(url), ClientTracingLive),
39		);
40		this.clientScope = this.runtime.runSync(Scope.make());
41		this.clientPromise = this.runtime.runPromise(
42		Scope.provide(this.clientScope)(makeWsRpcProtocolClient),
43		);
	56	this.session = this.createSession();
44	57	}
45	58
46	59	async request<TSuccess>(
...	...	@@ -52,8 +65,9 @@ export class WsTransport {
52	65	}
53	66
54	67	await this.tracingReady;
55		const client = await this.clientPromise;
56		return await this.runtime.runPromise(Effect.suspend(() => execute(client)));
	68	const session = this.session;
	69	const client = await session.clientPromise;
	70	return await session.runtime.runPromise(Effect.suspend(() => execute(client)));
57	71	}
58	72
59	73	async requestStream<TValue>(
...	...	@@ -65,8 +79,9 @@ export class WsTransport {
65	79	}
66	80
67	81	await this.tracingReady;
68		const client = await this.clientPromise;
69		await this.runtime.runPromise(
	82	const session = this.session;
	83	const client = await session.clientPromise;
	84	await session.runtime.runPromise(
70	85	Stream.runForEach(connect(client), (value) =>
71	86	Effect.sync(() => {
72	87	try {
...	...	@@ -89,16 +104,132 @@ export class WsTransport {
89	104	}
90	105
91	106	let active = true;
92		const retryDelayMs = options?.retryDelay ?? DEFAULT_SUBSCRIPTION_RETRY_DELAY_MS;
93		const cancel = this.runtime.runCallback(
	107	let hasReceivedValue = false;
	108	const retryDelayMs = Duration.toMillis(
	109	Duration.fromInputUnsafe(options?.retryDelay ?? DEFAULT_SUBSCRIPTION_RETRY_DELAY_MS),
	110	);
	111	let cancelCurrentStream: () => void = NOOP;
	112
	113	void (async () => {
	114	for (;;) {
	115	if (!active \|\| this.disposed) {
	116	return;
	117	}
	118
	119	try {
	120	if (hasReceivedValue) {
	121	try {
	122	options?.onResubscribe?.();
	123	} catch {
	124	// Swallow reconnect hook errors so the stream can recover.
	125	}
	126	}
	127
	128	const session = this.session;
	129	const runningStream = this.runStreamOnSession(
	130	session,
	131	connect,
	132	listener,
	133	() => active,
	134	() => {
	135	hasReceivedValue = true;
	136	},
	137	);
	138	cancelCurrentStream = runningStream.cancel;
	139	await runningStream.completed;
	140	cancelCurrentStream = NOOP;
	141	} catch (error) {
	142	cancelCurrentStream = NOOP;
	143	if (!active \|\| this.disposed) {
	144	return;
	145	}
	146
	147	console.warn("WebSocket RPC subscription disconnected", {
	148	error: formatErrorMessage(error),
	149	});
	150	await sleep(retryDelayMs);
	151	}
	152	}
	153	})();
	154
	155	return () => {
	156	active = false;
	157	cancelCurrentStream();
	158	};
	159	}
	160
	161	async reconnect() {
	162	if (this.disposed) {
	163	throw new Error("Transport disposed");
	164	}
	165
	166	const reconnectOperation = this.reconnectChain.then(async () => {
	167	if (this.disposed) {
	168	throw new Error("Transport disposed");
	169	}
	170
	171	const previousSession = this.session;
	172	this.session = this.createSession();
	173	await this.closeSession(previousSession);
	174	});
	175
	176	this.reconnectChain = reconnectOperation.catch(() => undefined);
	177	await reconnectOperation;
	178	}
	179
	180	async dispose() {
	181	if (this.disposed) {
	182	return;
	183	}
	184	this.disposed = true;
	185	await this.closeSession(this.session);
	186	}
	187
	188	private closeSession(session: TransportSession) {
	189	return session.runtime.runPromise(Scope.close(session.clientScope, Exit.void)).finally(() => {
	190	session.runtime.dispose();
	191	});
	192	}
	193
	194	private createSession(): TransportSession {
	195	const runtime = ManagedRuntime.make(
	196	Layer.mergeAll(createWsRpcProtocolLayer(this.url), ClientTracingLive),
	197	);
	198	const clientScope = runtime.runSync(Scope.make());
	199	return {
	200	runtime,
	201	clientScope,
	202	clientPromise: runtime.runPromise(Scope.provide(clientScope)(makeWsRpcProtocolClient)),
	203	};
	204	}
	205
	206	private runStreamOnSession<TValue>(
	207	session: TransportSession,
	208	connect: (client: WsRpcProtocolClient) => Stream.Stream<TValue, Error, never>,
	209	listener: (value: TValue) => void,
	210	isActive: () => boolean,
	211	markValueReceived: () => void,
	212	): {
	213	readonly cancel: () => void;
	214	readonly completed: Promise<void>;
	215	} {
	216	let resolveCompleted!: () => void;
	217	let rejectCompleted!: (error: unknown) => void;
	218	const completed = new Promise<void>((resolve, reject) => {
	219	resolveCompleted = resolve;
	220	rejectCompleted = reject;
	221	});
	222	const cancel = session.runtime.runCallback(
94	223	Effect.promise(() => this.tracingReady).pipe(
95		Effect.flatMap(() => Effect.promise(() => this.clientPromise)),
	224	Effect.flatMap(() => Effect.promise(() => session.clientPromise)),
96	225	Effect.flatMap((client) =>
97	226	Stream.runForEach(connect(client), (value) =>
98	227	Effect.sync(() => {
99		if (!active) {
	228	if (!isActive()) {
100	229	return;
101	230	}
	231
	232	markValueReceived();
102	233	try {
103	234	listener(value);
104	235	} catch {
...	...	@@ -107,33 +238,28 @@ export class WsTransport {
107	238	}),
108	239	),
109	240	),
110		Effect.catch((error) => {
111		if (!active \|\| this.disposed) {
112		return Effect.interrupt;
113		}
114		return Effect.sync(() => {
115		console.warn("WebSocket RPC subscription disconnected", {
116		error: formatErrorMessage(error),
117		});
118		}).pipe(Effect.andThen(Effect.sleep(retryDelayMs)));
119		}),
120		Effect.forever,
121	241	),
	242	{
	243	onExit: (exit) => {
	244	if (Exit.isSuccess(exit)) {
	245	resolveCompleted();
	246	return;
	247	}
	248
	249	rejectCompleted(Cause.squash(exit.cause));
	250	},
	251	},
122	252	);
123	253
124		return () => {
125		active = false;
126		cancel();
	254	return {
	255	cancel,
	256	completed,
127	257	};
128	258	}
	259	}
129	260
130		async dispose() {
131		if (this.disposed) {
132		return;
133		}
134		this.disposed = true;
135		await this.runtime.runPromise(Scope.close(this.clientScope, Exit.void)).finally(() => {
136		this.runtime.dispose();
137		});
138		}
	261	function sleep(ms: number): Promise<void> {
	262	return new Promise((resolve) => {
	263	setTimeout(resolve, ms);
	264	});
139	265	}

packages/contracts/src/ipc.ts+6-1

...	...	@@ -180,6 +180,11 @@ export interface NativeApi {
180	180	input: OrchestrationGetFullThreadDiffInput,
181	181	) => Promise<OrchestrationGetFullThreadDiffResult>;
182	182	replayEvents: (fromSequenceExclusive: number) => Promise<OrchestrationEvent[]>;
183		onDomainEvent: (callback: (event: OrchestrationEvent) => void) => () => void;
	183	onDomainEvent: (
	184	callback: (event: OrchestrationEvent) => void,
	185	options?: {
	186	onResubscribe?: () => void;
	187	},
	188	) => () => void;
184	189	};
185	190	}