fix(desktop): msg-queue drain settle-race — retry reservation error + requeue at head
This commit is contained in:
@@ -6635,17 +6635,43 @@ async function dispatchPrompt(sid, text, opts = {}) {
|
|||||||
const meta = state.sessions.get(sid)
|
const meta = state.sessions.get(sid)
|
||||||
if (meta) meta.running = true
|
if (meta) meta.running = true
|
||||||
try {
|
try {
|
||||||
await window.dsh.sendPrompt(sid, text)
|
await sendPromptSettleAware(sid, text, !!opts.settleRetry)
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
// Surface the error on the stream (matches the live-send path). We do
|
// Surface the error on the stream (matches the live-send path). We do
|
||||||
// NOT clear the rest of the queue — a failed drain leaves the remaining
|
// NOT clear the rest of the queue — a failed drain leaves the remaining
|
||||||
// items parked so the user can retry / edit rather than losing them.
|
// items parked so the user can retry / edit rather than losing them.
|
||||||
appendSystem(`error: ${err.message}`)
|
appendSystem(`error: ${err.message}`)
|
||||||
|
// A drained message that ultimately failed goes BACK to the queue head
|
||||||
|
// (enqueue + promote) instead of vanishing — the strip stays the source
|
||||||
|
// of truth for "not yet delivered". The already-painted optimistic
|
||||||
|
// bubble stays; the echo adopts it on the eventual successful send.
|
||||||
|
if (opts.requeueOnFailure && msgQueue) {
|
||||||
|
const id = msgQueue.enqueue(sid, text)
|
||||||
|
if (id) msgQueue.promote(sid, id)
|
||||||
|
renderMsgQueueStrip()
|
||||||
|
}
|
||||||
} finally {
|
} finally {
|
||||||
if (isActive) sendBtn.disabled = false
|
if (isActive) sendBtn.disabled = false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The daemon releases a session's prompt reservation at prompt SETTLEMENT,
|
||||||
|
// which can lag the turn/end EVENT by a beat — a drain firing exactly on
|
||||||
|
// turn/end can hit "session already has an active prompt" even though the
|
||||||
|
// turn is over (observed live 2026-07-20 on stdio-deepseek). Retry only that
|
||||||
|
// error, with a short backoff ladder; anything else throws immediately.
|
||||||
|
async function sendPromptSettleAware(sid, text, settleRetry) {
|
||||||
|
const delays = settleRetry ? [250, 500, 1000, 2000] : []
|
||||||
|
for (;;) {
|
||||||
|
try {
|
||||||
|
return await window.dsh.sendPrompt(sid, text)
|
||||||
|
} catch (err) {
|
||||||
|
if (delays.length === 0 || !/already has an active prompt/i.test(String(err && err.message))) throw err
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, delays.shift()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Drain exactly one queued message for `sessionId` and send it — guarded so
|
// Drain exactly one queued message for `sessionId` and send it — guarded so
|
||||||
// it fires at most once per turn even when both turn/end and
|
// it fires at most once per turn even when both turn/end and
|
||||||
// session.finished arrive for the same turn. The `_turnDrainPending` flag is
|
// session.finished arrive for the same turn. The `_turnDrainPending` flag is
|
||||||
@@ -6666,7 +6692,14 @@ async function drainMsgQueueOnce(sessionId) {
|
|||||||
const next = msgQueue.drain(sessionId)
|
const next = msgQueue.drain(sessionId)
|
||||||
renderMsgQueueStrip()
|
renderMsgQueueStrip()
|
||||||
if (!next) return
|
if (!next) return
|
||||||
await dispatchPrompt(sessionId, next.text, { clearComposer: false })
|
await dispatchPrompt(sessionId, next.text, {
|
||||||
|
clearComposer: false,
|
||||||
|
// Drains race the daemon's prompt-reservation release (turn/end event
|
||||||
|
// precedes settlement); retry that one error briefly, and park the
|
||||||
|
// message back at the queue head if it still can't get through.
|
||||||
|
settleRetry: true,
|
||||||
|
requeueOnFailure: true,
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
async function cancel() {
|
async function cancel() {
|
||||||
|
|||||||
@@ -194,3 +194,33 @@ test('drained send re-arms inflightTurn so a follow-up Enter queues again', asyn
|
|||||||
assert.equal(sendCalls(dsh).length, before, 'follow-up queued, not sent concurrently')
|
assert.equal(sendCalls(dsh).length, before, 'follow-up queued, not sent concurrently')
|
||||||
assert.deepEqual(renderer.listMsgQueue('s1').map((x) => x.text), ['two', 'three'])
|
assert.deepEqual(renderer.listMsgQueue('s1').map((x) => x.text), ['two', 'three'])
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test('drain retries "already has an active prompt" then delivers (settle race)', async () => {
|
||||||
|
const { renderer, document, dsh } = await loadRenderer()
|
||||||
|
const { input } = await activeInflight(renderer, document)
|
||||||
|
input.value = 'parked message'; await renderer.send()
|
||||||
|
let calls = 0
|
||||||
|
dsh.sendPrompt = async (...args) => {
|
||||||
|
dsh.__calls.push(['sendPrompt', ...args])
|
||||||
|
calls += 1
|
||||||
|
if (calls === 1) throw new Error("Error invoking remote method 'session:prompt': Error: session already has an active prompt: x")
|
||||||
|
return { accepted: true }
|
||||||
|
}
|
||||||
|
renderer.onSessionEvent('s1', { type: 'turn/end', seq: 2 })
|
||||||
|
// Retry ladder's first rung is 250ms — wait past it.
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 450))
|
||||||
|
assert.equal(calls, 2, 'second attempt fired after the settle retry')
|
||||||
|
assert.equal(renderer.listMsgQueue('s1').length, 0, 'queue emptied — message delivered, not dropped')
|
||||||
|
})
|
||||||
|
|
||||||
|
test('drain requeues at head when the send keeps failing (no message loss)', async () => {
|
||||||
|
const { renderer, document, dsh } = await loadRenderer()
|
||||||
|
const { input } = await activeInflight(renderer, document)
|
||||||
|
input.value = 'doomed message'; await renderer.send()
|
||||||
|
dsh.sendPrompt = async () => { throw new Error('some hard transport failure') }
|
||||||
|
renderer.onSessionEvent('s1', { type: 'turn/end', seq: 2 })
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 30))
|
||||||
|
const parked = renderer.listMsgQueue('s1')
|
||||||
|
assert.equal(parked.length, 1, 'failed drain went back to the queue')
|
||||||
|
assert.equal(parked[0].text, 'doomed message')
|
||||||
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user