src/vs/server/node/agentHostChannel.ts
304 LOC · 195 covered · 109 uncovered · 40 ranges · 5 concepts · 5 introducers · 3 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
>
/*---------------------------------------------------------------------------------------------
agentHostChannel.ts ×20
2
>
* Copyright (c) Microsoft Corporation. All rights reserved.
3
>
* Licensed under the MIT License. See License.txt in the project root for license information.
4
>
*--------------------------------------------------------------------------------------------*/
5
>
6
>
// Server-side IPC channel that proxies the agent host protocol from a
7
>
// renderer to the agent host process running on the server. For each
8
>
// renderer client (identified by IPC context) the channel opens its own
9
>
// AHP-over-WebSocket connection upstream and pipes raw JSON frames.
10
>
//
11
>
// The renderer-side counterpart is `AgentHostIpcChannelTransport` in
12
>
// `src/vs/platform/agentHost/browser/`. Together they reuse the existing
13
>
// `RemoteAgentHostProtocolClient` over IPC instead of a raw WebSocket.
14
>
15
>
import { Emitter, Event } from '../../base/common/event.js';
16
>
import { Disposable, IDisposable } from '../../base/common/lifecycle.js';
17
>
import { connectionTokenQueryName } from '../../base/common/network.js';
18
>
import { IPCServer, IServerChannel } from '../../base/parts/ipc/common/ipc.js';
19
>
import { ILogService } from '../../platform/log/common/log.js';
20
>
import type * as wsTypes from 'ws';
21
>
import type * as netTypes from 'net';
22
>
23
>
const agentHostProxyUnavailableMessage = 'Agent host proxy is not available because no upstream agent host endpoint was configured.';
24
>
25
>
/**
26
>
* Endpoint description for the upstream agent host. One of `port` or
27
>
* `socketPath` must be set, matching how `setWebSocketConfig` is called
28
>
* for `NodeAgentHostStarter`.
29
>
*/
30
>
export interface IAgentHostUpstreamEndpoint {
31
>
readonly host?: string;
32
>
readonly port?: string;
33
>
readonly socketPath?: string;
34
>
readonly connectionToken?: string;
35
>
}
36
>
37
>
/**
38
>
* Lazy-loaded `ws` module. Imported once on first connection so renderers
39
>
* that never touch the agent host don't pay the cost.
40
>
*/
41
>
let _wsModule: typeof wsTypes | undefined;
42
async function loadWs(): Promise<typeof wsTypes> {
43
return _wsModule ??= await import('ws');
44
}
46
>
let _netModule: typeof netTypes | undefined;
47
async function loadNet(): Promise<typeof netTypes> {
48
return _netModule ??= await import('net');
49
}
51
>
/**
52
>
* One upstream connection to the agent host, owned by a single renderer
53
>
* client. The default implementation wraps a `ws.WebSocket`; tests inject
54
>
* a custom factory via {@link AgentHostChannel.upstreamFactory}.
55
>
*/
56
>
export interface IUpstreamConnection extends IDisposable {
57
>
readonly onFrame: Event<string>;
58
>
readonly onClose: Event<void>;
59
>
connect(): Promise<void>;
60
>
send(frame: string): void;
61
>
}
62
>
63
>
export type UpstreamConnectionFactory = (endpoint: IAgentHostUpstreamEndpoint) => IUpstreamConnection;
64
>
65
>
/**
66
>
* IPC channel registered when the remote server has no agent host upstream.
67
>
* Keeping the channel present lets renderers fail explicitly without making
68
>
* the IPC layer report `Unknown channel: agentHostProxy`.
69
>
*/
70
>
export class UnavailableAgentHostChannel<TContext> implements IServerChannel<TContext> {
71
>
72
>
listen<T>(_ctx: TContext, event: string): Event<T> {
74
>
case 'frame':
75
>
case 'close':
76
>
return Event.None;
77
>
}
78
throw new Error(`Invalid listen: ${event}`);
81
>
call<T>(_ctx: TContext, command: string): Promise<T> {
83
>
case 'connect':
84
>
return Promise.reject(new Error(agentHostProxyUnavailableMessage));
85
>
case 'send':
86
>
case 'close':
87
>
return Promise.resolve(undefined as T);
88
>
}
89
return Promise.reject(new Error(`Invalid call: ${command}`));
92
>
93
>
/**
94
>
* Default upstream factory: opens an AHP WebSocket to the local agent host.
95
>
*/
96
>
const defaultUpstreamFactory = (logService: ILogService): UpstreamConnectionFactory =>
97
(endpoint) => new WebSocketUpstreamConnection(endpoint, logService);
99
>
class WebSocketUpstreamConnection extends Disposable implements IUpstreamConnection {
100
>
private readonly _onFrame = this._register(new Emitter<string>());
101
>
readonly onFrame: Event<string> = this._onFrame.event;
102
>
103
>
private readonly _onClose = this._register(new Emitter<void>());
104
>
readonly onClose: Event<void> = this._onClose.event;
105
>
106
>
private _ws: wsTypes.WebSocket | undefined;
107
>
private _connectPromise: Promise<void> | undefined;
108
>
private _closeFired = false;
109
>
110
>
constructor(
111
private readonly _endpoint: IAgentHostUpstreamEndpoint,
112
private readonly _logService: ILogService,
113
) {
114
super();
115
}
117
>
connect(): Promise<void> {
118
if (this._store.isDisposed) {
119
return Promise.reject(new Error('UpstreamConnection is disposed'));
120
}
121
return this._connectPromise ??= this._doConnect();
122
}
124
>
private async _doConnect(): Promise<void> {
125
const ws = await loadWs();
126
const url = this._buildUrl();
127
const wsOptions = await this._buildWsOptions();
128
129
this._logService.info(`[AgentHostChannel] Opening upstream to ${this._endpoint.socketPath ?? url}`);
130
const socket = new ws.WebSocket(url, wsOptions);
131
this._ws = socket;
132
133
return new Promise<void>((resolve, reject) => {
134
const onOpen = () => {
135
cleanup();
136
this._logService.trace('[AgentHostChannel] Upstream open');
137
socket.on('message', (data: Buffer | string) => {
138
const text = typeof data === 'string' ? data : data.toString('utf-8');
139
this._onFrame.fire(text);
140
});
141
socket.on('close', () => this._fireClose());
142
socket.on('error', err => {
143
this._logService.warn('[AgentHostChannel] Upstream error', err);
144
this._fireClose();
145
});
146
resolve();
147
};
148
149
const onError = (err: Error) => {
150
cleanup();
151
this._logService.warn('[AgentHostChannel] Upstream connection failed', err);
152
this._fireClose();
153
reject(err);
154
};
155
156
const onClose = () => {
157
cleanup();
158
this._fireClose();
159
reject(new Error('Upstream closed before connect'));
160
};
161
162
const cleanup = () => {
163
socket.removeListener('open', onOpen);
164
socket.removeListener('error', onError);
165
socket.removeListener('close', onClose);
166
};
167
168
socket.on('open', onOpen);
169
socket.on('error', onError);
170
socket.on('close', onClose);
171
});
172
}
174
>
send(frame: string): void {
175
const ws = this._ws;
176
if (!ws || ws.readyState !== ws.OPEN) {
177
this._logService.warn('[AgentHostChannel] Drop send: upstream not open');
178
this._fireClose();
179
return;
180
}
181
ws.send(frame);
182
}
184
>
override dispose(): void {
185
this._ws?.close();
186
this._fireClose();
187
super.dispose();
188
}
190
>
private _fireClose(): void {
191
if (this._closeFired) {
192
return;
193
}
194
this._closeFired = true;
195
this._onClose.fire();
196
}
198
>
private _buildUrl(): string {
199
const host = this._endpoint.host ?? 'localhost';
200
const port = this._endpoint.port ?? '0';
201
let url = `ws://${host}:${port}`;
202
if (this._endpoint.connectionToken) {
203
url += `?${connectionTokenQueryName}=${encodeURIComponent(this._endpoint.connectionToken)}`;
204
}
205
return url;
206
}
208
>
private async _buildWsOptions(): Promise<wsTypes.ClientOptions | undefined> {
209
if (!this._endpoint.socketPath) {
210
return undefined;
211
}
212
const net = await loadNet();
213
const socketPath = this._endpoint.socketPath;
214
// Note: `createConnection` shape required by `ws` differs slightly
215
// across versions; we cast through `unknown` to match the local typings.
216
const createConnection = (() => net.createConnection(socketPath)) as unknown as wsTypes.ClientOptions['createConnection'];
217
return { createConnection } satisfies wsTypes.ClientOptions;
218
}
220
>
221
>
/**
222
>
* IPC channel that proxies the agent host protocol. One channel instance
223
>
* serves all renderer clients; per-context state is tracked in `_perCtx`.
224
>
*/
225
>
export class AgentHostChannel<TContext> extends Disposable implements IServerChannel<TContext> {
226
>
227
>
private readonly _perCtx = new Map<TContext, IUpstreamConnection>();
228
>
private readonly _upstreamFactory: UpstreamConnectionFactory;
229
>
230
>
constructor(
232
>
private readonly _endpoint: IAgentHostUpstreamEndpoint,
233
>
private readonly _logService: ILogService,
234
>
upstreamFactory?: UpstreamConnectionFactory,
235
>
) {
236
>
super();
237
>
this._upstreamFactory = upstreamFactory ?? defaultUpstreamFactory(_logService);
238
>
this._register(ipcServer.onDidRemoveConnection(c => this._disposeCtx(c.ctx as unknown as TContext)));
239
>
}
241
>
listen<T>(ctx: TContext, event: string): Event<T> {
243
>
switch (event) {
244
>
case 'frame': return conn.onFrame as Event<unknown> as Event<T>;
245
>
case 'close': return conn.onClose as Event<unknown> as Event<T>;
246
>
}
247
throw new Error(`Invalid listen: ${event}`);
250
>
async call<T>(ctx: TContext, command: string, arg?: unknown): Promise<T> {
252
>
switch (command) {
253
>
case 'connect':
254
>
this._logService.info(`[AgentHostChannel] Renderer ctx=${String(ctx)} requested connect to upstream`);
255
>
await conn.connect();
256
>
return undefined as T;
257
>
case 'send':
259
throw new Error('send: arg must be a string frame');
260
}
262
>
return undefined as T;
264
this._disposeCtx(ctx);
265
return undefined as T;
267
throw new Error(`Invalid call: ${command}`);
270
>
override dispose(): void {
273
>
}
275
>
super.dispose();
276
>
}
278
>
private _getOrCreate(ctx: TContext): IUpstreamConnection {
280
>
if (!conn) {
281
>
conn = this._upstreamFactory(this._endpoint);
282
>
this._perCtx.set(ctx, conn);
283
>
// If the upstream closes on its own (e.g. agent host restart or
284
>
// connection drop), evict it from the cache so the next
285
>
// `connect()` call creates a fresh upstream rather than
286
>
// returning the stuck-closed one.
287
>
const sub = conn.onClose(() => {
288
>
sub.dispose();
289
>
if (this._perCtx.get(ctx) === conn) {
291
>
}
293
>
}
294
>
return conn;
295
>
}
297
>
private _disposeCtx(ctx: TContext): void {
299
>
if (conn) {
300
>
this._perCtx.delete(ctx);
301
>
conn.dispose();
302
>
}
303
>
}