Merge pull request 'Send the same request id when an outbox message is retried' (#37) from feature/pi-durable-borrow into main
Send the same request id when an outbox message is retried
This commit was merged in pull request #37.
This commit is contained in:
+2
-2
@@ -48,8 +48,8 @@ export const api = {
|
|||||||
conversationDetail: (id: string) => auth.fetch(`${BASE}/api/conversations/${id}`).then(r =>
|
conversationDetail: (id: string) => auth.fetch(`${BASE}/api/conversations/${id}`).then(r =>
|
||||||
json<{ conversation: Conversation; messages: Message[]; runs: Run[]; activeHitl: HITLCard[]; lastSeq: number }>(r)),
|
json<{ conversation: Conversation; messages: Message[]; runs: Run[]; activeHitl: HITLCard[]; lastSeq: number }>(r)),
|
||||||
|
|
||||||
sendMessage: (convId: string, text: string, attachmentIds: string[]) =>
|
sendMessage: (convId: string, text: string, attachmentIds: string[], requestId?: string) =>
|
||||||
auth.fetch(`${BASE}/api/conversations/${convId}/messages`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text, attachmentIds }) })
|
auth.fetch(`${BASE}/api/conversations/${convId}/messages`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text, attachmentIds, requestId }) })
|
||||||
.then(async r => {
|
.then(async r => {
|
||||||
if (!r.ok) throw new Error(`${r.status}: ${await r.text()}`)
|
if (!r.ok) throw new Error(`${r.status}: ${await r.text()}`)
|
||||||
return r.json() as Promise<{ message: Message; run?: Run; command?: string; reply?: string; conversation?: Conversation }>
|
return r.json() as Promise<{ message: Message; run?: Run; command?: string; reply?: string; conversation?: Conversation }>
|
||||||
|
|||||||
@@ -31,6 +31,12 @@ export function replaceQueuedText(queue: QueuedMessage[], id: string, text: stri
|
|||||||
return queue.map(q => q.id === id ? { ...q, text } : q)
|
return queue.map(q => q.id === id ? { ...q, text } : q)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// deliveryRequestId 是这条排队消息重发时必须复用的 requestId。
|
||||||
|
// 服务端按它去重,避免忙时 409 再入队后写出第二条用户消息。
|
||||||
|
export function deliveryRequestId(item: QueuedMessage): string {
|
||||||
|
return item.id
|
||||||
|
}
|
||||||
|
|
||||||
export function promoteQueuedMessage(queue: QueuedMessage[], id: string): QueuedMessage[] {
|
export function promoteQueuedMessage(queue: QueuedMessage[], id: string): QueuedMessage[] {
|
||||||
const item = queue.find(q => q.id === id)
|
const item = queue.find(q => q.id === id)
|
||||||
if (!item) return queue
|
if (!item) return queue
|
||||||
|
|||||||
+7
-6
@@ -4,7 +4,7 @@ import { useSyncExternalStore } from 'react'
|
|||||||
import { api, connectEvents } from './api'
|
import { api, connectEvents } from './api'
|
||||||
import { applyTextDelta, applyThinkingDelta, applyToolCall, applyToolCallDelta, applyToolResult, applyToolOutput, applyArtifactCreated, applyArtifactUpdated, applyCheck, applyFinalizing, type LiveItem } from './live'
|
import { applyTextDelta, applyThinkingDelta, applyToolCall, applyToolCallDelta, applyToolResult, applyToolOutput, applyArtifactCreated, applyArtifactUpdated, applyCheck, applyFinalizing, type LiveItem } from './live'
|
||||||
import { closePreviewTab, openPreviewTab, type PreviewTab } from './previewtabs'
|
import { closePreviewTab, openPreviewTab, type PreviewTab } from './previewtabs'
|
||||||
import { enqueueMessage, enqueueMessageFront, newQueuedMessage, promoteQueuedMessage, removeQueuedMessage, replaceQueuedText, type QueuedMessage } from './outbox'
|
import { deliveryRequestId, enqueueMessage, enqueueMessageFront, newQueuedMessage, promoteQueuedMessage, removeQueuedMessage, replaceQueuedText, type QueuedMessage } from './outbox'
|
||||||
import { flashElement } from './scrollsync'
|
import { flashElement } from './scrollsync'
|
||||||
import { isThemeChoice, resolveTheme, type ThemeChoice } from './theme'
|
import { isThemeChoice, resolveTheme, type ThemeChoice } from './theme'
|
||||||
import type {
|
import type {
|
||||||
@@ -468,10 +468,11 @@ export const store = {
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
||||||
async sendMessage(text: string, attachmentIds: string[]): Promise<'ok' | 'busy'> {
|
async sendMessage(text: string, attachmentIds: string[], requestId?: string): Promise<'ok' | 'busy'> {
|
||||||
if (!state.currentConvId) return 'ok'
|
if (!state.currentConvId) return 'ok'
|
||||||
|
const id = requestId ?? newQueuedMessage(text, attachmentIds, []).id
|
||||||
try {
|
try {
|
||||||
const out = await api.sendMessage(state.currentConvId, text, attachmentIds)
|
const out = await api.sendMessage(state.currentConvId, text, attachmentIds, id)
|
||||||
if (out.command) {
|
if (out.command) {
|
||||||
store.toast(out.reply ?? `命令 ${out.command} 已执行`)
|
store.toast(out.reply ?? `命令 ${out.command} 已执行`)
|
||||||
if (out.conversation) await store.openConversation(out.conversation.id)
|
if (out.conversation) await store.openConversation(out.conversation.id)
|
||||||
@@ -491,7 +492,7 @@ export const store = {
|
|||||||
return 'ok'
|
return 'ok'
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
if (isConversationBusy(e)) {
|
if (isConversationBusy(e)) {
|
||||||
set({ outbox: enqueueMessageFront(state.outbox, newQueuedMessage(text, attachmentIds, [])) })
|
set({ outbox: enqueueMessageFront(state.outbox, { id, text, attachmentIds, attachmentNames: [] }) })
|
||||||
return 'busy'
|
return 'busy'
|
||||||
}
|
}
|
||||||
store.toast(`发送失败:${e}`)
|
store.toast(`发送失败:${e}`)
|
||||||
@@ -522,7 +523,7 @@ export const store = {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
set({ outbox: rest })
|
set({ outbox: rest })
|
||||||
await store.sendMessage(item.text, item.attachmentIds)
|
await store.sendMessage(item.text, item.attachmentIds, deliveryRequestId(item))
|
||||||
},
|
},
|
||||||
|
|
||||||
async flushOutbox() {
|
async flushOutbox() {
|
||||||
@@ -533,7 +534,7 @@ export const store = {
|
|||||||
set({ outbox: state.outbox.slice(1), outboxDraining: true })
|
set({ outbox: state.outbox.slice(1), outboxDraining: true })
|
||||||
let result: 'ok' | 'busy' = 'ok'
|
let result: 'ok' | 'busy' = 'ok'
|
||||||
try {
|
try {
|
||||||
result = await store.sendMessage(next.text, next.attachmentIds)
|
result = await store.sendMessage(next.text, next.attachmentIds, deliveryRequestId(next))
|
||||||
} finally {
|
} finally {
|
||||||
set({ outboxDraining: false })
|
set({ outboxDraining: false })
|
||||||
}
|
}
|
||||||
|
|||||||
+10
-1
@@ -1,6 +1,6 @@
|
|||||||
import { describe, expect, it } from 'vitest'
|
import { describe, expect, it } from 'vitest'
|
||||||
import {
|
import {
|
||||||
enqueueMessage, enqueueMessageFront, newQueuedMessage, promoteQueuedMessage,
|
deliveryRequestId, enqueueMessage, enqueueMessageFront, newQueuedMessage, promoteQueuedMessage,
|
||||||
removeQueuedMessage, replaceQueuedText,
|
removeQueuedMessage, replaceQueuedText,
|
||||||
} from '../src/outbox'
|
} from '../src/outbox'
|
||||||
|
|
||||||
@@ -25,4 +25,13 @@ describe('outbox', () => {
|
|||||||
const again = enqueueMessageFront(q, b)
|
const again = enqueueMessageFront(q, b)
|
||||||
expect(again.map(x => x.id)).toEqual([b.id, a.id])
|
expect(again.map(x => x.id)).toEqual([b.id, a.id])
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('reuses the queued id as the request id when the send is retried', () => {
|
||||||
|
const item = newQueuedMessage('hi', ['f1'], ['note.md'])
|
||||||
|
const queued = enqueueMessageFront([], item)
|
||||||
|
expect(deliveryRequestId(queued[0])).toBe(item.id)
|
||||||
|
const retried = enqueueMessageFront(queued, { ...item, text: item.text })
|
||||||
|
expect(deliveryRequestId(retried[0])).toBe(item.id)
|
||||||
|
expect(retried).toHaveLength(1)
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user