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 * as DateTime from "effect/DateTime";
 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/connection.ts+1−1

@@ -8,7 +8,7 @@ import type { · 7 unmodified lines
8	} from "@t3tools/contracts";	8	} from "@t3tools/contracts";
9	import type { KnownEnvironment } from "@t3tools/client-runtime";	9	import type { KnownEnvironment } from "@t3tools/client-runtime";
10		10
11	import type { WsRpcClient } from "~/rpc/wsRpcClient";	11	import type { WsRpcClient } from "@t3tools/client-runtime";
12		12
13	export interface EnvironmentConnection {	13	export interface EnvironmentConnection {
14	readonly kind: "primary" \| "saved";	14	readonly kind: "primary" \| "saved";

apps/web/src/environments/runtime/service.addSavedEnvironment.test.ts+2−1

@@ -109,7 +109,8 @@ vi.mock("./connection", () => ({ · 108 unmodified lines
109	createEnvironmentConnection: mockCreateEnvironmentConnection,	109	createEnvironmentConnection: mockCreateEnvironmentConnection,
110	}));	110	}));
111		111
112	vi.mock("../../rpc/wsRpcClient", () => ({	112	vi.mock("@t3tools/client-runtime", async (importOriginal) => ({
		113	...(await importOriginal<typeof import("@t3tools/client-runtime")>()),
113	createWsRpcClient: vi.fn(() => ({	114	createWsRpcClient: vi.fn(() => ({
114	server: {	115	server: {
115	getConfig: mockClientGetConfig,	116	getConfig: mockClientGetConfig,

apps/web/src/environments/runtime/service.savedEnvironments.test.ts+2−1

@@ -66,7 +66,8 @@ vi.mock("./connection", () => ({ · 65 unmodified lines
66	createEnvironmentConnection: mockCreateEnvironmentConnection,	66	createEnvironmentConnection: mockCreateEnvironmentConnection,
67	}));	67	}));
68		68
69	vi.mock("../../rpc/wsRpcClient", () => ({	69	vi.mock("@t3tools/client-runtime", async (importOriginal) => ({
		70	...(await importOriginal<typeof import("@t3tools/client-runtime")>()),
70	createWsRpcClient: mockCreateWsRpcClient,	71	createWsRpcClient: mockCreateWsRpcClient,
71	}));	72	}));
72		73

apps/web/src/environments/runtime/service.threadSubscriptions.test.ts+2−1

@@ -70,7 +70,8 @@ vi.mock("./connection", () => ({ · 69 unmodified lines
70	createEnvironmentConnection: mockCreateEnvironmentConnection,	70	createEnvironmentConnection: mockCreateEnvironmentConnection,
71	}));	71	}));
72		72
73	vi.mock("../../rpc/wsRpcClient", () => ({	73	vi.mock("@t3tools/client-runtime", async (importOriginal) => ({
		74	...(await importOriginal<typeof import("@t3tools/client-runtime")>()),
74	createWsRpcClient: mockCreateWsRpcClient,	75	createWsRpcClient: mockCreateWsRpcClient,
75	}));	76	}));
76		77

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

   type ServerConfig,
   ThreadId,
 } from "@t3tools/contracts";
+import { createWsRpcClient as createBaseWsRpcClient, type 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 type { WsProtocolCloseContext } from "../../rpc/protocol";
 import { getServerConfig } from "../../rpc/serverState";
-import { WsTransport } from "../../rpc/wsTransport";
+import { WsTransport } from "~/rpc/wsTransport";
-import { createWsRpcClient, type WsRpcClient } from "../../rpc/wsRpcClient";
 import { appendVersionMismatchHint, resolveServerConfigVersionMismatch } from "../../versionSkew";
 import {
   deriveLogicalProjectKeyFromSettings,
 } from "../../logicalProject";
 import { getClientSettings } from "~/hooks/useSettings";
 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>,
 ) {
             lastErrorAt: isoNow(),
           });
         },
-        onClose: (
+        onClose: (details: { readonly code: number; readonly reason: string }) => {
-          details: { readonly code: number; readonly reason: string },
-          context: WsProtocolCloseContext,
-        ) => {
-          if (context.intentional) {
-            return;
-          }
           setRuntimeDisconnected(
             environmentId,
             appendVersionMismatchHint(

apps/web/src/lib/gitReactQuery.test.ts+0−5

   ensureEnvironmentApi: vi.fn(),
 }));
-vi.mock("../wsRpcClient", () => ({
-  getWsRpcClient: vi.fn(),
-  getWsRpcClientForEnvironment: vi.fn(),
-}));
 import type { InfiniteData } from "@tanstack/react-query";
 import { EnvironmentId, type VcsListRefsResult } from "@t3tools/contracts";

apps/web/src/lib/gitStatusState.ts+1−1

@@ -13,7 +13,7 @@ import { · 12 unmodified lines
13	readEnvironmentConnection,	13	readEnvironmentConnection,
14	subscribeEnvironmentConnections,	14	subscribeEnvironmentConnections,
15	} from "../environments/runtime";	15	} from "../environments/runtime";
16	import type { WsRpcClient } from "~/rpc/wsRpcClient";	16	import type { WsRpcClient } from "@t3tools/client-runtime";
17		17
18	interface GitStatusState {	18	interface GitStatusState {
19	readonly data: VcsStatusResult \| null;	19	readonly data: VcsStatusResult \| null;

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

 import type { ContextMenuItem, LocalApi } from "@t3tools/contracts";
+import type { WsRpcClient } from "@t3tools/client-runtime";
 import { resetGitStatusStateForTests } from "./lib/gitStatusState";
 import { resetSourceControlDiscoveryStateForTests } from "./lib/sourceControlDiscoveryState";
   resetEnvironmentServiceForTests,
 } from "./environments/runtime";
 import { getPrimaryKnownEnvironment } from "./environments/primary";
-import { type WsRpcClient } from "./rpc/wsRpcClient";
 import { showContextMenuFallback } from "./contextMenuFallback";
 import {
   readBrowserClientSettings,

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

@@ -1,328 +0,0 @@
1	import { WsRpcGroup } from "@t3tools/contracts";
2	import * as Duration from "effect/Duration";
3	import * as Effect from "effect/Effect";
4	import * as Layer from "effect/Layer";
5	import * as Schedule from "effect/Schedule";
6	import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
7	import * as Socket from "effect/unstable/socket/Socket";
8
9	import {
10	acknowledgeRpcRequest,
11	clearAllTrackedRpcRequests,
12	trackRpcRequestSent,
13	} from "./requestLatencyState";
14	import {
15	getWsReconnectDelayMsForRetry,
16	recordWsConnectionAttempt,
17	recordWsConnectionClosed,
18	recordWsConnectionErrored,
19	recordWsConnectionOpened,
20	type WsConnectionMetadata,
21	WS_RECONNECT_MAX_RETRIES,
22	} from "./wsConnectionState";
23
24	export interface WsProtocolCloseContext {
25	readonly intentional: boolean;
26	}
27
28	export interface WsProtocolLifecycleHandlers {
29	readonly getConnectionLabel?: () => string \| null;
30	readonly getVersionMismatchHint?: () => string \| null;
31	readonly isCloseIntentional?: () => boolean;
32	readonly isActive?: () => boolean;
33	readonly onAttempt?: (socketUrl: string) => void;
34	readonly onOpen?: () => void;
35	readonly onHeartbeatPing?: () => void;
36	readonly onHeartbeatPong?: () => void;
37	readonly onHeartbeatTimeout?: () => void;
38	readonly onRequestStart?: (info: {
39	readonly id: string;
40	readonly tag: string;
41	readonly stream: boolean;
42	}) => void;
43	readonly onRequestChunk?: (info: {
44	readonly id: string;
45	readonly tag: string;
46	readonly chunkCount: number;
47	}) => void;
48	readonly onRequestExit?: (info: {
49	readonly id: string;
50	readonly tag: string;
51	readonly stream: boolean;
52	}) => void;
53	readonly onRequestInterrupt?: (info: { readonly id: string; readonly tag?: string }) => void;
54	readonly onError?: (message: string) => void;
55	readonly onClose?: (
56	details: { readonly code: number; readonly reason: string },
57	context: WsProtocolCloseContext,
58	) => void;
59	}
60
61	export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);
62	type RpcClientFactory = typeof makeWsRpcProtocolClient;
63	export type WsRpcProtocolClient =
64	RpcClientFactory extends Effect.Effect<infer Client, any, any> ? Client : never;
65	export type WsRpcProtocolSocketUrlProvider = string \| (() => Promise<string>);
66
67	function formatSocketErrorMessage(error: unknown): string {
68	if (error instanceof Error && error.message.trim().length > 0) {
69	return error.message;
70	}
71	return String(error);
72	}
73
74	function resolveWsRpcSocketUrl(rawUrl: string): string {
75	const resolved = new URL(rawUrl);
76	if (resolved.protocol !== "ws:" && resolved.protocol !== "wss:") {
77	throw new Error(`Unsupported websocket transport URL protocol: ${resolved.protocol}`);
78	}
79
80	resolved.pathname = "/ws";
81	return resolved.toString();
82	}
83
84	function resolveConnectionMetadata(handlers?: WsProtocolLifecycleHandlers): WsConnectionMetadata {
85	return {
86	connectionLabel: handlers?.getConnectionLabel?.() ?? null,
87	versionMismatchHint: handlers?.getVersionMismatchHint?.() ?? null,
88	};
89	}
90
91	type ComposedWsProtocolLifecycleHandlers = Required<
92	Pick<WsProtocolLifecycleHandlers, "isActive" \| "onAttempt" \| "onOpen" \| "onError" \| "onClose">
93	>;
94
95	function defaultLifecycleHandlers(
96	handlers?: WsProtocolLifecycleHandlers,
97	): ComposedWsProtocolLifecycleHandlers {
98	return {
99	isActive: () => true,
100	onAttempt: (socketUrl) => {
101	recordWsConnectionAttempt(socketUrl, resolveConnectionMetadata(handlers));
102	},
103	onOpen: () => {
104	recordWsConnectionOpened(resolveConnectionMetadata(handlers));
105	},
106	onError: (message) => {
107	clearAllTrackedRpcRequests();
108	recordWsConnectionErrored(message, resolveConnectionMetadata(handlers));
109	},
110	onClose: (details, context) => {
111	clearAllTrackedRpcRequests();
112	if (context.intentional) {
113	return;
114	}
115	recordWsConnectionClosed(details, resolveConnectionMetadata(handlers));
116	},
117	};
118	}
119
120	function composeLifecycleHandlers(
121	handlers?: WsProtocolLifecycleHandlers,
122	): ComposedWsProtocolLifecycleHandlers {
123	const defaults = defaultLifecycleHandlers(handlers);
124	const isActive = handlers?.isActive ?? defaults.isActive;
125
126	return {
127	isActive,
128	onAttempt: (socketUrl) => {
129	if (!isActive()) {
130	return;
131	}
132	defaults.onAttempt(socketUrl);
133	handlers?.onAttempt?.(socketUrl);
134	},
135	onOpen: () => {
136	if (!isActive()) {
137	return;
138	}
139	defaults.onOpen();
140	handlers?.onOpen?.();
141	},
142	onError: (message) => {
143	if (!isActive()) {
144	return;
145	}
146	defaults.onError(message);
147	handlers?.onError?.(message);
148	},
149	onClose: (details, context) => {
150	if (!isActive()) {
151	return;
152	}
153	defaults.onClose(details, context);
154	handlers?.onClose?.(details, context);
155	},
156	};
157	}
158
159	export function createWsRpcProtocolLayer(
160	url: WsRpcProtocolSocketUrlProvider,
161	handlers?: WsProtocolLifecycleHandlers,
162	) {
163	const lifecycle = composeLifecycleHandlers(handlers);
164	const resolvedUrl =
165	typeof url === "function"
166	? Effect.promise(() => url()).pipe(
167	Effect.map((rawUrl) => resolveWsRpcSocketUrl(rawUrl)),
168	Effect.tapError((error) =>
169	Effect.sync(() => {
170	lifecycle.onError(formatSocketErrorMessage(error));
171	}),
172	),
173	Effect.orDie,
174	)
175	: resolveWsRpcSocketUrl(url);
176
177	const trackingWebSocketConstructorLayer = Layer.succeed(
178	Socket.WebSocketConstructor,
179	(socketUrl, protocols) => {
180	lifecycle.onAttempt(socketUrl);
181	const socket = new globalThis.WebSocket(socketUrl, protocols);
182
183	socket.addEventListener(
184	"open",
185	() => {
186	lifecycle.onOpen();
187	},
188	{ once: true },
189	);
190	socket.addEventListener(
191	"error",
192	() => {
193	lifecycle.onError("Unable to connect to the T3 server WebSocket.");
194	},
195	{ once: true },
196	);
197	socket.addEventListener(
198	"close",
199	(event) => {
200	lifecycle.onClose(
201	{
202	code: event.code,
203	reason: event.reason,
204	},
205	{
206	intentional: handlers?.isCloseIntentional?.() ?? false,
207	},
208	);
209	},
210	{ once: true },
211	);
212
213	return socket;
214	},
215	);
216	const socketLayer = Socket.layerWebSocket(resolvedUrl).pipe(
217	Layer.provide(trackingWebSocketConstructorLayer),
218	);
219	const retryPolicy = Schedule.addDelay(Schedule.recurs(WS_RECONNECT_MAX_RETRIES), (retryCount) =>
220	Effect.succeed(Duration.millis(getWsReconnectDelayMsForRetry(retryCount) ?? 0)),
221	);
222	const protocolLayer = Layer.effect(
223	RpcClient.Protocol,
224	Effect.map(
225	RpcClient.makeProtocolSocket({
226	retryPolicy,
227	retryTransientErrors: true,
228	}),
229	(protocol) => ({
230	...protocol,
231	run: (clientId, writeResponse) =>
232	protocol.run(clientId, (response) => {
233	if (response._tag === "ClientProtocolError" \|\| response._tag === "Defect") {
234	clearAllTrackedRpcRequests();
235	}
236	return writeResponse(response);
237	}),
238	}),
239	),
240	);
241	const requestHooksLayer = Layer.succeed(
242	RpcClient.RequestHooks,
243	RpcClient.RequestHooks.of({
244	onRequestStart: (info) =>
245	Effect.sync(() => {
246	if (!lifecycle.isActive()) {
247	return;
248	}
249	handlers?.onRequestStart?.({
250	id: String(info.id),
251	tag: info.tag,
252	stream: info.stream,
253	});
254	trackRpcRequestSent(String(info.id), info.tag);
255	}),
256	onRequestChunk: (info) =>
257	Effect.sync(() => {
258	if (!lifecycle.isActive()) {
259	return;
260	}
261	handlers?.onRequestChunk?.({
262	id: String(info.id),
263	tag: info.tag,
264	chunkCount: info.chunkCount,
265	});
266	acknowledgeRpcRequest(String(info.id));
267	}),
268	onRequestExit: (info) =>
269	Effect.sync(() => {
270	if (!lifecycle.isActive()) {
271	return;
272	}
273	handlers?.onRequestExit?.({
274	id: String(info.id),
275	tag: info.tag,
276	stream: info.stream,
277	});
278	acknowledgeRpcRequest(String(info.id));
279	}),
280	onRequestInterrupt: (info) =>
281	Effect.sync(() => {
282	if (!lifecycle.isActive()) {
283	return;
284	}
285	handlers?.onRequestInterrupt?.({
286	id: String(info.id),
287	...(info.tag === undefined ? {} : { tag: info.tag }),
288	});
289	acknowledgeRpcRequest(String(info.id));
290	}),
291	}),
292	);
293	const connectionHooksLayer = Layer.succeed(
294	RpcClient.ConnectionHooks,
295	RpcClient.ConnectionHooks.of({
296	onConnect: Effect.void,
297	onDisconnect: Effect.void,
298	onPing: Effect.sync(() => {
299	if (lifecycle.isActive()) {
300	handlers?.onHeartbeatPing?.();
301	}
302	}),
303	onPong: Effect.sync(() => {
304	if (lifecycle.isActive()) {
305	handlers?.onHeartbeatPong?.();
306	}
307	}),
308	onPingTimeout: Effect.sync(() => {
309	if (lifecycle.isActive()) {
310	clearAllTrackedRpcRequests();
311	recordWsConnectionErrored(
312	"WebSocket heartbeat timed out.",
313	resolveConnectionMetadata(handlers),
314	);
315	handlers?.onHeartbeatTimeout?.();
316	}
317	}),
318	}),
319	);
320
321	return Layer.mergeAll(
322	protocolLayer.pipe(
323	Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson, connectionHooksLayer)),
324	),
325	requestHooksLayer,
326	connectionHooksLayer,
327	);
328	}

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

@@ -14,7 +14,7 @@ import { DEFAULT_RESOLVED_KEYBINDINGS } from "@t3tools/shared/keybindings"; · 13 unmodified lines
14	import { Atom } from "effect/unstable/reactivity";	14	import { Atom } from "effect/unstable/reactivity";
15	import { useCallback, useRef } from "react";	15	import { useCallback, useRef } from "react";
16		16
17	import type { WsRpcClient } from "./wsRpcClient";	17	import type { WsRpcClient } from "@t3tools/client-runtime";
18	import { appAtomRegistry, resetAppAtomRegistryForTests } from "./atomRegistry";	18	import { appAtomRegistry, resetAppAtomRegistryForTests } from "./atomRegistry";
19		19
20	export type ServerConfigUpdateSource = ServerConfigStreamEvent["type"];	20	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−333

@@ -1,333 +0,0 @@
1	import {
2	type GitActionProgressEvent,
3	type GitRunStackedActionInput,
4	type GitRunStackedActionResult,
5	type VcsStatusResult,
6	type VcsStatusStreamEvent,
7	type LocalApi,
8	ORCHESTRATION_WS_METHODS,
9	type ServerSettingsPatch,
10	WS_METHODS,
11	} from "@t3tools/contracts";
12	import { applyGitStatusStreamEvent } from "@t3tools/shared/git";
13	import * as Effect from "effect/Effect";
14	import * as Stream from "effect/Stream";
15
16	import { type WsRpcProtocolClient } from "./protocol";
17	import { resetWsReconnectBackoff } from "./wsConnectionState";
18	import { WsTransport } from "./wsTransport";
19
20	type RpcTag = keyof WsRpcProtocolClient & string;
21	type RpcMethod<TTag extends RpcTag> = WsRpcProtocolClient[TTag];
22	type RpcInput<TTag extends RpcTag> = Parameters<RpcMethod<TTag>>[0];
23
24	interface StreamSubscriptionOptions {
25	readonly onResubscribe?: () => void;
26	}
27
28	type RpcUnaryMethod<TTag extends RpcTag> =
29	RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<infer TSuccess, any, any>
30	? (input: RpcInput<TTag>) => Promise<TSuccess>
31	: never;
32
33	type RpcUnaryNoArgMethod<TTag extends RpcTag> =
34	RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<infer TSuccess, any, any>
35	? () => Promise<TSuccess>
36	: never;
37
38	type RpcStreamMethod<TTag extends RpcTag> =
39	RpcMethod<TTag> extends (input: any, options?: any) => Stream.Stream<infer TEvent, any, any>
40	? (listener: (event: TEvent) => void, options?: StreamSubscriptionOptions) => () => void
41	: never;
42
43	type RpcInputStreamMethod<TTag extends RpcTag> =
44	RpcMethod<TTag> extends (input: any, options?: any) => Stream.Stream<infer TEvent, any, any>
45	? (
46	input: RpcInput<TTag>,
47	listener: (event: TEvent) => void,
48	options?: StreamSubscriptionOptions,
49	) => () => void
50	: never;
51
52	interface GitRunStackedActionOptions {
53	readonly onProgress?: (event: GitActionProgressEvent) => void;
54	}
55
56	export interface WsRpcClient {
57	readonly dispose: () => Promise<void>;
58	readonly reconnect: () => Promise<void>;
59	readonly isHeartbeatFresh: () => boolean;
60	readonly terminal: {
61	readonly open: RpcUnaryMethod<typeof WS_METHODS.terminalOpen>;
62	readonly write: RpcUnaryMethod<typeof WS_METHODS.terminalWrite>;
63	readonly resize: RpcUnaryMethod<typeof WS_METHODS.terminalResize>;
64	readonly clear: RpcUnaryMethod<typeof WS_METHODS.terminalClear>;
65	readonly restart: RpcUnaryMethod<typeof WS_METHODS.terminalRestart>;
66	readonly close: RpcUnaryMethod<typeof WS_METHODS.terminalClose>;
67	readonly onEvent: RpcStreamMethod<typeof WS_METHODS.subscribeTerminalEvents>;
68	};
69	readonly projects: {
70	readonly searchEntries: RpcUnaryMethod<typeof WS_METHODS.projectsSearchEntries>;
71	readonly writeFile: RpcUnaryMethod<typeof WS_METHODS.projectsWriteFile>;
72	};
73	readonly filesystem: {
74	readonly browse: RpcUnaryMethod<typeof WS_METHODS.filesystemBrowse>;
75	};
76	readonly sourceControl: {
77	readonly lookupRepository: RpcUnaryMethod<typeof WS_METHODS.sourceControlLookupRepository>;
78	readonly cloneRepository: RpcUnaryMethod<typeof WS_METHODS.sourceControlCloneRepository>;
79	readonly publishRepository: RpcUnaryMethod<typeof WS_METHODS.sourceControlPublishRepository>;
80	};
81	readonly shell: {
82	readonly openInEditor: (input: {
83	readonly cwd: Parameters<LocalApi["shell"]["openInEditor"]>[0];
84	readonly editor: Parameters<LocalApi["shell"]["openInEditor"]>[1];
85	}) => ReturnType<LocalApi["shell"]["openInEditor"]>;
86	};
87	readonly vcs: {
88	readonly pull: RpcUnaryMethod<typeof WS_METHODS.vcsPull>;
89	readonly refreshStatus: RpcUnaryMethod<typeof WS_METHODS.vcsRefreshStatus>;
90	readonly onStatus: (
91	input: RpcInput<typeof WS_METHODS.subscribeVcsStatus>,
92	listener: (status: VcsStatusResult) => void,
93	options?: StreamSubscriptionOptions,
94	) => () => void;
95	readonly listRefs: RpcUnaryMethod<typeof WS_METHODS.vcsListRefs>;
96	readonly createWorktree: RpcUnaryMethod<typeof WS_METHODS.vcsCreateWorktree>;
97	readonly removeWorktree: RpcUnaryMethod<typeof WS_METHODS.vcsRemoveWorktree>;
98	readonly createRef: RpcUnaryMethod<typeof WS_METHODS.vcsCreateRef>;
99	readonly switchRef: RpcUnaryMethod<typeof WS_METHODS.vcsSwitchRef>;
100	readonly init: RpcUnaryMethod<typeof WS_METHODS.vcsInit>;
101	};
102	/**
103	* Git-specific workflows. Local repository mechanics live under `vcs`.
104	*/
105	readonly git: {
106	readonly runStackedAction: (
107	input: GitRunStackedActionInput,
108	options?: GitRunStackedActionOptions,
109	) => Promise<GitRunStackedActionResult>;
110	readonly resolvePullRequest: RpcUnaryMethod<typeof WS_METHODS.gitResolvePullRequest>;
111	readonly preparePullRequestThread: RpcUnaryMethod<
112	typeof WS_METHODS.gitPreparePullRequestThread
113	>;
114	};
115	readonly server: {
116	readonly getConfig: RpcUnaryNoArgMethod<typeof WS_METHODS.serverGetConfig>;
117	/**
118	* Refresh provider snapshots. Pass `{ instanceId }` to refresh a single
119	* configured instance; pass no argument (or `{}`) to refresh all.
120	*/
121	readonly refreshProviders: (
122	input?: RpcInput<typeof WS_METHODS.serverRefreshProviders>,
123	) => ReturnType<RpcUnaryMethod<typeof WS_METHODS.serverRefreshProviders>>;
124	readonly updateProvider: RpcUnaryMethod<typeof WS_METHODS.serverUpdateProvider>;
125	readonly upsertKeybinding: RpcUnaryMethod<typeof WS_METHODS.serverUpsertKeybinding>;
126	readonly removeKeybinding: RpcUnaryMethod<typeof WS_METHODS.serverRemoveKeybinding>;
127	readonly getSettings: RpcUnaryNoArgMethod<typeof WS_METHODS.serverGetSettings>;
128	readonly updateSettings: (
129	patch: ServerSettingsPatch,
130	) => ReturnType<RpcUnaryMethod<typeof WS_METHODS.serverUpdateSettings>>;
131	readonly discoverSourceControl: RpcUnaryNoArgMethod<
132	typeof WS_METHODS.serverDiscoverSourceControl
133	>;
134	readonly getTraceDiagnostics: RpcUnaryNoArgMethod<typeof WS_METHODS.serverGetTraceDiagnostics>;
135	readonly getProcessDiagnostics: RpcUnaryNoArgMethod<
136	typeof WS_METHODS.serverGetProcessDiagnostics
137	>;
138	readonly getProcessResourceHistory: RpcUnaryMethod<
139	typeof WS_METHODS.serverGetProcessResourceHistory
140	>;
141	readonly signalProcess: RpcUnaryMethod<typeof WS_METHODS.serverSignalProcess>;
142	readonly subscribeConfig: RpcStreamMethod<typeof WS_METHODS.subscribeServerConfig>;
143	readonly subscribeLifecycle: RpcStreamMethod<typeof WS_METHODS.subscribeServerLifecycle>;
144	readonly subscribeAuthAccess: RpcStreamMethod<typeof WS_METHODS.subscribeAuthAccess>;
145	};
146	readonly orchestration: {
147	readonly dispatchCommand: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.dispatchCommand>;
148	readonly getTurnDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getTurnDiff>;
149	readonly getFullThreadDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getFullThreadDiff>;
150	readonly getArchivedShellSnapshot: RpcUnaryNoArgMethod<
151	typeof ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot
152	>;
153	readonly subscribeShell: RpcStreamMethod<typeof ORCHESTRATION_WS_METHODS.subscribeShell>;
154	readonly subscribeThread: RpcInputStreamMethod<typeof ORCHESTRATION_WS_METHODS.subscribeThread>;
155	};
156	}
157
158	export function createWsRpcClient(transport: WsTransport): WsRpcClient {
159	return {
160	dispose: () => transport.dispose(),
161	reconnect: async () => {
162	resetWsReconnectBackoff();
163	await transport.reconnect();
164	},
165	isHeartbeatFresh: () => transport.isHeartbeatFresh(),
166	terminal: {
167	open: (input) => transport.request((client) => client[WS_METHODS.terminalOpen](input)),
168	attach: (input, listener, options) =>
169	transport.subscribe(
170	(client) => client[WS_METHODS.terminalAttach](input),
171	listener,
172	options,
173	),
174	write: (input) => transport.request((client) => client[WS_METHODS.terminalWrite](input)),
175	resize: (input) => transport.request((client) => client[WS_METHODS.terminalResize](input)),
176	clear: (input) => transport.request((client) => client[WS_METHODS.terminalClear](input)),
177	restart: (input) => transport.request((client) => client[WS_METHODS.terminalRestart](input)),
178	close: (input) => transport.request((client) => client[WS_METHODS.terminalClose](input)),
179	onMetadata: (listener, options) =>
180	transport.subscribe(
181	(client) => client[WS_METHODS.subscribeTerminalMetadata]({}),
182	listener,
183	options,
184	),
185	},
186	projects: {
187	searchEntries: (input) =>
188	transport.request((client) => client[WS_METHODS.projectsSearchEntries](input)),
189	writeFile: (input) =>
190	transport.request((client) => client[WS_METHODS.projectsWriteFile](input)),
191	},
192	filesystem: {
193	browse: (input) => transport.request((client) => client[WS_METHODS.filesystemBrowse](input)),
194	},
195	sourceControl: {
196	lookupRepository: (input) =>
197	transport.request((client) => client[WS_METHODS.sourceControlLookupRepository](input)),
198	cloneRepository: (input) =>
199	transport.request((client) => client[WS_METHODS.sourceControlCloneRepository](input)),
200	publishRepository: (input) =>
201	transport.request((client) => client[WS_METHODS.sourceControlPublishRepository](input)),
202	},
203	shell: {
204	openInEditor: (input) =>
205	transport.request((client) => client[WS_METHODS.shellOpenInEditor](input)),
206	},
207	vcs: {
208	pull: (input) => transport.request((client) => client[WS_METHODS.vcsPull](input)),
209	refreshStatus: (input) =>
210	transport.request((client) => client[WS_METHODS.vcsRefreshStatus](input)),
211	onStatus: (input, listener, options) => {
212	let current: VcsStatusResult \| null = null;
213	return transport.subscribe(
214	(client) => client[WS_METHODS.subscribeVcsStatus](input),
215	(event: VcsStatusStreamEvent) => {
216	current = applyGitStatusStreamEvent(current, event);
217	listener(current);
218	},
219	options,
220	);
221	},
222	listRefs: (input) => transport.request((client) => client[WS_METHODS.vcsListRefs](input)),
223	createWorktree: (input) =>
224	transport.request((client) => client[WS_METHODS.vcsCreateWorktree](input)),
225	removeWorktree: (input) =>
226	transport.request((client) => client[WS_METHODS.vcsRemoveWorktree](input)),
227	createRef: (input) => transport.request((client) => client[WS_METHODS.vcsCreateRef](input)),
228	switchRef: (input) => transport.request((client) => client[WS_METHODS.vcsSwitchRef](input)),
229	init: (input) => transport.request((client) => client[WS_METHODS.vcsInit](input)),
230	},
231	git: {
232	runStackedAction: async (input, options) => {
233	let result: GitRunStackedActionResult \| null = null;
234
235	await transport.requestStream(
236	(client) => client[WS_METHODS.gitRunStackedAction](input),
237	(event) => {
238	options?.onProgress?.(event);
239	if (event.kind === "action_finished") {
240	result = event.result;
241	}
242	},
243	);
244
245	if (result) {
246	return result;
247	}
248
249	throw new Error("Git action stream completed without a final result.");
250	},
251	resolvePullRequest: (input) =>
252	transport.request((client) => client[WS_METHODS.gitResolvePullRequest](input)),
253	preparePullRequestThread: (input) =>
254	transport.request((client) => client[WS_METHODS.gitPreparePullRequestThread](input)),
255	},
256	server: {
257	getConfig: () => transport.request((client) => client[WS_METHODS.serverGetConfig]({})),
258	refreshProviders: (input) =>
259	transport.request((client) => client[WS_METHODS.serverRefreshProviders](input ?? {})),
260	updateProvider: (input) =>
261	transport.request((client) => client[WS_METHODS.serverUpdateProvider](input)),
262	upsertKeybinding: (input) =>
263	transport.request((client) => client[WS_METHODS.serverUpsertKeybinding](input)),
264	removeKeybinding: (input) =>
265	transport.request((client) => client[WS_METHODS.serverRemoveKeybinding](input)),
266	getSettings: () => transport.request((client) => client[WS_METHODS.serverGetSettings]({})),
267	updateSettings: (patch) =>
268	transport.request((client) => client[WS_METHODS.serverUpdateSettings]({ patch })),
269	discoverSourceControl: () =>
270	transport.request((client) => client[WS_METHODS.serverDiscoverSourceControl]({})),
271	getTraceDiagnostics: () =>
272	transport.request((client) =>
273	client[WS_METHODS.serverGetTraceDiagnostics]({}).pipe(Effect.withTracerEnabled(false)),
274	),
275	getProcessDiagnostics: () =>
276	transport.request((client) =>
277	client[WS_METHODS.serverGetProcessDiagnostics]({}).pipe(Effect.withTracerEnabled(false)),
278	),
279	getProcessResourceHistory: (input) =>
280	transport.request((client) =>
281	client[WS_METHODS.serverGetProcessResourceHistory](input).pipe(
282	Effect.withTracerEnabled(false),
283	),
284	),
285	signalProcess: (input) =>
286	transport.request((client) =>
287	client[WS_METHODS.serverSignalProcess](input).pipe(Effect.withTracerEnabled(false)),
288	),
289	subscribeConfig: (listener, options) =>
290	transport.subscribe(
291	(client) => client[WS_METHODS.subscribeServerConfig]({}),
292	listener,
293	options,
294	),
295	subscribeLifecycle: (listener, options) =>
296	transport.subscribe(
297	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
298	listener,
299	options,
300	),
301	subscribeAuthAccess: (listener, options) =>
302	transport.subscribe(
303	(client) => client[WS_METHODS.subscribeAuthAccess]({}),
304	listener,
305	options,
306	),
307	},
308	orchestration: {
309	dispatchCommand: (input) =>
310	transport.request((client) => client[ORCHESTRATION_WS_METHODS.dispatchCommand](input)),
311	getTurnDiff: (input) =>
312	transport.request((client) => client[ORCHESTRATION_WS_METHODS.getTurnDiff](input)),
313	getFullThreadDiff: (input) =>
314	transport.request((client) => client[ORCHESTRATION_WS_METHODS.getFullThreadDiff](input)),
315	getArchivedShellSnapshot: () =>
316	transport.request((client) =>
317	client[ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]({}),
318	),
319	subscribeShell: (listener, options) =>
320	transport.subscribe(
321	(client) => client[ORCHESTRATION_WS_METHODS.subscribeShell]({}),
322	listener,
323	options,
324	),
325	subscribeThread: (input, listener, options) =>
326	transport.subscribe(
327	(client) => client[ORCHESTRATION_WS_METHODS.subscribeThread](input),
328	listener,
329	options,
330	),
331	},
332	};
333	}

apps/web/src/rpc/wsTransport.test.ts+2−828

@@ -1,6 +1,4 @@
1	import { DEFAULT_SERVER_SETTINGS, ServerSettings, WS_METHODS } from "@t3tools/contracts";	1	import { DEFAULT_SERVER_SETTINGS, WS_METHODS } from "@t3tools/contracts";
2	import * as Schema from "effect/Schema";
3	import * as Stream from "effect/Stream";
4	import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";	2	import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
5		3
6	import {	4	import {
@@ -155,51 +153,7 @@ afterEach(async () => { · 148 unmodified lines
155	vi.restoreAllMocks();	153	vi.restoreAllMocks();
156	});	154	});
157		155
158	describe("WsTransport", () => {	156	describe("WsTransport (web instrumentation)", () => {
159	it("normalizes root websocket urls to /ws and preserves query params", async () => {
160	const transport = createTransport("ws://localhost:3020/?token=secret-token");
161
162	await waitFor(() => {
163	expect(sockets).toHaveLength(1);
164	});
165
166	expect(getSocket().url).toBe("ws://localhost:3020/ws?token=secret-token");
167	await transport.dispose();
168	});
169
170	it("uses an explicit secure websocket base url", async () => {
171	const transport = createTransport("wss://app.example.com");
172
173	await waitFor(() => {
174	expect(sockets).toHaveLength(1);
175	});
176
177	expect(getSocket().url).toBe("wss://app.example.com/ws");
178	await transport.dispose();
179	});
180
181	it("uses an explicit insecure websocket base url for remote backends", async () => {
182	const transport = createTransport("ws://192.168.1.44:3773");
183
184	await waitFor(() => {
185	expect(sockets).toHaveLength(1);
186	});
187
188	expect(getSocket().url).toBe("ws://192.168.1.44:3773/ws");
189	await transport.dispose();
190	});
191
192	it("supports async websocket url providers", async () => {
193	const transport = createTransport(async () => "wss://remote.example.com/?wsToken=dynamic");
194
195	await waitFor(() => {
196	expect(sockets).toHaveLength(1);
197	});
198
199	expect(getSocket().url).toBe("wss://remote.example.com/ws?wsToken=dynamic");
200	await transport.dispose();
201	});
202
203	it("tracks initial connection failures for the app error state", async () => {	157	it("tracks initial connection failures for the app error state", async () => {
204	const transport = createTransport("ws://localhost:3020");	158	const transport = createTransport("ws://localhost:3020");
205		159
@@ -306,192 +260,6 @@ describe("WsTransport", () => { · 100 unmodified lines
306	await transport.dispose();	260	await transport.dispose();
307	});	261	});
308		262
309	it("does not report an intentional dispose as a reconnectable disconnect", async () => {
310	const onClose = vi.fn();
311	const transport = createTransport("ws://localhost:3020", {
312	getConnectionLabel: () => "Remote Mac mini",
313	onClose,
314	});
315
316	await waitFor(() => {
317	expect(sockets).toHaveLength(1);
318	});
319
320	const socket = getSocket();
321	socket.open();
322
323	await waitFor(() => {
324	expect(getWsConnectionStatus()).toMatchObject({
325	connectionLabel: "Remote Mac mini",
326	hasConnected: true,
327	phase: "connected",
328	});
329	});
330
331	await transport.dispose();
332
333	expect(onClose).not.toHaveBeenCalled();
334	expect(getWsConnectionStatus()).toMatchObject({
335	connectionLabel: "Remote Mac mini",
336	phase: "connected",
337	reconnectPhase: "idle",
338	});
339	});
340
341	it("reconnects the websocket session without disposing the transport", async () => {
342	const transport = createTransport("ws://localhost:3020");
343
344	await waitFor(() => {
345	expect(sockets).toHaveLength(1);
346	});
347
348	const firstSocket = getSocket();
349	firstSocket.open();
350
351	await waitFor(() => {
352	expect(getWsConnectionStatus()).toMatchObject({
353	hasConnected: true,
354	phase: "connected",
355	});
356	});
357
358	await transport.reconnect();
359
360	await waitFor(() => {
361	expect(sockets).toHaveLength(2);
362	});
363
364	const secondSocket = getSocket();
365	expect(secondSocket).not.toBe(firstSocket);
366	expect(firstSocket.readyState).toBe(MockWebSocket.CLOSED);
367	expect(getWsConnectionStatus()).toMatchObject({
368	closeCode: null,
369	closeReason: null,
370	phase: "connecting",
371	});
372
373	const requestPromise = transport.request((client) =>
374	client[WS_METHODS.serverUpsertKeybinding]({
375	command: "terminal.toggle",
376	key: "ctrl+k",
377	}),
378	);
379
380	secondSocket.open();
381
382	await waitFor(() => {
383	expect(secondSocket.sent).toHaveLength(1);
384	});
385
386	const requestMessage = JSON.parse(secondSocket.sent[0] ?? "{}") as { id: string };
387	secondSocket.serverMessage(
388	JSON.stringify({
389	_tag: "Exit",
390	requestId: requestMessage.id,
391	exit: {
392	_tag: "Success",
393	value: {
394	keybindings: [],
395	issues: [],
396	},
397	},
398	}),
399	);
400
401	await expect(requestPromise).resolves.toEqual({
402	keybindings: [],
403	issues: [],
404	});
405
406	await transport.dispose();
407	});
408
409	it("ignores stale socket lifecycle events after a reconnect starts a new session", async () => {
410	const onClose = vi.fn();
411	const transport = createTransport("ws://localhost:3020", { onClose });
412
413	await waitFor(() => {
414	expect(sockets).toHaveLength(1);
415	});
416
417	const firstSocket = getSocket();
418	firstSocket.open();
419
420	await waitFor(() => {
421	expect(getWsConnectionStatus()).toMatchObject({
422	hasConnected: true,
423	phase: "connected",
424	});
425	});
426
427	await transport.reconnect();
428
429	await waitFor(() => {
430	expect(sockets).toHaveLength(2);
431	});
432
433	expect(onClose).not.toHaveBeenCalled();
434	expect(getWsConnectionStatus()).toMatchObject({
435	closeCode: null,
436	closeReason: null,
437	phase: "connecting",
438	});
439
440	const secondSocket = getSocket();
441	secondSocket.open();
442
443	await waitFor(() => {
444	expect(getWsConnectionStatus()).toMatchObject({
445	phase: "connected",
446	});
447	});
448
449	firstSocket.close(1006, "stale close");
450
451	expect(onClose).not.toHaveBeenCalled();
452	expect(getWsConnectionStatus()).toMatchObject({
453	closeCode: null,
454	closeReason: null,
455	phase: "connected",
456	});
457
458	await transport.dispose();
459	});
460
461	it("ignores stale socket close events after the transport is disposed", async () => {
462	const onClose = vi.fn();
463	const transport = createTransport("ws://localhost:3020", { onClose });
464
465	await waitFor(() => {
466	expect(sockets).toHaveLength(1);
467	});
468
469	const firstSocket = getSocket();
470	firstSocket.open();
471
472	await waitFor(() => {
473	expect(getWsConnectionStatus()).toMatchObject({
474	hasConnected: true,
475	phase: "connected",
476	});
477	});
478
479	await transport.reconnect();
480
481	await waitFor(() => {
482	expect(sockets).toHaveLength(2);
483	});
484
485	const secondSocket = getSocket();
486	secondSocket.open();
487	await transport.dispose();
488
489	onClose.mockClear();
490	firstSocket.close(1006, "stale close after dispose");
491
492	expect(onClose).not.toHaveBeenCalled();
493	});
494
495	it("marks unary requests as slow until the first server ack arrives", async () => {	263	it("marks unary requests as slow until the first server ack arrives", async () => {
496	const slowAckThresholdMs = 25;	264	const slowAckThresholdMs = 25;
497	setSlowRpcAckThresholdMsForTests(slowAckThresholdMs);	265	setSlowRpcAckThresholdMsForTests(slowAckThresholdMs);
@@ -598,600 +366,6 @@ describe("WsTransport", () => { · 100 unmodified lines
598	await transport.dispose();	366	await transport.dispose();
599	}, 5_000);	367	}, 5_000);
600		368
601	it("sends unary RPC requests and resolves successful exits", async () => {
602	const transport = createTransport("ws://localhost:3020");
603
604	const requestPromise = transport.request((client) =>
605	client[WS_METHODS.serverUpsertKeybinding]({
606	command: "terminal.toggle",
607	key: "ctrl+k",
608	}),
609	);
610
611	await waitFor(() => {
612	expect(sockets).toHaveLength(1);
613	});
614
615	const socket = getSocket();
616	socket.open();
617
618	await waitFor(() => {
619	expect(socket.sent).toHaveLength(1);
620	});
621
622	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as {
623	_tag: string;
624	id: string;
625	payload: unknown;
626	tag: string;
627	};
628	expect(requestMessage).toMatchObject({
629	_tag: "Request",
630	tag: WS_METHODS.serverUpsertKeybinding,
631	payload: {
632	command: "terminal.toggle",
633	key: "ctrl+k",
634	},
635	});
636
637	socket.serverMessage(
638	JSON.stringify({
639	_tag: "Exit",
640	requestId: requestMessage.id,
641	exit: {
642	_tag: "Success",
643	value: {
644	keybindings: [],
645	issues: [],
646	},
647	},
648	}),
649	);
650
651	await expect(requestPromise).resolves.toEqual({
652	keybindings: [],
653	issues: [],
654	});
655
656	await transport.dispose();
657	});
658
659	it("delivers stream chunks to subscribers", async () => {
660	const transport = createTransport("ws://localhost:3020");
661	const listener = vi.fn();
662
663	const unsubscribe = transport.subscribe(
664	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
665	listener,
666	);
667	await waitFor(() => {
668	expect(sockets).toHaveLength(1);
669	});
670
671	const socket = getSocket();
672	socket.open();
673
674	await waitFor(() => {
675	expect(socket.sent).toHaveLength(1);
676	});
677
678	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as { id: string; tag: string };
679	expect(requestMessage.tag).toBe(WS_METHODS.subscribeServerLifecycle);
680
681	const welcomeEvent = {
682	version: 1,
683	sequence: 1,
684	type: "welcome",
685	payload: {
686	environment: {
687	environmentId: "environment-local",
688	label: "Local environment",
689	platform: { os: "darwin", arch: "arm64" },
690	serverVersion: "0.0.0-test",
691	capabilities: { repositoryIdentity: true },
692	},
693	cwd: "/tmp/workspace",
694	projectName: "workspace",
695	},
696	};
697
698	socket.serverMessage(
699	JSON.stringify({
700	_tag: "Chunk",
701	requestId: requestMessage.id,
702	values: [welcomeEvent],
703	}),
704	);
705
706	await waitFor(() => {
707	expect(listener).toHaveBeenCalledWith(welcomeEvent);
708	});
709
710	unsubscribe();
711	await transport.dispose();
712	});
713
714	it("re-subscribes stream listeners after the stream exits", async () => {
715	const transport = createTransport("ws://localhost:3020");
716	const listener = vi.fn();
717	const onResubscribe = vi.fn();
718
719	const unsubscribe = transport.subscribe(
720	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
721	listener,
722	{ onResubscribe },
723	);
724	await waitFor(() => {
725	expect(sockets).toHaveLength(1);
726	});
727
728	const socket = getSocket();
729	socket.open();
730
731	await waitFor(() => {
732	expect(socket.sent).toHaveLength(1);
733	});
734
735	const firstRequest = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
736	socket.serverMessage(
737	JSON.stringify({
738	_tag: "Chunk",
739	requestId: firstRequest.id,
740	values: [
741	{
742	version: 1,
743	sequence: 1,
744	type: "welcome",
745	payload: {
746	environment: {
747	environmentId: "environment-local",
748	label: "Local environment",
749	platform: { os: "darwin", arch: "arm64" },
750	serverVersion: "0.0.0-test",
751	capabilities: { repositoryIdentity: true },
752	},
753	cwd: "/tmp/one",
754	projectName: "one",
755	},
756	},
757	],
758	}),
759	);
760	socket.serverMessage(
761	JSON.stringify({
762	_tag: "Exit",
763	requestId: firstRequest.id,
764	exit: {
765	_tag: "Success",
766	value: null,
767	},
768	}),
769	);
770
771	await waitFor(() => {
772	const nextRequest = socket.sent
773	.map((message) => JSON.parse(message) as { _tag?: string; id?: string })
774	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
775	expect(nextRequest).toBeDefined();
776	});
777	expect(onResubscribe).toHaveBeenCalledOnce();
778
779	const secondRequest = socket.sent
780	.map((message) => JSON.parse(message) as { _tag?: string; id?: string; tag?: string })
781	.find(
782	(message): message is { _tag: "Request"; id: string; tag: string } =>
783	message._tag === "Request" && message.id !== firstRequest.id,
784	);
785	if (!secondRequest) {
786	throw new Error("Expected a resubscribe request");
787	}
788	expect(secondRequest.tag).toBe(WS_METHODS.subscribeServerLifecycle);
789	expect(secondRequest.id).not.toBe(firstRequest.id);
790
791	const secondEvent = {
792	version: 1,
793	sequence: 2,
794	type: "welcome",
795	payload: {
796	environment: {
797	environmentId: "environment-local",
798	label: "Local environment",
799	platform: { os: "darwin", arch: "arm64" },
800	serverVersion: "0.0.0-test",
801	capabilities: { repositoryIdentity: true },
802	},
803	cwd: "/tmp/two",
804	projectName: "two",
805	},
806	};
807	socket.serverMessage(
808	JSON.stringify({
809	_tag: "Chunk",
810	requestId: secondRequest.id,
811	values: [secondEvent],
812	}),
813	);
814
815	await waitFor(() => {
816	expect(listener).toHaveBeenLastCalledWith(secondEvent);
817	});
818
819	unsubscribe();
820	await transport.dispose();
821	});
822
823	it("re-subscribes live stream listeners after an explicit transport reconnect", async () => {
824	const transport = createTransport("ws://localhost:3020");
825	const listener = vi.fn();
826	const onResubscribe = vi.fn();
827
828	const unsubscribe = transport.subscribe(
829	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
830	listener,
831	{ onResubscribe },
832	);
833
834	await waitFor(() => {
835	expect(sockets).toHaveLength(1);
836	});
837
838	const firstSocket = getSocket();
839	firstSocket.open();
840
841	await waitFor(() => {
842	expect(firstSocket.sent).toHaveLength(1);
843	});
844
845	const firstRequest = JSON.parse(firstSocket.sent[0] ?? "{}") as { id: string };
846	const firstEvent = {
847	version: 1,
848	sequence: 1,
849	type: "welcome",
850	payload: {
851	environment: {
852	environmentId: "environment-local",
853	label: "Local environment",
854	platform: { os: "darwin", arch: "arm64" },
855	serverVersion: "0.0.0-test",
856	capabilities: { repositoryIdentity: true },
857	},
858	cwd: "/tmp/one",
859	projectName: "one",
860	},
861	};
862
863	firstSocket.serverMessage(
864	JSON.stringify({
865	_tag: "Chunk",
866	requestId: firstRequest.id,
867	values: [firstEvent],
868	}),
869	);
870
871	await waitFor(() => {
872	expect(listener).toHaveBeenLastCalledWith(firstEvent);
873	});
874
875	await transport.reconnect();
876
877	await waitFor(() => {
878	expect(sockets).toHaveLength(2);
879	});
880
881	const secondSocket = getSocket();
882	expect(secondSocket).not.toBe(firstSocket);
883	expect(firstSocket.readyState).toBe(MockWebSocket.CLOSED);
884
885	secondSocket.open();
886
887	await waitFor(() => {
888	expect(secondSocket.sent).toHaveLength(1);
889	});
890
891	const secondRequest = JSON.parse(secondSocket.sent[0] ?? "{}") as {
892	id: string;
893	tag: string;
894	};
895	expect(secondRequest.tag).toBe(WS_METHODS.subscribeServerLifecycle);
896	expect(secondRequest.id).not.toBe(firstRequest.id);
897	expect(onResubscribe).toHaveBeenCalledOnce();
898
899	const secondEvent = {
900	version: 1,
901	sequence: 2,
902	type: "welcome",
903	payload: {
904	environment: {
905	environmentId: "environment-local",
906	label: "Local environment",
907	platform: { os: "darwin", arch: "arm64" },
908	serverVersion: "0.0.0-test",
909	capabilities: { repositoryIdentity: true },
910	},
911	cwd: "/tmp/two",
912	projectName: "two",
913	},
914	};
915
916	secondSocket.serverMessage(
917	JSON.stringify({
918	_tag: "Chunk",
919	requestId: secondRequest.id,
920	values: [secondEvent],
921	}),
922	);
923
924	await waitFor(() => {
925	expect(listener).toHaveBeenLastCalledWith(secondEvent);
926	});
927
928	unsubscribe();
929	await transport.dispose();
930	});
931
932	it("does not fire onResubscribe when the first stream attempt exits before any value", async () => {
933	const transport = createTransport("ws://localhost:3020");
934	const listener = vi.fn();
935	const onResubscribe = vi.fn();
936
937	const unsubscribe = transport.subscribe(
938	(client) => client[WS_METHODS.subscribeServerLifecycle]({}),
939	listener,
940	{ onResubscribe },
941	);
942	await waitFor(() => {
943	expect(sockets).toHaveLength(1);
944	});
945
946	const socket = getSocket();
947	socket.open();
948
949	await waitFor(() => {
950	expect(socket.sent).toHaveLength(1);
951	});
952
953	const firstRequest = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
954	socket.serverMessage(
955	JSON.stringify({
956	_tag: "Exit",
957	requestId: firstRequest.id,
958	exit: {
959	_tag: "Success",
960	value: null,
961	},
962	}),
963	);
964
965	await waitFor(() => {
966	const nextRequest = socket.sent
967	.map((message) => JSON.parse(message) as { _tag?: string; id?: string })
968	.find((message) => message._tag === "Request" && message.id !== firstRequest.id);
969	expect(nextRequest).toBeDefined();
970	});
971	expect(onResubscribe).not.toHaveBeenCalled();
972	expect(listener).not.toHaveBeenCalled();
973
974	unsubscribe();
975	await transport.dispose();
976	});
977
978	it("does not retry stream subscriptions after application-level failures", async () => {
979	const transport = createTransport("ws://localhost:3020");
980	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
981	let attempts = 0;
982
983	const unsubscribe = transport.subscribe(
984	() =>
985	Stream.suspend(() => {
986	attempts += 1;
987	return Stream.fail(new Error("Git command failed in GitCore.statusDetails"));
988	}),
989	vi.fn(),
990	{ retryDelay: 10 },
991	);
992
993	await waitFor(() => {
994	expect(sockets).toHaveLength(1);
995	});
996
997	getSocket().open();
998
999	await waitFor(() => {
1000	expect(attempts).toBe(1);
1001	});
1002	await new Promise((resolve) => setTimeout(resolve, 50));
1003
1004	expect(attempts).toBe(1);
1005	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription failed", {
1006	error: "Git command failed in GitCore.statusDetails",
1007	});
1008	expect(warnSpy).not.toHaveBeenCalledWith(
1009	"WebSocket RPC subscription disconnected",
1010	expect.anything(),
1011	);
1012
1013	unsubscribe();
1014	await transport.dispose();
1015	});
1016
1017	it("keeps retrying stream subscriptions after transport failures", async () => {
1018	const transport = createTransport("ws://localhost:3020");
1019	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
1020	let attempts = 0;
1021
1022	const unsubscribe = transport.subscribe(
1023	() =>
1024	Stream.suspend(() => {
1025	attempts += 1;
1026	return Stream.fail(new Error("SocketCloseError: WebSocket closed"));
1027	}),
1028	vi.fn(),
1029	{ retryDelay: 10 },
1030	);
1031
1032	await waitFor(() => {
1033	expect(sockets).toHaveLength(1);
1034	});
1035
1036	getSocket().open();
1037
1038	await waitFor(() => {
1039	expect(attempts).toBeGreaterThanOrEqual(2);
1040	});
1041
1042	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription disconnected", {
1043	error: "SocketCloseError: WebSocket closed",
1044	});
1045
1046	unsubscribe();
1047	await transport.dispose();
1048	});
1049
1050	it("logs a transport disconnect once even when multiple subscriptions fail together", async () => {
1051	const transport = createTransport("ws://localhost:3020");
1052	const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
1053
1054	const unsubscribeA = transport.subscribe(
1055	() => Stream.fail(new Error("SocketCloseError: 1006")),
1056	vi.fn(),
1057	{ retryDelay: 10 },
1058	);
1059	const unsubscribeB = transport.subscribe(
1060	() => Stream.fail(new Error("SocketCloseError: 1006")),
1061	vi.fn(),
1062	{ retryDelay: 10 },
1063	);
1064
1065	await waitFor(() => {
1066	expect(sockets).toHaveLength(1);
1067	});
1068
1069	getSocket().open();
1070
1071	await waitFor(() => {
1072	expect(warnSpy).toHaveBeenCalledTimes(1);
1073	});
1074	expect(warnSpy).toHaveBeenCalledWith("WebSocket RPC subscription disconnected", {
1075	error: "SocketCloseError: 1006",
1076	});
1077
1078	unsubscribeA();
1079	unsubscribeB();
1080	await transport.dispose();
1081	});
1082
1083	it("streams finite request events without re-subscribing", async () => {
1084	const transport = createTransport("ws://localhost:3020");
1085	const listener = vi.fn();
1086
1087	await waitFor(() => {
1088	expect(sockets).toHaveLength(1);
1089	});
1090	const socket = getSocket();
1091	socket.open();
1092
1093	const requestPromise = transport.requestStream(
1094	(client) =>
1095	client[WS_METHODS.gitRunStackedAction]({
1096	actionId: "action-1",
1097	cwd: "/repo",
1098	action: "commit",
1099	}),
1100	listener,
1101	);
1102
1103	await waitFor(() => {
1104	expect(socket.sent).toHaveLength(1);
1105	});
1106
1107	const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as { id: string };
1108	const progressEvent = {
1109	actionId: "action-1",
1110	cwd: "/repo",
1111	action: "commit",
1112	kind: "phase_started",
1113	phase: "commit",
1114	label: "Committing...",
1115	} as const;
1116
1117	socket.serverMessage(
1118	JSON.stringify({
1119	_tag: "Chunk",
1120	requestId: requestMessage.id,
1121	values: [progressEvent],
1122	}),
1123	);
1124	socket.serverMessage(
1125	JSON.stringify({
1126	_tag: "Exit",
1127	requestId: requestMessage.id,
1128	exit: {
1129	_tag: "Success",
1130	value: null,
1131	},
1132	}),
1133	);
1134
1135	await expect(requestPromise).resolves.toBeUndefined();
1136	expect(listener).toHaveBeenCalledWith(progressEvent);
1137	expect(
1138	socket.sent.filter((message) => {
1139	const parsed = JSON.parse(message) as { _tag?: string; tag?: string };
1140	return parsed._tag === "Request" && parsed.tag === WS_METHODS.gitRunStackedAction;
1141	}),
1142	).toHaveLength(1);
1143	await transport.dispose();
1144	});
1145
1146	it("closes the client scope on the transport runtime before disposing the runtime", async () => {
1147	const callOrder: string[] = [];
1148	let resolveClose!: () => void;
1149	const closePromise = new Promise<void>((resolve) => {
1150	resolveClose = resolve;
1151	});
1152
1153	const runtime = {
1154	runPromise: vi.fn(async () => {
1155	callOrder.push("close:start");
1156	await closePromise;
1157	callOrder.push("close:done");
1158	return undefined;
1159	}),
1160	dispose: vi.fn(async () => {
1161	callOrder.push("runtime:dispose");
1162	}),
1163	};
1164	const transport = {
1165	disposed: false,
1166	session: {
1167	clientScope: {} as never,
1168	runtime,
1169	},
1170	closeSession: (
1171	WsTransport.prototype as unknown as {
1172	closeSession: (session: {
1173	clientScope: unknown;
1174	runtime: { dispose: () => Promise<void>; runPromise: () => Promise<void> };
1175	}) => Promise<void>;
1176	}
1177	).closeSession,
1178	} as unknown as WsTransport;
1179
1180	void WsTransport.prototype.dispose.call(transport);
1181
1182	expect(runtime.runPromise).toHaveBeenCalledTimes(1);
1183	expect(runtime.dispose).not.toHaveBeenCalled();
1184	expect((transport as unknown as { disposed: boolean }).disposed).toBe(true);
1185
1186	resolveClose();
1187
1188	await waitFor(() => {
1189	expect(runtime.dispose).toHaveBeenCalledTimes(1);
1190	});
1191
1192	expect(callOrder).toEqual(["close:start", "close:done", "runtime:dispose"]);
1193	});
1194
1195	it("propagates OTLP trace ids for ws transport requests when client tracing is enabled", async () => {	369	it("propagates OTLP trace ids for ws transport requests when client tracing is enabled", async () => {
1196	await configureClientTracing({	370	await configureClientTracing({
1197	exportIntervalMs: 10,	371	exportIntervalMs: 10,

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

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