diff --git a/FEATURES.md b/FEATURES.md index d562e48..6344314 100644 --- a/FEATURES.md +++ b/FEATURES.md @@ -57,7 +57,7 @@ ### 工作管理、长期协作与工作流 - [x] **任务、活动与成果**：集中管理任务状态、审计活动和独立成果文件；普通聊天回复只保留在会话中，不再自动复制到成果栏，已有重复聊天 Markdown 从成果列表隐藏但不物理删除。Token 用量按 Runtime 与模型归类，并针对 OpenAI 兼容与 Anthropic Messages 的不同上报口径归一化展示缓存命中率；活动按会话分组并默认收起，避免长历史占满页面。 -- [x] **Task 与定制任务体验**：每个产品级 Task 只关联一条 Conversation，一条 Conversation 可承载多个 Task；左侧会话列表通过行首展开按钮显示带任务图标的 Task 子项，父会话行不重复任务标签，UI 只展示到 Task，不暴露 Job/Run 层级。新建定制任务可关联当前或新 Conversation，默认 Execute 并沿用 Runtime、工具和审批边界；重复触发复用同一 Task，文本结果回写 Conversation，独立文件和图片保留为成果。Task Center 继续作为完整索引，不建设独立 Automation Center。当前计划触发支持单次、每日和每周；高级时区、Cron、事件触发与重试治理仍按 PRD 逐步实现。详见 [Task Center PRD](./docs/prd/task-and-job/task-center-prd.md) 和 [Scheduled Task PRD](./docs/prd/task-and-job/scheduled-task-prd.md)。 +- [x] **Task 与定制任务体验**：每个产品级 Task 只关联一条 Conversation，一条 Conversation 可承载多个 Task；左侧会话列表通过行首展开按钮显示带共享状态点的 Task 子项，父会话行不重复任务标签，UI 只展示到 Task，不暴露 Job/Run 层级。新建定制任务可关联当前或新 Conversation，默认 Execute 并沿用 Runtime、工具和审批边界；重复触发复用同一 Task，文本结果回写 Conversation，独立文件和图片保留为成果。普通消息与到期 Scheduled Task 共用 Conversation 级持久发送队列，同一会话一次只执行一项；当前回复期间仍可继续发送，队列按顺序续跑，并允许删除或“立即中断并插入”。Task Center 继续作为完整索引，不建设独立 Automation Center。当前计划触发支持单次、每日和每周；高级时区、Cron、事件触发与重试治理仍按 PRD 逐步实现。详见 [Task Center PRD](./docs/prd/task-and-job/task-center-prd.md) 和 [Scheduled Task PRD](./docs/prd/task-and-job/scheduled-task-prd.md)。 - [x] **记忆与智能心跳**：当前提供周期回顾、建议记忆、洞察、后续任务和可审计运行轨迹。 - [x] **智能心跳入口与范围改善**：将“智能心跳 > 心跳计划”作为完整配置的唯一权威入口，支持创建和编辑 Global 或指定一个、多个 Project 的计划；旧单项目配置无损迁移，项目级记忆与行动输出必须显式指定范围内的 Project。Task Center 和设置不再复制心跳表单。“未来分区记忆”仍只是尚待独立设计的长期方向。详见 [智能心跳 PRD](./docs/prd/smart-heartbeat/smart-heartbeat-prd.md)。 - [ ] **通用监督**（规划中）：通过固定监督栏目观察用户选择的会话、任务、自动化或实验对象，提供带证据的评论与人工介入请求，但不自动发言、批准工具或切换 Execute。详见 [会话监督 PRD](./docs/prd/supervision/conversation-supervision-prd.md)。 diff --git a/UI-DESIGN.md b/UI-DESIGN.md index 9d7604b..1b7d616 100644 --- a/UI-DESIGN.md +++ b/UI-DESIGN.md @@ -283,14 +283,15 @@ - 当前导航项和当前会话必须同时使用至少三种信号中的两种：强调背景、可见边框、图标或文字强调。浅色主题的当前项优先使用更完整的蓝色选中表面和较高字重。 - 未选中项保持平整，不为每一行添加卡片边框或阴影。悬停反馈不得强于选中状态。 - 关联 Task 的 Conversation 在行最左侧显示独立展开按钮，父会话行不重复任务标签或数量； - 展开后的每个 Task 子项使用任务图标标识身份，展开层级只到 Task，不展示 Job、Subjob 或 Run。 + 展开后的每个 Task 子项使用共享状态点，展开层级只到 Task，不展示 Job、Subjob 或 Run。 - 会话标题保持单行且不挤压时间或操作按钮。标题实际溢出时，悬停会话行才在标题的固定裁切 区域内平滑横向滑动以展示完整名称；移开后快速复位。未溢出标题不产生位移动效， `prefers-reduced-motion` 下禁用滑动并保留完整标题提示。 - 新建 Task 后可以首次自动展开所属 Conversation；用户手动折叠后保持其选择，后台状态变化 只更新状态提示，不强制展开或抢占焦点。 -- Task 子项的任务图标保持稳定，运行中、等待审批、失败和暂停同时使用本地化状态文字， - 不能只靠图标颜色表达状态。 +- Task 子项复用任务中心的共享状态点：运行中使用脉冲强调色，已完成使用静态成功色， + 失败或中断使用危险色，等待审批使用警告色，暂停或取消使用禁用色；旁边同时显示 + 本地化状态文字，不能只靠颜色或动效表达状态。 - 账户与设置入口固定在侧栏底部。已有稳定设置入口时，不在顶栏重复提供同一入口。 ### 6.9 助手工作栏 @@ -505,6 +506,9 @@ GoodBuddy 是可调整窗口大小的桌面应用。响应式设计优先保证 - 设置中心只管理持久 Runtime 配置、默认值和能力清单；右侧 Runtime 栏目管理用户当前跟随或固定目标的生命周期。两处不得复制同一实时操作，栏目中的高风险操作仍须就地确认并保留取消、权限、用量和活动审计。 - Runtime Prompt 快捷操作只把模板填入输入草稿，用户可以继续编辑；OpenCode Command 由 Runtime 原生 API 执行，输入框只承载可选参数，不以普通斜杠文本冒充执行。 - Agent 回复进行中锁定模型、专家角色、工作模式和 Runtime 定制选择器，并关闭已打开的上下文菜单；回复结束或停止后再恢复选择，避免界面状态与本次运行实际使用的上下文不一致。 +- Agent 回复进行中不能禁用普通消息发送。新消息与到期 Scheduled Task 共用 Conversation 级待发送队列。队列位于 Composer 容器之外、与输入框等宽并保持独立间距；有待发送项时直接显示无标题栏的极简列表，空队列不保留占位。每项固定为一行，按进入顺序显示轻量来源图标和截断摘要；普通消息使用消息图标，Scheduled Task 统一使用淡出时钟图标。 +- 待发送项默认在当前回复结束后顺序执行。每项提供明确的“立即中断并插入”（空闲时为“立即运行”）和删除操作；前者先取消当前 Conversation 的活动执行，再将所选项提升为下一项，不允许同一 Conversation 并发写入时间线。 +- 队列区域向上增长并设置有界滚动高度，不提供展开、收起或冗余计数标题。宽容器中的操作采用弱化的行内样式，不渲染成强调卡片或胶囊；窄容器中保留来源、摘要和图标操作并隐藏冗长按钮文字。区域、来源和每个操作必须有可读无障碍名称。队列异步错误进入应用通知，不在 Composer 内复制通知样式。 - 支持上下文状态的 Runtime 在输入区下方复用同一紧凑用量条；文案必须区分“本次模型调用”和“压缩后对话估算”。手动压缩仅在当前 Runtime 明确支持且没有活动回复时显示，作为元信息区左下角的浮动次操作，不参与输入区高度计算；元信息区始终预留稳定高度，切换 Runtime 不得让输入框上下位移。元信息区与窗口底部只保留紧凑安全留白，不形成额外空白区。进行中禁用重复操作，结果通过应用通知反馈。 - 已选择的工作模式在触发按钮中只显示 `Ask` 或 `Execute`；完整中文含义和说明保留在菜单选项、可访问名称及输入区下方的模式说明中。 - 宽度大于 `700px` 时，通用工具栏内的添加内容、知识范围、专家、模式和 Runtime 选择保持同一行，Runtime 专属功能在自己的下一行横向排列。窄输入区中两行分别换行，专属选择器以至少 `220px` 的基准宽度换行而不是被挤压；不能把专属控件重新塞回通用工具栏。 @@ -521,7 +525,7 @@ GoodBuddy 是可调整窗口大小的桌面应用。响应式设计优先保证 - 侧栏最近会话的更新时间在当天显示本地时间，非当天显示本地日期；非当年记录必须同时显示年份。悬停时间信息时提供完整日期和时间。 - 侧栏中有关联 Task 的 Conversation 显示行首展开按钮，父行不显示任务标签或数量；展开后 最多直接显示 3 个 Task，“查看全部 N 个任务”打开该 Conversation 的完整任务区。 -- Task 子项以任务图标开头，并显示名称、本地化的 Ask/Execute、计划摘要和聚合状态。点击 +- Task 子项以共享状态点开头，并显示名称、本地化的 Ask/Execute、计划摘要和聚合状态。点击 子项打开同一 Conversation 并定位 Task，不继续展开 Job/Run。 - 删除入口使用 `danger-ghost`，并按数据可恢复性执行确认或撤销策略。 diff --git a/docs/architecture/automation-platform-architecture.md b/docs/architecture/automation-platform-architecture.md index 948d34b..40afbec 100644 --- a/docs/architecture/automation-platform-architecture.md +++ b/docs/architecture/automation-platform-architecture.md @@ -17,7 +17,9 @@ GoodBuddy 当前已经具备若干长期助手能力，但它们仍是彼此分 1. 当前 Schedule 已支持单次、每日和每周触发，并在创建时绑定稳定产品级 Task 与真实 Conversation；重复触发复用同一身份，文本结果写回 Conversation，独立文件和图片保存为 - Artifact。IANA 时区、Cron、事件触发、租约、重试和完整 Job/Run 抽象仍待实现。 + Artifact。到期执行与用户在回复期间继续发送的普通消息共用 Conversation 级持久队列， + 因而不会与当前回复并发写入同一时间线。IANA 时区、Cron、事件触发、租约、重试和完整 + Job/Run 抽象仍待实现。 2. 当前智能心跳支持全局或项目范围的每日、每周回顾，读取有界会话、任务和已确认记忆， 生成摘要、记忆建议和后续任务。 3. 专家执行的 Job 支持有限并发和只读综合，但没有实验变量、重复运行、统一指标和结果晋升。 @@ -283,7 +285,29 @@ Runtime 边界完成，用户可见结果通过所属 Task 汇入关联 Conversa 后台任务必须可被背压延后。延后记录为 `deferred`，不得丢失，也不得在系统恢复空闲时一次性 释放全部积压。 -### 7.3 幂等和租约 +### 7.3 Conversation 输入仲裁 + +当前实现以 Main 和 SQLite 中的 `conversation_queue_items` 作为每条 Conversation 的权威 +输入队列，而不是在 Renderer 分别维护聊天草稿队列和 Scheduled Task 队列： + +- 普通消息在发送时冻结 Runtime、工作模式、专家/团队、知识范围和附件上下文，再以 + `source=user` 入队；附件内容使用有界序列化保存，应用重启后仍可恢复。 +- 到期或手动启动的 Scheduled Task 先建立 `schedule_run`，再以 `source=schedule` 进入同一 + 队列。Scheduler 不再绕过队列直接调用 Runtime。 +- Main 对每条 Conversation 只保留一个活动请求或 Renderer 派发保留位。默认按 FIFO 认领； + 全局 Scheduled Task 执行仍受最多 4 项并发限制。 +- 用户消息由 Main 派发给 Renderer，由 Renderer 建立用户消息和流式助手消息后调用 + `agent.run`；Scheduled Task 由 Main 直接执行。两条路径共享同一 Conversation 活动锁。 +- 当前执行到达终态后再认领下一项。删除只移除尚未执行的项；“立即中断并插入”取消当前 + 请求并把所选项设为下一项，不重排其他项。 +- 每条 Conversation 最多保留 20 个用户可提交的待执行项。启动时将未完成的派发恢复为 + `pending`，但应用退出期间不会实际执行任务。 + +Renderer 只通过显式 IPC 列出、加入、删除、提升、释放和接收用户队列项；Main 在接受 +`agent.run` 时校验队列项仍处于 `dispatching` 且属于同一 Conversation，防止 Renderer +绕过顺序仲裁。 + +### 7.4 幂等和租约 - 每次计划触发使用 `planId + scheduledFor + planVersion` 形成幂等键。 - 手动触发使用调用方提供的单次幂等键。 diff --git a/docs/prd/task-and-job/scheduled-task-prd.md b/docs/prd/task-and-job/scheduled-task-prd.md index 44a6a8d..d74ecc2 100644 --- a/docs/prd/task-and-job/scheduled-task-prd.md +++ b/docs/prd/task-and-job/scheduled-task-prd.md @@ -4,8 +4,8 @@ | 项目 | 内容 | | --- | --- | -| 状态 | 首期稳定 Task 生命周期与创建体验已实现；高级触发和执行治理待实施 | -| 版本 | 0.6 | +| 状态 | 首期稳定 Task 生命周期、创建体验与 Conversation 输入仲裁已实现；高级触发和执行治理待实施 | +| 版本 | 0.7 | | 日期 | 2026-08-19 | | 依赖 | [Task 与 Job 统一领域模型](./task-and-job-model.md) | | 相关架构 | [自动化平台总体设计](../../architecture/automation-platform-architecture.md) | @@ -36,10 +36,15 @@ GoodBuddy 当前已实现首期统一生命周期： 为每次触发创建新的顶层 Task。 - 默认选择 Execute，并允许用户主动切换 Ask；不支持工具执行时明确禁用 Execute。 - 单次、每日和每周计划支持暂停、恢复、立即运行、应用重启恢复和最多 4 个独立计划并发。 +- 到期和手动运行先进入关联 Conversation 的持久输入队列，与回复期间继续发送的普通消息 + 顺序仲裁；默认不打断当前回复，也不与其并发写入时间线。 +- Composer 上沿显示待发送项和来源。用户可以删除尚未执行的 occurrence，或选择“立即 + 中断并插入”取消当前执行并将该项提升为下一项。 - 文本结果和失败写回关联 Conversation 并带 Task 来源；独立文件和图片继续保存为 Artifact。 - 左侧 Conversation 列表、Conversation Task 区和 Task Center 使用同一产品 Task；普通 模型请求、Subagent、委派和 Smart Heartbeat 内部 Task 不进入产品索引。 -- v22 迁移保留 Schedule 配置和历史运行，并为旧计划补齐稳定 Task 与 Conversation。 +- v22 迁移保留 Schedule 配置和历史运行，并为旧计划补齐稳定 Task 与 Conversation；v23 + 增加可恢复的统一 Conversation 输入队列。 尚未实现的高级能力包括 IANA 时区与 DST 墙上时间、每月/工作日/受限 Cron、事件触发、 可配置错过执行策略、租约、重试与结果未知治理、完整预算和权限快照，以及面向内部 @@ -222,6 +227,13 @@ type TimeTrigger = “立即运行”在当前 Task 内创建独立 Job 和 Run，不改变下一次计划时间，不创建新 Task。 重复点击使用调用级幂等键去重。 +### 7.4 与普通消息的顺序 + +同一 Conversation 的普通消息和 Scheduled Task occurrence 使用同一 FIFO 队列。Agent +正在回复时，到期 occurrence 只显示为待执行，不中断当前输出；当前执行结束后才认领下一项。 +用户显式选择“立即中断并插入”时，系统取消当前 Conversation 的活动请求，并让所选项成为 +下一项。删除待执行 occurrence 只取消该次运行，不删除稳定 Task、Conversation 或历史结果。 + ## 8. 一次触发的对象关系 ```text @@ -350,6 +362,9 @@ Task Center 显示 Scheduled Task 的范围、关联 Conversation、状态、模 - [ ] 夏令时不会造成漂移或双跑。 - [ ] 错过执行按配置跳过、补一次或有界补跑。 - [x] 手动运行不改变下次计划时间。 +- [x] Scheduled Task 与普通消息共用 Conversation 级队列，不并发写入同一时间线。 +- [x] 当前回复期间可以继续发送普通消息，并在 Composer 上沿查看、删除或提升待发送项。 +- [x] 应用重启恢复尚未执行的队列项和有界附件上下文。 - [x] 文本结果只写入 Conversation，独立交付物才进入成果。 - [ ] Task Center 和桌面通知可以打开正确 Conversation 并定位 Task。 - [ ] 应用重启不自动重放结果未知的副作用。 diff --git a/src/main/assistant/assistant-database.test.ts b/src/main/assistant/assistant-database.test.ts index 9ee4ee1..e0467fc 100644 --- a/src/main/assistant/assistant-database.test.ts +++ b/src/main/assistant/assistant-database.test.ts @@ -26,6 +26,64 @@ async function createDatabase(): Promise { return database } +function claimQueuedSchedules( + database: AssistantDatabase, + now: Date, + limit = 4 +): Array<{ + schedule: ReturnType[number] + runId: string +}> { + database.queueDueSchedules(now, limit) + const claims: Array<{ + schedule: ReturnType[number] + runId: string + }> = [] + const seenConversations = new Set() + for (const item of database.listConversationQueueItems()) { + if ( + item.source !== 'schedule' || + seenConversations.has(item.conversationId) || + claims.length >= limit + ) { + continue + } + const claimed = database.claimConversationQueueItem( + item.conversationId, + item.id + ) + if (claimed?.source === 'schedule') { + claims.push({ + schedule: claimed.schedule, + runId: claimed.runId + }) + seenConversations.add(item.conversationId) + } + } + return claims +} + +function claimManualScheduleQueueItem( + database: AssistantDatabase, + scheduleId: string +): { + schedule: ReturnType[number] + runId: string +} { + const item = database.queueScheduleNow(scheduleId) + const claimed = database.claimConversationQueueItem( + item.conversationId, + item.id + ) + if (claimed?.source !== 'schedule') { + throw new Error('Expected a claimed schedule queue item') + } + return { + schedule: claimed.schedule, + runId: claimed.runId + } +} + describe('AssistantDatabase', () => { it('rejects a newer unsupported schema without changing its version', async () => { const directory = await mkdtemp( @@ -98,7 +156,7 @@ describe('AssistantDatabase', () => { database.close() }) - it('migrates existing databases to schema version 22', async () => { + it('migrates existing databases to schema version 23', async () => { const directory = await mkdtemp( join(tmpdir(), 'goodbuddy-assistant-migration-') ) @@ -127,7 +185,7 @@ describe('AssistantDatabase', () => { user_version: number } ).user_version - ).toBe(22) + ).toBe(23) expect( current .prepare( @@ -198,6 +256,15 @@ describe('AssistantDatabase', () => { ) .get() ).toEqual({ name: 'magic_todos' }) + expect( + current + .prepare( + `SELECT name FROM sqlite_master + WHERE type = 'table' + AND name = 'conversation_queue_items'` + ) + .get() + ).toEqual({ name: 'conversation_queue_items' }) current.close() }) @@ -231,7 +298,7 @@ describe('AssistantDatabase', () => { user_version: number } ).user_version - ).toBe(22) + ).toBe(23) expect( current .prepare( @@ -369,7 +436,7 @@ describe('AssistantDatabase', () => { const inspected = new DatabaseSync(databasePath) expect( inspected.prepare('PRAGMA user_version').get() - ).toEqual({ user_version: 22 }) + ).toEqual({ user_version: 23 }) expect( inspected .prepare( @@ -1142,7 +1209,8 @@ describe('AssistantDatabase', () => { title: '每日摘要', messages: [] }) - const [claim] = database.claimDueSchedules( + const [claim] = claimQueuedSchedules( + database, new Date('2026-07-31T00:01:00.000Z') ) expect(claim?.schedule).toEqual( @@ -1169,12 +1237,17 @@ describe('AssistantDatabase', () => { status: 'idle', completedAt: undefined }) - const manualClaim = database.claimScheduleNow(schedule.id) + const manualClaim = claimManualScheduleQueueItem( + database, + schedule.id + ) expect(manualClaim.schedule).toMatchObject({ taskId: schedule.taskId, conversationId: schedule.conversationId }) - expect(() => database.claimScheduleNow(schedule.id)).toThrow( + expect(() => + claimManualScheduleQueueItem(database, schedule.id) + ).toThrow( '已有一次运行正在进行' ) database.completeScheduleRun( @@ -1254,7 +1327,8 @@ describe('AssistantDatabase', () => { recurrence: 'daily', nextRunAt: '2025-07-31T00:00:00.000Z' }) - const [overdueClaim] = database.claimDueSchedules( + const [overdueClaim] = claimQueuedSchedules( + database, new Date('2026-07-31T00:01:00.000Z') ) database.completeScheduleRun( @@ -1288,6 +1362,137 @@ describe('AssistantDatabase', () => { database.close() }) + it('persists and arbitrates a FIFO conversation input queue', async () => { + const directory = await mkdtemp( + join(tmpdir(), 'goodbuddy-conversation-queue-') + ) + temporaryDirectories.push(directory) + const databasePath = join(directory, 'assistant.sqlite') + const initial = new AssistantDatabase(databasePath) + initial.initialize('C:\\Workspace') + const project = initial.listProjects()[0]! + const conversationId = + '00000000-0000-4000-8000-000000000901' + initial.replaceConversations([ + { + id: conversationId, + projectId: project.id, + title: '排队对话', + updatedAt: Date.now(), + messages: [] + } + ]) + + const first = initial.enqueueConversationUserInput({ + conversationId, + label: '第一条消息', + payloadJson: JSON.stringify({ prompt: '第一条消息' }) + }) + const second = initial.enqueueConversationUserInput({ + conversationId, + label: '第二条消息', + payloadJson: JSON.stringify({ prompt: '第二条消息' }) + }) + expect( + initial + .listConversationQueueItems(conversationId) + .map((item) => item.id) + ).toEqual([first.id, second.id]) + + const preferred = initial.claimConversationQueueItem( + conversationId, + second.id + ) + expect(preferred).toMatchObject({ + source: 'user', + item: { id: second.id, source: 'user' }, + payloadJson: JSON.stringify({ prompt: '第二条消息' }) + }) + expect(initial.listConversationQueueItems(conversationId)).toEqual([ + expect.objectContaining({ id: first.id }) + ]) + initial.completeConversationUserQueueItem(second.id) + + const firstClaim = + initial.claimConversationQueueItem(conversationId) + expect(firstClaim).toMatchObject({ + source: 'user', + item: { id: first.id } + }) + initial.close() + + const recovered = new AssistantDatabase(databasePath) + recovered.initialize('C:\\Workspace') + expect( + recovered.listConversationQueueItems(conversationId) + ).toEqual([ + expect.objectContaining({ + id: first.id, + source: 'user', + label: '第一条消息' + }) + ]) + recovered.cancelConversationQueueItem(first.id) + expect( + recovered.listConversationQueueItems(conversationId) + ).toEqual([]) + recovered.close() + }) + + it('materializes due and manual schedule runs in the conversation queue', async () => { + const database = await createDatabase() + const schedule = database.createSchedule({ + title: '排队提醒', + prompt: '检查排队结果', + workMode: 'execute', + recurrence: 'daily', + nextRunAt: '2026-08-20T09:00:00.000Z' + }) + + const [dueItem] = database.queueDueSchedules( + new Date('2026-08-20T09:01:00.000Z') + ) + expect(dueItem).toMatchObject({ + conversationId: schedule.conversationId, + source: 'schedule', + scheduleId: schedule.id, + taskId: schedule.taskId + }) + expect(database.listConversationQueueItems()).toEqual([ + expect.objectContaining({ id: dueItem!.id }) + ]) + expect( + database.listPendingScheduleQueueConversationIds() + ).toEqual([schedule.conversationId]) + expect(database.listPendingConversationQueueIds()).toEqual([ + schedule.conversationId + ]) + + const claimed = database.claimConversationQueueItem( + schedule.conversationId + ) + expect(claimed).toMatchObject({ + source: 'schedule', + item: { id: dueItem!.id }, + schedule: { id: schedule.id }, + runId: dueItem!.id + }) + database.completeScheduleRun( + dueItem!.id, + 'completed', + new Date('2026-08-20T09:02:00.000Z') + ) + + const manualItem = database.queueScheduleNow(schedule.id) + expect(manualItem).toMatchObject({ + source: 'schedule', + scheduleId: schedule.id + }) + database.cancelConversationQueueItem(manualItem.id) + expect(database.listConversationQueueItems()).toEqual([]) + database.close() + }) + it('recovers a claimed schedule without swallowing its occurrence', async () => { const directory = await mkdtemp( join(tmpdir(), 'goodbuddy-schedule-recovery-') @@ -1303,7 +1508,8 @@ describe('AssistantDatabase', () => { recurrence: 'once', nextRunAt: '2026-08-13T00:00:00.000Z' }) - const [claimed] = initial.claimDueSchedules( + const [claimed] = claimQueuedSchedules( + initial, new Date('2026-08-13T00:01:00.000Z') ) expect(claimed?.schedule.id).toBe(schedule.id) @@ -1311,7 +1517,8 @@ describe('AssistantDatabase', () => { const recovered = new AssistantDatabase(databasePath) recovered.initialize('C:\\Workspace') - const [reclaimed] = recovered.claimDueSchedules( + const [reclaimed] = claimQueuedSchedules( + recovered, new Date('2026-08-13T00:02:00.000Z') ) expect(reclaimed).toMatchObject({ @@ -1350,7 +1557,8 @@ describe('AssistantDatabase', () => { }).id ) - const firstBatch = database.claimDueSchedules( + const firstBatch = claimQueuedSchedules( + database, new Date('2026-08-13T00:01:00.000Z'), 2 ) @@ -1359,7 +1567,8 @@ describe('AssistantDatabase', () => { new Set(firstBatch.map((claim) => claim.schedule.id)).size ).toBe(2) - const secondBatch = database.claimDueSchedules( + const secondBatch = claimQueuedSchedules( + database, new Date('2026-08-13T00:01:00.000Z'), 2 ) diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index f545881..317901f 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -16,6 +16,7 @@ import type { AssistantProject, AssistantSchedule, AssistantTask, + ConversationQueueItem, ConversationMessage, ConversationSnapshot, ExpertCreateInput, @@ -251,11 +252,32 @@ type ScheduleWithTaskRow = ScheduleRow & { conversation_id: string } -export type ClaimedSchedule = { - schedule: AssistantSchedule - runId: string +type ConversationQueueRow = { + id: string + conversation_id: string + source: ConversationQueueItem['source'] + label: string + payload_json: string + schedule_run_id: string | null + schedule_id: string | null + task_id: string | null + status: 'pending' | 'dispatching' + created_at: string } +export type ClaimedConversationQueueItem = + | { + source: 'user' + item: ConversationQueueItem & { source: 'user' } + payloadJson: string + } + | { + source: 'schedule' + item: ConversationQueueItem & { source: 'schedule' } + schedule: AssistantSchedule + runId: string + } + type ExpertRow = { id: string name: string @@ -554,6 +576,21 @@ function toSchedule(row: ScheduleWithTaskRow): AssistantSchedule { } } +function toConversationQueueItem( + row: ConversationQueueRow +): ConversationQueueItem { + return { + id: row.id, + conversationId: row.conversation_id, + source: row.source, + label: row.label, + createdAt: row.created_at, + scheduleRunId: row.schedule_run_id ?? undefined, + scheduleId: row.schedule_id ?? undefined, + taskId: row.task_id ?? undefined + } +} + function toExpert(row: ExpertRow): AssistantExpert { let routingKeywords: string[] let modelProfileId: string | undefined @@ -1012,6 +1049,24 @@ export class AssistantDatabase { WHERE status = 'running'` ) .run() + database + .prepare( + `UPDATE conversation_queue_items + SET status = 'pending' + WHERE status = 'dispatching'` + ) + .run() + database.exec(` + INSERT OR IGNORE INTO conversation_queue_items + (id, conversation_id, source, label, payload_json, + schedule_run_id, schedule_id, task_id, status, created_at) + SELECT sr.id, t.conversation_id, 'schedule', t.title, '{}', + sr.id, sr.schedule_id, t.id, 'pending', sr.scheduled_for + FROM schedule_runs sr + INNER JOIN tasks t ON t.schedule_id = sr.schedule_id + WHERE sr.status = 'pending' + AND t.conversation_id IS NOT NULL; + `) const interruptedTasks = database .prepare( `SELECT id, error @@ -1115,6 +1170,7 @@ export class AssistantDatabase { 'delegation_outbox', 'delegations', 'notifications', + 'conversation_queue_items', 'schedule_runs', 'schedules', 'memory_items', @@ -3517,6 +3573,317 @@ export class AssistantDatabase { } } + listConversationQueueItems( + conversationId?: string + ): ConversationQueueItem[] { + const database = this.requireDatabase() + const rows = conversationId + ? database + .prepare( + `SELECT * + FROM conversation_queue_items + WHERE conversation_id = ? AND status = 'pending' + ORDER BY created_at ASC, rowid ASC + LIMIT 100` + ) + .all(conversationId) + : database + .prepare( + `SELECT * + FROM conversation_queue_items + WHERE status = 'pending' + ORDER BY created_at ASC, rowid ASC + LIMIT 500` + ) + .all() + return (rows as ConversationQueueRow[]).map( + toConversationQueueItem + ) + } + + listPendingScheduleQueueConversationIds(limit = 4): string[] { + const safeLimit = Math.max(1, Math.min(20, Math.trunc(limit))) + const rows = this.requireDatabase() + .prepare( + `SELECT conversation_id + FROM conversation_queue_items + WHERE source = 'schedule' AND status = 'pending' + GROUP BY conversation_id + ORDER BY MIN(created_at) ASC, MIN(rowid) ASC + LIMIT ?` + ) + .all(safeLimit) as Array<{ conversation_id: string }> + return rows.map((row) => row.conversation_id) + } + + listPendingConversationQueueIds(): string[] { + const rows = this.requireDatabase() + .prepare( + `SELECT conversation_id + FROM conversation_queue_items + WHERE status = 'pending' + GROUP BY conversation_id + ORDER BY MIN(created_at) ASC, MIN(rowid) ASC` + ) + .all() as Array<{ conversation_id: string }> + return rows.map((row) => row.conversation_id) + } + + getConversationQueueItem( + itemId: string + ): ConversationQueueItem | undefined { + const row = this.requireDatabase() + .prepare( + `SELECT * + FROM conversation_queue_items + WHERE id = ?` + ) + .get(itemId) as ConversationQueueRow | undefined + return row ? toConversationQueueItem(row) : undefined + } + + getConversationUserQueuePayloadJson(itemId: string): string | undefined { + const row = this.requireDatabase() + .prepare( + `SELECT payload_json + FROM conversation_queue_items + WHERE id = ? AND source = 'user'` + ) + .get(itemId) as { payload_json: string } | undefined + return row?.payload_json + } + + isConversationUserQueueItemDispatching(itemId: string): boolean { + return Boolean( + this.requireDatabase() + .prepare( + `SELECT 1 + FROM conversation_queue_items + WHERE id = ? AND source = 'user' AND status = 'dispatching'` + ) + .get(itemId) + ) + } + + enqueueConversationUserInput(input: { + conversationId: string + label: string + payloadJson: string + }): ConversationQueueItem { + const database = this.requireDatabase() + const id = randomUUID() + const now = new Date().toISOString() + const label = input.label.trim().slice(0, 200) + if (!label) { + throw new Error('待发送消息标题不能为空') + } + database.exec('BEGIN IMMEDIATE') + try { + const conversation = database + .prepare( + `SELECT 1 + FROM conversations + WHERE id = ? AND status = 'active' AND channel IS NULL` + ) + .get(input.conversationId) + if (!conversation) { + throw new Error('对话不存在或不能加入发送队列') + } + const count = database + .prepare( + `SELECT COUNT(*) AS count + FROM conversation_queue_items + WHERE conversation_id = ? + AND status IN ('pending', 'dispatching')` + ) + .get(input.conversationId) as { count: number } + if (count.count >= 20) { + throw new Error('当前对话最多保留 20 条待执行项') + } + database + .prepare( + `INSERT INTO conversation_queue_items + (id, conversation_id, source, label, payload_json, + schedule_run_id, schedule_id, task_id, status, created_at) + VALUES (?, ?, 'user', ?, ?, NULL, NULL, NULL, 'pending', ?)` + ) + .run( + id, + input.conversationId, + label, + input.payloadJson, + now + ) + database.exec('COMMIT') + } catch (error) { + database.exec('ROLLBACK') + throw error + } + return this.getConversationQueueItem(id)! + } + + claimConversationQueueItem( + conversationId: string, + preferredItemId?: string + ): ClaimedConversationQueueItem | undefined { + const database = this.requireDatabase() + database.exec('BEGIN IMMEDIATE') + try { + const row = database + .prepare( + preferredItemId + ? `SELECT * + FROM conversation_queue_items + WHERE id = ? AND conversation_id = ? + AND status = 'pending'` + : `SELECT * + FROM conversation_queue_items + WHERE conversation_id = ? AND status = 'pending' + ORDER BY created_at ASC, rowid ASC + LIMIT 1` + ) + .get( + ...(preferredItemId + ? [preferredItemId, conversationId] + : [conversationId]) + ) as ConversationQueueRow | undefined + if (!row) { + database.exec('COMMIT') + return undefined + } + const claimed = database + .prepare( + `UPDATE conversation_queue_items + SET status = 'dispatching' + WHERE id = ? AND status = 'pending'` + ) + .run(row.id) + if (claimed.changes !== 1) { + database.exec('COMMIT') + return undefined + } + const item = toConversationQueueItem({ + ...row, + status: 'dispatching' + }) + if (row.source === 'user') { + database.exec('COMMIT') + return { + source: 'user', + item: { ...item, source: 'user' }, + payloadJson: row.payload_json + } + } + if (!row.schedule_run_id) { + throw new Error('定时任务队列项缺少运行记录') + } + const scheduleRow = database + .prepare( + `SELECT s.*, t.id AS task_id, + t.conversation_id AS conversation_id + FROM schedule_runs sr + INNER JOIN schedules s ON s.id = sr.schedule_id + INNER JOIN tasks t ON t.schedule_id = s.id + WHERE sr.id = ? AND sr.status = 'pending'` + ) + .get(row.schedule_run_id) as ScheduleWithTaskRow | undefined + if (!scheduleRow) { + throw new Error('定时任务队列项已失效') + } + const scheduleClaim = database + .prepare( + `UPDATE schedule_runs + SET status = 'running' + WHERE id = ? AND status = 'pending'` + ) + .run(row.schedule_run_id) + if (scheduleClaim.changes !== 1) { + throw new Error('定时任务运行记录无法认领') + } + database.exec('COMMIT') + return { + source: 'schedule', + item: { ...item, source: 'schedule' }, + schedule: toSchedule(scheduleRow), + runId: row.schedule_run_id + } + } catch (error) { + database.exec('ROLLBACK') + throw error + } + } + + completeConversationUserQueueItem(itemId: string): void { + const result = this.requireDatabase() + .prepare( + `DELETE FROM conversation_queue_items + WHERE id = ? AND source = 'user' AND status = 'dispatching'` + ) + .run(itemId) + if (result.changes !== 1) { + throw new Error('待发送消息不存在或状态已变化') + } + } + + releaseConversationUserQueueItem(itemId: string): void { + const result = this.requireDatabase() + .prepare( + `UPDATE conversation_queue_items + SET status = 'pending' + WHERE id = ? AND source = 'user' AND status = 'dispatching'` + ) + .run(itemId) + if ( + result.changes !== 1 && + !this.requireDatabase() + .prepare( + `SELECT 1 + FROM conversation_queue_items + WHERE id = ? AND source = 'user' AND status = 'pending'` + ) + .get(itemId) + ) { + throw new Error('待发送消息不存在或状态已变化') + } + } + + removeConversationUserQueueItem(itemId: string): void { + const result = this.requireDatabase() + .prepare( + `DELETE FROM conversation_queue_items + WHERE id = ? AND source = 'user' AND status = 'pending'` + ) + .run(itemId) + if (result.changes !== 1) { + throw new Error('待发送消息不存在或状态已变化') + } + } + + cancelConversationQueueItem( + itemId: string + ): ConversationQueueItem { + const row = this.requireDatabase() + .prepare( + `SELECT * + FROM conversation_queue_items + WHERE id = ? AND status = 'pending'` + ) + .get(itemId) as ConversationQueueRow | undefined + if (!row) { + throw new Error('待执行项不存在或状态已变化') + } + const item = toConversationQueueItem(row) + if (row.source === 'user') { + this.removeConversationUserQueueItem(itemId) + return item + } + if (!row.schedule_run_id) { + throw new Error('定时任务队列项缺少运行记录') + } + this.completeScheduleRun(row.schedule_run_id, 'cancelled') + return item + } + listSchedules(projectId?: string): AssistantSchedule[] { const rows = projectId ? this.requireDatabase() @@ -3772,21 +4139,22 @@ export class AssistantDatabase { } } - claimDueSchedules(now = new Date(), limit = 4): ClaimedSchedule[] { + queueDueSchedules( + now = new Date(), + limit = 100 + ): ConversationQueueItem[] { const database = this.requireDatabase() const nowIso = now.toISOString() - const safeLimit = Math.max(1, Math.min(16, Math.trunc(limit))) + const safeLimit = Math.max(1, Math.min(100, Math.trunc(limit))) database.exec('BEGIN IMMEDIATE') try { - const claims: ClaimedSchedule[] = [] - const pending = database + const due = database .prepare( - `SELECT sr.id AS run_id, s.*, t.id AS task_id, + `SELECT s.*, t.id AS task_id, t.conversation_id AS conversation_id - FROM schedule_runs sr - INNER JOIN schedules s ON s.id = sr.schedule_id + FROM schedules s INNER JOIN tasks t ON t.schedule_id = s.id - WHERE sr.status = 'pending' + WHERE s.enabled = 1 AND s.next_run_at <= ? AND ( s.project_id IS NULL OR EXISTS ( SELECT 1 FROM projects p @@ -3795,83 +4163,69 @@ export class AssistantDatabase { AND p.kind = 'user' ) ) - ORDER BY sr.scheduled_for + AND NOT EXISTS ( + SELECT 1 + FROM schedule_runs active + WHERE active.schedule_id = s.id + AND active.status IN ('pending', 'running') + ) + ORDER BY s.next_run_at LIMIT ?` ) - .all(safeLimit) as Array< - ScheduleWithTaskRow & { run_id: string } - > - const claimPending = database - .prepare( - `UPDATE schedule_runs - SET status = 'running' - WHERE id = ? AND status = 'pending'` - ) - for (const row of pending) { - if (claimPending.run(row.run_id).changes === 1) { - claims.push({ - schedule: toSchedule(row), - runId: row.run_id + .all(nowIso, safeLimit) as ScheduleWithTaskRow[] + const insertRun = database.prepare( + `INSERT OR IGNORE INTO schedule_runs + (id, schedule_id, scheduled_for, task_id, status) + VALUES (?, ?, ?, ?, 'pending')` + ) + const insertQueueItem = database.prepare( + `INSERT INTO conversation_queue_items + (id, conversation_id, source, label, payload_json, + schedule_run_id, schedule_id, task_id, status, created_at) + VALUES (?, ?, 'schedule', ?, '{}', ?, ?, ?, 'pending', ?)` + ) + const queued: ConversationQueueItem[] = [] + for (const row of due) { + const schedule = toSchedule(row) + const runId = randomUUID() + if ( + insertRun.run( + runId, + schedule.id, + schedule.nextRunAt, + schedule.taskId + ).changes === 1 + ) { + insertQueueItem.run( + runId, + schedule.conversationId, + schedule.title, + runId, + schedule.id, + schedule.taskId, + schedule.nextRunAt + ) + queued.push({ + id: runId, + conversationId: schedule.conversationId, + source: 'schedule', + label: schedule.title, + createdAt: schedule.nextRunAt, + scheduleRunId: runId, + scheduleId: schedule.id, + taskId: schedule.taskId }) } } - - const remaining = safeLimit - claims.length - if (remaining > 0) { - const due = database - .prepare( - `SELECT s.*, t.id AS task_id, - t.conversation_id AS conversation_id - FROM schedules s - INNER JOIN tasks t ON t.schedule_id = s.id - WHERE s.enabled = 1 AND s.next_run_at <= ? - AND ( - s.project_id IS NULL OR EXISTS ( - SELECT 1 FROM projects p - WHERE p.id = s.project_id - AND p.status = 'active' - AND p.kind = 'user' - ) - ) - AND NOT EXISTS ( - SELECT 1 - FROM schedule_runs active - WHERE active.schedule_id = s.id - AND active.status IN ('pending', 'running') - ) - ORDER BY s.next_run_at - LIMIT ?` - ) - .all(nowIso, remaining) as ScheduleWithTaskRow[] - const insertRun = database.prepare( - `INSERT OR IGNORE INTO schedule_runs - (id, schedule_id, scheduled_for, task_id, status) - VALUES (?, ?, ?, ?, 'running')` - ) - for (const row of due) { - const schedule = toSchedule(row) - const runId = randomUUID() - if ( - insertRun.run( - runId, - schedule.id, - schedule.nextRunAt, - schedule.taskId - ).changes === 1 - ) { - claims.push({ schedule, runId }) - } - } - } database.exec('COMMIT') - return claims + return queued } catch (error) { database.exec('ROLLBACK') throw error } } - claimScheduleNow(scheduleId: string): ClaimedSchedule { + queueScheduleNow(scheduleId: string): ConversationQueueItem { const schedule = this.getSchedule(scheduleId) const database = this.requireDatabase() const runId = randomUUID() @@ -3897,7 +4251,7 @@ export class AssistantDatabase { .prepare( `INSERT INTO schedule_runs (id, schedule_id, scheduled_for, task_id, status) - VALUES (?, ?, ?, ?, 'running') + VALUES (?, ?, ?, ?, 'pending') ON CONFLICT(schedule_id, scheduled_for) DO NOTHING` ) .run( @@ -3910,17 +4264,34 @@ export class AssistantDatabase { if (!inserted) { throw new Error('定时任务正在启动，请稍后重试') } + const createdAt = new Date(baseTime).toISOString() + database + .prepare( + `INSERT INTO conversation_queue_items + (id, conversation_id, source, label, payload_json, + schedule_run_id, schedule_id, task_id, status, created_at) + VALUES (?, ?, 'schedule', ?, '{}', ?, ?, ?, 'pending', ?)` + ) + .run( + runId, + schedule.conversationId, + schedule.title, + runId, + schedule.id, + schedule.taskId, + createdAt + ) database.exec('COMMIT') } catch (error) { database.exec('ROLLBACK') throw error } - return { schedule, runId } + return this.getConversationQueueItem(runId)! } completeScheduleRun( runId: string, - status: 'completed' | 'failed', + status: 'completed' | 'failed' | 'cancelled', now = new Date() ): void { const database = this.requireDatabase() @@ -3935,7 +4306,8 @@ export class AssistantDatabase { FROM schedule_runs sr INNER JOIN schedules s ON s.id = sr.schedule_id INNER JOIN tasks t ON t.schedule_id = s.id - WHERE sr.id = ? AND sr.status = 'running'` + WHERE sr.id = ? + AND sr.status IN ('pending', 'running')` ) .get(runId) as | (ScheduleWithTaskRow & { scheduled_for: string }) @@ -3950,6 +4322,12 @@ export class AssistantDatabase { WHERE id = ?` ) .run(status, runId) + database + .prepare( + `DELETE FROM conversation_queue_items + WHERE schedule_run_id = ?` + ) + .run(runId) const schedule = toSchedule(row) if (row.scheduled_for === row.next_run_at) { const next = new Date(row.scheduled_for) @@ -3972,7 +4350,12 @@ export class AssistantDatabase { .prepare( `UPDATE schedules SET enabled = CASE WHEN ? = 1 THEN 0 ELSE enabled END, - next_run_at = ?, last_run_at = ?, updated_at = ? + next_run_at = ?, + last_run_at = CASE + WHEN ? = 'cancelled' THEN last_run_at + ELSE ? + END, + updated_at = ? WHERE id = ? AND next_run_at = ?` ) .run( @@ -3980,19 +4363,22 @@ export class AssistantDatabase { schedule.recurrence === 'once' ? schedule.nextRunAt : next.toISOString(), + status, nowIso, nowIso, schedule.id, row.scheduled_for ) } else { - database - .prepare( - `UPDATE schedules - SET last_run_at = ?, updated_at = ? - WHERE id = ?` - ) - .run(nowIso, nowIso, schedule.id) + if (status !== 'cancelled') { + database + .prepare( + `UPDATE schedules + SET last_run_at = ?, updated_at = ? + WHERE id = ?` + ) + .run(nowIso, nowIso, schedule.id) + } } const updatedSchedule = database .prepare('SELECT enabled FROM schedules WHERE id = ?') @@ -4000,6 +4386,9 @@ export class AssistantDatabase { const finalTaskStatus = status === 'failed' ? 'failed' + : status === 'cancelled' && + schedule.recurrence === 'once' + ? 'cancelled' : updatedSchedule.enabled === 1 ? 'queued' : schedule.recurrence === 'once' @@ -4010,7 +4399,7 @@ export class AssistantDatabase { `UPDATE tasks SET status = ?, completed_at = CASE - WHEN ? IN ('completed', 'failed') THEN ? + WHEN ? IN ('completed', 'failed', 'cancelled') THEN ? ELSE NULL END WHERE id = ?` @@ -5352,12 +5741,12 @@ export class AssistantDatabase { const version = database .prepare('PRAGMA user_version') .get() as { user_version: number } - if (version.user_version > 22) { + if (version.user_version > 23) { throw new Error( `当前 GoodBuddy 不支持助理数据库版本 ${version.user_version}，请升级应用后重试` ) } - if (version.user_version === 22) { + if (version.user_version === 23) { return } if (version.user_version < 1) { @@ -6562,6 +6951,56 @@ export class AssistantDatabase { throw error } } + if (version.user_version < 23) { + database.exec('BEGIN IMMEDIATE') + try { + database.exec(` + CREATE TABLE IF NOT EXISTS conversation_queue_items ( + id TEXT PRIMARY KEY, + conversation_id TEXT NOT NULL + REFERENCES conversations(id) ON DELETE CASCADE, + source TEXT NOT NULL CHECK(source IN ('user', 'schedule')), + label TEXT NOT NULL, + payload_json TEXT NOT NULL DEFAULT '{}', + schedule_run_id TEXT UNIQUE + REFERENCES schedule_runs(id) ON DELETE CASCADE, + schedule_id TEXT + REFERENCES schedules(id) ON DELETE CASCADE, + task_id TEXT + REFERENCES tasks(id) ON DELETE SET NULL, + status TEXT NOT NULL + CHECK(status IN ('pending', 'dispatching')), + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS conversation_queue_pending_idx + ON conversation_queue_items( + conversation_id, status, created_at, id + ); + CREATE INDEX IF NOT EXISTS conversation_queue_schedule_pending_idx + ON conversation_queue_items( + source, status, created_at, conversation_id + ); + CREATE INDEX IF NOT EXISTS conversation_queue_status_created_idx + ON conversation_queue_items( + status, created_at, conversation_id + ); + INSERT OR IGNORE INTO conversation_queue_items + (id, conversation_id, source, label, payload_json, + schedule_run_id, schedule_id, task_id, status, created_at) + SELECT sr.id, t.conversation_id, 'schedule', t.title, '{}', + sr.id, sr.schedule_id, t.id, 'pending', sr.scheduled_for + FROM schedule_runs sr + INNER JOIN tasks t ON t.schedule_id = sr.schedule_id + WHERE sr.status IN ('pending', 'running') + AND t.conversation_id IS NOT NULL; + PRAGMA user_version = 23; + COMMIT; + `) + } catch (error) { + database.exec('ROLLBACK') + throw error + } + } } private requireDatabase(): DatabaseSync { diff --git a/src/main/assistant/heartbeat-database.test.ts b/src/main/assistant/heartbeat-database.test.ts index 34e000f..6478c77 100644 --- a/src/main/assistant/heartbeat-database.test.ts +++ b/src/main/assistant/heartbeat-database.test.ts @@ -102,7 +102,7 @@ describe('AssistantDatabase heartbeat persistence', () => { ).count check.close() migrated.close() - expect(version).toBe(22) + expect(version).toBe(23) expect(heartbeatTableCount).toBe(4) }) diff --git a/src/main/context-manager.test.ts b/src/main/context-manager.test.ts index 2ba3607..5f14779 100644 --- a/src/main/context-manager.test.ts +++ b/src/main/context-manager.test.ts @@ -121,6 +121,45 @@ describe('ContextManager', () => { ]) }) + it('serializes queued attachments and restores their bounded contents', async () => { + const manager = new ContextManager() + const content = Buffer.from('persisted queued context', 'utf8') + const attachment = await manager.ingestRemoteAttachment({ + name: 'queued.txt', + mimeType: 'text/plain', + size: content.byteLength, + kind: 'file', + dataBase64: content.toString('base64') + }) + const serialized = manager.serializeForQueue([attachment.id]) + + manager.clear() + manager.restoreFromQueue(serialized) + + expect( + manager.enrichRequest({ + requestId: '1f6a37b6-e0a3-449f-8878-b10d353fbfb4', + conversationId: 'conversation-1', + prompt: 'summarize', + contextIds: [attachment.id] + }).prompt + ).toContain('persisted queued context') + expect(() => + manager.restoreFromQueue( + JSON.stringify([ + { + id: 'bad', + name: 'bad.txt', + preview: '', + kind: 'text', + size: 99, + content: 'short' + } + ]) + ) + ).toThrow('待发送文本附件大小无效') + }) + it('only enriches prompts with files explicitly selected by the user', async () => { const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-context-')) temporaryDirectories.push(directory) diff --git a/src/main/context-manager.ts b/src/main/context-manager.ts index 553d661..79e662b 100644 --- a/src/main/context-manager.ts +++ b/src/main/context-manager.ts @@ -585,6 +585,138 @@ export class ContextManager { } } + serializeForQueue(contextIds: string[]): string { + if (contextIds.length > maximumAttachmentsPerMessage) { + throw new Error('单次消息最多添加 8 个附件') + } + const contexts = contextIds.map((contextId) => { + const context = this.contexts.get(contextId) + if (!context) { + throw new Error('附件上下文已失效，请重新添加') + } + return context + }) + return JSON.stringify(contexts) + } + + restoreFromQueue(serialized: string): void { + if ( + Buffer.byteLength(serialized) > + maximumContextBytes * 2 + 2_000_000 + ) { + throw new Error('待发送附件数据超过恢复上限') + } + const parsed = JSON.parse(serialized) as unknown + if ( + !Array.isArray(parsed) || + parsed.length > maximumAttachmentsPerMessage + ) { + throw new Error('待发送附件数据无效') + } + const restoredContexts: StoredContext[] = [] + const restoredIds = new Set() + for (const value of parsed) { + if (!value || typeof value !== 'object') { + throw new Error('待发送附件数据无效') + } + const candidate = value as Record + if ( + typeof candidate.id !== 'string' || + candidate.id.length === 0 || + candidate.id.length > 200 || + typeof candidate.name !== 'string' || + candidate.name.length === 0 || + candidate.name.length > 500 || + typeof candidate.preview !== 'string' || + candidate.preview.length > 500 || + (candidate.kind !== 'text' && candidate.kind !== 'image') + ) { + throw new Error('待发送附件数据无效') + } + if (this.contexts.has(candidate.id)) { + continue + } + if (restoredIds.has(candidate.id)) { + throw new Error('待发送附件数据包含重复项目') + } + let context: StoredContext + if (candidate.kind === 'text') { + if (typeof candidate.content !== 'string') { + throw new Error('待发送文本附件数据无效') + } + const size = Buffer.byteLength(candidate.content) + if ( + size === 0 || + size > maximumContextBytes || + candidate.size !== size + ) { + throw new Error('待发送文本附件大小无效') + } + context = { + id: candidate.id, + name: candidate.name, + preview: candidate.preview, + kind: 'text', + size, + content: candidate.content + } + } else { + if ( + candidate.mediaType !== 'image/jpeg' || + typeof candidate.data !== 'string' || + !/^[A-Za-z0-9+/]+={0,2}$/u.test(candidate.data) + ) { + throw new Error('待发送图片附件数据无效') + } + const image = Buffer.from(candidate.data, 'base64') + if ( + image.byteLength === 0 || + image.byteLength > maximumContextBytes || + candidate.size !== image.byteLength + ) { + throw new Error('待发送图片附件大小无效') + } + const thumbnailUrl = + typeof candidate.thumbnailUrl === 'string' && + candidate.thumbnailUrl.length <= 2_000_000 && + candidate.thumbnailUrl.startsWith( + 'data:image/jpeg;base64,' + ) + ? candidate.thumbnailUrl + : undefined + context = { + id: candidate.id, + name: candidate.name, + preview: candidate.preview, + kind: 'image', + size: image.byteLength, + mediaType: 'image/jpeg', + data: candidate.data, + ...(thumbnailUrl ? { thumbnailUrl } : {}) + } + } + restoredIds.add(context.id) + restoredContexts.push(context) + } + const restoredBytes = restoredContexts.reduce( + (total, context) => total + context.size, + 0 + ) + if ( + this.contexts.size + restoredContexts.length > + maximumContextCount + ) { + throw new Error('最多可暂存 16 个上下文项目') + } + if (this.totalBytes + restoredBytes > maximumContextBytes) { + throw new Error('上下文总大小不能超过 12MB') + } + for (const context of restoredContexts) { + this.contexts.set(context.id, context) + this.totalBytes += context.size + } + } + clear(): void { this.contexts.clear() this.totalBytes = 0 diff --git a/src/main/ipc.test.ts b/src/main/ipc.test.ts index 8c5c3ad..7d70e52 100644 --- a/src/main/ipc.test.ts +++ b/src/main/ipc.test.ts @@ -3,7 +3,10 @@ import { mkdtemp, realpath, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { ipcChannels } from '../shared/ipc-channels' -import type { AssistantProject } from '../shared/assistant-contracts' +import type { + AssistantProject, + ConversationQueueItem +} from '../shared/assistant-contracts' import type { AgentEvent, BrowserLiveState } from '../shared/contracts' import { defaultKnowledgeOntologySettings } from '../shared/knowledge-ontology' import { AssistantDatabase } from './assistant/assistant-database' @@ -163,7 +166,11 @@ describe('registerIpcHandlers computer capabilities', () => { capabilityService as never, { clear: vi.fn(), selectFiles } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, onRuntimeSettingsChanged, @@ -384,7 +391,11 @@ describe('registerIpcHandlers update source routing', () => { {} as never, { clear: vi.fn() } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => undefined), @@ -502,7 +513,11 @@ describe('registerIpcHandlers model download source routing', () => { {} as never, { clear: vi.fn() } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => undefined), @@ -694,7 +709,11 @@ describe('registerIpcHandlers DSH runtime extensions', () => { {} as never, { clear: vi.fn() } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, onRuntimeSettingsChanged, @@ -879,7 +898,9 @@ describe('registerIpcHandlers lifecycle tracking', () => { { clear: vi.fn() } as never, {} as never, { - claimDueSchedules: vi.fn(() => []), + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []), repairConversationRuntimeSelections: vi.fn() } as never, { clear: vi.fn() } as never, @@ -1007,7 +1028,11 @@ describe('registerIpcHandlers knowledge snapshot ontology', () => { {} as never, { clear: vi.fn() } as never, knowledgeService as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => undefined) @@ -1091,7 +1116,11 @@ describe('registerIpcHandlers knowledge embedding index', () => { {} as never, { clear: vi.fn() } as never, knowledgeService as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => undefined) @@ -1181,7 +1210,11 @@ describe('registerIpcHandlers knowledge task actions', () => { {} as never, { clear: vi.fn() } as never, knowledgeService as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => undefined) @@ -1280,7 +1313,11 @@ describe('registerIpcHandlers model ZIP dialogs', () => { {} as never, { clear: vi.fn() } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => undefined), @@ -1453,7 +1490,9 @@ describe('registerIpcHandlers document parsing', () => { } const createInlineArtifact = vi.fn((input) => input) const assistantDatabase = { - claimDueSchedules: vi.fn(() => []), + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []), createInlineArtifact } const webContents = { @@ -1622,7 +1661,11 @@ describe('registerIpcHandlers connection tests', () => { {} as never, contextManager as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, approvalBroker as never, {} as never, vi.fn(async () => {}) @@ -1739,7 +1782,11 @@ describe('registerIpcHandlers connection tests', () => { {} as never, { clear: vi.fn() } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => {}), @@ -1831,7 +1878,11 @@ describe('registerIpcHandlers Runtime config actions', () => { {} as never, { clear: vi.fn() } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => {}) @@ -1968,7 +2019,11 @@ describe('registerIpcHandlers window controls', () => { {} as never, { clear: vi.fn() } as never, {} as never, - { claimDueSchedules: vi.fn(() => []) } as never, + { + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) + } as never, { clear: vi.fn() } as never, {} as never, vi.fn(async () => {}) @@ -2028,7 +2083,9 @@ describe('registerIpcHandlers workspace files', () => { await writeFile(join(rootPath, 'README.md'), '# GoodBuddy\n') const projectId = '00000000-0000-4000-8000-000000000101' const assistantDatabase = { - claimDueSchedules: vi.fn(() => []), + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []), getProject: vi.fn(() => ({ id: projectId, rootPath })) } const webContents = { @@ -2107,14 +2164,17 @@ describe('registerIpcHandlers token usage', () => { records: [] } const assistantDatabase = { - claimDueSchedules: vi.fn(() => []), + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []), getTokenUsageSummary: vi.fn(() => summary) } const webContents = { mainFrame: { url: 'file:///goodbuddy/index.html' }, - getURL: vi.fn(() => 'file:///goodbuddy/index.html') + getURL: vi.fn(() => 'file:///goodbuddy/index.html'), + send: vi.fn() } const window = { webContents, @@ -2159,16 +2219,57 @@ describe('registerIpcHandlers local conversation persistence', () => { }) it('validates and forwards incremental saves and explicit deletions', async () => { + const conversationId = + '00000000-0000-4000-8000-000000000301' + const queuedAttachmentId = + '00000000-0000-4000-8000-000000000303' + const queuedItem = { + id: '00000000-0000-4000-8000-000000000304', + conversationId, + source: 'user' as const, + label: '待删除消息', + createdAt: '2026-08-20T09:01:00.000Z' + } const assistantDatabase = { - claimDueSchedules: vi.fn(() => []), + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => [queuedItem]), + listPendingConversationQueueIds: vi.fn(() => [conversationId]), + getConversationUserQueuePayloadJson: vi.fn(() => + JSON.stringify({ + input: { + conversationId, + runtimeSelection: { provider: 'auto' }, + workMode: 'ask', + includeMemoryContext: true, + prompt: queuedItem.label, + attachments: [ + { + id: queuedAttachmentId, + name: 'queued.txt', + size: 6, + preview: 'queued', + kind: 'text' + } + ], + knowledgeLibraryIds: [], + knowledgeRetrievalMode: 'auto' + }, + serializedContexts: '[]' + }) + ), saveLocalConversations: vi.fn(), deleteLocalConversation: vi.fn(() => true) } + const contextManager = { + clear: vi.fn(), + remove: vi.fn() + } const webContents = { mainFrame: { url: 'file:///goodbuddy/index.html' }, - getURL: vi.fn(() => 'file:///goodbuddy/index.html') + getURL: vi.fn(() => 'file:///goodbuddy/index.html'), + send: vi.fn() } const window = { webContents, @@ -2182,7 +2283,7 @@ describe('registerIpcHandlers local conversation persistence', () => { 'CommandOrControl+Shift+Space', {} as never, {} as never, - { clear: vi.fn() } as never, + contextManager as never, {} as never, assistantDatabase as never, { clear: vi.fn() } as never, @@ -2193,8 +2294,6 @@ describe('registerIpcHandlers local conversation persistence', () => { sender: webContents, senderFrame: webContents.mainFrame } - const conversationId = - '00000000-0000-4000-8000-000000000301' const messageId = '00000000-0000-4000-8000-000000000302' const batch = [ { @@ -2231,6 +2330,9 @@ describe('registerIpcHandlers local conversation persistence', () => { expect( assistantDatabase.deleteLocalConversation ).toHaveBeenCalledWith(conversationId) + expect(contextManager.remove).toHaveBeenCalledWith( + queuedAttachmentId + ) expect(() => electronMocks.handlers.get( @@ -2260,7 +2362,9 @@ describe('registerIpcHandlers local conversation persistence', () => { it('waits for the renderer persistence acknowledgement before removing handlers', async () => { const assistantDatabase = { - claimDueSchedules: vi.fn(() => []) + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []) } const webContents = { mainFrame: { @@ -2424,7 +2528,9 @@ describe('registerIpcHandlers Runtime customization', () => { | { provider: 'opencode' } | undefined = { provider: 'opencode' } const assistantDatabase = { - claimDueSchedules: vi.fn(() => []), + queueDueSchedules: vi.fn(() => []), + listConversationQueueItems: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []), getProject: vi.fn(() => ({ id: projectId, rootPath: 'C:\\ProjectWorkspace' @@ -2634,7 +2740,6 @@ describe('registerIpcHandlers agent terminal state', () => { capabilityServiceOverride?: Record ) { const assistantDatabase = { - claimDueSchedules: vi.fn(() => []), createTask: vi.fn(), appendTaskEvent: vi.fn(), updateTaskStatus: vi.fn(), @@ -2670,8 +2775,22 @@ describe('registerIpcHandlers agent terminal state', () => { })), appendConversationMessage: vi.fn(), appendRemoteConversationMessage: vi.fn(), + listConversationQueueItems: vi.fn< + () => ConversationQueueItem[] + >(() => []), + listPendingScheduleQueueConversationIds: vi.fn(() => []), + listPendingConversationQueueIds: vi.fn(() => []), + getConversationQueueItem: vi.fn(), + getConversationUserQueuePayloadJson: vi.fn(), + isConversationUserQueueItemDispatching: vi.fn(() => true), + enqueueConversationUserInput: vi.fn(), + claimConversationQueueItem: vi.fn(), + completeConversationUserQueueItem: vi.fn(), + releaseConversationUserQueueItem: vi.fn(), + cancelConversationQueueItem: vi.fn(), + queueDueSchedules: vi.fn(() => []), + queueScheduleNow: vi.fn(), completeScheduleRun: vi.fn(), - claimScheduleNow: vi.fn(), listSchedules: vi.fn(() => []), createSchedule: vi.fn(), setScheduleEnabled: vi.fn(), @@ -2711,6 +2830,8 @@ describe('registerIpcHandlers agent terminal state', () => { kind: attachment.kind === 'image' ? 'image' : 'text' })), remove: vi.fn(), + serializeForQueue: vi.fn(() => '[]'), + restoreFromQueue: vi.fn(), clear: vi.fn() } const approvalBroker = { @@ -2893,7 +3014,25 @@ describe('registerIpcHandlers agent terminal state', () => { createdAt: '2026-08-19T00:00:00.000Z', updatedAt: '2026-08-19T00:00:00.000Z' } - harness.assistantDatabase.claimScheduleNow.mockReturnValue({ + const queueItem = { + id: runId, + conversationId, + source: 'schedule' as const, + label: schedule.title, + scheduleRunId: runId, + scheduleId, + taskId, + createdAt: '2026-08-19T00:01:00.000Z' + } + harness.assistantDatabase.queueScheduleNow.mockReturnValue( + queueItem + ) + harness.assistantDatabase.listConversationQueueItems + .mockReturnValueOnce([queueItem]) + .mockReturnValue([]) + harness.assistantDatabase.claimConversationQueueItem.mockReturnValue({ + source: 'schedule', + item: queueItem, schedule, runId }) @@ -2940,6 +3079,298 @@ describe('registerIpcHandlers agent terminal state', () => { await harness.dispose() }) + it('serializes Agent runs that target the same Conversation', async () => { + let finishRun: (() => void) | undefined + const runtimeStarted = vi.fn() + const runtime = { + runtimeId: 'model', + capability: 'chat', + supportsToolExecution: false, + async *run(request: { requestId: string }) { + runtimeStarted(request.requestId) + await new Promise((resolve) => { + finishRun = resolve + }) + yield { + requestId: request.requestId, + type: 'done' + } as const + } + } + const harness = createHarness(runtime) + const conversationId = + '00000000-0000-4000-8000-000000000721' + const firstRun = harness.handler?.( + trustedEvent(harness.webContents), + { + requestId: '00000000-0000-4000-8000-000000000722', + conversationId, + prompt: '第一条', + workMode: 'ask', + knowledgeLibraryIds: [] + } + ) + await vi.waitFor(() => expect(runtimeStarted).toHaveBeenCalled()) + + await expect( + harness.handler?.(trustedEvent(harness.webContents), { + requestId: '00000000-0000-4000-8000-000000000723', + conversationId, + prompt: '第二条', + workMode: 'ask', + knowledgeLibraryIds: [] + }) + ).rejects.toThrow('当前对话已有执行中的请求') + + finishRun?.() + await firstRun + await harness.dispose() + }) + + it('reserves a Conversation while its Runtime is resolving', async () => { + const runtime = { + runtimeId: 'model', + capability: 'chat', + supportsToolExecution: false, + async *run(request: { requestId: string }) { + yield { + requestId: request.requestId, + type: 'done' + } as const + } + } + let resolveRuntime: + | ((value: typeof runtime) => void) + | undefined + const runtimeResolution = new Promise((resolve) => { + resolveRuntime = resolve + }) + const selectedRuntimes = { + getRuntime: vi.fn(() => runtimeResolution) + } + const harness = createHarness( + runtime, + undefined, + 'always', + undefined, + false, + selectedRuntimes + ) + const conversationId = + '00000000-0000-4000-8000-000000000724' + const firstRun = harness.handler?.( + trustedEvent(harness.webContents), + { + requestId: '00000000-0000-4000-8000-000000000728', + conversationId, + runtimeSelection: { provider: 'auto' }, + prompt: '等待 Runtime', + workMode: 'ask', + knowledgeLibraryIds: [] + } + ) + await vi.waitFor(() => + expect(selectedRuntimes.getRuntime).toHaveBeenCalledOnce() + ) + + await expect( + harness.handler?.(trustedEvent(harness.webContents), { + requestId: '00000000-0000-4000-8000-000000000729', + conversationId, + runtimeSelection: { provider: 'auto' }, + prompt: '不能并发', + workMode: 'ask', + knowledgeLibraryIds: [] + }) + ).rejects.toThrow('当前对话已有执行中的请求') + expect(selectedRuntimes.getRuntime).toHaveBeenCalledOnce() + + resolveRuntime?.(runtime) + await firstRun + await harness.dispose() + }) + + it('interrupts the active response and promotes the selected queue item', async () => { + const runtimeStarted = vi.fn() + const runtime = { + runtimeId: 'model', + capability: 'chat', + supportsToolExecution: false, + async *run( + request: { requestId: string }, + signal: AbortSignal + ) { + runtimeStarted(request.requestId) + await new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { + once: true + }) + }) + if (signal.aborted) { + throw signal.reason + } + yield { + requestId: request.requestId, + type: 'done' + } as const + } + } + const harness = createHarness(runtime) + const conversationId = + '00000000-0000-4000-8000-000000000725' + const item = { + id: '00000000-0000-4000-8000-000000000726', + conversationId, + source: 'user' as const, + label: '优先执行', + createdAt: '2026-08-20T09:01:00.000Z' + } + const input = { + conversationId, + runtimeSelection: { provider: 'auto' as const }, + workMode: 'ask' as const, + includeMemoryContext: true, + prompt: item.label, + attachments: [], + knowledgeLibraryIds: [], + knowledgeRetrievalMode: 'auto' as const + } + harness.assistantDatabase.getConversationQueueItem.mockReturnValue( + item + ) + harness.assistantDatabase.claimConversationQueueItem.mockReturnValueOnce({ + source: 'user', + item, + payloadJson: JSON.stringify(input) + }) + electronMocks.handlers.get( + ipcChannels.conversationQueueReady + )?.(trustedEvent(harness.webContents), conversationId) + + await harness.handler?.(trustedEvent(harness.webContents), { + requestId: '00000000-0000-4000-8000-000000000727', + conversationId, + prompt: '当前回复', + workMode: 'ask', + knowledgeLibraryIds: [] + }) + await vi.waitFor(() => expect(runtimeStarted).toHaveBeenCalled()) + + electronMocks.handlers.get( + ipcChannels.conversationQueueInterruptAndRun + )?.(trustedEvent(harness.webContents), item.id) + + await vi.waitFor(() => + expect(harness.webContents.send).toHaveBeenCalledWith( + ipcChannels.conversationQueueDispatch, + { item, input } + ) + ) + expect( + harness.assistantDatabase.claimConversationQueueItem + ).toHaveBeenCalledWith(conversationId, item.id) + await harness.dispose() + }) + + it('dispatches and accepts a queued user message through the renderer', async () => { + const runtime = { + runtimeId: 'model', + capability: 'chat', + supportsToolExecution: false, + async *run(request: { requestId: string }) { + yield { + requestId: request.requestId, + type: 'done' + } as const + } + } + const harness = createHarness(runtime) + const conversationId = + '00000000-0000-4000-8000-000000000731' + const itemId = '00000000-0000-4000-8000-000000000732' + const attachmentId = + '00000000-0000-4000-8000-000000000734' + const input = { + conversationId, + runtimeSelection: { provider: 'auto' as const }, + workMode: 'ask' as const, + includeMemoryContext: true, + prompt: '排队发送', + attachments: [ + { + id: attachmentId, + name: 'queued.txt', + size: 6, + preview: 'queued', + kind: 'text' as const + } + ], + knowledgeLibraryIds: [], + knowledgeRetrievalMode: 'auto' as const + } + const item = { + id: itemId, + conversationId, + source: 'user' as const, + label: input.prompt, + createdAt: '2026-08-19T00:01:00.000Z' + } + harness.assistantDatabase.enqueueConversationUserInput.mockReturnValue( + item + ) + harness.assistantDatabase.listConversationQueueItems + .mockReturnValueOnce([item]) + .mockReturnValue([]) + harness.assistantDatabase.claimConversationQueueItem.mockReturnValueOnce({ + source: 'user', + item, + payloadJson: JSON.stringify({ + input, + serializedContexts: '[]' + }) + }) + harness.assistantDatabase.getConversationQueueItem.mockReturnValue( + item + ) + + const enqueueHandler = electronMocks.handlers.get( + ipcChannels.conversationQueueEnqueueUser + ) + expect( + enqueueHandler?.(trustedEvent(harness.webContents), input) + ).toEqual(item) + await vi.waitFor(() => + expect(harness.webContents.send).toHaveBeenCalledWith( + ipcChannels.conversationQueueDispatch, + { item, input } + ) + ) + + await harness.handler?.(trustedEvent(harness.webContents), { + requestId: '00000000-0000-4000-8000-000000000733', + conversationId, + queueItemId: itemId, + runtimeSelection: input.runtimeSelection, + prompt: input.prompt, + workMode: input.workMode, + knowledgeLibraryIds: [] + }) + + expect( + harness.assistantDatabase.completeConversationUserQueueItem + ).toHaveBeenCalledWith(itemId) + expect(harness.contextManager.serializeForQueue).toHaveBeenCalledWith([ + attachmentId + ]) + expect(harness.contextManager.remove).toHaveBeenCalledWith( + attachmentId + ) + expect(harness.contextManager.restoreFromQueue).toHaveBeenCalledWith( + '[]' + ) + await harness.dispose() + }) + it('publishes Runtime usage as context metrics with one settings read', async () => { const runtime = { runtimeId: 'continue', diff --git a/src/main/ipc.ts b/src/main/ipc.ts index b6304b7..08d1b72 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -24,6 +24,7 @@ import { agentRequestSchema, browserInteractRequestSchema, browserStopRequestSchema, + conversationQueueUserInputSchema, defaultRuntimeSettings, knowledgeCreateSchema, knowledgeEntityUpdateSchema, @@ -52,6 +53,8 @@ import { type AgentRequest, type AppInfo, type BrowserLiveState, + type ConversationQueueDispatch, + type ConversationQueueUserInput, type KnowledgeSearchReference, type KnowledgeSnapshot, type RuntimeSettings @@ -883,6 +886,7 @@ export function registerIpcHandlers( runtimeExtensionStore?: RuntimeExtensionStore ): () => Promise { const activeRequests = new Map() + const activeRequestConversations = new Map() const activeEventBuffers = new Map() const pendingAgentQuestions = new Map< string, @@ -1247,6 +1251,220 @@ export function registerIpcHandlers( window.webContents.send(ipcChannels.conversationsChanged) } } + const publishConversationQueueChange = ( + conversationId?: string + ): void => { + if (!window.isDestroyed()) { + if (conversationId) { + window.webContents.send( + ipcChannels.conversationQueueChanged, + conversationId + ) + } else { + window.webContents.send( + ipcChannels.conversationQueueChanged + ) + } + } + } + const readyConversationQueues = new Set() + const preferredConversationQueueItems = new Map() + const reservedConversationQueueItems = new Map() + const preparingRequestConversations = new Map() + const rendererReadyConversationQueues = new Set() + const queueDispatchTimers = new Map() + const parseConversationQueueUserPayload = ( + payloadJson: string, + restoreContexts = false + ): ConversationQueueUserInput => { + const parsed = JSON.parse(payloadJson) as unknown + if ( + parsed && + typeof parsed === 'object' && + 'input' in parsed + ) { + const stored = parsed as { + input: unknown + serializedContexts?: unknown + } + const input = conversationQueueUserInputSchema.parse(stored.input) + if ( + stored.serializedContexts !== undefined && + typeof stored.serializedContexts !== 'string' + ) { + throw new Error('待发送附件数据无效') + } + if ( + restoreContexts && + typeof stored.serializedContexts === 'string' + ) { + contextManager.restoreFromQueue(stored.serializedContexts) + } + return input + } + return conversationQueueUserInputSchema.parse(parsed) + } + const pumpingConversationQueues = new Set() + const maximumConcurrentScheduleRuns = 4 + let activeScheduleRuns = 0 + + const isConversationExecuting = ( + conversationId: string + ): boolean => + reservedConversationQueueItems.has(conversationId) || + [...preparingRequestConversations.values()].some( + (candidate) => candidate === conversationId + ) || + [...activeRequestConversations.values()].some( + (candidate) => candidate === conversationId + ) + + const pumpConversationQueue = async ( + conversationId: string, + preferredItemId?: string + ): Promise => { + const preferred = + preferredItemId ?? + preferredConversationQueueItems.get(conversationId) + if ( + shuttingDown || + executionPaused || + pumpingConversationQueues.has(conversationId) || + isConversationExecuting(conversationId) + ) { + return + } + const pendingItem = preferred + ? assistantDatabase.getConversationQueueItem(preferred) + : assistantDatabase.listConversationQueueItems(conversationId)[0] + if (!pendingItem || pendingItem.conversationId !== conversationId) { + preferredConversationQueueItems.delete(conversationId) + return + } + if ( + pendingItem.source === 'user' && + !rendererReadyConversationQueues.has(conversationId) + ) { + return + } + if ( + pendingItem.source === 'schedule' && + activeScheduleRuns >= maximumConcurrentScheduleRuns + ) { + return + } + pumpingConversationQueues.add(conversationId) + try { + if (isConversationExecuting(conversationId)) { + return + } + const claimed = assistantDatabase.claimConversationQueueItem( + conversationId, + preferred + ) + if (!claimed) { + return + } + readyConversationQueues.delete(conversationId) + preferredConversationQueueItems.delete(conversationId) + publishConversationQueueChange(conversationId) + if (claimed.source === 'user') { + if (window.isDestroyed()) { + assistantDatabase.releaseConversationUserQueueItem( + claimed.item.id + ) + readyConversationQueues.add(conversationId) + return + } + let input: ConversationQueueUserInput + try { + input = parseConversationQueueUserPayload( + claimed.payloadJson, + true + ) + } catch { + assistantDatabase.releaseConversationUserQueueItem( + claimed.item.id + ) + readyConversationQueues.add(conversationId) + publishConversationQueueChange(conversationId) + return + } + reservedConversationQueueItems.set( + conversationId, + claimed.item.id + ) + const dispatchTimeout = setTimeout(() => { + queueDispatchTimers.delete(claimed.item.id) + if ( + reservedConversationQueueItems.get(conversationId) !== + claimed.item.id + ) { + return + } + reservedConversationQueueItems.delete(conversationId) + try { + assistantDatabase.releaseConversationUserQueueItem( + claimed.item.id + ) + } catch { + return + } + for (const attachment of input.attachments) { + contextManager.remove(attachment.id) + } + readyConversationQueues.add(conversationId) + publishConversationQueueChange(conversationId) + void pumpConversationQueue(conversationId) + }, 30_000) + queueDispatchTimers.set(claimed.item.id, dispatchTimeout) + const dispatch: ConversationQueueDispatch = { + item: claimed.item, + input + } + window.webContents.send( + ipcChannels.conversationQueueDispatch, + dispatch + ) + return + } + + activeScheduleRuns += 1 + const execution = (async () => { + const result = await executeTaskWork({ + origin: 'schedule', + schedule: claimed.schedule, + scheduleRunId: claimed.runId + }) + assistantDatabase.completeScheduleRun( + claimed.runId, + result.status + ) + publishConversationChange() + })() + publishConversationChange() + void trackExecution(execution) + .catch(() => undefined) + .finally(() => { + activeScheduleRuns -= 1 + readyConversationQueues.add(conversationId) + publishConversationQueueChange(conversationId) + void pumpConversationQueue(conversationId) + for (const pendingConversationId of + assistantDatabase.listPendingScheduleQueueConversationIds( + maximumConcurrentScheduleRuns + )) { + if ( + readyConversationQueues.has(pendingConversationId) + ) { + void pumpConversationQueue(pendingConversationId) + } + } + }) + } finally { + pumpingConversationQueues.delete(conversationId) + } + } type ExecutionTemplate = Omit< AssistantSchedule, @@ -1289,7 +1507,7 @@ export function registerIpcHandlers( const executeTaskWork = async ( input: TaskWorkExecution ): Promise<{ - status: 'completed' | 'failed' + status: 'completed' | 'failed' | 'cancelled' output?: string error?: string attachments?: ChannelMediaAttachment[] @@ -1303,7 +1521,7 @@ export function registerIpcHandlers( return { status: 'failed', error: '应用正在退出' } } if (externalSignal?.aborted) { - return { status: 'failed', error: '请求已取消' } + return { status: 'cancelled', error: '请求已取消' } } const taskId = input.origin === 'schedule' @@ -1327,6 +1545,7 @@ export function registerIpcHandlers( ? input.schedule.conversationId : undefined) ?? `${origin}:${schedule.id}` + activeRequestConversations.set(requestId, runtimeConversationId) if (input.origin !== 'delegation') { assistantDatabase.updateTaskStatus(taskId, 'running') } else { @@ -1661,9 +1880,10 @@ export function registerIpcHandlers( } catch (error) { eventBuffer.flush() const message = safeRuntimeError(error, '定时任务执行失败') + const cancelled = controller.signal.aborted assistantDatabase.updateTaskStatus( taskId, - controller.signal.aborted ? 'cancelled' : 'failed', + cancelled ? 'cancelled' : 'failed', message ) if (input.origin === 'schedule') { @@ -1672,7 +1892,7 @@ export function registerIpcHandlers( role: 'assistant', content: message, state: 'error', - status: '定时任务失败', + status: cancelled ? '定时任务已取消' : '定时任务失败', task: { id: taskId, title: schedule.title @@ -1684,13 +1904,18 @@ export function registerIpcHandlers( title: origin === 'channel' ? `${remoteContext?.channelLabel ?? '远程通道'}请求失败` - : `定时任务失败：${schedule.title}`, + : cancelled + ? `定时任务已取消：${schedule.title}` + : `定时任务失败：${schedule.title}`, body: origin === 'channel' ? '打开 GoodBuddy 查看远程通道会话详情。' : '打开 GoodBuddy 任务工作栏查看详情。' }) - return { status: 'failed', error: message } + return { + status: cancelled ? 'cancelled' : 'failed', + error: message + } } finally { eventBuffer.close() externalSignal?.removeEventListener( @@ -1700,6 +1925,7 @@ export function registerIpcHandlers( knowledgeGateway?.revoke(knowledgeCapabilityToken) goodbuddyConfigService?.revokeRequest(requestId) activeRequests.delete(requestId) + activeRequestConversations.delete(requestId) await flushGoodBuddyConfigReload().catch(() => undefined) } } @@ -1810,47 +2036,37 @@ export function registerIpcHandlers( yield { requestId: request.requestId, type: 'done' } } - const maximumConcurrentScheduleRuns = 4 - let scheduleClaimRunning = false - let activeScheduleRuns = 0 - const launchDueSchedules = (): void => { + let scheduleQueueTickRunning = false + const queueDueSchedules = (): void => { if ( - scheduleClaimRunning || + scheduleQueueTickRunning || shuttingDown || - executionPaused || - activeScheduleRuns >= maximumConcurrentScheduleRuns + executionPaused ) { return } - scheduleClaimRunning = true + scheduleQueueTickRunning = true try { - const claims = assistantDatabase.claimDueSchedules( - new Date(), - maximumConcurrentScheduleRuns - activeScheduleRuns + const queued = assistantDatabase.queueDueSchedules(new Date()) + const conversationIds = new Set( + queued.map((item) => item.conversationId) ) - activeScheduleRuns += claims.length - for (const claim of claims) { - const execution = (async () => { - const result = await executeTaskWork({ - origin: 'schedule', - schedule: claim.schedule, - scheduleRunId: claim.runId - }) - assistantDatabase.completeScheduleRun( - claim.runId, - result.status - ) - publishConversationChange() - })() - void trackExecution(execution) - .catch(() => undefined) - .finally(() => { - activeScheduleRuns -= 1 - launchDueSchedules() - }) + for (const conversationId of conversationIds) { + publishConversationQueueChange(conversationId) + if (!isConversationExecuting(conversationId)) { + readyConversationQueues.add(conversationId) + void pumpConversationQueue(conversationId) + } } } finally { - scheduleClaimRunning = false + scheduleQueueTickRunning = false + } + } + const resumePendingConversationQueues = (): void => { + for (const conversationId of + assistantDatabase.listPendingConversationQueueIds()) { + readyConversationQueues.add(conversationId) + void pumpConversationQueue(conversationId) } } let heartbeatTickRunning = false @@ -1870,10 +2086,11 @@ export function registerIpcHandlers( } } const runDueWork = (): void => { - launchDueSchedules() + queueDueSchedules() void trackExecution(runDueHeartbeats()).catch(() => undefined) } const scheduleInterval = setInterval(runDueWork, 30_000) + resumePendingConversationQueues() runDueWork() const delegationEndpoint = process.env.GOODBUDDY_DELEGATION_ENDPOINT?.trim() @@ -1911,7 +2128,14 @@ export function registerIpcHandlers( updatedAt: new Date().toISOString() } }) - ) + ).then((result) => ({ + status: + result.status === 'completed' + ? ('completed' as const) + : ('failed' as const), + ...(result.output ? { output: result.output } : {}), + ...(result.error ? { error: result.error } : {}) + })) }) : undefined remoteDelegation?.start() @@ -2355,6 +2579,16 @@ export function registerIpcHandlers( await executionTracker.drain() await onBeforeClearLocalData?.() assistantDatabase.clearAssistantData() + readyConversationQueues.clear() + preferredConversationQueueItems.clear() + reservedConversationQueueItems.clear() + preparingRequestConversations.clear() + rendererReadyConversationQueues.clear() + for (const timeout of queueDispatchTimers.values()) { + clearTimeout(timeout) + } + queueDispatchTimers.clear() + publishConversationQueueChange() } finally { executionPaused = false } @@ -2413,9 +2647,50 @@ export function registerIpcHandlers( throw new Error('本地数据维护期间暂不接受新任务') } const parsedInput = agentRequestSchema.parse(input) - if (activeRequests.has(parsedInput.requestId)) { + if ( + activeRequests.has(parsedInput.requestId) || + preparingRequestConversations.has(parsedInput.requestId) + ) { throw new Error('请求正在执行') } + const queuedItem = parsedInput.queueItemId + ? assistantDatabase.getConversationQueueItem( + parsedInput.queueItemId + ) + : undefined + if ( + parsedInput.queueItemId && + (!queuedItem || + queuedItem.source !== 'user' || + queuedItem.conversationId !== parsedInput.conversationId || + !assistantDatabase.isConversationUserQueueItemDispatching( + parsedInput.queueItemId + )) + ) { + throw new Error('待发送消息不存在或与当前对话不一致') + } + const reservationItemId = + reservedConversationQueueItems.get(parsedInput.conversationId) + if ( + [...activeRequestConversations.values()].some( + (conversationId) => + conversationId === parsedInput.conversationId + ) || + [...preparingRequestConversations.values()].some( + (conversationId) => + conversationId === parsedInput.conversationId + ) || + (parsedInput.queueItemId + ? reservationItemId !== parsedInput.queueItemId + : reservationItemId !== undefined) + ) { + throw new Error('当前对话已有执行中的请求') + } + preparingRequestConversations.set( + parsedInput.requestId, + parsedInput.conversationId + ) + try { const knowledgeLibraryIds = [ ...new Set(parsedInput.knowledgeLibraryIds) ] @@ -2549,6 +2824,41 @@ export function registerIpcHandlers( throw error } activeRequests.set(request.requestId, controller) + activeRequestConversations.set( + request.requestId, + request.conversationId + ) + if (parsedInput.queueItemId) { + const dispatchTimeout = queueDispatchTimers.get( + parsedInput.queueItemId + ) + if (dispatchTimeout) { + clearTimeout(dispatchTimeout) + queueDispatchTimers.delete(parsedInput.queueItemId) + } + if ( + reservedConversationQueueItems.get(request.conversationId) === + parsedInput.queueItemId + ) { + reservedConversationQueueItems.delete(request.conversationId) + } + try { + assistantDatabase.completeConversationUserQueueItem( + parsedInput.queueItemId + ) + publishConversationQueueChange(request.conversationId) + } catch (error) { + activeRequests.delete(request.requestId) + activeRequestConversations.delete(request.requestId) + assistantDatabase.updateTaskStatus( + request.requestId, + 'cancelled', + '待发送消息状态已变化' + ) + knowledgeGateway?.revoke(knowledgeCapabilityToken) + throw error + } + } const execution = (async () => { let completed = false @@ -3027,6 +3337,7 @@ export function registerIpcHandlers( } knowledgeGateway?.revoke(request.knowledgeCapabilityToken) activeRequests.delete(request.requestId) + activeRequestConversations.delete(request.requestId) const configReload = goodbuddyConfigService?.takePendingReload(request.requestId) ?? 'none' @@ -3035,9 +3346,15 @@ export function registerIpcHandlers( pendingGoodBuddyConfigReload = true } await flushGoodBuddyConfigReload().catch(() => undefined) + if (readyConversationQueues.has(request.conversationId)) { + void pumpConversationQueue(request.conversationId) + } } })() void trackExecution(execution) + } finally { + preparingRequestConversations.delete(parsedInput.requestId) + } }) registerHandler(ipcChannels.agentCancel, (event, input: unknown) => { @@ -3137,6 +3454,10 @@ export function registerIpcHandlers( 5 * 60_000 ) activeRequests.set(request.requestId, controller) + activeRequestConversations.set( + request.requestId, + request.conversationId + ) assistantDatabase.createTask({ id: request.requestId, projectId: request.projectId, @@ -3221,6 +3542,9 @@ export function registerIpcHandlers( } finally { clearTimeout(timeout) activeRequests.delete(request.requestId) + activeRequestConversations.delete(request.requestId) + readyConversationQueues.add(request.conversationId) + void pumpConversationQueue(request.conversationId) } } ) @@ -4150,9 +4474,185 @@ export function registerIpcHandlers( ipcChannels.conversationsDeleteLocal, (event, input: unknown) => { assertTrustedSender(event, window) - return assistantDatabase.deleteLocalConversation( - assistantIdSchema.parse(input) + const conversationId = assistantIdSchema.parse(input) + const queuedItems = + assistantDatabase.listConversationQueueItems(conversationId) + for (const item of queuedItems) { + if (item.source !== 'user') { + continue + } + const payloadJson = + assistantDatabase.getConversationUserQueuePayloadJson(item.id) + if (payloadJson) { + const queuedInput = + parseConversationQueueUserPayload(payloadJson) + for (const attachment of queuedInput.attachments) { + contextManager.remove(attachment.id) + } + } + } + const deleted = assistantDatabase.deleteLocalConversation( + conversationId ) + preferredConversationQueueItems.delete(conversationId) + readyConversationQueues.delete(conversationId) + rendererReadyConversationQueues.delete(conversationId) + publishConversationQueueChange(conversationId) + return deleted + } + ) + + registerHandler( + ipcChannels.conversationQueueList, + (event, input: unknown) => { + assertTrustedSender(event, window) + return assistantDatabase.listConversationQueueItems( + assistantIdSchema.optional().parse(input) + ) + } + ) + + registerHandler( + ipcChannels.conversationQueueEnqueueUser, + (event, input: unknown) => { + assertTrustedSender(event, window) + if (executionPaused || shuttingDown) { + throw new Error('本地数据维护期间暂不接受新消息') + } + const parsed = conversationQueueUserInputSchema.parse(input) + const serializedContexts = contextManager.serializeForQueue( + parsed.attachments.map((attachment) => attachment.id) + ) + const item = assistantDatabase.enqueueConversationUserInput({ + conversationId: parsed.conversationId, + label: parsed.prompt, + payloadJson: JSON.stringify({ + input: parsed, + serializedContexts + }) + }) + for (const attachment of parsed.attachments) { + contextManager.remove(attachment.id) + } + rendererReadyConversationQueues.add(item.conversationId) + publishConversationQueueChange(item.conversationId) + if (!isConversationExecuting(item.conversationId)) { + readyConversationQueues.add(item.conversationId) + setTimeout(() => { + void pumpConversationQueue(item.conversationId) + }, 0) + } + return item + } + ) + + registerHandler( + ipcChannels.conversationQueueRemove, + (event, input: unknown) => { + assertTrustedSender(event, window) + const itemId = assistantIdSchema.parse(input) + const item = assistantDatabase.getConversationQueueItem(itemId) + if (!item) { + throw new Error('待执行项不存在或状态已变化') + } + let queuedInput: ConversationQueueUserInput | undefined + if (item.source === 'user') { + const payloadJson = + assistantDatabase.getConversationUserQueuePayloadJson(item.id) + if (!payloadJson) { + throw new Error('待发送消息不存在或状态已变化') + } + queuedInput = + parseConversationQueueUserPayload(payloadJson) + } + assistantDatabase.cancelConversationQueueItem(itemId) + if ( + preferredConversationQueueItems.get(item.conversationId) === + itemId + ) { + preferredConversationQueueItems.delete(item.conversationId) + } + for (const attachment of queuedInput?.attachments ?? []) { + contextManager.remove(attachment.id) + } + publishConversationQueueChange(item.conversationId) + if (item.source === 'schedule') { + publishConversationChange() + } + if (readyConversationQueues.has(item.conversationId)) { + void pumpConversationQueue(item.conversationId) + } + } + ) + + registerHandler( + ipcChannels.conversationQueueInterruptAndRun, + (event, input: unknown) => { + assertTrustedSender(event, window) + const itemId = assistantIdSchema.parse(input) + const item = assistantDatabase.getConversationQueueItem(itemId) + if (!item) { + throw new Error('待执行项不存在或状态已变化') + } + preferredConversationQueueItems.set(item.conversationId, item.id) + readyConversationQueues.add(item.conversationId) + for (const [requestId, conversationId] of activeRequestConversations) { + if (conversationId === item.conversationId) { + activeRequests + .get(requestId) + ?.abort(new Error('用户中断当前回复并插入队列项')) + } + } + if (!isConversationExecuting(item.conversationId)) { + void pumpConversationQueue(item.conversationId, item.id) + } + } + ) + + registerHandler( + ipcChannels.conversationQueueReleaseUser, + (event, input: unknown) => { + assertTrustedSender(event, window) + const itemId = assistantIdSchema.parse(input) + const item = assistantDatabase.getConversationQueueItem(itemId) + const dispatchTimeout = queueDispatchTimers.get(itemId) + if (dispatchTimeout) { + clearTimeout(dispatchTimeout) + queueDispatchTimers.delete(itemId) + } + if (item) { + if ( + reservedConversationQueueItems.get(item.conversationId) === + itemId + ) { + reservedConversationQueueItems.delete(item.conversationId) + } + const payloadJson = + assistantDatabase.getConversationUserQueuePayloadJson(itemId) + if (payloadJson) { + const queuedInput = + parseConversationQueueUserPayload(payloadJson) + for (const attachment of queuedInput.attachments) { + contextManager.remove(attachment.id) + } + } + } + assistantDatabase.releaseConversationUserQueueItem(itemId) + if (item) { + readyConversationQueues.add(item.conversationId) + } + publishConversationQueueChange(item?.conversationId) + } + ) + + registerHandler( + ipcChannels.conversationQueueReady, + (event, input: unknown) => { + assertTrustedSender(event, window) + const conversationId = assistantIdSchema.parse(input) + rendererReadyConversationQueues.add(conversationId) + readyConversationQueues.add(conversationId) + void pumpConversationQueue(conversationId) } ) @@ -4397,6 +4897,7 @@ export function registerIpcHandlers( assertTrustedSender(event, window) assistantDatabase.removeSchedule(assistantIdSchema.parse(input)) publishConversationChange() + publishConversationQueueChange() }) registerHandler(ipcChannels.schedulesRunNow, (event, input: unknown) => { @@ -4404,22 +4905,14 @@ export function registerIpcHandlers( if (executionPaused || shuttingDown) { throw new Error('本地数据维护期间暂不接受新任务') } - const claim = assistantDatabase.claimScheduleNow( + const item = assistantDatabase.queueScheduleNow( assistantIdSchema.parse(input) ) - const execution = (async () => { - const result = await executeTaskWork({ - origin: 'schedule', - schedule: claim.schedule, - scheduleRunId: claim.runId - }) - assistantDatabase.completeScheduleRun( - claim.runId, - result.status - ) - publishConversationChange() - })() - void trackExecution(execution).catch(() => undefined) + publishConversationQueueChange(item.conversationId) + if (!isConversationExecuting(item.conversationId)) { + readyConversationQueues.add(item.conversationId) + void pumpConversationQueue(item.conversationId) + } }) registerHandler(ipcChannels.heartbeatsList, (event, input: unknown) => { @@ -5649,6 +6142,10 @@ export function registerIpcHandlers( shuttingDown = true removeBrowserStateListener?.() clearInterval(scheduleInterval) + for (const timeout of queueDispatchTimers.values()) { + clearTimeout(timeout) + } + queueDispatchTimers.clear() window.removeListener('maximize', notifyMaximizedChanged) window.removeListener('unmaximize', notifyMaximizedChanged) abortActiveRequests('应用正在退出') diff --git a/src/preload/index.ts b/src/preload/index.ts index ba61c06..23d104c 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -614,7 +614,10 @@ const desktopApi: DesktopApi = { conversationId ) as Promise, onChanged: (listener) => { - const handler = (): void => listener() + const handler = ( + _event: Electron.IpcRendererEvent, + conversationId?: string + ): void => listener(conversationId) ipcRenderer.on(ipcChannels.conversationsChanged, handler) return () => ipcRenderer.removeListener( @@ -623,6 +626,66 @@ const desktopApi: DesktopApi = { ) } }, + conversationQueue: { + list: (conversationId?: string) => + ipcRenderer.invoke( + ipcChannels.conversationQueueList, + conversationId + ), + enqueueUser: (input) => + ipcRenderer.invoke( + ipcChannels.conversationQueueEnqueueUser, + input + ), + remove: async (itemId: string) => { + await ipcRenderer.invoke( + ipcChannels.conversationQueueRemove, + itemId + ) + }, + interruptAndRun: async (itemId: string) => { + await ipcRenderer.invoke( + ipcChannels.conversationQueueInterruptAndRun, + itemId + ) + }, + releaseUser: async (itemId: string) => { + await ipcRenderer.invoke( + ipcChannels.conversationQueueReleaseUser, + itemId + ) + }, + ready: async (conversationId: string) => { + await ipcRenderer.invoke( + ipcChannels.conversationQueueReady, + conversationId + ) + }, + onChanged: (listener) => { + const handler = ( + _event: Electron.IpcRendererEvent, + conversationId?: string + ): void => listener(conversationId) + ipcRenderer.on(ipcChannels.conversationQueueChanged, handler) + return () => + ipcRenderer.removeListener( + ipcChannels.conversationQueueChanged, + handler + ) + }, + onDispatch: (listener) => { + const handler = ( + _event: Electron.IpcRendererEvent, + dispatch: Parameters[0] + ): void => listener(dispatch) + ipcRenderer.on(ipcChannels.conversationQueueDispatch, handler) + return () => + ipcRenderer.removeListener( + ipcChannels.conversationQueueDispatch, + handler + ) + } + }, workspace: { getChanges: (projectId: string) => ipcRenderer.invoke( diff --git a/src/preload/preload-sandbox.test.ts b/src/preload/preload-sandbox.test.ts index 34ae33a..c60e1ce 100644 --- a/src/preload/preload-sandbox.test.ts +++ b/src/preload/preload-sandbox.test.ts @@ -95,4 +95,18 @@ describe('sandboxed preload', () => { expect(source).toContain('ipcChannels.knowledgeTaskCancel') expect(source).toContain('ipcChannels.knowledgeTaskRetry') }) + + it('exposes only explicit Conversation queue operations and listeners', () => { + const source = readFileSync( + join(process.cwd(), 'src', 'preload', 'index.ts'), + 'utf8' + ) + expect(source).toContain('conversationQueue: {') + expect(source).toContain('ipcChannels.conversationQueueEnqueueUser') + expect(source).toContain( + 'ipcChannels.conversationQueueInterruptAndRun' + ) + expect(source).toContain('ipcChannels.conversationQueueDispatch') + expect(source).toContain('listener(conversationId)') + }) }) diff --git a/src/renderer/src/App.test.tsx b/src/renderer/src/App.test.tsx index 5970ff2..ace9a63 100644 --- a/src/renderer/src/App.test.tsx +++ b/src/renderer/src/App.test.tsx @@ -12,6 +12,7 @@ import type { AgentEvent, BrowserLiveState, ContextAttachment, + ConversationQueueDispatch, DesktopApi } from '../../shared/contracts' import type { ApplicationSettings } from '../../shared/application-settings-contracts' @@ -74,6 +75,14 @@ let fileSelectionProgressListener: let newConversationListener: (() => void) | undefined let maximizedChangedListener: ((maximized: boolean) => void) | undefined let beforeQuitListener: (() => Promise) | undefined +let conversationQueueDispatchListener: + | ((dispatch: ConversationQueueDispatch) => void) + | undefined +let conversationQueueChangeListener: + | Parameters< + DesktopApi['conversationQueue']['onChanged'] + >[0] + | undefined const removeMaximizedChangedListener = vi.fn() const run = vi.fn() const modelProfileId = '00000000-0000-4000-8000-000000000001' @@ -336,6 +345,38 @@ const api: DesktopApi = { deleteLocal: vi.fn(async () => true), onChanged: vi.fn(() => () => undefined) }, + conversationQueue: { + list: vi.fn(async () => []), + enqueueUser: vi.fn(async (input) => { + const item = { + id: crypto.randomUUID(), + conversationId: input.conversationId, + source: 'user' as const, + label: input.prompt, + createdAt: '2026-07-31T00:00:00.000Z' + } + queueMicrotask(() => + conversationQueueDispatchListener?.({ item, input }) + ) + return item + }), + remove: vi.fn(async () => {}), + interruptAndRun: vi.fn(async () => {}), + releaseUser: vi.fn(async () => {}), + ready: vi.fn(async () => {}), + onChanged: vi.fn((listener) => { + conversationQueueChangeListener = listener + return () => { + conversationQueueChangeListener = undefined + } + }), + onDispatch: vi.fn((listener) => { + conversationQueueDispatchListener = listener + return () => { + conversationQueueDispatchListener = undefined + } + }) + }, workspace: { getChanges: vi.fn(async () => ({ rootPath: 'C:\\Workspace', @@ -755,6 +796,18 @@ function selectComposerOption( fireEvent.click(option) } +function selectProjectOption(projectName: string): void { + fireEvent.click(screen.getByRole('button', { name: '当前项目' })) + const menu = screen.getByRole('menu', { name: '当前项目' }) + const option = within(menu) + .getByText(projectName, { selector: 'b' }) + .closest('[role="menuitemradio"]') + if (!option) { + throw new Error(`Missing project option: ${projectName}`) + } + fireEvent.click(option) +} + function deferred(): { promise: Promise resolve: (value: T) => void @@ -785,6 +838,54 @@ describe('App', () => { vi.mocked(api.conversations.onChanged) .mockReset() .mockReturnValue(() => undefined) + conversationQueueChangeListener = undefined + conversationQueueDispatchListener = undefined + vi.mocked(api.conversationQueue.list) + .mockReset() + .mockResolvedValue([]) + vi.mocked(api.conversationQueue.enqueueUser) + .mockReset() + .mockImplementation(async (input) => { + const item = { + id: crypto.randomUUID(), + conversationId: input.conversationId, + source: 'user' as const, + label: input.prompt, + createdAt: '2026-07-31T00:00:00.000Z' + } + queueMicrotask(() => + conversationQueueDispatchListener?.({ item, input }) + ) + return item + }) + vi.mocked(api.conversationQueue.remove) + .mockReset() + .mockResolvedValue() + vi.mocked(api.conversationQueue.interruptAndRun) + .mockReset() + .mockResolvedValue() + vi.mocked(api.conversationQueue.releaseUser) + .mockReset() + .mockResolvedValue() + vi.mocked(api.conversationQueue.ready) + .mockReset() + .mockResolvedValue() + vi.mocked(api.conversationQueue.onChanged) + .mockReset() + .mockImplementation((listener) => { + conversationQueueChangeListener = listener + return () => { + conversationQueueChangeListener = undefined + } + }) + vi.mocked(api.conversationQueue.onDispatch) + .mockReset() + .mockImplementation((listener) => { + conversationQueueDispatchListener = listener + return () => { + conversationQueueDispatchListener = undefined + } + }) vi.mocked(api.tasks.list).mockReset().mockResolvedValue([]) vi.mocked(api.schedules.list).mockReset().mockResolvedValue([]) api.channels = undefined @@ -908,6 +1009,143 @@ describe('App', () => { ).not.toBeInTheDocument() }) + it('renders the unified user and Scheduled Task queue above the Composer', async () => { + const conversationId = + '00000000-0000-4000-8000-000000000951' + vi.mocked(api.conversations.list).mockResolvedValue([ + { + id: conversationId, + projectId, + title: '排队对话', + updatedAt: Date.now(), + messages: [] + } + ]) + vi.mocked(api.conversationQueue.list).mockResolvedValue([ + { + id: '00000000-0000-4000-8000-000000000952', + conversationId, + source: 'schedule', + label: '每日汇总', + createdAt: '2026-08-20T09:00:00.000Z' + }, + { + id: '00000000-0000-4000-8000-000000000953', + conversationId, + source: 'user', + label: '补充说明', + createdAt: '2026-08-20T09:01:00.000Z' + } + ]) + + render() + + const queue = await screen.findByRole('region', { + name: '对话待发送队列' + }) + expect(queue.parentElement).toHaveClass('composer-wrap') + expect(queue.closest('.composer')).toBeNull() + expect(queue.nextElementSibling).toHaveClass('composer') + expect(within(queue).queryByText('待发送（2）')).not.toBeInTheDocument() + expect(within(queue).getByText('每日汇总')).toBeInTheDocument() + expect(within(queue).getByText('补充说明')).toBeInTheDocument() + fireEvent.click( + within(queue).getByRole('button', { + name: '立即运行“每日汇总”' + }) + ) + await waitFor(() => + expect( + api.conversationQueue.interruptAndRun + ).toHaveBeenCalledWith( + '00000000-0000-4000-8000-000000000952' + ) + ) + fireEvent.click( + within(queue).getByRole('button', { + name: '从待发送队列删除“补充说明”' + }) + ) + await waitFor(() => + expect(api.conversationQueue.remove).toHaveBeenCalledWith( + '00000000-0000-4000-8000-000000000953' + ) + ) + }) + + it('preserves scoped queue updates that arrive during another refresh', async () => { + const firstConversationId = + '00000000-0000-4000-8000-000000000954' + const secondConversationId = + '00000000-0000-4000-8000-000000000955' + vi.mocked(api.conversations.list).mockResolvedValue([ + { + id: firstConversationId, + projectId, + title: '第一条排队对话', + updatedAt: Date.now(), + messages: [] + }, + { + id: secondConversationId, + projectId, + title: '第二条排队对话', + updatedAt: Date.now() - 1, + messages: [] + } + ]) + const firstRefresh = deferred< + Awaited< + ReturnType + > + >() + vi.mocked(api.conversationQueue.list).mockImplementation( + async (conversationId) => { + if (conversationId === firstConversationId) { + return firstRefresh.promise + } + return [] + } + ) + + render() + await waitFor(() => + expect(conversationQueueChangeListener).toBeDefined() + ) + act(() => { + conversationQueueChangeListener?.(firstConversationId) + }) + await waitFor(() => + expect(api.conversationQueue.list).toHaveBeenCalledWith( + firstConversationId + ) + ) + await act(async () => { + conversationQueueChangeListener?.(secondConversationId) + await new Promise((resolve) => window.setTimeout(resolve, 0)) + }) + act(() => { + firstRefresh.resolve([ + { + id: '00000000-0000-4000-8000-000000000956', + conversationId: firstConversationId, + source: 'user', + label: '不能被后续刷新丢弃', + createdAt: '2026-08-20T09:03:00.000Z' + } + ]) + }) + + expect( + await screen.findByText('不能被后续刷新丢弃') + ).toBeInTheDocument() + await waitFor(() => + expect(api.conversationQueue.list).toHaveBeenCalledWith( + secondConversationId + ) + ) + }) + it('discovers product Tasks through their Conversation without exposing Runs', async () => { const conversationId = '00000000-0000-4000-8000-000000000821' @@ -969,8 +1207,10 @@ describe('App', () => { const taskChild = taskTitle.closest('button') expect(taskChild).not.toBeNull() expect( - taskChild?.querySelector('.conversation-task-child__icon') - ).toHaveClass('conversation-task-child__icon--idle') + taskChild?.querySelector('.task-status-dot') + ).toHaveClass('task-status-dot--idle') + expect(taskChild?.querySelector('.lucide-clock-fading')) + .not.toBeInTheDocument() expect( taskChild?.querySelector('.conversation-task-child__meta') ).toHaveTextContent('Execute · 每周 · 空闲') @@ -982,6 +1222,81 @@ describe('App', () => { expect(within(taskRegion).getByText('Execute')).toBeInTheDocument() }) + it('uses shared status dots for Conversation Task children', async () => { + const conversationId = + '00000000-0000-4000-8000-000000000824' + const conversation: ConversationSnapshot = { + id: conversationId, + projectId, + title: '任务状态会话', + updatedAt: Date.now(), + messages: [] + } + const tasks: AssistantTask[] = [ + { + id: '00000000-0000-4000-8000-000000000825', + projectId, + conversationId, + title: '已完成任务', + instructions: '检查完成状态', + origin: 'schedule', + status: 'completed', + createdAt: '2026-08-19T03:00:00.000Z' + }, + { + id: '00000000-0000-4000-8000-000000000826', + projectId, + conversationId, + title: '运行中任务', + instructions: '检查运行状态', + origin: 'schedule', + status: 'running', + createdAt: '2026-08-19T02:00:00.000Z' + }, + { + id: '00000000-0000-4000-8000-000000000827', + projectId, + conversationId, + title: '失败任务', + instructions: '检查失败状态', + origin: 'schedule', + status: 'failed', + createdAt: '2026-08-19T01:00:00.000Z' + } + ] + vi.mocked(api.conversations.list).mockResolvedValue([ + conversation + ]) + vi.mocked(api.tasks.list).mockResolvedValue(tasks) + + render() + + fireEvent.click( + await screen.findByLabelText( + '展开或折叠“任务状态会话”中的 3 个任务' + ) + ) + for (const [title, status, label] of [ + ['已完成任务', 'completed', '已完成'], + ['运行中任务', 'running', '运行中'], + ['失败任务', 'failed', '失败'] + ] as const) { + const taskTitle = await screen.findByText(title, { + selector: '.conversation-task-child__title' + }) + const taskChild = taskTitle.closest('button') + expect( + taskChild?.querySelector('.task-status-dot') + ).toHaveClass(`task-status-dot--${status}`) + expect( + taskChild?.querySelector('.conversation-task-child__meta') + ).toHaveTextContent(label) + } + expect( + document.querySelector('.conversation-task-child__icon') + ).not.toBeInTheDocument() + }) + it('routes scheduled Task approvals to the associated Conversation', async () => { const conversationId = '00000000-0000-4000-8000-000000000831' @@ -3436,6 +3751,86 @@ describe('App', () => { expect(runtimeButton).toBeEnabled() }) + it('queues another ordinary message while the Conversation is running', async () => { + render() + + fireEvent.change(screen.getByLabelText('向 GoodBuddy 提问'), { + target: { value: '第一条长消息' } + }) + fireEvent.click(await screen.findByLabelText('发送')) + await waitFor(() => expect(run).toHaveBeenCalledOnce()) + + const secondItem = { + id: '00000000-0000-4000-8000-000000000941', + conversationId: run.mock.calls[0]![0].conversationId, + source: 'user' as const, + label: '第二条排队消息', + createdAt: '2026-08-20T09:01:00.000Z' + } + vi.mocked(api.conversationQueue.enqueueUser) + .mockResolvedValueOnce(secondItem) + fireEvent.change(screen.getByLabelText('向 GoodBuddy 提问'), { + target: { value: secondItem.label } + }) + fireEvent.click( + await screen.findByLabelText('加入待发送队列') + ) + + await waitFor(() => + expect(api.conversationQueue.enqueueUser).toHaveBeenCalledTimes(2) + ) + expect(run).toHaveBeenCalledTimes(1) + expect( + screen.queryByText(/当前对话已有任务正在运行/u) + ).not.toBeInTheDocument() + + const firstRequest = run.mock.calls[0]![0] + act(() => { + agentListener?.({ + requestId: firstRequest.requestId, + type: 'done' + }) + }) + const secondInput = vi.mocked( + api.conversationQueue.enqueueUser + ).mock.calls[1]![0] + act(() => { + conversationQueueDispatchListener?.({ + item: secondItem, + input: secondInput + }) + }) + + await waitFor(() => expect(run).toHaveBeenCalledTimes(2)) + expect(run.mock.calls[1]![0]).toMatchObject({ + queueItemId: secondItem.id, + prompt: secondItem.label + }) + }) + + it('restores a queued message when Agent preflight rejects it', async () => { + run.mockRejectedValueOnce(new Error('Runtime 暂不可用')) + render() + + fireEvent.change(screen.getByLabelText('向 GoodBuddy 提问'), { + target: { value: '需要稍后重试的消息' } + }) + fireEvent.click(await screen.findByLabelText('发送')) + + await waitFor(() => + expect(api.conversationQueue.releaseUser).toHaveBeenCalledOnce() + ) + expect( + await screen.findByText('Runtime 暂不可用') + ).toBeInTheDocument() + expect( + screen.queryByText('需要稍后重试的消息') + ).not.toBeInTheDocument() + expect( + screen.queryByLabelText('停止生成') + ).not.toBeInTheDocument() + }) + it('keeps sent documents and images in conversation history', async () => { const documentAttachment = { id: '00000000-0000-4000-8000-000000000301', @@ -4007,9 +4402,7 @@ describe('App', () => { await waitFor(() => expect(api.workspace.getChanges).toHaveBeenCalledWith(projectId) ) - fireEvent.change(screen.getByLabelText('当前项目'), { - target: { value: secondProject.id } - }) + selectProjectOption(secondProject.name) await waitFor(() => expect(api.workspace.getChanges).toHaveBeenCalledWith( secondProject.id @@ -4296,18 +4689,16 @@ describe('App', () => { render() - expect(await screen.findByLabelText('当前项目')).toHaveValue( - secondProject.id - ) + expect( + await screen.findByRole('button', { name: '当前项目' }) + ).toHaveTextContent(secondProject.name) expect( screen.getByRole('button', { name: '工作模式：Execute · 受控执行' }) ).toBeEnabled() - fireEvent.change(screen.getByLabelText('当前项目'), { - target: { value: project.id } - }) + selectProjectOption(project.name) await waitFor(() => expect( localStorage.getItem('goodbuddy.active-project.v1') @@ -4333,10 +4724,8 @@ describe('App', () => { ]) render() - await screen.findByRole('option', { name: '微信 ClawBot' }) - fireEvent.change(await screen.findByLabelText('当前项目'), { - target: { value: channelProject.id } - }) + await screen.findByRole('button', { name: '当前项目' }) + selectProjectOption(channelProject.name) expect( screen.queryByRole('button', { name: /新建对话/u }) @@ -4482,9 +4871,8 @@ describe('App', () => { render() const weixinProject = channelProjects[0]! - fireEvent.change(await screen.findByLabelText('当前项目'), { - target: { value: weixinProject.id } - }) + await screen.findByRole('button', { name: '当前项目' }) + selectProjectOption(weixinProject.name) fireEvent.click( await screen.findByRole('button', { name: '打开设置' }) ) @@ -4646,10 +5034,8 @@ describe('App', () => { ]) render() - await screen.findByRole('option', { name: '微信 ClawBot' }) - fireEvent.change(await screen.findByLabelText('当前项目'), { - target: { value: channelProject.id } - }) + await screen.findByRole('button', { name: '当前项目' }) + selectProjectOption(channelProject.name) expect( screen.getAllByRole('button', { @@ -4672,9 +5058,9 @@ describe('App', () => { render() - expect(await screen.findByLabelText('当前项目')).toHaveValue( - project.id - ) + expect( + await screen.findByRole('button', { name: '当前项目' }) + ).toHaveTextContent(project.name) await waitFor(() => expect( localStorage.getItem('goodbuddy.active-project.v1') @@ -6360,8 +6746,8 @@ describe('App', () => { project.name ) ) - expect(screen.getByLabelText('当前项目')).toHaveValue( - secondProject.id + expect(screen.getByRole('button', { name: '当前项目' })).toHaveTextContent( + secondProject.name ) }) @@ -7230,9 +7616,7 @@ describe('App', () => { await screen.findByRole('tab', { name: '心跳计划' }) ) expect(await screen.findAllByText('旧项目心跳')).not.toHaveLength(0) - fireEvent.change(screen.getByLabelText('当前项目'), { - target: { value: secondProject.id } - }) + selectProjectOption(secondProject.name) fireEvent.click( screen.getByRole('button', { name: '智能心跳' }) ) diff --git a/src/renderer/src/App.tsx b/src/renderer/src/App.tsx index 13c7744..b99bfca 100644 --- a/src/renderer/src/App.tsx +++ b/src/renderer/src/App.tsx @@ -14,7 +14,6 @@ import { HeartPulse, Info, Library, - ListTodo, LoaderCircle, Maximize2, MessageSquarePlus, @@ -64,6 +63,8 @@ import type { BrowserLiveState, ContextAttachment, ContextFileSelectionProgress, + ConversationQueueDispatch, + ConversationQueueUserInput, KnowledgeSearchReference, KnowledgeSnapshot, RuntimeCustomizationSettings, @@ -104,6 +105,7 @@ import type { AssistantTask, TokenUsageSummary, ConversationMessage, + ConversationQueueItem, ConversationSnapshot, ConversationAttachment, ConversationContextCompressionMarker, @@ -165,6 +167,7 @@ import { type CustomTaskDestination } from './CustomTaskDialog' import { ConversationTaskStrip } from './ConversationTaskStrip' +import { ConversationInputQueue } from './ConversationInputQueue' import { OverflowMarquee } from './OverflowMarquee' import { findTaskSchedule } from './TaskScheduleActions' import type { SettingsCategoryId } from './settings-categories' @@ -242,6 +245,29 @@ const recentCachedConversations = 5 const maximumCachedWorkspaceViews = 4 const recentCachedWorkspaceViews = 3 +function sameConversationQueueItems( + current: ConversationQueueItem[], + next: ConversationQueueItem[] +): boolean { + return ( + current.length === next.length && + current.every((item, index) => { + const candidate = next[index] + return ( + candidate !== undefined && + item.id === candidate.id && + item.conversationId === candidate.conversationId && + item.source === candidate.source && + item.label === candidate.label && + item.createdAt === candidate.createdAt && + item.scheduleRunId === candidate.scheduleRunId && + item.scheduleId === candidate.scheduleId && + item.taskId === candidate.taskId + ) + }) + ) +} + type AppNotification = { id: string message: string @@ -1849,6 +1875,13 @@ function App(): React.JSX.Element { const [assistantSchedules, setAssistantSchedules] = useState< AssistantSchedule[] >([]) + const [conversationQueueItems, setConversationQueueItems] = useState< + ConversationQueueItem[] + >([]) + const dispatchedConversationQueueItems = useRef(new Set()) + const conversationQueueDispatchRef = useRef< + (dispatch: ConversationQueueDispatch) => void + >(() => undefined) const [selectedAssistantTaskId, setSelectedAssistantTaskId] = useState() const [expandedTaskConversationIds, setExpandedTaskConversationIds] = @@ -2307,6 +2340,32 @@ function App(): React.JSX.Element { requestAnimationFrame(() => sidebarToggleRef.current?.focus()) }, []) + useEffect(() => { + if (!conversationStoreReady) { + return + } + const conversationIds = [ + ...new Set( + conversationsRef.current + .filter((conversation) => !conversation.remote) + .map((conversation) => conversation.id) + ) + ] + void Promise.all( + conversationIds.map((conversationId) => + window.goodbuddy.conversationQueue.ready(conversationId) + ) + ).catch(() => { + notify({ + tone: 'error', + message: tRef.current( + 'notices.conversationQueueResumeFailed' + ), + dedupeKey: 'conversation-queue-resume' + }) + }) + }, [conversationStoreReady]) + useEffect(() => { const sweep = (): void => { const now = Date.now() @@ -4095,6 +4154,19 @@ function App(): React.JSX.Element { }) activeRuns.current.delete(event.requestId) setConversationActivity(run.conversationId, false) + requestAnimationFrame(() => { + void window.goodbuddy.conversationQueue + .ready(run.conversationId) + .catch(() => { + notify({ + tone: 'error', + message: tRef.current( + 'notices.conversationQueueResumeFailed' + ), + dedupeKey: 'conversation-queue-resume' + }) + }) + }) flushConversationPersistenceAfterRenderRef.current = true } }, @@ -4357,6 +4429,123 @@ function App(): React.JSX.Element { } }, [conversationStoreReady]) + useEffect(() => { + let active = true + let refreshInFlight = false + let refreshQueued = false + let refreshTimer: number | undefined + let refreshAll = false + const pendingConversationIds = new Set() + const scheduleRefresh = (): void => { + if (refreshTimer !== undefined || refreshInFlight) { + refreshQueued = true + return + } + refreshTimer = window.setTimeout(() => { + refreshTimer = undefined + refreshQueued = false + const conversationIds = refreshAll + ? undefined + : [...pendingConversationIds] + refreshAll = false + pendingConversationIds.clear() + refresh(conversationIds) + }, 0) + } + const refresh = (conversationIds?: string[]): void => { + refreshInFlight = true + const affectedConversationIds = new Set( + conversationIds ?? [] + ) + const reads = conversationIds + ? conversationIds.map(async (conversationId) => ({ + conversationId, + items: + await window.goodbuddy.conversationQueue.list( + conversationId + ) + })) + : [ + window.goodbuddy.conversationQueue + .list() + .then((items) => ({ + conversationId: undefined, + items + })) + ] + void Promise.all(reads) + .then((results) => { + if (!active) { + return + } + setConversationQueueItems((current) => { + const allItems = results.find( + (result) => result.conversationId === undefined + )?.items + const next = allItems + ? allItems + : [ + ...current.filter( + (item) => + !affectedConversationIds.has( + item.conversationId + ) + ), + ...results.flatMap((result) => result.items) + ].sort( + (left, right) => + left.createdAt.localeCompare(right.createdAt) || + left.id.localeCompare(right.id) + ) + return sameConversationQueueItems(current, next) + ? current + : next + }) + }) + .catch(() => { + if (active) { + notify({ + tone: 'error', + message: tRef.current( + 'notices.conversationQueueReadFailed' + ), + dedupeKey: 'conversation-queue-read' + }) + } + }) + .finally(() => { + refreshInFlight = false + if (active && refreshQueued) { + refreshQueued = false + scheduleRefresh() + } + }) + } + const queueRefresh = (conversationId?: string): void => { + if (conversationId) { + pendingConversationIds.add(conversationId) + } else { + refreshAll = true + } + scheduleRefresh() + } + refresh() + const remove = + window.goodbuddy.conversationQueue.onChanged(queueRefresh) + const removeDispatch = + window.goodbuddy.conversationQueue.onDispatch((dispatch) => + conversationQueueDispatchRef.current(dispatch) + ) + return () => { + active = false + if (refreshTimer !== undefined) { + window.clearTimeout(refreshTimer) + } + remove() + removeDispatch() + } + }, []) + useEffect(() => { activityRecordsRef.current = activityRecords const timeout = window.setTimeout(() => { @@ -5591,38 +5780,65 @@ function App(): React.JSX.Element { }) }, []) - const submit = async (): Promise => { + const submit = async ( + queuedDispatch?: ConversationQueueDispatch + ): Promise => { + const queuedInput = queuedDispatch?.input + const releaseQueuedItem = async (): Promise => { + if (!queuedDispatch) { + return + } + try { + await window.goodbuddy.conversationQueue.releaseUser( + queuedDispatch.item.id + ) + } catch { + notify({ + tone: 'error', + message: t('notices.conversationQueueReleaseFailed') + }) + } + } + const conversationSnapshot = queuedInput + ? conversationsRef.current.find( + (conversation) => + conversation.id === queuedInput.conversationId + ) + : activeConversation const command = + !queuedInput && activeRuntimeSelection?.provider === 'opencode' ? runtimeNativeSnapshot?.commands.find( (candidate) => candidate.id === selectedRuntimeCommand ) : undefined - const commandArguments = input.trim() - const prompt = command + const commandArguments = queuedInput ? '' : input.trim() + const prompt = queuedInput?.prompt ?? (command ? `/${command.name}${ commandArguments ? ` ${commandArguments}` : '' }` - : commandArguments - if (!prompt || !activeConversation) { + : commandArguments) + if (!prompt || !conversationSnapshot) { + await releaseQueuedItem() return } - if (selectingContextFilesRef.current) { + if (!queuedInput && selectingContextFilesRef.current) { notify({ tone: 'info', message: t('composer.attachmentProgress.waitBeforeSending') }) return } - if (activeConversation.remote) { + if (conversationSnapshot.remote) { notify({ tone: 'info', message: t('notices.remoteConversationReadOnly') }) + await releaseQueuedItem() return } - if (!runtime) { + if (!queuedInput && !runtime) { notify({ tone: 'info', message: t('runtime.loadingRetry') @@ -5630,8 +5846,9 @@ function App(): React.JSX.Element { return } if ( - runtimeSwitching || - runtimeStatusKey !== activeRuntimeSelectionKey + !queuedInput && + (runtimeSwitching || + runtimeStatusKey !== activeRuntimeSelectionKey) ) { notify({ tone: 'info', @@ -5639,73 +5856,145 @@ function App(): React.JSX.Element { }) return } - if (!runtime.available) { - return - } - if ( - preparingConversations.current.has(activeConversation.id) || - [...activeRuns.current.values()].some( - (run) => run.conversationId === activeConversation.id - ) - ) { - notify({ - tone: 'info', - message: t('notices.conversationAlreadyRunning') - }) + if (!queuedInput && !runtime?.available) { return } const requestId = crypto.randomUUID() - const conversationId = activeConversation.id - const attachmentSnapshot = attachments.slice(0, 8) - const historySnapshot = activeConversation.messages + const conversationId = conversationSnapshot.id + const attachmentSnapshot = ( + queuedInput?.attachments ?? attachments + ).slice(0, 8) + const historySnapshot = conversationSnapshot.messages const retainedHistorySnapshot = historySnapshot .filter( (message) => message.state === 'complete' && message.content.trim() ) .slice(-500) - const projectIdSnapshot = activeProjectId || undefined + const projectIdSnapshot = queuedInput + ? queuedInput.projectId + : activeProjectId || undefined const knowledgeRetrievalModeSnapshot = - activeConversation.knowledgeRetrievalMode ?? 'auto' - const runtimeSelectionSnapshot = activeRuntimeSelection + queuedInput?.knowledgeRetrievalMode ?? + conversationSnapshot.knowledgeRetrievalMode ?? + 'auto' + const runtimeSelectionSnapshot = + queuedInput?.runtimeSelection ?? activeRuntimeSelection if (!runtimeSelectionSnapshot) { notify({ tone: 'info', message: t('runtime.notSelected') }) + await releaseQueuedItem() return } const runtimeControlSnapshot: RuntimeControl | undefined = - runtimeSelectionSnapshot.provider === 'opencode' && - (selectedRuntimeAgent || command) - ? { - provider: 'opencode', - ...(selectedRuntimeAgent - ? { agent: selectedRuntimeAgent } - : {}), - ...(command - ? { - command: { - name: command.name, - arguments: commandArguments - } - } - : {}) - } - : runtimeSelectionSnapshot.provider === 'continue' && - selectedContinuePreset + queuedInput + ? queuedInput.runtimeControl + : runtimeSelectionSnapshot.provider === 'opencode' && + (selectedRuntimeAgent || command) ? { - provider: 'continue', - presetId: selectedContinuePreset + provider: 'opencode', + ...(selectedRuntimeAgent + ? { agent: selectedRuntimeAgent } + : {}), + ...(command + ? { + command: { + name: command.name, + arguments: commandArguments + } + } + : {}) } - : undefined - const selectedExpertSnapshot = - runtime.capability === 'image-generation' ? '' : selectedExpertId - const workModeSnapshot = effectiveWorkMode + : runtimeSelectionSnapshot.provider === 'continue' && + selectedContinuePreset + ? { + provider: 'continue', + presetId: selectedContinuePreset + } + : undefined + const selectedExpertSnapshot = queuedInput + ? queuedInput.teamMode + ? 'team' + : queuedInput.expertId ?? '' + : runtime?.capability === 'image-generation' + ? '' + : selectedExpertId + const workModeSnapshot = + normalizeInteractiveWorkMode( + queuedInput?.workMode ?? effectiveWorkMode + ) + const knowledgeLibraryIdsSnapshot = + queuedInput?.knowledgeLibraryIds ?? enabledKnowledgeLibraryIds + const smartRoutingSnapshot = + queuedInput?.smartRouting ?? + (!queuedInput && + runtime?.capability !== 'image-generation' && + runtimeSettings?.subagentSmartRoutingEnabled === true && + !selectedExpertSnapshot && + supportsSubagentSmartRouting(workModeSnapshot) + ? true + : undefined) + + if (!queuedInput) { + const queueInput: ConversationQueueUserInput = { + conversationId, + ...(projectIdSnapshot ? { projectId: projectIdSnapshot } : {}), + runtimeSelection: runtimeSelectionSnapshot, + ...(runtimeControlSnapshot + ? { runtimeControl: runtimeControlSnapshot } + : {}), + ...(selectedExpertSnapshot && + selectedExpertSnapshot !== 'team' + ? { expertId: selectedExpertSnapshot } + : {}), + ...(selectedExpertSnapshot === 'team' + ? { teamMode: true } + : {}), + ...(smartRoutingSnapshot + ? { smartRouting: true } + : {}), + workMode: workModeSnapshot, + includeMemoryContext: !command, + prompt, + attachments: attachmentSnapshot, + knowledgeLibraryIds: knowledgeLibraryIdsSnapshot, + knowledgeRetrievalMode: knowledgeRetrievalModeSnapshot + } + try { + await window.goodbuddy.conversationQueue.enqueueUser(queueInput) + setComposerMenuOpen(undefined) + setRuntimeMenuOpen(false) + setInput('') + updateAttachments([]) + if (command) { + setSelectedRuntimeCommand('') + } + } catch (reason) { + notify({ + tone: 'error', + message: + reason instanceof Error + ? reason.message + : t('notices.sendFailed') + }) + } + return + } + + if ( + dispatchedConversationQueueItems.current.has( + queuedDispatch.item.id + ) + ) { + return + } + dispatchedConversationQueueItems.current.add( + queuedDispatch.item.id + ) setComposerMenuOpen(undefined) setRuntimeMenuOpen(false) preparingConversations.current.add(conversationId) setConversationActivity(conversationId, true) - setInput('') - updateAttachments([]) const userMessage: Message = { id: crypto.randomUUID(), role: 'user', @@ -5733,13 +6022,10 @@ function App(): React.JSX.Element { : conversation ) ) - const memoryContext = - runtime.capability === 'image-generation' - ? '' - : buildMemoryContext(assistantMemories) - const executionPrompt = command - ? prompt - : memoryContext + const memoryContext = queuedInput.includeMemoryContext + ? buildMemoryContext(assistantMemories) + : '' + const executionPrompt = memoryContext ? `${prompt}\n\n${memoryContext}` : prompt const assistantMessage: Message = { @@ -5810,6 +6096,7 @@ function App(): React.JSX.Element { requestId, conversationId, projectId: projectIdSnapshot, + queueItemId: queuedDispatch.item.id, runtimeSelection: runtimeSelectionSnapshot, runtimeControl: runtimeControlSnapshot, expertId: @@ -5817,22 +6104,16 @@ function App(): React.JSX.Element { ? selectedExpertSnapshot : undefined, teamMode: selectedExpertSnapshot === 'team', - smartRouting: - runtime.capability !== 'image-generation' && - runtimeSettings?.subagentSmartRoutingEnabled === true && - !selectedExpertSnapshot && - supportsSubagentSmartRouting(workModeSnapshot) - ? true - : undefined, + smartRouting: smartRoutingSnapshot, workMode: workModeSnapshot, prompt: executionPrompt, - knowledgeLibraryIds: enabledKnowledgeLibraryIds, + knowledgeLibraryIds: knowledgeLibraryIdsSnapshot, knowledgeRetrievalMode: knowledgeRetrievalModeSnapshot, contextIds: attachmentSnapshot.map( (attachment) => attachment.id ), contextCompressionState: - activeConversation.contextCompressionState, + conversationSnapshot.contextCompressionState, history: retainedHistorySnapshot.map((message) => ({ role: message.role, content: message.content @@ -5846,24 +6127,58 @@ function App(): React.JSX.Element { for (const attachment of attachmentSnapshot) { void window.goodbuddy.context.remove(attachment.id) } - if (command) { - setSelectedRuntimeCommand('') - } } catch (error) { preparingConversations.current.delete(conversationId) + activeRuns.current.delete(requestId) + setConversationActivity(conversationId, false) for (const attachment of attachmentSnapshot) { void window.goodbuddy.context.remove(attachment.id) } - handleAgentEvent({ - requestId, - type: 'error', - status: 'failed', + setConversations((current) => + current.map((conversation) => + conversation.id === conversationId + ? { + ...conversation, + title: + conversationSnapshot.title === '新对话' && + conversation.title === prompt.slice(0, 24) + ? conversationSnapshot.title + : conversation.title, + updatedAt: Date.now(), + messages: conversation.messages.filter( + (message) => + message.id !== userMessage.id && + message.id !== assistantMessage.id + ) + } + : conversation + ) + ) + setAssistantTasks((current) => + current.filter((task) => task.id !== requestId) + ) + setActivityRecords((current) => + current.filter((record) => record.requestId !== requestId) + ) + notify({ + tone: 'error', message: error instanceof Error ? error.message : t('notices.sendFailed') }) + await releaseQueuedItem() + } finally { + dispatchedConversationQueueItems.current.delete( + queuedDispatch.item.id + ) } } + useLayoutEffect(() => { + conversationQueueDispatchRef.current = (dispatch) => { + void submit(dispatch) + } + }) + const compactRuntimeContext = async (): Promise => { if ( !activeConversation || @@ -6624,6 +6939,40 @@ function App(): React.JSX.Element { activeConversation?.messages.some( (message) => message.state === 'streaming' ) ?? false + const activeConversationQueueItems = useMemo( + () => + conversationQueueItems.filter( + (item) => item.conversationId === activeId + ), + [activeId, conversationQueueItems] + ) + const conversationExecutionRunning = + isRunning || + assistantTasks.some( + (task) => + task.conversationId === activeId && + (task.status === 'running' || + task.status === 'waiting_approval') + ) + const handleConversationQueueError = useCallback( + (message: string): void => { + notify({ + tone: 'error', + message + }) + }, + [notify] + ) + const interruptConversationQueueItem = useCallback( + (itemId: string) => + window.goodbuddy.conversationQueue.interruptAndRun(itemId), + [] + ) + const removeConversationQueueItem = useCallback( + (itemId: string) => + window.goodbuddy.conversationQueue.remove(itemId), + [] + ) const runtimeAgentControlAvailable = activeRuntimeSelection?.provider === 'opencode' && runtimeAgentOptions.length > 1 @@ -7171,10 +7520,9 @@ function App(): React.JSX.Element { onClick={() => openAssistantTask(task)} type="button" > -