Skip to content

Commit a0f9770

Browse files
myk1ytmyk1yt
committed
fix(task): hand a consumed submission's ID only when its entry is reserved
What: the pending-slot consumption hands queuedMessageId to the caller only when claimMessage(submittedId) succeeds; the tracker is cleared either way. When the entry is gone (user deleted the queued message between drain submission and consumption), the response text/images are still returned but with queuedMessageId undefined, so the consumer treats them as direct feedback via the say path. Why: with the reservation change, claimMessage can fail on an entry the webview removed mid-flight; handing the ID anyway made the durable ack persist feedback and then fail at removeMessage on the missing entry, turning sayUserFeedbackAndAckQueued into a throw (CodeRabbit 4174162031). Impact: no ID without a live reserved entry; deleted-mid-flight submissions degrade to direct feedback. Pin: drain-submit -> remove -> consume returns the text with queuedMessageId undefined.
1 parent d6d0b86 commit a0f9770

2 files changed

Lines changed: 30 additions & 3 deletions

File tree

‎src/core/task/Task.ts‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2535,15 +2535,22 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
25352535
// Hand the ID to the caller instead of removing inline: the queue
25362536
// entry is deleted only after the feedback is durably saved
25372537
// (persistQueuedFeedbackAndAcknowledge), so a failed history write
2538-
// cannot lose a message that was already dequeued.
2539-
queuedMessageId = this.pendingSubmittedQueuedMessageId
2538+
// cannot lose a message that was already dequeued. The ID is
2539+
// handed only when the entry can also be reserved: the user may
2540+
// have deleted the queued message between submission and
2541+
// consumption, and acking an entry that no longer exists would
2542+
// fail at removeMessage — without a reservation the response is
2543+
// treated as direct feedback instead.
2544+
const submittedId = this.pendingSubmittedQueuedMessageId
25402545
this.pendingSubmittedQueuedMessageId = undefined
25412546
// Reserve the entry through the durable ack: while persistence
25422547
// retries (which can take seconds), neither a second ask nor a
25432548
// background drain may claim it again and persist the same
25442549
// message twice. Persistence settles the reservation itself —
25452550
// removeMessage on success, releaseMessage on failure/abort.
2546-
this.messageQueueService.claimMessage(queuedMessageId)
2551+
if (this.messageQueueService.claimMessage(submittedId)) {
2552+
queuedMessageId = submittedId
2553+
}
25472554
} else if (
25482555
!this.messageQueueService.messages.some(
25492556
(message) => message.id === this.pendingSubmittedQueuedMessageId,

‎src/core/task/__tests__/ask-queued-message-drain.spec.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -481,6 +481,26 @@ describe("Task.ask queued message drain", () => {
481481
expect(String(resultA)).toContain("superseded")
482482
})
483483

484+
it("treats a consumed submission as direct feedback when the entry was deleted mid-flight", async () => {
485+
const task = await createTask({ getState: async () => ({}) })
486+
487+
// Interception: blocked ask, drain submits, then the user deletes the
488+
// queued message before the ask observes the pending slot.
489+
const askPromise = task.ask("completion_result", "Done", false)
490+
await new Promise((resolve) => setTimeout(resolve, 150))
491+
task.messageQueueService.addMessage("deleted before consume")
492+
await task.processQueuedMessages()
493+
const entry = task.messageQueueService.messages.at(0)
494+
if (!entry) throw new Error("queued message missing")
495+
task.messageQueueService.removeMessage(entry.id)
496+
497+
const result = await askPromise
498+
499+
// No entry remains to ack: the ID must not be handed back, and the
500+
// response text is delivered as direct feedback instead.
501+
expect(result).toMatchObject({ response: "messageResponse", text: "deleted before consume" })
502+
expect(result.queuedMessageId).toBeUndefined()
503+
})
484504
it("delivers an intercepted message exactly once when a consumer acks through the durable helper", async () => {
485505
const task = await createTask({ getState: async () => ({}) })
486506
const submitSpy = vi.spyOn(task, "submitUserMessage")

0 commit comments

Comments
 (0)