lael-labs/t3code

apps/web/src/components/settings/ConnectionsSettings.tsx+1−1

   type DesktopServerExposureState,
   type EnvironmentId,
 } from "@t3tools/contracts";
+import { WsRpcClient } from "@t3tools/client-runtime";
 import { DateTime } from "effect";
 import { useCopyToClipboard } from "../../hooks/useCopyToClipboard";
   type ServerClientSessionRecord,
   type ServerPairingLinkRecord,
 } from "~/environments/primary";
-import type { WsRpcClient } from "~/rpc/wsRpcClient";
 import {
   type SavedEnvironmentRecord,
   type SavedEnvironmentRuntimeState,

apps/web/src/environmentApi.ts+1−1

@@ -1,6 +1,6 @@
1	import type { EnvironmentId, EnvironmentApi } from "@t3tools/contracts";	1	import type { EnvironmentId, EnvironmentApi } from "@t3tools/contracts";
2		2
3	import type { WsRpcClient } from "./rpc/wsRpcClient";	3	import type { WsRpcClient } from "@t3tools/client-runtime";
4	import { readEnvironmentConnection } from "./environments/runtime";	4	import { readEnvironmentConnection } from "./environments/runtime";
5		5
6	const environmentApiOverridesForTests = new Map<EnvironmentId, EnvironmentApi>();	6	const environmentApiOverridesForTests = new Map<EnvironmentId, EnvironmentApi>();

apps/web/src/environments/runtime/connection.test.ts+1−1

@@ -2,7 +2,7 @@ import { EnvironmentId } from "@t3tools/contracts"; · 1 unmodified line
2	import { describe, expect, it, vi } from "vitest";	2	import { describe, expect, it, vi } from "vitest";
3		3
4	import { createEnvironmentConnection } from "./connection";	4	import { createEnvironmentConnection } from "./connection";
5	import type { WsRpcClient } from "~/rpc/wsRpcClient";	5	import type { WsRpcClient } from "@t3tools/client-runtime";
6		6
7	function createTestClient() {	7	function createTestClient() {
8	const lifecycleListeners = new Set<(event: any) => void>();	8	const lifecycleListeners = new Set<(event: any) => void>();

apps/web/src/environments/runtime/service.ts+10−2

   type ServerConfig,
   ThreadId,
 } from "@t3tools/contracts";
+import { createWsRpcClient as createBaseWsRpcClient, WsRpcClient } from "@t3tools/client-runtime";
 import { type QueryClient } from "@tanstack/react-query";
 import { Throttler } from "@tanstack/react-pacer";
 import {
 } from "~/store";
 import { useTerminalUiStateStore } from "~/terminalUiStateStore";
 import { useUiStateStore } from "~/uiStateStore";
-import { WsTransport } from "../../rpc/wsTransport";
+import { WsTransport } from "~/rpc/wsTransport";
-import { createWsRpcClient, type WsRpcClient } from "../../rpc/wsRpcClient";
 import { derivePhysicalProjectKey } from "../../logicalProject";
 import { subscribeTerminalMetadata, terminalSessionManager } from "../../terminalSessionState";
+import { resetWsReconnectBackoff } from "~/rpc/wsConnectionState";
 type EnvironmentServiceState = {
   readonly queryClient: QueryClient;
   };
 }
+function createWsRpcClient(transport: WsTransport): WsRpcClient {
+  return createBaseWsRpcClient(transport, {
+    beforeReconnect: () => resetWsReconnectBackoff(),
+  });
+}
 function createPrimaryEnvironmentClient(
   knownEnvironment: ReturnType<typeof getPrimaryKnownEnvironment>,
 ) {

apps/web/src/localApi.ts+2−1

@@ -1,4 +1,5 @@
1	import type { ContextMenuItem, LocalApi } from "@t3tools/contracts";	1	import type { ContextMenuItem, LocalApi } from "@t3tools/contracts";
		2	import type { WsRpcClient } from "@t3tools/client-runtime";
2		3
3	import { resetGitStatusStateForTests } from "./lib/gitStatusState";	4	import { resetGitStatusStateForTests } from "./lib/gitStatusState";
4	import { resetRequestLatencyStateForTests } from "./rpc/requestLatencyState";	5	import { resetRequestLatencyStateForTests } from "./rpc/requestLatencyState";
@@ -12,7 +13,7 @@ import { · 7 unmodified lines
12	getPrimaryEnvironmentConnection,	13	getPrimaryEnvironmentConnection,
13	resetEnvironmentServiceForTests,	14	resetEnvironmentServiceForTests,
14	} from "./environments/runtime";	15	} from "./environments/runtime";
15	import { type WsRpcClient } from "./rpc/wsRpcClient";	16
16	import { showContextMenuFallback } from "./contextMenuFallback";	17	import { showContextMenuFallback } from "./contextMenuFallback";
17	import {	18	import {
18	readBrowserClientSettings,	19	readBrowserClientSettings,

apps/web/src/rpc/protocol.ts+0−176

@@ -1,176 +0,0 @@
1	import { WsRpcGroup } from "@t3tools/contracts";
2	import {
3	DEFAULT_RECONNECT_BACKOFF,
4	getReconnectDelayMs,
5	type WsProtocolLifecycleHandlers,
6	} from "@t3tools/client-runtime";
7	import { Duration, Effect, Layer, Schedule } from "effect";
8	import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
9	import * as Socket from "effect/unstable/socket/Socket";
10
11	import {
12	acknowledgeRpcRequest,
13	clearAllTrackedRpcRequests,
14	trackRpcRequestSent,
15	} from "./requestLatencyState";
16	import {
17	recordWsConnectionAttempt,
18	recordWsConnectionClosed,
19	recordWsConnectionErrored,
20	recordWsConnectionOpened,
21	} from "./wsConnectionState";
22
23	export type { WsProtocolLifecycleHandlers };
24
25	export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);
26	type RpcClientFactory = typeof makeWsRpcProtocolClient;
27	export type WsRpcProtocolClient =
28	RpcClientFactory extends Effect.Effect<infer Client, any, any> ? Client : never;
29	export type WsRpcProtocolSocketUrlProvider = string \| (() => Promise<string>);
30
31	function formatSocketErrorMessage(error: unknown): string {
32	if (error instanceof Error && error.message.trim().length > 0) {
33	return error.message;
34	}
35	return String(error);
36	}
37
38	function resolveWsRpcSocketUrl(rawUrl: string): string {
39	const resolved = new URL(rawUrl);
40	if (resolved.protocol !== "ws:" && resolved.protocol !== "wss:") {
41	throw new Error(`Unsupported websocket transport URL protocol: ${resolved.protocol}`);
42	}
43
44	resolved.pathname = "/ws";
45	return resolved.toString();
46	}
47
48	function defaultLifecycleHandlers(): Required<WsProtocolLifecycleHandlers> {
49	return {
50	onAttempt: recordWsConnectionAttempt,
51	onOpen: recordWsConnectionOpened,
52	onError: (message) => {
53	clearAllTrackedRpcRequests();
54	recordWsConnectionErrored(message);
55	},
56	onClose: (details) => {
57	clearAllTrackedRpcRequests();
58	recordWsConnectionClosed(details);
59	},
60	};
61	}
62
63	function composeLifecycleHandlers(
64	handlers?: WsProtocolLifecycleHandlers,
65	): Required<WsProtocolLifecycleHandlers> {
66	const defaults = defaultLifecycleHandlers();
67
68	return {
69	onAttempt: (socketUrl) => {
70	defaults.onAttempt(socketUrl);
71	handlers?.onAttempt?.(socketUrl);
72	},
73	onOpen: () => {
74	defaults.onOpen();
75	handlers?.onOpen?.();
76	},
77	onError: (message) => {
78	defaults.onError(message);
79	handlers?.onError?.(message);
80	},
81	onClose: (details) => {
82	defaults.onClose(details);
83	handlers?.onClose?.(details);
84	},
85	};
86	}
87
88	export function createWsRpcProtocolLayer(
89	url: WsRpcProtocolSocketUrlProvider,
90	handlers?: WsProtocolLifecycleHandlers,
91	) {
92	const lifecycle = composeLifecycleHandlers(handlers);
93	const resolvedUrl =
94	typeof url === "function"
95	? Effect.promise(() => url()).pipe(
96	Effect.map((rawUrl) => resolveWsRpcSocketUrl(rawUrl)),
97	Effect.tapError((error) =>
98	Effect.sync(() => {
99	lifecycle.onError(formatSocketErrorMessage(error));
100	}),
101	),
102	Effect.orDie,
103	)
104	: resolveWsRpcSocketUrl(url);
105
106	const trackingWebSocketConstructorLayer = Layer.succeed(
107	Socket.WebSocketConstructor,
108	(socketUrl, protocols) => {
109	lifecycle.onAttempt(socketUrl);
110	const socket = new globalThis.WebSocket(socketUrl, protocols);
111
112	socket.addEventListener(
113	"open",
114	() => {
115	lifecycle.onOpen();
116	},
117	{ once: true },
118	);
119	socket.addEventListener(
120	"error",
121	() => {
122	lifecycle.onError("Unable to connect to the T3 server WebSocket.");
123	},
124	{ once: true },
125	);
126	socket.addEventListener(
127	"close",
128	(event) => {
129	lifecycle.onClose({
130	code: event.code,
131	reason: event.reason,
132	});
133	},
134	{ once: true },
135	);
136
137	return socket;
138	},
139	);
140	const socketLayer = Socket.layerWebSocket(resolvedUrl).pipe(
141	Layer.provide(trackingWebSocketConstructorLayer),
142	);
143	const retryPolicy = Schedule.addDelay(
144	Schedule.recurs(DEFAULT_RECONNECT_BACKOFF.maxRetries!),
145	(retryCount) => Effect.succeed(Duration.millis(getReconnectDelayMs(retryCount) ?? 0)),
146	);
147	const protocolLayer = Layer.effect(
148	RpcClient.Protocol,
149	Effect.map(
150	RpcClient.makeProtocolSocket({
151	retryPolicy,
152	retryTransientErrors: true,
153	}),
154	(protocol) => ({
155	...protocol,
156	run: (clientId, writeResponse) =>
157	protocol.run(clientId, (response) => {
158	if (response._tag === "Chunk" \|\| response._tag === "Exit") {
159	acknowledgeRpcRequest(response.requestId);
160	} else if (response._tag === "ClientProtocolError" \|\| response._tag === "Defect") {
161	clearAllTrackedRpcRequests();
162	}
163	return writeResponse(response);
164	}),
165	send: (clientId, request, transferables) => {
166	if (request._tag === "Request") {
167	trackRpcRequestSent(request.id, request.tag);
168	}
169	return protocol.send(clientId, request, transferables);
170	},
171	}),
172	),
173	);
174
175	return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));
176	}

