fix(chat): retry released queue inputs

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
+```

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,

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<boolean> {
     const state = await this.ports.sessionState.get(sessionId)
     if (!state) {

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' }
 

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<void> {
     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)

src/main/agent/manager/deepChatAgentBackend.ts

@@ -54,6 +54,10 @@ export interface DeepChatAgentBackendPort {
   listPendingInputs(sessionId: AppSessionId): Promise<PendingSessionInputRecord[]>
   isPendingQueueResumeAvailable(sessionId: AppSessionId): Promise<boolean>
   resumePendingQueue(sessionId: AppSessionId): Promise<boolean>
+  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)

src/main/agent/manager/sessionHandles.ts

@@ -85,6 +85,7 @@ export interface DeepChatControlFacet {
   compact(): Promise<{ compacted: boolean; state: SessionCompactionState }>
   isPendingQueueResumeAvailable(): Promise<boolean>
   resumePendingQueue(): Promise<boolean>
+  retryPendingQueueInput(itemId: string): Promise<{ accepted: boolean; started: boolean }>
 }
 
 export interface DeepChatSessionHandle extends AgentSessionHandle {

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 }
       }

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',