diff --git a/README.md b/README.md index 53f91f7..685bdcd 100644 --- a/README.md +++ b/README.md @@ -16,12 +16,13 @@ DeepSeek Harness 会自动调用工具并连续作答，虽然可以使用插话 ## 功能 +**暂停点**：模型完成一轮工具交互、发出下一次请求前，插件让它停下来等你的位置。 + - **可开关**：默认关闭，每个会话可独立开启。 -- **暂停在请求前**：模型完成一轮工具交互、正要发出下一次请求之前停下，不打断任何在途 API 调用。 +- **仅在暂停点暂停**：只有这个位置会停，也就是模型完成一轮工具交互、正要发出下一次请求之前。你发消息触发的第一次请求不停，也不打断任何在途 API 调用。 - **保持 composer 可用**：暂停期间输入框照常可打字，不遮挡、不切换。 -- **回车即放行**：在主输入框按裸 Enter 继续。草稿为空为无感放行，草稿有字或图片则作为插话随本次请求一并发出，随后清空草稿与附件。 -- **按钮同样放行**：也可直接点「已暂停·回车或点此继续」按钮，行为同上并读取当前草稿与附件。 -- **图片随插话**：草稿里的图片会先经附件服务持久化，再与文字组成同一条用户消息，不会只发文字而把图片留在输入框。文件附件暂不支持插话，会提示先移除。 +- **回车或按钮放行**：在主输入框按裸 Enter，或点「已暂停·回车或点此继续」按钮，都能放行暂停点。草稿为空无感放行，有字或图片则作为插话随本次请求一并发出。暂停点还没到时按回车也有效，消息留到下一个暂停点；这一轮已结束则作为新的一轮发出，按了就一定送达。 +- **图片随插话**：按下回车时文字与图片一起立刻离开输入框，图片在后台序列化后随消息发出，不会只发文字而把图片留在输入框。发送失败时文字与已序列化的图片暂存到下次发送，序列化失败的图片需要重新选择。文件附件暂不支持插话，会被直接移出输入框并提示重新选择。 - **模型无感知**：上下文完整、不插入多余消息，模型察觉不到暂停存在。 ## 它长什么样 @@ -59,9 +60,9 @@ dsh plugin --profile web remove dsh-pause ## 使用 1. 点 composer 工具行右侧的「暂停」按钮，变「暂停开」即该会话已开启。 -2. 之后该会话的工具续跑请求会在发出前停下等你。 -3. 需要放行时在主输入框按裸 Enter，空为无感继续，有字或有图为插话。也可点「已暂停·回车或点此继续」。 -4. 用完再点「暂停开」关掉。暂停中关掉会放行正挂着的门。 +2. 之后该会话的工具续跑请求会停在暂停点等你。 +3. 需要放行时在主输入框按裸 Enter，空为无感继续，有字或有图为插话。也可点「已暂停·回车或点此继续」。暂停点还没到时按回车同样有效，消息会留到下一个暂停点；这一轮已经结束则作为新的一轮发出。 +4. 用完再点「暂停开」关掉。暂停中关掉会放行当前暂停点。 ## 配置 @@ -77,13 +78,14 @@ plugins: ## 工作原理 -dsh 把一次模型请求当作一个 step，agent 连续调用工具时会经历 step、工具执行、下一次 step 的循环。本插件挂在 `agent/pre-step`，在「带工具结果的下一请求」正要进入前决定是否拉门。 +dsh 把一次模型请求当作一个 step，agent 连续调用工具时会经历 step、工具执行、下一次 step 的循环。本插件监听 `agent/pre-step`，由它决定何时出现暂停点。 - 暂停点设在请求真正发出之前，故无在途调用可打断。 -- 续跑判定用同一 turn 内 step 大于 1 近似，首条纯文字请求永不暂停。 +- 续跑判定用同一 turn 内 step 大于 1 近似，每个 turn 的第一次请求永不暂停。 - 浏览器端经 `conversation.input.right` 槽位在 composer 工具行渲染控件。 - 带图插话由浏览器把草稿图片序列化成 base64 随放行请求上传，主机经 `attachments.admitPromptContent` 换成持久引用后与文字并入同一条 user 消息；RPC 请求体上限据此放宽到 300 MiB。 -- 因 dsh 的 composer Enter 由 InputBar 私有 keymap 处理、第三方无 hook 点，暂停态用 DOM capture 拦截裸 Enter 改走放行，不改 DSH 源码。 +- 因 dsh 的 composer Enter 由 InputBar 私有 keymap 处理、第三方无 hook 点，开关开启期间用 DOM capture 常驻拦截裸 Enter，改走主机裁决，不改 DSH 源码。 +- 有没有暂停点由主机权威判定：有就放行，没有但 `agent/status` 为 running 就排进队列等下一个，idle 则直接 `followup` 成为新 turn；主机暂时发不出去时交回客户端暂存，不静默丢弃。 - 主机侧不经 `ctx.connection.rpc.handle`：该 API 自 0.1.2 起即不可用，改为向 webServer 自注册 `/dsh-pause` 前缀路由，复用 connection 的信任判定与浏览器鉴权，信封格式与官方一致。 ## 开发 diff --git a/package.json b/package.json index 11fed4c..cded4b4 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "dsh-pause", - "version": "0.1.4", + "version": "0.1.5", "packageManager": "pnpm@9.15.9", "engines": { "node": ">=26.3.1" @@ -54,6 +54,11 @@ "peerDependencies": { "@deepseek-ai/cordis": ">=4.0.0", "@deepseek-ai/dsh": "^0.2.0-rc.2", + "@deepseek-ai/dsh-client-connection": "^0.2.0-rc.2", + "@deepseek-ai/dsh-client-ui-conversation": "^0.2.0-rc.2", + "@deepseek-ai/dsh-client-ui-renderer": "^0.2.0-rc.2", + "@deepseek-ai/dsh-client-ui-slots": "^0.2.0-rc.2", + "@deepseek-ai/dsh-llm": "^0.2.0-rc.2", "react": "^18.2.0" }, "dependencies": { diff --git a/src/client/PauseBar.tsx b/src/client/PauseBar.tsx index c5f954a..930fa27 100644 --- a/src/client/PauseBar.tsx +++ b/src/client/PauseBar.tsx @@ -3,10 +3,10 @@ * * 目标交互： * - 开关按钮，本会话总开关。 - * - 暂停时在 host 报 paused 的状态下于 composer 主输入框按裸 Enter 即继续。草稿为空则无感放行，草稿有字则作为插话放行并清空草稿。 - * - 未暂停时不拦截任何按键，完全透明。 + * - 开关开启期间，在 composer 主输入框按裸 Enter 一律交主机裁决：有暂停点就继续并带插话，没有但在跑就排进下一个暂停点，没有且空闲则由主机 followup，失败内容暂存而不退回输入框。 + * - 开关关闭时不拦截任何按键，完全透明。 * - * dsh 的 composer Enter 由 InputBar 私有 keymap 处理，第三方没有正式 hook 点，故本组件在暂停态对所属 composer 挂 document 级 capture-phase keydown，抢在 Lexical 之前拦截裸 Enter。 + * dsh 的 composer Enter 由 InputBar 私有 keymap 处理，第三方没有正式 hook 点，故本组件在开关开启期间对所属 composer 挂 document 级 capture-phase keydown，抢在 Lexical 之前拦截裸 Enter。 * * @module dsh-pause/client */ @@ -23,6 +23,8 @@ export interface PauseBarInjected { } /** 把草稿附件序列化成可随插话提交的载荷。 */ readonly serializeAttachments?: (ids: readonly DraftAttachmentId[]) => Promise + /** 释放已经送出、不再留在草稿里的附件，避免底层对象与预览 URL 泄漏。 */ + readonly releaseAttachment?: (id: DraftAttachmentId) => void } /** conversation.input.right 的 session 作用域组件收到的标准 props + 注入面。 */ @@ -39,6 +41,10 @@ export interface PauseControlProps extends PauseBarInjected { interface PauseStatus { enabled?: boolean paused?: boolean + /** agent 是否在跑；决定没有暂停点时插话该排队还是直接 followup。 */ + busy?: boolean + /** 已排队、等下一个暂停点或 idle 补发的插话条数。 */ + pendingMessageCount?: number } async function fetchStatus(rpc: PauseBarInjected['rpc'], sessionId: string): Promise { @@ -47,13 +53,18 @@ async function fetchStatus(rpc: PauseBarInjected['rpc'], sessionId: string): Pro if (result.ok) return result.value as PauseStatus | undefined return undefined } catch { + // 本次轮询失败不更新状态，下一次轮询会重试。 return undefined } } -/** 一次放行的结果；error 存在表示主机拒绝，门未放行。 */ +/** 一次放行的结果；error 存在表示主机拒绝，暂停点未放行。 */ interface ReleaseOutcome { readonly released: boolean + /** 主机裁决时该会话的 agent 是否在跑。 */ + readonly busy: boolean + /** 主机已接住这条插话，草稿可以清空。 */ + readonly queued: boolean readonly error?: string } @@ -65,11 +76,11 @@ async function callRelease( ): Promise { try { const result = await rpc.call('/dsh-pause', 'release', { sessionId, text, images }) - if (!result.ok) return { released: false, error: result.error?.message ?? '放行失败' } - const value = result.value as { released?: boolean } | undefined - return { released: value?.released === true } + if (!result.ok) return { released: false, busy: false, queued: false, error: result.error?.message ?? '放行失败' } + const value = result.value as { released?: boolean; busy?: boolean; queued?: boolean } | undefined + return { released: value?.released === true, busy: value?.busy === true, queued: value?.queued === true } } catch (error) { - return { released: false, error: error instanceof Error ? error.message : String(error) } + return { released: false, busy: false, queued: false, error: error instanceof Error ? error.message : String(error) } } } @@ -82,50 +93,53 @@ async function callSetEnabled(rpc: PauseBarInjected['rpc'], sessionId: string, e } /** - * 放行一扇门：draft 与图片一起作为插话，空则无感继续。 - * 附件必须先序列化并随放行请求上传；主机拒绝时保持门与草稿原样，避免出现只发文字、图片残留的状态。 + * 把一次回车交给主机裁决：有暂停点就放行，没有但在跑就排进下一个暂停点，没有且空闲且有 agent 引用就由主机 followup。 + * 按下回车时文字与图片已经离开输入框，所以任何失败都不回填，改为暂存到下次发送时优先带出。 */ -async function releaseAndClear( +async function releaseOrQueue( rpc: PauseBarInjected['rpc'], sessionId: string, draft: string, - attachmentIds: readonly DraftAttachmentId[], - serializeAttachments: PauseBarInjected['serializeAttachments'], - inputActions: InputActions | undefined, + carried: readonly SubmitAttachment[], + attachmentsPromise: Promise | undefined, + onUnsent: (text: string, images: readonly SubmitAttachment[]) => void, onDone: () => void, onError: (message: string) => void, ): Promise { - let attachments: readonly SubmitAttachment[] = [] - if (attachmentIds.length > 0) { - if (serializeAttachments === undefined) { - onError('当前环境不支持附件插话，请先发送或移除附件') - return - } + let attachments: readonly SubmitAttachment[] = carried + if (attachmentsPromise !== undefined) { try { - attachments = await serializeAttachments(attachmentIds) + attachments = [...carried, ...await attachmentsPromise] } catch (error) { - onError('附件无法随插话发送：' + (error instanceof Error ? error.message : String(error))) + onUnsent(draft, carried) + onError('附件无法随插话发送，文字已保留：' + (error instanceof Error ? error.message : String(error))) return } } if (attachments.some((part) => part.type !== 'image')) { - onError('暂停中只支持图片插话，请先移除文件附件') + onUnsent(draft, carried) + onError('文件附件无法随插话发送，已从输入框移除，文字已保留') return } + const hasPayload = draft.trim() !== '' || attachments.length > 0 const outcome = await callRelease(rpc, sessionId, draft, attachments) if (outcome.error !== undefined) { - onError(outcome.error) + onUnsent(draft, attachments) + onError(outcome.error + '，已保留，下次发送时一并送出') return } - if (!outcome.released) { + if (outcome.released || outcome.queued) { onDone() return } - if (draft !== '' && inputActions !== undefined) inputActions.setDraft('') - if (inputActions !== undefined) { - for (const id of attachmentIds) inputActions.removeAttachment(id) + // 空草稿的回车本来就没有动作，重放只会触发一次空提交。 + if (!hasPayload) { + onDone() + return } - onDone() + // 主机没接住这条插话：不重放、不回填，暂存到下次发送时优先带出。 + onUnsent(draft, attachments) + onError('这条插话暂时没能送达主机，已保留，下次发送时一并送出') } /** @@ -141,7 +155,7 @@ function PauseIcon(): ReactElement { } /** - * 暂停控件。暂停态挂 document capture 拦截裸 Enter，落点在本 composer 卡内时改为放行。 + * 暂停控件。开关开启期间挂 document capture 拦截裸 Enter，落点在本 composer 卡内时交给主机裁决。 */ export const PauseControl = memo(function PauseControl({ rpc, @@ -149,11 +163,15 @@ export const PauseControl = memo(function PauseControl({ useInput, inputActions, serializeAttachments, + releaseAttachment, }: PauseControlProps) { const [enabled, setEnabled] = useState(false) const [paused, setPaused] = useState(false) + const [pendingCount, setPendingCount] = useState(0) const [error, setError] = useState(undefined) const rootRef = useRef(null) + // composer 卡标记缺失时只报一次，避免每次按键都刷控制台。 + const missingCardWarnedRef = useRef(false) // render 内 subscribe：草稿与附件变化会触发本组件重渲染并更新 ref。 // 注意：不能在 capture 事件处理器里调 useInput，因为 hook 只在 render 期合法，故存 ref。 @@ -163,27 +181,80 @@ export const PauseControl = memo(function PauseControl({ draftRef.current = draft const idsRef = useRef(attachmentIds) idsRef.current = attachmentIds - // capture 处理器要读最新 paused/rpc/sessionId/inputActions，也用 ref 规避闭包过期。 - const gateRef = useRef<{ paused: boolean; release: () => void }>({ paused: false, release: () => {} }) - // 放行中的重复 Enter 或连点不再发起第二次上传，避免同一批图片被提交两次。 + // 提交阶段会拒绝移除附件，capture 处理器要读最新阶段，同样存 ref。 + const phase = useInput((state: InputState) => state.phase) + const phaseRef = useRef(phase) + phaseRef.current = phase + // capture 处理器要读最新的放行入口，用 ref 规避闭包过期。 + const pausePointRef = useRef<{ release: () => void }>({ release: () => {} }) + // 放行中的重复 Enter 记到 deferred，不并发上传，但这次回车不能丢。 const releasingRef = useRef(false) - gateRef.current = { - paused, + const deferredRef = useRef(false) + // 已经交给在飞请求的附件，后续回车跳过同一批。 + const claimedRef = useRef>(new Set()) + // 没能送达主机的插话按原样暂存，下次按回车时优先带出；绝不回填输入框。 + const unsentRef = useRef>([]) + pausePointRef.current = { release: () => { - if (releasingRef.current) return - releasingRef.current = true + if (releasingRef.current) { + deferredRef.current = true + return + } + // 提交中的 composer 会拒绝移除附件，继续会把同一批图片留在草稿里重复发送。 + if (phaseRef.current === 'adjudicating' || phaseRef.current === 'submitting') { + setError('正在提交上一条消息，稍后再按回车') + return + } setError(undefined) - const pending = releaseAndClear( + const carried = unsentRef.current + unsentRef.current = [] + const carriedText = carried.map((item) => item.text.trim()).filter((text) => text !== '').join('\n') + const current = draftRef.current + const sent = carriedText === '' ? current : current === '' ? carriedText : carriedText + '\n' + current + const carriedImages = carried.flatMap((item) => item.images) + const ids = idsRef.current.filter((id: DraftAttachmentId) => !claimedRef.current.has(id)) + // 检查先于置位：此时请求还没发出，失败时不能留下在飞标记。 + if (ids.length > 0 && serializeAttachments === undefined) { + unsentRef.current = carried + setError('当前环境不支持附件插话，文字仍保留，图片请先移除') + return + } + releasingRef.current = true + for (const id of ids) claimedRef.current.add(id) + // 序列化在调用时就同步捕获了附件对象，所以可以立刻把图片也一起带走。 + const serialization = ids.length === 0 || serializeAttachments === undefined + ? undefined + : serializeAttachments(ids) + // 按了就带走：文字与图片一起立刻离开输入框，之后再输入的字属于下一条消息。 + draftRef.current = '' + inputActions?.setDraft('') + for (const id of ids) inputActions?.removeAttachment(id) + const pendingRelease = releaseOrQueue( rpc, sessionId, - draftRef.current, - idsRef.current, - serializeAttachments, - inputActions, + sent, + carriedImages, + serialization, + (text, images) => { unsentRef.current.push({ text, images }) }, () => setPaused(false), - (message) => setError(message), + (message) => { + deferredRef.current = false + setError(message) + }, ) - void pending.finally(() => { releasingRef.current = false }) + void pendingRelease.finally(() => { + releasingRef.current = false + for (const id of ids) { + claimedRef.current.delete(id) + // 序列化已经结束，底层附件不会再被读取，这时释放才不会泄漏。 + releaseAttachment?.(id) + } + // 请求期间被挡下的回车补一次，用那时的草稿重走裁决。 + if (deferredRef.current) { + deferredRef.current = false + pausePointRef.current.release() + } + }) }, } @@ -196,6 +267,7 @@ export const PauseControl = memo(function PauseControl({ if (disposed || s === undefined) return setEnabled(s.enabled ?? false) setPaused(s.paused ?? false) + setPendingCount(s.pendingMessageCount ?? 0) } void poll() timer = setInterval(() => void poll(), 400) @@ -205,23 +277,31 @@ export const PauseControl = memo(function PauseControl({ } }, [rpc, sessionId]) - // 暂停态激活 capture：document 捕获阶段抢在 Lexical 前拦裸 Enter。 + // 开关开启期间常驻 capture：暂停点挂上与否由主机裁决，客户端状态滞后不影响拦截。 useEffect(() => { - if (!paused) return + if (!enabled) return const onCapture = (event: KeyboardEvent): void => { if (event.key !== 'Enter' || event.shiftKey || event.ctrlKey || event.metaKey) return if (event.isComposing || event.keyCode === 229) return const root = rootRef.current if (root === null || !root.isConnected) return const card = root.closest('[data-composer-card]') - if (card === null || !card.contains(event.target as Node)) return + if (card === null) { + if (!missingCardWarnedRef.current) { + missingCardWarnedRef.current = true + console.warn('[dsh-pause] 找不到 [data-composer-card]，回车拦截未生效；请检查 DSH 版本是否改了 composer 结构') + } + return + } + missingCardWarnedRef.current = false + if (!card.contains(event.target as Node)) return event.preventDefault() event.stopPropagation() - void gateRef.current.release() + void pausePointRef.current.release() } document.addEventListener('keydown', onCapture, true) return () => document.removeEventListener('keydown', onCapture, true) - }, [paused]) + }, [enabled, sessionId]) const toggle = async (next: boolean): Promise => { setError(undefined) @@ -269,7 +349,7 @@ export const PauseControl = memo(function PauseControl({ {paused && ( )} + {pendingCount > 0 && ( + + {pendingCount} 条已排入下一步 + + )} {error !== undefined && ( {error} diff --git a/src/client/index.ts b/src/client/index.ts index b0359dc..f142c62 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -1,9 +1,9 @@ /** * dsh-pause 浏览器半身：暂停控件挂在 composer 工具行右侧 conversation.input.right。 * - * 目标：暂停时在 composer 主输入框里按 Enter 即继续。草稿为空则无感放行，草稿有字则作为插话一并放行并清空草稿。 - * 因为 dsh 的 composer Enter 由 InputBar 私有 keymap 处理，第三方没有正式 hook 点，所以本插件用 DOM capture 在暂停态拦截裸 Enter，阻止默认送信并改走 release。 - * 控件本身在 composer 卡内渲染，因此暂停条可见且可点。capture 监听只在暂停态激活，未暂停时不拦截任何按键。 + * 目标：开关开启期间在 composer 主输入框里按 Enter 即继续。有暂停点则放行并带上草稿，没有但在跑则排进下一个暂停点，没有且空闲则由主机 followup；主机发不出去时交回客户端暂存，失败内容不退回输入框。 + * 因为 dsh 的 composer Enter 由 InputBar 私有 keymap 处理，第三方没有正式 hook 点，所以本插件用 DOM capture 在开关开启期间常驻拦截裸 Enter，阻止默认送信并改走主机裁决。 + * 控件本身在 composer 卡内渲染，因此暂停条可见且可点。capture 监听只在开关开启期间激活，关闭时不拦截任何按键。 * * @module dsh-pause/client */ @@ -25,6 +25,8 @@ export interface PauseBarInjected { readonly rpc: ClientConnectionRpc /** 把草稿附件序列化成图片 base64 或文件 receipt；图片在主机侧持久化后随插话发出。 */ readonly serializeAttachments: (ids: readonly DraftAttachmentId[]) => Promise + /** 释放已经送出、不再留在草稿里的附件，避免底层对象与预览 URL 泄漏。 */ + readonly releaseAttachment: (id: DraftAttachmentId) => void } /** @@ -43,12 +45,15 @@ export function apply(ctx: ClientContext): void { inject: (): PauseBarInjected => ({ rpc: connection.rpc, serializeAttachments: async (ids: readonly DraftAttachmentId[]) => { - // 文件附件走官方提交与 receipt 绑定，本插件暂不接管，提前给出可操作的提示。 + // 文件附件走官方提交与 receipt 绑定，本插件暂不接管；这时附件已从输入框移除，只能请用户重选。 if (conversation.resolveDraftAttachments(ids).some((draft) => draft.kind !== 'image')) { - throw new Error('暂停中只支持图片插话，请先移除文件附件') + throw new Error('文件附件无法随插话发送，请重新选择') } return (await conversation.serializeDraftAttachments(ids)).attachments }, + releaseAttachment: (id: DraftAttachmentId) => { + conversation.releaseDraftAttachment(id) + }, }), }, PauseControl, diff --git a/src/controller.ts b/src/controller.ts index 3afba22..16c7000 100644 --- a/src/controller.ts +++ b/src/controller.ts @@ -2,10 +2,12 @@ * dsh-pause 主机半身：暂停控制器与判定。 * * 核心语义：在 agent 每轮 step 进入前的 agent/pre-step 事件里判断这次请求是否为工具交互后的续跑。 - * 若是续跑且开关开启，则把门拉住，等人类经 release 原语放行后才把这一次请求发出去。 + * 若是续跑且开关开启，则挂起一个暂停点，等人类经 release 原语放行后才把这一次请求发出去。 * 放行时可带补充文字，作为一条 user 消息并入本 step 的 messages，让模型在紧接着的请求里看到它。 * - * 门只卡在请求真正发出之前，因此不打断任何在途 API 调用。 + * 暂停点只卡在请求真正发出之前，因此不打断任何在途 API 调用。 + * 人类在暂停点挂上之前按回车时，插话先排队，下一个暂停点不再等回车而是直接带头并入本步，即预放行； + * turn 若在暂停点出现之前结束，队列由主机作为新的 followup 发出去；followup 抛错时放回队列等下一次重试，任何失败都不静默丢弃。 * driver 组装好的上下文原样放行，不插入、不删除、不改写任何模型可见的痕迹，模型无感知暂停。 * * @module dsh-pause @@ -14,18 +16,20 @@ import type { UserMessage } from '@deepseek-ai/dsh-llm' import { createUserMessage } from '@deepseek-ai/dsh-llm' -/** 一个会话的暂停开关与门控状态。会话以 agent 的 session id 为键。 */ +/** 一个会话的暂停开关与暂停点状态。会话以 agent 的 session id 为键。 */ export interface PauseSessionState { /** 用户开关：该会话是否启用暂停。false 时所有请求直接放行。 */ enabled: boolean - /** 当前是否有一扇门被拉住。 */ - paused: boolean - /** 正在等待放行的解析器，放行时若带文字则一并返回。 */ - waiting: ReleaseRequest | undefined + /** 该会话的 agent 是否在跑。idle 时暂停点不会再来，队列要当场发出去。 */ + busy: boolean + /** 当前挂着的暂停点，放行时若带文字则一并返回。 */ + pausePoint: PausePoint | undefined + /** 暂停点还没挂上时人类已按回车发出的插话，等下一个暂停点并入或 turn 结束时 followup。 */ + pendingMessages: UserMessage[] } -/** 一次放行请求：人类选择继续，可携带一条已构造好的补充用户消息。 */ -export interface ReleaseRequest { +/** 一个暂停点：人类选择继续时用它放行，可携带一条已构造好的补充用户消息。 */ +export interface PausePoint { /** 解析放行等待的入口，参数为待并入本步的补充消息，undefined 表示无感继续。 */ resolve: (message: UserMessage | undefined) => void reject: (reason: unknown) => void @@ -38,7 +42,7 @@ export interface ReleaseRequest { * dsh 的 driver 把一个模型请求当作一个 step。用户首条消息进入的 step 记为 turn 内的第 1 步。 * 模型在某个 step 里请求工具后，工具结果写回，driver 会为带工具结果的下一模型请求再开一个新 step，路径为 tools → pre-step → step。 * 因此一次工具交互后的续跑请求可用同一 turn 内 step 大于 1 来近似。此启发式无法区分工具续跑与人类中途插话造成的新 step，会把二者都视为续跑。 - * 对暂停语用而言，二者都是模型进一步作答前的一道门，行为可接受。首条纯文字请求 step 等于 1，永不触发暂停。 + * 对暂停语用而言，二者都是模型进一步作答前的一个暂停点，行为可接受。每个 turn 的第一次请求 step 等于 1，永不触发暂停。 * * @param enabled 该会话开关是否开启。 * @param step 本次 pre-step 的 step 序号，为 turn 内序号。 @@ -65,14 +69,14 @@ export function textToMessage(text: string): UserMessage | undefined { /** 会话状态在安装器里的默认初始值。 */ export function initialSessionState(): PauseSessionState { - return { enabled: false, paused: false, waiting: undefined } + return { enabled: false, busy: false, pausePoint: undefined, pendingMessages: [] } } /** 默认每会话开关。 */ export const DEFAULT_ENABLED = false /** - * 会话状态表，以 session id 为键存放每个会话的开关与门。 + * 会话状态表，以 session id 为键存放每个会话的开关与暂停点。 * 同一会话的 pre-step 与 release 串行访问，单线程无需加锁。 */ export class PauseRegistry { @@ -98,40 +102,79 @@ export class PauseRegistry { return this.stateFor(sessionId).enabled } - /** 是否已暂停，即当前挂着一扇在等放行的门。 */ + /** 是否已暂停，即当前挂着一个在等放行的暂停点。 */ isPaused(sessionId: string): boolean { - return this.stateFor(sessionId).paused + return this.stateFor(sessionId).pausePoint !== undefined + } + + /** 该会话的 agent 是否在跑。 */ + isBusy(sessionId: string): boolean { + return this.stateFor(sessionId).busy + } + + /** 待插话队列长度。 */ + pendingMessageCount(sessionId: string): number { + return this.stateFor(sessionId).pendingMessages.length + } + + /** 更新 agent 的运行状态。转为 idle 时暂停点不会再挂上，由安装器负责把队列当场发出去。 */ + setBusy(sessionId: string, busy: boolean): void { + const s = this.stateFor(sessionId) + if (s.busy === busy) return + this.states.set(sessionId, { ...s, busy }) + } + + /** 把一条还没送达的插话排进队列，等下一个暂停点并入，或等 turn 结束时 followup。 */ + queueMessage(sessionId: string, message: UserMessage): void { + const s = this.stateFor(sessionId) + this.states.set(sessionId, { ...s, pendingMessages: [...s.pendingMessages, message] }) + } + + /** 取走全部待插话。 */ + takePendingMessages(sessionId: string): UserMessage[] { + const s = this.stateFor(sessionId) + if (s.pendingMessages.length === 0) return [] + this.states.set(sessionId, { ...s, pendingMessages: [] }) + return s.pendingMessages + } + + /** 把没能并入的待插话放回队首，用于该步被 reject 或补发失败的情形。 */ + requeueMessages(sessionId: string, messages: readonly UserMessage[]): void { + if (messages.length === 0) return + const s = this.stateFor(sessionId) + this.states.set(sessionId, { ...s, pendingMessages: [...messages, ...s.pendingMessages] }) } /** - * 放行一扇门。若该会话当前正挂着门，用给定补充消息放行并返回 true；否则返回 false。 + * 放行暂停点。若该会话当前正挂着暂停点，用给定补充消息放行并返回 true；否则返回 false。 * 放行后状态复位为未暂停，因为这一步已经放行。 */ release(sessionId: string, message: UserMessage | undefined): boolean { const s = this.stateFor(sessionId) - if (s.waiting === undefined) return false - const resolve = s.waiting.resolve - this.states.set(sessionId, { enabled: s.enabled, paused: false, waiting: undefined }) + if (s.pausePoint === undefined) return false + const resolve = s.pausePoint.resolve + this.states.set(sessionId, { ...s, pausePoint: undefined }) resolve(message) return true } - /** 使一扇门进入等待态，并把拉门信号写入状态。 */ - armGate(sessionId: string, resolvers: Pick): void { + /** 挂起一个暂停点，并把放行入口写入状态。 */ + armPausePoint(sessionId: string, resolvers: Pick): void { const s = this.stateFor(sessionId) + // pre-step 一定发生在 running 内，插件热加载可能错过 idle→running 推送，这里顺手补上。 this.states.set(sessionId, { - enabled: s.enabled, - paused: true, - waiting: { resolve: resolvers.resolve, reject: resolvers.reject }, + ...s, + busy: true, + pausePoint: { resolve: resolvers.resolve, reject: resolvers.reject }, }) } - /** 中止等待中的门，例如 turn 被取消时。返回是否确有被中止的门。 */ - abortGate(sessionId: string, reason: unknown): boolean { + /** 中止等待中的暂停点，例如 turn 被取消时。返回是否确有被中止的暂停点。 */ + abortPausePoint(sessionId: string, reason: unknown): boolean { const s = this.stateFor(sessionId) - if (s.waiting === undefined) return false - const reject = s.waiting.reject - this.states.set(sessionId, { enabled: s.enabled, paused: false, waiting: undefined }) + if (s.pausePoint === undefined) return false + const reject = s.pausePoint.reject + this.states.set(sessionId, { ...s, pausePoint: undefined }) reject(reason) return true } diff --git a/src/index.ts b/src/index.ts index d5a9e99..67819b1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,6 +8,7 @@ import { mountRpcChannel, type RpcChannelHandler, type RpcChannelResult } from ' declare module '@deepseek-ai/cordis' { interface Events { 'agent/pre-step'(this: never, payload: PreStepPayload, next: () => Promise): Promise + 'agent/status'(this: never, payload: AgentStatusPayload): void } } @@ -28,6 +29,13 @@ interface PreStepPayload { } type PreStepDecision = { kind: 'reject' } | { kind: 'enter'; messages: UserMessage[]; startsRequestSeries?: true } +/** agent/status 只报 idle 与 running 两个值，用来判断 turn 是否还可能挂下一个暂停点。 */ +interface AgentStatusPayload { + /** followup 用于把 turn 结束时还没送出的插话当场发出去。 */ + agent: { id: string; followup(message: UserMessage): void } + status: 'idle' | 'running' +} + /** 浏览器可随插话提交的图片媒体类型，与附件服务的准入集合一致。 */ type ImageMediaType = 'image/png' | 'image/jpeg' | 'image/webp' | 'image/gif' const IMAGE_MEDIA_TYPES: ReadonlySet = new Set(['image/png', 'image/jpeg', 'image/webp', 'image/gif']) @@ -114,6 +122,9 @@ export function apply(ctx: Context, config: Config = {}): void { const defaultEnabled = config.defaultEnabled === true const registry = new PauseRegistry(defaultEnabled) + // 记住每个 agent 的活引用，空闲时按回车可以把插话直接 followup 成新 turn。 + const agents = new Map() + const handler: RpcChannelHandler = async (endpoint, payload) => { try { switch (endpoint) { @@ -121,22 +132,51 @@ export function apply(ctx: Context, config: Config = {}): void { const { sessionId, enabled } = payload as EnableBody if (typeof sessionId !== 'string' || typeof enabled !== 'boolean') return rpcError('setEnabled requires sessionId:string, enabled:boolean') registry.setEnabled(sessionId, enabled) - // 关闭开关时应放行任何正挂着的门，避免 driver 卡在暂停态无法前进。 + // 关闭开关时应放行任何正挂着的暂停点，避免 driver 卡在暂停态无法前进。 if (!enabled && registry.isPaused(sessionId)) registry.release(sessionId, undefined) return ok({ paused: registry.isPaused(sessionId) }) } case 'release': { const { sessionId, text, images } = payload as ReleaseBody if (typeof sessionId !== 'string') return rpcError('release requires sessionId:string') - // 没有门时直接返回，避免白存图片留下无主附件。 - if (!registry.isPaused(sessionId)) return ok({ released: false }) + const paused = registry.isPaused(sessionId) + const textLength = typeof text === 'string' ? text.trim().length : 0 + const imageCount = Array.isArray(images) ? images.length : 0 + if (!paused) { + // 没有暂停点时由 agent 是否在跑来裁决：还在跑说明下一个暂停点会来，消息先排队； + // 已经空闲则不会再有暂停点，有 agent 活引用就直接 followup 开一条新 turn。 + if (textLength === 0 && imageCount === 0) { + return ok({ released: false, busy: registry.isBusy(sessionId), queued: false }) + } + const message = await buildReleaseMessage(ctx, text, images) + if (message === undefined) return ok({ released: false, busy: registry.isBusy(sessionId), queued: false }) + // 构建消息期间 turn 可能已经结束，这种时候直接 followup 开新 turn。 + if (!registry.isBusy(sessionId)) { + const agent = agents.get(sessionId) + if (agent !== undefined) { + // 没有暂停点也不会再有：按了就发出去，直接开一条新 turn。 + agent.followup(message) + return ok({ released: false, busy: false, queued: true }) + } + // 没有 agent 活引用，主机现在发不出去。不静默入队，交回客户端暂存，下次按回车再试。 + return ok({ released: false, busy: false, queued: false }) + } + registry.queueMessage(sessionId, message) + return ok({ released: false, busy: true, queued: true }) + } const message = await buildReleaseMessage(ctx, text, images) - return ok({ released: registry.release(sessionId, message) }) + const released = registry.release(sessionId, message) + return ok({ released, busy: registry.isBusy(sessionId), queued: false }) } case 'status': { const { sessionId } = payload as StatusBody if (typeof sessionId !== 'string') return rpcError('status requires sessionId:string') - return ok({ enabled: registry.enabled(sessionId), paused: registry.isPaused(sessionId) }) + return ok({ + enabled: registry.enabled(sessionId), + paused: registry.isPaused(sessionId), + busy: registry.isBusy(sessionId), + pendingMessageCount: registry.pendingMessageCount(sessionId), + }) } default: return rpcError('unknown endpoint ' + endpoint) } @@ -147,29 +187,86 @@ export function apply(ctx: Context, config: Config = {}): void { // connection.rpc.handle 自 0.1.2-rc.1 起即不可用，登记路由时解析 webServer 会抛 without inject，故改为插件自注册通道。 mountRpcChannel(ctx, '/dsh-pause', handler) + // agent/status 是判断「暂停点还会不会来」的唯一依据：running 时下一个迟早挂上，idle 时永远不会。 + ctx.on('agent/status', (payload: AgentStatusPayload) => { + const sessionId = payload.agent.id + agents.set(sessionId, payload.agent) + const busy = payload.status === 'running' + if (registry.isBusy(sessionId) !== busy) registry.setBusy(sessionId, busy) + if (busy) return + // turn 已经结束，排队的插话等不到暂停点了。按了回车就一定发出去：直接 followup 开一条新 turn。 + // 不因 busy 未变而提前返回：队列里可能有暂停点该步被 reject 时放回的消息，要借这次派发再试。 + const leftover = registry.takePendingMessages(sessionId) + if (leftover.length === 0) return + // 让本次 status 派发先结束，避免 followup 唤醒 driver 时重入同一个事件。 + queueMicrotask(() => { + const failed: UserMessage[] = [] + for (const message of leftover) { + try { + payload.agent.followup(message) + } catch { + failed.push(message) + } + } + // 补发失败时放回队列，等下一次状态派发或暂停点再试。 + if (failed.length > 0) registry.requeueMessages(sessionId, failed) + }) + }) + ctx.on('agent/pre-step', async (payload: PreStepPayload, next: () => Promise) => { const sessionId = payload.agent.id const state = registry.stateFor(sessionId) if (payload.step <= 1 || !state.enabled) return next() + registry.setBusy(sessionId, true) - let resolveGate: (message: UserMessage | undefined) => void = () => {} - let rejectGate: (reason: unknown) => void = () => {} - const gate = new Promise((resolve, reject) => { resolveGate = resolve; rejectGate = reject }) - registry.armGate(sessionId, { resolve: resolveGate, reject: rejectGate }) + // 人类在暂停点挂上之前就按了回车，视作预放行这一步：不再挂暂停点，直接把插话并入本步。 + const preapproved = registry.takePendingMessages(sessionId) + if (preapproved.length > 0) { + try { + const decision = await next() + if (decision.kind === 'reject') { + registry.requeueMessages(sessionId, preapproved) + return decision + } + return { ...decision, messages: [...decision.messages, ...preapproved] } + } catch (error) { + // driver 抛错同样算这一步没发出去，队列放回，绝不让消息跟着异常一起消失。 + registry.requeueMessages(sessionId, preapproved) + throw error + } + } + + let resolvePausePoint: (message: UserMessage | undefined) => void = () => {} + let rejectPausePoint: (reason: unknown) => void = () => {} + const pausePoint = new Promise((resolve, reject) => { + resolvePausePoint = resolve + rejectPausePoint = reject + }) + registry.armPausePoint(sessionId, { resolve: resolvePausePoint, reject: rejectPausePoint }) - const onAbort = (): void => { registry.abortGate(sessionId, payload.signal.reason) } + const onAbort = (): void => { registry.abortPausePoint(sessionId, payload.signal.reason) } if (payload.signal.aborted) onAbort() else payload.signal.addEventListener('abort', onAbort, { once: true }) let carried: UserMessage | undefined - try { carried = await gate } catch { /* aborted */ } finally { payload.signal.removeEventListener('abort', onAbort) } + try { carried = await pausePoint } catch { /* aborted */ } finally { payload.signal.removeEventListener('abort', onAbort) } - const decision = await next() - if (decision.kind === 'reject') return decision + let decision: PreStepDecision + try { + decision = await next() + } catch (error) { + if (carried !== undefined) registry.requeueMessages(sessionId, [carried]) + throw error + } + if (decision.kind === 'reject') { + // 这一步被 reject 时放行的插话没有送达，放回队列而不是丢掉。 + if (carried !== undefined) registry.requeueMessages(sessionId, [carried]) + return decision + } return carried === undefined ? decision : { ...decision, messages: [...decision.messages, carried] } }) } export { PauseRegistry } from './controller.ts' export { shouldPause, textToMessage, initialSessionState, DEFAULT_ENABLED } from './controller.ts' -export type { PauseSessionState, ReleaseRequest } from './controller.ts' +export type { PauseSessionState, PausePoint } from './controller.ts' diff --git a/tests/controller.spec.ts b/tests/controller.spec.ts index 81cdaa6..a63e209 100644 --- a/tests/controller.spec.ts +++ b/tests/controller.spec.ts @@ -45,30 +45,71 @@ describe('PauseRegistry', () => { reg.setEnabled('s', true) expect(reg.enabled('s')).toBe(true) }) - it('arm/release 门控：release 携带补充消息并复位', async () => { + it('arm/release 暂停点：release 携带补充消息并复位', async () => { const reg = new PauseRegistry(false) reg.setEnabled('s', true) const message = textToMessage('补一句') - const gate = new Promise((resolve) => { - reg.armGate('s', { resolve, reject: () => {} }) + const pausePoint = new Promise((resolve) => { + reg.armPausePoint('s', { resolve, reject: () => {} }) }) expect(reg.isPaused('s')).toBe(true) const released = reg.release('s', message) expect(released).toBe(true) - await expect(gate).resolves.toBe(message) + await expect(pausePoint).resolves.toBe(message) expect(reg.isPaused('s')).toBe(false) }) - it('无门时 release 返回 false', () => { + it('没有暂停点时 release 返回 false', () => { const reg = new PauseRegistry(false) expect(reg.release('s', undefined)).toBe(false) }) - it('abort 中止门并复位', async () => { + it('abort 中止暂停点并复位', async () => { const reg = new PauseRegistry(false) - const gate = new Promise((_, reject) => { - reg.armGate('s', { resolve: () => {}, reject }) + const pausePoint = new Promise((_, reject) => { + reg.armPausePoint('s', { resolve: () => {}, reject }) }) - reg.abortGate('s', new Error('cancelled')) - await expect(gate).rejects.toThrow('cancelled') + reg.abortPausePoint('s', new Error('cancelled')) + await expect(pausePoint).rejects.toThrow('cancelled') + expect(reg.isPaused('s')).toBe(false) + }) + it('没有暂停点时 release 返回 false 且不动队列', () => { + const reg = new PauseRegistry(false) + expect(reg.release('s', undefined)).toBe(false) + expect(reg.pendingMessageCount('s')).toBe(0) + }) + it('待插话队列先进先出，take 一次取走全部', () => { + const reg = new PauseRegistry(false) + const first = textToMessage('第一条')! + const second = textToMessage('第二条')! + reg.queueMessage('s', first) + reg.queueMessage('s', second) + expect(reg.pendingMessageCount('s')).toBe(2) + expect(reg.takePendingMessages('s')).toEqual([first, second]) + expect(reg.pendingMessageCount('s')).toBe(0) + expect(reg.takePendingMessages('s')).toEqual([]) + }) + it('该步被 reject 时把待插话放回队首', () => { + const reg = new PauseRegistry(false) + const first = textToMessage('第一条')! + reg.queueMessage('s', first) + const taken = reg.takePendingMessages('s') + reg.requeueMessages('s', taken) + expect(reg.pendingMessageCount('s')).toBe(1) + expect(reg.takePendingMessages('s')[0]).toBe(first) + }) + it('busy 只记录 agent 是否在跑，不动队列', () => { + const reg = new PauseRegistry(false) + reg.setBusy('s', true) + reg.queueMessage('s', textToMessage('还在跑')!) + reg.setBusy('s', false) + expect(reg.isBusy('s')).toBe(false) + expect(reg.pendingMessageCount('s')).toBe(1) + }) + it('isPaused 直接由暂停点是否存在推出', () => { + const reg = new PauseRegistry(false) + expect(reg.isPaused('s')).toBe(false) + reg.armPausePoint('s', { resolve: () => {}, reject: () => {} }) + expect(reg.isPaused('s')).toBe(true) + reg.release('s', undefined) expect(reg.isPaused('s')).toBe(false) }) it('stateFor 幂等返回同一会话状态对象', () => { @@ -76,6 +117,11 @@ describe('PauseRegistry', () => { expect(reg.stateFor('x')).toBe(reg.stateFor('x')) }) it('initialSessionState 默认关且未暂停', () => { - expect(initialSessionState()).toEqual({ enabled: false, paused: false, waiting: undefined }) + expect(initialSessionState()).toEqual({ + enabled: false, + busy: false, + pausePoint: undefined, + pendingMessages: [], + }) }) }) diff --git a/tests/index.spec.ts b/tests/index.spec.ts index be1894f..45d18b6 100644 --- a/tests/index.spec.ts +++ b/tests/index.spec.ts @@ -8,6 +8,7 @@ import { afterAll, beforeAll, describe, expect, it } from 'vitest' import { apply } from '../src/index.ts' type PreStepListener = (payload: any, next: () => Promise) => Promise +type StatusListener = (payload: { agent: { id: string }; status: 'idle' | 'running' }) => void type RouteHandler = (req: http.IncomingMessage, res: http.ServerResponse) => void | Promise /** apply 自注册的路由处理器，由最近一次 harness 捕获。 */ @@ -53,6 +54,7 @@ function fakeAttachments() { /** 造一个桩 ctx：捕获 apply 注册的 pre-step 监听，并接住自注册的 prefix 路由。 */ function harness(enabledDefault = false) { let preStepListener: PreStepListener | undefined + let statusListener: StatusListener | undefined const scope = { connection: { requestRejection: () => undefined }, webServer: { @@ -67,17 +69,35 @@ function harness(enabledDefault = false) { const ctx = { on(_name: string, listener: any) { if (_name === 'agent/pre-step') preStepListener = listener as PreStepListener + if (_name === 'agent/status') statusListener = listener as StatusListener return () => true }, inject(_deps: string[], cb: (s: unknown) => void) { cb(scope) }, get(name: string) { return name === 'attachments' ? attachments.store : undefined }, } + const followed: { sessionId: string; message: any }[] = [] apply(ctx as any, { defaultEnabled: enabledDefault }) - return { get preStep() { return preStepListener! }, attachments } + return { + get preStep() { return preStepListener! }, + status(sessionId: string, status: 'idle' | 'running', followupThrows = false) { + statusListener!({ + agent: { + id: sessionId, + followup(message: any) { + if (followupThrows) throw new Error('followup failed') + followed.push({ sessionId, message }) + }, + }, + status, + }) + }, + followed, + attachments, + } } describe('apply host wiring', () => { - it('开关关时 pre-step 直接放行不拉门', async () => { + it('开关关时 pre-step 直接放行不挂暂停点', async () => { const h = harness(false) let calledNext = false const decision = await h.preStep( @@ -98,7 +118,7 @@ describe('apply host wiring', () => { expect(calledNext).toBe(true) }) - it('开启+续跑时拉门，release 空文字后原样放行', async () => { + it('开启+续跑时挂暂停点，release 空文字后原样放行', async () => { const h = harness(true) // 模拟 driver 在 await pre-step；先开 listener 后放行 const base = { kind: 'enter' as const, messages: [] } @@ -107,7 +127,7 @@ describe('apply host wiring', () => { { agent: { id: 's1' }, turn: 1, step: 2, signal: new AbortController().signal, messages: [] }, async () => { nextCalledAfter = true; return base }, ) - // next 尚未被调用，说明门被拉住 + // next 尚未被调用，说明暂停点被挂住 await Promise.resolve() expect(nextCalledAfter).toBe(false) // RPC status 应报 paused @@ -163,7 +183,7 @@ describe('apply host wiring', () => { ]) }) - it('图片类型非法时拒绝放行且门保持', async () => { + it('图片类型非法时拒绝放行且暂停点保持', async () => { const h = harness(true) const base = { kind: 'enter' as const, messages: [] } const p = h.preStep( @@ -183,10 +203,129 @@ describe('apply host wiring', () => { await p }) - it('无门时 release 返回 released:false', async () => { + it('没有暂停点且空闲时 release 返回 released:false 与 busy:false', async () => { harness(true) const rel = await rpc('release', { sessionId: 'nobody', text: '' }) - expect(rel.value.released).toBe(false) + expect(rel.value).toMatchObject({ released: false, busy: false, queued: false }) + }) + + it('没有暂停点但在跑时把插话排进队列，下一个暂停点自动并入', async () => { + const h = harness(true) + h.status('s6', 'running') + const queued = await rpc('release', { sessionId: 's6', text: ' 赶在暂停点前面 ' }) + expect(queued.value).toMatchObject({ released: false, busy: true, queued: true }) + expect((await rpc('status', { sessionId: 's6' })).value.pendingMessageCount).toBe(1) + + // 暂停点还没挂上时不必再按回车：这一步会直接带走队列里的消息。 + const p = h.preStep( + { agent: { id: 's6' }, turn: 1, step: 2, signal: new AbortController().signal, messages: [] }, + async () => ({ kind: 'enter', messages: [] }), + ) + const decision = await p + expect(decision.messages).toHaveLength(1) + expect(decision.messages[0].content).toMatchObject([{ type: 'text', text: '赶在暂停点前面' }]) + expect((await rpc('status', { sessionId: 's6' })).value.pendingMessageCount).toBe(0) + }) + + it('没有暂停点且在跑时空回车不排队', async () => { + const h = harness(true) + h.status('s7', 'running') + const rel = await rpc('release', { sessionId: 's7', text: '' }) + expect(rel.value).toMatchObject({ released: false, busy: true, queued: false }) + expect((await rpc('status', { sessionId: 's7' })).value.pendingMessageCount).toBe(0) + }) + + it('turn 结束时排队的插话由 followup 当场发出', async () => { + const h = harness(true) + h.status('s8', 'running') + await rpc('release', { sessionId: 's8', text: '没赶上' }) + expect(h.followed).toHaveLength(0) + h.status('s8', 'idle') + await Promise.resolve() + expect(h.followed).toHaveLength(1) + expect(h.followed[0]!.sessionId).toBe('s8') + expect(h.followed[0]!.message.content).toMatchObject([{ type: 'text', text: '没赶上' }]) + expect((await rpc('status', { sessionId: 's8' })).value.pendingMessageCount).toBe(0) + }) + + it('插话已被下一个暂停点带走时 idle 不再重复 followup', async () => { + const h = harness(true) + h.status('s9', 'running') + await rpc('release', { sessionId: 's9', text: '插一句' }) + await h.preStep( + { agent: { id: 's9' }, turn: 1, step: 2, signal: new AbortController().signal, messages: [] }, + async () => ({ kind: 'enter', messages: [] }), + ) + h.status('s9', 'idle') + await Promise.resolve() + expect(h.followed).toHaveLength(0) + }) + + it('没有暂停点、空闲且尚无 agent 引用时返回未送达，不入队', async () => { + const h = harness(true) + const rel = await rpc('release', { sessionId: 's10', text: '先别丢' }) + expect(rel.value).toMatchObject({ released: false, busy: false, queued: false }) + expect((await rpc('status', { sessionId: 's10' })).value.pendingMessageCount).toBe(0) + }) + + it('没有暂停点、空闲且已有 agent 引用时 release 直接 followup', async () => { + const h = harness(true) + h.status('s12', 'idle') + const rel = await rpc('release', { sessionId: 's12', text: '直接发' }) + expect(rel.value).toMatchObject({ released: false, busy: false, queued: true }) + expect(h.followed).toHaveLength(1) + expect(h.followed[0]!.message.content).toMatchObject([{ type: 'text', text: '直接发' }]) + }) + + it('暂停点放行后该步 reject，插话放回队列并在 idle 补发', async () => { + const h = harness(true) + const p = h.preStep( + { agent: { id: 's13' }, turn: 1, step: 2, signal: new AbortController().signal, messages: [] }, + async () => ({ kind: 'reject' as const }), + ) + await Promise.resolve() + await rpc('release', { sessionId: 's13', text: '别丢' }) + await expect(p).resolves.toEqual({ kind: 'reject' }) + expect((await rpc('status', { sessionId: 's13' })).value.pendingMessageCount).toBe(1) + h.status('s13', 'idle') + await Promise.resolve() + expect(h.followed).toHaveLength(1) + expect(h.followed[0]!.message.content).toMatchObject([{ type: 'text', text: '别丢' }]) + }) + + it('预放行时 next 抛错，队列放回', async () => { + const h = harness(true) + h.status('s14', 'running') + await rpc('release', { sessionId: 's14', text: '放回' }) + await expect(h.preStep( + { agent: { id: 's14' }, turn: 1, step: 2, signal: new AbortController().signal, messages: [] }, + async () => { throw new Error('driver failed') }, + )).rejects.toThrow('driver failed') + expect((await rpc('status', { sessionId: 's14' })).value.pendingMessageCount).toBe(1) + }) + + it('idle 补发失败时插话放回队列，下一次派发再试', async () => { + const h = harness(true) + h.status('s15', 'running') + await rpc('release', { sessionId: 's15', text: '重试' }) + h.status('s15', 'idle', true) + await Promise.resolve() + expect((await rpc('status', { sessionId: 's15' })).value.pendingMessageCount).toBe(1) + h.status('s15', 'idle') + await Promise.resolve() + expect(h.followed).toHaveLength(1) + expect(h.followed[0]!.message.content).toMatchObject([{ type: 'text', text: '重试' }]) + }) + + it('空闲状态重复推送时待发队列只补发一次', async () => { + const h = harness(true) + h.status('s11', 'running') + await rpc('release', { sessionId: 's11', text: '只发一次' }) + h.status('s11', 'idle') + await Promise.resolve() + h.status('s11', 'idle') + await Promise.resolve() + expect(h.followed).toHaveLength(1) }) it('RPC 参数校验返回 error', async () => { @@ -195,7 +334,7 @@ describe('apply host wiring', () => { expect(bad.ok).toBe(false) }) - it('暂停中把开关关闭会放行正挂着的门，不卡死', async () => { + it('暂停中把开关关闭会放行正挂着的暂停点，不卡死', async () => { const h = harness(true) const base = { kind: 'enter' as const, messages: [] } const p = h.preStep( @@ -205,7 +344,7 @@ describe('apply host wiring', () => { await Promise.resolve() const before = await rpc('status', { sessionId: 's3' }) expect(before.value.paused).toBe(true) - // 关闭开关：应放行门并让 pre-step 走完 + // 关闭开关：应放行暂停点并让 pre-step 走完 const off = await rpc('setEnabled', { sessionId: 's3', enabled: false }) expect(off.value.paused).toBe(false) const decision = await p