src/vs/platform/agentHost/node/codex/codexAppServerClient.ts
473 LOC · 426 covered · 47 uncovered · 73 ranges · 50 concepts · 25 introducers · 27 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
>
/*---------------------------------------------------------------------------------------------
codexAppServerClient.ts ×15
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
>
import type { Readable, Writable } from 'stream';
7
>
import { CancellationError } from '../../../../base/common/errors.js';
8
>
import { Emitter, Event } from '../../../../base/common/event.js';
9
>
import { Disposable, type IDisposable, toDisposable } from '../../../../base/common/lifecycle.js';
10
>
import { hasKey } from '../../../../base/common/types.js';
11
>
import type { ClientNotification } from './protocol/generated/ClientNotification.js';
12
>
import type { ClientRequest } from './protocol/generated/ClientRequest.js';
13
>
import type { RequestId } from './protocol/generated/RequestId.js';
14
>
import type { ServerNotification } from './protocol/generated/ServerNotification.js';
15
>
import type { ServerRequest } from './protocol/generated/ServerRequest.js';
16
>
17
>
// #region Wire types
18
>
//
19
>
// JSON-RPC 2.0 over NDJSON, with the `"jsonrpc": "2.0"` field *omitted on
20
>
// the wire* per the codex app-server convention. The generated
21
>
// `ClientRequest` / `ClientNotification` / `ServerRequest` /
22
>
// `ServerNotification` unions encode `method` + `params` shapes; we add
23
>
// `id` + `result` / `error` envelopes on top.
24
>
25
>
interface IWireResponseSuccess<R = unknown> {
26
>
readonly id: RequestId;
27
>
readonly result: R;
28
>
readonly error?: undefined;
29
>
readonly method?: undefined;
30
>
}
31
>
32
>
interface IWireResponseError {
33
>
readonly id: RequestId;
34
>
readonly result?: undefined;
35
>
readonly error: { readonly code: number; readonly message: string; readonly data?: unknown };
36
>
readonly method?: undefined;
37
>
}
38
>
39
>
type WireMessage = ClientNotification | ClientRequest | ServerNotification | ServerRequest | IWireResponseSuccess | IWireResponseError;
40
>
41
>
// #endregion
42
>
43
>
/**
44
>
* Standard JSON-RPC error codes.
45
>
*
46
>
* @see https://www.jsonrpc.org/specification#error_object
47
>
*/
48
>
export const enum JsonRpcErrorCode {
49
>
ParseError = -32700,
50
>
InvalidRequest = -32600,
51
>
MethodNotFound = -32601,
52
>
InvalidParams = -32602,
53
>
InternalError = -32603,
54
>
}
55
>
56
>
/**
57
>
* Error thrown when a remote request responds with an `error` envelope.
58
>
*/
59
>
export class JsonRpcError extends Error {
60
>
constructor(
62
>
message: string,
63
>
readonly data?: unknown,
64
>
) {
65
>
super(message);
66
>
this.name = 'JsonRpcError';
67
>
}
69
>
70
>
// #region Typed method projections
71
>
//
72
>
// Extract `<method>` → `params` / `result` for each direction. The
73
>
// generated unions have shape `{ method: "x/y", id?: RequestId, params: P
74
>
// }`, so a discriminated-union pick works as a method-keyed lookup.
75
>
76
>
type MethodOf<U> = U extends { method: infer M } ? M : never;
77
>
type ParamsOf<U, M> = U extends { method: M; params: infer P } ? P : never;
78
>
79
>
export type ClientRequestMethod = MethodOf<ClientRequest>;
80
>
export type ClientNotificationMethod = MethodOf<ClientNotification>;
81
>
export type ServerRequestMethod = MethodOf<ServerRequest>;
82
>
export type ServerNotificationMethod = MethodOf<ServerNotification>;
83
>
84
>
export type ClientRequestParams<M extends ClientRequestMethod> = ParamsOf<ClientRequest, M>;
85
>
export type ClientNotificationParams<M extends ClientNotificationMethod> = ParamsOf<ClientNotification, M>;
86
>
export type ServerRequestParams<M extends ServerRequestMethod> = ParamsOf<ServerRequest, M>;
87
>
export type ServerNotificationParams<M extends ServerNotificationMethod> = ParamsOf<ServerNotification, M>;
88
>
89
>
// `result` for client-issued requests doesn't have a single generated
90
>
// union; each method has its own `<X>Response` type. We surface the
91
>
// response shape as a generic parameter on `request<M, R>` so callers
92
>
// can name the response type explicitly, defaulting to `unknown`.
93
>
94
>
// #endregion
95
>
96
>
/**
97
>
* Result of a server→client request. Either a successful result payload
98
>
* or a JSON-RPC error envelope. Implementations of
99
>
* {@link ICodexAppServerClient.onRequest} return one of these.
100
>
*/
101
>
export type ServerRequestHandlerResult<R = unknown> =
102
>
| { readonly result: R; readonly error?: undefined }
103
>
| { readonly result?: undefined; readonly error: { readonly code: number; readonly message: string; readonly data?: unknown } };
104
>
105
>
/**
106
>
* Subset of `ChildProcessWithoutNullStreams` we actually use, so callers
107
>
* can pass either a real child process or an in-memory pair for tests.
108
>
*/
109
>
export interface ICodexAppServerTransport {
110
>
readonly stdin: Writable;
111
>
readonly stdout: Readable;
112
>
/** Force termination. Used as the 2 s grace force-kill fallback. */
113
>
kill(signal?: NodeJS.Signals): boolean;
114
>
/** Fires when the underlying process exits. */
115
>
readonly onExit: Event<{ readonly code: number | null; readonly signal: NodeJS.Signals | null }>;
116
>
/** Registers a one-shot exit listener that may outlive client disposal. */
117
>
onExitOnce(listener: (e: { readonly code: number | null; readonly signal: NodeJS.Signals | null }) => void): void;
118
>
}
119
>
120
>
/**
121
>
* Generic JSON-RPC client over a {@link ICodexAppServerTransport}.
122
>
*
123
>
* The client doesn't know anything about codex's domain — it just
124
>
* brokers typed requests and notifications in both directions. The
125
>
* `CodexAgent` layer above translates this into `IAgent` semantics.
126
>
*
127
>
* Lifecycle:
128
>
* - Construct with an active transport. The client immediately starts
129
>
* reading from `transport.stdout`.
130
>
* - Send requests / notifications via {@link request} / {@link notify}.
131
>
* - Register handlers for server-initiated traffic via
132
>
* {@link onNotification} / {@link onRequest}.
133
>
* - On `dispose()`: send EOF on stdin, wait up to 2 s for clean exit,
134
>
* then SIGKILL. Outstanding requests reject with `CancellationError`.
135
>
*/
136
>
export interface ICodexAppServerClient extends IDisposable {
137
>
/** Fires once when the transport exits (clean or otherwise). */
138
>
readonly onExit: Event<{ readonly code: number | null; readonly signal: NodeJS.Signals | null }>;
139
>
140
>
/** Fires when the underlying transport rejects further writes (process exited unexpectedly). */
141
>
readonly onTransportError: Event<Error>;
142
>
143
>
/**
144
>
* Issue a request. Resolves with the typed response payload, or
145
>
* rejects with {@link JsonRpcError} for protocol-level errors and
146
>
* {@link CancellationError} on dispose.
147
>
*/
148
>
request<M extends ClientRequestMethod, R = unknown>(
149
>
method: M,
150
>
params: ClientRequestParams<M>,
151
>
): Promise<R>;
152
>
153
>
/**
154
>
* Fire-and-forget notification. Does not throw if the transport
155
>
* already closed; lost notifications are surfaced via
156
>
* `onTransportError`.
157
>
*/
158
>
notify<M extends ClientNotificationMethod>(
159
>
method: M,
160
>
params: ClientNotificationParams<M>,
161
>
): void;
162
>
163
>
/**
164
>
* Register a handler for a server-pushed notification.
165
>
*
166
>
* Only one handler per method; subsequent registrations replace the
167
>
* previous handler.
168
>
*/
169
>
onNotification<M extends ServerNotificationMethod>(
170
>
method: M,
171
>
handler: (params: ServerNotificationParams<M>) => void,
172
>
): IDisposable;
173
>
174
>
/**
175
>
* Register a handler for a server-initiated request. The handler
176
>
* returns a typed result or an error envelope.
177
>
*
178
>
* Only one handler per method; subsequent registrations replace the
179
>
* previous handler. Unregistered methods reply with
180
>
* {@link JsonRpcErrorCode.MethodNotFound}.
181
>
*/
182
>
onRequest<M extends ServerRequestMethod, R = unknown>(
183
>
method: M,
184
>
handler: (params: ServerRequestParams<M>) => Promise<ServerRequestHandlerResult<R>> | ServerRequestHandlerResult<R>,
185
>
): IDisposable;
186
>
}
187
>
188
>
interface IPendingRequest {
189
>
resolve(value: unknown): void;
190
>
reject(reason: unknown): void;
191
>
readonly method: string;
192
>
}
193
>
194
>
const GRACE_KILL_MS = 2_000;
195
>
196
>
export class CodexAppServerClient extends Disposable implements ICodexAppServerClient {
197
>
198
>
private readonly _onExit = this._register(new Emitter<{ readonly code: number | null; readonly signal: NodeJS.Signals | null }>());
199
>
readonly onExit = this._onExit.event;
200
>
201
>
private readonly _onTransportError = this._register(new Emitter<Error>());
202
>
readonly onTransportError = this._onTransportError.event;
203
>
204
>
private _nextId = 1;
205
>
private readonly _pending = new Map<number, IPendingRequest>();
206
>
private readonly _notificationHandlers = new Map<string, (params: unknown) => void>();
207
>
private readonly _requestHandlers = new Map<string, (params: unknown) => Promise<ServerRequestHandlerResult<unknown>>>();
208
>
209
>
private _exited = false;
210
>
private _disposed = false;
211
>
private _buf = '';
212
>
213
>
constructor(
215
>
private readonly _onLog?: (level: 'info' | 'warn' | 'error', message: string) => void,
216
>
private readonly _graceKillMs = GRACE_KILL_MS,
217
>
) {
218
>
super();
219
>
this._register(this._transport.onExit(e => this._handleExit(e)));
220
>
this._transport.stdout.setEncoding?.('utf8');
221
>
this._register(this._listenToStdout());
222
>
}
224
>
private _listenToStdout(): IDisposable {
226
>
const text = typeof chunk === 'string' ? chunk : chunk.toString('utf8');
codexAppServerClient.ts ×7
227
>
this._buf += text;
228
>
let nl: number;
229
>
while ((nl = this._buf.indexOf('\n')) >= 0) {
230
>
const line = this._buf.slice(0, nl);
231
>
this._buf = this._buf.slice(nl + 1);
232
>
const trimmed = line.trim();
233
>
if (trimmed.length === 0) {
234
continue;
235
}
237
>
try {
238
>
parsed = JSON.parse(trimmed) as WireMessage;
239
>
} catch (err) {
240
this._log('error', `parse error on line: ${trimmed.slice(0, 200)}`);
241
continue;
242
}
244
>
}
245
>
};
247
>
return toDisposable(() => this._transport.stdout.off('data', onData));
248
>
}
250
>
private _dispatch(msg: WireMessage): void {
252
>
const hasMethod = hasKey(msg, { method: true });
253
>
// Response envelope (has id and either result or error, no method).
254
>
if (hasId && !hasMethod && (hasKey(msg, { result: true }) || hasKey(msg, { error: true }))) {
257
>
return;
258
>
}
260
>
const pending = this._pending.get(id);
261
>
if (!pending) {
262
this._log('warn', `unsolicited response id=${msg.id}`);
263
return;
264
}
266
>
if (hasKey(msg, { error: true }) && msg.error) {
267
>
pending.reject(new JsonRpcError(msg.error.code, msg.error.message, msg.error.data));
codexAppServerClient.ts ×1
270
>
}
272
>
}
274
>
// Server→client request (has method + id).
275
>
if (hasMethod && hasId && msg.id !== undefined && msg.method !== undefined) {
codexAppServerClient.ts ×7
277
>
return;
278
>
}
280
>
// Server→client notification (has method, no id).
283
>
return;
284
>
}
285
286
this._log('warn', `unrecognized message: ${JSON.stringify(msg).slice(0, 200)}`);
289
>
private async _handleServerRequest(msg: ServerRequest): Promise<void> {
291
>
if (!handler) {
293
>
id: msg.id,
294
>
error: {
295
>
code: JsonRpcErrorCode.MethodNotFound,
296
>
message: `Method not found: ${msg.method}`,
297
>
},
298
>
});
299
>
return;
300
>
}
302
>
const result = await handler(msg.params);
304
this._writeMessage({ id: msg.id, error: result.error });
306
>
this._writeMessage({ id: msg.id, result: result.result });
307
>
}
310
>
this._log('error', `handler for ${msg.method} threw: ${message}`);
311
>
this._writeMessage({
312
>
id: msg.id,
313
>
error: { code: JsonRpcErrorCode.InternalError, message },
314
>
});
315
>
}
318
>
private _handleServerNotification(msg: ServerNotification): void {
320
>
if (!handler) {
322
>
// notifications we don't handle yet). Warn so every dropped method is
323
>
// intentionally triaged by CodexAgent instead of disappearing silently.
324
>
this._log('warn', `dropping unhandled notification: ${msg.method}`);
325
>
return;
326
>
}
328
>
handler(msg.params);
329
>
} catch (err) {
330
this._log('error', `notification handler ${msg.method} threw: ${err instanceof Error ? err.message : String(err)}`);
331
}
334
>
private _writeMessage(message: unknown): boolean {
336
return false;
337
}
339
>
this._transport.stdin.write(JSON.stringify(message) + '\n');
340
>
return true;
341
>
} catch (err) {
342
this._onTransportError.fire(err instanceof Error ? err : new Error(String(err)));
343
return false;
344
}
347
>
private _handleExit(e: { code: number | null; signal: NodeJS.Signals | null }): void {
349
return;
350
}
352
>
const reason = `codex app-server exited (code=${e.code}, signal=${e.signal})`;
353
>
for (const [id, pending] of this._pending) {
354
>
pending.reject(new JsonRpcError(JsonRpcErrorCode.InternalError, `${reason}; request id=${id} (${pending.method}) aborted`));
355
>
}
356
>
this._pending.clear();
357
>
this._onExit.fire(e);
358
>
}
360
>
request<M extends ClientRequestMethod, R = unknown>(
362
>
params: ClientRequestParams<M>,
363
>
): Promise<R> {
364
>
if (this._disposed) {
365
return Promise.reject(new CancellationError());
366
}
368
return Promise.reject(new JsonRpcError(JsonRpcErrorCode.InternalError, 'transport has exited'));
369
}
371
>
return new Promise<R>((resolve, reject) => {
372
>
this._pending.set(id, { method, resolve: resolve as (v: unknown) => void, reject });
373
>
const ok = this._writeMessage({ id, method, params });
374
>
if (!ok) {
375
this._pending.delete(id);
376
reject(new JsonRpcError(JsonRpcErrorCode.InternalError, 'write failed; transport closed'));
377
}
379
>
}
381
>
notify<M extends ClientNotificationMethod>(
383
>
params: ClientNotificationParams<M>,
384
>
): void {
385
>
// `params` may be `undefined` (e.g. `initialized` carries no
386
>
// payload). Don't include the key in that case.
387
>
const payload: { method: string; params?: unknown } = { method };
388
>
if (params !== undefined) {
389
payload.params = params;
390
}
392
>
}
394
>
onNotification<M extends ServerNotificationMethod>(
396
>
handler: (params: ServerNotificationParams<M>) => void,
397
>
): IDisposable {
398
>
this._notificationHandlers.set(method, handler as (params: unknown) => void);
399
>
return toDisposable(() => {
400
>
if (this._notificationHandlers.get(method) === handler) {
401
>
this._notificationHandlers.delete(method);
402
>
}
403
>
});
404
>
}
406
>
onRequest<M extends ServerRequestMethod, R = unknown>(
408
>
handler: (params: ServerRequestParams<M>) => Promise<ServerRequestHandlerResult<R>> | ServerRequestHandlerResult<R>,
409
>
): IDisposable {
410
>
const wrapped = async (params: unknown): Promise<ServerRequestHandlerResult<unknown>> => {
411
>
return await handler(params as ServerRequestParams<M>);
412
>
};
413
>
this._requestHandlers.set(method, wrapped);
414
>
return toDisposable(() => {
415
>
if (this._requestHandlers.get(method) === wrapped) {
416
>
this._requestHandlers.delete(method);
417
>
}
418
>
});
419
>
}
421
>
override dispose(): void {
423
return;
424
}
426
>
// Reject anything still pending so callers don't hang.
427
>
for (const pending of this._pending.values()) {
429
>
}
431
>
// Try a graceful EOF on stdin; if the process doesn't exit in
432
>
// {@link GRACE_KILL_MS}, SIGKILL.
433
>
try {
434
>
this._transport.stdin.end();
435
>
} catch { /* already closed */ }
436
>
if (!this._exited) {
438
try {
439
this._transport.kill('SIGKILL');
440
} catch { /* already dead */ }
442
>
this._transport.onExitOnce(() => {
445
>
timer.unref?.();
446
>
}
448
>
}
450
>
private _log(level: 'info' | 'warn' | 'error', message: string): void {
452
>
}
454
>
455
>
/**
456
>
* Wrap a {@link import('node:child_process').ChildProcessWithoutNullStreams}
457
>
* into an {@link ICodexAppServerTransport}. Tests use a fake transport
458
>
* built around `node:stream`'s `PassThrough` or similar.
459
>
*/
460
>
export function transportFromChildProcess(
461
child: { stdin: Writable | null; stdout: Readable | null; kill: (signal?: NodeJS.Signals) => boolean; on: (event: 'exit', listener: (code: number | null, signal: NodeJS.Signals | null) => void) => unknown; once: (event: 'exit', listener: (code: number | null, signal: NodeJS.Signals | null) => void) => unknown; removeListener: (event: 'exit', listener: (code: number | null, signal: NodeJS.Signals | null) => void) => unknown },
462
): ICodexAppServerTransport {
463
if (!child.stdin || !child.stdout) {
464
throw new Error('Child process has no stdio pair');
465
}
466
return {
467
stdin: child.stdin,
468
stdout: child.stdout,
469
kill: signal => child.kill(signal),
470
onExit: Event.fromNodeEventEmitter(child, 'exit', (code: number | null, signal: NodeJS.Signals | null) => ({ code, signal })),
471
onExitOnce: listener => child.once('exit', (code, signal) => listener({ code, signal })),
472
};
473
}