src/vs/platform/agentHost/node/claude/claudeSdkPipeline.ts
703 LOC · 669 covered · 34 uncovered · 115 ranges · 398 concepts · 45 introducers · 217 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
>
/*---------------------------------------------------------------------------------------------
claudeSdkPipeline.ts ×26
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 { AgentInfo, McpServerStatus, PermissionMode, Query, SDKUserMessage, SlashCommand, WarmQuery } from '@anthropic-ai/claude-agent-sdk';
7
>
import { CancellationError, isCancellationError } from '../../../../base/common/errors.js';
8
>
import { Emitter, Event } from '../../../../base/common/event.js';
9
>
import { Disposable, IReference, toDisposable } from '../../../../base/common/lifecycle.js';
10
>
import { StopWatch } from '../../../../base/common/stopwatch.js';
11
>
import { URI } from '../../../../base/common/uri.js';
12
>
import { IInstantiationService } from '../../../instantiation/common/instantiation.js';
13
>
import { ILogService } from '../../../log/common/log.js';
14
>
import { ClaudeRuntimeEffortLevel } from '../../common/claudeModelConfig.js';
15
>
import { AgentSignal } from '../../common/agentService.js';
16
>
import { ISessionDatabase } from '../../common/sessionDataService.js';
17
>
import { ActionType } from '../../common/state/sessionActions.js';
18
>
import { DeferredPromise } from '../../../../base/common/async.js';
19
>
import { ClaudePromptQueue, IPendingSdkMessage } from './claudePromptQueue.js';
20
>
import { ClaudeSdkMessageRouter } from './claudeSdkMessageRouter.js';
21
>
import type { SubagentRegistry } from './claudeSubagentRegistry.js';
22
>
23
>
/**
24
>
* Callback the agent supplies via {@link ClaudeSdkPipeline.attachRematerializer}
25
>
* so the pipeline can rebuild its underlying {@link WarmQuery} /
26
>
* {@link AbortController} on abort or crash recovery without depending on
27
>
* the materializer service directly. The callback MUST start the SDK in
28
>
* `resume` mode (i.e. pass `Options.resume = sessionId` instead of
29
>
* `Options.sessionId`) and MUST NOT re-fire the agent's
30
>
* `onDidMaterializeSession` event — that event is once-per-provisional
31
>
* promotion (see `claudeAgent.ts` materialize path).
32
>
*/
33
>
export interface IRematerializer {
34
>
(reason: 'restart' | 'recover'): Promise<{ readonly warm: WarmQuery; readonly abortController: AbortController }>;
35
>
}
36
>
37
>
/**
38
>
* Owns one SDK Query lifecycle for a Claude session. Knows nothing about
39
>
* protocol turns, the workbench mapper, file-edit observers, or
40
>
* permission registries — the consuming session subscribes to
41
>
* {@link onDidProduceSignal} and fans out to its own collaborators.
42
>
*
43
>
* Responsibilities:
44
>
* • Hold the {@link WarmQuery} + {@link AbortController} for the
45
>
* active SDK subprocess. Both are mutable: rebind on abort/crash
46
>
* recovery via the supplied {@link IRematerializer}.
47
>
* • Drive a {@link ClaudePromptQueue} whose iterable is handed to
48
>
* `WarmQuery.query()`.
49
>
* • Apply the current model / effort / permissionMode to the SDK
50
>
* eagerly when the consumer calls {@link setModel} /
51
>
* {@link setEffort} / {@link setPermissionMode}. The SDK only takes
52
>
* these into account on the NEXT user request, so mid-turn calls
53
>
* are safe — no need to align the SDK setter with the prompt yield.
54
>
* Re-applied to a fresh Query on rebind.
55
>
* • Drain the SDK message stream, dispatch each message to the
56
>
* {@link ClaudeSdkMessageRouter}, settle the matching entry's
57
>
* deferred on `result`, and emit `ChatTurnComplete` only when
58
>
* the queue fully drains (intermediate results during steering
59
>
* preemption do NOT fire turn-complete — CONTEXT.md M10).
60
>
*
61
>
* Disposing the pipeline aborts the controller (terminating the SDK
62
>
* subprocess per `sdk.d.ts:982`) and async-disposes the WarmQuery.
63
>
*/
64
>
/**
65
>
* Snapshot of everything the SDK has currently resolved for this
66
>
* session. Returned by {@link ClaudeSdkPipeline.snapshotResolvedCustomizations}.
67
>
*/
68
>
export interface ISdkResolvedCustomizations {
69
>
readonly commands: readonly SlashCommand[];
70
>
readonly agents: readonly AgentInfo[];
71
>
readonly mcpServers: readonly McpServerStatus[];
72
>
/**
73
>
* Native plugins the live session actually loaded, as reported by the
74
>
* SDK `system/init` message. Used to filter the disk-discovered native
75
>
* plugins post-materialize: a plugin declared in `enabledPlugins` but
76
>
* absent here (bad path, manifest error, untrusted workspace) is hidden.
77
>
*
78
>
* `source` is the plugin id (`<plugin>@<marketplace>`) and is the
79
>
* authoritative match key — the SDK's `path` is unreliable for
80
>
* workspace-`local`-scoped plugins (it can report a non-cache path). The
81
>
* SDK `.d.ts` types the element as `{ name, path }` but the runtime adds
82
>
* `source`, so it is captured as optional.
83
>
*/
84
>
readonly plugins: readonly { readonly name: string; readonly path: string; readonly source?: string }[];
85
>
}
86
>
87
>
export class ClaudeSdkPipeline extends Disposable {
88
>
/**
89
>
* Phase 11 — hot-swap the SDK's plugin set in place via
90
>
* `Query.reloadPlugins()`. Commands / agents / mcpServers added or
91
>
* removed by the new plugin set become visible to the SDK
92
>
* immediately, without a session restart. Throws if the query is
93
>
* not yet bound (session not materialized).
94
>
*/
95
>
async reloadPlugins(): Promise<void> {
96
>
const query = await this._ensureQueryBound();
97
>
await query.reloadPlugins();
98
>
}
99
>
100
>
/**
101
>
* Phase 11 — snapshot the SDK's currently-resolved customization
102
>
* surface (slash commands / skills, subagents, MCP servers). This
103
>
* is the SDK's view of "what does this session actually have
104
>
* access to right now" — covers everything the SDK loaded itself
105
>
* (`~/.claude/**`, `.claude/agents/`, `settings.json` MCP) AND
106
>
* anything we fed in via `Options.plugins`. The host overlays
107
>
* client-side enablement separately.
108
>
*/
109
>
async snapshotResolvedCustomizations(): Promise<ISdkResolvedCustomizations> {
111
>
const [commands, agents, mcpServers] = await Promise.all([
112
>
query.supportedCommands(),
113
>
query.supportedAgents(),
114
>
query.mcpServerStatus(),
115
>
]);
119
>
async startMcpServer(serverName: string): Promise<boolean> {
120
const query = await this._ensureQueryBound();
121
return this._applyMcpServerEnablement(query, serverName, true);
122
}
124
>
async stopMcpServer(serverName: string): Promise<boolean> {
125
const query = await this._ensureQueryBound();
126
return this._applyMcpServerEnablement(query, serverName, false);
127
}
129
>
async reconcileMcpServerEnablement(desired: ReadonlyMap<string, boolean>): Promise<boolean> {
131
>
const observed = new Map((await query.mcpServerStatus()).map(server => [server.name, server.status !== 'disabled']));
132
>
for (const [serverName, enabled] of desired) {
133
>
if (observed.get(serverName) === enabled) {
134
>
continue;
135
>
}
136
>
if (!await this._applyMcpServerEnablement(query, serverName, enabled)) {
137
return false;
138
}
140
>
return true;
141
>
}
143
>
private async _applyMcpServerEnablement(query: Query, serverName: string, enabled: boolean): Promise<boolean> {
144
>
if (!query.toggleMcpServer || (enabled && !query.reconnectMcpServer)) {
claudeSdkPipeline.ts ×5
145
return false;
146
}
148
>
if (enabled) {
149
>
await query.reconnectMcpServer!(serverName);
150
>
}
151
>
return true;
152
>
}
154
>
/**
155
>
* Bind the SDK Query if needed, recovering a dead one first. Mirrors the
156
>
* gate in {@link send}: if the pipeline is marked for rebind (after an
157
>
* abort/crash the `_query` handle is retained for teardown but its stream
158
>
* is dead), rebuild via the rematerializer so pre-flight helpers never
159
>
* operate on a disposed stream. Then lazily bind if nothing is bound yet.
160
>
*/
161
>
private async _ensureQueryBound(): Promise<Query> {
164
>
}
167
>
await this._replayCurrentConfig();
168
>
}
170
>
}
172
>
/**
173
>
* Bind a fresh SDK stream off the current warm subprocess. The stream is
174
>
* long-lived: it spans every turn until a rebind swaps the subprocess (the
175
>
* prompt iterable parks between turns rather than ending), so {@link _query}
176
>
* tracks the lifetime of {@link _warm} and is only swapped here.
177
>
*/
178
>
private _bindWarmQuery(): Query {
180
>
this._query = query;
181
>
return query;
182
>
}
184
>
/**
185
>
* The SDK stream bound to the current {@link _warm} subprocess, or
186
>
* `undefined` before the first bind. Health is tracked separately by
187
>
* {@link _needsRebind}: a non-`undefined` `_query` with `_needsRebind`
188
>
* set is a *dead* stream awaiting rebuild. Cleared only on dispose.
189
>
*/
190
>
private _query: Query | undefined;
191
>
private _warm: WarmQuery;
192
>
private _abortController: AbortController;
193
>
194
>
private readonly _queue: ClaudePromptQueue;
195
>
196
>
/** Flips to `true` on the first `system:init` SDK message. Drives `Options.resume` decisions for downstream phases. */
197
>
private _isResumed = false;
198
>
199
>
/**
200
>
* Native plugins reported by the most recent `system:init` message.
201
>
* Captured on *every* init (including resume) so the post-materialize
202
>
* native-plugin filter always reflects the live set. `source` is the
203
>
* plugin id and is the reliable match key (see {@link ISdkResolvedCustomizations}).
204
>
*/
205
>
private _initPlugins: readonly { readonly name: string; readonly path: string; readonly source?: string }[] = [];
206
>
207
>
/** Last model / effort / permission mode applied to the SDK via the runtime setters. Reset on rebind. */
208
>
private _appliedModel: string | undefined;
209
>
private _appliedEffort: ClaudeRuntimeEffortLevel | undefined;
210
>
private _appliedPermissionMode: PermissionMode | undefined;
211
>
212
>
/** Current values the consumer has asked for. Replayed to a fresh Query on bind / rebind. */
213
>
private _currentModel: string | undefined;
214
>
private _currentEffort: ClaudeRuntimeEffortLevel | undefined;
215
>
private _currentPermissionMode: PermissionMode | undefined;
216
>
217
>
private _rematerializer: IRematerializer | undefined;
218
>
219
>
/** Set when the consumer loop ends in error (cancellation OR crash). Read by {@link send} to trigger rebind. */
220
>
private _needsRebind = false;
221
>
222
>
/** Tracks whether the consumer loop is currently draining {@link _query}. */
223
>
private _consumerLoopRunning = false;
224
>
225
>
private readonly _onDidProduceSignal = this._register(new Emitter<AgentSignal>());
226
>
/**
227
>
* Single fan-out for every {@link AgentSignal} this session produces:
228
>
* • Router-mapped per-message signals (response parts, tool calls,
229
>
* pending confirmations, etc.).
230
>
* • `ChatTurnComplete` action, fired when the LAST entry in the
231
>
* queue drains via `result` (intermediate results during steering
232
>
* preempt do NOT fire — CONTEXT.md M10).
233
>
* • `steering_consumed` signal, fired the moment the iterable yields
234
>
* a steering entry to the SDK.
235
>
*/
236
>
readonly onDidProduceSignal: Event<AgentSignal> = this._onDidProduceSignal.event;
237
>
238
>
private readonly _router: ClaudeSdkMessageRouter;
239
>
240
>
constructor(
242
>
readonly sessionUri: URI,
243
>
readonly chatChannelUri: URI,
244
>
warm: WarmQuery,
245
>
abortController: AbortController,
246
>
dbRef: IReference<ISessionDatabase>,
247
>
subagents: SubagentRegistry,
248
>
clientToolOwner: ((toolName: string) => string | undefined) | undefined = undefined,
249
>
@IInstantiationService instantiationService: IInstantiationService,
250
>
@ILogService private readonly _logService: ILogService,
251
>
) {
252
>
super();
253
>
this._warm = warm;
254
>
this._abortController = abortController;
255
>
this._wireAbortHandler(abortController);
256
>
this._queue = this._register(instantiationService.createInstance(
257
>
ClaudePromptQueue,
258
>
sessionId,
259
>
() => this._abortController.signal,
260
>
(pendingId: string) => this._onDidProduceSignal.fire({
262
>
chat: this.chatChannelUri,
263
>
id: pendingId,
264
>
}),
266
>
this._router = this._register(instantiationService.createInstance(
267
>
ClaudeSdkMessageRouter, sessionUri, chatChannelUri, dbRef, subagents, clientToolOwner,
268
>
));
269
>
this._register(this._router.onDidProduceSignal(s => this._onDidProduceSignal.fire(s)));
270
>
// Dispose chain → abort → SDK cleanup. Reads the *current*
271
>
// `_abortController` so a swap aborts the live subprocess.
272
>
this._register(toDisposable(() => this._abortController.abort()));
273
>
this._register(toDisposable(() => {
274
>
void Promise.resolve(this._warm[Symbol.asyncDispose]()).catch((err: unknown) =>
275
>
this._logService.warn(`[ClaudeSdkPipeline] WarmQuery dispose failed: ${err}`));
276
>
}));
277
>
}
279
>
get isResumed(): boolean { return this._isResumed; }
280
>
281
>
get isAborted(): boolean { return this._abortController.signal.aborted; }
282
>
283
>
/**
284
>
* Whether a turn is currently in flight or queued. False between turns (the
285
>
* warm query parks with a drained queue). Used by non-destructive idle
286
>
* release to avoid tearing the pipeline down mid-turn.
287
>
*/
288
>
get hasActiveTurn(): boolean { return !this._queue.isEmpty; }
289
>
290
>
/**
291
>
* Abort the live SDK subprocess and **await its actual exit**.
292
>
*
293
>
* `WarmQuery[Symbol.asyncDispose]()` calls the query's `close()`, which
294
>
* *fires* the SDK cleanup but does not await it — so it returns while the
295
>
* subprocess is still shutting down (and still re-flushing its transcript).
296
>
* `Query.return()` awaits the same (memoized) cleanup, which in turn awaits
297
>
* `transport.waitForExit()` — the OS process actually exiting after its
298
>
* final transcript flush. Awaiting that is what lets a caller safely reuse
299
>
* the `--session-id` (the CLI rejects a fresh spawn while `<id>.jsonl`
300
>
* still exists, and the dying process would otherwise recreate it).
301
>
*/
302
>
async shutdownAndWait(): Promise<void> {
304
>
try {
305
>
await this._warm[Symbol.asyncDispose]();
306
>
await this._query?.return(undefined);
307
>
} catch (err) {
308
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] shutdownAndWait: teardown failed`, err);
309
}
312
>
/**
313
>
* Phase 10 \u2014 narrow public wrapper around the internal
314
>
* {@link _rebindQuery} so {@link ClaudeAgentSession.rebindForClientTools}
315
>
* can drive a yield-restart without exposing the private rebind
316
>
* machinery to every collaborator.
317
>
*/
318
>
rebindForRestart(): Promise<void> {
320
>
}
322
>
/**
323
>
* Phase 10 — update the resolver the stream mapper uses to stamp the
324
>
* owning workbench `clientId` onto subsequent `ChatToolCallStart` events.
325
>
*/
326
>
setClientToolOwner(clientToolOwner: ((toolName: string) => string | undefined) | undefined): void {
327
this._router.setClientToolOwner(clientToolOwner);
328
}
330
>
/** Attach the rematerializer hook for abort / crash recovery. Optional — tests that exercise only the dispose path skip this. */
331
>
attachRematerializer(rematerializer: IRematerializer): void {
333
>
}
335
>
/**
336
>
* Seed the current + applied config from materialize-time `Options`.
337
>
* The SDK already starts with these values, so we mark them as both
338
>
* "current" (what the consumer wants) and "applied" (what the SDK has)
339
>
* to avoid a redundant `setModel` / `applyFlagSettings` on first use.
340
>
*/
341
>
seedCurrentConfig(model: string | undefined, effort: ClaudeRuntimeEffortLevel | undefined, permissionMode: PermissionMode | undefined): void {
343
>
this._currentEffort = effort;
344
>
this._currentPermissionMode = permissionMode;
345
>
this._appliedModel = model;
346
>
this._appliedEffort = effort;
347
>
this._appliedPermissionMode = permissionMode;
348
>
}
350
>
/**
351
>
* Eagerly push a model change to the SDK. Safe to call mid-turn:
352
>
* `Query.setModel` only takes effect on the NEXT user request. No-op
353
>
* if the value is unchanged. Buffered as `_currentModel` until the
354
>
* Query is bound (and replayed on rebind).
355
>
*/
356
>
async setModel(model: string): Promise<void> {
358
>
if (this._query && !this._needsRebind && model !== this._appliedModel) {
360
>
await this._query.setModel(model);
361
>
this._appliedModel = model;
362
>
} catch (err) {
363
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] setModel failed: ${err}`);
364
}
368
>
/**
369
>
* Eagerly push an effort-level change to the SDK via
370
>
* `applyFlagSettings({ effortLevel })`. Same mid-turn safety as
371
>
* {@link setModel}.
372
>
*
373
>
* `undefined` means "clear the effort the SDK is currently applying" —
374
>
* issued as `applyFlagSettings({ effortLevel: null })` (sdk.d.ts:2263:
375
>
* passing `null` clears a key from the flag layer). This is what makes a
376
>
* switch to a model that does not support reasoning effort (e.g. Haiku)
377
>
* drop a `'high'` left over from a prior effort-capable model instead of
378
>
* replaying it onto a model the API will 400 on.
379
>
*/
380
>
async setEffort(effort: ClaudeRuntimeEffortLevel | undefined): Promise<void> {
382
>
if (this._query && !this._needsRebind && effort !== this._appliedEffort) {
384
>
await this._query.applyFlagSettings({ effortLevel: effort ?? null });
385
>
this._appliedEffort = effort;
386
>
} catch (err) {
387
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] setEffort failed: ${err}`);
388
}
392
>
/**
393
>
* Queue a user prompt for the SDK. Resolves when the matching
394
>
* `result` message arrives.
395
>
*
396
>
* If a previous turn aborted or crashed, this triggers a rebind via
397
>
* the attached rematerializer before queueing.
398
>
*/
399
>
async send(prompt: SDKUserMessage, turnId: string): Promise<void> {
404
throw new CancellationError();
405
}
408
>
await this._replayCurrentConfig();
409
>
}
411
>
const entry: IPendingSdkMessage = {
412
>
sdkMessage: prompt,
414
>
turnId,
415
>
stopWatch: StopWatch.create(false),
416
>
deferred: new DeferredPromise<void>(),
417
>
};
418
>
return this._queue.push(entry);
419
>
}
421
>
/**
422
>
* Push a `priority: 'now'` steering message into the iterable. The
423
>
* caller pre-builds the {@link SDKUserMessage} (the pipeline is SDK
424
>
* messaging-shaped, not protocol-shaped). `pendingMessageId` is the
425
>
* protocol `PendingMessage.id` that {@link onSteeringConsumed} will
426
>
* carry when the SDK accepts the message.
427
>
*
428
>
* No-op if the pipeline is aborted or no in-flight / queued request
429
>
* exists to inherit a `turnId` from (CONTEXT.md M10: steering folds
430
>
* into the in-progress protocol Turn).
431
>
*/
432
>
injectSteering(prompt: SDKUserMessage, pendingMessageId: string): void {
434
this._logService.warn(`[Claude:${this.sessionId}] injectSteering: dropped (controller aborted) id=${pendingMessageId}`);
435
return;
436
}
438
>
if (!parent) {
439
>
this._logService.warn(`[Claude:${this.sessionId}] injectSteering: dropped (no in-flight turn) id=${pendingMessageId}`);
claudeAgent.ts ×1
440
>
return;
441
>
}
442
>
const sdkUuid = typeof prompt.uuid === 'string' ? prompt.uuid : pendingMessageId;
claudeSdkPipeline.ts ×3
443
>
// Steering deferreds aren't observed by anyone (the agent's send
444
>
// promise is the original entry's deferred); attach a no-op catch
445
>
// so a `failAll` rejection on abort/crash doesn't surface as an
446
>
// unhandled rejection.
447
>
this._queue.push({
448
>
sdkMessage: prompt,
449
>
sdkUuid,
450
>
turnId: parent.turnId,
451
>
stopWatch: parent.stopWatch,
452
>
deferred: new DeferredPromise<void>(),
453
>
steeringPendingId: pendingMessageId,
454
>
}).catch(() => { /* expected on abort/crash */ });
455
>
this._logService.info(`[Claude:${this.sessionId}] injectSteering: enqueued id=${pendingMessageId} sdkUuid=${sdkUuid}`);
456
>
}
458
>
/**
459
>
* Cancel the in-flight SDK turn via the abort controller. Mirrors
460
>
* the production reference (`claudeCodeAgent.ts:719`). Drops every
461
>
* pending entry's deferred (rejected with `CancellationError`),
462
>
* marks the pipeline for rebind on next {@link send}. Idempotent.
463
>
*
464
>
* Safe to call during rebind: {@link _rebindQuery} swaps in a fresh
465
>
* placeholder {@link AbortController} before awaiting the
466
>
* rematerializer, so an abort issued during recovery lands on that
467
>
* placeholder and is honored when the freshly-built pair arrives
468
>
* (the rebind discards the new pair and surfaces a cancellation).
469
>
*/
470
>
abort(): void {
473
>
}
475
>
this._queue.failAll(new CancellationError());
476
>
// Mark unhealthy but keep the `_query` handle: the next `send` rebinds,
477
>
// and `shutdownAndWait` still needs it to await the subprocess exit.
478
>
this._needsRebind = true;
479
>
}
481
>
/**
482
>
* Forwards to {@link Query.setPermissionMode} once the query is
483
>
* bound; the value is also remembered so it's re-applied after a
484
>
* rebind. Permission mode is whole-session (not per-entry).
485
>
*/
486
>
async setPermissionMode(mode: PermissionMode): Promise<void> {
488
>
if (this._query && !this._needsRebind && mode !== this._appliedPermissionMode) {
490
>
this._appliedPermissionMode = mode;
491
>
}
494
>
private _wireAbortHandler(controller: AbortController): void {
496
>
this._queue.notifyAborted();
497
>
}, { once: true });
498
>
}
500
>
private _ensureConsumerLoop(): void {
503
>
}
505
>
this._runConsumerLoop();
506
>
}
508
>
/**
509
>
* Runs one {@link _processMessages} pass over the live {@link _query} and,
510
>
* when it ends, decides whether to hand off to a fresh pass.
511
>
*
512
>
* A rebind ({@link _rebindQuery}) swaps in a new `_query` while the loop is
513
>
* still draining the OLD (now-disposed) one; that old pass then ends with
514
>
* the "stream ended without a result" guard. Because `_consumerLoopRunning`
515
>
* stays `true` for the whole handoff, the {@link send} that queued the
516
>
* post-rebind prompt already saw {@link _ensureConsumerLoop} no-op — so if
517
>
* this pass just stopped, nothing would ever read the new query and `send`
518
>
* would hang. Detect the swap (current `_query` differs from the one this
519
>
* pass bound) and re-arm for it instead. Abort / crash / dispose leave
520
>
* `_query` cleared (or the store disposed), so they fall through to stop.
521
>
*/
522
>
private _runConsumerLoop(): void {
524
>
void this._processMessages()
525
>
.catch(err => this._logService.error(`[ClaudeSdkPipeline:${this.sessionId}] _processMessages crashed: ${err}`))
526
>
.finally(() => {
527
>
if (!this._store.isDisposed && this._query && this._query !== boundQuery) {
530
>
this._consumerLoopRunning = false;
531
>
}
532
>
});
533
>
}
535
>
/**
536
>
* Push the current model / effort / permissionMode to the SDK if they
537
>
* diverge from what was last applied. Called after binding a fresh
538
>
* Query (initial first-send and after rebind). Failures are logged.
539
>
*/
540
>
private async _replayCurrentConfig(): Promise<void> {
542
>
if (this._currentModel !== undefined && this._currentModel !== this._appliedModel) {
544
>
this._appliedModel = this._currentModel;
545
>
}
546
>
if (this._currentEffort !== undefined && this._currentEffort !== this._appliedEffort) {
claudeSdkPipeline.ts ×17
547
>
await this._query?.applyFlagSettings({ effortLevel: this._currentEffort });
claudeSdkPipeline.ts ×1
548
>
this._appliedEffort = this._currentEffort;
549
>
}
550
>
if (this._currentPermissionMode !== undefined && this._currentPermissionMode !== this._appliedPermissionMode) {
claudeSdkPipeline.ts ×17
552
>
this._appliedPermissionMode = this._currentPermissionMode;
553
>
}
555
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] _replayCurrentConfig failed: ${err}`);
556
}
559
>
/**
560
>
* Dispose the dead SDK plumbing and rebuild via the agent-supplied
561
>
* rematerializer in `resume` mode. Re-applies the current model /
562
>
* effort / permission mode to the fresh Query.
563
>
*/
564
>
private async _rebindQuery(reason: 'restart' | 'recover'): Promise<void> {
566
>
throw new Error(`ClaudeSdkPipeline.rebind: no rematerializer attached (reason=${reason})`);
claudeSdkPipeline.ts ×1
567
>
}
569
>
// Install a placeholder controller BEFORE awaiting the
570
>
// rematerializer so a concurrent {@link abort} has a live target
571
>
// instead of returning early as idempotent against the already-
572
>
// aborted old controller.
573
>
const placeholder = new AbortController();
574
>
this._abortController = placeholder;
575
>
const built = await this._rematerializer(reason);
577
>
// The dispose chain has already torn down the OLD warm/controller;
578
>
// the freshly-built pair would otherwise leak its subprocess. Mirror
579
>
// the post-await abort gate in `_materializeProvisional`.
580
>
if (this._store.isDisposed) {
581
built.abortController.abort();
582
void Promise.resolve(built.warm[Symbol.asyncDispose]()).catch((err: unknown) =>
583
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] rebind-after-dispose: warm dispose failed: ${err}`));
584
throw new CancellationError();
585
}
587
>
// the placeholder. Discard the freshly-built pair and surface a
588
>
// cancellation to the in-flight `send`.
589
>
if (placeholder.signal.aborted) {
591
>
void Promise.resolve(built.warm[Symbol.asyncDispose]()).catch((err: unknown) =>
592
>
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] rebind-aborted: warm dispose failed: ${err}`));
593
>
void Promise.resolve(oldWarm[Symbol.asyncDispose]()).catch((err: unknown) =>
594
>
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] previous WarmQuery dispose failed during aborted rebind: ${err}`));
595
>
this._queue.failAll(new CancellationError());
596
>
this._needsRebind = true;
597
>
throw new CancellationError();
598
>
}
599
>
void Promise.resolve(oldWarm[Symbol.asyncDispose]()).catch((err: unknown) =>
claudeSdkPipeline.ts ×1
600
>
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] previous WarmQuery dispose failed during rebind: ${err}`));
601
>
this._warm = built.warm;
602
>
this._abortController = built.abortController;
603
>
this._wireAbortHandler(built.abortController);
604
>
this._queue.resetForRebind();
605
>
this._needsRebind = false;
606
>
// New SDK starts with the materializer's `Options.model` / effort /
607
>
// permissionMode but we don't trust that to match `_currentModel`
608
>
// etc. — reset the applied cache and let `_replayCurrentConfig`
609
>
// push whatever the consumer last set.
610
>
this._appliedModel = undefined;
611
>
this._appliedEffort = undefined;
612
>
this._appliedPermissionMode = undefined;
613
>
this._bindWarmQuery();
614
>
await this._replayCurrentConfig();
617
>
/**
618
>
* Consumer loop. Drains the SDK iterator, dispatches each message
619
>
* to the {@link ClaudeSdkMessageRouter} (awaited so async file-edit
620
>
* observation completes before the next message), settles the head
621
>
* entry's deferred on `result`, and fires `ChatTurnComplete` only
622
>
* when the queue fully drains.
623
>
*
624
>
* On any uncaught error (cancellation, transport failure, or the
625
>
* post-loop "stream ended without result" guard) the catch block
626
>
* rejects every pending entry's deferred with the same error and
627
>
* marks `_needsRebind=true`. Cancellation is swallowed (don't
628
>
* rethrow); other errors propagate to the void caller's `.catch` for
629
>
* logging.
630
>
*/
631
>
private async _processMessages(): Promise<void> {
633
>
if (!query) {
634
throw new Error('ClaudeSdkPipeline._processMessages called before query was bound');
635
}
637
>
for await (const message of query) {
640
>
}
643
>
// resume / post-rebind) so the post-materialize filter is fresh.
644
>
this._initPlugins = message.plugins ?? [];
645
>
if (!this._isResumed) {
646
>
this._isResumed = true;
647
>
}
648
>
}
649
>
const turnId = this._queue.peekParent()?.turnId;
651
>
try {
652
>
await this._router.handle(message, turnId, turnDuration);
654
this._logService.warn(`[ClaudeSdkPipeline:${this.sessionId}] router threw, skipping: ${handlerErr}`);
655
}
658
>
this._logService.info(`[Claude:${this.sessionId}] result for sdkUuid=${completed?.sdkUuid}`);
659
>
// Final result: queue fully drained → protocol turn done.
660
>
// Intermediate result (still pending entries from a
661
>
// steering preempt) does NOT fire ChatTurnComplete.
662
>
if (completed && this._queue.isEmpty) {
663
>
this._onDidProduceSignal.fire({
664
>
kind: 'action',
665
>
resource: this.chatChannelUri,
666
>
action: {
667
>
type: ActionType.ChatTurnComplete,
668
>
turnId: completed.turnId,
669
>
duration: Math.max(0, completed.stopWatch.elapsed()),
670
>
},
671
>
});
672
>
}
673
>
}
677
>
}
679
>
// disposes the old one, ending THIS pass's stream cleanly. That is
680
>
// expected — return quietly and let {@link _runConsumerLoop} hand
681
>
// off to the new query. Only an unexpected end of the *current*
682
>
// query (no swap) is the real "stream ended without a result"
683
>
// failure that should mark the pipeline for recovery.
684
>
if (this._query !== query) {
686
>
}
689
>
const fatal = err instanceof Error ? err : new Error(String(err));
690
>
// Only the loop that still owns the live query reacts: a later
691
>
// unwinding pass whose query was already swapped by a rebind must
692
>
// not clobber the fresh one. Mark unhealthy (keep the handle for
693
>
// teardown); the next `send` rebinds.
694
>
if (this._query === query) {
695
>
this._queue.failAll(fatal);
696
>
this._needsRebind = true;
697
>
}
698
>
if (!isCancellationError(fatal)) {
700
>
}
702
>
}