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..b962ab252 --- /dev/null +++ b/docs/issues/pending-input-explicit-retry/spec.md @@ -0,0 +1,104 @@ +# 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`. Schema version 67 adds + `retry_required_at`; the row is stored as the legacy-safe `blocked` state plus that marker and is + projected as `retry_required` by the current store. Migration rewrites prerelease raw + `retry_required` values to this representation. A downgraded binary therefore fences the row and + can release it through its existing blocked-input Retry path instead of dispatching past an + unknown state. +- 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. Claimed Queue + rows retain their occupied order slot during move, delete, and Queue-to-Steer resequencing, so a + later release cannot create duplicate order values or let another row overtake it. +- 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. The addressed row need not be the current head: an explicit + Send can be claimed behind an older restart-held draft or as a question follow-up, then fail + before committing its user fact. Retry authorizes that exact row while the pump continues to + enforce FIFO execution of any earlier rows. +- 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. Claimed Queue rows keep a unique occupied order slot across every waiting-row resequence. +4. Duplicate Retry requests create at most one state transition, claim, and turn. +5. Retry schedules only the addressed Session and never claims directly from the route. +6. A manually resumed Queue item retains its consume-before-provider marker if a pre-user-fact + failure makes it retry-required. +7. A mixed restart queue with a retry-required head and restart-held tail cannot release the tail + through Resume Queue. +8. Attachment-blocked Retry and Send without image content keep their existing state machine. +9. Downgrade sees retry-required rows as blocked work; an old Retry changes the persisted state to + `pending`, which the current projection honors even if the additive marker remains. +10. 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..49b4644eb 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,49 @@ 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 } + } + + 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/data/schemaCatalog.ts b/src/main/data/schemaCatalog.ts index b48df2b72..63d7d7fb4 100644 --- a/src/main/data/schemaCatalog.ts +++ b/src/main/data/schemaCatalog.ts @@ -225,9 +225,20 @@ const CATALOG_DEFINITIONS: CatalogDefinition[] = [ message_ids_json: "ALTER TABLE deepchat_pending_inputs ADD COLUMN message_ids_json TEXT NOT NULL DEFAULT '[]';", assistant_message_id: - 'ALTER TABLE deepchat_pending_inputs ADD COLUMN assistant_message_id TEXT;' + 'ALTER TABLE deepchat_pending_inputs ADD COLUMN assistant_message_id TEXT;', + retry_required_at: 'ALTER TABLE deepchat_pending_inputs ADD COLUMN retry_required_at INTEGER;' }, - typeCheckedColumns: ['blocking_json', 'message_ids_json', 'assistant_message_id'] + typeCheckedColumns: [ + 'blocking_json', + 'message_ids_json', + 'assistant_message_id', + 'retry_required_at' + ], + afterRepair: (db, addedColumns) => { + if (addedColumns.has('retry_required_at')) { + new DeepChatPendingInputsTable(db).normalizeRetryRequiredRows() + } + } }, { name: 'deepchat_usage_stats', 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..ea168fff3 100644 --- a/src/main/session/data/pendingInputStore.ts +++ b/src/main/session/data/pendingInputStore.ts @@ -90,13 +90,16 @@ export class SessionPendingInputStore { ): PendingSessionInputRecord { const id = nanoid() const nextQueueOrder = this.getNextQueueOrder(sessionId) - const claimedAt = state === 'claimed' ? Date.now() : null + const now = Date.now() + const claimedAt = state === 'claimed' ? now : null + const retryRequiredAt = state === 'retry_required' ? now : null this.database.deepchatPendingInputsTable.insert({ id, sessionId, mode: 'queue', - state, + state: state === 'retry_required' ? 'blocked' : state, payloadJson: JSON.stringify(input), + retryRequiredAt, queueOrder: nextQueueOrder, claimedAt }) @@ -207,13 +210,19 @@ 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') { + const state = this.getRowState(row) + if (state !== 'pending' && state !== 'blocked' && 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' - ? { state: 'pending' as const, blocking_json: null, claimed_at: null } + ...(state !== 'pending' + ? { + state: 'pending' as const, + blocking_json: null, + retry_required_at: null, + claimed_at: null + } : {}) }) return this.toRecord(this.requireRow(itemId, row.session_id)) @@ -225,6 +234,9 @@ export class SessionPendingInputStore { if (fromIndex === -1) { throw new Error(`Pending queue item not found: ${itemId}`) } + if (queueRows[0] && this.isRetryRequiredRow(queueRows[0])) { + 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) { @@ -233,7 +245,7 @@ export class SessionPendingInputStore { const [moved] = queueRows.splice(fromIndex, 1) queueRows.splice(clampedIndex, 0, moved) - this.resequenceQueueRows(queueRows) + this.resequenceQueueRows(sessionId, queueRows) return this.listPendingInputs(sessionId) } @@ -288,7 +300,7 @@ export class SessionPendingInputStore { hasBlockingInput(sessionId: string): boolean { return this.database.deepchatPendingInputsTable .listActiveBySession(sessionId) - .some((row) => row.state === 'blocked') + .some((row) => row.state === 'blocked' && !this.isRetryRequiredRow(row)) } hasClaimedInput(sessionId: string): boolean { @@ -334,19 +346,27 @@ 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 (!this.isRetryRequiredRow(row)) { + throw new Error(`Pending queue item ${itemId} does not require retry.`) } this.database.deepchatPendingInputsTable.update(itemId, { state: 'pending', claimed_at: null, blocking_json: null, - message_ids_json: '[]' + retry_required_at: null }) return this.toRecord(this.requireRow(itemId, row.session_id)) } @@ -395,6 +415,7 @@ export class SessionPendingInputStore { this.database.deepchatPendingInputsTable.update(itemId, { state: 'blocked', blocking_json: JSON.stringify(bodyFreeBlocking), + retry_required_at: null, claimed_at: null }) return this.toRecord(this.requireRow(itemId, row.session_id)) @@ -402,12 +423,13 @@ export class SessionPendingInputStore { retryBlockedInput(itemId: string): PendingSessionInputRecord { const row = this.requireRow(itemId) - if (row.state !== 'blocked') { + if (row.state !== 'blocked' || this.isRetryRequiredRow(row)) { throw new Error(`Pending input ${itemId} is not blocked.`) } this.database.deepchatPendingInputsTable.update(itemId, { state: 'pending', blocking_json: null, + retry_required_at: null, claimed_at: null }) return this.toRecord(this.requireRow(itemId, row.session_id)) @@ -415,7 +437,7 @@ export class SessionPendingInputStore { degradeBlockedInput(itemId: string): PendingSessionInputRecord { const row = this.requireRow(itemId) - if (row.state !== 'blocked') { + if (row.state !== 'blocked' || this.isRetryRequiredRow(row)) { throw new Error(`Pending input ${itemId} is not blocked.`) } const payload = this.decodePayload(row) @@ -426,6 +448,7 @@ export class SessionPendingInputStore { attachmentFallbackPolicy: 'send_without_image_content' }), blocking_json: null, + retry_required_at: null, claimed_at: null }) return this.toRecord(this.requireRow(itemId, row.session_id)) @@ -460,9 +483,7 @@ export class SessionPendingInputStore { } private getWaitingQueueRows(sessionId: string): DeepChatPendingInputRow[] { - return this.getQueueRows(sessionId).filter( - (row) => row.state === 'pending' || row.state === 'blocked' - ) + return this.getQueueRows(sessionId).filter((row) => this.isWaitingQueueRow(row)) } private getSteerRows(sessionId: string): DeepChatPendingInputRow[] { @@ -479,17 +500,45 @@ export class SessionPendingInputStore { } private resequenceQueue(sessionId: string): void { - this.resequenceQueueRows(this.getWaitingQueueRows(sessionId)) + this.resequenceQueueRows(sessionId, this.getWaitingQueueRows(sessionId)) } - private resequenceQueueRows(rows: DeepChatPendingInputRow[]): void { - rows.forEach((row, index) => { + private resequenceQueueRows(sessionId: string, waitingRows: DeepChatPendingInputRow[]): void { + let waitingIndex = 0 + const orderedRows = this.getQueueRows(sessionId) + .filter((row) => row.state !== 'consumed') + .map((row) => (this.isWaitingQueueRow(row) ? (waitingRows[waitingIndex++] ?? row) : row)) + + orderedRows.forEach((row, index) => { this.database.deepchatPendingInputsTable.update(row.id, { queue_order: index + 1 }) }) } + 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) + } + + const retryRequiredAt = state === 'retry_required' ? Date.now() : null + this.database.deepchatPendingInputsTable.update(itemId, { + state: state === 'retry_required' ? 'blocked' : state, + claimed_at: null, + blocking_json: null, + retry_required_at: retryRequiredAt, + 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) { @@ -506,7 +555,7 @@ export class SessionPendingInputStore { id: row.id, sessionId: row.session_id, mode: row.mode, - state: row.state as PendingSessionInputState, + state: this.getRowState(row), payload: this.decodePayload(row), messageIds: this.decodeMessageIds(row), assistantMessageId: row.assistant_message_id, @@ -519,6 +568,20 @@ export class SessionPendingInputStore { } } + private getRowState(row: DeepChatPendingInputRow): PendingSessionInputState { + return this.isRetryRequiredRow(row) ? 'retry_required' : row.state + } + + private isRetryRequiredRow(row: DeepChatPendingInputRow): boolean { + return ( + row.state === 'retry_required' || (row.state === 'blocked' && row.retry_required_at != null) + ) + } + + private isWaitingQueueRow(row: DeepChatPendingInputRow): boolean { + return row.state === 'pending' || row.state === 'blocked' || row.state === 'retry_required' + } + private decodePayload(row: DeepChatPendingInputRow): SendMessageInput { let parsed: unknown try { diff --git a/src/main/session/data/pendingInputs.ts b/src/main/session/data/pendingInputs.ts index 6eb038f20..e6a4eb941 100644 --- a/src/main/session/data/pendingInputs.ts +++ b/src/main/session/data/pendingInputs.ts @@ -229,15 +229,19 @@ export class SessionPendingInputs { } moveQueuedInput(sessionId: string, itemId: string, toIndex: number): PendingSessionInputRecord[] { - this.assertQueueInput(sessionId, itemId) - const records = this.store.moveQueueInput(sessionId, itemId, toIndex) + const records = this.store.runInTransaction(() => { + this.assertQueueInput(sessionId, itemId) + return this.store.moveQueueInput(sessionId, itemId, toIndex) + }) this.emitUpdated(sessionId) return records } deletePendingInput(sessionId: string, itemId: string): void { - this.assertDeletablePendingInput(sessionId, itemId) - this.store.deleteInput(itemId) + this.store.runInTransaction(() => { + this.assertDeletablePendingInput(sessionId, itemId) + this.store.deleteInput(itemId) + }) this.emitUpdated(sessionId) } @@ -307,6 +311,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) @@ -347,8 +367,10 @@ export class SessionPendingInputs { } consumeQueuedInput(sessionId: string, itemId: string): void { - this.assertQueueInputForSession(sessionId, itemId) - this.store.consumeQueueInput(itemId) + this.store.runInTransaction(() => { + this.assertQueueInputForSession(sessionId, itemId) + this.store.consumeQueueInput(itemId) + }) this.emitUpdated(sessionId) } @@ -385,9 +407,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/data/tables/deepchatPendingInputs.ts b/src/main/session/data/tables/deepchatPendingInputs.ts index af6a9f357..03dbc40d7 100644 --- a/src/main/session/data/tables/deepchatPendingInputs.ts +++ b/src/main/session/data/tables/deepchatPendingInputs.ts @@ -2,6 +2,8 @@ import Database from 'better-sqlite3-multiple-ciphers' import { BaseTable } from '@/data/baseTable' import type { PendingSessionInputState } from '@shared/types/agent-interface' +export const PENDING_INPUT_RETRY_SCHEMA_VERSION = 67 + export interface DeepChatPendingInputRow { id: string session_id: string @@ -11,6 +13,7 @@ export interface DeepChatPendingInputRow { message_ids_json: string assistant_message_id: string | null blocking_json: string | null + retry_required_at: number | null queue_order: number | null claimed_at: number | null consumed_at: number | null @@ -34,6 +37,7 @@ export class DeepChatPendingInputsTable extends BaseTable { message_ids_json TEXT NOT NULL DEFAULT '[]', assistant_message_id TEXT, blocking_json TEXT, + retry_required_at INTEGER, queue_order INTEGER, claimed_at INTEGER, consumed_at INTEGER, @@ -60,11 +64,30 @@ export class DeepChatPendingInputsTable extends BaseTable { ADD COLUMN assistant_message_id TEXT; ` } + if (version === PENDING_INPUT_RETRY_SCHEMA_VERSION) { + return 'ALTER TABLE deepchat_pending_inputs ADD COLUMN retry_required_at INTEGER;' + } return null } getLatestVersion(): number { - return 46 + return PENDING_INPUT_RETRY_SCHEMA_VERSION + } + + finalizeMigration(version: number): void { + if (version === PENDING_INPUT_RETRY_SCHEMA_VERSION) { + this.normalizeRetryRequiredRows() + } + } + + normalizeRetryRequiredRows(): void { + this.db.exec(` + UPDATE deepchat_pending_inputs + SET state = 'blocked', + retry_required_at = COALESCE(retry_required_at, updated_at, created_at), + blocking_json = NULL + WHERE state = 'retry_required'; + `) } insert(row: { @@ -76,6 +99,7 @@ export class DeepChatPendingInputsTable extends BaseTable { messageIdsJson?: string assistantMessageId?: string | null blockingJson?: string | null + retryRequiredAt?: number | null queueOrder?: number | null claimedAt?: number | null consumedAt?: number | null @@ -96,12 +120,13 @@ export class DeepChatPendingInputsTable extends BaseTable { message_ids_json, assistant_message_id, blocking_json, + retry_required_at, queue_order, claimed_at, consumed_at, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` ) .run( row.id, @@ -112,6 +137,7 @@ export class DeepChatPendingInputsTable extends BaseTable { row.messageIdsJson ?? '[]', row.assistantMessageId ?? null, row.blockingJson ?? null, + row.retryRequiredAt ?? null, row.queueOrder ?? null, row.claimedAt ?? null, row.consumedAt ?? null, @@ -196,6 +222,7 @@ export class DeepChatPendingInputsTable extends BaseTable { | 'message_ids_json' | 'assistant_message_id' | 'blocking_json' + | 'retry_required_at' | 'queue_order' | 'claimed_at' | 'consumed_at' @@ -229,6 +256,10 @@ export class DeepChatPendingInputsTable extends BaseTable { setClauses.push('blocking_json = ?') params.push(fields.blocking_json) } + if (fields.retry_required_at !== undefined) { + setClauses.push('retry_required_at = ?') + params.push(fields.retry_required_at) + } if (fields.queue_order !== undefined) { setClauses.push('queue_order = ?') params.push(fields.queue_order) 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" />