diff --git a/docs/architecture/deepchat-agent-harness-boundaries/spec.md b/docs/architecture/deepchat-agent-harness-boundaries/spec.md index 6f5b42a2c3..38c687303e 100644 --- a/docs/architecture/deepchat-agent-harness-boundaries/spec.md +++ b/docs/architecture/deepchat-agent-harness-boundaries/spec.md @@ -98,11 +98,11 @@ fence that matches their operation. existing order: `sessions.status.changed`, `sessions.updated`, internal session update, then UI refresh. - `PendingInputAdmissionCoordinator` owns send/steer normalization, attachment acceptance lanes, - capacity and interaction gates, queue mutation commands, steer promotion, and interrupt - requests. + capacity and interaction gates, queue mutation commands, steer promotion, explicit Queue resume, + and interrupt requests. Listing pending inputs is a pure read. - `PendingInputPump` owns steer-before-queue selection, per-session single-flight drain, claim, - release, consume, recovery, and starting a claimed turn. Durable input records remain owned by - `SessionPendingInputs`. + release, consume, process-local restart holds, recovery, and starting a claimed turn. Durable + input records remain owned by `SessionPendingInputs`. - `TurnCoordinator` continues to own the distinct initial and resume preparation algorithms, but delegates lifecycle settlement and pending-input disposition instead of mutating those owners through parent callbacks. @@ -126,6 +126,32 @@ event bus is not used because claim settlement and the next drain must observe a order. Public `MessageStartResult`, including `attachmentPreparation`, remains unchanged and is mapped losslessly from the internal completion. +### Pending Input Restart Boundary + +Harness construction reconciles active pending inputs before transcript recovery and before any +Session scope is hydrated. `SessionPendingInputs` returns the affected Session IDs and Queue input +IDs that must be held. It terminalizes unread Steer messages, their Tape replacements, and their +pending rows atomically instead of handing them to a later global recovery phase. + +The pump installs Queue IDs in a process-local hold set. Automatic `enqueue` and `completed` wakes +stop at a held Queue FIFO head; pending-input listing and Session hydration never schedule a drain. +An explicit composer Send may start without releasing historical drafts. The typed +`sessions.resumePendingQueue` operation validates normal Session gates, releases the Session hold, +and drains with a `manual` wake reason. Before that manually resumed turn enters the provider Run, +the pump consumes its Queue claim; a returned provider error therefore cannot restore the same item +to Queue. Manual intent is tracked by Queue item ID, so attachment block and resolution cannot erase +that boundary. The pending-input list projects authoritative resume availability, and Resume returns +`false` without draining when no hold exists. Failures before the provider boundary still release +safely. Durable Queue rows remain the only ordering and content source, and a later restart derives +a new hold from whatever rows remain. + +An unclaimed Steer belongs to the previous process's interrupted delivery attempt. Startup changes +its linked messages to retryable `error`, appends their Tape replacements, and consumes its pending +row in one transaction. The renderer hides its receipt and adds no recovery-specific action; the +standard message toolbar remains unchanged. A claimed Steer keeps its sent user message and exposes +the interruption through the assistant error. Neither path adds a persisted recovery marker, age +policy, schema migration, or hydration-triggered execution. + ### Test Boundary Before production extraction, tests stop reflecting private coordinator members. Existing cases diff --git a/docs/architecture/session-management.md b/docs/architecture/session-management.md index 21618f4fe9..cf97f5adc5 100644 --- a/docs/architecture/session-management.md +++ b/docs/architecture/session-management.md @@ -66,13 +66,28 @@ Steer: user message (Unread) -> claim (Read) -> new assistant message 仍是 source of truth。 - receipt 只由持久化 `readAt` 派生为 `Unread` 或 `Read`。`Read` 的短时显示和淡出属于 renderer， 不产生延迟数据库写入。 +- 冷启动是单独的终止边界：上个进程尚未 claim 的 Steer 会消费 pending row，并把 linked user + message 内部终止为 `error`。renderer 不展示恢复专用 receipt 或按钮，仍保留普通消息工具栏；已经 + claim 的 Steer 保留 sent user fact，由中断的 assistant message 承载失败。 ## 恢复与查询 - `sessions.restore` 返回最近一页，`sessions.listMessagesPage` 使用 keyset pagination 拉取旧历史。 - 普通 list/history/binding query 不 hydrate Agent instance。 -- active session 的 pending-input restore 若发现 `pending` Steer，会唤醒对应 backend 的正常 drain； - 这是重启前已发送 `Unread` 消息的执行恢复点，不改变普通 transcript/history query。 +- DeepChat harness 构造时会先 reconcile active pending inputs。历史 Queue row 保持 durable，并进入 + process-local restart hold；Session 打开、hydrate、消息查询和 pending-input list 都不会释放 hold + 或触发执行。用户通过 Queue lane 的 `Resume queue` 显式释放当前 Session 的 hold，之后继续沿用 + Steer-first 与 Queue FIFO drain 规则。手动恢复的 Queue head 在写入 user/assistant fact 并进入 + provider Run 前消费；manual 标记绑定 Queue item ID，附件 block 后的 retry/degrade 不会丢失该语义。 + 此后的 provider error 只保留在 transcript，不会把同一条目放回 Queue。pending-input list 同时返回 + 后端计算的 resume availability；普通 live Queue 不显示该操作，无 hold 的 Resume 不触发 drain。 +- 重启前尚未 claim 的 Steer 不再自动 drain：reconciliation 消费对应 pending row，并把 linked + pending user message、Tape replacement 与 row 消费在同一事务内终结为 `error`，UI 不显示 + receipt。用户通过普通消息工具栏 Retry 时走普通 turn，且不会释放同 Session 的历史 Queue hold。 +- claimed Queue 创建 user message 时会在同一事务内回写 pending row 的 message ID，使冷启动可以 + 明确区分尚未开始的 draft 与已物化的 transcript fact。 +- restart hold 仅由现有 active Queue ID 派生，不持久化、不改变 Queue 排序和容量，也不新增 schema； + 再次重启会从剩余 durable row 重建。pending-input list 是纯读。 - DeepChat harness 构造时在 pending-input 与 transcript recovery 之前分类 Execution Journal。存在 dispatch-without-outcome、corruption 或缺失 terminal 的 Run 只输出结构化 parked 诊断，不依据该报告 自动重放遗留 operation；分类先于 runtime graph 构建，Journal 读取失败会阻止 harness 构造。诊断最多 diff --git a/docs/features/im-style-steer-messages/spec.md b/docs/features/im-style-steer-messages/spec.md index 11a8b31e64..3eb3f7e5cb 100644 --- a/docs/features/im-style-steer-messages/spec.md +++ b/docs/features/im-style-steer-messages/spec.md @@ -25,8 +25,10 @@ Queue = mutable draft, normal admission time, no receipt Steer = immutable sent message, interrupt admission time, Unread -> Read receipt ``` -The receipt has only two named states: `Unread` and `Read`. It disappears after `Read`; disappearance -is presentation, not a third state. User-facing Chinese copy is `未读` / `已读`. +The live receipt has only two named states: `Unread` and `Read`. It disappears after `Read`; +disappearance is presentation, not a third live state. User-facing Chinese copy is `未读` / `已读`. +Cold restart before claim terminalizes the message internally as `error` and suppresses the receipt; +it does not add a third delivery label or a recovery-specific action. ## Problem @@ -135,6 +137,7 @@ Promoting a Queue item to Steer is a one-way admission transition: | `pending` | `pending` | `readAt: null` | `Unread` | Copy | | `claimed` | `pending` | `readAt: ` | `Read`, then disappears | Copy | | `consumed` | `sent` | Original `readAt` retained | No receipt after timeout | Normal message actions | +| Cold restart before claim | `error` | `readAt: null` retained | No receipt | Normal message actions | `ChatMessageRecord.status = 'pending'` prevents an accepted-but-unsettled Steer from entering later context as historical input. `isContextHistoryRecord` already excludes pending user messages. @@ -159,6 +162,12 @@ The claim transition is irreversible for UI semantics. A failure after claim pro assistant error or interruption below the Steer; it does not move the message back to `Unread` or silently delete it. +A cold restart before claim is different from a live failure after claim. The accepting runtime no +longer exists, so startup consumes the active Steer row and recovers its linked user messages to +`error`. The renderer uses that status to hide the stale live receipt and otherwise keeps the normal +user-message rendering. The standard toolbar Retry action starts a normal turn because there is no +active turn left to steer. + ### Rapid consecutive Steers The existing payload merge is retained, but the UI no longer merges user messages: @@ -455,12 +464,13 @@ It receives no `Unread` / `Read` receipt. | Persistence fails before acceptance | No partial pending row or message; keep draft and show error | | Renderer misses acceptance event | Route result inserts the persisted message; later restore is authoritative | | Renderer misses claim event | Session restore reconstructs `readAt` and the new assistant row | -| App restarts during pre-stream handoff | Keep the materialized source user fact, consume its claimed Queue record, and resume the `Unread` Steer | +| App restarts during pre-stream handoff before Steer claim | Keep the materialized source user fact, consume its claimed Queue record, and terminalize the Steer internally as `error` without a receipt | | Previous DeepChat turn errors | Open the safe boundary and drain the durable Steer unless an interaction blocks it | | Previous ACP turn cancellation fails | Keep Steer `Unread`; do not claim until the old operation is terminal | | Runtime fails after claim | Keep user messages `Read`; settle a new assistant error row; do not delete or retry silently | -| App restarts before claim | Restore the Steer as `Unread` and resume normal pending-input drain | +| App restarts before claim | Atomically fail linked user messages, append Tape replacements, consume the Steer row, hide the receipt, and leave the standard toolbar Retry action available | | App restarts after claim | Restore the persisted `Read` receipt and settlement facts; never duplicate user rows | +| App restarts with Queue drafts | Keep rows in Queue and hold them from automatic drain until explicit `Resume queue`; once the manually resumed head enters the provider Run, consume it so a provider error cannot restore it to Queue | | Session is switched | Keep lifecycle in main; active renderer derives the state when restored | | Session is deleted | Delete transcript and pending-input facts through the existing session deletion transaction | @@ -477,10 +487,15 @@ It receives no `Unread` / `Read` receipt. 7. The claimed batch payload is supplied exactly once as the new loop input. 8. Settlement marks all linked user messages `sent` and appends the corresponding Tape replacement facts. -9. Search and transcript restore may show an accepted `Unread` Steer because it is a real sent fact. +9. Reload in the same process may show an accepted `Unread` Steer because it is a real sent fact; + cold-start recovery terminalizes an unclaimed Steer as `error` and shows no receipt instead. 10. Event delivery is a cache update, never the source of truth. 11. Pre-stream acceptance materializes and links the current claimed Queue user fact in the same transaction as the Steer, before assigning the Steer's `orderSeq`. +12. Normal claimed Queue materialization also creates the user fact and links its pending row in one + transaction. +13. Restart terminalization changes every unread Steer message to `error`, appends its Tape + replacement, and consumes the Steer row in one transaction. ## Acceptance Criteria @@ -502,6 +517,10 @@ It receives no `Unread` / `Read` receipt. - Queue items remain editable and reorderable in the bottom lane. - Queue promotion creates a visible `Unread` Steer only after successful preparation. - Normal Queue drain creates no receipt. +- Queue drafts retained across cold restart do not drain from hydration or lifecycle wakes and expose + `Resume queue` only when the backend reports an actual restart hold while the Session is idle. +- A manually resumed Queue head is consumed before its provider Run and does not return to Queue + after a provider error, including after attachment Retry or Send without image content. ### Reliability @@ -509,6 +528,9 @@ It receives no `Unread` / `Read` receipt. the last bubble. - A Steer accepted after claim belongs below the newly reserved assistant message. - Reload and restart never duplicate, hide, or reorder accepted Steer messages. +- Cold restart never executes an unclaimed Steer implicitly; it renders without a receipt and the + standard toolbar Retry action starts one normal turn. +- Retrying a restart-failed Steer does not release retained Queue drafts. - A post-claim failure never removes or reverts the user message. - DeepChat and ACP satisfy the same visible ordering contract. diff --git a/docs/issues/pending-input-restart-recovery/spec.md b/docs/issues/pending-input-restart-recovery/spec.md new file mode 100644 index 0000000000..cf681157ad --- /dev/null +++ b/docs/issues/pending-input-restart-recovery/spec.md @@ -0,0 +1,536 @@ +# Pending Inputs Lack a Deterministic Restart Disposition + +## GitHub + +- Issue: https://github.com/ThinkInAIXYZ/deepchat/issues/2111 +- Classification: complex reliability bug +- Priority: P1, with P0-level user perception when an accepted input appears lost +- Status: implemented on `codex/issue-2111-pending-input-recovery` + +## Decision + +A cold restart must not automatically execute any input accepted by the previous process. + +- Queue remains an unsent draft in the normal Queue lane. The user explicitly resumes the Queue. +- An unread Steer becomes an internal terminal `error` transcript message without recovery-specific + UI. The existing message toolbar remains available if the user wants to retry it. +- A Steer that had already been claimed remains sent; its interrupted assistant response becomes the + failure surface and can be retried normally. +- Full Session restoration only projects the reconciled state. It does not wake the pending-input + pump. + +This replaces the earlier age-based recovery proposal. There is no 24-hour policy, recovery card, +sidebar warning, recovery timestamp, or database migration. + +## Issue + +Queue and Steer inputs are persisted in `deepchat_pending_inputs`, but a process restart rebuilds +only part of their durable state. The startup path repairs rows left in `claimed` and recovers +pending messages, then returns the runtime graph without giving every remaining input a stable +user-visible disposition. + +The current behavior is inconsistent: + +- a pending Queue can remain stored without an explicit way to start it while the Session is idle; +- a pending Steer can remain visible as `Unread` even though the process that accepted it no longer + exists; +- `PendingInputAdmissionCoordinator.list()` may schedule a pending Steer merely because the + renderer listed it; +- a later lifecycle wake can execute an old Queue even if opening the restored Session did not. + +The defect is therefore not only a missing wakeup. The restart boundary fails to distinguish an +unsent Queue draft from a sent Steer attempt. + +## Confirmed Code Path + +```text +createDeepChatRuntimeServices() + -> bind PendingInputWakeup to PendingInputPump + -> recoverInputsAfterRestart() + -> install process-local Queue holds + -> atomically fail unread Steer messages and consume their pending rows + -> recoverPendingMessages() + -> return runtime services + +later, when the renderer restores a Session + -> sessions.restore + -> SessionQuery.getSession() + -> DeepChat session handle snapshot() + -> SessionStateResolver.get() + -> rebuild runtime state + -> renderer lists pending inputs + -> list() may schedule pending Steer as a side effect +``` + +`PendingInputPump.drain()` requires a hydrated runtime scope, so scheduling during startup is not a +reliable repair. Scheduling after full hydration would make historical inputs execute without a +fresh user decision. The selected fix removes that requirement instead of moving the automatic +wake to another lifecycle edge. + +## User-visible Semantics + +| State at process exit | Cold-start reconciliation | Restored Session | User action | +| --- | --- | --- | --- | +| Queue `pending` | Keep the durable row and hold it from automatic drain | Normal Queue item | Resume Queue | +| Queue `claimed`, no user message | Release to `pending` and hold it | Normal Queue item | Resume Queue | +| Queue `claimed`, user message exists | Consume the row; recover the interrupted transcript turn | Interrupted message/response | Retry message | +| Steer `pending` / `Unread` | Mark linked user messages `error`; consume the Steer row | Normal user message without a receipt | Existing message Retry action | +| Steer `claimed` / `Read` | Settle the user message; consume the row; recover assistant as interrupted | Failed assistant response | Retry response | +| Attachment-blocked input | Preserve the existing Queue/attachment recovery behavior | Queue item with attachment actions | Resolve attachment | + +The distinction is semantic: + +- Queue means “save this for later”; no transcript fact exists until it is claimed. +- Steer means “I already sent this into the active conversation”; its transcript fact must remain, + but the interrupted delivery attempt must not stay `Unread` forever. + +## Root Cause + +1. Startup repair treats pending Queue and pending Steer as work that may still be executed, even + though they represent different user commitments. +2. A durable `pending` row is level-triggered evidence, while pump scheduling is an in-memory edge. + Losing or later recreating that edge produces inconsistent behavior. +3. `PendingInputAdmissionCoordinator.list()` compensates only for Steer and turns a read into an + execution trigger. +4. The Queue UI has mutation actions but no explicit idle-state action that resumes a restored + Queue. +5. Merely leaving a historical Queue as `pending` is insufficient: a later enqueue or completion + wake can still drain it. The runtime must temporarily hold startup Queue rows until the user + resumes them. + +## Governing Invariants + +1. Cold startup, Session hydration, message listing, and pending-input listing never execute a + historical input. +2. `deepchat_pending_inputs` remains the durable source of Queue content and ordering. +3. A startup Queue hold is process-local safety state derived from existing active Queue IDs. It is + not a second durable queue and does not require a schema field. +4. Restarting again safely rebuilds the hold from the same durable rows. +5. A restart-failed Steer remains a transcript fact, is excluded from model context while its + message status is `error`, renders without a receipt, and is removed only by the existing + retry/truncation contract. +6. `PendingInputPump` remains the only owner of Queue selection, lease acquisition, durable claim, + and turn start. +7. Interaction, attachment, active-run, claimed-input, and instance-fence gates remain + authoritative. +8. Duplicate Resume or Retry actions may request duplicate work but must create at most one claim + and one turn. +9. Diagnostics contain no message text, attachment paths, serialized payloads, or other user + content. +10. Materializing a claimed Queue user message and linking that message to its pending-input row + commit or roll back together. +11. Failing unread Steer messages, recording their Tape replacements, and consuming the Steer row + commit or roll back together. +12. Manual-resume intent belongs to the resumed Queue item, not to a transient pump wake reason; it + survives attachment block and resolution until that item is consumed or explicitly removed. + +## Fix Design + +### 1. Reconcile restart state before any Session is restored + +Replace the claimed-only startup repair with one deterministic reconciliation pass over active +pending inputs. + +For Queue: + +- keep `pending` rows unchanged; +- release `claimed` rows without a materialized user message back to `pending`; +- treat a linked but missing or foreign user message as a dangling rollback association, clear the + link while releasing the row, and keep the Queue draft; +- preserve the existing terminal recovery for `claimed` rows that already materialized a user + message; +- record the IDs of retained/released Queue rows in a process-local `restartHeldQueueInputIds` set. + +For Steer: + +- for each `pending` Steer, ensure its linked user message exists, change every linked message from + `pending` to `error`, append the corresponding Tape replacements, and mark the Steer row + `consumed` in one SQLite transaction; +- for `claimed` Steer, retain the existing behavior: settle linked user messages, consume the row, + and let pending assistant recovery create the interrupted error response; +- preserve the existing attachment-blocked conversion and resolution path. + +The pass is idempotent. Consumed Steer rows are not active on the next restart, and retained Queue +rows simply receive a new process-local hold. + +No Session runtime is created during this scan. Startup logs report aggregate counts only. + +### 2. Hold startup Queue rows until an explicit Queue resume + +The process-local hold closes a subtle hole in the simple “leave it in Queue” approach. Without the +hold, any later `enqueue` or `completed` wake could claim the historical head row automatically. + +The hold has narrow behavior: + +- Queue listing, editing, moving, deleting, and capacity counting continue to use the durable rows; +- pump Queue selection stops when the FIFO head is held; +- pending Steer selection remains unaffected during normal live operation; +- an explicit new composer Send may start independently and does not silently release historical + Queue rows; +- deleting or promoting a held Queue row removes its ID from the hold; +- once no held IDs remain, the overlay disappears naturally; +- another process restart derives a new hold from whatever Queue rows remain. + +Queue transcript materialization also closes its crash window: creating the claimed Queue's user +message and linking that message ID back to the claimed row use one transaction. Startup can then +distinguish an unstarted draft from a turn whose transcript fact already exists without guessing +from timing. + +Keep the set inside the existing pending-input runtime owner. Do not add a new service, database +column, timer, or global event bus. + +### 3. Add an explicit Resume Queue action + +Add a typed Session route named `sessions.resumePendingQueue`. It performs these steps under the +existing Session operation gate: + +1. require a fully restored Session in `idle` or recoverable `error` state; +2. reject active interactions, attachment blockers, claimed inputs, or an owned drain lease; +3. release the Session's startup Queue hold; +4. request the existing pump to drain with a new explicit `manual` reason. + +The list response exposes whether that Session actually has a restart-held Queue row. The renderer +uses this authoritative value instead of inferring recovery state from the presence of an ordinary +pending Queue item, and Resume returns `false` without starting a drain when no hold was released. + +The action resumes the existing Steer-before-Queue and Queue FIFO rules. One click means “continue +this Queue”; after the first item completes, later Queue items may continue normally. Reorder or +delete remains available before resuming. + +Repeated clicks are harmless because the instance drain lease and durable claim remain the +exact-once fences. If the turn cannot start after the hold is released, the durable row remains +pending and can be selected by the next valid lifecycle wake. A later process restart holds it +again. + +Once the manually resumed head has persisted its user and assistant facts and is about to enter the +provider Run, its Queue claim is consumed like an explicit Send. A returned provider error therefore +stays in the transcript and cannot put the same item back into Queue. Failures before that boundary +still release the claim so no accepted draft is lost. + +If attachment preparation blocks the resumed head, the process-local manual marker remains attached +to that item ID. Retrying or degrading the attachment can wake the pump with the ordinary `enqueue` +reason without losing manual-resume semantics. The marker is cleared only after consumption or an +explicit delete/Steer conversion. + +### 4. Terminalize pending Steer as a retryable failure + +A Steer accepted by the previous process must not be demoted to Queue: it already has a visible user +message and Tape fact. It also must not remain `Unread`, because the accepting runtime is gone. + +Cold-start reconciliation therefore applies this terminal transition: + +```text +pending Steer row + pending user message + -> Steer row consumed + -> user message status error + -> receipt hidden +``` + +The existing `inputReceipt.mode = 'steer'` metadata and `message.status = 'error'` are sufficient to +exclude the interrupted attempt from later model context and suppress its live receipt. No new +persisted failure marker is required. + +Retry uses the existing message retry semantics: + +- remove the failed user message and later invalidated transcript suffix; +- resend its original content as a normal turn because there is no active turn to steer after + restart; +- keep restart-held Queue drafts intact; +- allow this narrowly identified restart-failed Steer retry to coexist with held Queue rows; +- keep the ordinary retry prohibition for genuinely executable pending inputs. + +For a Steer already claimed before the crash, the user message was successfully read. It remains +`sent`; the interrupted assistant message carries the failure and the existing assistant Retry +action restarts the turn. + +### 5. Make pending-input listing a pure read + +Remove the scheduling side effect from `PendingInputAdmissionCoordinator.list()`. + +After the change: + +- Session restoration reads messages and Queue rows; +- lightweight Session summaries remain non-hydrating; +- full Session hydration does not schedule pending work; +- only live admission, run settlement, attachment resolution, and explicit Resume/Retry actions can + request a pump drain. + +### 6. Expose Queue resume without recovery-specific Steer chrome + +Do not add a recovery panel, modal, sidebar warning, or separate recovery list. + +The Queue lane gains one action while the Session is idle and contains a restart-held Queue: + +- `Resume queue` in the lane header; +- disabled while the Session is generating or an interaction/attachment blocker owns the Session; +- existing edit, reorder, Steer, and delete controls remain unchanged. + +The restart-failed Steer remains where the user originally sent it, but renders as an ordinary +historical user message: no `Unread`, no `Failed`, and no inline recovery button. The standard +message toolbar below it remains unchanged and continues to provide the normal Retry command. + +The Queue action copy uses vue-i18n and the existing `DcButton` primitive. + +## UI Change + +BEFORE + +```text +Restored Session ++--------------------------------------------------+ +| You 10:21 Unread | +| Please change the output format | +| | +| Queue (2) | +| 1. Add a short example [Edit] [Delete] | +| 2. Translate it to Chinese [Edit] [Delete] | +| | +| [composer] | ++--------------------------------------------------+ + +The Unread Steer may execute from a read side effect. +The Queue has no clear idle-state resume action. +``` + +AFTER + +```text +Restored Session ++--------------------------------------------------+ +| You 10:21 | +| Please change the output format | +| | +| Queue (2) [Resume queue] | +| 1. Add a short example [Edit] [Delete] | +| 2. Translate it to Chinese [Edit] [Delete] | +| | +| [composer] | ++--------------------------------------------------+ + +Nothing executes merely because the Session was opened. +``` + +Resume availability correction: + +```text +BEFORE — ordinary live Queue ++--------------------------------------------------+ +| Queue (1) [Resume queue] | +| 1. Follow up after this turn | ++--------------------------------------------------+ + +AFTER — ordinary live Queue ++--------------------------------------------------+ +| Queue (1) | +| 1. Follow up after this turn | ++--------------------------------------------------+ + +The action is projected only for a real restart-held Queue. +``` + +## Data and Contract Changes + +| Boundary | Change | +| --- | --- | +| SQLite | No schema or migration change. | +| Pending-input store | Atomically link claimed Queue messages and terminalize unread Steer messages with their rows. | +| Pump | Track startup-held Queue IDs and manually resumed item IDs; skip held FIFO heads; consume a manually resumed head at the provider boundary. | +| Admission | Make `list()` pure and expose Queue resume validation. | +| Session route/client | Add `sessions.resumePendingQueue` and authoritative `resumeAvailable` projection. | +| Transcript recovery | Persist unread Steer `error` status and Tape replacements inside reconciliation. | +| Renderer | Add the Queue resume action and suppress recovery-specific Steer receipt chrome. | + +No provider, model request, permission, attachment-preparation, Session summary, or sidebar contract +changes. + +## Runtime Sequences + +### Restored Queue + +```text +cold startup + -> find active Queue row + -> keep/release it as pending + -> add its ID to restartHeldQueueInputIds +user opens Session + -> full state hydration + -> list Queue as normal + -> no pump wake +user clicks Resume queue + -> validate Session gates + -> release hold + -> pump acquires instance lease + -> durable row is claimed once + -> turn persists user and assistant facts + -> Queue row is consumed before the provider Run + -> a later provider error remains only in the transcript +``` + +### Restored pending Steer + +```text +cold startup + -> find pending Steer and linked user message + -> consume Steer row + -> recover user message to error +user opens Session + -> message renders as a normal historical user message with no receipt + -> no pump wake +user clicks the existing toolbar Retry action + -> existing retry preparation reads original content + -> failed transcript suffix is replaced + -> one normal turn starts + -> held Queue remains held +``` + +### Duplicate user action + +```text +Resume click --------+ +Resume click --------+--> same instance drain lease --> one durable claim --> one turn +later wake -----------+ +``` + +## Affected Ownership + +| Owner | Responsibility in this fix | +| --- | --- | +| `SessionPendingInputs` / store | Deterministic startup reconciliation and Steer terminalization inputs. | +| `PendingInputPump` | Startup Queue hold, explicit resume, manual-run claim policy, gating, and exact-once drain ownership. | +| `PendingInputAdmissionCoordinator` | Pure pending list and typed manual Queue resume validation. | +| Transcript recovery | Convert pending Steer messages to durable error messages. | +| Renderer pending-input/message UI | Render Queue resume and hide recovery-specific Steer chrome. | + +`SessionStateResolver` is deliberately not changed. Full hydration is not an execution signal in +this design. + +## Constraints and Non-goals + +- Do not scan and hydrate every Session at startup. +- Do not execute Queue or Steer because a Session was listed, opened, or fully hydrated. +- Do not add `recovery_required_at`, an age threshold, a recovery lifecycle state, or a schema + migration. +- Do not add a recovery card, modal, sidebar indicator, or recovery-specific delete command. +- Do not delete or demote an accepted Steer transcript fact. +- Do not overload attachment `blocked` state for restart interruption. +- Do not change live Queue FIFO, Steer priority, capacity, provider execution, or permission + semantics. +- Do not add a second durable queue, scheduler, event bus, or state-machine dependency. +- Do not sync or rewrite the GitHub issue; it is already linked and contains the source report. + +## Task Checklist + +- [x] Extend startup pending-input reconciliation to collect held Queue IDs and terminalize pending + Steer rows. +- [x] Atomically fail unread Steer messages, append Tape replacements, and consume their rows. +- [x] Add the process-local startup Queue hold to `PendingInputPump` selection and cleanup paths. +- [x] Remove the scheduling side effect from `PendingInputAdmissionCoordinator.list()`. +- [x] Add the typed `sessions.resumePendingQueue` route/client and Session operation-gate handling. +- [x] Add the Queue lane `Resume queue` action and its disabled/busy states. +- [x] Hide the receipt for `error + steer receipt` and rely on the standard message Retry action. +- [x] Allow restart-failed Steer Retry while restart-held Queue drafts remain non-executable. +- [x] Update `docs/architecture/session-management.md`, + `docs/features/im-style-steer-messages/spec.md`, and + `docs/architecture/deepchat-agent-harness-boundaries/spec.md` with the implemented contract. +- [x] Add focused reconciliation, pump, admission, retry, store, and component tests. +- [x] Run format, i18n, lint, typecheck, and focused main/renderer suites. +- [x] Atomically create and link claimed Queue user messages. +- [x] Preserve manual-resume semantics across attachment block and resolution. +- [x] Project authoritative Queue resume availability and reject no-hold Resume calls. +- [x] Add crash-window and attachment-resolution regressions, then rerun validation. + +## Validation Plan + +### Restart reconciliation + +- A pending Queue survives restart unchanged and is added to the process-local hold. +- A claimed Queue without a user message is released and held. +- A claimed Queue with only a dangling message ID is released, unlinked, and held. +- A claimed Queue with a user message follows existing interrupted-turn recovery. +- A pending Steer with one or multiple linked messages consumes its row and marks every linked user + message `error`. +- Injecting a failure during multi-message Steer terminalization rolls back every message, Tape + replacement, and the pending-row transition; the next reconciliation succeeds once. +- A claimed Steer keeps read user messages sent and recovers its assistant response to error. +- Repeating startup reconciliation changes no already-terminal Steer row and safely rebuilds Queue + holds. +- Attachment-blocked inputs retain their existing resolution behavior. + +### No implicit execution + +- Harness construction, lightweight Session listing, full Session hydration, message listing, and + pending-input listing start zero turns. +- An enqueue or completion wake cannot bypass a held Queue FIFO head. +- Sending a new composer message does not silently release the historical Queue hold. +- Removing every held Queue row clears the overlay without leaving the Session blocked. + +### Explicit Queue resume + +- Resume Queue starts the FIFO head exactly once. +- An ordinary live Queue does not expose Resume, and calling Resume without a restart hold returns + `false` without draining. +- Once the manually resumed head reaches the provider boundary, it is consumed and cannot reappear + after a returned provider error. +- Attachment Retry and Send without image content preserve that same consumption boundary for the + manually resumed item even though resolution wakes the pump as `enqueue`. +- Later Queue items follow the existing completion/FIFO contract after explicit resume. +- Duplicate Resume calls, a held drain lease, stale instance, active interaction, attachment block, + or concurrent lifecycle wake cannot create a duplicate claim or turn. +- A restart before claim safely holds the row again. + +### Steer failure and retry + +- A restored pending Steer renders no receipt or recovery-specific action and has no active pending + row. +- Failed Steer content is excluded from later model context until retry. +- Retry replaces the failed message through the existing transcript retry contract and starts one + normal turn. +- Restart-held Queue rows remain intact and held while the failed Steer is retried. +- A claimed Steer renders the existing interrupted assistant error and retries without duplicating + the user message. + +### Focused suites + +- `test/main/session/data/pendingInputs.test.ts` +- `test/main/session/data/pendingInputStore.test.ts` +- `test/main/agent/deepchat/runtime/pendingInputPump.test.ts` +- `test/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.test.ts` +- `test/main/agent/deepchat/harness/deepChatAgentHarness.test.ts` +- `test/main/session/turn.test.ts` +- `test/renderer/stores/pendingInputStore.test.ts` +- `test/renderer/features/chat-page/composables/usePendingInputActions.test.ts` +- `test/renderer/components/PendingInputLane.test.ts` +- `test/renderer/components/message/MessageItemUser.test.ts` + +## Validation Results + +- `pnpm format`, `pnpm i18n`, `pnpm lint`, and `pnpm typecheck` pass. +- The focused main-process run passes 515/515 tests across ten affected and route-boundary files. +- The focused renderer run passes 34/34 tests; the full renderer run passes 2056/2056 tests. +- The full main-process run is not green in this Windows environment: 6544 tests pass, 394 are + skipped, and 60 fail across 27 unchanged files. The failures are dominated by POSIX path + expectations, Windows symbolic-link permissions, executable-bit checks, and packaging-script + fixtures. None of the failing files is part of this change, and every affected main-process suite + passes in the focused run. +- The installed Node.js version is 24.15.0 while the repository requests at least 24.18.0; pnpm emits + an engine warning, but the completed checks above run successfully. + +## Acceptance Criteria + +1. No historical Queue or Steer starts because the app launched, a Session hydrated, or a read route + ran. +2. Restored Queue drafts remain visible, editable, reorderable, deletable, and explicitly + resumable from the existing Queue lane. +3. A restored Queue remains held across unrelated lifecycle wakes until the user resumes it. +4. Explicit Queue resume starts the FIFO drain exactly once; the resumed head is consumed at the + provider boundary, cannot reappear after a provider error, and retains existing subsequent Queue + behavior. +5. A pending Steer becomes an internal terminal transcript message, renders without a receipt, and + no longer occupies active pending-input state. +6. The standard Retry action for that user message starts one normal turn while preserving + restart-held Queue drafts. +7. A claimed Steer retains its sent user fact and exposes failure through the interrupted assistant + response. +8. Pending-input listing is a pure read, and no Session hydration-specific wake is added. +9. No schema migration, age policy, recovery marker, recovery panel, or sidebar indicator is + introduced. diff --git a/src/main/agent/deepchat/harness/createDeepChatAgentHarness.ts b/src/main/agent/deepchat/harness/createDeepChatAgentHarness.ts index e2241e02b1..3902004e6d 100644 --- a/src/main/agent/deepchat/harness/createDeepChatAgentHarness.ts +++ b/src/main/agent/deepchat/harness/createDeepChatAgentHarness.ts @@ -377,6 +377,7 @@ function createDeepChatRuntimeServices(deps: DeepChatHarnessDependencies): DeepC toolService, sessionStore, messageStore, + pendingInputs: pendingInputCoordinator, tapeReconciliation: tapeService, toolResolver, compactionService, @@ -499,10 +500,11 @@ function createDeepChatRuntimeServices(deps: DeepChatHarnessDependencies): DeepC input ) - const recoveredPendingInputs = pendingInputCoordinator.recoverClaimedInputsAfterRestart() - if (recoveredPendingInputs > 0) { + const pendingInputRecovery = pendingInputCoordinator.recoverInputsAfterRestart() + pendingInputPump.holdRestartedQueueInputs(pendingInputRecovery.heldQueueInputIds) + if (pendingInputRecovery.affectedSessionIds.size > 0) { logger.info( - `DeepChatAgent: recovered ${recoveredPendingInputs} sessions with claimed pending inputs` + `DeepChatAgent: reconciled ${pendingInputRecovery.affectedSessionIds.size} sessions with pending inputs` ) } diff --git a/src/main/agent/deepchat/harness/deepChatAgentHarness.ts b/src/main/agent/deepchat/harness/deepChatAgentHarness.ts index c2070d97b5..cfb5e2b3d4 100644 --- a/src/main/agent/deepchat/harness/deepChatAgentHarness.ts +++ b/src/main/agent/deepchat/harness/deepChatAgentHarness.ts @@ -114,6 +114,14 @@ export class DeepChatAgentHarness return this.services.pendingInputAdmission.list(sessionId) } + async isPendingQueueResumeAvailable(sessionId: string): Promise { + return this.services.pendingInputAdmission.isPendingQueueResumeAvailable(sessionId) + } + + async resumePendingQueue(sessionId: string): Promise { + return await this.services.pendingInputAdmission.resumePendingQueue(sessionId) + } + async queuePendingInput( sessionId: string, content: string | SendMessageInput, @@ -259,8 +267,11 @@ export class DeepChatAgentHarness this.services.transcriptMutation.finishClearMessages(sessionId) } - prepareRetry(sessionId: string): Promise<{ projectDir: string | null }> { - return this.services.transcriptMutation.prepareRetry(sessionId) + prepareRetry( + sessionId: string, + options?: { allowRestartHeldQueue?: boolean } + ): Promise<{ projectDir: string | null }> { + return this.services.transcriptMutation.prepareRetry(sessionId, options) } cancelForTranscriptMutation(sessionId: string): Promise { diff --git a/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts b/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts index c321d4292f..6827547d13 100644 --- a/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts +++ b/src/main/agent/deepchat/runtime/pendingInputAdmissionCoordinator.ts @@ -36,6 +36,7 @@ export type PendingInputAdmissionStorePort = Pick< | 'getInput' | 'hasActiveInputs' | 'hasBlockingInput' + | 'hasClaimedInput' | 'isAtCapacity' | 'listPendingInputs' | 'moveQueuedInput' @@ -69,6 +70,10 @@ export interface PendingInputAdmissionPumpPort { sessionId: string, itemId: string ): ClaimedPendingInputHandle + releaseRestartHoldForInput(itemId: string): void + releaseRestartHoldForSession(sessionId: string): boolean + hasRestartHeldQueueInputs(sessionId: string): boolean + hasOnlyRestartHeldQueueInputs(sessionId: string): boolean } export interface PendingInputAdmissionCoordinatorPorts { @@ -88,11 +93,11 @@ export class PendingInputAdmissionCoordinator { constructor(private readonly ports: PendingInputAdmissionCoordinatorPorts) {} list(sessionId: string): PendingSessionInputRecord[] { - const inputs = this.ports.pendingInputs.listPendingInputs(sessionId) - if (inputs.some((input) => input.mode === 'steer' && input.state === 'pending')) { - this.ports.pump.schedule(sessionId, 'enqueue') - } - return inputs + return this.ports.pendingInputs.listPendingInputs(sessionId) + } + + isPendingQueueResumeAvailable(sessionId: string): boolean { + return this.ports.pump.hasRestartHeldQueueInputs(sessionId) } async queue( @@ -356,8 +361,12 @@ export class PendingInputAdmissionCoordinator { if (activeGeneration) { this.requireActiveAssistantMessage(sessionId, activeGeneration.messageId) - return this.ports.pendingInputs.promoteQueuedInputToSteerMessage(sessionId, itemId) - .pendingInput + const promoted = this.ports.pendingInputs.promoteQueuedInputToSteerMessage( + sessionId, + itemId + ).pendingInput + this.ports.pump.releaseRestartHoldForInput(itemId) + return promoted } if ( @@ -379,6 +388,7 @@ export class PendingInputAdmissionCoordinator { if (!preStreamController.signal.aborted) { preStreamController.abort(PENDING_INPUT_ABORT_REASON) } + this.ports.pump.releaseRestartHoldForInput(itemId) return accepted.pendingInput } @@ -386,6 +396,7 @@ export class PendingInputAdmissionCoordinator { sessionId, itemId ).pendingInput + this.ports.pump.releaseRestartHoldForInput(itemId) releaseAcceptanceLane() const started = await this.ports.pump.drain(sessionId, 'enqueue') if (!started) { @@ -400,6 +411,7 @@ export class PendingInputAdmissionCoordinator { async deletePendingInput(sessionId: string, itemId: string): Promise { await this.ensureSessionReady(sessionId) this.ports.pendingInputs.deletePendingInput(sessionId, itemId) + this.ports.pump.releaseRestartHoldForInput(itemId) this.ports.pump.schedule(sessionId, 'enqueue') } @@ -417,10 +429,39 @@ export class PendingInputAdmissionCoordinator { return record } - assertNoActiveInputs(sessionId: string): void { + async resumePendingQueue(sessionId: string): Promise { + const state = await this.ports.sessionState.get(sessionId) + if (!state) { + throw new Error(`Session ${sessionId} not found`) + } + if (this.ports.pendingInputs.hasBlockingInput(sessionId)) { + throw new Error('Resolve the blocked attachment input before resuming the queue.') + } + if (this.ports.pendingInputs.hasClaimedInput(sessionId)) { + return false + } + if (!this.ports.pump.canDrain(sessionId, state.status, 'manual')) { + return false + } + if (!this.ports.pump.releaseRestartHoldForSession(sessionId)) { + return false + } + return await this.ports.pump.drain(sessionId, 'manual') + } + + assertNoActiveInputs( + sessionId: string, + options?: { allowRestartHeldQueue?: boolean } + ): void { if (!this.ports.pendingInputs.hasActiveInputs(sessionId)) { return } + if ( + options?.allowRestartHeldQueue === true && + this.ports.pump.hasOnlyRestartHeldQueueInputs(sessionId) + ) { + return + } throw new Error('Please clear the waiting lane before mutating chat history.') } diff --git a/src/main/agent/deepchat/runtime/pendingInputPump.ts b/src/main/agent/deepchat/runtime/pendingInputPump.ts index c85b587a8f..9227f4dacf 100644 --- a/src/main/agent/deepchat/runtime/pendingInputPump.ts +++ b/src/main/agent/deepchat/runtime/pendingInputPump.ts @@ -75,6 +75,7 @@ export interface PendingInputTurnStarter { export interface PendingInputTurnContext { projectDir: string | null claimedInput: ClaimedPendingInputHandle + consumeClaimBeforeProviderStream: boolean } export interface PendingInputPumpPorts { @@ -206,9 +207,60 @@ class DurablePendingInputClaim implements ClaimedPendingInputHandle { export class PendingInputPump { private readonly launchedDrains = new Map() private readonly deferredWakeups = new Map() + private readonly restartHeldQueueInputIds = new Set() + private readonly manuallyResumedQueueInputIds = new Set() constructor(private readonly ports: PendingInputPumpPorts) {} + holdRestartedQueueInputs(inputIds: Iterable): void { + for (const inputId of inputIds) { + this.restartHeldQueueInputIds.add(inputId) + } + } + + releaseRestartHoldForInput(itemId: string): void { + this.restartHeldQueueInputIds.delete(itemId) + this.manuallyResumedQueueInputIds.delete(itemId) + } + + releaseRestartHoldForSession(sessionId: string): boolean { + const heldQueueInputs = this.ports.pendingInputs + .listPendingInputs(sessionId) + .filter((input) => input.mode === 'queue' && this.restartHeldQueueInputIds.has(input.id)) + .sort((left, right) => (left.queueOrder ?? 0) - (right.queueOrder ?? 0)) + const resumedHead = heldQueueInputs[0] + if (!resumedHead) { + return false + } + for (const input of heldQueueInputs) { + this.restartHeldQueueInputIds.delete(input.id) + } + this.manuallyResumedQueueInputIds.add(resumedHead.id) + return true + } + + hasRestartHeldQueueInputs(sessionId: string): boolean { + return this.ports.pendingInputs + .listPendingInputs(sessionId) + .some((input) => input.mode === 'queue' && this.restartHeldQueueInputIds.has(input.id)) + } + + hasOnlyRestartHeldQueueInputs(sessionId: string): boolean { + if ( + this.ports.pendingInputs.hasBlockingInput(sessionId) || + this.ports.pendingInputs.hasClaimedInput(sessionId) + ) { + return false + } + const inputs = this.ports.pendingInputs.listPendingInputs(sessionId) + return ( + inputs.length > 0 && + inputs.every( + (input) => input.mode === 'queue' && this.restartHeldQueueInputIds.has(input.id) + ) + ) + } + hasInteractionBlocker(sessionId: string): boolean { const snapshot = this.readGateSnapshot(sessionId) return ( @@ -243,8 +295,12 @@ export class PendingInputPump { if (!this.canDrainWithSnapshot(sessionId, status, 'enqueue', snapshot)) { return false } + const hasWaitingTurnInput = + source === 'send' + ? this.hasExecutablePendingTurnInput(sessionId) + : this.ports.pendingInputs.hasPendingTurnInput(sessionId) return ( - !this.ports.pendingInputs.hasPendingTurnInput(sessionId) && + !hasWaitingTurnInput && !this.ports.pendingInputs.hasBlockingInput(sessionId) && !this.ports.pendingInputs.hasClaimedInput(sessionId) ) @@ -353,6 +409,9 @@ export class PendingInputPump { const nextQueuedInput = nextSteerInput ? null : this.ports.pendingInputs.getNextQueuedInput(sessionId) + if (nextQueuedInput && this.restartHeldQueueInputIds.has(nextQueuedInput.id)) { + return false + } const nextPendingInput = nextSteerInput ?? nextQueuedInput if (!nextPendingInput) { return false @@ -436,7 +495,9 @@ export class PendingInputPump { try { turn = this.ports.turnStarter.start(claimedInput.sessionId, claimedInput.payload, { projectDir, - claimedInput: claim + claimedInput: claim, + consumeClaimBeforeProviderStream: + claim.source === 'queue' && this.manuallyResumedQueueInputIds.has(claimedInput.id) }) } catch (error) { turn = Promise.reject(error) @@ -467,6 +528,9 @@ export class PendingInputPump { new Error(`Turn left pending input ${claimedInput.id} unsettled.`) ) } + if (claim.disposition?.kind === 'consume') { + this.manuallyResumedQueueInputIds.delete(claimedInput.id) + } this.clearLaunchedDrain(claimedInput.sessionId, scope.instance, drainLease) scope.instance.releasePendingQueueDrain(drainLease) this.flushDeferredWakeup(claimedInput.sessionId) @@ -534,9 +598,14 @@ export class PendingInputPump { private rememberDeferredWakeup(sessionId: string, reason: PendingInputWakeReason): void { const current = this.deferredWakeups.get(sessionId) + const priority: Record = { + completed: 0, + enqueue: 1, + manual: 2 + } this.deferredWakeups.set( sessionId, - current === 'enqueue' || reason === 'enqueue' ? 'enqueue' : 'completed' + !current || priority[reason] > priority[current] ? reason : current ) } @@ -643,7 +712,15 @@ export class PendingInputPump { status: DeepChatSessionState['status'], reason: PendingInputWakeReason ): boolean { - return status === 'idle' || (reason === 'enqueue' && status === 'error') + return status === 'idle' || (reason !== 'completed' && status === 'error') + } + + private hasExecutablePendingTurnInput(sessionId: string): boolean { + if (this.ports.pendingInputs.getNextSteerInput(sessionId)) { + return true + } + const nextQueueInput = this.ports.pendingInputs.getNextQueuedInput(sessionId) + return Boolean(nextQueueInput && !this.restartHeldQueueInputIds.has(nextQueueInput.id)) } private logDrainError( diff --git a/src/main/agent/deepchat/runtime/runLifecycleCoordinator.ts b/src/main/agent/deepchat/runtime/runLifecycleCoordinator.ts index 9c5d6fdb6b..ef6b4bfad8 100644 --- a/src/main/agent/deepchat/runtime/runLifecycleCoordinator.ts +++ b/src/main/agent/deepchat/runtime/runLifecycleCoordinator.ts @@ -29,7 +29,7 @@ import type { MessageProjectionService } from './messageProjectionService' import { resolveStreamRequestId as resolveRegistryStreamRequestId } from './streamRequestId' import type { ProcessResult } from './types' -export type PendingInputWakeReason = 'enqueue' | 'completed' +export type PendingInputWakeReason = 'enqueue' | 'completed' | 'manual' type RunLifecycleTranscript = Pick< SessionTranscript, diff --git a/src/main/agent/deepchat/runtime/transcriptMutationCoordinator.ts b/src/main/agent/deepchat/runtime/transcriptMutationCoordinator.ts index a8e0c35457..8dd99931fb 100644 --- a/src/main/agent/deepchat/runtime/transcriptMutationCoordinator.ts +++ b/src/main/agent/deepchat/runtime/transcriptMutationCoordinator.ts @@ -58,7 +58,10 @@ export class TranscriptMutationCoordinator { ) } - async prepareRetry(sessionId: string): Promise<{ projectDir: string | null }> { + async prepareRetry( + sessionId: string, + options?: { allowRestartHeldQueue?: boolean } + ): Promise<{ projectDir: string | null }> { const instance = this.deps.registry.getOrHydrateScope(toAppSessionId(sessionId)).instance const state = await this.deps.sessionState.get(sessionId) if (!state) { @@ -71,7 +74,7 @@ export class TranscriptMutationCoordinator { if (this.deps.runLifecycle.hasPendingInteractions(sessionId)) { throw new Error('Please resolve pending tool interactions before retrying.') } - this.assertNoActivePendingInputs(sessionId) + this.deps.admission.assertNoActiveInputs(sessionId, options) this.deps.runLifecycle.assertCurrentInstance(sessionId, instance) return { projectDir: this.deps.sessionSettings.resolveProjectDir(sessionId, undefined, instance) diff --git a/src/main/agent/deepchat/runtime/turnCoordinator.ts b/src/main/agent/deepchat/runtime/turnCoordinator.ts index 37054b63ad..193d844f85 100644 --- a/src/main/agent/deepchat/runtime/turnCoordinator.ts +++ b/src/main/agent/deepchat/runtime/turnCoordinator.ts @@ -88,6 +88,7 @@ import type { } from './pendingInputContracts' import { createDeepSeekResponsesReplayProjector } from '@/provider/deepseekResponsesAdapter' import type { CommandShellService } from '@/agent/shared/process/commandShellService' +import type { SessionPendingInputs } from '@/session/data/pendingInputs' type TurnRunLifecyclePort = Pick< RunLifecycleCoordinator, @@ -119,6 +120,7 @@ export interface TurnStartContext { export interface TurnExecutionContext extends TurnStartContext { claimedInput?: ClaimedPendingInputHandle + consumeClaimBeforeProviderStream?: boolean } export interface TurnCoordinatorPorts { @@ -128,6 +130,7 @@ export interface TurnCoordinatorPorts { toolService: Pick sessionStore: SessionSettingsStore messageStore: SessionTranscript + pendingInputs: Pick tapeReconciliation: TapeReconciliationPort toolResolver: DeepChatToolResolver compactionService: CompactionService @@ -657,11 +660,17 @@ export class TurnCoordinator { sessionId, 'user-message-create', () => - this.ports.messageStore.createUserMessage( - sessionId, - this.ports.messageStore.getNextOrderSeq(sessionId), - userContent - ) + claimedInput && claimedInput.source !== 'steer' + ? this.ports.pendingInputs.createClaimedQueueUserMessage( + sessionId, + claimedInput.id, + userContent + ) + : this.ports.messageStore.createUserMessage( + sessionId, + this.ports.messageStore.getNextOrderSeq(sessionId), + userContent + ) ) userMessageId = createdUserMessageId instance.setPreStreamTranscriptAnchorId(createdUserMessageId) @@ -862,7 +871,16 @@ export class TurnCoordinator { contextBuilderVersion: contextBuild.assemblerVersion, syntheticContributions: contextBuild.metadata.syntheticContributions }, - onBeforeProviderStream: providerBoundary.complete, + onBeforeProviderStream: () => { + if ( + context?.consumeClaimBeforeProviderStream && + claimedInput && + !claimedInput.disposition + ) { + claimedInput.settle({ kind: 'consume' }) + } + providerBoundary.complete() + }, onRunRegistered: (runId) => { streamRunId = runId } diff --git a/src/main/agent/manager/deepChatAgentBackend.ts b/src/main/agent/manager/deepChatAgentBackend.ts index 7c60ea0d70..1b5bba8ec1 100644 --- a/src/main/agent/manager/deepChatAgentBackend.ts +++ b/src/main/agent/manager/deepChatAgentBackend.ts @@ -52,6 +52,8 @@ export interface DeepChatAgentBackendPort { options?: { signal?: AbortSignal } ): Promise listPendingInputs(sessionId: AppSessionId): Promise + isPendingQueueResumeAvailable(sessionId: AppSessionId): Promise + resumePendingQueue(sessionId: AppSessionId): Promise queuePendingInput( sessionId: AppSessionId, content: SendMessageInput, @@ -197,7 +199,9 @@ export function createDeepChatAgentBackend( setSessionAgentContext: (config) => port.setSessionAgentContext(sessionId, config), setModel: (providerId, modelId) => port.setSessionModel(sessionId, providerId, modelId), getCompactionState: () => port.getSessionCompactionState(sessionId), - compact: () => port.compactSession(sessionId) + compact: () => port.compactSession(sessionId), + isPendingQueueResumeAvailable: () => port.isPendingQueueResumeAvailable(sessionId), + resumePendingQueue: () => port.resumePendingQueue(sessionId) } } handles.set(sessionId, handle) diff --git a/src/main/agent/manager/sessionHandles.ts b/src/main/agent/manager/sessionHandles.ts index 2a7242829d..696e19930c 100644 --- a/src/main/agent/manager/sessionHandles.ts +++ b/src/main/agent/manager/sessionHandles.ts @@ -83,6 +83,8 @@ export interface DeepChatControlFacet { setModel(providerId: string, modelId: string): Promise getCompactionState(): Promise compact(): Promise<{ compacted: boolean; state: SessionCompactionState }> + isPendingQueueResumeAvailable(): Promise + resumePendingQueue(): Promise } export interface DeepChatSessionHandle extends AgentSessionHandle { diff --git a/src/main/app/composition.ts b/src/main/app/composition.ts index faa407c986..fe1816099e 100644 --- a/src/main/app/composition.ts +++ b/src/main/app/composition.ts @@ -1598,7 +1598,9 @@ export async function createMainProcessControl(dependencies: { compaction: { getState: () => handle.deepchat.getCompactionState(), compact: () => handle.deepchat.compact() - } + }, + isPendingQueueResumeAvailable: () => handle.deepchat.isPendingQueueResumeAvailable(), + resumePendingQueue: () => handle.deepchat.resumePendingQueue() } : { ...turn, kind: handle.kind } } diff --git a/src/main/session/contracts.ts b/src/main/session/contracts.ts index a2db8b591d..a8ef0c71ad 100644 --- a/src/main/session/contracts.ts +++ b/src/main/session/contracts.ts @@ -241,6 +241,8 @@ export type SessionTurnRuntimeSession = getState(): Promise compact(): Promise<{ compacted: boolean; state: SessionCompactionState }> } + isPendingQueueResumeAvailable(): Promise + resumePendingQueue(): Promise }) | (SessionTurnRuntimeBase & { readonly kind: 'acp' }) @@ -289,6 +291,8 @@ export interface SessionTurnPort { options?: { signal?: AbortSignal } ): Promise listPendingInputs(sessionId: string): Promise + isPendingQueueResumeAvailable(sessionId: string): Promise + resumePendingQueue(sessionId: string): Promise queuePendingInput( sessionId: string, content: string | SendMessageInput diff --git a/src/main/session/data/pendingInputStore.ts b/src/main/session/data/pendingInputStore.ts index 7e2faa82e1..f372d9dcfc 100644 --- a/src/main/session/data/pendingInputStore.ts +++ b/src/main/session/data/pendingInputStore.ts @@ -338,7 +338,17 @@ export class SessionPendingInputStore { if (row.mode !== 'queue') { throw new Error(`Pending input ${itemId} is not a queue item.`) } - return this.releaseClaimedInput(itemId, row) + if (row.state !== 'claimed') { + return this.toRecord(row) + } + + this.database.deepchatPendingInputsTable.update(itemId, { + state: 'pending', + claimed_at: null, + blocking_json: null, + message_ids_json: '[]' + }) + return this.toRecord(this.requireRow(itemId, row.session_id)) } releaseClaimedInput( diff --git a/src/main/session/data/pendingInputs.ts b/src/main/session/data/pendingInputs.ts index 125138a5ab..6eb038f202 100644 --- a/src/main/session/data/pendingInputs.ts +++ b/src/main/session/data/pendingInputs.ts @@ -11,6 +11,11 @@ import type { SessionTranscript } from './transcript' const MAX_ACTIVE_PENDING_INPUTS = 5 +export interface PendingInputRestartRecovery { + affectedSessionIds: Set + heldQueueInputIds: Set +} + function toUserMessageContent(input: SendMessageInput): UserMessageContent { return { text: input.text, @@ -48,6 +53,35 @@ export class SessionPendingInputs { return record } + createClaimedQueueUserMessage( + sessionId: string, + itemId: string, + content: UserMessageContent + ): string { + return this.store.runInTransaction(() => { + const input = this.store.getInput(itemId) + if ( + !input || + input.sessionId !== sessionId || + input.mode !== 'queue' || + input.state !== 'claimed' + ) { + throw new Error(`Pending input ${itemId} is not a claimed queue item for ${sessionId}.`) + } + if (input.messageIds.length > 0) { + throw new Error(`Claimed queue item ${itemId} already has a linked message.`) + } + + const messageId = this.transcript.createUserMessage( + sessionId, + this.transcript.getNextOrderSeq(sessionId), + content + ) + this.store.linkClaimedQueueMessage(itemId, messageId) + return messageId + }) + } + queuePendingInput( sessionId: string, input: SendMessageInput, @@ -333,63 +367,80 @@ export class SessionPendingInputs { this.events.publishMessagesChanged(sessionId, changedMessages) } - recoverClaimedInputsAfterRestart(): number { - const sessionIds = new Set() + recoverInputsAfterRestart(): PendingInputRestartRecovery { + const affectedSessionIds = new Set() + const heldQueueInputIds = new Set() this.store.runInTransaction(() => { for (const input of this.store.listActiveInputs()) { if (input.mode === 'queue') { if (input.state === 'claimed') { - if (input.messageIds.length > 0) { + const hasMaterializedUserFact = input.messageIds.some((messageId) => { + const message = this.transcript.getMessage(messageId) + return message?.sessionId === input.sessionId && message.role === 'user' + }) + if (hasMaterializedUserFact) { this.store.consumeQueueInput(input.id) } else { this.store.releaseClaimedQueueInput(input.id) + heldQueueInputIds.add(input.id) } - sessionIds.add(input.sessionId) + affectedSessionIds.add(input.sessionId) + } else { + heldQueueInputIds.add(input.id) + affectedSessionIds.add(input.sessionId) } continue } if (input.state === 'blocked') { this.store.convertSteerInputToQueue(input.id) - sessionIds.add(input.sessionId) + heldQueueInputIds.add(input.id) + affectedSessionIds.add(input.sessionId) continue } if (input.state === 'claimed' && input.messageIds.length > 0) { this.transcript.settleSteerMessages(input.messageIds) this.store.consumeSteerInput(input.id) - sessionIds.add(input.sessionId) + affectedSessionIds.add(input.sessionId) continue } if (input.state === 'claimed') { this.store.releaseClaimedInput(input.id) } - if (input.messageIds.length > 0) { - continue - } - const messageId = this.transcript.createUserMessage( - input.sessionId, - this.transcript.getNextOrderSeq(input.sessionId), - toUserMessageContent(input.payload), - { - status: 'pending', - metadata: { - inputReceipt: { - mode: 'steer', - readAt: null + const messageIds = [...input.messageIds] + if (messageIds.length === 0) { + messageIds.push( + this.transcript.createUserMessage( + input.sessionId, + this.transcript.getNextOrderSeq(input.sessionId), + toUserMessageContent(input.payload), + { + status: 'pending', + metadata: { + inputReceipt: { + mode: 'steer', + readAt: null + } + } } - } - } - ) - this.store.linkSteerMessage(input.id, messageId) - sessionIds.add(input.sessionId) + ) + ) + this.store.linkSteerMessage(input.id, messageIds[0]) + } + this.transcript.failPendingSteerMessages(messageIds) + this.store.consumeSteerInput(input.id) + affectedSessionIds.add(input.sessionId) } }) - for (const sessionId of sessionIds) { + for (const sessionId of affectedSessionIds) { this.emitUpdated(sessionId) } - return sessionIds.size + return { + affectedSessionIds, + heldQueueInputIds + } } hasActiveInputs(sessionId: string): boolean { diff --git a/src/main/session/data/transcript.ts b/src/main/session/data/transcript.ts index f4d1fa63fe..d57d970491 100644 --- a/src/main/session/data/transcript.ts +++ b/src/main/session/data/transcript.ts @@ -363,6 +363,26 @@ export class SessionTranscript { return messageIds.map((messageId) => this.requireMessage(messageId)) } + failPendingSteerMessages(messageIds: string[]): ChatMessageRecord[] { + for (const messageId of messageIds) { + const message = this.getMessage(messageId) + if (!message || message.role !== 'user' || message.status !== 'pending') { + throw new Error(`Pending steer message not found: ${messageId}`) + } + const metadata = parseMessageMetadata(message.metadata) + if (metadata.inputReceipt?.mode !== 'steer' || metadata.inputReceipt.readAt !== null) { + throw new Error(`Message ${messageId} is not an unread steer message.`) + } + this.database.deepchatMessagesTable.updateStatus(messageId, 'error') + const updated = this.requireMessage(messageId) + this.tapeFacts.appendMessageReplacement(updated, { + reason: 'steer_message_restart_failed', + revisionKind: 'record' + }) + } + return messageIds.map((messageId) => this.requireMessage(messageId)) + } + finalizeAssistantMessage( messageId: string, blocks: AssistantMessageBlock[], diff --git a/src/main/session/routes.ts b/src/main/session/routes.ts index d29c11656b..bbcafe8e0d 100644 --- a/src/main/session/routes.ts +++ b/src/main/session/routes.ts @@ -41,6 +41,7 @@ import { sessionsMoveToAgentRoute, sessionsQueuePendingInputRoute, sessionsRenameRoute, + sessionsResumePendingQueueRoute, sessionsRestoreRoute, sessionsResolveBlockedPendingInputRoute, sessionsRetryMessageRoute, @@ -245,8 +246,22 @@ export function createSessionRoutes(deps: { sessionsListPendingInputsRoute.name, async (rawInput) => { const input = sessionsListPendingInputsRoute.input.parse(rawInput) + const [items, resumeAvailable] = await Promise.all([ + deps.turn.listPendingInputs(input.sessionId), + deps.turn.isPendingQueueResumeAvailable(input.sessionId) + ]) return sessionsListPendingInputsRoute.output.parse({ - items: await deps.turn.listPendingInputs(input.sessionId) + items, + resumeAvailable + }) + } + ], + [ + sessionsResumePendingQueueRoute.name, + async (rawInput) => { + const input = sessionsResumePendingQueueRoute.input.parse(rawInput) + return sessionsResumePendingQueueRoute.output.parse({ + started: await deps.turn.resumePendingQueue(input.sessionId) }) } ], diff --git a/src/main/session/transcriptMutations.ts b/src/main/session/transcriptMutations.ts index bc924ea0a1..73d5b1532a 100644 --- a/src/main/session/transcriptMutations.ts +++ b/src/main/session/transcriptMutations.ts @@ -3,11 +3,15 @@ import type { SessionPendingInputs } from './data/pendingInputs' import type { SessionSettingsStore } from './data/settings' import type { SessionTranscript } from './data/transcript' import { buildEditedUserContent, extractUserMessageInput } from './data/userMessageContent' +import { parseMessageMetadata } from './usageStats' export interface SessionTranscriptRuntimePort { prepareClearMessages(sessionId: string): Promise finishClearMessages(sessionId: string): void - prepareRetry(sessionId: string): Promise<{ projectDir: string | null }> + prepareRetry( + sessionId: string, + options?: { allowRestartHeldQueue?: boolean } + ): Promise<{ projectDir: string | null }> assertNoActivePendingInputs(sessionId: string): void cancelForTranscriptMutation(sessionId: string): Promise invalidateTranscriptFrom(sessionId: string, orderSeq: number): void @@ -40,13 +44,18 @@ export class SessionTranscriptMutations { sessionId: string, messageId: string ): Promise<{ content: SendMessageInput; projectDir: string | null; sourceOrderSeq: number }> { - const { projectDir } = await this.dependencies.runtime.prepareRetry(sessionId) const target = this.requireMessage(sessionId, messageId) const sourceUserMessage = target.role === 'user' ? target : this.dependencies.transcript.getLastUserMessageBeforeOrAt(sessionId, target.orderSeq) if (!sourceUserMessage) throw new Error('No user message found for retry.') + const sourceMetadata = parseMessageMetadata(sourceUserMessage.metadata) + const allowRestartHeldQueue = + target.status === 'error' && sourceMetadata.inputReceipt?.mode === 'steer' + const { projectDir } = await this.dependencies.runtime.prepareRetry(sessionId, { + allowRestartHeldQueue + }) const content = extractUserMessageInput(sourceUserMessage.content) if (!content.text.trim() && (content.files?.length ?? 0) === 0) { diff --git a/src/main/session/turn.ts b/src/main/session/turn.ts index 6648fb15b0..07fb48da9f 100644 --- a/src/main/session/turn.ts +++ b/src/main/session/turn.ts @@ -210,6 +210,23 @@ export class SessionTurn implements SessionTurnPort, SessionInitialTurnPort { return await this.dependencies.runtime.resolveSession(toAppSessionId(sessionId)).pending.list() } + async isPendingQueueResumeAvailable(sessionId: string): Promise { + if (!this.dependencies.sessions.get(sessionId)) return false + const runtime = this.dependencies.runtime.resolveSession(toAppSessionId(sessionId)) + return runtime.kind === 'deepchat' && (await runtime.isPendingQueueResumeAvailable()) + } + + async resumePendingQueue(sessionId: string): Promise { + 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 resume is only available for DeepChat sessions.') + } + return await runtime.resumePendingQueue() + }) + } + async queuePendingInput( sessionId: string, content: string | SendMessageInput diff --git a/src/renderer/api/SessionClient.ts b/src/renderer/api/SessionClient.ts index c2cf8bd658..7b2aeaf81a 100644 --- a/src/renderer/api/SessionClient.ts +++ b/src/renderer/api/SessionClient.ts @@ -48,6 +48,7 @@ import { sessionsQueuePendingInputRoute, sessionsRenameRoute, sessionsResolveBlockedPendingInputRoute, + sessionsResumePendingQueueRoute, sessionsRetryRtkHealthCheckRoute, sessionsRetryMessageRoute, sessionsRestoreRoute @@ -154,8 +155,11 @@ export function createSessionClient(bridge: DeepchatBridge = getDeepchatBridge() } async function listPendingInputs(sessionId: string) { - const result = await bridge.invoke(sessionsListPendingInputsRoute.name, { sessionId }) - return result.items + return await bridge.invoke(sessionsListPendingInputsRoute.name, { sessionId }) + } + + async function resumePendingQueue(sessionId: string) { + return await bridge.invoke(sessionsResumePendingQueueRoute.name, { sessionId }) } async function queuePendingInput(sessionId: string, content: string | SendMessageInput) { @@ -586,6 +590,7 @@ export function createSessionClient(bridge: DeepchatBridge = getDeepchatBridge() getLightweightByIds, ensureAcpDraftSession, listPendingInputs, + resumePendingQueue, queuePendingInput, updateQueuedInput, moveQueuedInput, diff --git a/src/renderer/src/components/chat/PendingInputLane.vue b/src/renderer/src/components/chat/PendingInputLane.vue index c24a08b75e..0e5587974e 100644 --- a/src/renderer/src/components/chat/PendingInputLane.vue +++ b/src/renderer/src/components/chat/PendingInputLane.vue @@ -22,6 +22,22 @@ {{ t('chat.attachments.pending.blockedCount', { count: blockedCount }) }} + + + {{ t('chat.pendingInput.resume') }} +
(), { activeLimit: 5, disableSteerAction: false, - disableQueueSteerAction: false + disableQueueSteerAction: false, + showResumeAction: false, + resumeDisabled: false, + resumeLoading: false } ) @@ -250,6 +272,7 @@ const emit = defineEmits<{ 'move-queue': [payload: { itemId: string; toIndex: number }] 'steer-queue': [itemId: string] 'delete-queue': [itemId: string] + 'resume-queue': [] 'resolve-blocked': [payload: { itemId: string; action: 'retry' | 'send_without_image_content' }] }>() const { t } = useI18n() diff --git a/src/renderer/src/components/message/MessageItemUser.vue b/src/renderer/src/components/message/MessageItemUser.vue index a30fac02cf..63fe184ae4 100644 --- a/src/renderer/src/components/message/MessageItemUser.vue +++ b/src/renderer/src/components/message/MessageItemUser.vue @@ -396,8 +396,8 @@ watch( ) watch( - () => [props.message.id, props.message.inputReceipt?.readAt] as const, - ([, readAt]) => { + () => [props.message.id, props.message.status, props.message.inputReceipt?.readAt] as const, + ([, status, readAt]) => { if (receiptTimer) { clearTimeout(receiptTimer) receiptTimer = null @@ -406,6 +406,10 @@ watch( receipt.value = null return } + if (status === 'error') { + receipt.value = null + return + } if (readAt === null || readAt === undefined) { receipt.value = 'unread' return diff --git a/src/renderer/src/features/chat-page/ChatPage.vue b/src/renderer/src/features/chat-page/ChatPage.vue index fc567cd9aa..c17362761a 100644 --- a/src/renderer/src/features/chat-page/ChatPage.vue +++ b/src/renderer/src/features/chat-page/ChatPage.vue @@ -147,11 +147,15 @@ :queue-items="pendingInputStore.queueItems" :disable-steer-action="pendingInputStore.isAtCapacity" :disable-queue-steer-action="disableQueueSteerAction" + :show-resume-action="showPendingQueueResume" + :resume-disabled="disablePendingQueueResume" + :resume-loading="pendingInputStore.resumingQueue" class="mx-auto mb-1.5 max-w-4xl" @update-queue="onPendingInputUpdate" @move-queue="onPendingInputMove" @steer-queue="onPendingInputSteer" @delete-queue="onPendingInputDelete" + @resume-queue="onPendingInputResume" @resolve-blocked="onPendingInputResolve" />