From 60d189b6742ea12a68b88fd459a89cdba591be2b Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Mon, 10 Aug 2026 21:49:46 +0800 Subject: [PATCH 1/4] fix(chat): retry released queue inputs --- .../pending-input-explicit-retry/spec.md | 93 ++++++++++++++ .../deepchat/harness/deepChatAgentHarness.ts | 7 ++ .../pendingInputAdmissionCoordinator.ts | 53 +++++++- .../deepchat/runtime/pendingInputContracts.ts | 1 + .../deepchat/runtime/pendingInputPump.ts | 60 ++++++--- .../agent/manager/deepChatAgentBackend.ts | 7 +- src/main/agent/manager/sessionHandles.ts | 1 + src/main/app/composition.ts | 3 +- src/main/session/contracts.ts | 5 + src/main/session/data/pendingInputStore.ts | 48 ++++++-- src/main/session/data/pendingInputs.ts | 20 ++- src/main/session/routes.ts | 10 ++ src/main/session/turn.ts | 14 +++ src/renderer/api/SessionClient.ts | 6 + .../src/components/chat/PendingInputLane.vue | 55 ++++++++- .../src/features/chat-page/ChatPage.vue | 3 + .../composables/usePendingInputActions.ts | 25 +++- src/renderer/src/i18n/da-DK/chat.json | 4 + src/renderer/src/i18n/de-DE/chat.json | 4 + src/renderer/src/i18n/en-US/chat.json | 6 +- src/renderer/src/i18n/es-ES/chat.json | 4 + src/renderer/src/i18n/fa-IR/chat.json | 4 + src/renderer/src/i18n/fr-FR/chat.json | 4 + src/renderer/src/i18n/he-IL/chat.json | 4 + src/renderer/src/i18n/id-ID/chat.json | 4 + src/renderer/src/i18n/it-IT/chat.json | 4 + src/renderer/src/i18n/ja-JP/chat.json | 4 + src/renderer/src/i18n/ko-KR/chat.json | 4 + src/renderer/src/i18n/ms-MY/chat.json | 4 + src/renderer/src/i18n/pl-PL/chat.json | 4 + src/renderer/src/i18n/pt-BR/chat.json | 4 + src/renderer/src/i18n/ru-RU/chat.json | 4 + src/renderer/src/i18n/tr-TR/chat.json | 4 + src/renderer/src/i18n/vi-VN/chat.json | 4 + src/renderer/src/i18n/zh-CN/chat.json | 6 +- src/renderer/src/i18n/zh-HK/chat.json | 4 + src/renderer/src/i18n/zh-TW/chat.json | 4 + src/renderer/src/stores/ui/pendingInput.ts | 29 +++++ src/shared/contracts/routes.ts | 2 + .../contracts/routes/sessions.routes.ts | 12 ++ src/shared/types/agent-interface.d.ts | 7 +- .../harness/deepChatAgentHarness.test.ts | 12 +- .../pendingInputAdmissionCoordinator.test.ts | 109 +++++++++++++++++ .../deepchat/runtime/pendingInputPump.test.ts | 114 ++++++++++++++++-- .../manager/deepChatAgentBackend.test.ts | 12 ++ test/main/routes/dispatcher.test.ts | 19 +++ .../session/data/pendingInputStore.test.ts | 59 +++++++++ test/main/session/data/pendingInputs.test.ts | 38 ++++++ .../tables/deepchatPendingInputsTable.test.ts | 32 +++++ test/main/session/turn.test.ts | 29 ++++- .../components/PendingInputLane.test.ts | 40 ++++++ .../usePendingInputActions.test.ts | 49 +++++++- .../renderer/stores/pendingInputStore.test.ts | 29 +++++ 53 files changed, 1029 insertions(+), 58 deletions(-) create mode 100644 docs/issues/pending-input-explicit-retry/spec.md diff --git a/docs/issues/pending-input-explicit-retry/spec.md b/docs/issues/pending-input-explicit-retry/spec.md new file mode 100644 index 000000000..d605f0fc5 --- /dev/null +++ b/docs/issues/pending-input-explicit-retry/spec.md @@ -0,0 +1,93 @@ +# Released Queue Inputs Lack an Explicit Retry + +## GitHub + +- Issue: https://github.com/ThinkInAIXYZ/deepchat/issues/2112 +- Classification: complex reliability bug +- Priority: P1 +- Status: implemented on `fix/issue-2112-pending-input-retry` + +## Issue And Impact + +`PendingInputPump` deliberately leaves a Queue row at the FIFO head when a claimed turn is +released before its user fact is committed. The durable row currently returns to the ordinary +`pending` state, so the renderer cannot distinguish it from an unsent draft and exposes no Retry +action. The row can therefore block every later Queue input with no explanation or direct recovery +path. + +Editing the row happens to schedule another drain, but that mutation is not the explicit retry +contract owned by the pump. + +## Root Cause + +PR https://github.com/ThinkInAIXYZ/deepchat/pull/2023 centralized Queue claim, release, and +single-flight draining in `PendingInputPump`. Commit `b2729141` added the contract and regression +that a released Queue head waits for explicit retry, but release persisted the same `pending` state +used by ordinary drafts and the renderer contract was not extended. + +PR https://github.com/ThinkInAIXYZ/deepchat/pull/2129 did not introduce the defect. It added the +separate restart-held Queue and Resume Queue semantics that this fix must preserve. + +## Fix Design + +- Add durable `retry_required` to `PendingSessionInputState`. SQLite already stores unconstrained + text, so no schema migration is required. +- Failed DeepChat Queue settlements transition `claimed -> retry_required`. Intentional temporary + release for Queue-to-Steer mutation and cold-start repair continue to transition to `pending`. +- Keep `retry_required` in Queue ordering while excluding it from claimable work. This makes it the + authoritative FIFO head without adding a renderer flag or process-local retry set. +- Add an idempotent per-item DeepChat retry operation under the existing Session operation gate. + The first request transitions `retry_required -> pending` and asks the existing pump to drain; + stale repeated requests are no-ops. +- Preserve restart behavior: an existing `retry_required` row remains retry-required and is not + added to the restart hold. Ordinary pending rows and recovered claimed rows remain restart-held. + Resume Queue is available only when the actual FIFO head is restart-held. +- Keep ACP release behavior unchanged. ACP continues to release failed Queue claims to `pending`. +- Label retry-required rows in the existing Queue lane, expose Retry, and suppress invalid Steer and + reorder actions. Attachment-blocked actions remain unchanged. +- Editing a retry-required row is an explicit content mutation that authorizes a new attempt and + transitions the row back to `pending`, preserving existing behavior without relying on it as the + only recovery path. + +## Compatibility And Safety Invariants + +1. Ordinary live Queue drafts remain `pending` and retain their existing actions. +2. A later Queue row is never claimed while a retry-required row is ahead of it. +3. Duplicate Retry requests create at most one state transition, claim, and turn. +4. Retry schedules only the addressed Session and never claims directly from the route. +5. A manually resumed Queue item retains its consume-before-provider marker if a pre-user-fact + failure makes it retry-required. +6. A mixed restart queue with a retry-required head and restart-held tail cannot release the tail + through Resume Queue. +7. Attachment-blocked Retry and Send without image content keep their existing state machine. +8. No message text, attachment path, or payload is added to diagnostics. + +## Tasks + +- [x] Add and enforce the durable retry-required state and distinct release transitions. +- [x] Add the gated, typed, idempotent retry route through the existing pump. +- [x] Project retry-required state and action in the renderer with localized copy. +- [x] Add focused persistence, pump, restart, route, and renderer regressions. +- [x] Run format, i18n, lint, typecheck, and relevant main/renderer tests. +- [x] Review the staged change for side effects, compatibility, boundaries, performance, security, + naming, coverage, and maintenance cost before committing. + +## Validation + +```bash +pnpm exec vitest run --config vitest.config.ts \ + test/main/session/data/tables/deepchatPendingInputsTable.test.ts \ + test/main/session/data/pendingInputs.test.ts \ + test/main/agent/deepchat/runtime/pendingInputPump.test.ts \ + test/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.test.ts \ + test/main/session/turn.test.ts \ + test/main/routes/dispatcher.test.ts +pnpm exec vitest run --config vitest.config.renderer.ts \ + test/renderer/components/PendingInputLane.test.ts \ + test/renderer/features/chat-page/composables/usePendingInputActions.test.ts \ + test/renderer/stores/pendingInputStore.test.ts +pnpm run format +pnpm run i18n +pnpm run lint +pnpm run typecheck +``` diff --git a/src/main/agent/deepchat/harness/deepChatAgentHarness.ts b/src/main/agent/deepchat/harness/deepChatAgentHarness.ts index cfb5e2b3d..87cae2ede 100644 --- a/src/main/agent/deepchat/harness/deepChatAgentHarness.ts +++ b/src/main/agent/deepchat/harness/deepChatAgentHarness.ts @@ -122,6 +122,13 @@ export class DeepChatAgentHarness return await this.services.pendingInputAdmission.resumePendingQueue(sessionId) } + async retryPendingQueueInput( + sessionId: string, + itemId: string + ): Promise<{ accepted: boolean; started: boolean }> { + return await this.services.pendingInputAdmission.retryPendingQueueInput(sessionId, itemId) + } + async queuePendingInput( sessionId: string, content: string | SendMessageInput, diff --git a/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts b/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts index 6827547d1..c81f9e65b 100644 --- a/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts +++ b/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts @@ -44,6 +44,7 @@ export type PendingInputAdmissionStorePort = Pick< | 'promoteQueuedInputToSteerMessage' | 'queuePendingInput' | 'retryBlockedInput' + | 'retryReleasedQueueInput' | 'updateQueuedInput' > @@ -352,7 +353,7 @@ export class PendingInputAdmissionCoordinator { } } - claim.settle({ kind: 'release-before-user-fact' }) + claim.settle({ kind: 'release-for-mutation' }) this.ports.pendingInputs.updateQueuedInput(sessionId, itemId, prepared.content) const instance = this.ports.registry.getHydratedScope(toAppSessionId(sessionId))?.instance @@ -429,6 +430,56 @@ export class PendingInputAdmissionCoordinator { return record } + async retryPendingQueueInput( + sessionId: string, + itemId: string + ): Promise<{ accepted: boolean; started: boolean }> { + await this.ensureSessionReady(sessionId) + const input = this.ports.pendingInputs.getInput(sessionId, itemId) + if (!input) { + throw new Error(`Pending input not found: ${itemId}`) + } + if (input.mode !== 'queue') { + throw new Error('Only a released queue input can be retried.') + } + if (input.state !== 'retry_required') { + return { accepted: false, started: false } + } + const queueHead = this.ports.pendingInputs + .listPendingInputs(sessionId) + .filter((item) => item.mode === 'queue') + .sort((left, right) => (left.queueOrder ?? 0) - (right.queueOrder ?? 0))[0] + if (queueHead?.id !== itemId) { + return { accepted: false, started: false } + } + + try { + this.ports.pendingInputs.retryReleasedQueueInput(sessionId, itemId) + } catch (error) { + const persisted = this.ports.pendingInputs.getInput(sessionId, itemId) + if (persisted?.mode !== 'queue' || persisted.state !== 'pending') { + throw error + } + logger.error( + `[DeepChatAgent] retry pending queue publication failed session=${sessionId}`, + redactRuntimeErrorForLog(error) + ) + } + let started = false + try { + started = await this.ports.pump.drain(sessionId, 'manual') + } catch (error) { + logger.error( + `[DeepChatAgent] retry pending queue drain failed session=${sessionId}`, + redactRuntimeErrorForLog(error) + ) + } + if (!started) { + this.ports.pump.schedule(sessionId, 'enqueue') + } + return { accepted: true, started } + } + async resumePendingQueue(sessionId: string): Promise { const state = await this.ports.sessionState.get(sessionId) if (!state) { diff --git a/src/main/agent/deepchat/runtime/pendingInputContracts.ts b/src/main/agent/deepchat/runtime/pendingInputContracts.ts index 40fb3c5ff..6b773479c 100644 --- a/src/main/agent/deepchat/runtime/pendingInputContracts.ts +++ b/src/main/agent/deepchat/runtime/pendingInputContracts.ts @@ -10,6 +10,7 @@ export type PendingInputTurnSource = PendingInputEnqueueSource | 'steer' export type ClaimedInputDisposition = | { kind: 'consume' } | { kind: 'block'; attachmentPreparation: AttachmentPreparationSummary } + | { kind: 'release-for-mutation' } | { kind: 'release-before-user-fact' } | { kind: 'release-after-rollback' } diff --git a/src/main/agent/deepchat/runtime/pendingInputPump.ts b/src/main/agent/deepchat/runtime/pendingInputPump.ts index 9227f4dac..8401b8e15 100644 --- a/src/main/agent/deepchat/runtime/pendingInputPump.ts +++ b/src/main/agent/deepchat/runtime/pendingInputPump.ts @@ -46,6 +46,7 @@ export type PendingInputPumpStorePort = Pick< | 'listPendingInputs' | 'releaseClaimedInput' | 'releaseClaimedQueueInput' + | 'releaseClaimedQueueInputForRetry' > type PendingInputPumpLifecyclePort = Pick< @@ -160,11 +161,15 @@ class DurablePendingInputClaim implements ClaimedPendingInputHandle { this.id, disposition.attachmentPreparation ) + case 'release-for-mutation': + return this.source === 'steer' + ? this.pendingInputs.releaseClaimedInput(this.sessionId, this.id) + : this.pendingInputs.releaseClaimedQueueInput(this.sessionId, this.id) case 'release-before-user-fact': case 'release-after-rollback': return this.source === 'steer' ? this.pendingInputs.releaseClaimedInput(this.sessionId, this.id) - : this.pendingInputs.releaseClaimedQueueInput(this.sessionId, this.id) + : this.pendingInputs.releaseClaimedQueueInputForRetry(this.sessionId, this.id) } } @@ -197,9 +202,14 @@ class DurablePendingInputClaim implements ClaimedPendingInputHandle { : record === null case 'block': return record?.mode === expectedMode && record.state === 'blocked' + case 'release-for-mutation': + return record?.mode === expectedMode && record.state === 'pending' case 'release-before-user-fact': case 'release-after-rollback': - return record?.mode === expectedMode && record.state === 'pending' + return ( + record?.mode === expectedMode && + record.state === (this.source === 'queue' ? 'retry_required' : 'pending') + ) } } } @@ -224,25 +234,35 @@ export class PendingInputPump { } releaseRestartHoldForSession(sessionId: string): boolean { - const heldQueueInputs = this.ports.pendingInputs + const queueInputs = this.ports.pendingInputs .listPendingInputs(sessionId) - .filter((input) => input.mode === 'queue' && this.restartHeldQueueInputIds.has(input.id)) + .filter((input) => input.mode === 'queue') .sort((left, right) => (left.queueOrder ?? 0) - (right.queueOrder ?? 0)) - const resumedHead = heldQueueInputs[0] - if (!resumedHead) { + const resumedHead = queueInputs[0] + if ( + !resumedHead || + resumedHead.state !== 'pending' || + !this.restartHeldQueueInputIds.has(resumedHead.id) + ) { return false } - for (const input of heldQueueInputs) { - this.restartHeldQueueInputIds.delete(input.id) + for (const input of queueInputs) { + if (this.restartHeldQueueInputIds.has(input.id)) { + this.restartHeldQueueInputIds.delete(input.id) + } } this.manuallyResumedQueueInputIds.add(resumedHead.id) return true } hasRestartHeldQueueInputs(sessionId: string): boolean { - return this.ports.pendingInputs + const head = this.ports.pendingInputs .listPendingInputs(sessionId) - .some((input) => input.mode === 'queue' && this.restartHeldQueueInputIds.has(input.id)) + .filter((input) => input.mode === 'queue') + .sort((left, right) => (left.queueOrder ?? 0) - (right.queueOrder ?? 0))[0] + return Boolean( + head?.state === 'pending' && this.restartHeldQueueInputIds.has(head.id) + ) } hasOnlyRestartHeldQueueInputs(sessionId: string): boolean { @@ -288,17 +308,21 @@ export class PendingInputPump { !this.ports.pendingInputs.hasBlockingInput(sessionId) && !this.ports.pendingInputs.hasClaimedInput(sessionId) && !instance?.isPendingQueueDraining() + const hasRetryRequiredInput = this.ports.pendingInputs + .listPendingInputs(sessionId) + .some((input) => input.mode === 'queue' && input.state === 'retry_required') - if (isUnclaimedQuestionFollowUp) { + if (isUnclaimedQuestionFollowUp && !hasRetryRequiredInput) { return true } if (!this.canDrainWithSnapshot(sessionId, status, 'enqueue', snapshot)) { return false } const hasWaitingTurnInput = - source === 'send' + hasRetryRequiredInput || + (source === 'send' ? this.hasExecutablePendingTurnInput(sessionId) - : this.ports.pendingInputs.hasPendingTurnInput(sessionId) + : this.ports.pendingInputs.hasPendingTurnInput(sessionId)) return ( !hasWaitingTurnInput && !this.ports.pendingInputs.hasBlockingInput(sessionId) && @@ -572,11 +596,15 @@ export class PendingInputPump { reason: PendingInputWakeReason ): Promise { try { - const releasedInputIsWaitingForRetry = this.ports.pendingInputs + const releasedInputIsStillWaiting = this.ports.pendingInputs .listPendingInputs(sessionId) - .some((item) => item.id === claimedInputId && item.state === 'pending') + .some( + (item) => + item.id === claimedInputId && + (item.state === 'pending' || item.state === 'retry_required') + ) if ( - !releasedInputIsWaitingForRetry && + !releasedInputIsStillWaiting && this.ports.pendingInputs.hasPendingTurnInput(sessionId) && (await this.ports.sessionState.get(sessionId))?.status === 'idle' && !this.hasInteractionBlocker(sessionId) diff --git a/src/main/agent/manager/deepChatAgentBackend.ts b/src/main/agent/manager/deepChatAgentBackend.ts index 1b5bba8ec..0632f87c5 100644 --- a/src/main/agent/manager/deepChatAgentBackend.ts +++ b/src/main/agent/manager/deepChatAgentBackend.ts @@ -54,6 +54,10 @@ export interface DeepChatAgentBackendPort { listPendingInputs(sessionId: AppSessionId): Promise isPendingQueueResumeAvailable(sessionId: AppSessionId): Promise resumePendingQueue(sessionId: AppSessionId): Promise + retryPendingQueueInput( + sessionId: AppSessionId, + itemId: string + ): Promise<{ accepted: boolean; started: boolean }> queuePendingInput( sessionId: AppSessionId, content: SendMessageInput, @@ -201,7 +205,8 @@ export function createDeepChatAgentBackend( getCompactionState: () => port.getSessionCompactionState(sessionId), compact: () => port.compactSession(sessionId), isPendingQueueResumeAvailable: () => port.isPendingQueueResumeAvailable(sessionId), - resumePendingQueue: () => port.resumePendingQueue(sessionId) + resumePendingQueue: () => port.resumePendingQueue(sessionId), + retryPendingQueueInput: (itemId) => port.retryPendingQueueInput(sessionId, itemId) } } handles.set(sessionId, handle) diff --git a/src/main/agent/manager/sessionHandles.ts b/src/main/agent/manager/sessionHandles.ts index 696e19930..8aff60d80 100644 --- a/src/main/agent/manager/sessionHandles.ts +++ b/src/main/agent/manager/sessionHandles.ts @@ -85,6 +85,7 @@ export interface DeepChatControlFacet { compact(): Promise<{ compacted: boolean; state: SessionCompactionState }> isPendingQueueResumeAvailable(): Promise resumePendingQueue(): Promise + retryPendingQueueInput(itemId: string): Promise<{ accepted: boolean; started: boolean }> } export interface DeepChatSessionHandle extends AgentSessionHandle { diff --git a/src/main/app/composition.ts b/src/main/app/composition.ts index 6758642ce..e268fb522 100644 --- a/src/main/app/composition.ts +++ b/src/main/app/composition.ts @@ -1663,7 +1663,8 @@ export async function createMainProcessControl(dependencies: { compact: () => handle.deepchat.compact() }, isPendingQueueResumeAvailable: () => handle.deepchat.isPendingQueueResumeAvailable(), - resumePendingQueue: () => handle.deepchat.resumePendingQueue() + resumePendingQueue: () => handle.deepchat.resumePendingQueue(), + retryPendingQueueInput: (itemId) => handle.deepchat.retryPendingQueueInput(itemId) } : { ...turn, kind: handle.kind } } diff --git a/src/main/session/contracts.ts b/src/main/session/contracts.ts index a8ef0c71a..92dd6a7ab 100644 --- a/src/main/session/contracts.ts +++ b/src/main/session/contracts.ts @@ -243,6 +243,7 @@ export type SessionTurnRuntimeSession = } isPendingQueueResumeAvailable(): Promise resumePendingQueue(): Promise + retryPendingQueueInput(itemId: string): Promise<{ accepted: boolean; started: boolean }> }) | (SessionTurnRuntimeBase & { readonly kind: 'acp' }) @@ -293,6 +294,10 @@ export interface SessionTurnPort { listPendingInputs(sessionId: string): Promise isPendingQueueResumeAvailable(sessionId: string): Promise resumePendingQueue(sessionId: string): Promise + retryPendingQueueInput( + sessionId: string, + itemId: string + ): Promise<{ accepted: boolean; started: boolean }> queuePendingInput( sessionId: string, content: string | SendMessageInput diff --git a/src/main/session/data/pendingInputStore.ts b/src/main/session/data/pendingInputStore.ts index f372d9dcf..0a61c093d 100644 --- a/src/main/session/data/pendingInputStore.ts +++ b/src/main/session/data/pendingInputStore.ts @@ -207,12 +207,12 @@ export class SessionPendingInputStore { if (row.mode !== 'queue') { throw new Error(`Pending input ${itemId} is not a queue item.`) } - if (row.state !== 'pending' && row.state !== 'blocked') { + if (row.state !== 'pending' && row.state !== 'blocked' && row.state !== 'retry_required') { throw new Error(`Pending queue item ${itemId} is not editable.`) } this.database.deepchatPendingInputsTable.update(itemId, { payload_json: JSON.stringify(input), - ...(row.state === 'blocked' + ...(row.state !== 'pending' ? { state: 'pending' as const, blocking_json: null, claimed_at: null } : {}) }) @@ -225,6 +225,9 @@ export class SessionPendingInputStore { if (fromIndex === -1) { throw new Error(`Pending queue item not found: ${itemId}`) } + if (queueRows[0]?.state === 'retry_required') { + throw new Error('Retry or edit the released queue input before reordering the queue.') + } const clampedIndex = Math.max(0, Math.min(toIndex, queueRows.length - 1)) if (fromIndex === clampedIndex) { @@ -334,19 +337,29 @@ export class SessionPendingInputStore { } releaseClaimedQueueInput(itemId: string): PendingSessionInputRecord { + return this.releaseClaimedQueueInputTo(itemId, 'pending') + } + + releaseClaimedQueueInputForRetry(itemId: string): PendingSessionInputRecord { + return this.releaseClaimedQueueInputTo(itemId, 'retry_required') + } + + retryReleasedQueueInput(itemId: string): PendingSessionInputRecord { const row = this.requireRow(itemId) if (row.mode !== 'queue') { throw new Error(`Pending input ${itemId} is not a queue item.`) } - if (row.state !== 'claimed') { - return this.toRecord(row) + if (row.state !== 'retry_required') { + throw new Error(`Pending queue item ${itemId} does not require retry.`) + } + if (this.getWaitingQueueRows(row.session_id)[0]?.id !== itemId) { + throw new Error(`Pending queue item ${itemId} is not the queue head.`) } this.database.deepchatPendingInputsTable.update(itemId, { state: 'pending', claimed_at: null, - blocking_json: null, - message_ids_json: '[]' + blocking_json: null }) return this.toRecord(this.requireRow(itemId, row.session_id)) } @@ -461,7 +474,7 @@ export class SessionPendingInputStore { private getWaitingQueueRows(sessionId: string): DeepChatPendingInputRow[] { return this.getQueueRows(sessionId).filter( - (row) => row.state === 'pending' || row.state === 'blocked' + (row) => row.state === 'pending' || row.state === 'blocked' || row.state === 'retry_required' ) } @@ -490,6 +503,27 @@ export class SessionPendingInputStore { }) } + private releaseClaimedQueueInputTo( + itemId: string, + state: 'pending' | 'retry_required' + ): PendingSessionInputRecord { + const row = this.requireRow(itemId) + if (row.mode !== 'queue') { + throw new Error(`Pending input ${itemId} is not a queue item.`) + } + if (row.state !== 'claimed') { + return this.toRecord(row) + } + + this.database.deepchatPendingInputsTable.update(itemId, { + state, + claimed_at: null, + blocking_json: null, + message_ids_json: '[]' + }) + return this.toRecord(this.requireRow(itemId, row.session_id)) + } + private requireRow(itemId: string, expectedSessionId?: string): DeepChatPendingInputRow { const row = this.database.deepchatPendingInputsTable.get(itemId) if (!row) { diff --git a/src/main/session/data/pendingInputs.ts b/src/main/session/data/pendingInputs.ts index 6eb038f20..aa01d1fa4 100644 --- a/src/main/session/data/pendingInputs.ts +++ b/src/main/session/data/pendingInputs.ts @@ -307,6 +307,22 @@ export class SessionPendingInputs { return record } + releaseClaimedQueueInputForRetry(sessionId: string, itemId: string): PendingSessionInputRecord { + this.assertQueueInputForSession(sessionId, itemId) + const record = this.store.releaseClaimedQueueInputForRetry(itemId) + this.emitUpdated(sessionId) + return record + } + + retryReleasedQueueInput(sessionId: string, itemId: string): PendingSessionInputRecord { + const record = this.store.runInTransaction(() => { + this.assertQueueInputForSession(sessionId, itemId) + return this.store.retryReleasedQueueInput(itemId) + }) + this.emitUpdated(sessionId) + return record + } + releaseClaimedInput(sessionId: string, itemId: string): PendingSessionInputRecord { this.assertInputOwnedBySession(sessionId, itemId) const claimed = this.store.getInput(itemId) @@ -385,9 +401,11 @@ export class SessionPendingInputs { heldQueueInputIds.add(input.id) } affectedSessionIds.add(input.sessionId) - } else { + } else if (input.state !== 'retry_required') { heldQueueInputIds.add(input.id) affectedSessionIds.add(input.sessionId) + } else { + affectedSessionIds.add(input.sessionId) } continue } diff --git a/src/main/session/routes.ts b/src/main/session/routes.ts index bbcafe8e0..b76097eb8 100644 --- a/src/main/session/routes.ts +++ b/src/main/session/routes.ts @@ -45,6 +45,7 @@ import { sessionsRestoreRoute, sessionsResolveBlockedPendingInputRoute, sessionsRetryMessageRoute, + sessionsRetryPendingQueueInputRoute, sessionsRetryRtkHealthCheckRoute, sessionsSearchHistoryRoute, sessionsSetAcpSessionConfigOptionRoute, @@ -265,6 +266,15 @@ export function createSessionRoutes(deps: { }) } ], + [ + sessionsRetryPendingQueueInputRoute.name, + async (rawInput) => { + const input = sessionsRetryPendingQueueInputRoute.input.parse(rawInput) + return sessionsRetryPendingQueueInputRoute.output.parse( + await deps.turn.retryPendingQueueInput(input.sessionId, input.itemId) + ) + } + ], [ sessionsQueuePendingInputRoute.name, async (rawInput) => { diff --git a/src/main/session/turn.ts b/src/main/session/turn.ts index 07fb48da9..a839ad973 100644 --- a/src/main/session/turn.ts +++ b/src/main/session/turn.ts @@ -227,6 +227,20 @@ export class SessionTurn implements SessionTurnPort, SessionInitialTurnPort { }) } + async retryPendingQueueInput( + sessionId: string, + itemId: string + ): Promise<{ accepted: boolean; started: boolean }> { + return await this.dependencies.workdir.runWithSessionOperationGate(sessionId, async () => { + this.requireSession(sessionId) + const runtime = this.dependencies.runtime.resolveSession(toAppSessionId(sessionId)) + if (runtime.kind !== 'deepchat') { + throw new Error('Pending queue retry is only available for DeepChat sessions.') + } + return await runtime.retryPendingQueueInput(itemId) + }) + } + async queuePendingInput( sessionId: string, content: string | SendMessageInput diff --git a/src/renderer/api/SessionClient.ts b/src/renderer/api/SessionClient.ts index 7b2aeaf81..3f62d2843 100644 --- a/src/renderer/api/SessionClient.ts +++ b/src/renderer/api/SessionClient.ts @@ -49,6 +49,7 @@ import { sessionsRenameRoute, sessionsResolveBlockedPendingInputRoute, sessionsResumePendingQueueRoute, + sessionsRetryPendingQueueInputRoute, sessionsRetryRtkHealthCheckRoute, sessionsRetryMessageRoute, sessionsRestoreRoute @@ -162,6 +163,10 @@ export function createSessionClient(bridge: DeepchatBridge = getDeepchatBridge() return await bridge.invoke(sessionsResumePendingQueueRoute.name, { sessionId }) } + async function retryPendingQueueInput(sessionId: string, itemId: string) { + return await bridge.invoke(sessionsRetryPendingQueueInputRoute.name, { sessionId, itemId }) + } + async function queuePendingInput(sessionId: string, content: string | SendMessageInput) { const input = sessionsQueuePendingInputRoute.input.parse({ sessionId, content }) const result = await bridge.invoke(sessionsQueuePendingInputRoute.name, input) @@ -591,6 +596,7 @@ export function createSessionClient(bridge: DeepchatBridge = getDeepchatBridge() ensureAcpDraftSession, listPendingInputs, resumePendingQueue, + retryPendingQueueInput, queuePendingInput, updateQueuedInput, moveQueuedInput, diff --git a/src/renderer/src/components/chat/PendingInputLane.vue b/src/renderer/src/components/chat/PendingInputLane.vue index 0e5587974..c950ce894 100644 --- a/src/renderer/src/components/chat/PendingInputLane.vue +++ b/src/renderer/src/components/chat/PendingInputLane.vue @@ -55,7 +55,7 @@ item-key="id" handle=".pending-input-drag" :animation="150" - :disabled="Boolean(editingItemId) || hasBlockedQueueItem" + :disabled="Boolean(editingItemId) || hasBlockedQueueItem || hasRetryRequiredQueueItem" ghost-class="pending-input-ghost" class="space-y-1" @end="onDragEnd" @@ -82,7 +82,11 @@ type="button" class="pending-input-drag inline-flex h-6 w-5 shrink-0 items-center justify-center rounded-md text-muted-foreground transition hover:bg-muted/80 hover:text-foreground" :title="t('chat.pendingInput.reorder')" - :disabled="Boolean(editingItemId) || element.state === 'blocked'" + :disabled=" + Boolean(editingItemId) || + element.state === 'blocked' || + element.state === 'retry_required' + " > @@ -146,6 +150,12 @@ > {{ formatBlockingText(element) }} + + {{ t('chat.pendingInput.retryRequiredDescription') }} + @@ -195,6 +205,29 @@ + (), { activeLimit: 5, @@ -263,7 +297,8 @@ const props = withDefaults( disableQueueSteerAction: false, showResumeAction: false, resumeDisabled: false, - resumeLoading: false + resumeLoading: false, + retryingItemId: null } ) @@ -273,6 +308,7 @@ const emit = defineEmits<{ 'steer-queue': [itemId: string] 'delete-queue': [itemId: string] 'resume-queue': [] + 'retry-queue': [itemId: string] 'resolve-blocked': [payload: { itemId: string; action: 'retry' | 'send_without_image_content' }] }>() const { t } = useI18n() @@ -288,6 +324,9 @@ const blockedCount = computed( const hasBlockedQueueItem = computed(() => props.queueItems.some((item) => item.state === 'blocked') ) +const hasRetryRequiredQueueItem = computed(() => + props.queueItems.some((item) => item.state === 'retry_required') +) const isScrollable = computed(() => props.queueItems.length > 3 || Boolean(editingItemId.value)) const listMaxHeightClass = computed(() => (editingItemId.value ? 'max-h-[220px]' : 'max-h-[116px]')) const editingQueueItem = computed( @@ -327,9 +366,13 @@ function formatPayloadText(item: PendingSessionInputRecord): string { } function formatPayloadTitle(item: PendingSessionInputRecord): string { - return item.state === 'blocked' - ? `${formatPayloadText(item)} — ${formatBlockingText(item)}` - : formatPayloadText(item) + if (item.state === 'blocked') { + return `${formatPayloadText(item)} — ${formatBlockingText(item)}` + } + if (item.state === 'retry_required') { + return `${formatPayloadText(item)} — ${t('chat.pendingInput.retryRequiredDescription')}` + } + return formatPayloadText(item) } function beginEdit(item: PendingSessionInputRecord): void { diff --git a/src/renderer/src/features/chat-page/ChatPage.vue b/src/renderer/src/features/chat-page/ChatPage.vue index c17362761..5771cd7d0 100644 --- a/src/renderer/src/features/chat-page/ChatPage.vue +++ b/src/renderer/src/features/chat-page/ChatPage.vue @@ -150,12 +150,14 @@ :show-resume-action="showPendingQueueResume" :resume-disabled="disablePendingQueueResume" :resume-loading="pendingInputStore.resumingQueue" + :retrying-item-id="pendingInputStore.retryingItemId" class="mx-auto mb-1.5 max-w-4xl" @update-queue="onPendingInputUpdate" @move-queue="onPendingInputMove" @steer-queue="onPendingInputSteer" @delete-queue="onPendingInputDelete" @resume-queue="onPendingInputResume" + @retry-queue="onPendingInputRetry" @resolve-blocked="onPendingInputResolve" />