Skip to content

Commit 0bbba8d

Browse files
myk1ytmyk1yt
committed
[Fix] Close the queued-drain approval and delivery defects in #937
What: - Task: a drain-retained message is now consumable only by a conversational ask. `queuedClaimExclusionsForAsk` returns the retained IDs for any ask whose `queuedResponseForAsk` mapping answers with `yesButtonClicked` (command, tool, use_mcp_server), and both ask claim sites pass that set to `MessageQueueService.claimNextMessage` / `hasUnclaimed`. The retention record is cleared only on the consume edge, so exactly-once delivery is provable. - Task: a drain no longer submits while an `ask()` is in flight (`pendingAskCount` brackets the whole ask body). A background completion landing during a prompt previously answered that ask with text written for another turn, bypassing the ask's claim gate, blanket-deny latch, and protected-ask exemption, and left the message queued for a second delivery. - ExecuteCommandTool: the submitted-command shell-integration warning arm now drains, and only after its warning result publishes. - ExecuteCommandTool: `commandSubmitted` is tracked from the `shell_execution_started` callback instead of being hardcoded true, so a user timeout that fires before `TerminalProcess.run()` reached `sendText`/`executeCommand` reports false and skips the drain. - ExecuteCommandTool: the background-completion drain is gated on the final `command_output` say succeeding. That rejection is swallowed by the output chain's `.catch`, so the drain reads a recorded publication outcome. - MessageQueueService: dropped the now-orphaned `dequeueMessage` (knip). Why: The retention semantics the #937 fix introduced made a drain-submitted message reachable by the pre-existing approval mapper. A stale "ship it" queued during an earlier command could approve a later `rm -rf build` with no prompt even with auto-approval off. The remaining four defects were confirmed by the review quartet: a warning arm that never drained, a pre-start timeout that wrongly drained, a background drain that ran after the final output failed to publish, and untested cancellation guards. Impact: Execution-gating asks always get a real prompt; between-turn feedback is delivered by the next conversational ask instead. No public signature change beyond the already-shipped `TaskLike.submitUserMessage` boolean. Non-test changed lines: 495 (was 329, budget 500). Tests: core/task + core/tools + core/message-queue = 84 files, 1707 passed, 5 skipped. Full suite: 551 files passed, 3 skipped; 11,862 passed, 37 skipped, 0 failed. Every new test was mutation-checked: reverting its fix line fails it.
1 parent 8761ddc commit 0bbba8d

12 files changed

Lines changed: 524 additions & 64 deletions

‎src/core/message-queue/MessageQueueService.ts‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -73,18 +73,18 @@ export class MessageQueueService extends EventEmitter<QueueEvents> {
7373
return true
7474
}
7575

