diff --git a/doc/ai-harness/human-harness.md b/doc/ai-harness/human-harness.md index b349b8cd..d3a4aeb2 100644 --- a/doc/ai-harness/human-harness.md +++ b/doc/ai-harness/human-harness.md @@ -87,7 +87,8 @@ Conversation 行（`user` / `assistant` / `tool`）进 `cache.messages` 只经 ` ### 1.6 运行中追加消息的安全交接 - 普通 `running` run 可在 `ai_run.pending_messages_json` 保存多条待处理消息。交互处于审批或提问等待时，Workbench 展示交互卡并移除普通 Composer，Main 同时拒绝追加。 -- `context_budget` 先确保当前 user log 已写入，再检查待处理消息。若存在追加消息，graph 以 `yielded` 结束当前调用；该节点只会在前序工具结果已写入 transcript 且 dispatch 队列清空后到达，因此下一次 brain 不会越过尚未记录的工具结果。 +- Main 在追加消息校验开始前同步登记一次待处理提交。`context_budget` 和 Brain 入口都等待这些提交结束，再检查 `pending_messages_json`；若校验成功且消息已入队，graph 以 `yielded` 结束当前调用。Brain 入口复查用于覆盖 `context_budget` 已通过、Brain 尚未开始的调度间隙；校验失败则原 run 继续。 +- 交接只会在前序工具结果已写入 transcript 且 dispatch 队列清空后发生，因此下一次 brain 不会越过尚未记录的工具结果。Brain 已开始后才到达的追加请求由下一处 `context_budget` 边界处理。 - 若最终回答已经开始，human graph 正常完成原交付；Main 在 `run_completed` 事件后创建一次续接 run。批次内的 user 按 FIFO 进入 transcript，每条消息保留原始输入、参考资料和附件；最后一条消息决定续接模型、偏好与执行权限。 - 安全交接时较早的追加消息以 `context_only` run 和 user log 落库，最后一条由续接 run 写入。运行正常结束时自动启动续接；停止、失败和应用重启恢复时只把待处理消息转成普通 user log，不自动调用模型。 - 停止若发生在工具调用中途，已落库结果保持原样。`repairInterruptedToolCalls` 只在模型上下文投影中为没有结果的悬空调用补充合成 ToolMessage（“The run was interrupted before this tool call could be executed.”）；不会写成真实工具失败记录。 diff --git a/doc/prd/workbench.md b/doc/prd/workbench.md index e9474937..b7b9e81d 100644 --- a/doc/prd/workbench.md +++ b/doc/prd/workbench.md @@ -430,7 +430,7 @@ #### WB-COMPOSER-017 运行中追加消息 -- 系统行为：普通 `running` 状态下 Composer 保持可用，排队消息能力不变。右侧只显示一个主按钮：没有可发送草稿时按钮为停止；输入文字、附件或引用等可发送内容后，该按钮切换为发送，提交后消息按提交顺序暂存在当前 run 上，输入清空后按钮恢复为停止。消息气泡只以浅灰视觉表示尚未交接，不显示等待说明，不提供编辑或撤回入口。主模型继续当前任务；当前 Brain 已开始最终回答时先完成交付，再处理追加消息。当前 run 的工具调用、结果日志和统一投递队列全部完成后，human graph 在下一次 `context_budget` 结束处安全交接，追加消息按 FIFO 顺序进入伙伴 transcript，组成一次续接请求。审批和提问状态由交互卡替代 Composer，不显示输入框或普通主按钮，不能排队。用户停止 run、任务失败或应用中断时，已追加消息作为普通用户历史记录保存，不自动启动 AI；未完成的工具调用继续按中断恢复规则在下一轮模型上下文投影中补齐“未执行”结果。 +- 系统行为：普通 `running` 状态下 Composer 保持可用，排队消息能力不变。右侧只显示一个主按钮：没有可发送草稿时按钮为停止；输入文字、附件或引用等可发送内容后，该按钮切换为发送，提交后消息按提交顺序暂存在当前 run 上，输入清空后按钮恢复为停止。消息气泡只以浅灰视觉表示尚未交接，不显示等待说明，不提供编辑或撤回入口。Main 在追加消息校验开始前登记待处理提交；`context_budget` 与 Brain 入口等待校验结束并复查队列，避免提交正在校验时启动下一轮 Brain。校验失败则当前任务继续。主模型继续当前任务；当前 Brain 已开始最终回答时先完成交付，再处理追加消息。当前 run 的工具调用、结果日志和统一投递队列全部完成后，human graph 在下一次 `context_budget` 结束处安全交接，追加消息按 FIFO 顺序进入伙伴 transcript，组成一次续接请求。交接后的旧 run 显示固定用时；助手过程气泡按其用户气泡的投影时间加 1 毫秒排序。审批和提问状态由交互卡替代 Composer，不显示输入框或普通主按钮，不能排队。用户停止 run、任务失败或应用中断时，已追加消息作为普通用户历史记录保存，不自动启动 AI；未完成的工具调用继续按中断恢复规则在下一轮模型上下文投影中补齐“未执行”结果。 - 验收： - `WB-COMPOSER-017-AC1` 普通 run 运行中只显示一个主按钮；草稿无可发送内容时按钮显示停止，有可发送内容时按钮显示发送，提交后新增用户消息立即显示在消息数组末尾，采用浅灰样式且没有状态文字、编辑或撤回入口；发送后草稿清空且按钮恢复为停止。 - `WB-COMPOSER-017-AC2` 工具结果与投递完成前不会启动续接 Brain；安全边界后，历史工具结果先于追加用户消息，多个追加消息按 FIFO 保留为独立 user 内容；续接 run 使用最后一条追加消息的模型、偏好和执行权限。 @@ -438,6 +438,7 @@ - `WB-COMPOSER-017-AC4` 等待审批或提问时由交互卡片替代 Composer，不出现普通输入框或 Composer 主按钮；当前阻塞交互只能按交互卡片既有路径提交或停止。 - `WB-COMPOSER-017-AC5` 用户停止、运行失败或应用进程中断后，已提交追加消息保留为普通用户历史，不自动启动续接 AI；随后由用户显式发送新消息时，模型从 transcript 继续。 - `WB-COMPOSER-017-AC6` 停止发生在工具执行中途时，真实已返回的工具结果保留；只有缺少结果的悬空工具调用才在下一次上下文投影时补合成“工具调用未执行”结果。 + - `WB-COMPOSER-017-AC7` 追加提交处于后台校验期间时，不会启动下一轮 Brain；校验成功后当前轮安全交接，校验失败后当前 run 继续。交接完成后旧助手过程停止计时并显示固定用时；其过程气泡位于对应用户气泡之后，排序时间取该用户气泡投影时间加 1 毫秒。 #### WB-COMPOSER-010 附件保存与传递规则 diff --git a/doc/test-plan/ai-test.md b/doc/test-plan/ai-test.md index 11af0cd5..062eae45 100644 --- a/doc/test-plan/ai-test.md +++ b/doc/test-plan/ai-test.md @@ -647,10 +647,10 @@ DOCX tool 回归以 7 个公开工具、story 内容树、稳定 `nodeId`、segm ##### `human-queued-message-safe-boundary` - 测试类型：`backend / mock-llm` -- What：human 工具执行期间出现追加消息时，在工具结果写入后从 `context_budget` 安全让出；下一次 brain 的数组保留历史工具结果并把追加 user 放在末尾。 +- What：human 工具执行期间出现追加提交时，提交校验完成后不启动下一次 brain；在工具结果写入后安全让出，续接上下文保留历史工具结果并把追加 user 放在末尾。 - Why：追加消息若在 tool result / dispatch 完成前启动下一轮，会造成上下文顺序错乱或漏掉已执行事实。 -- How：脚本化 brain 决策调用真实 `kb_list_documents`；工具启动事件后将 pending callback 设为真，验证当前 graph 让出，再用第二个真实 graph run 捕获首轮 brain messages。 -- Oracle：第一轮状态为 `yielded` 且真实工具结果已写入 log；第二轮状态为 `completed`，历史中工具结果先于尾部追加 user；第二轮不重复调用中断前工具。 +- How：脚本化 brain 决策调用真实 `kb_list_documents`；在工具结果写入、`context_budget` 已完成检查后模拟一条正在校验的提交，延迟确认入队；验证 Brain 入口等待校验并回到交接边界，再用第二个真实 graph run 捕获首轮 brain messages。 +- Oracle：第一轮状态为 `yielded` 且真实工具结果已写入 log；未确认入队前不会启动第二次 brain 决策；第二轮状态为 `completed`，历史中工具结果先于尾部追加 user；第二轮不重复调用中断前工具。 ##### `ai-run-service-embedding-preflight` diff --git a/doc/test-plan/workbench-test.md b/doc/test-plan/workbench-test.md index 1bd49b4a..8f456c6a 100644 --- a/doc/test-plan/workbench-test.md +++ b/doc/test-plan/workbench-test.md @@ -325,10 +325,10 @@ pnpm test:update:e2e #### `queue-message-during-run` - 测试类型：`e2e / live` -- What：真实模型读取资料期间从 Composer 追加消息，验证单一主按钮在停止与发送间切换、待处理用户气泡浅灰，以及工具结果完成后安全续接。 +- What：真实模型读取资料期间从 Composer 追加消息，验证单一主按钮、浅灰待处理气泡、安全续接，以及交接后旧过程固定用时和消息显示顺序。 - Why：AI backend case 不覆盖 Renderer 输入、运行态单按钮切换、pending 气泡投影和 IPC 续接；发送草稿时停止入口暂时切换为排队发送，发送后必须恢复。 -- How：在临时工作区创建三个文本输入，通过真实 UI 发送跨文件汇总请求；观察 `text_file_read` 开始后检查空草稿时唯一主按钮为停止，输入第一条消息后同一按钮切换为发送并提交排队，再输入第二条消息重复提交；确认每次清空草稿后按钮恢复为停止；等待原 run 正常结束及续接 run 完成。记录真实模型、run 事件与最终回答。 -- Oracle：任一时刻 Composer 仅有一个主按钮；空草稿时标签为停止，有可发送内容时同一按钮标签为发送；发送后草稿清空且标签恢复为停止。两条待处理消息都按 FIFO 显示为浅灰气泡且无编辑入口；续接使用第二条消息作为最新诉求并完成。工具结果先于追加消息由 AI live `human-queued-message-safe-boundary-live` 覆盖；FIFO transcript、停止 / 中断后不自动启动由 unit `chat-run-service-contract` 覆盖，对齐 `WB-COMPOSER-017`、`WB-COMPOSER-006`。 +- How：在临时工作区创建三个文本输入，通过真实 UI 发送跨文件汇总请求；观察 `text_file_read` 开始后检查空草稿时唯一主按钮为停止，输入第一条消息后同一按钮切换为发送并提交排队，再输入第二条消息重复提交；确认每次清空草稿后按钮恢复为停止；等待原 run 正常结束及续接 run 完成，并检查原 user 气泡位于对应助手过程卡之前。记录真实模型、run 事件与最终回答。 +- Oracle：任一时刻 Composer 仅有一个主按钮；空草稿时标签为停止，有可发送内容时同一按钮标签为发送；发送后草稿清空且标签恢复为停止。两条待处理消息都按 FIFO 显示为浅灰气泡且无编辑入口；续接使用第二条消息作为最新诉求并完成。初始 run 交接后原过程行显示固定“用时”，对应用户气泡排在其助手过程行之前。工具结果先于追加消息由 AI live `human-queued-message-safe-boundary-live` 覆盖；FIFO transcript、停止 / 中断后不自动启动由 unit `chat-run-service-contract` 覆盖，对齐 `WB-COMPOSER-017`、`WB-COMPOSER-006`。 #### `llm-retry-process-row` diff --git a/src/main/services/ai/ChatRunService.ts b/src/main/services/ai/ChatRunService.ts index d1cd9657..027380f8 100644 --- a/src/main/services/ai/ChatRunService.ts +++ b/src/main/services/ai/ChatRunService.ts @@ -93,6 +93,7 @@ type ActiveRun = { abortController: AbortController eventQueue: Promise projectionFlushTimer: ReturnType | null + pendingUserMessageSubmissions: Set> } type ChatRunServiceDeps = { @@ -412,9 +413,17 @@ export class ChatRunService { const activeRun = Array.from(this.activeRuns.values()).find( (entry) => entry.run.partnerId === partnerId && isActiveRunRecord(entry.run) ) - const response = activeRun - ? await this.enqueueMessage(activeRun, { ...payload, partnerId, input }) - : await this.submitAi({ ...payload, partnerId, input }) + let response: ChatSubmitResponse + if (activeRun) { + const finishSubmission = this.registerPendingUserMessageSubmission(activeRun) + try { + response = await this.enqueueMessage(activeRun, { ...payload, partnerId, input }) + } finally { + finishSubmission() + } + } else { + response = await this.submitAi({ ...payload, partnerId, input }) + } this.trackSubmitResult(response) return response } catch (error) { @@ -428,6 +437,29 @@ export class ChatRunService { } } + private registerPendingUserMessageSubmission(activeRun: ActiveRun): () => void { + let resolveSubmission: () => void = () => {} + const submission = new Promise((resolve) => { + resolveSubmission = resolve + }) + activeRun.pendingUserMessageSubmissions.add(submission) + let released = false + return () => { + if (released) return + released = true + activeRun.pendingUserMessageSubmissions.delete(submission) + resolveSubmission() + } + } + + private async waitForPendingUserMessageSubmissions(runId: string): Promise { + while (true) { + const activeRun = this.activeRuns.get(runId) + if (!activeRun || activeRun.pendingUserMessageSubmissions.size === 0) return + await Promise.all([...activeRun.pendingUserMessageSubmissions]) + } + } + private async enqueueMessage( activeRun: ActiveRun, payload: ChatSubmitRequest & { partnerId: string; input: string } @@ -465,10 +497,8 @@ export class ChatRunService { const workspaces = await this.deps.workspaceStore.listWorkspaces() if ( this.activeRuns.get(run.id) !== activeRun || - activeRun.finalizing || activeRun.cancelled || - (run.status !== 'running' && - !(run.status === 'continued' && (run.pendingMessages?.length || 0) > 0)) || + (run.status !== 'running' && run.status !== 'continued') || this.activeApprovals.has(run.id) || this.activeQuestions.has(run.id) ) { @@ -534,50 +564,44 @@ export class ChatRunService { messages: PendingChatMessage[], partnerId: string ): Promise { - const writes: Promise[] = [] for (const message of messages) { const run = this.createContextOnlyRun(partnerId, message) - writes.push(this.deps.aiRunStore.saveRun(run)) - writes.push( - (async () => { - const existingRecords = await this.deps.chatMessageLogStore.listByRun(run.id) - if ( - existingRecords.some( - (record) => record.entry.kind === 'user' && record.entry.messageId === message.id - ) - ) { - return + await this.deps.aiRunStore.saveRun(run) + const existingRecords = await this.deps.chatMessageLogStore.listByRun(run.id) + if ( + existingRecords.some( + (record) => record.entry.kind === 'user' && record.entry.messageId === message.id + ) + ) { + continue + } + await this.deps.chatMessageLogStore.appendBatch({ + partnerId, + runId: run.id, + createdAt: message.createdAt, + entries: [ + { + kind: 'user', + messageId: message.id, + content: buildBrainUserMessageForInput({ + userInput: message.input, + preference: message.preference, + referenceScope: { + selectedReferences: message.selectedReferences, + attachments: message.attachments + }, + workspaceRoots: message.workspaceRoots, + runStartedAt: message.createdAt + }), + input: message.input, + context: { + selectedReferences: message.selectedReferences, + attachments: message.attachments + } } - await this.deps.chatMessageLogStore.appendBatch({ - partnerId, - runId: run.id, - createdAt: message.createdAt, - entries: [ - { - kind: 'user', - messageId: message.id, - content: buildBrainUserMessageForInput({ - userInput: message.input, - preference: message.preference, - referenceScope: { - selectedReferences: message.selectedReferences, - attachments: message.attachments - }, - workspaceRoots: message.workspaceRoots, - runStartedAt: message.createdAt - }), - input: message.input, - context: { - selectedReferences: message.selectedReferences, - attachments: message.attachments - } - } - ] - }) - })() - ) + ] + }) } - await Promise.all(writes) } private async createQueuedContinuation( @@ -623,7 +647,8 @@ export class ChatRunService { deleted: false, abortController: new AbortController(), eventQueue: Promise.resolve(), - projectionFlushTimer: null + projectionFlushTimer: null, + pendingUserMessageSubmissions: new Set() } previousRun.pendingMessages = [] @@ -916,7 +941,8 @@ export class ChatRunService { deleted: false, abortController: new AbortController(), eventQueue: Promise.resolve(), - projectionFlushTimer: null + projectionFlushTimer: null, + pendingUserMessageSubmissions: new Set() } this.activeRuns.set(runId, activeRun) @@ -997,6 +1023,8 @@ export class ChatRunService { isCancelled: () => Boolean(this.activeRuns.get(runId)?.cancelled), hasPendingUserMessages: () => (this.activeRuns.get(runId)?.run.pendingMessages?.length || 0) > 0, + waitForPendingUserMessageSubmissions: async () => + await this.waitForPendingUserMessageSubmissions(runId), abortSignal: this.activeRuns.get(runId)?.abortController.signal, sendEvent, recordLlmIo: this.deps.recordLlmIo, @@ -1266,7 +1294,10 @@ export class ChatRunService { event.type === 'run_completed' || event.type === 'run_failed' || event.type === 'run_cancelled' - if (isTerminalEvent && entry) entry.finalizing = true + if (isTerminalEvent && entry) { + entry.finalizing = true + await this.waitForPendingUserMessageSubmissions(runId) + } let projectedEvent: AiRunEvent = event.type === 'run_completed' && entry?.cancelled ? { type: 'run_cancelled', runId } : event const run = await this.getMutableRun(runId) diff --git a/src/main/services/ai/human/node/brainNode/index.ts b/src/main/services/ai/human/node/brainNode/index.ts index bf10ce86..b9236fc3 100644 --- a/src/main/services/ai/human/node/brainNode/index.ts +++ b/src/main/services/ai/human/node/brainNode/index.ts @@ -49,6 +49,17 @@ export const createBrainNode = (input: { }) => { return async (state: HumanState): Promise => { input.ctx.checkCancelled() + await input.ctx.waitForPendingUserMessageSubmissions() + input.ctx.checkCancelled() + if (input.ctx.hasPendingUserMessages()) { + return { + ...state, + execution: { + ...state.execution, + stage: 'context_budget' + } + } + } assertExternalLearnedSkillIds(state.skills.learnedSkillIds) input.cache.currentState = state diff --git a/src/main/services/ai/human/node/contextBudgetNode/index.ts b/src/main/services/ai/human/node/contextBudgetNode/index.ts index b5f77104..8e52c82f 100644 --- a/src/main/services/ai/human/node/contextBudgetNode/index.ts +++ b/src/main/services/ai/human/node/contextBudgetNode/index.ts @@ -99,6 +99,9 @@ export const createContextBudgetNode = (input: { ctx: AiRunContext; cache: Human input.cache.currentUserLogAppended = true } + await input.ctx.waitForPendingUserMessageSubmissions() + input.ctx.checkCancelled() + if (input.ctx.hasPendingUserMessages()) { return { ...state, diff --git a/src/main/services/ai/runtime.ts b/src/main/services/ai/runtime.ts index 5f3d5190..09f5dbd7 100644 --- a/src/main/services/ai/runtime.ts +++ b/src/main/services/ai/runtime.ts @@ -113,6 +113,9 @@ export const createAiRunContext = (payload: AiRunStartPayload, deps: AiRunDeps): }), checkCancelled, hasPendingUserMessages: () => deps.hasPendingUserMessages?.() === true, + waitForPendingUserMessageSubmissions: async () => { + await deps.waitForPendingUserMessageSubmissions?.() + }, abortSignal: deps.abortSignal, chatStructured: llmExecutor.chatStructured, chatText: llmExecutor.chatText, diff --git a/src/main/services/ai/types.ts b/src/main/services/ai/types.ts index d5bb66fe..aa8a1765 100644 --- a/src/main/services/ai/types.ts +++ b/src/main/services/ai/types.ts @@ -307,6 +307,7 @@ export type AiRunDeps = { listFileChangeRecords: (input: { runId: string }) => Promise isCancelled: () => boolean hasPendingUserMessages?: () => boolean + waitForPendingUserMessageSubmissions?: () => Promise abortSignal?: AbortSignal sendEvent: (event: AiRunEvent) => void recordLlmIo?: (record: AiLlmIoRecord) => void @@ -351,6 +352,7 @@ export type AiRunContext = { workspace?: AiWorkspaceBinding checkCancelled: () => void hasPendingUserMessages: () => boolean + waitForPendingUserMessageSubmissions: () => Promise abortSignal?: AbortSignal chatStructured: >(input: { caller: string diff --git a/src/renderer/src/pages/ChatPage.tsx b/src/renderer/src/pages/ChatPage.tsx index e08be660..4741dcef 100644 --- a/src/renderer/src/pages/ChatPage.tsx +++ b/src/renderer/src/pages/ChatPage.tsx @@ -625,9 +625,12 @@ const applyRunRecordToTurn = ( ): AssistantTurnViewModel => { const artifacts = run.artifacts || [] const runFinished = - run.status === 'completed' || run.status === 'failed' || run.status === 'cancelled' + run.status === 'completed' || + run.status === 'failed' || + run.status === 'cancelled' || + run.status === 'continued' const status = - run.status === 'completed' + run.status === 'completed' || run.status === 'continued' ? 'completed' : run.status === 'failed' ? 'failed' @@ -691,7 +694,8 @@ const buildAssistantTurnFromRunSnapshot = ( runFinished: snapshot.run.status === 'completed' || snapshot.run.status === 'failed' || - snapshot.run.status === 'cancelled' + snapshot.run.status === 'cancelled' || + snapshot.run.status === 'continued' }) const approvalReviews = projectApprovalReviews({ records: snapshot.messageLogRecords || [] diff --git a/src/shared/workbench/conversationProjection.ts b/src/shared/workbench/conversationProjection.ts index 282651d0..cb10aa90 100644 --- a/src/shared/workbench/conversationProjection.ts +++ b/src/shared/workbench/conversationProjection.ts @@ -49,6 +49,10 @@ export const projectAiConversationMessages = (input: { attachments: record.entry.context.attachments as ChatAttachment[] }) } + + const runUserMessage = messages.find( + (message) => message.id === run.userMessageId && message.role === 'user' + ) if (!loggedMessageIds.has(run.userMessageId)) { const userRecord = userRecords[0] const userEntry = userRecord?.entry.kind === 'user' ? userRecord.entry : undefined @@ -86,7 +90,7 @@ export const projectAiConversationMessages = (input: { content: deliveryContent, timestamp: run.status === 'continued' - ? run.createdAt + 1 + ? (runUserMessage?.timestamp ?? run.createdAt) + 1 : (deliveryRecord?.createdAt ?? run.finishedAt ?? run.updatedAt ?? run.createdAt + 1), artifacts: run.artifacts as AiArtifactRef[] | undefined, linkAccess: deliveryEntry?.linkAccess, diff --git a/tests/case/ai/human/mock-llm/human-queued-message-safe-boundary.test.ts b/tests/case/ai/human/mock-llm/human-queued-message-safe-boundary.test.ts index a60e1c6c..c93524d6 100644 --- a/tests/case/ai/human/mock-llm/human-queued-message-safe-boundary.test.ts +++ b/tests/case/ai/human/mock-llm/human-queued-message-safe-boundary.test.ts @@ -19,13 +19,36 @@ describe('ai human mock-llm backend', () => { runMode: 'single', execute: async (runtime) => { let hasQueuedMessage = false + let queueSubmissionStarted = false + let toolResultRecorded = false + let firstBrainCallCount = 0 let firstBrainMessages: BaseMessage[] | null = null + const countFirstBrainCalls: NonNullable = + (realChatTools) => async (request) => { + if (request.caller === 'human.brain') firstBrainCallCount += 1 + return await realChatTools(request) + } const firstRunner = createHumanTestRunner(runtime, { runId: `${runtime.runId}-queue-first`, partnerId: 'queued-message-partner', hasPendingUserMessages: () => hasQueuedMessage, + waitForPendingUserMessageSubmissions: async () => { + if (!queueSubmissionStarted || hasQueuedMessage) return + await new Promise((resolve) => setTimeout(resolve, 10)) + hasQueuedMessage = true + }, + chatToolsWrapper: countFirstBrainCalls, observeEvent: (event) => { - if (event.type === 'tool_call_started') hasQueuedMessage = true + if (event.type === 'tool_call_result') toolResultRecorded = true + if ( + event.type === 'context_budget_evaluated' && + event.decision === 'continue' && + toolResultRecorded + ) { + // Simulate a send arriving after context_budget checked the queue, + // while the pending-message validation is still in flight. + queueSubmissionStarted = true + } }, structuredOutputScript: { 'human.brain': [ @@ -93,6 +116,7 @@ describe('ai human mock-llm backend', () => { secondOutput, firstLog, firstEvents: runtime.eventCollector.snapshot(), + firstBrainCallCount, firstToolResultIndex, secondUserMessageIndex, lastMessageType: secondLastMessage?.getType() @@ -102,7 +126,8 @@ describe('ai human mock-llm backend', () => { const toolResults = filterAiRunEventsByType(result.firstEvents, 'tool_call_result') const toolLogEntries = result.firstLog.filter((record) => record.entry.kind === 'tool') runtime.assertThat( - result.firstOutput.status === 'yielded' && + result.firstOutput.status === 'yielded' && + result.firstBrainCallCount === 1 && toolResults.some((event) => event.toolName === 'kb_list_documents') && toolLogEntries.some( (record) => @@ -118,6 +143,7 @@ describe('ai human mock-llm backend', () => { message: '安全交接必须等真实工具结果写入后发生，续接消息再按顺序追加到历史上下文末尾。', data: { firstOutput: result.firstOutput, + firstBrainCallCount: result.firstBrainCallCount, secondOutput: result.secondOutput, toolResults, toolLogEntries: toolLogEntries.map((record) => record.entry), diff --git a/tests/case/unit/ai/persistence/chat-run-service-contract.test.ts b/tests/case/unit/ai/persistence/chat-run-service-contract.test.ts index 0700a9d4..88361f50 100644 --- a/tests/case/unit/ai/persistence/chat-run-service-contract.test.ts +++ b/tests/case/unit/ai/persistence/chat-run-service-contract.test.ts @@ -27,6 +27,7 @@ type TestService = { } type ChatRunServicePrivate = ChatRunService & { + activeRuns: Map queueRunEvent: (runId: string, event: AiRunEvent) => Promise requestApprovalBatch: ( runId: string, @@ -220,8 +221,10 @@ describe('unit ai persistence backend', () => { run.status = 'continued' await setup.storage.aiRun.saveRun(run) const second = await submitQueued('第二条追加消息') + const third = await submitQueued('第三条追加消息') expect(first).toMatchObject({ success: true, queued: true, userMessage: { pending: true } }) expect(second).toMatchObject({ success: true, queued: true, userMessage: { pending: true } }) + expect(third).toMatchObject({ success: true, queued: true, userMessage: { pending: true } }) expect(first.run).toBeUndefined() expect(first.assistantMessage).toBeUndefined() expect(first.userMessage?.timestamp).toBeLessThan(second.userMessage?.timestamp ?? 0) @@ -240,16 +243,22 @@ describe('unit ai persistence backend', () => { const messages = snapshot.messages.filter((item) => item.role === 'user') const firstQueuedMessage = first.userMessage const secondQueuedMessage = second.userMessage + const thirdQueuedMessage = third.userMessage const firstQueuedContextRun = firstQueuedMessage ? await setup.storage.aiRun.getRun(`context-${firstQueuedMessage.id}`) : null const log = firstQueuedMessage ? await setup.storage.chatMessageLog.list('partner-1') : [] const firstQueuedLog = log.find((record) => record.runId === firstQueuedContextRun?.id) + const secondQueuedContextRun = secondQueuedMessage + ? await setup.storage.aiRun.getRun(`context-${secondQueuedMessage.id}`) + : null + const secondQueuedLog = log.find((record) => record.runId === secondQueuedContextRun?.id) expect(messages.map((item) => item.content)).toEqual([ '生成测试回答', '第一条追加消息', - '第二条追加消息' + '第二条追加消息', + '第三条追加消息' ]) expect(messages.slice(1).every((item) => item.pending !== true)).toBe(true) expect(firstQueuedContextRun?.status).toBe('context_only') @@ -258,11 +267,75 @@ describe('unit ai persistence backend', () => { messageId: firstQueuedMessage?.id, input: '第一条追加消息' }) - expect(snapshot.activeRuns.map((item) => item.run.input)).toEqual(['第二条追加消息']) - expect(snapshot.activeRuns[0]?.run.userMessageId).toBe(secondQueuedMessage?.id) + expect(secondQueuedLog?.entry).toMatchObject({ + kind: 'user', + messageId: secondQueuedMessage?.id, + input: '第二条追加消息' + }) + expect(firstQueuedLog?.seq).toBeLessThan(secondQueuedLog?.seq ?? 0) + expect(snapshot.activeRuns.map((item) => item.run.input)).toEqual(['第三条追加消息']) + expect(snapshot.activeRuns[0]?.run.userMessageId).toBe(thirdQueuedMessage?.id) expect((await setup.storage.aiRun.getRun(run.id))?.pendingMessages).toEqual([]) }) + it('waits for an in-flight queued submission before finalizing the run', async () => { + const setup = await createTestService() + const run = await submitAiRun(setup) + const workspaceStore = setup.storage.workspace + const listWorkspaces = workspaceStore.listWorkspaces.bind(workspaceStore) + let releaseValidation: () => void = () => {} + let markValidationStarted: () => void = () => {} + const validationGate = new Promise((resolve) => { + releaseValidation = resolve + }) + const validationStarted = new Promise((resolve) => { + markValidationStarted = resolve + }) + workspaceStore.listWorkspaces = async () => { + markValidationStarted() + await validationGate + return await listWorkspaces() + } + + const service = setup.service as ChatRunServicePrivate + const submission = setup.service.submit({ + partnerId: 'partner-1', + input: '校验期间到达的追加消息', + preference: 'default', + executionPermission: 'review', + mainModel: { configId: 'mock-config', modelName: 'mock-model' } + }) + let terminalFinished = false + const terminal = service + .queueRunEvent(run.id, { type: 'run_completed', runId: run.id }) + .then(() => { + terminalFinished = true + }) + + try { + await validationStarted + await waitForCondition(() => service.activeRuns.get(run.id)?.finalizing === true) + expect(terminalFinished).toBe(false) + releaseValidation() + + const queued = await submission + await terminal + const snapshot = await setup.service.getPartnerSnapshot('partner-1') + + expect(queued).toMatchObject({ + success: true, + queued: true, + userMessage: { content: '校验期间到达的追加消息', pending: true } + }) + expect(snapshot.activeRuns.map((item) => item.run.input)).toEqual([ + '校验期间到达的追加消息' + ]) + expect((await setup.storage.aiRun.getRun(run.id))?.pendingMessages).toEqual([]) + } finally { + releaseValidation() + } + }) + it('stopped-run-parks-queued-messages-without-starting-ai', async () => { const setup = await createTestService() const run = await submitAiRun(setup) diff --git a/tests/case/unit/workbench/conversation-projection.test.ts b/tests/case/unit/workbench/conversation-projection.test.ts index 54c93446..ee1ca394 100644 --- a/tests/case/unit/workbench/conversation-projection.test.ts +++ b/tests/case/unit/workbench/conversation-projection.test.ts @@ -155,6 +155,47 @@ describe('conversationProjection', () => { expect(messages[1]).toMatchObject({ role: 'assistant', content: '' }) }) + it('orders a continued assistant process immediately after its projected user message', () => { + const messages = projectAiConversationMessages({ + partnerId: 'partner-1', + runs: [ + buildRun({ + id: 'run-continued', + userMessageId: 'continued-user', + assistantMessageId: 'continued-assistant', + status: 'continued', + input: '继续查财报', + createdAt: 10, + updatedAt: 200, + finishedAt: 200, + partialAnswer: '', + finalAnswer: undefined + }) + ], + records: [ + buildLogRecord({ + runId: 'run-continued', + seq: 1, + createdAt: 100, + entry: { + kind: 'user', + messageId: 'continued-user', + content: '继续查财报', + input: '继续查财报', + context: { selectedReferences: [], attachments: [] } + } + }) + ] + }) + const ordered = [...messages].sort((left, right) => left.timestamp - right.timestamp) + + expect(ordered.map((message) => message.id)).toEqual([ + 'continued-user', + 'continued-assistant' + ]) + expect(ordered[1]?.timestamp).toBe(ordered[0]!.timestamp + 1) + }) + it('keeps pending messages after active assistant activity in FIFO order', () => { const runs = [ buildRun({ diff --git a/tests/case/workbench/run/process/queue-message-during-run.e2e.test.ts b/tests/case/workbench/run/process/queue-message-during-run.e2e.test.ts index eb6deced..6efa93b8 100644 --- a/tests/case/workbench/run/process/queue-message-during-run.e2e.test.ts +++ b/tests/case/workbench/run/process/queue-message-during-run.e2e.test.ts @@ -64,6 +64,17 @@ describe('workbench run e2e', () => { event.runId === firstRunStarted.runId && event.toolName === 'text_file_read' }) + const firstRunDurationTestId = await runtime.page + .waitForFunction( + (prefix) => + document + .querySelector(`[data-testid^="${prefix}-"]`) + ?.getAttribute('data-testid') || + null, + workbenchSelectors.assistantRunDurationPrefix, + { timeout: 10_000 } + ) + .then(async (handle) => await handle.jsonValue()) await setWorkbenchInput(runtime.page, FIRST_QUEUED_INPUT) await runtime.page.waitForFunction( @@ -155,6 +166,39 @@ describe('workbench run e2e', () => { if (firstRunTerminal.type !== 'run_completed') { throw new Error(`initial run ended with ${firstRunTerminal.type}`) } + await runtime.page.waitForFunction( + (testId) => + document.querySelector(`[data-testid="${testId}"]`)?.textContent?.includes('用时 '), + firstRunDurationTestId, + { timeout: 10_000 } + ) + const firstRunDurationText = + (await runtime.page.locator(`[data-testid="${firstRunDurationTestId}"]`).textContent()) || + '' + const firstRunAssistantMessageId = firstRunDurationTestId.slice( + `${workbenchSelectors.assistantRunDurationPrefix}-`.length + ) + const firstRunUserBeforeAssistant = await runtime.page.evaluate( + ({ assistantMessageId, userInput, userMessagePrefix }) => { + const userMessage = Array.from( + document.querySelectorAll(`[data-testid^="${userMessagePrefix}-"]`) + ).find((element) => element.textContent?.includes(userInput)) + const assistantTurn = document.querySelector( + `[data-testid="workbench-assistant-turn-${assistantMessageId}"]` + ) + return Boolean( + userMessage && + assistantTurn && + (userMessage.compareDocumentPosition(assistantTurn) & + Node.DOCUMENT_POSITION_FOLLOWING) + ) + }, + { + assistantMessageId: firstRunAssistantMessageId, + userInput: FIRST_INPUT, + userMessagePrefix: workbenchSelectors.userMessagePrefix + } + ) const continuationStarted = await waitForWorkbenchAiEvent(runtime, { name: 'queued-continuation-started', @@ -208,6 +252,8 @@ describe('workbench run e2e', () => { firstRunId: firstRunStarted.runId, continuationInput: continuationStarted.type === 'run_started' ? continuationStarted.input : '', + firstRunDurationText, + firstRunUserBeforeAssistant, firstQueuedInput: FIRST_QUEUED_INPUT, secondQueuedInput: SECOND_QUEUED_INPUT, toolResults: toolResults.map((event) => ({ @@ -230,6 +276,8 @@ describe('workbench run e2e', () => { result.primaryButtonCount === 1 && result.sendButtonCountWithDraft === 1 && result.continuationInput === SECOND_QUEUED_INPUT && + result.firstRunDurationText.includes('用时 ') && + result.firstRunUserBeforeAssistant && result.toolResults.some((event) => event.toolName === 'text_file_read' && event.ok), { name: 'workbench-queues-messages-during-ai-run', diff --git a/tests/harness/backend/realAiDepsFactory.ts b/tests/harness/backend/realAiDepsFactory.ts index 62cf3720..72c1801b 100644 --- a/tests/harness/backend/realAiDepsFactory.ts +++ b/tests/harness/backend/realAiDepsFactory.ts @@ -233,6 +233,7 @@ export const createRealAiDepsFactory = async (input: { (await aiFileChangeService.listRecords(request)).items, isCancelled: contextInput.checkCancelledOverride || (() => false), hasPendingUserMessages: contextInput.hasPendingUserMessages, + waitForPendingUserMessageSubmissions: contextInput.waitForPendingUserMessageSubmissions, sendEvent: (event) => { input.eventCollector.add(event) contextInput.observeEvent?.(event) diff --git a/tests/harness/backend/test-types.ts b/tests/harness/backend/test-types.ts index 0e92bd28..21616445 100644 --- a/tests/harness/backend/test-types.ts +++ b/tests/harness/backend/test-types.ts @@ -191,6 +191,7 @@ export type CreateAiRunContextInput = { questionResolver?: QuestionResolver checkCancelledOverride?: () => void hasPendingUserMessages?: () => boolean + waitForPendingUserMessageSubmissions?: () => Promise observeEvent?: (event: AiRunEvent) => void humanRuntimeOverrides?: { graphRecursionLimit?: number