mirror of
https://github.com/RooVetGit/Roo-Code.git
synced 2026-09-14 23:21:19 +00:00
fix: resolve message queue race condition during LLM processing
- Move queue check before status mutation logic to prevent race condition - Add continuous queue monitoring during pWaitFor to catch messages that arrive during processing - Process queued messages immediately when detected during wait period - Add comprehensive tests for queue race condition scenarios Fixes #8536
This commit is contained in:
parent
5a3f911321
commit
a1f583a0ab
2 changed files with 187 additions and 24 deletions
|
|
@ -799,10 +799,36 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
|
|||
// The state is mutable if the message is complete and the task will
|
||||
// block (via the `pWaitFor`).
|
||||
const isBlocking = !(this.askResponse !== undefined || this.lastMessageTs !== askTs)
|
||||
const isMessageQueued = !this.messageQueueService.isEmpty()
|
||||
const isStatusMutable = !partial && isBlocking && !isMessageQueued
|
||||
let statusMutationTimeouts: NodeJS.Timeout[] = []
|
||||
|
||||
// Process any queued messages first
|
||||
let processedQueuedMessage = false
|
||||
if (!partial && isBlocking && !this.messageQueueService.isEmpty()) {
|
||||
console.log("Task#ask will process message queue")
|
||||
|
||||
const message = this.messageQueueService.dequeueMessage()
|
||||
|
||||
if (message) {
|
||||
processedQueuedMessage = true
|
||||
// Check if this is a tool approval ask that needs to be handled
|
||||
if (
|
||||
type === "tool" ||
|
||||
type === "command" ||
|
||||
type === "browser_action_launch" ||
|
||||
type === "use_mcp_server"
|
||||
) {
|
||||
// For tool approvals, we need to approve first, then send the message if there's text/images
|
||||
this.handleWebviewAskResponse("yesButtonClicked", message.text, message.images)
|
||||
} else {
|
||||
// For other ask types (like followup), fulfill the ask directly
|
||||
this.setMessageResponse(message.text, message.images)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Only set status mutations if we didn't process a queued message and there are no more queued messages
|
||||
const isStatusMutable = !partial && isBlocking && !processedQueuedMessage && this.messageQueueService.isEmpty()
|
||||
|
||||
if (isStatusMutable) {
|
||||
console.log(`Task#ask will block -> type: ${type}`)
|
||||
|
||||
|
|
@ -840,30 +866,41 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
|
|||
}, 1_000),
|
||||
)
|
||||
}
|
||||
} else if (isMessageQueued) {
|
||||
console.log("Task#ask will process message queue")
|
||||
|
||||
const message = this.messageQueueService.dequeueMessage()
|
||||
|
||||
if (message) {
|
||||
// Check if this is a tool approval ask that needs to be handled
|
||||
if (
|
||||
type === "tool" ||
|
||||
type === "command" ||
|
||||
type === "browser_action_launch" ||
|
||||
type === "use_mcp_server"
|
||||
) {
|
||||
// For tool approvals, we need to approve first, then send the message if there's text/images
|
||||
this.handleWebviewAskResponse("yesButtonClicked", message.text, message.images)
|
||||
} else {
|
||||
// For other ask types (like followup), fulfill the ask directly
|
||||
this.setMessageResponse(message.text, message.images)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for askResponse to be set.
|
||||
await pWaitFor(() => this.askResponse !== undefined || this.lastMessageTs !== askTs, { interval: 100 })
|
||||
// Wait for askResponse to be set, but also check for new queued messages periodically
|
||||
await pWaitFor(
|
||||
() => {
|
||||
// If response is ready, we're done
|
||||
if (this.askResponse !== undefined || this.lastMessageTs !== askTs) {
|
||||
return true
|
||||
}
|
||||
|
||||
// Check if new messages were queued while waiting
|
||||
if (!this.messageQueueService.isEmpty()) {
|
||||
console.log("Task#ask detected new queued message while waiting")
|
||||
const message = this.messageQueueService.dequeueMessage()
|
||||
|
||||
if (message) {
|
||||
// Process the newly queued message
|
||||
if (
|
||||
type === "tool" ||
|
||||
type === "command" ||
|
||||
type === "browser_action_launch" ||
|
||||
type === "use_mcp_server"
|
||||
) {
|
||||
this.handleWebviewAskResponse("yesButtonClicked", message.text, message.images)
|
||||
} else {
|
||||
this.setMessageResponse(message.text, message.images)
|
||||
}
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
},
|
||||
{ interval: 100 },
|
||||
)
|
||||
|
||||
if (this.lastMessageTs !== askTs) {
|
||||
// Could happen if we send multiple asks in a row i.e. with
|
||||
|
|
|
|||
|
|
@ -1776,4 +1776,130 @@ describe("Cline", () => {
|
|||
consoleErrorSpy.mockRestore()
|
||||
})
|
||||
})
|
||||
|
||||
describe("Message Queue Race Condition Fix", () => {
|
||||
it("should process messages from queue when available", async () => {
|
||||
const task = new Task({
|
||||
provider: mockProvider,
|
||||
apiConfiguration: mockApiConfig,
|
||||
task: "test task",
|
||||
startTask: false,
|
||||
})
|
||||
|
||||
// Add a message to the queue
|
||||
task.messageQueueService.addMessage("queued message", ["image.png"])
|
||||
|
||||
// Call ask which should process the queued message
|
||||
const result = await task.ask("followup", "Initial question")
|
||||
|
||||
// Verify the queued message was processed
|
||||
expect(result.response).toBe("messageResponse")
|
||||
expect(result.text).toBe("queued message")
|
||||
expect(result.images).toEqual(["image.png"])
|
||||
|
||||
// Verify queue is now empty
|
||||
expect(task.messageQueueService.isEmpty()).toBe(true)
|
||||
})
|
||||
|
||||
it("should handle tool approval messages from queue", async () => {
|
||||
const task = new Task({
|
||||
provider: mockProvider,
|
||||
apiConfiguration: mockApiConfig,
|
||||
task: "test task",
|
||||
startTask: false,
|
||||
})
|
||||
|
||||
// Add a message to the queue
|
||||
task.messageQueueService.addMessage("approve with context", ["image.png"])
|
||||
|
||||
// Call ask for tool approval - should auto-approve with queued message
|
||||
const result = await task.ask("tool", "Do you want to use this tool?")
|
||||
|
||||
// Verify the queued message was processed as tool approval
|
||||
expect(result.response).toBe("yesButtonClicked")
|
||||
expect(result.text).toBe("approve with context")
|
||||
expect(result.images).toEqual(["image.png"])
|
||||
|
||||
// Verify queue is now empty
|
||||
expect(task.messageQueueService.isEmpty()).toBe(true)
|
||||
})
|
||||
|
||||
it("should check for new messages during wait period", async () => {
|
||||
const task = new Task({
|
||||
provider: mockProvider,
|
||||
apiConfiguration: mockApiConfig,
|
||||
task: "test task",
|
||||
startTask: false,
|
||||
})
|
||||
|
||||
// Mock pWaitFor to simulate adding a message during the wait
|
||||
const originalPWaitFor = (await import("p-wait-for")).default
|
||||
let conditionCheckCount = 0
|
||||
vi.mocked(originalPWaitFor).mockImplementation(async (condition, options) => {
|
||||
// Simulate checking the condition multiple times
|
||||
while (true) {
|
||||
conditionCheckCount++
|
||||
|
||||
// On the second check, add a message to the queue
|
||||
if (conditionCheckCount === 2) {
|
||||
task.messageQueueService.addMessage("delayed message")
|
||||
// The condition should now detect the message and process it
|
||||
task.setMessageResponse("delayed message")
|
||||
}
|
||||
|
||||
// Check the condition
|
||||
const result = await condition()
|
||||
if (result) {
|
||||
return
|
||||
}
|
||||
|
||||
// Prevent infinite loop
|
||||
if (conditionCheckCount > 5) {
|
||||
// Force completion
|
||||
task.setMessageResponse("forced completion")
|
||||
return
|
||||
}
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 10))
|
||||
}
|
||||
})
|
||||
|
||||
// Call ask - initially no messages in queue
|
||||
const result = await task.ask("followup", "Question")
|
||||
|
||||
// Should have processed the message that was added during wait
|
||||
expect(result.response).toBe("messageResponse")
|
||||
expect(result.text).toBe("delayed message")
|
||||
|
||||
// Verify condition was checked multiple times
|
||||
expect(conditionCheckCount).toBeGreaterThan(1)
|
||||
})
|
||||
|
||||
it("should handle multiple messages in queue", async () => {
|
||||
const task = new Task({
|
||||
provider: mockProvider,
|
||||
apiConfiguration: mockApiConfig,
|
||||
task: "test task",
|
||||
startTask: false,
|
||||
})
|
||||
|
||||
// Add multiple messages to the queue
|
||||
task.messageQueueService.addMessage("first message")
|
||||
task.messageQueueService.addMessage("second message")
|
||||
|
||||
// First ask should process first message
|
||||
const result1 = await task.ask("followup", "Question 1")
|
||||
expect(result1.text).toBe("first message")
|
||||
|
||||
// Queue should still have one message
|
||||
expect(task.messageQueueService.isEmpty()).toBe(false)
|
||||
|
||||
// Second ask should process second message
|
||||
const result2 = await task.ask("followup", "Question 2")
|
||||
expect(result2.text).toBe("second message")
|
||||
|
||||
// Queue should now be empty
|
||||
expect(task.messageQueueService.isEmpty()).toBe(true)
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue