src/vs/platform/agentHost/node/claude/claudePromptQueue.ts
171 LOC · 171 covered · 0 uncovered · 38 ranges · 428 concepts · 22 introducers · 229 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
>
/*---------------------------------------------------------------------------------------------
claudePromptQueue.ts ×10
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 { SDKUserMessage } from '@anthropic-ai/claude-agent-sdk';
7
>
import { DeferredPromise } from '../../../../base/common/async.js';
8
>
import { Disposable } from '../../../../base/common/lifecycle.js';
9
>
import { StopWatch } from '../../../../base/common/stopwatch.js';
10
>
import { ILogService } from '../../../log/common/log.js';
11
>
12
>
/**
13
>
* One {@link SDKUserMessage} the queue has handed to (or is about to
14
>
* hand to) the SDK. Lifecycle:
15
>
* 1. Created by the caller and pushed via {@link ClaudePromptQueue.push}.
16
>
* 2. Shifted off the to-yield list and pushed to the yielded list when
17
>
* the prompt iterable hands it to the SDK.
18
>
* 3. Shifted off the yielded list and {@link deferred} settled when
19
>
* the matching SDK `result` message arrives (via
20
>
* {@link ClaudePromptQueue.settleHead}).
21
>
*/
22
>
export interface IPendingSdkMessage {
23
>
readonly sdkMessage: SDKUserMessage;
24
>
readonly sdkUuid: string;
25
>
readonly turnId: string;
26
>
readonly stopWatch: StopWatch;
27
>
readonly deferred: DeferredPromise<void>;
28
>
readonly steeringPendingId?: string;
29
>
}
30
>
31
>
/**
32
>
* Owns the prompt queue + the async iterable handed to
33
>
* `WarmQuery.query()`. Knows nothing about the SDK Query lifecycle,
34
>
* config push, or message dispatch — those live on the pipeline.
35
>
*
36
>
* Invariants:
37
>
* • Pushing wakes the iterable's parked `next()`.
38
>
* • The iterable returns `done` when the supplied `getAbortSignal()`
39
>
* is aborted; pipeline calls {@link notifyAborted} after flipping
40
>
* the controller so the parked `next()` returns immediately.
41
>
* • {@link settleHead} pops the head of the yielded list (called by
42
>
* the consumer loop on every `result` message).
43
>
* • {@link failAll} rejects every pending deferred and clears both
44
>
* lists; used by abort and crash fan-out.
45
>
* • {@link resetForRebind} re-creates the parked deferred for a fresh
46
>
* Query binding (the queue itself survives across rebinds).
47
>
*/
48
>
export class ClaudePromptQueue extends Disposable {
49
>
50
>
private _toYield: IPendingSdkMessage[] = [];
51
>
private _yielded: IPendingSdkMessage[] = [];
52
>
/**
53
>
* Entries that have been popped by {@link settleHead} during the
54
>
* current turn but whose deferreds haven't been completed yet — we
55
>
* batch-complete them when the turn fully drains so an intermediate
56
>
* `result` (steering preempt; CONTEXT.md M10) does NOT settle the
57
>
* original `sendMessage`'s deferred.
58
>
*/
59
>
private _popped: IPendingSdkMessage[] = [];
60
>
private _pendingPromptDeferred = new DeferredPromise<void>();
61
>
62
>
readonly iterable: AsyncIterable<SDKUserMessage> = {
63
>
[Symbol.asyncIterator]: () => ({
65
>
while (true) {
66
>
if (this._getAbortSignal().aborted) {
68
>
}
71
>
this._yielded.push(entry);
72
>
this._logService.info(`[Claude:${this._sessionId}] queue yielded sdkUuid=${entry.sdkUuid} turnId=${entry.turnId}${entry.steeringPendingId ? ` steeringPendingId=${entry.steeringPendingId}` : ''}`);
73
>
if (entry.steeringPendingId) {
75
>
}
77
>
}
80
>
}
82
>
}),
84
>
85
>
constructor(
87
>
private readonly _getAbortSignal: () => AbortSignal,
88
>
private readonly _onSteeringYielded: (pendingId: string) => void,
89
>
@ILogService private readonly _logService: ILogService,
90
>
) {
91
>
super();
92
>
}
94
>
/** True iff no entries are queued or in-flight. */
95
>
get isEmpty(): boolean {
97
>
}
99
>
* Push an entry. Resolves with the entry's deferred (which the
100
>
* consumer settles on `result` via {@link settleHead}).
101
>
*/
102
>
push(entry: IPendingSdkMessage): Promise<void> {
104
>
this._pendingPromptDeferred.complete();
105
>
return entry.deferred.p;
106
>
}
108
>
/**
109
>
* Most-recent in-flight or queued entry, used by steering to inherit
110
>
* its parent's `turnId`. Prefers the in-flight head over the latest
111
>
* queued entry (matches CONTEXT.md M10: steering folds into the
112
>
* in-progress protocol Turn).
113
>
*/
114
>
peekParent(): IPendingSdkMessage | undefined {
116
>
}
118
>
/**
119
>
* Pop the head of the yielded list. If the queue is now fully
120
>
* drained (no more pending or in-flight entries), batch-complete
121
>
* every popped-but-deferred deferred from this turn including the
122
>
* one we just popped. Otherwise hold the popped entry's deferred
123
>
* until the turn ends — the M10 invariant for steering preempt.
124
>
* Called by the consumer on every `result` message.
125
>
*/
126
>
settleHead(): IPendingSdkMessage | undefined {
128
>
if (!completed) {
130
>
}
133
>
for (const e of this._popped) {
135
>
e.deferred.complete();
136
>
}
137
>
}
141
>
}
145
>
/** Reject every pending deferred with `err` and clear all lists. */
146
>
failAll(err: Error): void {
148
>
for (const entry of list) {
150
>
entry.deferred.error(err);
151
>
}
152
>
}
154
>
rejectAll(this._toYield);
155
>
rejectAll(this._yielded);
156
>
rejectAll(this._popped);
157
>
this._toYield = [];
158
>
this._yielded = [];
159
>
this._popped = [];
160
>
}
162
>
/** Wake any parked `next()` — call after the controller is aborted so the iterable returns `done`. */
163
>
notifyAborted(): void {
165
>
}
167
>
/** Re-create the parked deferred for a fresh Query binding. */
168
>
resetForRebind(): void {
170
>
}