apps/web/src/rpc/serverState.ts+1−1

@@ -13,7 +13,7 @@ import { · 12 unmodified lines
13	import { Atom } from "effect/unstable/reactivity";	13	import { Atom } from "effect/unstable/reactivity";
14	import { useCallback, useRef } from "react";	14	import { useCallback, useRef } from "react";
15		15
16	import type { WsRpcClient } from "./wsRpcClient";	16	import type { WsRpcClient } from "@t3tools/client-runtime";
17	import { appAtomRegistry, resetAppAtomRegistryForTests } from "./atomRegistry";	17	import { appAtomRegistry, resetAppAtomRegistryForTests } from "./atomRegistry";
18		18
19	export type ServerConfigUpdateSource = ServerConfigStreamEvent["type"];	19	export type ServerConfigUpdateSource = ServerConfigStreamEvent["type"];

apps/web/src/rpc/transportError.test.ts+0−24

@@ -1,24 +0,0 @@
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/wsRpcClient.ts+0−156

@@ -1,156 +0,0 @@
1	import {
2	type GitRunStackedActionResult,
3	type GitStatusResult,
4	type GitStatusStreamEvent,
5	ORCHESTRATION_WS_METHODS,
6	WS_METHODS,
7	} from "@t3tools/contracts";
8	import type { WsRpcClient } from "@t3tools/client-runtime";
9	import { applyGitStatusStreamEvent } from "@t3tools/shared/git";
10
11	import { resetWsReconnectBackoff } from "./wsConnectionState";
12	import { WsTransport } from "./wsTransport";
13
14	export type { WsRpcClient };
15
16	export function createWsRpcClient(transport: WsTransport): WsRpcClient {
17	return {
18	dispose: () => transport.dispose(),
19	reconnect: async () => {
20	resetWsReconnectBackoff();
21	await transport.reconnect();
22	},
23	terminal: {
24	open: (input) => transport.request((client) => client[WS_METHODS.terminalOpen](input)),
25	attach: (input, listener, options) =>
26	transport.subscribe(
27	(client) => client[WS_METHODS.terminalAttach](input),
28	listener,
29	options,
30	),
31	write: (input) => transport.request((client) => client[WS_METHODS.terminalWrite](input)),
32	resize: (input) => transport.request((client) => client[WS_METHODS.terminalResize](input)),
33	clear: (input) => transport.request((client) => client[WS_METHODS.terminalClear](input)),
34	restart: (input) => transport.request((client) => client[WS_METHODS.terminalRestart](input)),
35	close: (input) => transport.request((client) => client[WS_METHODS.terminalClose](input)),
36	onMetadata: (listener, options) =>
37	transport.subscribe(
38	(client) => client[WS_METHODS.subscribeTerminalMetadata]({}),
39	listener,
40	options,
41	),
42	},
43	projects: {
44	searchEntries: (input) =>
45	transport.request((client) => client[WS_METHODS.projectsSearchEntries](input)),
46	writeFile: (input) =>
47	transport.request((client) => client[WS_METHODS.projectsWriteFile](input)),
48	},
49	filesystem: {
50	browse: (input) => transport.request((client) => client[WS_METHODS.filesystemBrowse](input)),
51	},
52	shell: {
53	openInEditor: (input) =>
54	transport.request((client) => client[WS_METHODS.shellOpenInEditor](input)),
55	},
56	git: {
57	pull: (input) => transport.request((client) => client[WS_METHODS.gitPull](input)),
58	refreshStatus: (input) =>
59	transport.request((client) => client[WS_METHODS.gitRefreshStatus](input)),
60	onStatus: (input, listener, options) => {
61	let current: GitStatusResult \| null = null;
62	return transport.subscribe(
63	(client) => client[WS_METHODS.subscribeGitStatus](input),
64	(event: GitStatusStreamEvent) => {
65	current = applyGitStatusStreamEvent(current, event);
66	listener(current);
67	},
68	options,
69	);
70	},
71	runStackedAction: async (input, options) => {
72	let result: GitRunStackedActionResult \| null = null;
73
74	await transport.requestStream(
75	(client) => client[WS_METHODS.gitRunStackedAction](input),
76	(event) => {
77	options?.onProgress?.(event);
78	if (event.kind === "action_finished") {
79	result = event.result;
80	}
81	},
82	);
83
84	if (result) {
85	return result;
86	}
87
88	throw new Error("Git action stream completed without a final result.");
89	},
90	listBranches: (input) =>
91	transport.request((client) => client[WS_METHODS.gitListBranches](input)),
92	createWorktree: (input) =>
93	transport.request((client) => client[WS_METHODS.gitCreateWorktree](input)),
94	removeWorktree: (input) =>
95	transport.request((client) => client[WS_METHODS.gitRemoveWorktree](input)),
96	createBranch: (input) =>
97	transport.request((client) => client[WS_METHODS.gitCreateBranch](input)),
98	checkout: (input) => transport.request((client) => client[WS_METHODS.gitCheckout](input)),
99	init: (input) => transport.request((client) => client[WS_METHODS.gitInit](input)),
100	resolvePullRequest: (input) =>
101	transport.request((client) => client[WS_METHODS.gitResolvePullRequest](input)),
102	preparePullRequestThread: (input) =>
103	transport.request((client) => client[WS_METHODS.gitPreparePullRequestThread](input)),
104	getReviewDiffs: (input) =>
105	transport.request((client) => client[WS_METHODS.gitGetReviewDiffs](input)),
106	},
107	server: {
108	getConfig: () => transport.request((client) => client[WS_METHODS.serverGetConfig]({})),
109	refreshProviders: () =>
110	transport.request((client) => client[WS_METHODS.serverRefreshProviders]({})),
111	upsertKeybinding: (input) =>
112	transport.request((client) => client[WS_METHODS.serverUpsertKeybinding](input)),
113	getSettings: () => transport.request((client) => client[WS_METHODS.serverGetSettings]({})),
114	updateSettings: (patch) =>
115	transport.request((client) => client[WS_METHODS.serverUpdateSettings]({ patch })),
116	subscribeConfig: (listener, options) =>
117	transport.subscribe(
118	(client) => client[WS_METHODS.subscribeServerConfig]({}),
119	listener,
120	options,
121	),
122	subscribeLifecycle: (listener, options) =>
123	transport.subscribe(
124	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
125	listener,
126	options,
127	),
128	subscribeAuthAccess: (listener, options) =>
129	transport.subscribe(
130	(client) => client[WS_METHODS.subscribeAuthAccess]({}),
131	listener,
132	options,
133	),
134	},
135	orchestration: {
136	dispatchCommand: (input) =>
137	transport.request((client) => client[ORCHESTRATION_WS_METHODS.dispatchCommand](input)),
138	getTurnDiff: (input) =>
139	transport.request((client) => client[ORCHESTRATION_WS_METHODS.getTurnDiff](input)),
140	getFullThreadDiff: (input) =>
141	transport.request((client) => client[ORCHESTRATION_WS_METHODS.getFullThreadDiff](input)),
142	subscribeShell: (listener, options) =>
143	transport.subscribe(
144	(client) => client[ORCHESTRATION_WS_METHODS.subscribeShell]({}),
145	listener,
146	options,
147	),
148	subscribeThread: (input, listener, options) =>
149	transport.subscribe(
150	(client) => client[ORCHESTRATION_WS_METHODS.subscribeThread](input),
151	listener,
152	options,
153	),
154	},
155	};
156	}

apps/web/src/rpc/wsTransport.test.ts+1−703

@@ -1,5 +1,4 @@
1	import { DEFAULT_SERVER_SETTINGS, WS_METHODS } from "@t3tools/contracts";	1	import { DEFAULT_SERVER_SETTINGS, WS_METHODS } from "@t3tools/contracts";
2	import { Stream } from "effect";
3	import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";	2	import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
4		3
5	import {	4	import {
@@ -152,51 +151,7 @@ afterEach(async () => { · 146 unmodified lines
152	vi.restoreAllMocks();	151	vi.restoreAllMocks();
153	});	152	});
154		153
155	describe("WsTransport", () => {	154	describe("WsTransport (web instrumentation)", () => {
156	it("normalizes root websocket urls to /ws and preserves query params", async () => {
157	const transport = createTransport("ws://localhost:3020/?token=secret-token");
158
159	await waitFor(() => {
160	expect(sockets).toHaveLength(1);
161	});
162
163	expect(getSocket().url).toBe("ws://localhost:3020/ws?token=secret-token");
164	await transport.dispose();
165	});
166
167	it("uses an explicit secure websocket base url", async () => {
168	const transport = createTransport("wss://app.example.com");
169
170	await waitFor(() => {
171	expect(sockets).toHaveLength(1);
172	});
173
174	expect(getSocket().url).toBe("wss://app.example.com/ws");
175	await transport.dispose();
176	});
177
178	it("uses an explicit insecure websocket base url for remote backends", async () => {
179	const transport = createTransport("ws://192.168.1.44:3773");
180
181	await waitFor(() => {
182	expect(sockets).toHaveLength(1);
183	});
184
185	expect(getSocket().url).toBe("ws://192.168.1.44:3773/ws");
186	await transport.dispose();
187	});
188
189	it("supports async websocket url providers", async () => {
190	const transport = createTransport(async () => "wss://remote.example.com/?wsToken=dynamic");
191
192	await waitFor(() => {
193	expect(sockets).toHaveLength(1);
194	});
195
196	expect(getSocket().url).toBe("wss://remote.example.com/ws?wsToken=dynamic");
197	await transport.dispose();
198	});
199
200	it("tracks initial connection failures for the app error state", async () => {	155	it("tracks initial connection failures for the app error state", async () => {
201	const transport = createTransport("ws://localhost:3020");	156	const transport = createTransport("ws://localhost:3020");
202		157
@@ -298,69 +253,6 @@ describe("WsTransport", () => { · 95 unmodified lines
298	await transport.dispose();	253	await transport.dispose();
299	});	254	});
300		255
301	it("reconnects the websocket session without disposing the transport", async () => {
302	const transport = createTransport("ws://localhost:3020");
303
304	await waitFor(() => {
305	expect(sockets).toHaveLength(1);
306	});
307
308	const firstSocket = getSocket();
309	firstSocket.open();
310
311	await waitFor(() => {
312	expect(getWsConnectionStatus()).toMatchObject({
313	hasConnected: true,
314	phase: "connected",
315	});
316	});
317
318	await transport.reconnect();
319
320	await waitFor(() => {
321	expect(sockets).toHaveLength(2);
322	});
323
324	const secondSocket = getSocket();
325	expect(secondSocket).not.toBe(firstSocket);
326	expect(firstSocket.readyState).toBe(MockWebSocket.CLOSED);
327
328	const requestPromise = transport.request((client) =>
329	client[WS_METHODS.serverUpsertKeybinding]({
330	command: "terminal.toggle",
331	key: "ctrl+k",
332	}),
333	);
334
335	secondSocket.open();
336
337	await waitFor(() => {
338	expect(secondSocket.sent).toHaveLength(1);
339	});
340
341	const requestMessage = JSON.parse(secondSocket.sent[0] ?? "{}") as { id: string };
342	secondSocket.serverMessage(
343	JSON.stringify({
344	_tag: "Exit",
345	requestId: requestMessage.id,
346	exit: {
347	_tag: "Success",
348	value: {
349	keybindings: [],
350	issues: [],
351	},
352	},
353	}),
354	);
355
356	await expect(requestPromise).resolves.toEqual({
357	keybindings: [],
358	issues: [],
359	});
360
361	await transport.dispose();
362	});
363
364	it("marks unary requests as slow until the first server ack arrives", async () => {	256	it("marks unary requests as slow until the first server ack arrives", async () => {
365	const slowAckThresholdMs = 25;	257	const slowAckThresholdMs = 25;
366	setSlowRpcAckThresholdMsForTests(slowAckThresholdMs);	258	setSlowRpcAckThresholdMsForTests(slowAckThresholdMs);
@@ -467,600 +359,6 @@ describe("WsTransport", () => { · 100 unmodified lines
467	await transport.dispose();	359	await transport.dispose();
468	}, 5_000);	360	}, 5_000);
469		361
470	it("sends unary RPC requests and resolves successful exits", async () => {
471	const transport = createTransport("ws://localhost:3020");
472
473	const requestPromise = transport.request((client) =>
474	client[WS_METHODS.serverUpsertKeybinding]({
475	command: "terminal.toggle",
476	key: "ctrl+k",
477	}),
478	);
479
480	await waitFor(() => {
481	expect(sockets).toHaveLength(1);
482	});
483
484	const socket = getSocket();
485	socket.open();
486
487	await waitFor(() => {
488	expect(socket.sent).toHaveLength(1);
489	});
490
491	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as {
492	_tag: string;
493	id: string;
494	payload: unknown;
495	tag: string;
496	};
497	expect(requestMessage).toMatchObject({
498	_tag: "Request",
499	tag: WS_METHODS.serverUpsertKeybinding,
500	payload: {
501	command: "terminal.toggle",
502	key: "ctrl+k",
503	},
504	});
505
506	socket.serverMessage(
507	JSON.stringify({
508	_tag: "Exit",
509	requestId: requestMessage.id,
510	exit: {
511	_tag: "Success",
512	value: {
513	keybindings: [],
514	issues: [],
515	},
516	},
517	}),
518	);
519
520	await expect(requestPromise).resolves.toEqual({
521	keybindings: [],
522	issues: [],
523	});
524
525	await transport.dispose();
526	});
527
528	it("delivers stream chunks to subscribers", async () => {
529	const transport = createTransport("ws://localhost:3020");
530	const listener = vi.fn();
531
532	const unsubscribe = transport.subscribe(
533	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
534	listener,
535	);
536	await waitFor(() => {
537	expect(sockets).toHaveLength(1);
538	});
539
540	const socket = getSocket();
541	socket.open();
542
543	await waitFor(() => {
544	expect(socket.sent).toHaveLength(1);
545	});
546
547	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as { id: string; tag: string };
548	expect(requestMessage.tag).toBe(WS_METHODS.subscribeServerLifecycle);
549
550	const welcomeEvent = {
551	version: 1,
552	sequence: 1,
553	type: "welcome",
554	payload: {
555	environment: {
556	environmentId: "environment-local",
557	label: "Local environment",
558	platform: { os: "darwin", arch: "arm64" },
559	serverVersion: "0.0.0-test",
560	capabilities: { repositoryIdentity: true },
561	},
562	cwd: "/tmp/workspace",
563	projectName: "workspace",
564	},
565	};
566
567	socket.serverMessage(
568	JSON.stringify({
569	_tag: "Chunk",
570	requestId: requestMessage.id,
571	values: [welcomeEvent],
572	}),
573	);
574
575	await waitFor(() => {
576	expect(listener).toHaveBeenCalledWith(welcomeEvent);
577	});
578
579	unsubscribe();
580	await transport.dispose();
581	});
582
583	it("re-subscribes stream listeners after the stream exits", async () => {
584	const transport = createTransport("ws://localhost:3020");
585	const listener = vi.fn();
586	const onResubscribe = vi.fn();
587
588	const unsubscribe = transport.subscribe(
589	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
590	listener,
591	{ onResubscribe },
592	);
593	await waitFor(() => {
594	expect(sockets).toHaveLength(1);
595	});
596
597	const socket = getSocket();
598	socket.open();
599
600	await waitFor(() => {
601	expect(socket.sent).toHaveLength(1);
602	});
603
604	const firstRequest = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
605	socket.serverMessage(
606	JSON.stringify({
607	_tag: "Chunk",
608	requestId: firstRequest.id,
609	values: [
610	{
611	version: 1,
612	sequence: 1,
613	type: "welcome",
614	payload: {
615	environment: {
616	environmentId: "environment-local",
617	label: "Local environment",
618	platform: { os: "darwin", arch: "arm64" },
619	serverVersion: "0.0.0-test",
620	capabilities: { repositoryIdentity: true },
621	},
622	cwd: "/tmp/one",
623	projectName: "one",
624	},
625	},
626	],
627	}),
628	);
629	socket.serverMessage(
630	JSON.stringify({
631	_tag: "Exit",
632	requestId: firstRequest.id,
633	exit: {
634	_tag: "Success",
635	value: null,
636	},
637	}),
638	);
639
640	await waitFor(() => {
641	const nextRequest = socket.sent
642	.map((message) => JSON.parse(message) as { _tag?: string; id?: string })
643	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
644	expect(nextRequest).toBeDefined();
645	});
646	expect(onResubscribe).toHaveBeenCalledOnce();
647
648	const secondRequest = socket.sent
649	.map((message) => JSON.parse(message) as { _tag?: string; id?: string; tag?: string })
650	.find(
651	(message): message is { _tag: "Request"; id: string; tag: string } =>
652	message._tag === "Request" && message.id !== firstRequest.id,
653	);
654	if (!secondRequest) {
655	throw new Error("Expected a resubscribe request");
656	}
657	expect(secondRequest.tag).toBe(WS_METHODS.subscribeServerLifecycle);
658	expect(secondRequest.id).not.toBe(firstRequest.id);
659
660	const secondEvent = {
661	version: 1,
662	sequence: 2,
663	type: "welcome",
664	payload: {
665	environment: {
666	environmentId: "environment-local",
667	label: "Local environment",
668	platform: { os: "darwin", arch: "arm64" },
669	serverVersion: "0.0.0-test",
670	capabilities: { repositoryIdentity: true },
671	},
672	cwd: "/tmp/two",
673	projectName: "two",
674	},
675	};
676	socket.serverMessage(
677	JSON.stringify({
678	_tag: "Chunk",
679	requestId: secondRequest.id,
680	values: [secondEvent],
681	}),
682	);
683
684	await waitFor(() => {
685	expect(listener).toHaveBeenLastCalledWith(secondEvent);
686	});
687
688	unsubscribe();
689	await transport.dispose();
690	});
691
692	it("re-subscribes live stream listeners after an explicit transport reconnect", async () => {
693	const transport = createTransport("ws://localhost:3020");
694	const listener = vi.fn();
695	const onResubscribe = vi.fn();
696
697	const unsubscribe = transport.subscribe(
698	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
699	listener,
700	{ onResubscribe },
701	);
702
703	await waitFor(() => {
704	expect(sockets).toHaveLength(1);
705	});
706
707	const firstSocket = getSocket();
708	firstSocket.open();
709
710	await waitFor(() => {
711	expect(firstSocket.sent).toHaveLength(1);
712	});
713
714	const firstRequest = JSON.parse(firstSocket.sent[0] ?? "{}") as { id: string };
715	const firstEvent = {
716	version: 1,
717	sequence: 1,
718	type: "welcome",
719	payload: {
720	environment: {
721	environmentId: "environment-local",
722	label: "Local environment",
723	platform: { os: "darwin", arch: "arm64" },
724	serverVersion: "0.0.0-test",
725	capabilities: { repositoryIdentity: true },
726	},
727	cwd: "/tmp/one",
728	projectName: "one",
729	},
730	};
731
732	firstSocket.serverMessage(
733	JSON.stringify({
734	_tag: "Chunk",
735	requestId: firstRequest.id,
736	values: [firstEvent],
737	}),
738	);
739
740	await waitFor(() => {
741	expect(listener).toHaveBeenLastCalledWith(firstEvent);
742	});
743
744	await transport.reconnect();
745
746	await waitFor(() => {
747	expect(sockets).toHaveLength(2);
748	});
749
750	const secondSocket = getSocket();
751	expect(secondSocket).not.toBe(firstSocket);
752	expect(firstSocket.readyState).toBe(MockWebSocket.CLOSED);
753
754	secondSocket.open();
755
756	await waitFor(() => {
757	expect(secondSocket.sent).toHaveLength(1);
758	});
759
760	const secondRequest = JSON.parse(secondSocket.sent[0] ?? "{}") as {
761	id: string;
762	tag: string;
763	};
764	expect(secondRequest.tag).toBe(WS_METHODS.subscribeServerLifecycle);
765	expect(secondRequest.id).not.toBe(firstRequest.id);
766	expect(onResubscribe).toHaveBeenCalledOnce();
767
768	const secondEvent = {
769	version: 1,
770	sequence: 2,
771	type: "welcome",
772	payload: {
773	environment: {
774	environmentId: "environment-local",
775	label: "Local environment",
776	platform: { os: "darwin", arch: "arm64" },
777	serverVersion: "0.0.0-test",
778	capabilities: { repositoryIdentity: true },
779	},
780	cwd: "/tmp/two",
781	projectName: "two",
782	},
783	};
784
785	secondSocket.serverMessage(
786	JSON.stringify({
787	_tag: "Chunk",
788	requestId: secondRequest.id,
789	values: [secondEvent],
790	}),
791	);
792
793	await waitFor(() => {
794	expect(listener).toHaveBeenLastCalledWith(secondEvent);
795	});
796
797	unsubscribe();
798	await transport.dispose();
799	});
800
801	it("does not fire onResubscribe when the first stream attempt exits before any value", async () => {
802	const transport = createTransport("ws://localhost:3020");
803	const listener = vi.fn();
804	const onResubscribe = vi.fn();
805
806	const unsubscribe = transport.subscribe(
807	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
808	listener,
809	{ onResubscribe },
810	);
811	await waitFor(() => {
812	expect(sockets).toHaveLength(1);
813	});
814
815	const socket = getSocket();
816	socket.open();
817
818	await waitFor(() => {
819	expect(socket.sent).toHaveLength(1);
820	});
821
822	const firstRequest = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
823	socket.serverMessage(
824	JSON.stringify({
825	_tag: "Exit",
826	requestId: firstRequest.id,
827	exit: {
828	_tag: "Success",
829	value: null,
830	},
831	}),
832	);
833
834	await waitFor(() => {
835	const nextRequest = socket.sent
836	.map((message) => JSON.parse(message) as { _tag?: string; id?: string })
837	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
838	expect(nextRequest).toBeDefined();
839	});
840	expect(onResubscribe).not.toHaveBeenCalled();
841	expect(listener).not.toHaveBeenCalled();
842
843	unsubscribe();
844	await transport.dispose();
845	});
846
847	it("does not retry stream subscriptions after application-level failures", async () => {
848	const transport = createTransport("ws://localhost:3020");
849	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
850	let attempts = 0;
851
852	const unsubscribe = transport.subscribe(
853	() =>
854	Stream.suspend(() => {
855	attempts += 1;
856	return Stream.fail(new Error("Git command failed in GitCore.statusDetails"));
857	}),
858	vi.fn(),
859	{ retryDelay: 10 },
860	);
861
862	await waitFor(() => {
863	expect(sockets).toHaveLength(1);
864	});
865
866	getSocket().open();
867
868	await waitFor(() => {
869	expect(attempts).toBe(1);
870	});
871	await new Promise((resolve) => setTimeout(resolve, 50));
872
873	expect(attempts).toBe(1);
874	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription failed", {
875	error: "Git command failed in GitCore.statusDetails",
876	});
877	expect(warnSpy).not.toHaveBeenCalledWith(
878	"WebSocket RPC subscription disconnected",
879	expect.anything(),
880	);
881
882	unsubscribe();
883	await transport.dispose();
884	});
885
886	it("keeps retrying stream subscriptions after transport failures", async () => {
887	const transport = createTransport("ws://localhost:3020");
888	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
889	let attempts = 0;
890
891	const unsubscribe = transport.subscribe(
892	() =>
893	Stream.suspend(() => {
894	attempts += 1;
895	return Stream.fail(new Error("SocketCloseError: WebSocket closed"));
896	}),
897	vi.fn(),
898	{ retryDelay: 10 },
899	);
900
901	await waitFor(() => {
902	expect(sockets).toHaveLength(1);
903	});
904
905	getSocket().open();
906
907	await waitFor(() => {
908	expect(attempts).toBeGreaterThanOrEqual(2);
909	});
910
911	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription disconnected", {
912	error: "SocketCloseError: WebSocket closed",
913	});
914
915	unsubscribe();
916	await transport.dispose();
917	});
918
919	it("logs a transport disconnect once even when multiple subscriptions fail together", async () => {
920	const transport = createTransport("ws://localhost:3020");
921	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
922
923	const unsubscribeA = transport.subscribe(
924	() => Stream.fail(new Error("SocketCloseError: 1006")),
925	vi.fn(),
926	{ retryDelay: 10 },
927	);
928	const unsubscribeB = transport.subscribe(
929	() => Stream.fail(new Error("SocketCloseError: 1006")),
930	vi.fn(),
931	{ retryDelay: 10 },
932	);
933
934	await waitFor(() => {
935	expect(sockets).toHaveLength(1);
936	});
937
938	getSocket().open();
939
940	await waitFor(() => {
941	expect(warnSpy).toHaveBeenCalledTimes(1);
942	});
943	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription disconnected", {
944	error: "SocketCloseError: 1006",
945	});
946
947	unsubscribeA();
948	unsubscribeB();
949	await transport.dispose();
950	});
951
952	it("streams finite request events without re-subscribing", async () => {
953	const transport = createTransport("ws://localhost:3020");
954	const listener = vi.fn();
955
956	await waitFor(() => {
957	expect(sockets).toHaveLength(1);
958	});
959	const socket = getSocket();
960	socket.open();
961
962	const requestPromise = transport.requestStream(
963	(client) =>
964	client[WS_METHODS.gitRunStackedAction]({
965	actionId: "action-1",
966	cwd: "/repo",
967	action: "commit",
968	}),
969	listener,
970	);
971
972	await waitFor(() => {
973	expect(socket.sent).toHaveLength(1);
974	});
975
976	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
977	const progressEvent = {
978	actionId: "action-1",
979	cwd: "/repo",
980	action: "commit",
981	kind: "phase_started",
982	phase: "commit",
983	label: "Committing...",
984	} as const;
985
986	socket.serverMessage(
987	JSON.stringify({
988	_tag: "Chunk",
989	requestId: requestMessage.id,
990	values: [progressEvent],
991	}),
992	);
993	socket.serverMessage(
994	JSON.stringify({
995	_tag: "Exit",
996	requestId: requestMessage.id,
997	exit: {
998	_tag: "Success",
999	value: null,
1000	},
1001	}),
1002	);
1003
1004	await expect(requestPromise).resolves.toBeUndefined();
1005	expect(listener).toHaveBeenCalledWith(progressEvent);
1006	expect(
1007	socket.sent.filter((message) => {
1008	const parsed = JSON.parse(message) as { _tag?: string; tag?: string };
1009	return parsed._tag === "Request" && parsed.tag === WS_METHODS.gitRunStackedAction;
1010	}),
1011	).toHaveLength(1);
1012	await transport.dispose();
1013	});
1014
1015	it("closes the client scope on the transport runtime before disposing the runtime", async () => {
1016	const callOrder: string[] = [];
1017	let resolveClose!: () => void;
1018	const closePromise = new Promise<void>((resolve) => {
1019	resolveClose = resolve;
1020	});
1021
1022	const runtime = {
1023	runPromise: vi.fn(async () => {
1024	callOrder.push("close:start");
1025	await closePromise;
1026	callOrder.push("close:done");
1027	return undefined;
1028	}),
1029	dispose: vi.fn(async () => {
1030	callOrder.push("runtime:dispose");
1031	}),
1032	};
1033	const transport = {
1034	disposed: false,
1035	session: {
1036	clientScope: {} as never,
1037	runtime,
1038	},
1039	closeSession: (
1040	WsTransport.prototype as unknown as {
1041	closeSession: (session: {
1042	clientScope: unknown;
1043	runtime: { dispose: () => Promise<void>; runPromise: () => Promise<void> };
1044	}) => Promise<void>;
1045	}
1046	).closeSession,
1047	} as unknown as WsTransport;
1048
1049	void WsTransport.prototype.dispose.call(transport);
1050
1051	expect(runtime.runPromise).toHaveBeenCalledTimes(1);
1052	expect(runtime.dispose).not.toHaveBeenCalled();
1053	expect((transport as unknown as { disposed: boolean }).disposed).toBe(true);
1054
1055	resolveClose();
1056
1057	await waitFor(() => {
1058	expect(runtime.dispose).toHaveBeenCalledTimes(1);
1059	});
1060
1061	expect(callOrder).toEqual(["close:start", "close:done", "runtime:dispose"]);
1062	});
1063
1064	it("propagates OTLP trace ids for ws transport requests when client tracing is enabled", async () => {	362	it("propagates OTLP trace ids for ws transport requests when client tracing is enabled", async () => {
1065	await configureClientTracing({	363	await configureClientTracing({
1066	exportIntervalMs: 10,	364	exportIntervalMs: 10,

apps/web/src/rpc/wsTransport.ts+47−274

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

packages/client-runtime/src/transportError.test.ts+3−0

@@ -41,6 +41,9 @@ describe("sanitizeThreadErrorMessage", () => { · 40 unmodified lines
41		41
42	it("preserves non-transport errors", () => {	42	it("preserves non-transport errors", () => {
43	expect(sanitizeThreadErrorMessage("Thread not found")).toBe("Thread not found");	43	expect(sanitizeThreadErrorMessage("Thread not found")).toBe("Thread not found");
		44	expect(sanitizeThreadErrorMessage("Select a base branch before sending.")).toBe(
		45	"Select a base branch before sending.",
		46	);
44	});	47	});
45		48
46	it("returns null for null/undefined", () => {	49	it("returns null for null/undefined", () => {

packages/client-runtime/src/wsRpcClient.test.tsrenamed from apps/web/src/rpc/wsRpcClient.test.ts+29−4

@@ -5,7 +5,7 @@ import type { · 4 unmodified lines
5	} from "@t3tools/contracts";	5	} from "@t3tools/contracts";
6	import { describe, expect, it, vi } from "vitest";	6	import { describe, expect, it, vi } from "vitest";
7		7
8	vi.mock("./wsTransport", () => ({	8	vi.mock("./wsTransport.ts", () => ({
9	WsTransport: class WsTransport {	9	WsTransport: class WsTransport {
10	dispose = vi.fn(async () => undefined);	10	dispose = vi.fn(async () => undefined);
11	reconnect = vi.fn(async () => undefined);	11	reconnect = vi.fn(async () => undefined);
@@ -15,8 +15,8 @@ vi.mock("./wsTransport", () => ({ · 3 unmodified lines
15	},	15	},
16	}));	16	}));
17		17
18	import { createWsRpcClient } from "./wsRpcClient";	18	import { createWsRpcClient } from "./wsRpcClient.ts";
19	import { type WsTransport } from "./wsTransport";	19	import type { WsTransport } from "./wsTransport.ts";
20		20
21	const baseLocalStatus: GitStatusLocalResult = {	21	const baseLocalStatus: GitStatusLocalResult = {
22	isRepo: true,	22	isRepo: true,
@@ -34,7 +34,32 @@ const baseRemoteStatus: GitStatusRemoteResult = { · 11 unmodified lines
34	pr: null,	34	pr: null,
35	};	35	};
36		36
37	describe("wsRpcClient", () => {	37	describe("createWsRpcClient", () => {
		38	it("runs beforeReconnect before awaiting transport.reconnect", async () => {
		39	const order: string[] = [];
		40	const transport = {
		41	dispose: vi.fn(async () => undefined),
		42	reconnect: vi.fn(async () => {
		43	order.push("reconnect");
		44	}),
		45	request: vi.fn(),
		46	requestStream: vi.fn(),
		47	subscribe: vi.fn(() => () => undefined),
		48	} satisfies Pick<
		49	WsTransport,
		50	"dispose" \| "reconnect" \| "request" \| "requestStream" \| "subscribe"
		51	>;
		52
		53	const client = createWsRpcClient(transport as unknown as WsTransport, {
		54	beforeReconnect: () => {
		55	order.push("beforeReconnect");
		56	},
		57	});
		58
		59	await client.reconnect();
		60	expect(order).toEqual(["beforeReconnect", "reconnect"]);
		61	});
		62
38	it("reduces git status stream events into flat status snapshots", () => {	63	it("reduces git status stream events into flat status snapshots", () => {
39	const subscribe = vi.fn(<TValue>(_connect: unknown, listener: (value: TValue) => void) => {	64	const subscribe = vi.fn(<TValue>(_connect: unknown, listener: (value: TValue) => void) => {
40	for (const event of [	65	for (const event of [

packages/client-runtime/src/wsRpcClient.ts+13−2

@@ -122,10 +122,21 @@ export interface WsRpcClient { · 121 unmodified lines
122	};	122	};
123	}	123	}
124		124
125	export function createWsRpcClient(transport: WsTransport): WsRpcClient {	125	export interface CreateWsRpcClientOptions {
		126	/** Runs immediately before `transport.reconnect()` (e.g. reset reconnect UI/backoff state). */
		127	readonly beforeReconnect?: () => void;
		128	}
		129
		130	export function createWsRpcClient(
		131	transport: WsTransport,
		132	options?: CreateWsRpcClientOptions,
		133	): WsRpcClient {
126	return {	134	return {
127	dispose: () => transport.dispose(),	135	dispose: () => transport.dispose(),
128	reconnect: () => transport.reconnect(),	136	reconnect: async () => {
		137	options?.beforeReconnect?.();
		138	await transport.reconnect();
		139	},
129	terminal: {	140	terminal: {
130	open: (input) => transport.request((client) => client[WS_METHODS.terminalOpen](input)),	141	open: (input) => transport.request((client) => client[WS_METHODS.terminalOpen](input)),
131	attach: (input, listener, options) =>	142	attach: (input, listener, options) =>

packages/client-runtime/src/wsRpcProtocol.ts+79−8

@@ -16,9 +16,22 @@ export interface WsProtocolLifecycleHandlers { · 15 unmodified lines
16	readonly onClose?: (details: { readonly code: number; readonly reason: string }) => void;	16	readonly onClose?: (details: { readonly code: number; readonly reason: string }) => void;
17	}	17	}
18		18
		19	export interface WsRpcProtocolRequestTelemetry {
		20	readonly onRequestSent?: (requestId: string, tag: string) => void;
		21	readonly onRequestAcknowledged?: (requestId: string) => void;
		22	readonly onClearTrackedRequests?: () => void;
		23	}
		24
19	export interface WsRpcProtocolOptions {	25	export interface WsRpcProtocolOptions {
20	/** Backoff configuration for reconnect retries. */	26	/** Backoff configuration for reconnect retries. */
21	readonly backoff?: ReconnectBackoffConfig;	27	readonly backoff?: ReconnectBackoffConfig;
		28	/**
		29	* Invoked before user {@link WsProtocolLifecycleHandlers} for each socket lifecycle event.
		30	* Use for additive telemetry (connection state, clearing request trackers on disconnect).
		31	*/
		32	readonly telemetryLifecycle?: WsProtocolLifecycleHandlers;
		33	/** Optional hooks around outbound requests and inbound RPC responses (latency tracking, etc.). */
		34	readonly requestTelemetry?: WsRpcProtocolRequestTelemetry;
22	}	35	}
23		36
24	export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);	37	export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);
@@ -54,16 +67,49 @@ function defaultLifecycleHandlers(): Required<WsProtocolLifecycleHandlers> { · 29 unmodified lines
54	};	67	};
55	}	68	}
56		69
		70	function resolveLifecycleHandlers(
		71	handlers: WsProtocolLifecycleHandlers \| undefined,
		72	telemetryLifecycle: WsProtocolLifecycleHandlers \| undefined,
		73	): Required<WsProtocolLifecycleHandlers> {
		74	if (telemetryLifecycle === undefined) {
		75	return {
		76	...defaultLifecycleHandlers(),
		77	...handlers,
		78	};
		79	}
		80
		81	return {
		82	onAttempt: (socketUrl) => {
		83	telemetryLifecycle.onAttempt?.(socketUrl);
		84	handlers?.onAttempt?.(socketUrl);
		85	},
		86	onOpen: () => {
		87	telemetryLifecycle.onOpen?.();
		88	handlers?.onOpen?.();
		89	},
		90	onError: (message) => {
		91	telemetryLifecycle.onError?.(message);
		92	handlers?.onError?.(message);
		93	},
		94	onClose: (details) => {
		95	telemetryLifecycle.onClose?.(details);
		96	handlers?.onClose?.(details);
		97	},
		98	};
		99	}
		100
57	export function createWsRpcProtocolLayer(	101	export function createWsRpcProtocolLayer(
58	url: WsRpcProtocolSocketUrlProvider,	102	url: WsRpcProtocolSocketUrlProvider,
59	handlers?: WsProtocolLifecycleHandlers,	103	handlers?: WsProtocolLifecycleHandlers,
60	options?: WsRpcProtocolOptions,	104	options?: WsRpcProtocolOptions,
61	) {	105	) {
62	const lifecycle = {	106	const lifecycle = resolveLifecycleHandlers(handlers, options?.telemetryLifecycle);
63	...defaultLifecycleHandlers(),
64	...handlers,
65	};
66	const backoff = options?.backoff ?? DEFAULT_RECONNECT_BACKOFF;	107	const backoff = options?.backoff ?? DEFAULT_RECONNECT_BACKOFF;
		108	const requestTelemetry = options?.requestTelemetry;
		109	const instrumentRequests =
		110	requestTelemetry?.onRequestSent !== undefined \|\|
		111	requestTelemetry?.onRequestAcknowledged !== undefined \|\|
		112	requestTelemetry?.onClearTrackedRequests !== undefined;
67		113
68	const resolvedUrl =	114	const resolvedUrl =
69	typeof url === "function"	115	typeof url === "function"
@@ -123,10 +169,35 @@ export function createWsRpcProtocolLayer( · 53 unmodified lines
123	);	169	);
124	const protocolLayer = Layer.effect(	170	const protocolLayer = Layer.effect(
125	RpcClient.Protocol,	171	RpcClient.Protocol,
126	RpcClient.makeProtocolSocket({	172	instrumentRequests
127	retryPolicy,	173	? Effect.map(
128	retryTransientErrors: true,	174	RpcClient.makeProtocolSocket({
129	}),	175	retryPolicy,
		176	retryTransientErrors: true,
		177	}),
		178	(protocol) => ({
		179	...protocol,
		180	run: (clientId, writeResponse) =>
		181	protocol.run(clientId, (response) => {
		182	if (response._tag === "Chunk" \|\| response._tag === "Exit") {
		183	requestTelemetry?.onRequestAcknowledged?.(response.requestId);
		184	} else if (response._tag === "ClientProtocolError" \|\| response._tag === "Defect") {
		185	requestTelemetry?.onClearTrackedRequests?.();
		186	}
		187	return writeResponse(response);
		188	}),
		189	send: (clientId, request, transferables) => {
		190	if (request._tag === "Request") {
		191	requestTelemetry?.onRequestSent?.(request.id, request.tag);
		192	}
		193	return protocol.send(clientId, request, transferables);
		194	},
		195	}),
		196	)
		197	: RpcClient.makeProtocolSocket({
		198	retryPolicy,
		199	retryTransientErrors: true,
		200	}),
130	);	201	);
131		202
132	return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));	203	return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));

packages/client-runtime/src/wsTransport.test.ts+861−0

@@ -0,0 +1,861 @@
	1	import { WS_METHODS } from "@t3tools/contracts";
	2	import { Stream } from "effect";
	3	import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
	4
	5	import { WsTransport } from "./wsTransport.ts";
	6
	7	type WsEventType = "open" \| "message" \| "close" \| "error";
	8	type WsEvent = { code?: number; data?: unknown; reason?: string; type?: string };
	9	type WsListener = (event?: WsEvent) => void;
	10
	11	const sockets: MockWebSocket[] = [];
	12
	13	class MockWebSocket {
	14	static readonly CONNECTING = 0;
	15	static readonly OPEN = 1;
	16	static readonly CLOSING = 2;
	17	static readonly CLOSED = 3;
	18
	19	readyState = MockWebSocket.CONNECTING;
	20	readonly sent: string[] = [];
	21	readonly url: string;
	22	private readonly listeners = new Map<WsEventType, Set<WsListener>>();
	23
	24	constructor(url: string) {
	25	this.url = url;
	26	sockets.push(this);
	27	}
	28
	29	addEventListener(type: WsEventType, listener: WsListener) {
	30	const listeners = this.listeners.get(type) ?? new Set<WsListener>();
	31	listeners.add(listener);
	32	this.listeners.set(type, listeners);
	33	}
	34
	35	removeEventListener(type: WsEventType, listener: WsListener) {
	36	this.listeners.get(type)?.delete(listener);
	37	}
	38
	39	send(data: string) {
	40	this.sent.push(data);
	41	}
	42
	43	close(code = 1000, reason = "") {
	44	this.readyState = MockWebSocket.CLOSED;
	45	this.emit("close", { code, reason, type: "close" });
	46	}
	47
	48	open() {
	49	this.readyState = MockWebSocket.OPEN;
	50	this.emit("open", { type: "open" });
	51	}
	52
	53	serverMessage(data: unknown) {
	54	this.emit("message", { data, type: "message" });
	55	}
	56
	57	error() {
	58	this.emit("error", { type: "error" });
	59	}
	60
	61	private emit(type: WsEventType, event?: WsEvent) {
	62	const listeners = this.listeners.get(type);
	63	if (!listeners) return;
	64	for (const listener of listeners) {
	65	listener(event);
	66	}
	67	}
	68	}
	69
	70	const originalWebSocket = globalThis.WebSocket;
	71	const originalFetch = globalThis.fetch;
	72	const transports: WsTransport[] = [];
	73
	74	function getSocket(): MockWebSocket {
	75	const socket = sockets.at(-1);
	76	if (!socket) {
	77	throw new Error("Expected a websocket instance");
	78	}
	79	return socket;
	80	}
	81
	82	async function waitFor(assertion: () => void, timeoutMs = 1_000): Promise<void> {
	83	const startedAt = Date.now();
	84	for (;;) {
	85	try {
	86	assertion();
	87	return;
	88	} catch (error) {
	89	if (Date.now() - startedAt >= timeoutMs) {
	90	throw error;
	91	}
	92	await new Promise((resolve) => setTimeout(resolve, 10));
	93	}
	94	}
	95	}
	96
	97	function createTransport(...args: ConstructorParameters<typeof WsTransport>): WsTransport {
	98	const transport = new WsTransport(...args);
	99	transports.push(transport);
	100	return transport;
	101	}
	102
	103	beforeEach(() => {
	104	vi.useRealTimers();
	105	sockets.length = 0;
	106	transports.length = 0;
	107
	108	Object.defineProperty(globalThis, "window", {
	109	configurable: true,
	110	value: {
	111	location: {
	112	origin: "http://localhost:3020",
	113	hostname: "localhost",
	114	port: "3020",
	115	protocol: "http:",
	116	},
	117	desktopBridge: undefined,
	118	},
	119	});
	120	Object.defineProperty(globalThis, "navigator", {
	121	configurable: true,
	122	value: { onLine: true },
	123	});
	124
	125	globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket;
	126	});
	127
	128	afterEach(async () => {
	129	await Promise.allSettled(transports.map((transport) => transport.dispose()));
	130	transports.length = 0;
	131	globalThis.WebSocket = originalWebSocket;
	132	globalThis.fetch = originalFetch;
	133	vi.restoreAllMocks();
	134	});
	135
	136	describe("WsTransport", () => {
	137	it("normalizes root websocket urls to /ws and preserves query params", async () => {
	138	const transport = createTransport("ws://localhost:3020/?token=secret-token");
	139
	140	await waitFor(() => {
	141	expect(sockets).toHaveLength(1);
	142	});
	143
	144	expect(getSocket().url).toBe("ws://localhost:3020/ws?token=secret-token");
	145	await transport.dispose();
	146	});
	147
	148	it("uses an explicit secure websocket base url", async () => {
	149	const transport = createTransport("wss://app.example.com");
	150
	151	await waitFor(() => {
	152	expect(sockets).toHaveLength(1);
	153	});
	154
	155	expect(getSocket().url).toBe("wss://app.example.com/ws");
	156	await transport.dispose();
	157	});
	158
	159	it("uses an explicit insecure websocket base url for remote backends", async () => {
	160	const transport = createTransport("ws://192.168.1.44:3773");
	161
	162	await waitFor(() => {
	163	expect(sockets).toHaveLength(1);
	164	});
	165
	166	expect(getSocket().url).toBe("ws://192.168.1.44:3773/ws");
	167	await transport.dispose();
	168	});
	169
	170	it("supports async websocket url providers", async () => {
	171	const transport = createTransport(async () => "wss://remote.example.com/?wsToken=dynamic");
	172
	173	await waitFor(() => {
	174	expect(sockets).toHaveLength(1);
	175	});
	176
	177	expect(getSocket().url).toBe("wss://remote.example.com/ws?wsToken=dynamic");
	178	await transport.dispose();
	179	});
	180
	181	it("invokes optional lifecycle handlers when the socket opens and closes", async () => {
	182	const onOpen = vi.fn();
	183	const onClose = vi.fn();
	184	const transport = createTransport("ws://localhost:3020", {
	185	onOpen,
	186	onClose,
	187	});
	188
	189	await waitFor(() => {
	190	expect(sockets).toHaveLength(1);
	191	});
	192
	193	const socket = getSocket();
	194	socket.open();
	195
	196	await waitFor(() => {
	197	expect(onOpen).toHaveBeenCalledOnce();
	198	});
	199
	200	socket.close(1012, "service restart");
	201
	202	await waitFor(() => {
	203	expect(onClose).toHaveBeenCalledWith({
	204	code: 1012,
	205	reason: "service restart",
	206	});
	207	});
	208
	209	await transport.dispose();
	210	});
	211
	212	it("reconnects the websocket session without disposing the transport", async () => {
	213	const transport = createTransport("ws://localhost:3020");
	214
	215	await waitFor(() => {
	216	expect(sockets).toHaveLength(1);
	217	});
	218
	219	const firstSocket = getSocket();
	220	firstSocket.open();
	221
	222	await transport.reconnect();
	223
	224	await waitFor(() => {
	225	expect(sockets).toHaveLength(2);
	226	});
	227
	228	const secondSocket = getSocket();
	229	expect(secondSocket).not.toBe(firstSocket);
	230	expect(firstSocket.readyState).toBe(MockWebSocket.CLOSED);
	231
	232	const requestPromise = transport.request((client) =>
	233	client[WS_METHODS.serverUpsertKeybinding]({
	234	command: "terminal.toggle",
	235	key: "ctrl+k",
	236	}),
	237	);
	238
	239	secondSocket.open();
	240
	241	await waitFor(() => {
	242	expect(secondSocket.sent).toHaveLength(1);
	243	});
	244
	245	const requestMessage = JSON.parse(secondSocket.sent[0] ?? "{}") as { id: string };
	246	secondSocket.serverMessage(
	247	JSON.stringify({
	248	_tag: "Exit",
	249	requestId: requestMessage.id,
	250	exit: {
	251	_tag: "Success",
	252	value: {
	253	keybindings: [],
	254	issues: [],
	255	},
	256	},
	257	}),
	258	);
	259
	260	await expect(requestPromise).resolves.toEqual({
	261	keybindings: [],
	262	issues: [],
	263	});
	264
	265	await transport.dispose();
	266	});
	267
	268	it("sends unary RPC requests and resolves successful exits", async () => {
	269	const transport = createTransport("ws://localhost:3020");
	270
	271	const requestPromise = transport.request((client) =>
	272	client[WS_METHODS.serverUpsertKeybinding]({
	273	command: "terminal.toggle",
	274	key: "ctrl+k",
	275	}),
	276	);
	277
	278	await waitFor(() => {
	279	expect(sockets).toHaveLength(1);
	280	});
	281
	282	const socket = getSocket();
	283	socket.open();
	284
	285	await waitFor(() => {
	286	expect(socket.sent).toHaveLength(1);
	287	});
	288
	289	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as {
	290	_tag: string;
	291	id: string;
	292	payload: unknown;
	293	tag: string;
	294	};
	295	expect(requestMessage).toMatchObject({
	296	_tag: "Request",
	297	tag: WS_METHODS.serverUpsertKeybinding,
	298	payload: {
	299	command: "terminal.toggle",
	300	key: "ctrl+k",
	301	},
	302	});
	303
	304	socket.serverMessage(
	305	JSON.stringify({
	306	_tag: "Exit",
	307	requestId: requestMessage.id,
	308	exit: {
	309	_tag: "Success",
	310	value: {
	311	keybindings: [],
	312	issues: [],
	313	},
	314	},
	315	}),
	316	);
	317
	318	await expect(requestPromise).resolves.toEqual({
	319	keybindings: [],
	320	issues: [],
	321	});
	322
	323	await transport.dispose();
	324	});
	325
	326	it("delivers stream chunks to subscribers", async () => {
	327	const transport = createTransport("ws://localhost:3020");
	328	const listener = vi.fn();
	329
	330	const unsubscribe = transport.subscribe(
	331	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
	332	listener,
	333	);
	334	await waitFor(() => {
	335	expect(sockets).toHaveLength(1);
	336	});
	337
	338	const socket = getSocket();
	339	socket.open();
	340
	341	await waitFor(() => {
	342	expect(socket.sent).toHaveLength(1);
	343	});
	344
	345	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as { id: string; tag: string };
	346	expect(requestMessage.tag).toBe(WS_METHODS.subscribeServerLifecycle);
	347
	348	const welcomeEvent = {
	349	version: 1,
	350	sequence: 1,
	351	type: "welcome",
	352	payload: {
	353	environment: {
	354	environmentId: "environment-local",
	355	label: "Local environment",
	356	platform: { os: "darwin", arch: "arm64" },
	357	serverVersion: "0.0.0-test",
	358	capabilities: { repositoryIdentity: true },
	359	},
	360	cwd: "/tmp/workspace",
	361	projectName: "workspace",
	362	},
	363	};
	364
	365	socket.serverMessage(
	366	JSON.stringify({
	367	_tag: "Chunk",
	368	requestId: requestMessage.id,
	369	values: [welcomeEvent],
	370	}),
	371	);
	372
	373	await waitFor(() => {
	374	expect(listener).toHaveBeenCalledWith(welcomeEvent);
	375	});
	376
	377	unsubscribe();
	378	await transport.dispose();
	379	});
	380
	381	it("re-subscribes stream listeners after the stream exits", async () => {
	382	const transport = createTransport("ws://localhost:3020");
	383	const listener = vi.fn();
	384	const onResubscribe = vi.fn();
	385
	386	const unsubscribe = transport.subscribe(
	387	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
	388	listener,
	389	{ onResubscribe },
	390	);
	391	await waitFor(() => {
	392	expect(sockets).toHaveLength(1);
	393	});
	394
	395	const socket = getSocket();
	396	socket.open();
	397
	398	await waitFor(() => {
	399	expect(socket.sent).toHaveLength(1);
	400	});
	401
	402	const firstRequest = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
	403	socket.serverMessage(
	404	JSON.stringify({
	405	_tag: "Chunk",
	406	requestId: firstRequest.id,
	407	values: [
	408	{
	409	version: 1,
	410	sequence: 1,
	411	type: "welcome",
	412	payload: {
	413	environment: {
	414	environmentId: "environment-local",
	415	label: "Local environment",
	416	platform: { os: "darwin", arch: "arm64" },
	417	serverVersion: "0.0.0-test",
	418	capabilities: { repositoryIdentity: true },
	419	},
	420	cwd: "/tmp/one",
	421	projectName: "one",
	422	},
	423	},
	424	],
	425	}),
	426	);
	427	socket.serverMessage(
	428	JSON.stringify({
	429	_tag: "Exit",
	430	requestId: firstRequest.id,
	431	exit: {
	432	_tag: "Success",
	433	value: null,
	434	},
	435	}),
	436	);
	437
	438	await waitFor(() => {
	439	const nextRequest = socket.sent
	440	.map((message) => JSON.parse(message) as { _tag?: string; id?: string })
	441	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
	442	expect(nextRequest).toBeDefined();
	443	});
	444	expect(onResubscribe).toHaveBeenCalledOnce();
	445
	446	const secondRequest = socket.sent
	447	.map((message) => JSON.parse(message) as { _tag?: string; id?: string; tag?: string })
	448	.find(
	449	(message): message is { _tag: "Request"; id: string; tag: string } =>
	450	message._tag === "Request" && message.id !== firstRequest.id,
	451	);
	452	if (!secondRequest) {
	453	throw new Error("Expected a resubscribe request");
	454	}
	455	expect(secondRequest.tag).toBe(WS_METHODS.subscribeServerLifecycle);
	456	expect(secondRequest.id).not.toBe(firstRequest.id);
	457
	458	const secondEvent = {
	459	version: 1,
	460	sequence: 2,
	461	type: "welcome",
	462	payload: {
	463	environment: {
	464	environmentId: "environment-local",
	465	label: "Local environment",
	466	platform: { os: "darwin", arch: "arm64" },
	467	serverVersion: "0.0.0-test",
	468	capabilities: { repositoryIdentity: true },
	469	},
	470	cwd: "/tmp/two",
	471	projectName: "two",
	472	},
	473	};
	474	socket.serverMessage(
	475	JSON.stringify({
	476	_tag: "Chunk",
	477	requestId: secondRequest.id,
	478	values: [secondEvent],
	479	}),
	480	);
	481
	482	await waitFor(() => {
	483	expect(listener).toHaveBeenLastCalledWith(secondEvent);
	484	});
	485
	486	unsubscribe();
	487	await transport.dispose();
	488	});
	489
	490	it("re-subscribes live stream listeners after an explicit transport reconnect", async () => {
	491	const transport = createTransport("ws://localhost:3020");
	492	const listener = vi.fn();
	493	const onResubscribe = vi.fn();
	494
	495	const unsubscribe = transport.subscribe(
	496	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
	497	listener,
	498	{ onResubscribe },
	499	);
	500
	501	await waitFor(() => {
	502	expect(sockets).toHaveLength(1);
	503	});
	504
	505	const firstSocket = getSocket();
	506	firstSocket.open();
	507
	508	await waitFor(() => {
	509	expect(firstSocket.sent).toHaveLength(1);
	510	});
	511
	512	const firstRequest = JSON.parse(firstSocket.sent[0] ?? "{}") as { id: string };
	513	const firstEvent = {
	514	version: 1,
	515	sequence: 1,
	516	type: "welcome",
	517	payload: {
	518	environment: {
	519	environmentId: "environment-local",
	520	label: "Local environment",
	521	platform: { os: "darwin", arch: "arm64" },
	522	serverVersion: "0.0.0-test",
	523	capabilities: { repositoryIdentity: true },
	524	},
	525	cwd: "/tmp/one",
	526	projectName: "one",
	527	},
	528	};
	529
	530	firstSocket.serverMessage(
	531	JSON.stringify({
	532	_tag: "Chunk",
	533	requestId: firstRequest.id,
	534	values: [firstEvent],
	535	}),
	536	);
	537
	538	await waitFor(() => {
	539	expect(listener).toHaveBeenLastCalledWith(firstEvent);
	540	});
	541
	542	await transport.reconnect();
	543
	544	await waitFor(() => {
	545	expect(sockets).toHaveLength(2);
	546	});
	547
	548	const secondSocket = getSocket();
	549	expect(secondSocket).not.toBe(firstSocket);
	550	expect(firstSocket.readyState).toBe(MockWebSocket.CLOSED);
	551
	552	secondSocket.open();
	553
	554	await waitFor(() => {
	555	expect(secondSocket.sent).toHaveLength(1);
	556	});
	557
	558	const secondRequest = JSON.parse(secondSocket.sent[0] ?? "{}") as {
	559	id: string;
	560	tag: string;
	561	};
	562	expect(secondRequest.tag).toBe(WS_METHODS.subscribeServerLifecycle);
	563	expect(secondRequest.id).not.toBe(firstRequest.id);
	564	expect(onResubscribe).toHaveBeenCalledOnce();
	565
	566	const secondEvent = {
	567	version: 1,
	568	sequence: 2,
	569	type: "welcome",
	570	payload: {
	571	environment: {
	572	environmentId: "environment-local",
	573	label: "Local environment",
	574	platform: { os: "darwin", arch: "arm64" },
	575	serverVersion: "0.0.0-test",
	576	capabilities: { repositoryIdentity: true },
	577	},
	578	cwd: "/tmp/two",
	579	projectName: "two",
	580	},
	581	};
	582
	583	secondSocket.serverMessage(
	584	JSON.stringify({
	585	_tag: "Chunk",
	586	requestId: secondRequest.id,
	587	values: [secondEvent],
	588	}),
	589	);
	590
	591	await waitFor(() => {
	592	expect(listener).toHaveBeenLastCalledWith(secondEvent);
	593	});
	594
	595	unsubscribe();
	596	await transport.dispose();
	597	});
	598
	599	it("does not fire onResubscribe when the first stream attempt exits before any value", async () => {
	600	const transport = createTransport("ws://localhost:3020");
	601	const listener = vi.fn();
	602	const onResubscribe = vi.fn();
	603
	604	const unsubscribe = transport.subscribe(
	605	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
	606	listener,
	607	{ onResubscribe },
	608	);
	609	await waitFor(() => {
	610	expect(sockets).toHaveLength(1);
	611	});
	612
	613	const socket = getSocket();
	614	socket.open();
	615
	616	await waitFor(() => {
	617	expect(socket.sent).toHaveLength(1);
	618	});
	619
	620	const firstRequest = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
	621	socket.serverMessage(
	622	JSON.stringify({
	623	_tag: "Exit",
	624	requestId: firstRequest.id,
	625	exit: {
	626	_tag: "Success",
	627	value: null,
	628	},
	629	}),
	630	);
	631
	632	await waitFor(() => {
	633	const nextRequest = socket.sent
	634	.map((message) => JSON.parse(message) as { _tag?: string; id?: string })
	635	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
	636	expect(nextRequest).toBeDefined();
	637	});
	638	expect(onResubscribe).not.toHaveBeenCalled();
	639	expect(listener).not.toHaveBeenCalled();
	640
	641	unsubscribe();
	642	await transport.dispose();
	643	});
	644
	645	it("does not retry stream subscriptions after application-level failures", async () => {
	646	const transport = createTransport("ws://localhost:3020");
	647	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
	648	let attempts = 0;
	649
	650	const unsubscribe = transport.subscribe(
	651	() =>
	652	Stream.suspend(() => {
	653	attempts += 1;
	654	return Stream.fail(new Error("Git command failed in GitCore.statusDetails"));
	655	}),
	656	vi.fn(),
	657	{ retryDelay: 10 },
	658	);
	659
	660	await waitFor(() => {
	661	expect(sockets).toHaveLength(1);
	662	});
	663
	664	getSocket().open();
	665
	666	await waitFor(() => {
	667	expect(attempts).toBe(1);
	668	});
	669	await new Promise((resolve) => setTimeout(resolve, 50));
	670
	671	expect(attempts).toBe(1);
	672	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription failed", {
	673	error: "Git command failed in GitCore.statusDetails",
	674	});
	675	expect(warnSpy).not.toHaveBeenCalledWith(
	676	"WebSocket RPC subscription disconnected",
	677	expect.anything(),
	678	);
	679
	680	unsubscribe();
	681	await transport.dispose();
	682	});
	683
	684	it("keeps retrying stream subscriptions after transport failures", async () => {
	685	const transport = createTransport("ws://localhost:3020");
	686	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
	687	let attempts = 0;
	688
	689	const unsubscribe = transport.subscribe(
	690	() =>
	691	Stream.suspend(() => {
	692	attempts += 1;
	693	return Stream.fail(new Error("SocketCloseError: WebSocket closed"));
	694	}),
	695	vi.fn(),
	696	{ retryDelay: 10 },
	697	);
	698
	699	await waitFor(() => {
	700	expect(sockets).toHaveLength(1);
	701	});
	702
	703	getSocket().open();
	704
	705	await waitFor(() => {
	706	expect(attempts).toBeGreaterThanOrEqual(2);
	707	});
	708
	709	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription disconnected", {
	710	error: "SocketCloseError: WebSocket closed",
	711	});
	712
	713	unsubscribe();
	714	await transport.dispose();
	715	});
	716
	717	it("logs a transport disconnect once even when multiple subscriptions fail together", async () => {
	718	const transport = createTransport("ws://localhost:3020");
	719	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
	720
	721	const unsubscribeA = transport.subscribe(
	722	() => Stream.fail(new Error("SocketCloseError: 1006")),
	723	vi.fn(),
	724	{ retryDelay: 10 },
	725	);
	726	const unsubscribeB = transport.subscribe(
	727	() => Stream.fail(new Error("SocketCloseError: 1006")),
	728	vi.fn(),
	729	{ retryDelay: 10 },
	730	);
	731
	732	await waitFor(() => {
	733	expect(sockets).toHaveLength(1);
	734	});
	735
	736	getSocket().open();
	737
	738	await waitFor(() => {
	739	expect(warnSpy).toHaveBeenCalledTimes(1);
	740	});
	741	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription disconnected", {
	742	error: "SocketCloseError: 1006",
	743	});
	744
	745	unsubscribeA();
	746	unsubscribeB();
	747	await transport.dispose();
	748	});
	749
	750	it("streams finite request events without re-subscribing", async () => {
	751	const transport = createTransport("ws://localhost:3020");
	752	const listener = vi.fn();
	753
	754	await waitFor(() => {
	755	expect(sockets).toHaveLength(1);
	756	});
	757	const socket = getSocket();
	758	socket.open();
	759
	760	const requestPromise = transport.requestStream(
	761	(client) =>
	762	client[WS_METHODS.gitRunStackedAction]({
	763	actionId: "action-1",
	764	cwd: "/repo",
	765	action: "commit",
	766	}),
	767	listener,
	768	);
	769
	770	await waitFor(() => {
	771	expect(socket.sent).toHaveLength(1);
	772	});
	773
	774	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
	775	const progressEvent = {
	776	actionId: "action-1",
	777	cwd: "/repo",
	778	action: "commit",
	779	kind: "phase_started",
	780	phase: "commit",
	781	label: "Committing...",
	782	} as const;
	783
	784	socket.serverMessage(
	785	JSON.stringify({
	786	_tag: "Chunk",
	787	requestId: requestMessage.id,
	788	values: [progressEvent],
	789	}),
	790	);
	791	socket.serverMessage(
	792	JSON.stringify({
	793	_tag: "Exit",
	794	requestId: requestMessage.id,
	795	exit: {
	796	_tag: "Success",
	797	value: null,
	798	},
	799	}),
	800	);
	801
	802	await expect(requestPromise).resolves.toBeUndefined();
	803	expect(listener).toHaveBeenCalledWith(progressEvent);
	804	expect(
	805	socket.sent.filter((message) => {
	806	const parsed = JSON.parse(message) as { _tag?: string; tag?: string };
	807	return parsed._tag === "Request" && parsed.tag === WS_METHODS.gitRunStackedAction;
	808	}),
	809	).toHaveLength(1);
	810	await transport.dispose();
	811	});
	812
	813	it("closes the client scope on the transport runtime before disposing the runtime", async () => {
	814	const callOrder: string[] = [];
	815	let resolveClose!: () => void;
	816	const closePromise = new Promise<void>((resolve) => {
	817	resolveClose = resolve;
	818	});
	819
	820	const runtime = {
	821	runPromise: vi.fn(async () => {
	822	callOrder.push("close:start");
	823	await closePromise;
	824	callOrder.push("close:done");
	825	return undefined;
	826	}),
	827	dispose: vi.fn(async () => {
	828	callOrder.push("runtime:dispose");
	829	}),
	830	};
	831	const transport = {
	832	disposed: false,
	833	session: {
	834	clientScope: {} as never,
	835	runtime,
	836	},
	837	closeSession: (
	838	WsTransport.prototype as unknown as {
	839	closeSession: (session: {
	840	clientScope: unknown;
	841	runtime: { dispose: () => Promise<void>; runPromise: () => Promise<void> };
	842	}) => Promise<void>;
	843	}
	844	).closeSession,
	845	} as unknown as WsTransport;
	846
	847	void WsTransport.prototype.dispose.call(transport);
	848
	849	expect(runtime.runPromise).toHaveBeenCalledTimes(1);
	850	expect(runtime.dispose).not.toHaveBeenCalled();
	851	expect((transport as unknown as { disposed: boolean }).disposed).toBe(true);
	852
	853	resolveClose();
	854
	855	await waitFor(() => {
	856	expect(runtime.dispose).toHaveBeenCalledTimes(1);
	857	});
	858
	859	expect(callOrder).toEqual(["close:start", "close:done", "runtime:dispose"]);
	860	});
	861	});

packages/client-runtime/src/wsTransport.ts+36−2

-import { Cause, Duration, Effect, Exit, ManagedRuntime, Scope, Stream } from "effect";
+import { Cause, Duration, Effect, Exit, Layer, ManagedRuntime, Scope, Stream } from "effect";
 import { RpcClient } from "effect/unstable/rpc";
 import { isTransportConnectionErrorMessage } from "./transportError.ts";
   type WsRpcProtocolSocketUrlProvider,
 } from "./wsRpcProtocol.ts";
+export interface WsTransportOptions {
+  /**
+   * Merged into the transport `ManagedRuntime` alongside the RPC protocol layer
+   * (for example a `Tracer` layer for OTLP).
+   */
+  readonly tracingLayer?: Layer.Layer<never, never, never>;
+  /**
+   * Override protocol construction (defaults to {@link createWsRpcProtocolLayer}).
+   * The web app supplies its instrumented layer factory.
+   */
+  readonly createProtocolLayer?: (
+    url: WsRpcProtocolSocketUrlProvider,
+    lifecycleHandlers?: WsProtocolLifecycleHandlers,
+  ) => Layer.Layer<RpcClient.Protocol, never, never>;
+  /**
+   * Invoked at the start of {@link WsTransport.reconnect} before the session is replaced.
+   */
+  readonly onBeforeReconnect?: () => void;
+}
 interface SubscribeOptions {
   readonly retryDelay?: Duration.Input;
   readonly onResubscribe?: () => void;
 export class WsTransport {
   private readonly url: WsRpcProtocolSocketUrlProvider;
   private readonly lifecycleHandlers: WsProtocolLifecycleHandlers | undefined;
+  private readonly options: WsTransportOptions | undefined;
   private disposed = false;
   private hasReportedTransportDisconnect = false;
   private reconnectChain: Promise<void> = Promise.resolve();
   constructor(
     url: WsRpcProtocolSocketUrlProvider,
     lifecycleHandlers?: WsProtocolLifecycleHandlers,
+    options?: WsTransportOptions,
   ) {
     this.url = url;
     this.lifecycleHandlers = lifecycleHandlers;
+    this.options = options;
     this.session = this.createSession();
   }
         throw new Error("Transport disposed");
       }
+      try {
+        this.options?.onBeforeReconnect?.();
+      } catch {
+        // Ignore hook failures so reconnect can proceed.
+      }
       const previousSession = this.session;
       this.session = this.createSession();
       await this.closeSession(previousSession);
   }
   private createSession(): TransportSession {
-    const runtime = ManagedRuntime.make(createWsRpcProtocolLayer(this.url, this.lifecycleHandlers));
+    const protocolFactory = this.options?.createProtocolLayer ?? createWsRpcProtocolLayer;
+    const protocolLayer = protocolFactory(this.url, this.lifecycleHandlers);
+    const rootLayer = this.options?.tracingLayer
+      ? Layer.mergeAll(protocolLayer, this.options.tracingLayer)
+      : protocolLayer;
+    const runtime = ManagedRuntime.make(rootLayer);
     const clientScope = runtime.runSync(Scope.make());
     return {
       runtime,

packages/client-runtime/vitest.config.ts+8−0

@@ -0,0 +1,8 @@
	1	import { defineConfig } from "vitest/config";
	2
	3	export default defineConfig({
	4	test: {
	5	include: ["src/*/.test.ts"],
	6	environment: "node",
	7	},
	8	});