76-
public dequeueMessage(): QueuedMessage | undefined {
77-
const index = this._messages.findIndex((message) => !this.claimedMessageIds.has(message.id))
78-
if (index === -1) {
79-
return undefined
80-
}
81-
const [message] = this._messages.splice(index, 1)
82-
this.emit("stateChanged", this._messages)
83-
return message
84-
}
85-
86-
public claimNextMessage(): QueuedMessage | undefined {
87-
const message = this._messages.find((candidate) => !this.claimedMessageIds.has(candidate.id))
76+
/**
77+
* Claims the first message that is neither already claimed nor listed in
78+
* `excludeIds`.
79+
*
80+
* `excludeIds` lets a consumer that must not take certain messages skip them
81+
* without removing them: an execution-gating ask passes the IDs a drain has
82+
* already submitted so stale feedback can never stand in for an approval.
83+
*/
84+
public claimNextMessage(excludeIds?: ReadonlySet<string>): QueuedMessage | undefined {
85+
const message = this._messages.find(
86+
(candidate) => !this.claimedMessageIds.has(candidate.id) && !excludeIds?.has(candidate.id),
87+
)
8888
if (message) {
8989
this.claimedMessageIds.add(message.id)
9090
}
@@ -115,9 +115,13 @@ export class MessageQueueService extends EventEmitter<QueueEvents> {
115115
* report a non-empty queue. Consumers that are about to take a message need
116116
* this instead: it answers "is there anything I may take", which is false
117117
* when every remaining message is already spoken for.
118+
*
119+
* `excludeIds` mirrors `claimNextMessage`: a message the caller may not take
120+
* is not available, so an execution-gating ask does not report availability
121+
* for a message only a conversational ask may consume.
118122
*/
119-
public hasUnclaimed(): boolean {
120-
return this._messages.some((message) => !this.claimedMessageIds.has(message.id))
123+
public hasUnclaimed(excludeIds?: ReadonlySet<string>): boolean {
124+
return this._messages.some((message) => !this.claimedMessageIds.has(message.id) && !excludeIds?.has(message.id))
121125
}
122126

123127
public dispose(): void {

‎src/core/message-queue/__tests__/MessageQueueService.spec.ts‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ describe("MessageQueueService claims", () => {
88

99
expect(queue.claimNextMessage()).toEqual(first)
1010
expect(queue.claimNextMessage()).toEqual(second)
11-
expect(queue.dequeueMessage()).toBeUndefined()
11+
expect(queue.claimNextMessage()).toBeUndefined()
1212
expect(queue.messages).toEqual([first, second])
1313
})
1414

@@ -67,4 +67,32 @@ describe("MessageQueueService claims", () => {
6767
queue.releaseMessage(message.id)
6868
expect(queue.hasUnclaimed()).toBe(true)
6969
})
70+
71+
it("skips excluded messages without claiming or removing them", () => {
72+
const queue = new MessageQueueService()
73+
const retained = queue.addMessage("retained by a drain")!
74+
const fresh = queue.addMessage("fresh")!
75+
const excluded = new Set([retained.id])
76+
77+
// Availability must agree with the claim: the excluded message is not
78+
// available to this consumer even though the queue holds it.
79+
expect(queue.hasUnclaimed(excluded)).toBe(true)
80+
expect(queue.claimNextMessage(excluded)).toEqual(fresh)
81+
82+
// The excluded message stays queued and unclaimed, so a consumer allowed
83+
// to take it (a conversational ask) still can.
84+
expect(queue.messages).toEqual([retained, fresh])
85+
expect(queue.hasUnclaimed(excluded)).toBe(false)
86+
expect(queue.claimNextMessage()).toEqual(retained)
87+
})
88+
89+
it("reports no availability when every message is excluded", () => {
90+
const queue = new MessageQueueService()
91+
const only = queue.addMessage("only")!
92+
const excluded = new Set([only.id])
93+
94+
expect(queue.hasUnclaimed(excluded)).toBe(false)
95+
expect(queue.claimNextMessage(excluded)).toBeUndefined()
96+
expect(queue.messages).toEqual([only])
97+
})
7098
})

‎src/core/task/Task.ts‎

Lines changed: 117 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -515,11 +515,18 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
515515
// slot (which would drop the earlier response after both messages were
516516
// already removed from the queue).
517517
private queuedMessageDrainChain: Promise<unknown> = Promise.resolve()
518-
// ID of the last drain-submitted queued message. A successful submit only
519-
// posts into the pending ask-response slot; the message stays queued until
520-
// an ask consumes that response. A later drain that claims this same ID
521-
// must not re-post it, or it would overwrite a distinct pending response.
522-
private pendingSubmittedQueuedMessageId: string | undefined
518+
// IDs of queued messages a drain submitted into the ask-response slot but
519+
// no ask has consumed yet. They stay queued on purpose, so two consumers
520+
// must read this set: a later drain, so it never re-posts the same message
521+
// into the slot, and an execution-gating ask (`queuedResponseForAsk` would
522+
// answer it with `yesButtonClicked`), which must never let between-turn
523+
// feedback stand in for an approval. Cleared only on consumption.
524+
private drainRetainedQueuedMessageIds = new Set<string>()
525+
// Number of `ask()` calls in flight. A drain must not submit while one is
526+
// pending: a background completion can land during any prompt, and its
527+
// submission would answer that ask with text written for another turn while
528+
// the retained message stayed queued for a second delivery.
529+
private pendingAskCount = 0
523530

524531
// Streaming
525532
isWaitingForFirstChunk = false
@@ -1245,10 +1252,9 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
12451252
private handleQueuedAskResponse(message: QueuedMessage, resolution: QueuedAskResolution): string | undefined {
12461253
this.handleWebviewAskResponse(resolution.response, message.text, message.images)
12471254
// This ask consumed the message, so a later drain may submit the next
1248-
// one instead of treating it as still pending.
1249-
if (this.pendingSubmittedQueuedMessageId === message.id) {
1250-
this.pendingSubmittedQueuedMessageId = undefined
1251-
}
1255+
// one instead of treating it as still pending. The retention record is
1256+
// cleared on the same edge: a consumed message is no longer retained.
1257+
this.drainRetainedQueuedMessageIds.delete(message.id)
12521258
if (resolution.requiresDurableAck) {
12531259
return message.id
12541260
}
@@ -1379,9 +1385,38 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
13791385
* `isMessageQueued`/`isStatusMutable` keep `isEmpty()` semantics on purpose:
13801386
* flipping those would re-enable interactive prompt timers whenever a claim
13811387
* is outstanding.
1388+
*
1389+
* `excludeIds` carries the messages this ask may not consume at all; see
1390+
* `queuedClaimExclusionsForAsk`.
1391+
*/
1392+
private mayDrainQueuedMessageForAsk(excludeIds?: ReadonlySet<string>): boolean {
1393+
return !this.blanketDeniedCommandThisTurn && this.messageQueueService.hasUnclaimed(excludeIds)
1394+
}
1395+
1396+
/**
1397+
* Queued messages this ask must not claim. `queuedResponseForAsk` answers a
1398+
* command, tool, or use_mcp_server ask with `yesButtonClicked`, so letting one
1399+
* of those claim a drain-retained message would turn between-turn feedback
1400+
* into an approval: the ask skips `checkAutoApproval`, and a stale "ship it"
1401+
* would run the next command with no prompt even with auto-approval off. Such
1402+
* asks exclude every retained ID and get a real prompt; the message stays
1403+
* queued for a conversational ask, the only consumer that carries its text
1404+
* into the model as feedback.
13821405
*/
1383-
private mayDrainQueuedMessageForAsk(): boolean {
1384-
return !this.blanketDeniedCommandThisTurn && this.messageQueueService.hasUnclaimed()
1406+
private queuedClaimExclusionsForAsk(type: ClineAsk, text?: string): ReadonlySet<string> | undefined {
1407+
// Drop IDs the queue no longer holds (deleted from the webview or the API,
1408+
// or cleared by dispose) so the record cannot outlive its message.
1409+
const queuedIds = new Set(this.messageQueueService.messages.map((message) => message.id))
1410+
for (const id of this.drainRetainedQueuedMessageIds) {
1411+
if (!queuedIds.has(id)) {
1412+
this.drainRetainedQueuedMessageIds.delete(id)
1413+
}
1414+
}
1415+
1416+
return this.drainRetainedQueuedMessageIds.size > 0 &&
1417+
queuedResponseForAsk(type, text)?.response === "yesButtonClicked"
1418+
? this.drainRetainedQueuedMessageIds
1419+
: undefined
13851420
}
13861421

13871422
/**
@@ -1864,6 +1899,36 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
18641899
* rejection: the denial is scoped to its own tool call.
18651900
*/
18661901
autoDenyDetail?: AutoDenyDetail
1902+
}> {
1903+
// Bracket the whole ask, not just the wait: the counter is the drain's
1904+
// "a prompt is in flight" signal, and a submission landing during any
1905+
// await below would answer this ask with text written for another turn.
1906+
this.pendingAskCount++
1907+
try {
1908+
return await this.askWithPendingCount(type, text, partial, progressStatus, isProtected, autoApprovalContext)
1909+
} finally {
1910+
this.pendingAskCount--
1911+
}
1912+
}
1913+
1914+
private async askWithPendingCount(
1915+
type: ClineAsk,
1916+
text?: string,
1917+
partial?: boolean,
1918+
progressStatus?: ToolProgressStatus,
1919+
isProtected?: boolean,
1920+
autoApprovalContext?: AutoApprovalContext,
1921+
): Promise<{
1922+
response: ClineAskResponse
1923+
text?: string
1924+
images?: string[]
1925+
queuedMessageId?: string
1926+
/**
1927+
* Present when the ask was resolved by an automatic (policy) denial
1928+
* rather than a user click. Consumers must not treat this as a user
1929+
* rejection: the denial is scoped to its own tool call.
1930+
*/
1931+
autoDenyDetail?: AutoDenyDetail
18671932
}> {
18681933
// If this Cline instance was aborted by the provider, then the only
18691934
// thing keeping us alive is a promise still running in the background,
@@ -1902,6 +1967,9 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
19021967
// has to see and stalling a hands-free session. Leaving the message
19031968
// unclaimed lets the policy decision stand and keeps the message queued.
19041969
const queueMayAnswerThisAsk = !(blanketDenyEngaged && type === "command")
1970+
// Retained drain submissions are off-limits to any ask that would answer
1971+
// with `yesButtonClicked`; see `queuedClaimExclusionsForAsk`.
1972+
const queuedClaimExclusions = this.queuedClaimExclusionsForAsk(type, text)
19051973
// A queued message must not short-circuit protected asks (e.g.
19061974
// DCG-blocked commands): claiming one here skips checkAutoApproval, and
19071975
// the drain below would auto-approve what protection intentionally
@@ -1911,9 +1979,9 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
19111979
type === "command_output" ||
19121980
isProtected ||
19131981
!queueMayAnswerThisAsk ||
1914-
!this.mayDrainQueuedMessageForAsk()
1982+
!this.mayDrainQueuedMessageForAsk(queuedClaimExclusions)
19151983
? undefined
1916-
: this.messageQueueService.claimNextMessage()
1984+
: this.messageQueueService.claimNextMessage(queuedClaimExclusions)
19171985
const queuedAskResolution = queuedMessage ? queuedResponseForAsk(type, text) : undefined
19181986
// `this.cwd`, not `provider.cwd`:
19191987
// The path inside `text` was made relative to this task's workspace,
@@ -2318,9 +2386,9 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
23182386
shouldDrainQueuedMessageForAsk &&
23192387
!isProtected &&
23202388
!queuedCommandPolicyCheck &&
2321-
this.mayDrainQueuedMessageForAsk()
2389+
this.mayDrainQueuedMessageForAsk(queuedClaimExclusions)
23222390
) {
2323-
const message = this.messageQueueService.claimNextMessage()
2391+
const message = this.messageQueueService.claimNextMessage(queuedClaimExclusions)
23242392
const resolution = message ? queuedResponseForAsk(type, text) : undefined
23252393
if (message && resolution) {
23262394
if (type === "command") {
@@ -2416,11 +2484,10 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
24162484
this.askResponse = askResponse
24172485
this.askResponseText = text
24182486
this.askResponseImages = images
2419-
// A direct response overwrites whatever a drain submitted into the slot,
2420-
// so that submission was not consumed and must not block a later drain.
2421-
// The drain's own submit lands here before it records the ID, so this
2422-
// cannot clear the tracker for the submission it just made.
2423-
this.pendingSubmittedQueuedMessageId = undefined
2487+
// No retention bookkeeping here: a direct response overwrites the slot a
2488+
// drain submitted into, which means that submission was NOT consumed, so
2489+
// the message stays queued and stays excluded from approval asks. Only
2490+
// `handleQueuedAskResponse` clears retention, on the consume edge.
24242491

24252492
// Create a checkpoint whenever the user sends a message.
24262493
// Use allowEmpty=true to ensure a checkpoint is recorded even if there are no file changes.
@@ -6488,14 +6555,24 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
64886555
* delivered (task stopping, provider gone), the claim is released and the
64896556
* run resolves false.
64906557
*
6491-
* Drains are serialized per task: each run claims only after the previous
6492-
* run's submission handoff completed, so a background-completion drain and
6493-
* a post-result drain cannot interleave two different messages into the
6494-
* single pending ask-response.
6558+
* Retention is restricted to conversational consumption. A retained message
6559+
* is recorded in `drainRetainedQueuedMessageIds`, and every ask whose
6560+
* `queuedResponseForAsk` mapping would answer with `yesButtonClicked`
6561+
* (command, tool, use_mcp_server) excludes those IDs from its claim. Without
6562+
* that exclusion the retained text would silently approve a later execution
6563+
* ask with no prompt, even with auto-approval off.
6564+
*
6565+
* A drain also never submits while an `ask()` is pending: the submission
6566+
* would answer that ask with text written for another turn, bypassing the
6567+
* ask's own latch and denial authority, and the retained message would then
6568+
* be delivered a second time.
6569+
*
6570+
* Drains are serialized per task, so a background-completion drain and a
6571+
* post-result drain cannot interleave two messages into the single slot.
64956572
*
64966573
* @returns Promise resolving to true when a queued message was submitted
6497-
* (and remains queued until consumed); false when the queue was empty or
6498-
* the task is stopping.
6574+
* (and remains queued until consumed); false when the queue was empty, an
6575+
* ask was pending, or the task is stopping.
64996576
*/
65006577
public processQueuedMessages(): Promise<boolean> {
65016578
const run = this.queuedMessageDrainChain.then(() => this.claimAndSubmitNextQueuedMessage())
@@ -6516,6 +6593,14 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
65166593
if (this.abort || this.abandoned) {
65176594
return false
65186595
}
6596+
// An ask is waiting for its response. Submitting now would resolve that
6597+
// prompt with text written for another turn — the ask's own claim gate,
6598+
// blanket-deny latch, and protected-ask exemption never run — and the
6599+
// message would stay queued for a second delivery. Defer: the message
6600+
// keeps its place and the next post-ask drain submits it.
6601+
if (this.pendingAskCount > 0) {
6602+
return false
6603+
}
65196604
const queued = this.messageQueueService.claimNextMessage()
65206605
if (!queued) {
65216606
return false
@@ -6524,7 +6609,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
65246609
// consumed it yet: re-posting would overwrite a distinct pending
65256610
// response and lose it, so release the claim and keep the message
65266611
// queued for the ask.
6527-
if (this.pendingSubmittedQueuedMessageId === queued.id) {
6612+
if (this.drainRetainedQueuedMessageIds.has(queued.id)) {
65286613
this.messageQueueService.releaseMessage(queued.id)
65296614
return true
65306615
}
@@ -6543,10 +6628,11 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
65436628
throw error
65446629
}
65456630
// Submission succeeded, but the message is only removed once an ask
6546-
// consumes it. Track the ID so a later drain can tell "already
6547-
// submitted, still pending" from "not submitted yet", and release the
6548-
// claim so the ask path can claim it.
6549-
this.pendingSubmittedQueuedMessageId = queued.id
6631+
// consumes it. Record the ID so a later drain can tell "already
6632+
// submitted, still pending" from "not submitted yet" and so an
6633+
// execution-gating ask refuses to claim it, then release the claim so a
6634+
// conversational ask can take it.
6635+
this.drainRetainedQueuedMessageIds.add(queued.id)
65506636
this.messageQueueService.releaseMessage(queued.id)
65516637
return true
65526638
}

0 commit comments

Comments
 (0)