src/vs/platform/agentHost/node/protocolServerHandler.ts
1630 LOC · 1415 covered · 215 uncovered · 366 ranges · 171 concepts · 126 introducers · 77 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
>
/*---------------------------------------------------------------------------------------------
protocolServerHandler.ts ×78
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 { disposableTimeout } from '../../../base/common/async.js';
7
>
import { Emitter } from '../../../base/common/event.js';
8
>
import { isJsonRpcResponse } from '../../../base/common/jsonRpcProtocol.js';
9
>
import { Disposable, DisposableMap, DisposableStore } from '../../../base/common/lifecycle.js';
10
>
import { hasKey } from '../../../base/common/types.js';
11
>
import { URI } from '../../../base/common/uri.js';
12
>
import { ILogService } from '../../log/common/log.js';
13
>
import { AHPFileSystemProvider } from '../common/agentHostFileSystemProvider.js';
14
>
import { AgentSession, type IAgentCreateChatOptions, type IAgentService, type IMcpNotification } from '../common/agentService.js';
15
>
import { isActionEnvelopeRelevantToSubscriptionUris } from '../common/state/agentSubscription.js';
16
>
import { ChatSourceKind } from '../common/state/protocol/channels-chat/commands.js';
17
>
import type { CommandMap } from '../common/state/protocol/messages.js';
18
>
import { ActionEnvelope, ActionType, INotification, isAnnotationsAction, isChangesetAction, isChatAction, isSessionAction, isTerminalAction, type ChatAction, type ClientAnnotationsAction, type ClientChangesetAction, type IRootConfigChangedAction, type SessionAction, type TerminalAction } from '../common/state/sessionActions.js';
19
>
import { PROTOCOL_VERSION } from '../common/state/protocol/version/registry.js';
20
>
import { negotiateProtocolVersion } from '../common/state/protocol/version/negotiation.js';
21
>
import { VSCODE_UPGRADE_METHOD, type UnsupportedProtocolVersionErrorDataEx } from '../common/state/protocolUpgrade.js';
22
>
import { getAgentHostManagementSocketPath, requestAgentHostUpgrade } from './agentHostUpgradeChannel.js';
23
>
import {
24
>
AHP_AUTH_REQUIRED,
25
>
AhpErrorCodes,
26
>
AHP_PROVIDER_NOT_FOUND,
27
>
AHP_SESSION_NOT_FOUND,
28
>
AHP_UNSUPPORTED_PROTOCOL_VERSION,
29
>
JsonRpcRequest,
30
>
isJsonRpcNotification,
31
>
isJsonRpcRequest,
32
>
JSON_RPC_INTERNAL_ERROR,
33
>
JsonRpcErrorCodes,
34
>
ProtocolError,
35
>
type AhpServerNotification,
36
>
type InitializeParams,
37
>
type JsonRpcResponse,
38
>
type ReconnectParams,
39
>
type IStateSnapshot,
40
>
type SubscribeResult,
41
>
type ListSessionsResult,
42
>
} from '../common/state/sessionProtocol.js';
43
>
import { isAhpResourceWatchChannel, isAhpRootChannel, ResponsePartKind, SessionStatus, ToolCallConfirmationReason, ToolCallContributorKind, ToolCallStatus, ToolResultContentType, buildDefaultChatUri, isAhpChatChannel, parseChatUri, parseRequiredSessionUriFromChatUri, type ISessionWithDefaultChat, type SessionState } from '../common/state/sessionState.js';
44
>
import type { IProtocolServer, IProtocolTransport } from '../common/state/sessionTransport.js';
45
>
import { AgentHostStateManager } from './agentHostStateManager.js';
46
>
import {
47
>
buildOtlpLogsChannelUri,
48
>
extractLevelFromOtlpLogsUri,
49
>
levelToSeverityNumber,
50
>
OTLP_CHANNEL_SCHEME,
51
>
OTLP_LOGS_CHANNEL_TEMPLATE,
52
>
OtlpLogEmitter,
53
>
toResourceLogsPayload,
54
>
type IOtlpLogRecord,
55
>
type OtlpLogLevelName,
56
>
} from '../common/otlp/otlpLogEmitter.js';
57
>
import { isFileResourceRead } from '../common/resourceReadLogging.js';
58
>
59
>
/** Default capacity of the server-side action replay buffer. */
60
>
const REPLAY_BUFFER_CAPACITY = 1000;
61
>
62
>
const CLIENT_TOOL_CALL_DISCONNECT_TIMEOUT = 30_000;
63
>
64
>
/**
65
>
* Client-dispatchable actions that are declared in the protocol but not yet
66
>
* operational in this build. The multiroot working-directory mutations
67
>
* (`session|chat/workingDirectorySet|Removed`) would mutate the synchronized
68
>
* working-directory set without reconfiguring the agent's actual directory
69
>
* access, so they are rejected in the dispatch path until capability-backed
70
>
* multiroot support lands.
71
>
*/
72
>
const UNSUPPORTED_CLIENT_ACTION_TYPES: ReadonlySet<ActionType> = new Set([
73
>
ActionType.SessionWorkingDirectorySet,
74
>
ActionType.SessionWorkingDirectoryRemoved,
75
>
ActionType.ChatWorkingDirectorySet,
76
>
ActionType.ChatWorkingDirectoryRemoved,
77
>
]);
78
>
79
>
/** A client tool call in any of these statuses is still awaiting its result. */
80
>
function isPendingToolCallStatus(status: ToolCallStatus): boolean {
protocolServerHandler.ts ×9
81
>
return status === ToolCallStatus.Streaming
86
>
/** Build a JSON-RPC success response suitable for transport.send(). */
87
>
function jsonRpcSuccess(id: number, result: unknown): JsonRpcResponse {
protocolServerHandler.ts ×1
88
>
return { jsonrpc: '2.0', id, result };
89
>
}
91
>
/** Build a JSON-RPC error response suitable for transport.send(). */
92
>
function jsonRpcError(id: number, code: number, message: string, data?: unknown): JsonRpcResponse {
protocolServerHandler.ts ×1
93
>
return { jsonrpc: '2.0', id, error: { code, message, ...(data !== undefined ? { data } : {}) } };
94
>
}
96
>
/** Build a JSON-RPC error response from an unknown thrown value, preserving {@link ProtocolError} fields. */
97
>
function jsonRpcErrorFrom(id: number, err: unknown): JsonRpcResponse {
protocolServerHandler.ts ×1
98
>
if (err instanceof ProtocolError) {
99
>
return jsonRpcError(id, err.code, err.message, err.data);
100
>
}
101
>
const message = err instanceof Error ? (err.stack ?? err.message) : String(err);
102
>
return jsonRpcError(id, JSON_RPC_INTERNAL_ERROR, message);
103
>
}
105
>
function shouldLogFailedRequest(method: string, params: unknown, err: unknown): boolean {
protocolServerHandler.ts ×3
106
>
if (!(err instanceof ProtocolError) || err.code !== AhpErrorCodes.NotFound || !isFileResourceRead(method, params)) {
108
>
}
110
>
}
112
>
/** True when `value` is a non-null params object (as opposed to an array or primitive). */
113
>
function isParamsObject(value: unknown): value is Record<string, unknown> {
protocolServerHandler.ts ×5
114
>
return typeof value === 'object' && value !== null && !Array.isArray(value);
115
>
}
117
>
/**
118
>
* Returns the `channel` URI carried on a request's params when it is an
119
>
* `mcp://` channel — the AHP routing envelope for raw MCP requests
120
>
* tunnelled over the JSON-RPC connection. Returns `undefined` for any
121
>
* other params shape.
122
>
*/
124
>
if (!isParamsObject(params)) {
125
return undefined;
126
}
128
>
if (typeof channel !== 'string' || !channel.startsWith('mcp://')) {
129
>
return undefined;
130
>
}
131
return channel;
132
}
134
>
/**
135
>
* Methods handled by the request dispatcher. Excludes `initialize`,
136
>
* `reconnect`, and `ping`, which are handled directly during message
137
>
* dispatch without requiring an established client context.
138
>
*/
139
>
type RequestMethod = Exclude<keyof CommandMap, 'initialize' | 'reconnect' | 'ping'>;
140
>
141
>
/**
142
>
* Typed handler map: each key is a request method, each value is a handler
143
>
* that receives the correctly-typed params and must return the correctly-typed
144
>
* result. The compiler will error if a handler returns the wrong shape.
145
>
*/
146
>
type RequestHandlerMap = {
147
>
[M in RequestMethod]: (client: IConnectedClient, params: CommandMap[M]['params']) => Promise<CommandMap[M]['result']>;
148
>
};
149
>
150
>
/**
151
>
* Discriminant for {@link ChannelSubscription}. Distinguishes a regular
152
>
* state-bearing channel (root, session, terminal, changeset) from the
153
>
* stateless OTLP signal channels so each subscribe/unsubscribe path can
154
>
* dispatch through a single typed lookup.
155
>
*/
156
>
const enum ChannelKind {
157
>
/**
158
>
* Subscribed via {@link IAgentService.subscribe} and tracked by the
159
>
* server-side refcount. Carries replayable state, participates in
160
>
* action broadcasts ({@link _broadcastAction}) and reconnect
161
>
* snapshot/replay.
162
>
*/
163
>
State = 'state',
164
>
/**
165
>
* Resource-watch channels (`ahp-resource-watch:/<id>`). Tracked
166
>
* separately so subscribe/unsubscribe routes through the agent
167
>
* service's per-watch refcount + grace timer rather than the
168
>
* session-shaped {@link IAgentService.subscribe} path.
169
>
*/
170
>
ResourceWatch = 'resource-watch',
171
>
/**
172
>
* Subscribed against the OTLP logs channel template advertised in
173
>
* {@link InitializeResult.telemetry}. Stateless — no snapshot, no
174
>
* agent-service refcount. The `level` field records the minimum
175
>
* severity the client asked to receive.
176
>
*/
177
>
OtlpLogs = 'otlp-logs',
178
>
}
179
>
180
>
/**
181
>
* Per-channel server-side subscription record. Stored on every
182
>
* {@link IConnectedClient} so each subscribed channel can be routed by
183
>
* its `kind` without re-deriving it from the URI on every dispatch.
184
>
*
185
>
* `uri` is the canonical channel URI string used everywhere a subscription
186
>
* is referenced — the same string is broadcast on outbound notifications
187
>
* and persists across reconnects.
188
>
*/
189
>
type ChannelSubscription =
190
>
| { readonly kind: ChannelKind.State; readonly uri: string }
191
>
| { readonly kind: ChannelKind.ResourceWatch; readonly uri: string }
192
>
| { readonly kind: ChannelKind.OtlpLogs; readonly uri: string; readonly level: OtlpLogLevelName };
193
>
194
>
/**
195
>
* Represents a connected protocol client with its subscription state.
196
>
*/
197
>
interface IConnectedClient {
198
>
readonly clientId: string;
199
>
readonly protocolVersion: string;
200
>
readonly transport: IProtocolTransport;
201
>
/**
202
>
* Every channel the client is currently subscribed to, keyed by the
203
>
* canonical channel URI. OTLP channel URIs are canonicalised to
204
>
* `buildOtlpLogsChannelUri(level)` so URI variants that resolve to
205
>
* the same logical channel collapse to one entry.
206
>
*/
207
>
readonly subscriptions: Map<string, ChannelSubscription>;
208
>
readonly disposables: DisposableStore;
209
>
}
210
>
211
>
/**
212
>
* Per-client server-side record, keyed by clientId in
213
>
* {@link ProtocolServerHandler._clients}. Unlike {@link IConnectedClient},
214
>
* the record OUTLIVES individual transports: multiple overlapping transports
215
>
* for the same logical client are held oldest-first, with the active transport
216
>
* at the end. When the last transport disconnects, the record is retained
217
>
* (until pruned) so the tool-call disconnect-grace machinery can compute the
218
>
* remaining window and hold any armed timeouts.
219
>
*
220
>
* A client is in exactly one of two states, which makes the core invariant
221
>
* unrepresentable in the wrong shape: a client either has one or more live
222
>
* transports ({@link IActiveClientRecord}, never any disconnect-grace timers)
223
>
* or has no transport and is within its disconnect-grace window
224
>
* ({@link IGraceClientRecord}, never any connections). Transitions happen only
225
>
* in {@link ProtocolServerHandler._attachConnection} (→ active, which disposes
226
>
* any grace timers) and the transport `onClose` handler (→ grace, once the last
227
>
* transport is gone).
228
>
*/
229
>
type IClientRecord = IActiveClientRecord | IGraceClientRecord;
230
>
231
>
interface IActiveClientRecord {
232
>
readonly state: 'active';
233
>
/**
234
>
* Live transports for this client, oldest first. The active connection is
235
>
* the last entry (most recent wins). Older entries are kept so that if a
236
>
* reconnecting client registers `A`, then `B`, then `B` closes first, we can
237
>
* fall back to `A` instead of treating the client as disconnected. Never
238
>
* empty: removing the last transport promotes the record to a grace record.
239
>
*/
240
>
readonly connections: IConnectedClient[];
241
>
}
242
>
243
>
interface IGraceClientRecord {
244
>
readonly state: 'grace';
245
>
/**
246
>
* Epoch ms when the client last had a live transport, or when this record
247
>
* was created for a never-connected orphan tool-call stamp. Pins the grace
248
>
* clock so re-arms triggered by later orphaned tool calls shrink the
249
>
* remaining window instead of resetting it. Drives the disconnect-timeout
250
>
* delay (residual window from this instant).
251
>
*/
252
>
lastSeenAt: number;
253
>
/**
254
>
* Pending tool-call disconnect timeouts owned by this client, keyed by
255
>
* session URI. Armed when the client owns a pending client tool call but is
256
>
* not connected; fires a failing completion if it does not (re)connect
257
>
* within the grace window. Reconnecting promotes the record to active and
258
>
* disposes these timers (the grace window no longer applies once a transport
259
>
* is live). Disposing an entry (or the whole map) clears the timer.
260
>
*/
261
>
readonly disconnectTimeouts: DisposableMap<string>;
262
>
}
263
>
264
>
/**
265
>
* Classifies a raw channel URI string into its {@link ChannelKind} and
266
>
* returns the canonical URI to key subscriptions by. Returns `undefined`
267
>
* when the channel is OTLP-flavoured but the URI does not parse into a
268
>
* supported shape (unknown level, missing path) so the caller can
269
>
* silently drop the subscribe rather than installing a broken entry.
270
>
*
271
>
* For state channels the canonical URI is just the input verbatim — the
272
>
* agent service is the authoritative deduplication point and tolerates
273
>
* whatever URI form the client sent.
274
>
*/
275
>
function classifyChannel(channel: string): ChannelSubscription | undefined {
protocolServerHandler.ts ×1
276
>
if (channel.toLowerCase().startsWith(`${OTLP_CHANNEL_SCHEME}:`)) {
278
>
if (!level) {
280
>
}
281
>
return { kind: ChannelKind.OtlpLogs, uri: buildOtlpLogsChannelUri(level), level };
protocolServerHandler.ts ×3
282
>
}
285
>
}
287
>
}
289
>
/**
290
>
* Configuration for protocol-level concerns outside of IAgentService.
291
>
*/
292
>
export interface IProtocolServerConfig {
293
>
/** Default directory returned to clients during the initialize handshake. */
294
>
readonly defaultDirectory?: string;
295
>
/**
296
>
* Whether to expose VS Code extension methods outside the Agent Host Protocol.
297
>
* Defaults to `true` for existing remote listeners.
298
>
*/
299
>
readonly allowExtensionMethods?: boolean;
300
>
/**
301
>
* Characters that, when typed in a {@link UserMessage} input, SHOULD
302
>
* cause the client to issue a `completions` request. Announced to
303
>
* clients in the `initialize` response.
304
>
*/
305
>
readonly completionTriggerCharacters?: readonly string[];
306
>
/**
307
>
* Prefix that marks a user message as a host terminal command.
308
>
*/
309
>
readonly terminalCommandPrefix?: string;
310
>
/**
311
>
* Optional emitter to use as the source for the OTLP logs channel
312
>
* advertised via `InitializeResult.telemetry.logs`. When present, this
313
>
* handler will route `subscribe`/`unsubscribe` requests on
314
>
* `ahp-otlp:` channels to its internal OTLP subscription registry and
315
>
* broadcast every record fed into the emitter as an
316
>
* `otlp/exportLogs` notification. When absent, the OTLP channel is
317
>
* not advertised and any inbound `ahp-otlp:` subscribe request is
318
>
* rejected.
319
>
*/
320
>
readonly otlpLogEmitter?: OtlpLogEmitter;
321
>
}
322
>
323
>
/**
324
>
* Server-side handler that manages protocol connections, routes JSON-RPC
325
>
* messages to the agent service, and broadcasts actions/notifications
326
>
* to subscribed clients.
327
>
*/
328
>
export class ProtocolServerHandler extends Disposable {
329
>
330
>
/**
331
>
* Per-client records keyed by clientId. Holds both connected clients
332
>
* (`connections` non-empty) and recently-disconnected ones retained for the
333
>
* tool-call disconnect-grace window (`connections.length === 0`). See
334
>
* {@link IClientRecord}.
335
>
*/
336
>
private readonly _clients = new Map<string, IClientRecord>();
337
>
private readonly _replayBuffer: ActionEnvelope[] = [];
338
>
339
>
private readonly _onDidChangeConnectionCount = this._register(new Emitter<number>());
340
>
341
>
/** Fires with the current client count whenever a client connects or disconnects. */
342
>
readonly onDidChangeConnectionCount = this._onDidChangeConnectionCount.event;
343
>
344
>
constructor(
345
>
private readonly _agentService: IAgentService,
346
>
private readonly _stateManager: AgentHostStateManager,
347
>
private readonly _server: IProtocolServer,
348
>
private readonly _config: IProtocolServerConfig,
349
>
private readonly _clientFileSystemProvider: AHPFileSystemProvider,
350
>
@ILogService private readonly _logService: ILogService,
351
>
) {
352
>
super();
353
>
354
>
this._register(this._server.onConnection(transport => {
357
>
358
>
this._register(this._stateManager.onDidEmitEnvelope(envelope => {
360
>
if (this._replayBuffer.length > REPLAY_BUFFER_CAPACITY) {
362
>
}
364
>
// A client tool call may be issued for a client that is no longer
365
>
// connected — e.g. a stale stamp from a window that reloaded. The
366
>
// live-disconnect path (`_handleClientDisconnected`) does not cover
367
>
// these because no disconnect event fires for an already-gone
368
>
// client. Detect the orphan at issuance time and arm the same
369
>
// grace-period timeout so the call cannot hang forever. Calls
370
>
// stamped while no client is connected are failed immediately by
371
>
// the provider, so they never reach this path.
372
>
if (envelope.action.type === ActionType.ChatToolCallStart || envelope.action.type === ActionType.ChatToolCallReady) {
374
throw new Error(`[ProtocolServer] Chat tool-call action emitted on non-chat channel: ${envelope.channel}`);
375
}
376
>
this._checkOrphanedClientToolCalls(parseRequiredSessionUriFromChatUri(envelope.channel), envelope.channel);
protocolServerHandler.ts ×9
377
>
}
379
>
380
>
this._register(this._stateManager.onDidEmitNotification(notification => {
383
>
384
>
this._register(this._agentService.onMcpNotification(notification => {
385
this._broadcastMcpNotification(notification);
387
>
388
>
if (this._config.otlpLogEmitter) {
389
>
this._register(this._config.otlpLogEmitter.onDidLog(record => this._broadcastOtlpLog(record)));
protocolServerHandler.ts ×1
390
>
}
392
>
393
>
// ---- Connection handling -------------------------------------------------
394
>
395
>
private _handleNewConnection(transport: IProtocolTransport): void {
397
>
let client: IConnectedClient | undefined;
398
>
399
>
disposables.add(transport.onMessage(msg => {
400
>
if (isJsonRpcRequest(msg)) {
401
>
this._logService.trace(`[ProtocolServer] request: method=${msg.method} id=${msg.id}`);
402
>
403
>
// Ping is stateless and MUST be answerable regardless of whether
404
>
// the connection has been initialized. Carries no payload — the
405
>
// round-trip itself is the liveness signal.
406
>
if (msg.method === 'ping') {
408
>
return;
409
>
}
411
>
// Handle initialize/reconnect as requests that set up the client
414
>
const result = this._handleInitialize(msg.params, transport, disposables);
415
>
client = result.client;
416
>
transport.send(jsonRpcSuccess(msg.id, result.response));
417
>
} catch (err) {
419
>
}
421
>
}
424
>
try {
425
>
const result = this._handleReconnect(msg.params, transport, disposables);
426
>
client = result.client;
427
>
responsePromise = result.responsePromise;
428
>
} catch (err) {
429
transport.send(jsonRpcErrorFrom(msg.id, err));
430
return;
431
}
433
>
response => transport.send(jsonRpcSuccess(msg.id, response)),
434
>
err => transport.send(jsonRpcErrorFrom(msg.id, err)),
435
>
);
436
>
return;
437
>
}
439
>
// The VS Code upgrade request rides on the same transport but
440
>
// is callable pre-`initialize`: by definition we get here when
441
>
// the client's protocol version was rejected, so the client
442
>
// never managed to complete the handshake.
443
>
if ((msg.method as string) === VSCODE_UPGRADE_METHOD) {
445
>
return;
446
>
}
448
>
if (!client) {
449
>
transport.send(jsonRpcError(msg.id, JsonRpcErrorCodes.MethodNotFound, `Method not found: ${msg.method}`));
protocolServerHandler.ts ×2
450
>
return;
451
>
}
454
>
this._logService.trace(`[ProtocolServer] notification: method=${msg.method}`);
protocolServerHandler.ts ×3
455
>
// Notification — fire-and-forget
456
>
switch (msg.method) {
457
>
case 'unsubscribe':
459
>
this._removeSubscription(client, msg.params.channel);
460
>
}
461
>
break;
464
>
this._logService.trace(`[ProtocolServer] dispatchAction: ${JSON.stringify(msg.params.action.type)}`);
465
>
const action = msg.params.action as SessionAction | ChatAction | TerminalAction | ClientChangesetAction | ClientAnnotationsAction | IRootConfigChangedAction;
466
>
const channel = msg.params.channel;
467
>
// Multiroot working-directory mutations are declared in the
468
>
// protocol but not yet supported: they would mutate the
469
>
// synchronized access set without reconfiguring the agent's
470
>
// actual directory access. Reject them through the normal
471
>
// reconciliation path (preserving the client's origin) so the
472
>
// client rolls back its optimistic action instead of leaving
473
>
// it pending, until capability-backed multiroot support lands.
474
>
if (UNSUPPORTED_CLIENT_ACTION_TYPES.has(action.type)) {
475
>
this._logService.warn(`[ProtocolServer] rejecting unsupported client action: ${action.type}`);
agentHostStateManager.ts ×1
476
>
this._stateManager.rejectClientAction(
477
>
channel,
478
>
action,
479
>
{ clientId: client.clientId, clientSeq: msg.params.clientSeq },
480
>
`Unsupported action: ${action.type}`,
481
>
);
482
>
} else if (isSessionAction(action) || isChatAction(action) || isTerminalAction(action) || isChangesetAction(action) || isAnnotationsAction(action) || action.type === ActionType.RootConfigChanged) {
protocolServerHandler.ts ×3
483
>
this._agentService.dispatchAction(channel, action, client.clientId, msg.params.clientSeq);
protocolServerHandler.ts ×1
484
>
}
486
>
break;
490
>
if (pending && pending.client === client) {
491
>
this._pendingReverseRequests.delete(msg.id);
492
>
if (hasKey(msg, { error: true })) {
493
pending.reject(new ProtocolError(
494
msg.error?.code ?? -32000,
495
msg.error?.message ?? 'Reverse RPC error',
496
msg.error?.data,
497
));
499
>
pending.resolve(msg.result);
500
>
}
501
>
}
502
>
}
504
>
505
>
disposables.add(transport.onClose(() => {
506
>
const record = client ? this._clients.get(client.clientId) : undefined;
protocolServerHandler.ts ×2
507
>
if (client && record?.state === 'active') {
509
>
if (connectionIndex !== -1) {
510
>
const subscriptionCount = client.subscriptions.size;
511
>
record.connections.splice(connectionIndex, 1);
512
>
this._releaseClientSubscriptions(client, record);
513
>
this._rejectPendingReverseRequestsForConnection(client);
514
>
if (record.connections.length === 0) {
515
>
this._logService.info(`[ProtocolServer] Client disconnected: ${client.clientId}, subscriptions=${subscriptionCount}`);
protocolServerHandler.ts ×3
516
>
this._clients.set(client.clientId, { state: 'grace', lastSeenAt: Date.now(), disconnectTimeouts: new DisposableMap() });
517
>
this._handleClientDisconnected(client.clientId);
518
>
this._onDidChangeConnectionCount.fire(this._connectedClientCount);
519
>
}
521
>
}
524
>
525
>
disposables.add(transport);
526
>
}
528
>
// ---- Handshake handlers ----------------------------------------------------
529
>
530
>
private _handleInitialize(
532
>
transport: IProtocolTransport,
533
>
disposables: DisposableStore,
534
>
): { client: IConnectedClient; response: unknown } {
535
>
const offered = Array.isArray(params.protocolVersions) ? params.protocolVersions : [];
536
>
this._logService.info(`[ProtocolServer] Initialize: clientId=${params.clientId}, protocolVersions=[${offered.join(', ')}]`);
537
>
538
>
const negotiated = negotiateProtocolVersion(offered, PROTOCOL_VERSION);
539
>
if (!negotiated) {
541
>
supportedVersions: [`^${PROTOCOL_VERSION}`],
542
>
// Only advertise the in-band upgrade method when the agent
543
>
// host was spawned by a VS Code CLI that is listening for
544
>
// management requests (presence of the env var). Otherwise
545
>
// there is no supervisor to actually act on it, so don't
546
>
// lie to the client.
547
>
_meta: getAgentHostManagementSocketPath()
551
>
throw new ProtocolError(
552
>
AHP_UNSUPPORTED_PROTOCOL_VERSION,
553
>
`Client offered protocol versions [${offered.join(', ')}], none of which are compatible with this server's version ${PROTOCOL_VERSION} (server accepts ^${PROTOCOL_VERSION}).`,
554
>
data,
555
>
);
556
>
}
558
>
const client: IConnectedClient = {
559
>
clientId: params.clientId,
560
>
protocolVersion: negotiated,
561
>
transport,
562
>
subscriptions: new Map(),
563
>
disposables,
564
>
};
565
>
this._attachConnection(params.clientId, client);
566
>
567
>
this._registerClientFileSystemAuthority(params.clientId, disposables);
568
>
569
>
570
>
const snapshots: IStateSnapshot[] = [];
571
>
if (params.initialSubscriptions) {
573
>
const snapshot = this._addInitialSubscription(client, uri.toString());
574
>
if (snapshot) {
575
>
snapshots.push(snapshot);
576
>
}
577
>
}
578
>
}
580
>
return {
581
>
client,
582
>
response: {
583
>
protocolVersion: negotiated,
584
>
serverSeq: this._stateManager.serverSeq,
585
>
snapshots,
586
>
defaultDirectory: this._config.defaultDirectory,
587
>
completionTriggerCharacters: this._config.completionTriggerCharacters,
588
>
terminalCommandPrefix: this._config.terminalCommandPrefix,
589
>
telemetry: this._config.otlpLogEmitter ? { logs: OTLP_LOGS_CHANNEL_TEMPLATE } : undefined,
protocolServerHandler.ts ×4
590
>
},
591
>
};
592
>
}
594
>
/**
595
>
* Helper for `initialize` and `reconnect` initial-subscription
596
>
* processing: classify `channel`, install the matching subscription
597
>
* on the client, and return the snapshot to include in the handshake
598
>
* response (or `undefined` for stateless channels and missing state).
599
>
*
600
>
* Side effects:
601
>
* - State channels: register with the agent service and clear any
602
>
* pending tool-call disconnect timeout.
603
>
* - OTLP channels: install the canonical entry on the client's
604
>
* {@link IConnectedClient.subscriptions} map.
605
>
*
606
>
* Channels with unsupported shapes (e.g. `ahp-otlp://logs/verbose`
607
>
* with no recognised level, or a state channel the state manager
608
>
* does not know about) are silently dropped.
609
>
*/
610
>
private _addInitialSubscription(client: IConnectedClient, channel: string): IStateSnapshot | undefined {
612
>
if (!sub) {
613
return undefined;
614
}
616
if (!this._config.otlpLogEmitter) {
617
this._logService.warn(`[ProtocolServer] Ignoring OTLP initialSubscription ${channel}: no OTLP emitter configured.`);
618
return undefined;
619
}
620
client.subscriptions.set(sub.uri, sub);
621
return undefined;
622
}
624
>
if (!snapshot) {
625
return undefined;
626
}
628
>
this._agentService.addSubscriber(URI.parse(sub.uri), client.clientId);
629
>
this._clearClientToolCallDisconnectTimeout(client.clientId, sub.uri);
630
>
return snapshot;
631
>
}
633
>
/**
634
>
* Forwards a client's upgrade request to the hosting VS Code CLI's
635
>
* HTTP management API (advertised via the {@link VSCODE_AGENT_HOST_MANAGEMENT_SOCKET_ENV}).
636
>
* Returns the CLI's parsed response verbatim so the client can render
637
>
* a meaningful status (already up-to-date, restart scheduled, etc.).
638
>
*
639
>
* When the server was not spawned by a managing CLI, responds with
640
>
* `MethodNotFound` — the upgrade method is only meaningfully callable
641
>
* on CLI-hosted servers.
642
>
*/
643
>
private _handleVscodeUpgrade(id: number, transport: IProtocolTransport): void {
645
>
if (!socketPath) {
646
>
transport.send(jsonRpcError(
647
>
id,
648
>
JsonRpcErrorCodes.MethodNotFound,
649
>
`No upgrade supervisor is available for this agent host.`,
650
>
));
651
>
return;
652
>
}
653
requestAgentHostUpgrade(socketPath).then(
654
(result) => transport.send(jsonRpcSuccess(id, result)),
655
(err: unknown) => {
656
this._logService.warn(`[ProtocolServer] vscodeUpgrade signal failed: ${err instanceof Error ? err.message : String(err)}`);
657
transport.send(jsonRpcErrorFrom(id, err));
658
},
659
);
662
>
private _handleReconnect(
664
>
transport: IProtocolTransport,
665
>
disposables: DisposableStore,
666
>
): { client: IConnectedClient; responsePromise: Promise<unknown> } {
667
>
this._logService.info(`[ProtocolServer] Reconnect: clientId=${params.clientId}, lastSeenSeq=${params.lastSeenServerSeq}`);
668
>
669
>
// Synchronously install the client so messages arriving on this transport
670
>
// while we restore subscriptions can find a valid client object. The
671
>
// reconnect response is only sent once `responsePromise` resolves below.
672
>
const client: IConnectedClient = {
673
>
clientId: params.clientId,
674
>
protocolVersion: PROTOCOL_VERSION,
675
>
transport,
676
>
subscriptions: new Map(),
677
>
disposables,
678
>
};
679
>
this._attachConnection(params.clientId, client);
680
>
681
>
// Re-establish the reverse-RPC filesystem authority for this client.
682
>
// The prior transport's `onClose` disposed the previous registration,
683
>
// so without this step any subsequent `resourceRead` / `resourceWrite`
684
>
// / etc. from the agent host would fail with "no connection registered
685
>
// for authority" until the client disconnected and re-initialized.
686
>
this._registerClientFileSystemAuthority(params.clientId, disposables);
687
>
688
>
const oldestBuffered = this._replayBuffer.length > 0 ? this._replayBuffer[0].serverSeq : this._stateManager.serverSeq;
689
>
const canReplay = params.lastSeenServerSeq >= oldestBuffered;
690
>
691
>
const responsePromise = this._restoreReconnectSubscriptions(client, params, canReplay);
692
>
return { client, responsePromise };
693
>
}
695
>
/**
696
>
* Wires the reverse-RPC filesystem callbacks for `clientId` and binds
697
>
* the unregister to `disposables` (the transport's per-connection
698
>
* store). The callbacks dispatch through {@link _sendReverseRequest},
699
>
* which looks up the *current* connected client by id — so re-binding
700
>
* after a reconnect picks up the new transport without rebuilding the
701
>
* closures.
702
>
*/
703
>
private _registerClientFileSystemAuthority(clientId: string, disposables: DisposableStore): void {
704
>
disposables.add(this._clientFileSystemProvider.registerAuthority(clientId, {
protocolServerHandler.ts ×12
705
>
resourceList: (uri) => this._sendReverseRequest(clientId, 'resourceList', { uri: uri.toString() }),
706
>
resourceRead: (uri) => this._sendReverseRequest(clientId, 'resourceRead', { uri: uri.toString() }),
707
>
resourceWrite: (params_) => this._sendReverseRequest(clientId, 'resourceWrite', params_),
708
>
resourceCopy: (params_) => this._sendReverseRequest(clientId, 'resourceCopy', params_),
709
>
resourceDelete: (params_) => this._sendReverseRequest(clientId, 'resourceDelete', params_),
710
>
resourceMove: (params_) => this._sendReverseRequest(clientId, 'resourceMove', params_),
711
>
resourceRequest: (params_) => this._sendReverseRequest(clientId, 'resourceRequest', params_),
712
>
resourceResolve: (params_) => this._sendReverseRequest(clientId, 'resourceResolve', params_),
713
>
resourceMkdir: (params_) => this._sendReverseRequest(clientId, 'resourceMkdir', params_),
714
>
}));
715
>
}
717
>
/**
718
>
* Re-establish each of the client's prior subscriptions on the server side.
719
>
* Uses {@link IAgentService.subscribe} (rather than a bare `addSubscriber`
720
>
* + `getSnapshot`) so any session state that was evicted while the client
721
>
* was disconnected is restored. Returns the appropriate reconnect response
722
>
* payload — `replay` actions when the client's last-seen seq is still in
723
>
* the buffer, otherwise fresh `snapshot`s.
724
>
*/
725
>
private async _restoreReconnectSubscriptions(
727
>
params: ReconnectParams,
728
>
canReplay: boolean,
729
>
): Promise<unknown> {
730
>
const missing: string[] = [];
731
>
const snapshots = await Promise.all(params.subscriptions.map(async sub => {
733
>
const classified = classifyChannel(key);
734
>
if (!classified) {
735
return undefined;
736
}
738
if (!this._config.otlpLogEmitter) {
739
this._logService.warn(`[ProtocolServer] Reconnect: dropping OTLP subscription ${key}: no OTLP emitter configured.`);
740
return undefined;
741
}
742
// Stateless: re-install without going through the agent service.
743
client.subscriptions.set(classified.uri, classified);
744
return undefined;
745
}
747
const descriptor = this._agentService.onResourceWatchSubscribed(classified.uri);
748
if (!descriptor) {
749
this._logService.info(`[ProtocolServer] Reconnect: resource watch ${key} no longer parses`);
750
missing.push(sub);
751
return undefined;
752
}
753
client.subscriptions.set(classified.uri, classified);
754
return {
755
resource: classified.uri,
756
state: descriptor,
757
fromSeq: this._stateManager.serverSeq,
758
};
759
}
761
>
const snapshot = await this._agentService.subscribe(URI.parse(key), client.clientId);
762
>
client.subscriptions.set(classified.uri, classified);
763
>
this._clearClientToolCallDisconnectTimeout(client.clientId, classified.uri);
764
>
return snapshot;
765
>
} catch (err) {
766
this._logService.info(`[ProtocolServer] Reconnect: failed to restore subscription ${key}: ${err instanceof Error ? err.message : String(err)}`);
767
missing.push(sub);
768
return undefined;
769
}
771
>
772
>
this._reconcileActiveClientsAfterReconnect(client);
773
>
774
>
if (canReplay) {
776
>
for (const envelope of this._replayBuffer) {
780
>
}
784
>
}
785
>
return { type: 'snapshot', snapshots: snapshots.filter((s): s is IStateSnapshot => s !== undefined) };
protocolServerHandler.ts ×2
788
>
/**
789
>
* Release a client from every session where it is still an active client
790
>
* but did not resubscribe during a reconnect. The set of resubscribed
791
>
* sessions is gathered from every live connection the client currently
792
>
* holds (not just the reconnecting one) so an overlapping connection that
793
>
* still subscribes to a session keeps the client active there.
794
>
*/
795
>
private _reconcileActiveClientsAfterReconnect(client: IConnectedClient): void {
797
>
const resubscribed = new Set<string>();
798
>
for (const connection of record?.state === 'active' ? record.connections : [client]) {
799
>
for (const sub of connection.subscriptions.values()) {
801
>
resubscribed.add(sub.uri);
802
>
}
803
>
}
805
>
for (const session of this._stateManager.getSessionUris()) {
807
>
if (state && this._isActiveClient(state, client.clientId)) {
809
>
if (!resubscribed.has(session) && !resubscribed.has(chat.resource)) {
810
>
this._releaseActiveClientForSession(session, client.clientId, chat.resource);
protocolServerHandler.ts ×1
811
>
}
813
>
}
817
>
private _handleClientDisconnected(clientId: string): void {
820
>
const isActive = state ? this._isActiveClient(state, clientId) : false;
821
>
const ownsPendingToolCall = state ? this._hasPendingClientToolCall(state, clientId) : false;
822
>
// Keep the client marked active during the grace window so a quick
823
>
// reconnect that resubscribes can retain its slot. The disconnect
824
>
// timeout removes the active client (and fails its pending tool
825
>
// calls) if it never returns; an explicit unsubscribe or a
826
>
// reconnect without resubscription removes it sooner.
827
>
if (isActive || ownsPendingToolCall) {
829
>
this._startClientToolCallDisconnectTimeout(clientId, session, chat.resource);
830
>
}
831
>
}
835
>
/** Whether `clientId` is one of the session's active clients. */
836
>
private _isActiveClient(state: SessionState, clientId: string): boolean {
838
>
}
840
>
/**
841
>
* Remove `clientId` from a session's active clients, if present. Dispatched
842
>
* as a server action so the removal is reflected in state and broadcast to
843
>
* the remaining subscribers.
844
>
*/
845
>
private _removeActiveClient(session: string, clientId: string): void {
847
>
if (state && this._isActiveClient(state, clientId)) {
849
>
type: ActionType.SessionActiveClientRemoved,
850
>
clientId,
851
>
});
852
>
}
855
>
/**
856
>
* Release a client from a session: clear its pending disconnect timeout,
857
>
* fail any client tool calls it still owns, and remove it from the active
858
>
* clients. Used by the explicit-unsubscribe and reconnect-reconciliation
859
>
* paths to drop a client that has left a session.
860
>
*/
861
>
private _releaseActiveClientForSession(session: string, clientId: string, chatChannel: string): void {
862
>
this._clearClientToolCallDisconnectTimeout(clientId, chatChannel);
protocolServerHandler.ts ×9
863
>
this._completeDisconnectedClientToolCalls(clientId, session, chatChannel);
864
>
this._removeActiveClient(session, clientId);
865
>
}
867
>
/**
868
>
* Yields every still-pending client-contributed tool call in `state`'s
869
>
* active turn, paired with its owning `clientId`. Single source of truth
870
>
* for the disconnect-grace machinery: detect ownership
871
>
* ({@link _hasPendingClientToolCall}), arm timeouts
872
>
* ({@link _checkOrphanedClientToolCalls}), and fail orphaned calls
873
>
* ({@link _completeDisconnectedClientToolCalls}).
874
>
*/
875
>
private *_pendingClientToolCalls(state: ISessionWithDefaultChat | undefined) {
877
>
if (!activeTurn) {
879
>
}
881
>
if (part.kind !== ResponsePartKind.ToolCall) {
882
continue;
883
}
885
>
const contributor = toolCall.contributor;
886
>
if (contributor?.kind === ToolCallContributorKind.Client && isPendingToolCallStatus(toolCall.status)) {
887
>
yield { toolCall, clientId: contributor.clientId };
888
>
}
889
>
}
892
>
private _hasPendingClientToolCall(state: ISessionWithDefaultChat | undefined, clientId: string): boolean {
895
>
return true;
896
>
}
897
>
}
901
>
private _hasReplacementActiveClientTool(state: SessionState, clientId: string, toolName: string): boolean {
905
>
}
907
>
/**
908
>
* Arm (or re-arm) the per-(clientId, session) timeout that fails pending
909
>
* client tool calls owned by `clientId` if it does not reconnect before the
910
>
* grace window elapses. Only meaningful for a client with no live transport:
911
>
* a connected client is handled by {@link _attachConnection}, which disposes
912
>
* any armed timers, so this is a no-op when the client is active. The delay
913
>
* is the remaining grace measured from when the client disconnected — so a
914
>
* client that disconnected a while before the call was issued gets the
915
>
* residual window rather than a fresh one, and a stamp from a long-disconnected
916
>
* client fails promptly. Re-arms triggered by later orphaned tool calls in the
917
>
* same session shrink the remaining window instead of resetting it.
918
>
*/
919
>
private _startClientToolCallDisconnectTimeout(clientId: string, session: string, chatChannel: string): void {
921
>
if (!record) {
922
// Client is connected; the grace machinery does not apply.
923
return;
924
}
926
>
const elapsed = Date.now() - record.lastSeenAt;
927
>
const delay = Math.max(0, CLIENT_TOOL_CALL_DISCONNECT_TIMEOUT - elapsed);
928
>
record.disconnectTimeouts.set(chatChannel, disposableTimeout(() => {
929
>
this._releaseActiveClientForSession(session, clientId, chatChannel);
protocolServerHandler.ts ×2
931
>
}
933
>
/**
934
>
* Scan a chat for pending client tool calls owned by a disconnected client
935
>
* of this protocol server, and arm the disconnect timeout for each owner.
936
>
* Called when a `ChatToolCallStart` / `ChatToolCallReady` envelope is
937
>
* observed — covering calls issued for an already-gone client, which the
938
>
* live disconnect path never sees. Ownerless client tool calls (no client
939
>
* connected at stamp time) are failed immediately by the provider, so they
940
>
* never reach a pending state here. Unknown client ids are ignored because
941
>
* they may belong to another transport such as local IPC.
942
>
*/
943
>
private _checkOrphanedClientToolCalls(session: string, chatChannel: string): void {
945
>
const orphanOwners = new Set<string>();
946
>
for (const { clientId } of this._pendingClientToolCalls(state)) {
947
>
const ownerRecord = this._clients.get(clientId);
948
>
if (ownerRecord?.state === 'grace') {
950
>
}
952
>
for (const ownerId of orphanOwners) {
953
>
this._startClientToolCallDisconnectTimeout(ownerId, session, chatChannel);
protocolServerHandler.ts ×2
954
>
}
957
>
/**
958
>
* Register a freshly connected (or reconnected) transport for `clientId`,
959
>
* promoting the record to {@link IActiveClientRecord}. Promoting a grace
960
>
* record back to active disposes its pending disconnect timers: the
961
>
* disconnect-grace window only applies while the client has no live
962
>
* transport. This is the single place that maintains the "active records
963
>
* hold no grace timers" invariant.
964
>
*/
965
>
private _attachConnection(clientId: string, client: IConnectedClient): void {
967
>
if (existing?.state === 'active') {
970
>
existing?.disconnectTimeouts.dispose();
971
>
this._clients.set(clientId, { state: 'active', connections: [client] });
972
>
}
973
>
this._pruneClientRecords();
974
>
this._onDidChangeConnectionCount.fire(this._connectedClientCount);
975
>
}
977
>
/**
978
>
* Return the existing grace record for `clientId`, creating one for a
979
>
* never-connected client (an orphan tool-call stamp). Returns `undefined`
980
>
* when the client is currently active — the grace machinery does not apply
981
>
* to a connected client. A newly created record pins its grace clock to now.
982
>
*/
983
>
private _ensureGraceRecord(clientId: string): IGraceClientRecord | undefined {
985
>
if (record?.state === 'active') {
986
return undefined;
987
}
989
>
return record;
990
>
}
991
const created: IGraceClientRecord = { state: 'grace', lastSeenAt: Date.now(), disconnectTimeouts: new DisposableMap() };
992
this._clients.set(clientId, created);
993
return created;
996
>
private _getActiveClient(clientId: string): IConnectedClient | undefined {
997
>
return this._getActiveClientFromRecord(this._clients.get(clientId));
protocolServerHandler.ts ×3
998
>
}
1000
>
private _getActiveClientFromRecord(record: IClientRecord | undefined): IConnectedClient | undefined {
1003
>
}
1007
>
private _releaseClientSubscriptions(client: IConnectedClient, record: IActiveClientRecord): void {
1010
>
if (this._hasSubscriptionInOtherConnection(record, client, sub.uri)) {
protocolServerHandler.ts ×2
1012
>
}
1013
>
this._agentService.unsubscribe(URI.parse(sub.uri), client.clientId);
protocolServerHandler.ts ×2
1016
>
}
1019
>
}
1021
>
private _hasSubscriptionInOtherConnection(record: IClientRecord, client: IConnectedClient, uri: string): boolean {
1023
return false;
1024
}
1028
>
}
1031
>
}
1033
>
/** Number of clients that currently have a live connection. */
1034
>
private get _connectedClientCount(): number {
1036
>
for (const record of this._clients.values()) {
1037
>
if (record.state === 'active') {
1038
>
count++;
1039
>
}
1040
>
}
1041
>
return count;
1042
>
}
1044
>
/**
1045
>
* Drop grace records whose timers have all fired and whose last-seen time is
1046
>
* stale beyond the retention window (10× the disconnect timeout). This
1047
>
* covers both genuinely-disconnected clients and never-connected orphan
1048
>
* stamps. Bounds {@link _clients} without tracking liveness precisely — a
1049
>
* pruned-then-resurfacing stamp simply falls back to the full grace window.
1050
>
* Active records are never pruned; they persist until their last transport
1051
>
* closes.
1052
>
*/
1053
>
private _pruneClientRecords(): void {
1054
>
const cutoff = Date.now() - CLIENT_TOOL_CALL_DISCONNECT_TIMEOUT * 10;
protocolServerHandler.ts ×12
1055
>
for (const [clientId, record] of this._clients) {
1056
>
if (record.state === 'grace'
1057
&& record.disconnectTimeouts.size === 0
1059
this._clients.delete(clientId);
1060
}
1062
>
}
1064
>
private _clearClientToolCallDisconnectTimeout(clientId: string, channel: string): void {
1066
>
if (record?.state === 'grace') {
1068
>
}
1071
>
private _completeDisconnectedClientToolCalls(clientId: string, session: string, chatChannel: string): void {
1073
>
const activeTurn = state?.activeTurn;
1074
>
if (!state || !activeTurn) {
1075
return;
1076
}
1077
>
for (const { toolCall, clientId: ownerId } of this._pendingClientToolCalls(state)) {
protocolServerHandler.ts ×9
1078
>
if (ownerId !== clientId) {
1079
continue;
1080
}
1081
>
const mayRetryWithReplacementClient = this._hasReplacementActiveClientTool(state, clientId, toolCall.toolName);
protocolServerHandler.ts ×9
1082
>
if (toolCall.status === ToolCallStatus.Streaming) {
1084
>
type: ActionType.ChatToolCallReady,
1085
>
turnId: activeTurn.id,
1086
>
toolCallId: toolCall.toolCallId,
1087
>
invocationMessage: toolCall.invocationMessage ?? toolCall.displayName,
1088
>
confirmed: ToolCallConfirmationReason.NotNeeded,
1089
>
});
1090
>
}
1092
>
type: ActionType.ChatToolCallComplete,
1093
>
turnId: activeTurn.id,
1094
>
toolCallId: toolCall.toolCallId,
1095
>
result: {
1096
>
success: false,
1097
>
pastTenseMessage: `${toolCall.displayName} failed`,
1098
>
...(mayRetryWithReplacementClient ? { content: [{ type: ToolResultContentType.Text, text: `The client that was running ${toolCall.displayName} disconnected, but another active client now provides ${toolCall.displayName}. You may try calling the tool again.` }] } : {}),
1099
>
error: { message: `Client ${clientId} disconnected before completing ${toolCall.displayName}` },
1100
>
},
1101
>
});
1102
>
}
1103
>
}
1105
>
// ---- Requests (expect a response) ---------------------------------------
1106
>
1107
>
/**
1108
>
* Methods handled by the request dispatcher (excludes initialize/reconnect
1109
>
* which are handled during the handshake phase).
1110
>
*/
1111
>
private readonly _requestHandlers: RequestHandlerMap = {
1112
>
subscribe: async (client, params) => {
1114
>
if (!classified) {
1116
>
// level). Acknowledge as stateless so the client doesn't
1117
>
// hang, but install nothing.
1118
>
return {};
1119
>
}
1122
this._logService.warn(`[ProtocolServer] Ignoring OTLP subscribe for ${params.channel}: no OTLP emitter configured.`);
1123
return {};
1124
}
1126
>
return {};
1127
>
}
1129
>
const descriptor = this._agentService.onResourceWatchSubscribed(classified.uri);
protocolServerHandler.ts ×2
1130
>
if (!descriptor) {
1131
>
throw new ProtocolError(AHP_SESSION_NOT_FOUND, `Resource watch not found: ${params.channel}`);
protocolServerHandler.ts ×1
1132
>
}
1134
>
return {
1135
>
snapshot: {
1136
>
resource: classified.uri,
1137
>
state: descriptor,
1138
>
fromSeq: this._stateManager.serverSeq,
1139
>
},
1140
>
};
1141
>
}
1143
>
const snapshot = await this._agentService.subscribe(URI.parse(params.channel), client.clientId);
1144
>
client.subscriptions.set(classified.uri, classified);
1145
>
this._clearClientToolCallDisconnectTimeout(client.clientId, classified.uri);
1146
>
// `IStateSnapshot` is widened with `ChatState` (see sessionProtocol.ts);
1147
>
// the generated wire `Snapshot` union does not list it yet. The value
1148
>
// is JSON over the wire, so narrowing at this boundary is safe.
1149
>
return { snapshot: snapshot as SubscribeResult['snapshot'] };
1150
>
} catch (err) {
1151
if (err instanceof ProtocolError) {
1152
throw err;
1153
}
1154
throw new ProtocolError(AHP_SESSION_NOT_FOUND, `Resource not found: ${params.channel}`);
1155
}
1159
>
// Resolve fork turnId to a 0-based index using the source session's
1160
>
// turn list in the state manager.
1161
>
let fork: { session: URI; turnIndex: number; turnId: string } | undefined;
1162
>
if (params.fork) {
1163
const sourceState = this._stateManager.getSessionState(params.fork.session);
1164
if (!sourceState) {
1165
throw new ProtocolError(AHP_SESSION_NOT_FOUND, `Fork source session not found: ${params.fork.session}`);
1166
}
1167
const turnIndex = sourceState.turns.findIndex(t => t.id === params.fork!.turnId);
1168
if (turnIndex < 0) {
1169
throw new ProtocolError(AHP_SESSION_NOT_FOUND, `Fork turn ID ${params.fork.turnId} not found in session ${params.fork.session}`);
1170
}
1171
fork = { session: URI.parse(params.fork.session), turnIndex, turnId: params.fork.turnId };
1172
}
1173
>
// If the client eagerly claimed the active client role, validate
protocolServerHandler.ts ×4
1174
>
// the clientId matches the connection before forwarding.
1175
>
if (params.activeClient && params.activeClient.clientId !== _client.clientId) {
1176
>
throw new ProtocolError(JsonRpcErrorCodes.InvalidParams, `createSession.activeClient.clientId must match the connection's clientId`);
protocolServerHandler.ts ×1
1177
>
}
1179
>
createdSession = await this._agentService.createSession({
1180
>
provider: params.provider,
1181
>
workingDirectory: params.workingDirectories?.[0] ? URI.parse(params.workingDirectories[0]) : undefined,
protocolServerHandler.ts ×4
1182
>
session: URI.parse(params.channel),
1183
>
fork,
1184
>
config: params.config,
1185
>
activeClient: params.activeClient,
1186
>
progressToken: params.progressToken,
1187
>
});
1189
if (err instanceof ProtocolError) {
1190
throw err;
1191
}
1192
throw new ProtocolError(AHP_PROVIDER_NOT_FOUND, err instanceof Error ? err.message : String(err));
1193
}
1194
>
// Verify the provider honored the client-chosen session URI per the protocol contract
protocolServerHandler.ts ×4
1195
>
if (createdSession.toString() !== URI.parse(params.channel).toString()) {
1196
this._logService.warn(`[ProtocolServer] createSession: provider returned URI ${createdSession.toString()} but client requested ${params.channel}`);
1197
}
1201
await this._agentService.disposeSession(URI.parse(params.channel));
1202
return null;
1203
},
1205
>
const state = this._stateManager.getSessionState(params.channel);
protocolServerHandler.ts ×3
1206
>
if (!state) {
1207
>
throw new ProtocolError(AHP_SESSION_NOT_FOUND, `Session not found: ${params.channel}`);
protocolServerHandler.ts ×1
1208
>
}
1209
>
const defaultChat = state.defaultChat ?? buildDefaultChatUri(params.channel);
protocolServerHandler.ts ×1
1210
>
// The default chat is created alongside its session; creating it
protocolServerHandler.ts ×3
1211
>
// again is a no-op. Any other chat URI spins up an additional chat.
1212
>
if (URI.parse(params.chat).toString() === URI.parse(defaultChat).toString()) {
1214
>
}
1216
>
let options: IAgentCreateChatOptions | undefined;
1217
>
if (source) {
1219
>
case ChatSourceKind.Fork:
1220
>
options = { fork: { source: URI.parse(source.chat), turnId: source.turnId } };
protocolServerHandler.ts ×1
1221
>
break;
1224
>
sideChat: {
1225
>
source: URI.parse(source.chat),
1226
>
turnId: source.turnId,
1227
>
...(source.selection ? { selection: source.selection } : {}),
1228
>
},
1229
>
};
1230
>
break;
1232
>
throw new ProtocolError(JsonRpcErrorCodes.InvalidParams, `Unsupported createChat source kind: ${String((source as { kind?: unknown }).kind)}`);
protocolServerHandler.ts ×1
1234
>
}
1236
>
URI.parse(params.channel),
1237
>
URI.parse(params.chat),
1238
>
options,
1239
>
);
1240
>
return null;
1244
>
const parsed = parseChatUri(chat);
1245
>
if (!parsed) {
1246
return null;
1247
}
1248
>
await this._agentService.disposeChat(URI.parse(parsed.session), chat);
protocolServerHandler.ts ×2
1249
>
return null;
1250
>
},
1252
return this._agentService.resourceWrite(params);
1253
},
1256
>
const items = sessions.map(s => {
1257
>
const provider = AgentSession.provider(s.session);
1258
>
if (!provider) {
1259
throw new Error(`Agent session URI has no provider scheme: ${s.session.toString()}`);
1260
}
1262
>
let status = s.status ?? SessionStatus.Idle;
1263
>
if (s.isRead) {
1264
status |= SessionStatus.IsRead;
1265
}
1267
status |= SessionStatus.IsArchived;
1268
}
1270
>
resource: s.session.toString(),
1271
>
provider,
1272
>
title: s.summary ?? 'Session',
1273
>
status,
1274
>
activity: s.activity,
1275
>
createdAt: new Date(s.startTime).toISOString(),
1276
>
modifiedAt: new Date(s.modifiedTime).toISOString(),
1277
>
...(s.project ? { project: { uri: s.project.uri.toString(), displayName: s.project.displayName } } : {}),
1278
>
workingDirectories: s.workingDirectory ? [s.workingDirectory.toString()] : undefined,
1279
>
changes: s.changes,
1280
>
} satisfies ListSessionsResult['items'][number];
1281
>
});
1282
>
return { items };
1283
>
},
1285
return this._agentService.resolveSessionConfig({
1286
provider: params.provider,
1287
workingDirectory: params.workingDirectory ? URI.parse(params.workingDirectory) : undefined,
1288
config: params.config,
1289
});
1290
},
1292
return this._agentService.sessionConfigCompletions({
1293
provider: params.provider,
1294
workingDirectory: params.workingDirectory ? URI.parse(params.workingDirectory) : undefined,
1295
config: params.config,
1296
property: params.property,
1297
query: params.query,
1298
});
1299
},
1301
return this._agentService.completions(params);
1302
},
1304
const state = this._stateManager.getChatState(params.channel);
1305
if (!state) {
1306
throw new ProtocolError(AHP_SESSION_NOT_FOUND, `Session not found: ${params.channel}`);
1307
}
1308
if (params.cursor && params.cursor !== state.turnsNextCursor) {
1309
throw new ProtocolError(JsonRpcErrorCodes.InvalidParams, `Unrecognized fetchTurns cursor`);
1310
}
1311
this._stateManager.dispatchServerAction(params.channel, {
1312
type: ActionType.ChatTurnsLoaded,
1313
turns: [],
1314
});
1315
return {};
1316
},
1319
>
},
1322
>
},
1324
return this._agentService.resourceCopy(params);
1325
},
1327
return this._agentService.resourceDelete(params);
1328
},
1330
return this._agentService.resourceMove(params);
1331
},
1333
return this._agentService.resourceResolve(params);
1334
},
1336
return this._agentService.resourceMkdir(params);
1337
},
1339
return this._agentService.createResourceWatch(params);
1340
},
1342
// The local agent host does not yet enforce per-resource grants
1343
// for client → server access. Always grant; receivers MAY rescind
1344
// access by returning `PermissionDenied` on subsequent operations.
1345
return {};
1346
},
1350
throw new ProtocolError(AHP_AUTH_REQUIRED, `Authentication failed for resource: ${params.resource}`);
1351
}
1355
await this._agentService.createTerminal(params);
1356
return null;
1357
},
1359
await this._agentService.disposeTerminal(URI.parse(params.channel));
1360
return null;
1361
},
1363
return this._agentService.invokeChangesetOperation(params);
1364
},
1366
>
1367
>
1368
>
// ---- Reverse RPC (server → client requests) ----------------------------
1369
>
1370
>
private _reverseRequestId = 0;
1371
>
private readonly _pendingReverseRequests = new Map<number, { client: IConnectedClient; resolve: (value: unknown) => void; reject: (reason: unknown) => void }>();
1372
>
1373
>
/**
1374
>
* Sends a JSON-RPC request to a connected client and waits for the response.
1375
>
* Used for reverse-RPC operations like reading client-side files.
1376
>
* Rejects if the client disconnects or the server is disposed.
1377
>
*/
1378
>
private _sendReverseRequest<T>(clientId: string, method: string, params: unknown): Promise<T> {
1380
>
if (!client) {
1381
return Promise.reject(new Error(`Client ${clientId} is not connected`));
1382
}
1384
>
return new Promise<T>((resolve, reject) => {
1385
>
this._pendingReverseRequests.set(id, { client, resolve: resolve as (value: unknown) => void, reject });
1386
>
const request: JsonRpcRequest = { jsonrpc: '2.0', id, method, params };
1387
>
client.transport.send(request);
1388
>
});
1389
>
}
1391
>
/**
1392
>
* Rejects and clears all pending reverse-RPC requests sent over a given
1393
>
* connection.
1394
>
*/
1395
>
private _rejectPendingReverseRequestsForConnection(client: IConnectedClient): void {
1398
>
this._pendingReverseRequests.delete(id);
1399
>
pending.reject(new Error(`Client ${client.clientId} disconnected`));
1400
>
}
1401
>
}
1404
>
private _handleRequest(client: IConnectedClient, method: string, params: unknown, id: number): void {
1405
>
const handler = this._requestHandlers.hasOwnProperty(method) ? this._requestHandlers[method as RequestMethod] : undefined;
protocolServerHandler.ts ×4
1406
>
if (handler) {
1407
>
(handler as (client: IConnectedClient, params: unknown) => Promise<unknown>)(client, params).then(result => {
protocolServerHandler.ts ×3
1408
>
this._logService.trace(`[ProtocolServer] Request '${method}' id=${id} succeeded`);
protocolServerHandler.ts ×1
1409
>
client.transport.send(jsonRpcSuccess(id, result ?? null));
1412
>
this._logService.error(`[ProtocolServer] Request '${method}' failed`, err);
protocolServerHandler.ts ×2
1413
>
}
1416
>
return;
1417
>
}
1419
>
// VS Code extension methods (not in the typed protocol maps yet)
1420
>
const extensionResult = this._handleExtensionRequest(method, params);
1421
>
if (extensionResult) {
1423
>
client.transport.send(jsonRpcSuccess(id, result ?? null));
1424
>
}).catch(err => {
1425
this._logService.error(`[ProtocolServer] Extension request '${method}' failed`, err);
1426
client.transport.send(jsonRpcErrorFrom(id, err));
1428
>
return;
1429
>
}
1431
>
// MCP side-channel: requests targeting an `mcp://` channel carry the
1432
>
// channel URI in `params.channel`. We forward them through the
1433
>
// agent service, which routes by `<providerId>/<sessionId>/<serverName>`
1434
>
// to the owning agent's MCP App implementation. Unknown channels and
1435
>
// unknown methods are rejected with `-32601`.
1436
>
const mcpChannel = readMcpChannel(params);
1437
>
if (mcpChannel !== undefined) {
1438
const paramsObj = isParamsObject(params) ? params : undefined;
1439
this._agentService.handleMcpRequest(mcpChannel, method, paramsObj).then(result => {
1440
client.transport.send(jsonRpcSuccess(id, result ?? null));
1441
}).catch(err => {
1442
if (err instanceof Error && err.message.startsWith('Method not found')) {
1443
client.transport.send(jsonRpcError(id, JsonRpcErrorCodes.MethodNotFound, err.message));
1444
return;
1445
}
1446
this._logService.error(`[ProtocolServer] mcp:// request '${method}' on ${mcpChannel} failed`, err);
1447
client.transport.send(jsonRpcErrorFrom(id, err));
1448
});
1449
return;
1450
}
1452
>
client.transport.send(jsonRpcError(id, JsonRpcErrorCodes.MethodNotFound, `Method not found: ${method}`));
1455
>
/**
1456
>
* Handle VS Code extension methods that are not yet part of the typed
1457
>
* protocol. Returns a Promise if the method was recognized, undefined
1458
>
* otherwise.
1459
>
*/
1460
>
private _handleExtensionRequest(method: string, params: unknown): Promise<unknown> | undefined {
1463
>
}
1465
>
switch (method) {
1466
>
case 'shutdown':
1469
return this._agentService.getNetworkDiagnosticsInfo();
1473
return this._agentService.diagnosticsFetch((params as { url: string }).url);
1477
>
}
1479
>
// ---- Broadcasting -------------------------------------------------------
1480
>
1481
>
private _broadcastAction(envelope: ActionEnvelope): void {
1482
>
this._logService.trace(`[ProtocolServer] Broadcasting action: ${envelope.action.type}`);
protocolServerHandler.ts ×4
1483
>
const msg: AhpServerNotification<'action'> = { jsonrpc: '2.0', method: 'action', params: envelope };
1484
>
for (const record of this._clients.values()) {
1486
>
if (client && this._isRelevantToClient(client, envelope)) {
1488
>
}
1492
>
private _broadcastNotification(notification: INotification): void {
1493
>
// Each protocol notification now ships as its own top-level method. The
protocolServerHandler.ts ×3
1494
>
// `type` discriminant on our local {@link ProtocolNotification} union is
1495
>
// the wire-level method name, so we can route it directly.
1496
>
const { type, ...params } = notification;
1497
>
// eslint-disable-next-line local/code-no-dangerous-type-assertions
1498
>
const msg = { jsonrpc: '2.0', method: type, params } as AhpServerNotification;
1499
>
for (const record of this._clients.values()) {
1501
>
}
1504
>
/**
1505
>
* Forward an MCP server-originated notification (e.g.
1506
>
* `notifications/tools/list_changed`) over the AHP transport. The
1507
>
* `channel` field on `params` is the AHP routing envelope; the
1508
>
* receiving client demultiplexes by it. Notifications are broadcast
1509
>
* to every connected client — per-channel subscription filtering is
1510
>
* left to the client, since MCP notifications are cheap and the
1511
>
* client already knows which channels it cares about.
1512
>
*/
1513
>
private _broadcastMcpNotification(notification: IMcpNotification): void {
1514
const params: Record<string, unknown> = { ...(notification.params ?? {}), channel: notification.channel };
1515
// MCP notifications don't share a discriminated `method` literal
1516
// with the known {@link AhpServerNotification} union, so cast
1517
// through `unknown` to satisfy the transport contract.
1518
// eslint-disable-next-line local/code-no-dangerous-type-assertions
1519
const msg = { jsonrpc: '2.0' as const, method: notification.method, params } as unknown as AhpServerNotification;
1520
for (const record of this._clients.values()) {
1521
this._getActiveClientFromRecord(record)?.transport.send(msg);
1522
}
1523
}
1525
>
/**
1526
>
* Drop a subscription identified by `channel` from `client`. Handles
1527
>
* canonicalisation for OTLP URIs (so an `unsubscribe` with a URI
1528
>
* variant collapses to the same entry as the original `subscribe`)
1529
>
* and tears down the agent-service refcount for state channels.
1530
>
*/
1531
>
private _removeSubscription(client: IConnectedClient, channel: string): void {
1533
>
if (!classified) {
1534
// OTLP-flavoured URI with an unknown level — there can never
1535
// have been a matching subscription. Silently ignore.
1536
return;
1537
}
1539
>
if (!sub) {
1540
return;
1541
}
1543
>
if (sub.kind === ChannelKind.State) {
1545
>
if (record && this._hasSubscriptionInOtherConnection(record, client, sub.uri)) {
1546
return;
1547
}
1548
>
this._agentService.unsubscribe(URI.parse(sub.uri), client.clientId);
protocolServerHandler.ts ×3
1549
>
if (isAhpChatChannel(sub.uri)) {
1550
this._releaseActiveClientForSession(parseRequiredSessionUriFromChatUri(sub.uri), client.clientId, sub.uri);
1552
>
const state = this._stateManager.getSessionState(sub.uri);
1553
>
for (const chat of state?.chats ?? []) {
1554
>
this._releaseActiveClientForSession(sub.uri, client.clientId, chat.resource);
1555
>
}
1556
>
}
1559
>
}
1562
>
/**
1563
>
* Fan out an OTLP log record to every connected client that has
1564
>
* subscribed to a logs channel whose `{level}` band includes the
1565
>
* record's `severityNumber`. The notification's `channel` field is
1566
>
* the canonical URI the client subscribed against — clients can
1567
>
* route by URI without re-deriving the level.
1568
>
*/
1569
>
private _broadcastOtlpLog(record: IOtlpLogRecord): void {
1571
>
for (const clientRecord of this._clients.values()) {
1572
>
const client = this._getActiveClientFromRecord(clientRecord);
1573
>
if (!client) {
1575
>
}
1578
continue;
1579
}
1582
>
}
1584
>
jsonrpc: '2.0',
1585
>
method: 'otlp/exportLogs',
1586
>
params: { channel: sub.uri, payload },
1587
>
};
1588
>
client.transport.send(msg);
1589
>
}
1593
>
private _isRelevantToClient(client: IConnectedClient, envelope: ActionEnvelope): boolean {
1595
>
if (sub?.kind === ChannelKind.State || sub?.kind === ChannelKind.ResourceWatch) {
1597
>
}
1600
>
}
1601
>
return isActionEnvelopeRelevantToSubscriptionUris(envelope, this._stateAndResourceWatchUris(client));
protocolServerHandler.ts ×2
1604
>
private *_stateAndResourceWatchUris(client: IConnectedClient): Iterable<string> {
1606
>
if (sub.kind === ChannelKind.State || sub.kind === ChannelKind.ResourceWatch) {
1607
>
yield sub.uri;
1608
>
}
1609
>
}
1610
>
}
1612
>
override dispose(): void {
1613
>
for (const record of this._clients.values()) {
1616
>
connection.disposables.dispose();
1617
>
}
1620
>
}
1623
>
for (const [, pending] of this._pendingReverseRequests) {
1624
pending.reject(new Error('ProtocolServerHandler disposed'));
1625
}
1627
>
this._replayBuffer.length = 0;
1628
>
super.dispose();
1629
>
}
1630
>
}