From 68fb17304136f019f344967d1e416d2cc3a7d4bb Mon Sep 17 00:00:00 2001 From: qqqys Date: Tue, 3 Nov 2026 01:52:03 -0800 Subject: [PATCH] feat(acp): run cron and background-notification turns on the shared loop (#16190) * feat(acp): run cron and background-notification turns on the shared loop With experimental.unifiedAgentLoop on, ACP's cron turns (loop ticks included) run as Cron rounds and its background-notification turns as Notification rounds on the shared round, the shared tool batch and the client's stop point, with the Todo Stop Guard as their stop policy. The prompt loop's shared round moves into #runSharedRound, which the prompt, cron and notification loops share; the prompt loop passes in what only a prompt has. ACP keeps the scheduling around these turns, records the cron prompt once it reached history and the notification with its task, and shows each notification response as one background message. userTurnRecordedByHost also covers Cron and Notification sends, whose record the host writes. The ACP parity suite gains a background-shell scenario whose result starts a notification turn. Part of #15327. * fix(acp): settle shared cron and notification turns as the current path does A cron turn that loop protection or a declined approval stopped is no longer delivered to its channel, as the current path returns before delivery. The cron prompt is recorded as the first event of its response arrives, before the client records the response, so the record keeps the prompt ahead of its answer. A cron or notification round that threw on its aborted signal, which only the session aborts, ends as cancelled rather than firing StopFailure. The cron turn counts its rounds as they start, so a round that throws still counts. The automatic loop emits a response's usage through the shared round's emitResponseUsage instead of repeating it, and the shared round destructures what it aliased. * docs(design): list the cron prompt a cancel before its first request keeps --- docs/design/2026-10-23-unified-agent-loop.md | 33 +- .../2026-10-23-unified-agent-loop.zh-CN.md | 33 +- .../cli/acp-agent-loop-parity.test.ts | 58 +- .../acp-integration/session/Session.test.ts | 357 +++- .../src/acp-integration/session/Session.ts | 1760 ++++++++++------- packages/core/src/core/client.test.ts | 33 + packages/core/src/core/client.ts | 40 +- 7 files changed, 1593 insertions(+), 721 deletions(-) diff --git a/docs/design/2026-10-23-unified-agent-loop.md b/docs/design/2026-10-23-unified-agent-loop.md index 9a8135ceda7..61fd84b7553 100644 --- a/docs/design/2026-10-23-unified-agent-loop.md +++ b/docs/design/2026-10-23-unified-agent-loop.md @@ -176,11 +176,11 @@ A policy affects only the surface that passes it. Some surfaces did part of the interaction step themselves before they moved, and keep doing it on the shared path for a while. A send option then tells the client not to do that part, so it does not run twice. Each such option is a transition step, not a lasting API: it is listed here with the step that removes it or turns it into an input. -| Option | Why the host keeps the part | Removed by | -| ------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `toolLoopGuardsByHost` | ACP's tool step runs the daemon tool-loop guards: the per-turn tool-call cap, repeated calls, the exploration reminder and the repeated-tool-failure guard | PR 2b for a fresh prompt's tool batches: the client's guards apply, and ACP keeps the ones only it has, its parameter-error, partial-edit and repeated-tool-failure stops, as its policy. A session with live tools and an Agent Host session keep ACP's tool step and this option until PR 2d | -| `interactionSpanByHost` | ACP's span is parented to the daemon's trace | an input instead: the client joins the span the host started for the prompt; before the default changes for ACP | -| `userTurnRecordedByHost` | ACP records the prompt as typed, with its display text, attachments, resource links and daemon prompt id, before slash commands run, and takes the turn's file-history snapshot before its first send, so a prompt cancelled before it is sent still has one; on a retry, a continue or a restore it has started the todo work chain; on a Goal round it recorded the runtime continuation's message | an input instead: the host passes that record and the client writes it; before the default changes for ACP | +| Option | Why the host keeps the part | Removed by | +| ------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `toolLoopGuardsByHost` | ACP's tool step runs the daemon tool-loop guards: the per-turn tool-call cap, repeated calls, the exploration reminder and the repeated-tool-failure guard | PR 2b for a fresh prompt's tool batches: the client's guards apply, and ACP keeps the ones only it has, its parameter-error, partial-edit and repeated-tool-failure stops, as its policy. A session with live tools and an Agent Host session keep ACP's tool step and this option until PR 2d | +| `interactionSpanByHost` | ACP's span is parented to the daemon's trace | an input instead: the client joins the span the host started for the prompt; before the default changes for ACP | +| `userTurnRecordedByHost` | ACP records the prompt as typed, with its display text, attachments, resource links and daemon prompt id, before slash commands run, and takes the turn's file-history snapshot before its first send, so a prompt cancelled before it is sent still has one; on a retry, a continue or a restore it has started the todo work chain; on a Goal round it recorded the runtime continuation's message; on a cron round it records the prompt once its request reached history, and on a background-notification round the notification with the task it reports | an input instead: the host passes that record and the client writes it; before the default changes for ACP | `toolCallPreparations`, `memoryRecallQuery`, `stopPolicies` and `historyPromptId` are not in the table: the first asks for an extra event, the second gives recall the prompt's text, which the host expanded before the send, the third adds the host's reasons to continue at a stop point ([Stop-point continuations](#stop-point-continuations)), and the fourth gives a `Retry` the history identity of the attempt it replaces; none turns anything off. PR 2c-1 removed `stopHookRunByHost`, with which ACP fired the Stop hook in its own Stop loop. @@ -351,6 +351,21 @@ Input a round took as its response finished is answered within the turn, as toda The core changes of this step: `userTurnRecordedByHost` also covers a `Goal` round, whose message the host recorded, and a stop policy's continuation keeps the Goal fields of its send, so it runs under the same permit. +### ACP cron and background-notification turns (PR 2c-4) + +PR 2c-4 puts ACP's last two kinds of turn on the shared round, the shared tool batch and the client's stop point: the cron turns the scheduler fires, loop ticks included, and the turns a background task's result starts. ACP keeps the scheduling around the turn as today: the queues, the admission of a notification turn, the echo of a cron prompt, the start and end notifications, and the channel delivery of a scheduled reply. + +The prompt loop's shared round becomes a method that the three loops share. The prompt loop passes it what only a prompt has: putting a re-entered turn back when its request never went out, the Goal marks, and the loop stop that pauses a Goal. The cron and notification loops run their tool batches around it as the prompt loop does. + +| Turn | With the setting on | +| ----------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| cron | a `Cron` round; the client resets the turn's memory state, adds its system and todo reminders and recalls memory, as for the other surfaces; ACP records the prompt once its request reached history, as today | +| background notification | a `Notification` round; ACP records the notification and adds its system and todo reminders, as today, and shows each response as one background message once it ended | + +Both kinds of turn run the Todo Stop Guard as their stop policy, after the Stop hook. Queued input does not cut their responses short, as today, and is taken at a tool boundary; it no longer continues a notification turn at its stop point, but starts the next turn, as on a prompt since PR 2c-1. A cron turn keeps ACP's interaction span; a notification turn, which has none on ACP's side, gets the client's. + +The core change of this step: `userTurnRecordedByHost` also covers a `Cron` and a `Notification` send, whose prompt or notification the host records. + ## Migration | PR | Content | @@ -428,7 +443,8 @@ A difference fails the scenario unless it is listed with a linked issue. - a UserPromptSubmit hook that blocks; - a Stop hook that blocks once, and one that blocks up to the cap; - the Todo Stop Guard continuing past unfinished todos until its attempts are spent; -- a Goal whose first continuation stops on its own and whose second proposes it complete, which the verifier accepts. +- a Goal whose first continuation stops on its own and whose second proposes it complete, which the verifier accepts; +- a background shell whose result arrives after the prompt ended and starts a notification turn. A continuation's request runs under a prompt id of its own on the current loop and under the turn's prompt id on the shared loop; the suite compares them as one. @@ -471,6 +487,11 @@ The suite runs in the integration job. A scheduled job also runs the whole headl - a retry, a continue and a restore start the client's interaction: its loop detector, exploration budget and turn budget start over, and their first round does not count toward the session turn cap; - a continue whose first request never goes out, cancelled or failed, puts the interrupted user run back as it was, instead of keeping it as one merged entry, and the worktree notice the merged entry took stays pending for the next turn; - from PR 2c-3, loop protection that stops a Goal turn, the client's or ACP's, pauses the Goal with the loop as the reason, as on the other surfaces, instead of finishing the turn and letting the runtime schedule the next continuation; + - from PR 2c-4, a cron or background-notification turn runs what a prompt runs on the shared path: the client's loop heuristics and session turn cap, the Stop hook and its continuations, StopFailure when its model request fails, and the shared tool batch with the client's tool-call guards, where a tool that ends the turn ends it, unless the cron job delivers its reply to a channel; + - a cron turn's system reminders are the client's, and it carries the client's memory recall; a notification turn gets an interaction span; + - input queued during a notification turn, and the turn's own background results, no longer continue it at its stop point; they start the next turn; + - a cron turn cancelled between rounds keeps the results of its last tool batch in history, as a prompt does, and a notification turn whose response is retried shows only the retried text; + - a cron turn cancelled after its echo and before its first request keeps its prompt in history and in the record, as a cron turn cancelled while its request is going out does today; The ACP parity suite covers the common cases unchanged and asserts the date reminder, UserPromptSubmit on tool results, cancelling while that hook runs, and that a batch's approvals come before its calls run. diff --git a/docs/design/2026-10-23-unified-agent-loop.zh-CN.md b/docs/design/2026-10-23-unified-agent-loop.zh-CN.md index 5418fd17992..daedc634ba2 100644 --- a/docs/design/2026-10-23-unified-agent-loop.zh-CN.md +++ b/docs/design/2026-10-23-unified-agent-loop.zh-CN.md @@ -176,11 +176,11 @@ headless 的主循环以 `maxRounds: 1` 运行。因此它为每一轮做的准 有些入口在迁移前自己做了交互步骤中的一部分工作，迁到共享路径后还会继续自己做一段时间。这时用一个发送选项告诉 client 跳过这部分，避免执行两次。这类选项只是迁移期的过渡手段，不是长期 API。下表逐项列出，并写明由哪一步把它删除或改成输入。 -| 选项 | 宿主为什么保留这部分 | 由哪一步移除 | -| ------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `toolLoopGuardsByHost` | ACP 的工具步骤运行 daemon 工具循环守卫：每轮工具调用上限、重复调用、探索提醒，以及重复工具失败守卫 | PR 2b 迁移全新 prompt 的工具批：client 的守卫生效，只有 ACP 才有的守卫，即它的参数错误、半截编辑和重复工具失败停止，作为 ACP 的策略保留。带 live 工具的会话和 Agent Host 会话保留 ACP 的工具步骤和这个选项，直到 PR 2d | -| `interactionSpanByHost` | ACP 的 span 挂在 daemon 的 trace 下 | 改为输入：client 加入宿主为该 prompt 开启的 span；ACP 修改默认值之前完成 | -| `userTurnRecordedByHost` | ACP 在 slash 命令执行前，按用户输入的原样记录 prompt，连同显示文本、附件、resource link 和 daemon prompt id 一起记录，并在首次发送前为这一轮建立文件历史快照，所以发出前就被取消的 prompt 也有快照；在 retry、continue 或 restore 上，宿主已经启动了 todo 工作链；在 Goal 模型轮上，宿主已经记录了 runtime 续写的消息 | 改为输入：宿主传入这条记录，由 client 写入；ACP 修改默认值之前完成 | +| 选项 | 宿主为什么保留这部分 | 由哪一步移除 | +| ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `toolLoopGuardsByHost` | ACP 的工具步骤运行 daemon 工具循环守卫：每轮工具调用上限、重复调用、探索提醒，以及重复工具失败守卫 | PR 2b 迁移全新 prompt 的工具批：client 的守卫生效，只有 ACP 才有的守卫，即它的参数错误、半截编辑和重复工具失败停止，作为 ACP 的策略保留。带 live 工具的会话和 Agent Host 会话保留 ACP 的工具步骤和这个选项，直到 PR 2d | +| `interactionSpanByHost` | ACP 的 span 挂在 daemon 的 trace 下 | 改为输入：client 加入宿主为该 prompt 开启的 span；ACP 修改默认值之前完成 | +| `userTurnRecordedByHost` | ACP 在 slash 命令执行前，按用户输入的原样记录 prompt，连同显示文本、附件、resource link 和 daemon prompt id 一起记录，并在首次发送前为这一轮建立文件历史快照，所以发出前就被取消的 prompt 也有快照；在 retry、continue 或 restore 上，宿主已经启动了 todo 工作链；在 Goal 模型轮上，宿主已经记录了 runtime 续写的消息；在 cron 模型轮上，宿主在请求进入历史后记录 prompt；在后台通知模型轮上，宿主记录通知及其报告的任务 | 改为输入：宿主传入这条记录，由 client 写入；ACP 修改默认值之前完成 | `toolCallPreparations`、`memoryRecallQuery`、`stopPolicies` 和 `historyPromptId` 不在表中：第一个只是请求一个额外的事件，第二个把宿主在发送前展开的 prompt 原文交给召回，第三个加入宿主在 stop 点继续的理由（见 [stop 点续发](#stop-点续发)），第四个让 `Retry` 沿用它所替代那次尝试的历史标识；四者都不关闭任何行为。PR 2c-1 删除了 `stopHookRunByHost`，ACP 原来用它在自己的 Stop 循环里触发 Stop hook。 @@ -351,6 +351,21 @@ PR 2c-3 让 ACP 的 Goal 轮次使用共享模型轮、共享工具批和 client 这一步对 core 的改动：`userTurnRecordedByHost` 也覆盖 `Goal` 模型轮，表示宿主已经记录了它的消息；stop policy 的续发请求保留原请求的 Goal 字段，因此在同一个 permit 下运行。 +### ACP 的 cron 和后台通知轮次（PR 2c-4） + +PR 2c-4 让 ACP 最后两类轮次使用共享模型轮、共享工具批和 client 的 stop 点：调度器触发的 cron 轮次（包括 loop tick），以及后台任务结果启动的轮次。轮次外围的调度和现在一样留在 ACP：队列、通知轮次的准入、cron prompt 的回显、开始和结束通知，以及定时回复的 channel 投递。 + +prompt 循环里的共享模型轮改为三个循环共用的方法。prompt 循环把只有 prompt 才有的部分传给它：请求没有发出时把重新进入的轮次放回原样、Goal 标记，以及暂停 Goal 的循环停止。cron 和通知循环像 prompt 循环一样，在它外面运行自己的工具批。 + +| 轮次 | 设置开启后 | +| -------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | +| cron | `Cron` 模型轮；和其他界面一样，由 client 重置这一轮的记忆状态、加上它的系统提醒和 todo 提醒并召回记忆；ACP 和现在一样在请求进入历史后记录 prompt | +| 后台通知 | `Notification` 模型轮；ACP 和现在一样记录通知、加上它的系统提醒和 todo 提醒，并在每个响应结束后把它作为一条后台消息显示 | + +两类轮次都在 Stop hook 之后运行 Todo Stop Guard 作为 stop 策略。排队的输入和现在一样不会打断它们的响应，在工具边界取入；在 stop 点它不再让通知轮次继续，而是启动下一轮，和 PR 2c-1 以来的 prompt 一样。cron 轮次保留 ACP 的交互 span；通知轮次在 ACP 一侧没有 span，改用 client 的。 + +这一步对 core 的改动：`userTurnRecordedByHost` 也覆盖 `Cron` 和 `Notification` 发送，表示由宿主记录 prompt 或通知。 + ## 迁移计划 | PR | 内容 | @@ -428,7 +443,8 @@ PR 3–6 每一项之后，同样各自跟随默认开启和删除旧 loop 两 - 拦截 prompt 的 UserPromptSubmit hook； - 拦截一次的 Stop hook，以及一直拦截到上限的 Stop hook； - Todo Stop Guard 在还有未完成 todo 时继续，直到用完尝试次数； -- 一个 Goal：第一次续写自行停下，第二次续写提出 Goal 已完成，verifier 接受。 +- 一个 Goal：第一次续写自行停下，第二次续写提出 Goal 已完成，verifier 接受； +- 一个后台 shell：它的结果在 prompt 结束后到达，启动一个通知轮次。 续发请求在现有 loop 上使用自己的 prompt id，在共享 loop 上沿用这一轮的 prompt id；对拍测试把两者视为相同。 @@ -471,6 +487,11 @@ PR 3–6 每一项之后，同样各自跟随默认开启和删除旧 loop 两 - retry、continue 和 restore 会开启 client 的一次交互：循环检测器、探索预算和轮次预算重新开始，它们的第一轮不计入会话轮数上限； - continue 的第一条请求一直没有发出时（被取消或失败），被打断的用户输入按原样放回，而不是作为一条合并后的条目保留，合并条目取走的 worktree 提醒也留给下一轮； - 从 PR 2c-3 起，循环保护（client 的或 ACP 的）停下 Goal 轮次时，和其他界面一样以循环为原因暂停 Goal，而不是结束这一轮、让 runtime 安排下一次续写； + - 从 PR 2c-4 起，cron 或后台通知轮次运行 prompt 在共享路径上运行的内容：client 的循环启发式检查和会话轮次上限、Stop hook 及其续发、模型请求失败时的 StopFailure，以及带 client 工具调用守卫的共享工具批，其中结束轮次的工具会结束这一轮，除非这个 cron 任务把回复投递到 channel； + - cron 轮次的系统提醒改为 client 的，并带上 client 的记忆召回；通知轮次有了交互 span； + - 通知轮次期间排队的输入，以及这一轮自己的后台结果，不再在 stop 点让它继续，而是启动下一轮； + - 在两个模型轮之间被取消的 cron 轮次，和 prompt 一样把最后一个工具批的结果保留在历史中；响应被重试的通知轮次只显示重试后的文本； + - 在回显之后、第一个请求发出之前被取消的 cron 轮次，会把 prompt 保留在历史和记录中，和现在请求发出过程中被取消的 cron 轮次一样； ACP 对拍测试覆盖了保持不变的常见场景，并对日期提醒、工具结果上的 UserPromptSubmit、该 hook 运行时取消，以及一批的审批先于它的调用执行做了断言。 diff --git a/integration-tests/cli/acp-agent-loop-parity.test.ts b/integration-tests/cli/acp-agent-loop-parity.test.ts index 20e7107857c..d6410cd564b 100644 --- a/integration-tests/cli/acp-agent-loop-parity.test.ts +++ b/integration-tests/cli/acp-agent-loop-parity.test.ts @@ -57,6 +57,8 @@ interface Scenario { * this many Goal turns to end. */ goalTurns?: number; + /** Waits after the prompts for this many background-notification turns to end. */ + backgroundTurns?: number; /** The model's reply to a request that does not stream, such as a verifier's. */ sideReply?: (body: JsonObject) => FakeOpenAIResponse; /** Sends `session/cancel` once this file exists in the test directory. */ @@ -94,6 +96,8 @@ interface Run { permissionsAt: number[]; /** The stop reason of each Goal turn, in order. */ goalTurnEnds: unknown[]; + /** The stop reason of each background-notification turn, in order. */ + backgroundTurnEnds: unknown[]; } const PROMPT_TIMEOUT_MS = 60_000; @@ -102,6 +106,11 @@ const PROMPT_TIMEOUT_MS = 60_000; const RANDOM_ID = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/g; +// A background task's id, its process and the run's own home directory. +const BACKGROUND_TASK_ID = /bg_[0-9a-f]{8}/g; +const PROCESS_ID = /(pid>|pid: |pid )\d+/g; +const RUN_HOME = new RegExp(`${ACP_HOME_PREFIX}[A-Za-z0-9]+`, 'g'); + // Fields that differ between two runs of the same scenario. const VOLATILE_KEYS = new Set([ 'sessionId', @@ -155,7 +164,13 @@ function normalize(value: unknown, dir: string, sessionId = ''): unknown { .join('') .replace(CONTINUATION_PROMPT_ID, '$1'); if (sessionId) text = text.split(sessionId).join(''); - const replaced = JSON.parse(text.replace(RANDOM_ID, '')); + const replaced = JSON.parse( + text + .replace(RANDOM_ID, '') + .replace(BACKGROUND_TASK_ID, 'bg_') + .replace(PROCESS_ID, '$1') + .replace(RUN_HOME, ''), + ); const strip = (node: unknown): unknown => { if (Array.isArray(node)) return node.map(strip); if (node && typeof node === 'object') { @@ -312,6 +327,7 @@ const allowOnce = () => ({ const updates: unknown[] = []; const goalTurnEnds: unknown[] = []; + const backgroundTurnEnds: unknown[] = []; const permissions: unknown[] = []; const permissionsAt: number[] = []; const pending = new Map< @@ -378,6 +394,13 @@ const allowOnce = () => ({ goalTurnEnds.push(params['reason']); return; } + if ( + method === '_qwencode/end_turn' && + params?.['source'] === 'background_notification' + ) { + backgroundTurnEnds.push(params['reason']); + return; + } if (typeof id !== 'number' || typeof method !== 'string') return; if (method === 'session/request_permission') { permissions.push(params?.['toolCall']); @@ -461,7 +484,8 @@ const allowOnce = () => ({ } const goalDeadline = Date.now() + PROMPT_TIMEOUT_MS; while ( - goalTurnEnds.length < (scenario.goalTurns ?? 0) && + (goalTurnEnds.length < (scenario.goalTurns ?? 0) || + backgroundTurnEnds.length < (scenario.backgroundTurns ?? 0)) && Date.now() < goalDeadline ) { await new Promise((resolve) => setTimeout(resolve, 50)); @@ -492,6 +516,7 @@ const allowOnce = () => ({ permissions: permissions.map((call) => normalize(call, dir, sessionId)), permissionsAt, goalTurnEnds, + backgroundTurnEnds, }; } @@ -708,6 +733,31 @@ const allowOnce = () => ({ content: '{"decision":"accept","reason":"The forecast was given."}', }), }, + { + name: 'a background shell whose result starts a notification turn', + prompts: ['Run the job in the background.'], + backgroundTurns: 1, + // The job outlasts the prompt, so its result starts a turn of its + // own on both loops. + reply: (_messages, _dir, index) => + index === 0 + ? { + toolCalls: [ + fakeToolCall( + 'run_shell_command', + { + command: 'sleep 2; echo job done', + description: 'Run the job', + is_background: true, + }, + 'call_job', + ), + ], + } + : index === 1 + ? { content: 'The job is running.' } + : { content: 'The job is done.' }, + }, { name: 'a model error', prompts: ['Fail please.'], @@ -796,6 +846,10 @@ const allowOnce = () => ({ expect(shared.stopReasons).toEqual(current.stopReasons); expect(shared.requests).toEqual(current.requests); expect(shared.goalTurnEnds).toEqual(current.goalTurnEnds); + expect(shared.backgroundTurnEnds).toEqual(current.backgroundTurnEnds); + expect(current.backgroundTurnEnds).toHaveLength( + scenario.backgroundTurns ?? 0, + ); if (scenario.goalTurns) { expect(current.goalTurnEnds).toHaveLength(scenario.goalTurns); } diff --git a/packages/cli/src/acp-integration/session/Session.test.ts b/packages/cli/src/acp-integration/session/Session.test.ts index c56e16f042e..dba96705608 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/packages/cli/src/acp-integration/session/Session.test.ts @@ -5499,19 +5499,15 @@ describe('Session', () => { const result = await prompt(); expect(result.stopReason).toBe('end_turn'); - // The prompt ends with its response, as the client's stop point does. - expect(send).toHaveBeenCalledOnce(); - await vi.waitFor(() => - expect(mockChat.sendMessageStream).toHaveBeenCalledWith( - expect.anything(), - expect.objectContaining({ - message: expect.arrayContaining([{ text: 'WORKER_RESULT' }]), - }), - expect.stringContaining('notification'), - undefined, - undefined, - ), - ); + // The prompt ends with its response, as the client's stop point does, + // and the result starts a notification turn of its own. + await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(2)); + expect(send.mock.calls[1][0]).toContainEqual({ text: 'WORKER_RESULT' }); + expect(send.mock.calls[1][2]).toContain('notification'); + expect(send.mock.calls[1][3]).toMatchObject({ + type: core.SendMessageType.Notification, + }); + expect(mockChat.sendMessageStream).not.toHaveBeenCalled(); }); describe('the Todo Stop Guard as a stop policy', () => { @@ -5977,6 +5973,341 @@ describe('Session', () => { ).toBe(true); }); }); + + describe('cron and background-notification turns', () => { + /** + * Runs each round on the mocked client; a round pushes its request + * unless told not to, and runs `afterEvents` once its events are out. + */ + function automaticRounds( + events: (index: number) => RoundEvent[] = () => [], + pushes: (index: number) => boolean = () => true, + afterEvents: (index: number) => Promise | void = () => {}, + ) { + const sends: Array< + [Part[], AbortSignal, string, Record] + > = []; + let pushCount = 0; + mockConfig.isUnifiedAgentLoopEnabled = vi.fn().mockReturnValue(true); + Object.assign(mockChat, { + getUserContentPushCount: vi.fn(() => pushCount), + }); + Object.assign(mockLlmClient, { + admitUserPrompt: vi.fn().mockResolvedValue({ blocked: false }), + finishManagedAutoMemoryRecall: vi.fn(), + sendMessageStream: vi.fn( + ( + parts: Part[], + signal: AbortSignal, + promptId: string, + options: Record, + ) => { + sends.push([parts, signal, promptId, options]); + const index = sends.length - 1; + return (async function* () { + if (pushes(index)) pushCount++; + for (const event of events(index)) yield event; + await afterEvents(index); + })(); + }, + ), + }); + mockChat.sendMessageStream = vi.fn(); + return sends; + } + + /** Fires one in-session cron job once the session starts its scheduler. */ + function fireCron(prompt: string, job: object = {}) { + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue({ + size: 1, + hasPendingWork: true, + start: vi.fn((callback: (job: { prompt: string }) => void) => + callback({ prompt, ...job }), + ), + stop: vi.fn(), + getExitSummary: vi.fn().mockReturnValue(undefined), + }); + } + + const notify = () => + session.enqueueBackgroundNotification({ + displayText: 'Agent completed.', + modelText: '', + taskId: 'task-1', + status: 'completed', + kind: 'agent', + }); + + const backgroundResponses = () => + updates().filter( + (update) => + update.sessionUpdate === 'agent_message_chunk' && + (update._meta as { source?: string } | undefined)?.source === + 'background_notification_response', + ); + + it('runs a cron fire as a Cron round and records its prompt once sent', async () => { + fireCron('nightly report'); + const sends = automaticRounds((index) => + index === 1 ? [text('cron result'), finished()] : [finished()], + ); + + await prompt(); + await vi.waitFor(() => + expect(mockChatRecordingService.recordCronPrompt).toHaveBeenCalled(), + ); + + expect(sends[1][0]).toEqual([{ text: 'nightly report' }]); + expect(sends[1][3]).toMatchObject({ + type: core.SendMessageType.Cron, + userTurnRecordedByHost: true, + interactionSpanByHost: true, + toolCallPreparations: true, + }); + expect(sends[1][3]['stopPolicies']).toHaveLength(1); + expect(mockChatRecordingService.recordCronPrompt).toHaveBeenCalledWith( + [{ text: 'nightly report' }], + 'nightly report', + ); + await vi.waitFor(() => expect(agentText()).toContain('cron result')); + expect(mockChat.sendMessageStream).not.toHaveBeenCalled(); + }); + + it('records a cron prompt before the response its round records', async () => { + fireCron('nightly report'); + const order: string[] = []; + mockChatRecordingService.recordCronPrompt.mockImplementation(() => { + order.push('cron prompt'); + }); + // The client records a response once its stream ends. + automaticRounds( + (index) => + index === 1 ? [text('cron result'), finished()] : [finished()], + () => true, + (index) => { + if (index === 1) order.push('cron response'); + }, + ); + + await prompt(); + await vi.waitFor(() => expect(order).toContain('cron response')); + + expect(order).toEqual(['cron prompt', 'cron response']); + }); + + describe('scheduled delivery', () => { + const scheduledJob = { + id: 'task-1', + lastFiredAt: 1_750_000_000_000, + delivery: { + kind: 'channel', + target: { channelName: 'dingtalk', type: 'chat', id: 'chat-1' }, + }, + }; + const deliveries = () => + vi + .mocked(mockClient.extMethod) + .mock.calls.filter( + ([method]) => method === 'qwen/control/channel-delivery', + ); + + it('delivers a scheduled cron turn that ended on its own', async () => { + fireCron('nightly report', scheduledJob); + automaticRounds((index) => + index === 1 ? [text('cron result'), finished()] : [finished()], + ); + + await prompt(); + + await vi.waitFor(() => expect(deliveries()).toHaveLength(1)); + expect(deliveries()[0][1]).toMatchObject({ + deliveryId: 'task-1:1750000000000', + text: 'cron result', + }); + }); + + it('does not deliver a scheduled cron turn loop protection stopped', async () => { + fireCron('nightly report', scheduledJob); + automaticRounds((index) => + index === 1 + ? [ + text('looping'), + { + type: core.LlmEventType.LoopDetected, + value: { + loopType: core.LoopType.CONSECUTIVE_IDENTICAL_TOOL_CALLS, + }, + } as RoundEvent, + ] + : [finished()], + ); + + await prompt(); + await vi.waitFor(() => + expect( + mockLlmClient.finishManagedAutoMemoryRecall, + ).toHaveBeenCalled(), + ); + // Delivery is submitted on the next tick. + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(deliveries()).toHaveLength(0); + }); + }); + + it('ends a cron turn cancelled mid-response as cancelled, not as a failure', async () => { + fireCron('nightly report'); + const fireStopFailureEvent = vi.fn().mockResolvedValue(undefined); + mockConfig.getHookSystem = vi.fn().mockReturnValue({ + fireStopFailureEvent, + }); + mockConfig.hasHooksForEvent = vi + .fn() + .mockImplementation((name: string) => name === 'StopFailure'); + let release!: () => void; + const released = new Promise((resolve) => { + release = resolve; + }); + automaticRounds( + (index) => (index === 1 ? [text('cron partial')] : [finished()]), + () => true, + async (index) => { + if (index !== 1) return; + await released; + const error = new Error('aborted'); + error.name = 'AbortError'; + throw error; + }, + ); + + await prompt(); + await vi.waitFor(() => expect(agentText()).toContain('cron partial')); + await session.cancelPendingPrompt(); + release(); + await vi.waitFor(() => + expect( + mockLlmClient.finishManagedAutoMemoryRecall, + ).toHaveBeenCalled(), + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(fireStopFailureEvent).not.toHaveBeenCalled(); + expect(agentText()).not.toContain('cron error'); + }); + + it('keeps a cron prompt whose request never went out out of the record', async () => { + fireCron('nightly report'); + const sends = automaticRounds( + () => [finished()], + (index) => index !== 1, + ); + + await prompt(); + await vi.waitFor(() => expect(sends).toHaveLength(2)); + await vi.waitFor(() => + expect( + mockLlmClient.finishManagedAutoMemoryRecall, + ).toHaveBeenCalled(), + ); + + expect( + mockChatRecordingService.recordCronPrompt, + ).not.toHaveBeenCalled(); + }); + + it("sends a cron fire's tool results back as a ToolResult round", async () => { + fireCron('nightly report'); + const execute = useTool(core.Kind.Read, 'read_file'); + const sends = automaticRounds((index) => + index === 1 + ? [ + toolCall({ id: 'call-1', name: 'read_file', args: {} }), + finished(), + ] + : [finished()], + ); + + await prompt(); + await vi.waitFor(() => expect(sends).toHaveLength(3)); + + expect(execute).toHaveBeenCalledOnce(); + expect(sends[2][3]).toMatchObject({ + type: core.SendMessageType.ToolResult, + interactionSpanByHost: true, + }); + expect(sends[2][3]).not.toHaveProperty('userTurnRecordedByHost'); + expect(sends[2][0]).toContainEqual( + expect.objectContaining({ + functionResponse: expect.objectContaining({ id: 'call-1' }), + }), + ); + }); + + it('runs a background notification as a Notification round and shows its response once', async () => { + const sends = automaticRounds(() => [ + text('agent '), + text('done'), + finished(), + ]); + + await notify(); + await vi.waitFor(() => + expect( + backgroundResponses().some( + (update) => + ( + update._meta as { + backgroundTask?: { turnComplete?: boolean }; + } + ).backgroundTask?.turnComplete === true, + ), + ).toBe(true), + ); + + expect(sends).toHaveLength(1); + expect(sends[0][3]).toMatchObject({ + type: core.SendMessageType.Notification, + userTurnRecordedByHost: true, + }); + expect(sends[0][3]).not.toHaveProperty('interactionSpanByHost'); + expect(sends[0][0]).toContainEqual({ text: '' }); + expect( + backgroundResponses().map( + (update) => (update.content as { text: string }).text, + ), + ).toEqual(['agent done', '']); + // Not streamed piece by piece as a prompt's response is. + expect(agentText()).not.toContain('agent '); + }); + + it('ends a notification turn that loop protection stopped as partial', async () => { + automaticRounds(() => [ + text('looping'), + { + type: core.LlmEventType.LoopDetected, + value: { loopType: core.LoopType.CONSECUTIVE_IDENTICAL_TOOL_CALLS }, + } as RoundEvent, + ]); + + await notify(); + await vi.waitFor(() => + expect(mockClient.extNotification).toHaveBeenCalledWith( + '_qwencode/end_turn', + expect.objectContaining({ source: 'background_notification' }), + ), + ); + + const last = backgroundResponses().at(-1)!; + expect( + ( + last._meta as { + backgroundTask: { turnComplete?: boolean; partial?: boolean }; + } + ).backgroundTask, + ).toMatchObject({ turnComplete: true, partial: true }); + }); + }); }); describe('Managed Runtime batch close', () => { diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index 8124cd66729..a78c374b2db 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -949,6 +949,83 @@ const newModelResponseFacts = (): ModelResponseFacts => ({ refusedPartialEdits: new Set(), }); +/** Queued input that may cut a response short; see `#openResponseInterrupt`. */ +interface ResponseInterrupt { + signal: AbortSignal; + watch: (stream: AsyncGenerator) => AsyncGenerator; + /** Opens and closes the window `watch` keeps around a stream. */ + begin: () => void; + end: () => void; + takeInput: ( + onFullTurnModel?: (model: string) => boolean, + ) => Promise; + /** Whether input was taken that `takeInput` will answer. */ + hasTakenInput: () => Promise; + /** + * Ends the round's signal without cutting its response, which already + * finished: the turn goes on to answer the input taken. + */ + keepTurnForInput: () => void; +} + +/** A round of a turn on the shared agent loop; see `#runSharedRound`. */ +interface SharedRound { + message: Content; + promptId: string; + type: SendMessageType; + /** The send options of the round's type and of a Goal turn. */ + options?: Partial; + modelOverride?: string; + stopPolicies?: StopPolicy[]; + sharedToolBatch: boolean; + /** The turn's signal. */ + signal: AbortSignal; + /** + * Only the session aborts the turn's signal, without a reason, so a round + * that threw on it was cancelled, whatever the reason. + */ + signalAbortedBySession?: boolean; + responseCapture: AgentResponseCapture; + responseInterrupt: ResponseInterrupt; + /** Filled with the calls the round's last response made. */ + functionCalls: FunctionCall[]; + toolRequests: ToolCallRequestInfo[]; + preparationTracker: ToolCallPreparationTracker; + /** Names the round in the logs of its tool-call preparations. */ + label: string; + /** Keeps a message whose request never went out. */ + keepUnsent: (message: Content, full: boolean) => void; + /** Whether a request a limit stopped is kept whole, not only its results. */ + keepFullOnStop?: boolean; + /** Called before the request goes out. */ + onOpen?: (open: { + requestUnsent: () => boolean; + emitResponseUsage: () => Promise; + }) => void; + /** Called on each model event and once the round ends. */ + onModelStarted?: () => void; + /** Called once the round ends, with whether its request reached history. */ + onSettled?: (pushed: boolean) => void; + /** Called when the Stop hook continues the turn. */ + onContinuation?: () => void; + /** Called when loop protection stopped the round; may throw. */ + onLoopStop?: ( + loopStop: ({ loopType: LoopType } & LoopStopDetail) | undefined, + ) => PromptResponse['stopReason']; +} + +type SharedRoundResult = + | { ended: { stopReason: PromptResponse['stopReason'] } } + | { + ended?: undefined; + usageMetadata: GenerateContentResponseUsageMetadata | null; + streamStartTime: number; + requestRouteKey: string; + channelDeliveryResponseBlock: ChannelDeliveryResponseBlock | undefined; + /** The round's last response stopped on its own. */ + stoppedOnItsOwn: boolean; + }; + const userPromptBlockedNotice = (reason: string | undefined) => `✗ **UserPromptSubmit blocked**: ${reason || 'No reason provided'}`; @@ -8121,23 +8198,6 @@ export class Session implements SessionContext { stopHookCapped = false; const message = nextMessage; nextMessage = null; - requestRouteKey = await this.#requestRouteKeyForModel( - fullTurnModelOverride ?? this.config.getModel(), - ); - this.#syncPromptTokenCountWithCurrentChat(requestRouteKey); - channelDeliveryResponseBlock = - beginChannelDeliveryResponseBlock(responseCapture); - channelDeliveryCheckpoint = - channelDeliveryResponseBlock?.parts.length ?? 0; - const pushCountBefore = - this.#getCurrentChat().getUserContentPushCount?.() ?? 0; - requestUnsent = () => { - const pushCount = - this.#getCurrentChat().getUserContentPushCount?.(); - return ( - pushCount !== undefined && pushCount <= pushCountBefore - ); - }; // A first request that never went out leaves history as the // re-entered turn found it: a retry's stripped attempt is put // back when the prompt ends, and a continue's orphaned @@ -8155,23 +8215,6 @@ export class Session implements SessionContext { this.#clearPendingRestoreNotices(); } }; - let attempt: ModelResponseFacts = newModelResponseFacts(); - // The user-content push a response answers. A continuation - // sends another request within the round, and its response - // starts its own facts, capture and usage, as a new send - // does on the current path. - let responsePushCount: number | undefined; - let loopStop: - | ({ loopType: LoopType } & LoopStopDetail) - | undefined; - let roundStop: AgentLoopStopReason | undefined; - let streamFailed = false; - let responseOpened = false; - // A response cut short for queued input is given up, as - // `responseInterrupt.watch` gives it up on the current path. - const cutForInput = () => - responseInterrupt.signal.aborted && - !pendingSend.signal.aborted; // Input taken while a response streamed is user text at a // model boundary. const roundType = @@ -8180,31 +8223,6 @@ export class Session implements SessionContext { : message.parts?.some((part) => part.functionResponse) ? SendMessageType.ToolResult : SendMessageType.Steer; - // The error a response reported, thrown once the client has - // finished the round. - let streamError: StructuredError | undefined; - // The usage of a response the round continues past. - emitResponseUsage = async () => { - if (!usageMetadata) return; - this.#recordPromptTokenCount( - usageMetadata, - requestRouteKey, - ); - this.messageRewriter?.flushTurn(pendingSend.signal); - await this.messageEmitter.emitUsageMetadata( - usageMetadata, - '', - Date.now() - streamStartTime, - ); - usageMetadata = null; - }; - roundTookInput = responseInterrupt.hasTakenInput; - roundKeepTurnForInput = responseInterrupt.keepTurnForInput; - // Open from the send, as `watch` is open once the request is - // out: input sent before the model answers cuts the wait. - responseInterrupt.begin(); - // Whether the round's request reached history, on every exit. - let pushed = false; // A Goal turn's model started once its request went out, // before the client can finish the turn at its stop point. const markGoalModelStarted = () => { @@ -8214,373 +8232,101 @@ export class Session implements SessionContext { this.#markGoalModelStarted(goalTurn); }; roundMarkGoalModelStarted = markGoalModelStarted; - try { - ({ stop: roundStop } = await runModelRound( - this.config.getLlmClient(), - { - parts: message.parts ?? [], - promptId, - options: { - type: roundType, - modelOverride: fullTurnModelOverride, - stopPolicies, - interactionSpanByHost: true, - ...(!sharedToolBatch && { - toolLoopGuardsByHost: true, + roundTookInput = responseInterrupt.hasTakenInput; + roundKeepTurnForInput = responseInterrupt.keepTurnForInput; + const roundResult = await this.#runSharedRound({ + message, + promptId, + type: roundType, + options: { + interactionSpanByHost: true, + ...(roundType === SendMessageType.UserQuery && { + userTurnRecordedByHost: true, + workflowModeEligible, + promptAdmission, + // Recall searches the user's words, as the current + // path does, not the resolved files and notices + // `parts` carries. + memoryRecallQuery: promptText, + ...(workflowHumanInput && { + humanInput: workflowHumanInput, + }), + ...(isUserSubmissionTurn && + submittedPrompt.trim().length > 0 && { + submittedPrompt, }), - toolCallPreparations: true, - ...(roundType === SendMessageType.UserQuery && { - userTurnRecordedByHost: true, - workflowModeEligible, - promptAdmission, - // Recall searches the user's words, as the - // current path does, not the resolved files and - // notices `parts` carries. - memoryRecallQuery: promptText, - ...(workflowHumanInput && { - humanInput: workflowHumanInput, - }), - ...(isUserSubmissionTurn && - submittedPrompt.trim().length > 0 && { - submittedPrompt, - }), - }), - ...(roundType === SendMessageType.Retry && { - userTurnRecordedByHost: true, - ...(historyPromptId !== undefined && { - historyPromptId, - }), - ...(promptAdmission && { promptAdmission }), - }), - // The session recorded the continuation's message. - ...(roundType === SendMessageType.Goal && { - userTurnRecordedByHost: true, - }), - ...(goalTurn && - !goalPermitReleasedByClient && - this.#sharedGoalRoundOptions( - goalTurn, - responseInterrupt.signal, - )), - }, - }, - responseInterrupt.signal, - { - onModelEvent: async (event) => { - markGoalModelStarted(); - if (pendingSend.signal.aborted) return; - const pushCount = - this.#getCurrentChat().getUserContentPushCount?.(); - if (pushCount !== responsePushCount) { - if (responsePushCount !== undefined) { - attempt = newModelResponseFacts(); - await emitResponseUsage(); - streamStartTime = Date.now(); - channelDeliveryResponseBlock = - beginChannelDeliveryResponseBlock( - responseCapture, - ); - channelDeliveryCheckpoint = - channelDeliveryResponseBlock?.parts.length ?? 0; - } - responsePushCount = pushCount; - } - // Closed when a response finishes and opened by the - // next model response, not by hook diagnostics. The - // first response event also takes input that arrived - // before the request went out. - if (event.type === LlmEventType.Finished) { - responseOpened = false; - responseInterrupt.end(); - } else if ( - !responseOpened && - (event.type === LlmEventType.Content || - event.type === LlmEventType.Thought || - event.type === LlmEventType.Retry || - event.type === LlmEventType.ModelFallback || - event.type === LlmEventType.ToolCallPreparation || - event.type === LlmEventType.ToolCallRequest || - event.type === LlmEventType.Citation) - ) { - responseOpened = true; - responseInterrupt.begin(); - } - switch (event.type) { - case LlmEventType.Content: - if (!event.value) break; - void this.messageEmitter.emitMessage( - event.value, - 'assistant', - false, - ); - responseCapture.agentOutput.appendText( - event.value, - ); - appendChannelDeliveryResponseText( - channelDeliveryResponseBlock, - event.value, - ); - break; - case LlmEventType.Thought: - if (event.text) { - void this.messageEmitter.emitMessage( - event.text, - 'assistant', - true, - ); - } - break; - case LlmEventType.ToolCallPreparation: - await preparationTracker.prepare(event.value); - break; - case LlmEventType.ToolCallRequest: - if (event.functionCall) { - this.#callResponses.set( - event.functionCall, - attempt, - ); - preparationTracker.resolve([ - event.functionCall, - ]); - functionCalls.push(event.functionCall); - toolRequests.push(event.value); - } - break; - case LlmEventType.Finished: - usageMetadata = - event.value.usageMetadata ?? usageMetadata; - responseCapture.agentOutput.observeFinishReason( - event.value.reason, - ); - if ( - event.value.reason === FinishReason.MAX_TOKENS - ) { - attempt.outputTruncated = true; - } - break; - case LlmEventType.Retry: - case LlmEventType.ModelFallback: { - const continuation = - event.type === LlmEventType.Retry && - event.isContinuation === true; - responseCapture.agentOutput.restartAttempt( - continuation, - ); - if (!continuation) { - rewindChannelDeliveryResponseBlock( - channelDeliveryResponseBlock, - channelDeliveryCheckpoint, - ); - } - await finalizeToolCallPreparations( - preparationTracker, - true, - `main prompt ${event.type}`, - ); - functionCalls.length = 0; - toolRequests.length = 0; - attempt = newModelResponseFacts(); - break; - } - case LlmEventType.ChatCompressed: - if (!event.value) break; - this.loopTickResolver?.resetCache(); - this.#recordCompressionTokenCount( - event.value, - requestRouteKey, - ); - await this.#emitAgentDiagnosticMessageSafely( - formatCompressionDiagnostic( - event.value, - this.config.getModel(), - ), - `Failed to emit compression notification for prompt ${promptId}`, - ); - break; - case LlmEventType.SessionTokenLimitExceeded: - await this.#emitAgentDiagnosticMessageSafely( - `Session token limit exceeded: ${event.value.currentTokens} tokens > ${event.value.limit} limit. ` + - 'Please start a new session or increase the sessionTokenLimit in your settings.json.', - `Failed to emit token limit diagnostic for prompt ${promptId}`, - ); - break; - case LlmEventType.LoopDetected: - loopStop = event.value; - break; - case LlmEventType.UserPromptSubmitBlocked: - // A tool-result round runs UserPromptSubmit too. - await this.messageEmitter.emitAgentMessage( - userPromptBlockedNotice(event.value.reason), - ); - break; - case LlmEventType.HookSystemMessage: - await emitResponseUsage(); - await this.messageEmitter.emitAgentMessage( - event.value, - ); - break; - case LlmEventType.StopHookLoop: - await emitResponseUsage(); - resetToolLoopState(); - // Reported from the second block on, as the - // session's own Stop loop does. - if (event.value.iterationCount > 1) { - await this.messageEmitter.emitStopHookLoop( - event.value.iterationCount, - event.value.reasons, - event.value.stopHookCount, - ); - } - break; - case LlmEventType.Error: - streamError = event.value.error; - break; - default: - break; - } - }, - }, - )); - if (streamError) throw modelStreamError(streamError); - } catch (error) { - streamFailed = true; - if (!cutForInput()) { - // A round that threw before its request pushed the - // message, such as on a tool-result round whose - // UserPromptSubmit hook timed out, keeps it as the stops - // below do. - if (requestUnsent()) { - if (pendingSend.signal.aborted) { - markApiHistoryPrompt(message, responseCapture.turnId); - } - keepUnsent(message, pendingSend.signal.aborted); - } - return this.#endFailedModelRound( - error, - pendingSend.signal, - ); - } - } finally { - markGoalModelStarted(); - pushed = - (this.#getCurrentChat().getUserContentPushCount?.() ?? - 0) > pushCountBefore; + }), + ...(roundType === SendMessageType.Retry && { + userTurnRecordedByHost: true, + ...(historyPromptId !== undefined && { + historyPromptId, + }), + ...(promptAdmission && { promptAdmission }), + }), + // The session recorded the continuation's message. + ...(roundType === SendMessageType.Goal && { + userTurnRecordedByHost: true, + }), + ...(goalTurn && + !goalPermitReleasedByClient && + this.#sharedGoalRoundOptions( + goalTurn, + responseInterrupt.signal, + )), + }, + modelOverride: fullTurnModelOverride, + stopPolicies, + sharedToolBatch, + signal: pendingSend.signal, + responseCapture, + responseInterrupt, + functionCalls, + toolRequests, + preparationTracker, + label: 'main prompt', + keepUnsent, + keepFullOnStop: isContinue, + onOpen: (open) => { + requestUnsent = open.requestUnsent; + emitResponseUsage = open.emitResponseUsage; + }, + onModelStarted: markGoalModelStarted, // A first request in history replaces what the turn // stripped and carries the restore's notices, even when // its response then fails. - if (pushed && turnCount === 1) { + onSettled: (pushed) => { + if (!pushed || turnCount !== 1) return; if (restoresOriginalRetry) retryHistoryReplaced = true; strippedOrphanEntries = null; if (restorePostAnswerNoticesAttached) { this.#clearPendingRestoreNotices(); } - } - responseInterrupt.end(); - // A round that stopped, such as on the client's loop - // guard, runs none of the calls it already yielded. The - // loop guard's calls are skipped, as ACP's guard skips - // them. - if ( - roundStop === 'loop_detected' && - !streamFailed && - !pendingSend.signal.aborted - ) { - await this.#emitLoopSkippedToolCalls( - preparationTracker - .takePending() - .map(({ callId, toolName }) => ({ - callId, - name: toolName, - })), - ); - } else { - await finalizeToolCallPreparations( - preparationTracker, - streamFailed || - pendingSend.signal.aborted || - roundStop !== undefined, - 'main prompt', - ); - } - } - // A round that stopped before its request pushed the message - // keeps it in history, as a send the current path drops does. - if (pendingSend.signal.aborted) { - this.todoStopGuard.suspend(); - if (!pushed) { - markApiHistoryPrompt(message, responseCapture.turnId); - keepUnsent(message, true); - } - return { stopReason: 'cancelled' }; - } - if (roundStop === 'user_prompt_blocked') { - this.todoStopGuard.suspend(); - // A tool-result round the hook blocked closes its calls - // with the block, not with the results it withheld from - // the model, nor with the repair's "retry if needed". - const blockedCalls = (message.parts ?? []).flatMap( - ({ functionResponse }) => - functionResponse?.id - ? [ - { - callId: functionResponse.id, - name: functionResponse.name ?? '', - }, - ] - : [], - ); - // Input taken with the results, already shown as - // delivered, stays in history with the closed calls, as a - // stopped tool run keeps it. - if (!pushed && blockedCalls.length > 0) { - this.#getCurrentChat().addHistory({ - role: 'user', - parts: [ - ...buildSyntheticToolResponseParts( - blockedCalls, - USER_PROMPT_BLOCKED_TOOL_RESULT, - ), - ...(message.parts ?? []).filter( - (part) => !part.functionResponse, - ), - ], - }); - } - return { stopReason: 'end_turn' }; - } - if ( - roundStop === 'session_token_limit' || - roundStop === 'max_session_turns' || - roundStop === 'loop_detected' - ) { - this.todoStopGuard.suspend(); - if (!pushed) { - keepUnsent(message, isContinue); - } - if (roundStop === 'session_token_limit') { - return { stopReason: 'max_tokens' }; - } - if (roundStop === 'max_session_turns') { - return { stopReason: 'max_turn_requests' }; - } - // The client has already paused the Goal for its guard; - // this covers a pause or release that failed there, so - // `#settleGoalTurn` still pauses with the loop reason. - pauseGoalForLoopStop(); - if (rejectOnLoopDetected) { - throw createLoopDetectedTurnError({ - loopType: loopStop?.loopType, - loopDetail: loopStop, - }); - } - return { stopReason: 'end_turn' }; - } + }, + onContinuation: resetToolLoopState, + onLoopStop: (loopStop) => { + // The client has already paused the Goal for its guard; + // this covers a pause or release that failed there, so + // `#settleGoalTurn` still pauses with the loop reason. + pauseGoalForLoopStop(); + if (rejectOnLoopDetected) { + throw createLoopDetectedTurnError({ + loopType: loopStop?.loopType, + loopDetail: loopStop, + }); + } + return 'end_turn'; + }, + }); + if (roundResult.ended) return roundResult.ended; + ({ + usageMetadata, + streamStartTime, + requestRouteKey, + channelDeliveryResponseBlock, + } = roundResult); memoryTasksRanInClient = - roundType !== SendMessageType.Steer && - functionCalls.length === 0 && + roundResult.stoppedOnItsOwn && !stopHookCapped && - !streamFailed && - !cutForInput() && // The client's Goal stop finishes the turn before its // memory tasks, so a round under the permit leaves them // to the session. @@ -8997,21 +8743,617 @@ export class Session implements SessionContext { }); } + /** + * One round of a turn on the shared agent loop: a request through + * `LlmClient` and the continuations the client sends at its stop point, + * shown, counted and kept in history as the session's own rounds are. The + * turn around it runs the tool batch and decides what is sent next. + */ + async #runSharedRound(round: SharedRound): Promise { + const { + message, + promptId, + type: roundType, + signal, + responseCapture, + responseInterrupt, + functionCalls, + toolRequests, + preparationTracker, + keepUnsent, + } = round; + let usageMetadata: GenerateContentResponseUsageMetadata | null = null; + let streamStartTime = Date.now(); + const requestRouteKey = await this.#requestRouteKeyForModel( + round.modelOverride ?? this.config.getModel(), + ); + this.#syncPromptTokenCountWithCurrentChat(requestRouteKey); + let channelDeliveryResponseBlock = + beginChannelDeliveryResponseBlock(responseCapture); + let channelDeliveryCheckpoint = + channelDeliveryResponseBlock?.parts.length ?? 0; + const pushCountBefore = + this.#getCurrentChat().getUserContentPushCount?.() ?? 0; + const requestUnsent = () => { + const pushCount = this.#getCurrentChat().getUserContentPushCount?.(); + return pushCount !== undefined && pushCount <= pushCountBefore; + }; + let attempt: ModelResponseFacts = newModelResponseFacts(); + // The user-content push a response answers. A continuation + // sends another request within the round, and its response + // starts its own facts, capture and usage, as a new send + // does on the current path. + let responsePushCount: number | undefined; + let loopStop: ({ loopType: LoopType } & LoopStopDetail) | undefined; + let roundStop: AgentLoopStopReason | undefined; + let streamFailed = false; + let responseOpened = false; + // A response cut short for queued input is given up, as + // `responseInterrupt.watch` gives it up on the current path. + const cutForInput = () => + responseInterrupt.signal.aborted && !signal.aborted; + // The error a response reported, thrown once the client has + // finished the round. + let streamError: StructuredError | undefined; + // A background turn shows each response as one message once it ended. + let shownResponseBlock: ChannelDeliveryResponseBlock | undefined; + const showBackgroundResponse = async () => { + if ( + !responseCapture.backgroundResponse || + !channelDeliveryResponseBlock || + shownResponseBlock === channelDeliveryResponseBlock + ) { + return; + } + shownResponseBlock = channelDeliveryResponseBlock; + await responseCapture.backgroundResponse( + channelDeliveryResponseBlock.parts.join(''), + ); + }; + // The usage of a response the round continues past. + const emitResponseUsage = async () => { + await showBackgroundResponse(); + if (!usageMetadata) return; + this.#recordPromptTokenCount(usageMetadata, requestRouteKey); + this.messageRewriter?.flushTurn(signal); + await this.messageEmitter.emitUsageMetadata( + usageMetadata, + '', + Date.now() - streamStartTime, + ); + usageMetadata = null; + }; + round.onOpen?.({ requestUnsent, emitResponseUsage }); + // Open from the send, as `watch` is open once the request is + // out: input sent before the model answers cuts the wait. + responseInterrupt.begin(); + // Whether the round's request reached history, on every exit. + let pushed = false; + try { + ({ stop: roundStop } = await runModelRound( + this.config.getLlmClient(), + { + parts: message.parts ?? [], + promptId, + options: { + type: roundType, + modelOverride: round.modelOverride, + stopPolicies: round.stopPolicies, + ...(!round.sharedToolBatch && { + toolLoopGuardsByHost: true, + }), + toolCallPreparations: true, + ...round.options, + }, + }, + responseInterrupt.signal, + { + onModelEvent: async (event) => { + // A Goal turn's model started once its request went out, before + // the client can finish the turn at its stop point; the host + // checks. + round.onModelStarted?.(); + if (signal.aborted) return; + const pushCount = + this.#getCurrentChat().getUserContentPushCount?.(); + if (pushCount !== responsePushCount) { + if (responsePushCount !== undefined) { + attempt = newModelResponseFacts(); + await emitResponseUsage(); + streamStartTime = Date.now(); + channelDeliveryResponseBlock = + beginChannelDeliveryResponseBlock(responseCapture); + channelDeliveryCheckpoint = + channelDeliveryResponseBlock?.parts.length ?? 0; + } + responsePushCount = pushCount; + } + // Closed when a response finishes and opened by the + // next model response, not by hook diagnostics. The + // first response event also takes input that arrived + // before the request went out. + if (event.type === LlmEventType.Finished) { + responseOpened = false; + responseInterrupt.end(); + } else if ( + !responseOpened && + (event.type === LlmEventType.Content || + event.type === LlmEventType.Thought || + event.type === LlmEventType.Retry || + event.type === LlmEventType.ModelFallback || + event.type === LlmEventType.ToolCallPreparation || + event.type === LlmEventType.ToolCallRequest || + event.type === LlmEventType.Citation) + ) { + responseOpened = true; + responseInterrupt.begin(); + } + switch (event.type) { + case LlmEventType.Content: + if (!event.value) break; + // A background response is shown once it ended. + if (!responseCapture.backgroundResponse) { + void this.messageEmitter.emitMessage( + event.value, + 'assistant', + false, + ); + } + responseCapture.agentOutput.appendText(event.value); + appendChannelDeliveryResponseText( + channelDeliveryResponseBlock, + event.value, + ); + break; + case LlmEventType.Thought: + if (event.text) { + void this.messageEmitter.emitMessage( + event.text, + 'assistant', + true, + ); + } + break; + case LlmEventType.ToolCallPreparation: + await preparationTracker.prepare(event.value); + break; + case LlmEventType.ToolCallRequest: + if (event.functionCall) { + this.#callResponses.set(event.functionCall, attempt); + preparationTracker.resolve([event.functionCall]); + functionCalls.push(event.functionCall); + toolRequests.push(event.value); + } + break; + case LlmEventType.Finished: + usageMetadata = event.value.usageMetadata ?? usageMetadata; + responseCapture.agentOutput.observeFinishReason( + event.value.reason, + ); + if (event.value.reason === FinishReason.MAX_TOKENS) { + attempt.outputTruncated = true; + } + break; + case LlmEventType.Retry: + case LlmEventType.ModelFallback: { + const continuation = + event.type === LlmEventType.Retry && + event.isContinuation === true; + responseCapture.agentOutput.restartAttempt(continuation); + if (!continuation) { + rewindChannelDeliveryResponseBlock( + channelDeliveryResponseBlock, + channelDeliveryCheckpoint, + ); + } + await finalizeToolCallPreparations( + preparationTracker, + true, + `${round.label} ${event.type}`, + ); + functionCalls.length = 0; + toolRequests.length = 0; + attempt = newModelResponseFacts(); + break; + } + case LlmEventType.ChatCompressed: + if (!event.value) break; + this.loopTickResolver?.resetCache(); + this.#recordCompressionTokenCount(event.value, requestRouteKey); + await this.#emitAgentDiagnosticMessageSafely( + formatCompressionDiagnostic( + event.value, + this.config.getModel(), + ), + `Failed to emit compression notification for prompt ${promptId}`, + ); + break; + case LlmEventType.SessionTokenLimitExceeded: + await this.#emitAgentDiagnosticMessageSafely( + `Session token limit exceeded: ${event.value.currentTokens} tokens > ${event.value.limit} limit. ` + + 'Please start a new session or increase the sessionTokenLimit in your settings.json.', + `Failed to emit token limit diagnostic for prompt ${promptId}`, + ); + break; + case LlmEventType.LoopDetected: + loopStop = event.value; + break; + case LlmEventType.UserPromptSubmitBlocked: + // A tool-result round runs UserPromptSubmit too. + await this.messageEmitter.emitAgentMessage( + userPromptBlockedNotice(event.value.reason), + ); + break; + case LlmEventType.HookSystemMessage: + await emitResponseUsage(); + await this.messageEmitter.emitAgentMessage(event.value); + break; + case LlmEventType.StopHookLoop: + await emitResponseUsage(); + round.onContinuation?.(); + // Reported from the second block on, as the + // session's own Stop loop does. + if (event.value.iterationCount > 1) { + await this.messageEmitter.emitStopHookLoop( + event.value.iterationCount, + event.value.reasons, + event.value.stopHookCount, + ); + } + break; + case LlmEventType.Error: + streamError = event.value.error; + break; + default: + break; + } + }, + }, + )); + if (streamError) throw modelStreamError(streamError); + } catch (error) { + streamFailed = true; + if (!cutForInput()) { + // A round that threw before its request pushed the + // message, such as on a tool-result round whose + // UserPromptSubmit hook timed out, keeps it as the stops + // below do. + if (requestUnsent()) { + if (signal.aborted) { + markApiHistoryPrompt(message, responseCapture.turnId); + } + keepUnsent(message, signal.aborted); + } + return { + ended: await this.#endFailedModelRound( + error, + signal, + round.signalAbortedBySession, + ), + }; + } + } finally { + round.onModelStarted?.(); + pushed = + (this.#getCurrentChat().getUserContentPushCount?.() ?? 0) > + pushCountBefore; + round.onSettled?.(pushed); + responseInterrupt.end(); + // A round that stopped, such as on the client's loop + // guard, runs none of the calls it already yielded. The + // loop guard's calls are skipped, as ACP's guard skips + // them. + if (roundStop === 'loop_detected' && !streamFailed && !signal.aborted) { + await this.#emitLoopSkippedToolCalls( + preparationTracker.takePending().map(({ callId, toolName }) => ({ + callId, + name: toolName, + })), + ); + } else { + await finalizeToolCallPreparations( + preparationTracker, + streamFailed || signal.aborted || roundStop !== undefined, + round.label, + ); + } + } + // A round that stopped before its request pushed the message + // keeps it in history, as a send the current path drops does. + if (signal.aborted) { + this.todoStopGuard.suspend(); + if (!pushed) { + markApiHistoryPrompt(message, responseCapture.turnId); + keepUnsent(message, true); + } + return { ended: { stopReason: 'cancelled' } }; + } + if (roundStop === 'user_prompt_blocked') { + this.todoStopGuard.suspend(); + // A tool-result round the hook blocked closes its calls + // with the block, not with the results it withheld from + // the model, nor with the repair's "retry if needed". + const blockedCalls = (message.parts ?? []).flatMap( + ({ functionResponse }) => + functionResponse?.id + ? [ + { + callId: functionResponse.id, + name: functionResponse.name ?? '', + }, + ] + : [], + ); + // Input taken with the results, already shown as + // delivered, stays in history with the closed calls, as a + // stopped tool run keeps it. + if (!pushed && blockedCalls.length > 0) { + this.#getCurrentChat().addHistory({ + role: 'user', + parts: [ + ...buildSyntheticToolResponseParts( + blockedCalls, + USER_PROMPT_BLOCKED_TOOL_RESULT, + ), + ...(message.parts ?? []).filter((part) => !part.functionResponse), + ], + }); + } + return { ended: { stopReason: 'end_turn' } }; + } + if ( + roundStop === 'session_token_limit' || + roundStop === 'max_session_turns' || + roundStop === 'loop_detected' + ) { + this.todoStopGuard.suspend(); + await showBackgroundResponse(); + if (!pushed) { + keepUnsent(message, round.keepFullOnStop === true); + } + if (roundStop === 'session_token_limit') { + return { ended: { stopReason: 'max_tokens' } }; + } + if (roundStop === 'max_session_turns') { + return { ended: { stopReason: 'max_turn_requests' } }; + } + return { + ended: { + stopReason: round.onLoopStop + ? round.onLoopStop(loopStop) + : 'end_turn', + }, + }; + } + await showBackgroundResponse(); + return { + usageMetadata, + streamStartTime, + requestRouteKey, + channelDeliveryResponseBlock, + // The client runs the turn's memory tasks when a response stops on + // its own. + stoppedOnItsOwn: + roundType !== SendMessageType.Steer && + functionCalls.length === 0 && + !streamFailed && + !cutForInput(), + }; + } + + /** + * A cron or background-notification turn on the shared agent loop: its + * rounds go through `LlmClient`, its tool batches run as a prompt's do, and + * its stop point is the client's, with the Todo Stop Guard as the session's + * stop policy. Queued input never cuts its responses short and does not + * continue it at its stop point; input is taken at a tool boundary as + * today. The session keeps the turn's message, its record and its + * settlement. + */ + async #runAutomaticTurnOnSharedLoop(turn: { + message: Content; + type: SendMessageType.Cron | SendMessageType.Notification; + /** The send options of the turn's first round. */ + options: Partial; + interactionSpanByHost: boolean; + promptId: string; + signal: AbortSignal; + responseCapture: AgentResponseCapture; + label: string; + /** + * Called once the turn's message reached history: sent, before its + * response is recorded, or kept whole when it was not. + */ + onMessageKept?: (sent: boolean) => void; + /** Called as each round starts. */ + onRound?: () => void; + }): Promise<{ + stopReason: PromptResponse['stopReason']; + /** Loop protection or a declined approval stopped the turn. */ + stoppedShort?: boolean; + }> { + const { promptId, signal, responseCapture } = turn; + const sharedToolBatch = this.#runsToolsOnSharedBatch(); + let toolLoopState = createDaemonToolLoopState('off'); + const resetToolLoopState = () => { + toolLoopState = createDaemonToolLoopState('off'); + }; + let emitResponseUsage = async () => {}; + const stopPolicies = [ + this.#todoStopGuardStopPolicy( + () => emitResponseUsage(), + async () => false, + resetToolLoopState, + () => {}, + () => {}, + ), + ]; + let messageKept = false; + const keepMessage = (sent: boolean) => { + if (messageKept) return; + messageKept = true; + turn.onMessageKept?.(sent); + }; + let nextMessage: Content | null = turn.message; + let rounds = 0; + try { + while (nextMessage !== null) { + rounds++; + turn.onRound?.(); + const first = rounds === 1; + const message: Content = nextMessage; + nextMessage = null; + if (signal.aborted) { + this.todoStopGuard.suspend(); + // The results of the tools that ran stay in history, as a + // cancelled prompt keeps them. + this.#preserveUnsentMessageHistory(message, true); + if (first) keepMessage(false); + return { stopReason: 'cancelled' }; + } + const functionCalls: FunctionCall[] = []; + const toolRequests: ToolCallRequestInfo[] = []; + const preparationTracker = new ToolCallPreparationTracker( + this.toolCallEmitter, + ); + let stoppedShort = false; + let requestUnsent = () => true; + const round = await this.#runSharedRound({ + message, + promptId, + type: first + ? turn.type + : message.parts?.some((part) => part.functionResponse) + ? SendMessageType.ToolResult + : SendMessageType.Steer, + options: { + ...(turn.interactionSpanByHost && { interactionSpanByHost: true }), + ...(first && turn.options), + }, + stopPolicies, + sharedToolBatch, + signal, + signalAbortedBySession: true, + responseCapture, + responseInterrupt: this.#openResponseInterrupt( + signal, + () => true, + false, + ), + functionCalls, + toolRequests, + preparationTracker, + label: turn.label, + keepUnsent: (unsent, full) => { + this.#preserveUnsentMessageHistory(unsent, full); + if (first && full) keepMessage(false); + }, + onOpen: (open) => { + requestUnsent = open.requestUnsent; + emitResponseUsage = open.emitResponseUsage; + }, + // Kept as the first event of its response arrives, before the + // client records the response. + onModelStarted: () => { + if (first && !requestUnsent()) keepMessage(true); + }, + onSettled: (pushed) => { + if (first && pushed) keepMessage(true); + }, + onContinuation: resetToolLoopState, + onLoopStop: () => { + stoppedShort = true; + return 'end_turn'; + }, + }); + if (round.ended) { + return { ...round.ended, stoppedShort }; + } + commitChannelDeliveryResponseBlock( + responseCapture, + round.channelDeliveryResponseBlock, + functionCalls.length > 0, + ); + await emitResponseUsage(); + if (functionCalls.length === 0) break; + const toolRun = sharedToolBatch + ? await this.#runSharedToolBatch( + signal, + promptId, + toolRequests, + toolLoopState, + ) + : await this.runToolCalls( + signal, + promptId, + functionCalls, + toolLoopState, + ); + if (toolRun.stopAfterPermissionCancel || signal.aborted) { + this.todoStopGuard.suspend(); + await this.#preserveStoppedToolRun(toolRun, signal); + return { + stopReason: getAbortAwareEndTurnStopReason(signal), + stoppedShort: true, + }; + } + if ( + await this.#endTurnAfterToolRun( + toolRun, + undefined, + false, + responseCapture.channelDelivery !== undefined, + ) + ) { + return { stopReason: getAbortAwareEndTurnStopReason(signal) }; + } + const next = await this.#buildNextMessageAfterToolRun( + toolRun, + signal, + promptId, + toolLoopState, + undefined, + false, + sharedToolBatch, + ); + nextMessage = next.message; + if (toolRun.loopDetected) { + this.todoStopGuard.suspend(); + await this.#preserveStoppedToolRun(toolRun, signal); + return { + stopReason: getAbortAwareEndTurnStopReason(signal), + stoppedShort: true, + }; + } + } + await this.messageRewriter?.waitForPendingRewrites(); + return { stopReason: getAbortAwareEndTurnStopReason(signal) }; + } finally { + // The client started a cron turn's memory recall. + if (turn.type === SendMessageType.Cron) { + this.config.getLlmClient().finishManagedAutoMemoryRecall(); + } + } + } + /** * Ends a turn whose model round threw. Explicit user cancellation and * session disposal are controlled aborts, and so is the bridge's forwarded - * interruption token; other AbortErrors still surface so infrastructure - * failures are not hidden as cancellations. A failure fires StopFailure - * (fire-and-forget, in place of Stop, as `use-llm-stream.ts` + * interruption token, as is any abort of a cron or notification turn's + * signal, which only the session aborts; other AbortErrors still surface so + * infrastructure failures are not hidden as cancellations. A failure fires + * StopFailure (fire-and-forget, in place of Stop, as `use-llm-stream.ts` * handleFinishedWithErrorEvent does) and is thrown. */ #endFailedModelRound( error: unknown, turnSignal: AbortSignal, + signalAbortedBySession = false, ): { stopReason: 'cancelled' } { if ( turnSignal.aborted && - (turnSignal.reason === USER_CANCEL_ABORT_REASON || + (signalAbortedBySession || + turnSignal.reason === USER_CANCEL_ABORT_REASON || turnSignal.reason === SESSION_DISPOSE_ABORT_REASON || turnSignal.reason === INTERRUPTED_PROMPT_ABORT_REASON) ) { @@ -11723,23 +12065,7 @@ export class Session implements SessionContext { turnSignal: AbortSignal, hasFunctionCalls: () => boolean, enabled: boolean, - ): { - signal: AbortSignal; - watch: (stream: AsyncGenerator) => AsyncGenerator; - /** Opens and closes the window `watch` keeps around a stream. */ - begin: () => void; - end: () => void; - takeInput: ( - onFullTurnModel?: (model: string) => boolean, - ) => Promise; - /** Whether input was taken that `takeInput` will answer. */ - hasTakenInput: () => Promise; - /** - * Ends the round's signal without cutting its response, which already - * finished: the turn goes on to answer the input taken. - */ - keepTurnForInput: () => void; - } { + ): ResponseInterrupt { if (!enabled) { return { signal: turnSignal, @@ -12644,257 +12970,308 @@ export class Session implements SessionContext { _meta: { source: item.source }, }); - // Cron-fired prompts stream through the chat directly and never - // enter LlmClient.sendMessageStream, so core's per-turn reset is - // not on this path; without it a claimed search_memory signature - // or exhausted body ref would persist across cron turns. - this.config - .getMemoryManager() - .resetExhaustedBodyRefsForCurrentTurn(); - - // Prepend session-level system reminders (same rationale as the - // user-query path in #executePrompt). - const cronReminders = await this.#buildInitialSystemReminders(); - const activeTodoReminder = this.config.takeActiveTodoReminder( - promptId, - true, - ); - let nextMessage: Content | null = { - role: 'user', - parts: [ - ...cronReminders, - ...(activeTodoReminder ? [{ text: activeTodoReminder }] : []), - { text: modelText }, - ], - }; - const toolLoopState = createDaemonToolLoopState('off'); - - while (nextMessage !== null) { - turnCount++; - if (ac.signal.aborted) { - this.todoStopGuard.suspend(); - return; + // `experimental.unifiedAgentLoop`: the turn's rounds go through + // `LlmClient` as a Cron send, which resets the turn's memory + // state, adds the system and todo reminders and recalls memory, + // as for the other surfaces. The session records the prompt + // once it reached history. + if (this.config.isUnifiedAgentLoopEnabled?.() === true) { + const result = await this.#runAutomaticTurnOnSharedLoop({ + message: { role: 'user', parts: [{ text: modelText }] }, + type: SendMessageType.Cron, + options: { + userTurnRecordedByHost: true, + todoWorkChainId: item.todoWorkChainId, + }, + interactionSpanByHost: true, + promptId, + signal: ac.signal, + responseCapture, + label: 'cron/loop tick', + onMessageKept: (sent) => { + this.config + .getChatRecordingService() + ?.recordCronPrompt([{ text: modelText }], echoText); + // Committed once the block reached the model, so the next + // tick can tell it unchanged. + if (sent && loopTick) { + this.loopTickResolver?.markDelivered(); + } + }, + onRound: () => { + turnCount++; + }, + }); + if (result.stopReason === 'max_tokens') { + this.#stopCronAfterTokenLimit(); } + // A turn loop protection or a declined approval stopped is + // not delivered, as the current path returns before it. + cronCompleted = + result.stopReason === 'end_turn' && + result.stoppedShort !== true && + !ac.signal.aborted; + } else { + // Cron-fired prompts stream through the chat directly and never + // enter LlmClient.sendMessageStream, so core's per-turn reset is + // not on this path; without it a claimed search_memory signature + // or exhausted body ref would persist across cron turns. + this.config + .getMemoryManager() + .resetExhaustedBodyRefsForCurrentTurn(); - const functionCalls: FunctionCall[] = []; - const preparationTracker = new ToolCallPreparationTracker( - this.toolCallEmitter, + // Prepend session-level system reminders (same rationale as the + // user-query path in #executePrompt). + const cronReminders = await this.#buildInitialSystemReminders(); + const activeTodoReminder = this.config.takeActiveTodoReminder( + promptId, + true, ); - let usageMetadata: GenerateContentResponseUsageMetadata | null = - null; - const streamStartTime = Date.now(); - const sendResult = - await this.#sendMessageStreamWithAutoCompression( - promptId, - nextMessage.parts ?? [], + let nextMessage: Content | null = { + role: 'user', + parts: [ + ...cronReminders, + ...(activeTodoReminder + ? [{ text: activeTodoReminder }] + : []), + { text: modelText }, + ], + }; + const toolLoopState = createDaemonToolLoopState('off'); + + while (nextMessage !== null) { + turnCount++; + if (ac.signal.aborted) { + this.todoStopGuard.suspend(); + return; + } + + const functionCalls: FunctionCall[] = []; + const preparationTracker = new ToolCallPreparationTracker( + this.toolCallEmitter, + ); + let usageMetadata: GenerateContentResponseUsageMetadata | null = + null; + const streamStartTime = Date.now(); + const sendResult = + await this.#sendMessageStreamWithAutoCompression( + promptId, + nextMessage.parts ?? [], + ac.signal, + ); + const responseStream = sendResult.responseStream; + const preserveFullMessage = + !responseStream && sendResult.stopReason === 'cancelled'; + if (!responseStream) { + this.todoStopGuard.suspend(); + this.#preserveUnsentMessageHistory( + nextMessage, + preserveFullMessage, + ); + if (sendResult.stopReason === 'max_tokens') { + this.#stopCronAfterTokenLimit(); + } + } + // Keep transcript replay aligned with model history: persist + // only if the prompt reached the model or was preserved in full. + if ( + turnCount === 1 && + (responseStream || preserveFullMessage) + ) { + this.config + .getChatRecordingService() + ?.recordCronPrompt([{ text: modelText }], echoText); + } + if (!responseStream) return; + const requestRouteKey = sendResult.requestRouteKey; + const channelDeliveryResponseBlock: + | ChannelDeliveryResponseBlock + | undefined = + beginChannelDeliveryResponseBlock(responseCapture); + const channelDeliveryCheckpoint = + channelDeliveryResponseBlock?.parts.length ?? 0; + if (loopTick && turnCount === 1) { + // The block reached the model (the send started); commit it so + // the next tick can detect "unchanged". Deferring the commit + // to here keeps an abort before delivery from poisoning the + // cache into a dangling short reminder. + this.loopTickResolver?.markDelivered(); + } + nextMessage = null; + const messageDisplay = this.#createMessageDisplayDispatcher( ac.signal, ); - const responseStream = sendResult.responseStream; - const preserveFullMessage = - !responseStream && sendResult.stopReason === 'cancelled'; - if (!responseStream) { - this.todoStopGuard.suspend(); - this.#preserveUnsentMessageHistory( - nextMessage, - preserveFullMessage, + + let streamFailed = false; + try { + for await (const resp of responseStream) { + if (ac.signal.aborted) { + this.todoStopGuard.suspend(); + return; + } + + if ( + resp.type === StreamEventType.CHUNK && + resp.value.candidates && + resp.value.candidates.length > 0 + ) { + const candidate = resp.value.candidates[0]; + for (const part of candidate.content?.parts ?? []) { + if (!part.text) continue; + this.messageEmitter.emitMessage( + part.text, + 'assistant', + part.thought, + ); + if (!part.thought) { + responseCapture.agentOutput.appendText(part.text); + appendChannelDeliveryResponseText( + channelDeliveryResponseBlock, + part.text, + ); + messageDisplay?.addChunk(part.text); + } + } + responseCapture.agentOutput.observeFinishReason( + candidate.finishReason, + ); + } + + if ( + resp.type === StreamEventType.CHUNK && + resp.value.usageMetadata + ) { + usageMetadata = resp.value.usageMetadata; + } + + if (resp.type === StreamEventType.CHUNK) { + await preparationTracker.observe(resp.value); + if (resp.value.functionCalls) { + preparationTracker.resolve(resp.value.functionCalls); + functionCalls.push(...resp.value.functionCalls); + } + } + if ( + resp.type === StreamEventType.RETRY || + resp.type === StreamEventType.MODEL_FALLBACK + ) { + responseCapture.agentOutput.restartAttempt( + resp.type === StreamEventType.RETRY && + resp.isContinuation === true, + ); + if ( + resp.type === StreamEventType.MODEL_FALLBACK || + !resp.isContinuation + ) { + rewindChannelDeliveryResponseBlock( + channelDeliveryResponseBlock, + channelDeliveryCheckpoint, + ); + } + await finalizeToolCallPreparations( + preparationTracker, + true, + `cron/loop tick ${resp.type}`, + ); + functionCalls.length = 0; + } + if (resp.type === StreamEventType.COMPRESSED) { + // In-send compression rewrote the shared history; + // invalidate every retained route count (the + // pre-send hook never sees this path). + this.#recordCompressionTokenCount( + resp.info, + requestRouteKey, + ); + } + } + } catch (error) { + streamFailed = true; + throw error; + } finally { + try { + await finalizeToolCallPreparations( + preparationTracker, + streamFailed || ac.signal.aborted, + 'cron/loop tick', + ); + } finally { + // is_final (skipped on abort) delivered and drained on + // every exit path, same as the interactive prompt loops. + await messageDisplay?.finish(); + } + } + + commitChannelDeliveryResponseBlock( + responseCapture, + channelDeliveryResponseBlock, + functionCalls.length > 0, ); - if (sendResult.stopReason === 'max_tokens') { + + if (usageMetadata) { + this.#recordPromptTokenCount( + usageMetadata, + requestRouteKey, + ); + if (this.messageRewriter) { + this.messageRewriter.flushTurn(ac.signal); + } + const durationMs = Date.now() - streamStartTime; + await this.messageEmitter.emitUsageMetadata( + usageMetadata, + '', + durationMs, + ); + } + + if (functionCalls.length > 0) { + const toolRun = await this.runToolCalls( + ac.signal, + promptId, + functionCalls, + toolLoopState, + ); + if ( + toolRun.stopAfterPermissionCancel || + ac.signal.aborted + ) { + this.todoStopGuard.suspend(); + await this.#preserveStoppedToolRun(toolRun, ac.signal); + return; + } + const nextAfterTools = + await this.#buildNextMessageAfterToolRun( + toolRun, + ac.signal, + promptId, + toolLoopState, + ); + nextMessage = nextAfterTools.message; + if (toolRun.loopDetected) { + this.todoStopGuard.suspend(); + await this.#preserveStoppedToolRun(toolRun, ac.signal); + return; + } + } + } + let stopReason: PromptResponse['stopReason'] = 'end_turn'; + if (this.todoStopGuard.needsStopInspection) { + const guardStop = await this.#handleStopHookLoop( + ac, + promptId, + false, + undefined, + false, + undefined, + responseCapture, + ); + stopReason = guardStop.stopReason; + if (guardStop.stopReason === 'max_tokens') { this.#stopCronAfterTokenLimit(); } } - // Keep transcript replay aligned with model history: persist - // only if the prompt reached the model or was preserved in full. - if ( - turnCount === 1 && - (responseStream || preserveFullMessage) - ) { - this.config - .getChatRecordingService() - ?.recordCronPrompt([{ text: modelText }], echoText); - } - if (!responseStream) return; - const requestRouteKey = sendResult.requestRouteKey; - const channelDeliveryResponseBlock: - | ChannelDeliveryResponseBlock - | undefined = - beginChannelDeliveryResponseBlock(responseCapture); - const channelDeliveryCheckpoint = - channelDeliveryResponseBlock?.parts.length ?? 0; - if (loopTick && turnCount === 1) { - // The block reached the model (the send started); commit it so - // the next tick can detect "unchanged". Deferring the commit - // to here keeps an abort before delivery from poisoning the - // cache into a dangling short reminder. - this.loopTickResolver?.markDelivered(); - } - nextMessage = null; - const messageDisplay = this.#createMessageDisplayDispatcher( - ac.signal, - ); - - let streamFailed = false; - try { - for await (const resp of responseStream) { - if (ac.signal.aborted) { - this.todoStopGuard.suspend(); - return; - } - - if ( - resp.type === StreamEventType.CHUNK && - resp.value.candidates && - resp.value.candidates.length > 0 - ) { - const candidate = resp.value.candidates[0]; - for (const part of candidate.content?.parts ?? []) { - if (!part.text) continue; - this.messageEmitter.emitMessage( - part.text, - 'assistant', - part.thought, - ); - if (!part.thought) { - responseCapture.agentOutput.appendText(part.text); - appendChannelDeliveryResponseText( - channelDeliveryResponseBlock, - part.text, - ); - messageDisplay?.addChunk(part.text); - } - } - responseCapture.agentOutput.observeFinishReason( - candidate.finishReason, - ); - } - - if ( - resp.type === StreamEventType.CHUNK && - resp.value.usageMetadata - ) { - usageMetadata = resp.value.usageMetadata; - } - - if (resp.type === StreamEventType.CHUNK) { - await preparationTracker.observe(resp.value); - if (resp.value.functionCalls) { - preparationTracker.resolve(resp.value.functionCalls); - functionCalls.push(...resp.value.functionCalls); - } - } - if ( - resp.type === StreamEventType.RETRY || - resp.type === StreamEventType.MODEL_FALLBACK - ) { - responseCapture.agentOutput.restartAttempt( - resp.type === StreamEventType.RETRY && - resp.isContinuation === true, - ); - if ( - resp.type === StreamEventType.MODEL_FALLBACK || - !resp.isContinuation - ) { - rewindChannelDeliveryResponseBlock( - channelDeliveryResponseBlock, - channelDeliveryCheckpoint, - ); - } - await finalizeToolCallPreparations( - preparationTracker, - true, - `cron/loop tick ${resp.type}`, - ); - functionCalls.length = 0; - } - if (resp.type === StreamEventType.COMPRESSED) { - // In-send compression rewrote the shared history; - // invalidate every retained route count (the - // pre-send hook never sees this path). - this.#recordCompressionTokenCount( - resp.info, - requestRouteKey, - ); - } - } - } catch (error) { - streamFailed = true; - throw error; - } finally { - try { - await finalizeToolCallPreparations( - preparationTracker, - streamFailed || ac.signal.aborted, - 'cron/loop tick', - ); - } finally { - // is_final (skipped on abort) delivered and drained on - // every exit path, same as the interactive prompt loops. - await messageDisplay?.finish(); - } - } - - commitChannelDeliveryResponseBlock( - responseCapture, - channelDeliveryResponseBlock, - functionCalls.length > 0, - ); - - if (usageMetadata) { - this.#recordPromptTokenCount(usageMetadata, requestRouteKey); - if (this.messageRewriter) { - this.messageRewriter.flushTurn(ac.signal); - } - const durationMs = Date.now() - streamStartTime; - await this.messageEmitter.emitUsageMetadata( - usageMetadata, - '', - durationMs, - ); - } - - if (functionCalls.length > 0) { - const toolRun = await this.runToolCalls( - ac.signal, - promptId, - functionCalls, - toolLoopState, - ); - if (toolRun.stopAfterPermissionCancel || ac.signal.aborted) { - this.todoStopGuard.suspend(); - await this.#preserveStoppedToolRun(toolRun, ac.signal); - return; - } - const nextAfterTools = - await this.#buildNextMessageAfterToolRun( - toolRun, - ac.signal, - promptId, - toolLoopState, - ); - nextMessage = nextAfterTools.message; - if (toolRun.loopDetected) { - this.todoStopGuard.suspend(); - await this.#preserveStoppedToolRun(toolRun, ac.signal); - return; - } - } + cronCompleted = stopReason === 'end_turn' && !ac.signal.aborted; } - let stopReason: PromptResponse['stopReason'] = 'end_turn'; - if (this.todoStopGuard.needsStopInspection) { - const guardStop = await this.#handleStopHookLoop( - ac, - promptId, - false, - undefined, - false, - undefined, - responseCapture, - ); - stopReason = guardStop.stopReason; - if (guardStop.stopReason === 'max_tokens') { - this.#stopCronAfterTokenLimit(); - } - } - cronCompleted = stopReason === 'end_turn' && !ac.signal.aborted; } catch (error) { if (ac.signal.aborted) { this.todoStopGuard.suspend(); @@ -13997,6 +14374,35 @@ export class Session implements SessionContext { ...notificationParts, ], }; + // `experimental.unifiedAgentLoop`: the turn's rounds go through + // `LlmClient` as a Notification send. The session keeps the + // notification's record and reminders, and shows each response + // as one background message once it ended. + if (this.config.isUnifiedAgentLoopEnabled?.() === true) { + const result = await this.#runAutomaticTurnOnSharedLoop({ + message: nextMessage, + type: SendMessageType.Notification, + options: { + userTurnRecordedByHost: true, + todoWorkChainId: item.todoWorkChainId, + }, + interactionSpanByHost: false, + promptId, + signal: ac.signal, + responseCapture: { + agentOutput: new AgentOutputMessageCapture(this.config), + backgroundResponse: emitBackgroundResponse, + }, + label: 'background notification', + }); + await finishBackgroundNotificationTurn( + ac.signal.aborted ? 'cancelled' : result.stopReason, + ac.signal.aborted || + result.stopReason !== 'end_turn' || + result.stoppedShort === true, + ); + return; + } const toolLoopState = createDaemonToolLoopState('off'); while (nextMessage !== null) { diff --git a/packages/core/src/core/client.test.ts b/packages/core/src/core/client.test.ts index d629968a581..bbad34bc1dd 100644 --- a/packages/core/src/core/client.test.ts +++ b/packages/core/src/core/client.test.ts @@ -11099,6 +11099,39 @@ Other open files: ); }); + it.each([ + { type: SendMessageType.Cron, record: 'recordCronPrompt' as const }, + { + type: SendMessageType.Notification, + record: 'recordNotification' as const, + }, + ])( + 'leaves the record of a $type send to a host that records it', + async ({ type, record }) => { + const recorder = { + recordCronPrompt: vi.fn(), + recordNotification: vi.fn(), + recordAttributionSnapshot: vi.fn(), + recordFileHistorySnapshot: vi.fn(), + }; + vi.mocked(mockConfig.getChatRecordingService).mockReturnValue( + recorder as unknown as ReturnType, + ); + mockTurnRunFn.mockReturnValue(textTurn('Hello')); + installChat(); + + await run([{ text: 'tick' }], `prompt-host-${type}`, { + type, + userTurnRecordedByHost: true, + }); + expect(mockTurnRunFn).toHaveBeenCalledOnce(); + expect(recorder[record]).not.toHaveBeenCalled(); + + await run([{ text: 'tick' }], `prompt-client-${type}`, { type }); + expect(recorder[record]).toHaveBeenCalledOnce(); + }, + ); + /** A recall that never settles; returns a spy on its abort listener. */ const recallAbortHandler = () => { const abortHandler = vi.fn(); diff --git a/packages/core/src/core/client.ts b/packages/core/src/core/client.ts index 4867e54e064..f364d6519c5 100644 --- a/packages/core/src/core/client.ts +++ b/packages/core/src/core/client.ts @@ -342,7 +342,9 @@ export interface SendMessageOptions { * a prompt, so the client does none of them. On a Retry, the host started * the todo work chain and keeps the attempt's recorded user message, so the * client neither starts the chain nor records the message again. On a Goal - * turn, the host recorded the runtime continuation's message. + * turn, the host recorded the runtime continuation's message. On a Cron or + * Notification send, the host records the prompt or the notification + * itself. */ userTurnRecordedByHost?: boolean; /** @@ -4559,15 +4561,17 @@ export class LlmClient { // under them without losing the resumed info item this record exists to // restore. See `ChatRecord.deliveredTurn` for that accepted imprecision // and the test pinning it. - this.config - .getChatRecordingService() - ?.recordNotification( - request, - options?.notificationDisplayText, - undefined, - goalPermit, - /* deliveredTurn */ true, - ); + if (!options?.userTurnRecordedByHost) { + this.config + .getChatRecordingService() + ?.recordNotification( + request, + options?.notificationDisplayText, + undefined, + goalPermit, + /* deliveredTurn */ true, + ); + } } // Notifications start a fresh Turn with a new prompt_id, so the loop @@ -4720,13 +4724,15 @@ export class LlmClient { // record user/cron message for session management if (messageType === SendMessageType.Cron) { - this.config - .getChatRecordingService() - ?.recordCronPrompt( - request, - options?.notificationDisplayText, - goalPermit, - ); + if (!options?.userTurnRecordedByHost) { + this.config + .getChatRecordingService() + ?.recordCronPrompt( + request, + options?.notificationDisplayText, + goalPermit, + ); + } } else if (!options?.userTurnRecordedByHost) { const recorder = this.config.getChatRecordingService(); recorder?.recordUserMessage(