From a85370b38f04165351bfe7239e02bfa1736174b9 Mon Sep 17 00:00:00 2001 From: qqqys Date: Tue, 3 Nov 2026 23:36:48 -0800 Subject: [PATCH] feat(cli): run the Ink TUI's model rounds on the shared loop's round (#16206) * feat(cli): run the Ink TUI's model rounds on the shared loop's round With experimental.unifiedAgentLoop on, every model round of the Ink TUI goes through the shared loop's runModelRound, which collects the tool calls, drops those of an attempt a retry or fallback replaced, and turns the client's terminal events into a stop reason. The TUI keeps its stream processor and its tool step. Adds the Ink parity suite. Part of #15327. * test(cli): tighten the Ink parity suite and fix the no-AK script pin - add the suite to the no-AK script pin - check each run's request count, and through the debug log that only the run with the setting on sends its rounds on the shared loop - compare the system prompt, the tools offered and recorded usage - run each scenario once, without the root config's retries, with a source pin - describe the suite as it is in the design doc - say in the setting's description that the Ink TUI sends its model requests through the shared loop --- docs/design/2026-10-23-unified-agent-loop.md | 42 +- .../2026-10-23-unified-agent-loop.zh-CN.md | 42 +- docs/users/configuration/settings.md | 2 +- .../interactive/ink-agent-loop-parity.test.ts | 562 +++++++++++++ package.json | 2 +- packages/cli/src/config/settingsSchema.ts | 2 +- .../cli/src/ui/hooks/use-llm-stream.test.tsx | 152 ++++ packages/cli/src/ui/hooks/use-llm-stream.ts | 794 ++++++++++-------- packages/core/src/core/agent-loop.test.ts | 66 +- .../schemas/settings.schema.json | 2 +- .../web-shell/client/settings/messages.ts | 2 +- .../tests/ink-agent-loop-parity-retry.test.ts | 31 + scripts/tests/no-ak-integration-ci.test.js | 1 + 13 files changed, 1304 insertions(+), 396 deletions(-) create mode 100644 integration-tests/interactive/ink-agent-loop-parity.test.ts create mode 100644 scripts/tests/ink-agent-loop-parity-retry.test.ts diff --git a/docs/design/2026-10-23-unified-agent-loop.md b/docs/design/2026-10-23-unified-agent-loop.md index 6fc290aaf4c..ad1f42e15f5 100644 --- a/docs/design/2026-10-23-unified-agent-loop.md +++ b/docs/design/2026-10-23-unified-agent-loop.md @@ -366,6 +366,28 @@ Both kinds of turn run the Todo Stop Guard as their stop policy, after the Stop The core change of this step: `userTurnRecordedByHost` also covers a `Cron` and a `Notification` send, whose prompt or notification the host records. +### The Ink TUI on the shared loop (PR 3) + +The Ink TUI already sends every model round through `LlmClient.sendMessageStream`, from one function, `submitQuery`, for every kind of turn. Its loop is split by React instead: a response that requested tools hands them to the scheduler and returns, and the scheduler's completion callback sends the results as a new `ToolResult` turn. Moving the TUI means awaiting that batch, so it moves in these steps: + +| Step | Content | +| ---- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| 3a | every model round on the loop's `runModelRound`; the TUI keeps its tool step; the Ink parity suite | +| 3b | the tool batch of a turn awaited on the shared batch, over the TUI's `InteractiveModToolScheduler`, with the display fed from callbacks; duplicates, finalizing and recording are the loop's; the TUI's own steps (its history repair, the tool-use summary, the stops for a quota model switch and for spent background-agent capacity, the split by interaction owner) become host hooks | +| 3c-1 | input taken at a tool boundary through the loop's next round; a retry | +| 3c-2 | cron, background-notification, teammate and Goal turns | +| 3d | default on, then remove the current loop | + +PR 3a puts every model round of the TUI on `runModelRound`. The round sends the same request with the same options, so the TUI's stream processor keeps handling every event as today. What moves is the part of the round the loop already owns: + +| The TUI today | With the setting on | +| ---------------------------------------------------------------------- | ------------------------------------------------------------ | +| the stream processor collects `ToolCallRequest` events | `runModelRound` collects them | +| the processor drops the collected calls on `Retry` and `ModelFallback` | `runModelRound` drops them | +| a detected loop keeps the collected calls from being scheduled | any stop reason of the round keeps them from being scheduled | + +A stop reason other than a detected loop comes from an event the client sends before it asks the model (the session turn cap, the session token limit, a blocked prompt) or from a cancel, which ends the round as today, so its round has no calls to schedule. The TUI therefore behaves the same with the setting on and off. + ## Migration | PR | Content | @@ -381,13 +403,17 @@ The core change of this step: `userTurnRecordedByHost` also covers a `Cron` and | 2c-3 | ACP: Goal turns | | 2c-4 | ACP: cron and background-notification turns | | 2d | ACP: default on, then remove its current loop | -| 3 | Ink TUI | +| 3a | Ink TUI: every model round on the loop's round; the TUI keeps its tool step; the Ink parity suite | +| 3b | Ink TUI: the tool batch awaited on the shared batch | +| 3c-1 | Ink TUI: input at a tool boundary through the loop's next round; a retry | +| 3c-2 | Ink TUI: cron, background-notification, teammate and Goal turns | +| 3d | Ink TUI: default on, then remove its current loop | | 4 | OpenTUI | | 5 | AgentCore on a `Turn` round (subagents, background and forked agents, workflows) | | 6 | speculation | | — | Hosted Harness: separate evaluation | -Each of PRs 3–6 is followed by its own default-on and removal steps. +Each of PRs 4–6 is followed by its own default-on and removal steps. A surface counts as moved when it passes its parity suite with the setting on and its known differences are listed. Each surface then changes its default on its own, without waiting for the others. Its default changes when: @@ -448,6 +474,18 @@ A difference fails the scenario unless it is listed with a linked issue. 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. +**Ink scenarios** (`integration-tests/interactive/ink-agent-loop-parity.test.ts`). The TUI runs in a pseudo-terminal against the fake server. The suite compares the two runs' request bodies as sent, with the system prompt, the tools offered and each message's `reasoning_content`, and their session's chat records, with usage. It also checks each run's number of requests, and, through the debug log, that only the run with the setting on sends its rounds on the shared loop. A scenario that sends and records nothing, such as a blocked prompt, checks the notice on each run's last screen instead; the screen is not compared otherwise, since it repaints on timers: + +- plain text, and streamed text with a thought; +- a thought before a tool, whose second request replays the reasoning; +- one tool, and two reads in one response; +- an edit cut off by the output limit; +- a stream that drops and is retried; +- two prompts in one session; +- a model error; +- a UserPromptSubmit hook that blocks; +- a Stop hook that blocks once. + The suite runs in the integration job. The nightly E2E run, and one started by hand, also runs the whole integration suite with the setting on, in a Linux leg without a sandbox; it is the scheduled job a surface's default change waits on. ## Compatibility 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 4b674656751..dd3f4856b48 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 @@ -366,6 +366,28 @@ prompt 循环里的共享模型轮改为三个循环共用的方法。prompt 循 这一步对 core 的改动：`userTurnRecordedByHost` 也覆盖 `Cron` 和 `Notification` 发送，表示由宿主记录 prompt 或通知。 +### Ink TUI 迁到共享 loop（PR 3） + +Ink TUI 的每个模型轮已经通过 `LlmClient.sendMessageStream` 发送，所有类型的轮次都由同一个函数 `submitQuery` 发出。它的 loop 是被 React 拆开的：请求了工具的响应把工具交给 scheduler 后就返回，由 scheduler 的完成回调把结果作为新的 `ToolResult` 轮次发出。迁移 TUI 就是要等待这个工具批，因此分以下几步： + +| 步骤 | 内容 | +| ---- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| 3a | 每个模型轮改用 loop 的 `runModelRound`；TUI 保留自己的工具步骤；Ink 对拍测试 | +| 3b | 一个轮次的工具批改为在共享工具批上等待完成，底层仍是 TUI 的 `InteractiveModToolScheduler`，界面通过回调更新；去重、收尾和记录由 loop 负责；TUI 自己的步骤（历史修复、工具使用摘要、额度切换模型时的停止和后台 agent 容量用尽时的停止、按交互 owner 拆分）改为宿主钩子 | +| 3c-1 | 工具边界取入的输入改走 loop 的下一轮；retry | +| 3c-2 | cron、后台通知、teammate 和 Goal 轮次 | +| 3d | 默认开启，随后删除现有 loop | + +PR 3a 把 TUI 的每个模型轮都放到 `runModelRound` 上。模型轮用相同的选项发送相同的请求，因此 TUI 的流处理器仍和现在一样处理每个事件。迁走的是 loop 本来就负责的那部分： + +| TUI 现在 | 开启设置后 | +| -------------------------------------------------------- | ----------------------------------------- | +| 流处理器收集 `ToolCallRequest` 事件 | 由 `runModelRound` 收集 | +| 流处理器在 `Retry` 和 `ModelFallback` 时丢弃已收集的调用 | 由 `runModelRound` 丢弃 | +| 检测到循环时不调度已收集的调用 | 模型轮的任何 stop reason 都不调度这些调用 | + +除检测到循环外，其他 stop reason 要么来自 client 在请求模型之前发出的事件（会话轮次上限、会话 token 上限、被拦截的 prompt），要么来自取消，而取消和现在一样结束这一轮，所以这些模型轮都没有要调度的调用。因此设置开启和关闭时，TUI 的行为相同。 + ## 迁移计划 | PR | 内容 | @@ -381,13 +403,17 @@ prompt 循环里的共享模型轮改为三个循环共用的方法。prompt 循 | 2c-3 | ACP：Goal 轮次 | | 2c-4 | ACP：cron 和后台通知轮次 | | 2d | ACP：默认开启，随后删除现有 loop | -| 3 | Ink TUI | +| 3a | Ink TUI：每个模型轮改用 loop 的模型轮；TUI 保留自己的工具步骤；Ink 对拍测试 | +| 3b | Ink TUI：工具批改为在共享工具批上等待完成 | +| 3c-1 | Ink TUI：工具边界的输入改走 loop 的下一轮；retry | +| 3c-2 | Ink TUI：cron、后台通知、teammate 和 Goal 轮次 | +| 3d | Ink TUI：默认开启，随后删除现有 loop | | 4 | OpenTUI | | 5 | AgentCore 改用基于 `Turn` 的模型轮（subagent、后台与 fork agent、workflow） | | 6 | speculation | | — | Hosted Harness：单独评估 | -PR 3–6 每一项之后，同样各自跟随默认开启和删除旧 loop 两步。 +PR 4–6 每一项之后，同样各自跟随默认开启和删除旧 loop 两步。 一个入口在设置开启时通过它的对拍测试，并列出了已知差异，才算完成迁移。之后每个入口各自修改默认值，不必等其他入口。修改默认值需要同时满足： @@ -448,6 +474,18 @@ PR 3–6 每一项之后，同样各自跟随默认开启和删除旧 loop 两 续发请求在现有 loop 上使用自己的 prompt id，在共享 loop 上沿用这一轮的 prompt id；对拍测试把两者视为相同。 +**Ink 场景**（`integration-tests/interactive/ink-agent-loop-parity.test.ts`）。TUI 在伪终端里对着假服务器运行。测试比较两次运行实际发送的请求体（包括 system prompt、提供的工具和每条消息的 `reasoning_content`）以及会话的聊天记录（包括 usage）。它还检查每次运行的请求数，并通过 debug 日志确认只有开启设置的那次运行在共享 loop 上发送模型轮。对于既不发送也不记录任何内容的场景（例如被拦截的 prompt），改为检查每次运行最后屏幕上的提示；除此之外不比较屏幕，因为屏幕按定时器重绘： + +- 纯文本，以及带 thought 的流式文本； +- 先有 thought 再调用工具，第二个请求会回传这段 reasoning； +- 一个工具，以及一次响应中的两次读取； +- 被输出上限截断的编辑； +- 断开后重试的流； +- 同一会话中的两个 prompt； +- 模型错误； +- 拦截的 UserPromptSubmit hook； +- 拦截一次的 Stop hook。 + 对拍测试在 integration 任务中运行。每晚的 E2E 运行（以及手动触发的运行）还会在一个不带沙箱的 Linux 分支上，在设置开启的情况下运行完整的集成测试；入口修改默认值时等待的就是这个定时任务。 ## 兼容性 diff --git a/docs/users/configuration/settings.md b/docs/users/configuration/settings.md index 3cc36b34871..098d143a037 100644 --- a/docs/users/configuration/settings.md +++ b/docs/users/configuration/settings.md @@ -690,7 +690,7 @@ Persistent [`qwen serve`](../qwen-serve) settings. Changes require restarting th | `experimental.sessionWriterLease` | boolean | Enable cross-process write fencing for persisted ACP and daemon sessions. The value is frozen when the ACP or daemon process starts. All concurrent ACP writers must enable the setting; mixed versions or configurations remain unsafe. Interactive and headless recorders are unaffected. Requires process restart. | `false` | | `experimental.cronRecurringMaxAgeDays` | number | Days a recurring cron/loop job lives before auto-expiring (it fires one final time, then is deleted). Set to `0` to disable expiry so jobs run until deleted — useful for long-running daemon deployments. Can be overridden via the `QWEN_CODE_CRON_MAX_AGE_DAYS` environment variable. Requires restart. | `7` | | `experimental.agentTeam` | boolean | Enable agent-team collaboration tools (`team_create`, `task_create`, `task_update`, `send_message`, etc.) for multi-agent coordination. Can also be enabled via `QWEN_CODE_ENABLE_AGENT_TEAM=1`. Requires restart. | `false` | -| `experimental.unifiedAgentLoop` | boolean | Run headless sessions (`-p` and stream-json) and ACP sessions (Zed, VS Code, Web Shell, channels) on the shared agent loop instead of their own loop. The interactive TUI and subagents keep their current loop for now. Can also be enabled via `QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1`. Requires restart. | `false` | +| `experimental.unifiedAgentLoop` | boolean | Run headless sessions (`-p` and stream-json) and ACP sessions (Zed, VS Code, Web Shell, channels) on the shared agent loop instead of their own loop. The Ink TUI sends its model requests through it, while its tool runs, OpenTUI and subagents keep their current loop for now. Can also be enabled via `QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1`. Requires restart. | `false` | | `experimental.artifact` | boolean | Enable artifact tools. Enabled by default. In interactive, non-SDK sessions, the model can publish a self-contained HTML page as an interactive Artifact and open it in the browser. Non-SDK daemon sessions can use metadata-only `record_artifact`. Set this to `false` or use `QWEN_CODE_DISABLE_ARTIFACT=1` to disable both. Requires restart. | `true` | | `experimental.emitToolUseSummaries` | boolean | Generate a short LLM-based label after each tool-call batch completes. See [Tool-Use Summaries](../features/tool-use-summaries). Requires a fast model to be configured (`fastModel`); silently skipped otherwise. Can be overridden per-session with `QWEN_CODE_EMIT_TOOL_USE_SUMMARIES=0` or `=1`. | `true` | diff --git a/integration-tests/interactive/ink-agent-loop-parity.test.ts b/integration-tests/interactive/ink-agent-loop-parity.test.ts new file mode 100644 index 00000000000..dddba317b9a --- /dev/null +++ b/integration-tests/interactive/ink-agent-loop-parity.test.ts @@ -0,0 +1,562 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * The Ink TUI runs each scripted scenario twice, with + * `QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP` unset and set, and the two runs must + * send the model the same requests and record the same session. The screen is + * not compared: it repaints on timers, and what it shows comes from the same + * events the session records. + */ + +import { + existsSync, + mkdirSync, + mkdtempSync, + readdirSync, + readFileSync, + rmSync, + writeFileSync, +} from 'node:fs'; +import { tmpdir } from 'node:os'; +import { basename, join } from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { E2E_MEMORY_SETTINGS_DEFAULTS } from '../e2e-memory-defaults.js'; +import { + fakeToolCall, + startFakeOpenAIServer, + type FakeOpenAIResponse, + type FakeOpenAIServer, +} from '../fake-openai-server.js'; +import { pickE2eRenderer } from '../renderer-matrix.js'; +import { IS_CONTAINER_SANDBOX } from '../test-helper.js'; +import { InteractiveSession } from './interactive-session.js'; + +type JsonObject = Record; + +interface Scenario { + name: string; + prompts: string[]; + /** Files to create in the workspace, by relative path. */ + files?: Record; + settings?: JsonObject; + /** The model requests each run must send, retries included. */ + requests: number; + /** + * The model rounds the TUI sends; a retry or a Stop-hook continuation runs + * inside its round. + */ + rounds: number; + /** + * Text both runs must show at the end, for a scenario that sends and + * records nothing. + */ + shows?: string; + /** Text both runs' session records must contain. */ + recorded?: string; + /** Text both runs' model requests must contain. */ + sent?: string; + /** + * The model's reply to a streaming request, from its message history; + * `index` counts this run's streaming requests, retries included. + */ + reply: ( + messages: JsonObject[], + dir: string, + index: number, + ) => FakeOpenAIResponse; +} + +interface Run { + requests: unknown[]; + records: unknown[]; + /** The screen after the last prompt. */ + screen: string; + /** How many model rounds the debug log says ran on the shared loop. */ + sharedRounds: number; +} + +/** + * The root config's `retry: 2` would rerun both TUI sessions after a stall, + * tripling its cost in the required gate. + */ +const PARITY_CASE_OPTIONS = { retry: 0, timeout: 240_000 }; + +/** What the TUI logs for each model round it sends on the shared loop. */ +const SHARED_ROUND_LOG = 'model round on the shared agent loop'; + +const sleep = (ms: number) => new Promise((done) => setTimeout(done, ms)); + +async function until( + check: () => boolean | Promise, + reason: string, + timeout = 30_000, +) { + const deadline = Date.now() + timeout; + while (!(await check())) { + if (Date.now() > deadline) throw new Error(reason); + await sleep(100); + } +} + +function files(directory: string): string[] { + return readdirSync(directory, { withFileTypes: true }).flatMap((entry) => { + const file = join(directory, entry.name); + return entry.isDirectory() ? files(file) : [file]; + }); +} + +const RANDOM_ID = + /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/g; + +// Fields that differ between two runs of the same scenario. +const VOLATILE_KEYS = new Set([ + 'uuid', + 'parentUuid', + 'sessionId', + 'timestamp', + 'cwd', + 'version', + 'gitBranch', + 'durationMs', + 'duration_ms', + 'apiTimeMs', + 'startedAt', + 'endedAt', + 'event.timestamp', +]); + +// Timings, such as the telemetry recorded with each response. +const TIMING_KEY = /(_ms|Ms)$/; + +const DATE_REMINDER = '\nThe current date is:'; + +const toolResults = (messages: JsonObject[]) => + messages.filter((message) => message['role'] === 'tool').length; + +const userTurns = (messages: JsonObject[]) => + messages.filter( + (message) => + message['role'] === 'user' && + !JSON.stringify(message['content']).includes(''), + ).length; + +const isDateReminder = (part: unknown) => + typeof (part as JsonObject)?.['text'] === 'string' && + ((part as JsonObject)['text'] as string).startsWith(DATE_REMINDER); + +/** `dir` is the workspace; `scratch`, which holds it, is the run's home. */ +function normalize(value: unknown, dir: string, scratch: string): unknown { + const replaced = JSON.parse( + JSON.stringify(value ?? null) + .split(dir) + .join('') + .split(scratch) + .join('') + .replace(RANDOM_ID, ''), + ); + const strip = (node: unknown): unknown => { + if (Array.isArray(node)) return node.map(strip); + if (node && typeof node === 'object') { + return Object.fromEntries( + Object.entries(node as JsonObject) + .filter(([key]) => !VOLATILE_KEYS.has(key) && !TIMING_KEY.test(key)) + .map(([key, child]) => [key, strip(child)]), + ); + } + return node; + }; + return strip(replaced); +} + +/** + * Each request body as the client sent it, without the date reminder: every + * message field, `reasoning_content` included, the tools offered and the + * sampling options. + */ +function modelRequests( + server: FakeOpenAIServer, + dir: string, + scratch: string, +): unknown[] { + return server.requests + .filter((request) => request.body['stream'] === true) + .map((request) => + normalize( + { + ...request.body, + messages: ((request.body['messages'] ?? []) as JsonObject[]) + .map((message) => ({ + ...message, + content: Array.isArray(message['content']) + ? message['content'].filter((part) => !isDateReminder(part)) + : message['content'], + })) + .filter( + (message) => + !Array.isArray(message.content) || message.content.length > 0, + ), + }, + dir, + scratch, + ), + ); +} + +/** The session's chat records: what was said, run and reported, in order. */ +function sessionRecords( + runtimeDir: string, + dir: string, + scratch: string, +): unknown[] { + return files(runtimeDir) + .filter((file) => file.endsWith('.jsonl') && file.includes('chats')) + .flatMap((file) => + readFileSync(file, 'utf8') + .split('\n') + .filter(Boolean) + .map((line) => JSON.parse(line) as JsonObject), + ) + .map((record) => + normalize( + { + type: record['type'], + subtype: record['subtype'], + message: record['message'], + toolCallResult: record['toolCallResult'], + systemPayload: record['systemPayload'], + usageMetadata: record['usageMetadata'], + model: record['model'], + contextWindowSize: record['contextWindowSize'], + }, + dir, + scratch, + ), + ); +} + +const hooks = (event: string, script: string) => ({ + [event]: [ + { + matcher: '*', + hooks: [{ type: 'command', command: `node ${script}` }], + }, + ], +}); + +const describeInk = + IS_CONTAINER_SANDBOX || pickE2eRenderer() !== 'ink' + ? describe.skip + : describe; + +describeInk('Ink TUI agent loop parity (experimental.unifiedAgentLoop)', () => { + let server: FakeOpenAIServer | undefined; + let session: InteractiveSession | undefined; + let scratch: string | undefined; + + afterEach(async () => { + await session?.close(); + session = undefined; + await server?.close(); + server = undefined; + if (scratch) rmSync(scratch, { recursive: true, force: true }); + scratch = undefined; + }); + + async function run(scenario: Scenario, shared: boolean): Promise { + await session?.close(); + await server?.close(); + if (scratch) rmSync(scratch, { recursive: true, force: true }); + scratch = mkdtempSync(join(tmpdir(), 'qwen-ink-agent-loop-parity-')); + const workspace = join(scratch, 'workspace'); + const home = join(scratch, 'home'); + const runtimeDir = join(scratch, 'runtime'); + for (const directory of [workspace, home, runtimeDir]) + mkdirSync(directory, { recursive: true }); + for (const [file, content] of Object.entries(scenario.files ?? {})) + writeFileSync(join(workspace, file), content); + const json = (file: string, value: unknown) => + writeFileSync(file, JSON.stringify(value, null, 2) + '\n'); + json(join(home, 'settings.json'), { + $version: 4, + ...scenario.settings, + security: { + auth: { selectedType: 'openai' }, + folderTrust: { enabled: false }, + }, + general: { + enableAutoUpdate: false, + showSessionRecap: false, + chatRecording: true, + }, + ui: { enableFollowupSuggestions: false }, + memory: E2E_MEMORY_SETTINGS_DEFAULTS, + experimental: { mods: false }, + }); + for (const file of ['system.json', 'defaults.json']) + json(join(scratch, file), { $version: 4 }); + + let streamingIndex = 0; + server = await startFakeOpenAIServer(({ body }) => + body['stream'] === true + ? scenario.reply( + (body['messages'] ?? []) as JsonObject[], + workspace, + streamingIndex++, + ) + : { content: '{"selected_memories":[]}' }, + ); + session = await InteractiveSession.start({ + cwd: workspace, + cols: 100, + rows: 50, + inheritEnv: false, + env: { + PATH: process.env['PATH'], + HOME: scratch, + QWEN_HOME: home, + QWEN_RUNTIME_DIR: runtimeDir, + QWEN_CODE_SYSTEM_SETTINGS_PATH: join(scratch, 'system.json'), + QWEN_CODE_SYSTEM_DEFAULTS_PATH: join(scratch, 'defaults.json'), + TERM: 'xterm-256color', + LANG: 'en_US.UTF-8', + QWEN_CODE_LANG: 'en', + QWEN_SANDBOX: 'false', + QWEN_CODE_DISABLE_CRON: '1', + QWEN_CODE_SKIP_UPDATE_CHECK_ONCE: 'true', + QWEN_CODE_MODELS_DEV_REFRESH: 'off', + QWEN_DEBUG_LOG_FILE: '1', + NO_PROXY: 'localhost,127.0.0.1', + no_proxy: 'localhost,127.0.0.1', + ...(shared ? { QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP: '1' } : {}), + }, + args: [ + '--auth-type', + 'openai', + '--openai-api-key', + 'fake-key', + '--openai-base-url', + server.baseUrl, + '--model', + 'fake-model', + '--approval-mode', + 'yolo', + ], + }); + await until( + async () => !(await session!.viewport()).includes('Initializing...'), + 'initialization completes', + ); + await session.idle(300, 5_000); + + for (const prompt of scenario.prompts) { + await session.send(prompt); + // A turn is over once the spinner stops and no request is in flight. + await sleep(500); + await session.idle(2_000, 90_000); + await until( + () => + server!.requests.every( + (request) => request.transport.closedAt !== undefined, + ), + 'model requests complete', + ); + await session.idle(1_000, 30_000); + } + + const screen = await session.screen(); + await session.send('/quit'); + await until( + () => session!.exitResult() !== undefined, + '/quit exits the TUI', + ); + const debugDir = join(runtimeDir, 'debug'); + const debugLog = existsSync(debugDir) + ? files(debugDir) + // `latest` points at the session's own log. + .filter((file) => basename(file) !== 'latest') + .map((file) => readFileSync(file, 'utf8')) + .join('') + : ''; + return { + requests: modelRequests(server, workspace, scratch), + records: sessionRecords(runtimeDir, workspace, scratch), + screen, + sharedRounds: debugLog.split(SHARED_ROUND_LOG).length - 1, + }; + } + + const read = (file: string, id: string) => + fakeToolCall('read_file', { file_path: file }, id); + + const scenarios: Scenario[] = [ + { + name: 'plain text', + requests: 1, + rounds: 1, + prompts: ['Say hello.'], + reply: () => ({ content: 'Hello there.' }), + }, + { + name: 'streamed text and a thought', + requests: 1, + rounds: 1, + prompts: ['Think, then answer.'], + reply: () => ({ + reasoningChunks: ['Weighing ', 'the options.'], + contentChunks: ['Hel', 'lo ', 'there.'], + usage: { prompt_tokens: 40, completion_tokens: 9, total_tokens: 49 }, + }), + recorded: '"totalTokenCount":49', + }, + { + // The second request replays the assistant turn that carried the + // reasoning, so its `reasoning_content` is compared. + name: 'a thought before a tool', + requests: 2, + rounds: 2, + prompts: ['Think, then read the readme.'], + files: { 'readme.txt': 'readme body\n' }, + sent: '"reasoning_content":"Deciding to read."', + reply: (messages, dir) => + toolResults(messages) === 0 + ? { + reasoningChunks: ['Deciding ', 'to read.'], + toolCalls: [read(join(dir, 'readme.txt'), 'call_think_read')], + } + : { content: 'It says readme body.' }, + }, + { + name: 'one tool', + requests: 2, + rounds: 2, + prompts: ['Read the readme.'], + files: { 'readme.txt': 'readme body\n' }, + reply: (messages, dir) => + toolResults(messages) === 0 + ? { toolCalls: [read(join(dir, 'readme.txt'), 'call_read')] } + : { content: 'It says readme body.' }, + }, + { + name: 'two reads in one response', + requests: 2, + rounds: 2, + prompts: ['Read both files.'], + files: { 'a.txt': 'alpha\n', 'b.txt': 'beta\n' }, + reply: (messages, dir) => + toolResults(messages) === 0 + ? { + toolCalls: [ + read(join(dir, 'a.txt'), 'call_a'), + read(join(dir, 'b.txt'), 'call_b'), + ], + } + : { content: 'Both read.' }, + }, + { + name: 'an edit cut off by the output limit', + // The client asks again for the cut-off call inside its round. + requests: 3, + rounds: 2, + prompts: ['Write the notes.'], + reply: (messages, dir) => + toolResults(messages) === 0 + ? { + toolCalls: [ + { + id: 'call_cut', + type: 'function', + function: { + name: 'write_file', + arguments: `{"file_path":${JSON.stringify(join(dir, 'notes.txt'))},"content":"line one\\nline thr`, + }, + }, + ], + finishReason: 'length', + } + : { content: 'Understood.' }, + }, + { + name: 'a stream that drops and is retried', + requests: 2, + rounds: 1, + prompts: ['Tell me a story.'], + reply: (_messages, _dir, index) => + index === 0 + ? { + contentChunks: ['Once ', 'upon'], + disconnectAfterContentChunks: 1, + } + : { contentChunks: ['Once ', 'upon a time.'] }, + }, + { + name: 'two prompts in one session', + requests: 2, + rounds: 2, + prompts: ['First question.', 'Second question.'], + reply: (messages) => ({ content: `Answer ${userTurns(messages)}.` }), + }, + { + name: 'a model error', + requests: 1, + rounds: 1, + prompts: ['Fail please.'], + reply: () => ({ errorContent: 'upstream exploded' }), + }, + { + name: 'a UserPromptSubmit hook that blocks', + requests: 0, + rounds: 1, + prompts: ['Do the blocked thing.'], + settings: { hooks: hooks('UserPromptSubmit', 'block-prompt.mjs') }, + files: { + 'block-prompt.mjs': + "process.stdout.write(JSON.stringify({ decision: 'block', reason: 'not today' }));\n", + }, + shows: 'not today', + reply: () => ({ content: 'Should not be asked.' }), + }, + { + name: 'a Stop hook that blocks once', + requests: 2, + rounds: 1, + prompts: ['Answer, then keep going once.'], + settings: { hooks: hooks('Stop', 'stop-once.mjs') }, + files: { + 'stop-once.mjs': [ + "import { existsSync, writeFileSync } from 'node:fs';", + "const marker = 'stop-hook-fired';", + 'if (existsSync(marker)) process.stdout.write("{}");', + 'else {', + " writeFileSync(marker, '');", + " process.stdout.write(JSON.stringify({ decision: 'block', reason: 'Add a closing line.' }));", + '}', + '', + ].join('\n'), + }, + reply: (messages) => ({ content: `Reply ${userTurns(messages)}.` }), + }, + ]; + + it.each(scenarios)('$name', PARITY_CASE_OPTIONS, async (scenario) => { + const current = await run(scenario, false); + const shared = await run(scenario, true); + // Only the run with the setting on sends its rounds on the shared loop. + expect(current.sharedRounds).toBe(0); + expect(shared.sharedRounds).toBe(scenario.rounds); + for (const run of [current, shared]) { + expect(run.requests).toHaveLength(scenario.requests); + if (scenario.shows) expect(run.screen).toContain(scenario.shows); + else expect(run.records.length).toBeGreaterThan(0); + if (scenario.recorded) + expect(JSON.stringify(run.records)).toContain(scenario.recorded); + if (scenario.sent) + expect(JSON.stringify(run.requests)).toContain(scenario.sent); + } + expect(shared.requests).toEqual(current.requests); + expect(shared.records).toEqual(current.records); + }); +}); diff --git a/package.json b/package.json index 5c4494b98fd..207ade74c8c 100644 --- a/package.json +++ b/package.json @@ -99,7 +99,7 @@ "test:integration:sandbox:docker": "cross-env QWEN_SANDBOX=docker npm run build:sandbox && QWEN_SANDBOX=docker vitest run --root ./integration-tests", "test:integration:sandbox:podman": "cross-env QWEN_SANDBOX=podman vitest run --root ./integration-tests", "test:integration:hosted:sandbox:none": "cross-env QWEN_SANDBOX=false vitest run --root ./integration-tests --config ./vitest.hosted.config.ts", - "test:integration:no-ak:sandbox:none": "cross-env QWEN_SANDBOX=false vitest run --root ./integration-tests --poolOptions.forks.maxForks 2 ./fake-openai-server.test.ts ./test-helper.test.ts ./chat-transcript-contract.test.ts ./skill-hooks-invocation-parity.test.ts ./skill-hooks-resume.test.ts ./qwen-live-m4-acp-call.test.ts ./qwen-live-m4-acp-permission.test.ts ./qwen-live-m4-acp-steering.test.ts ./qwen-live-m4-acp-multibackend.test.ts ./qwen-live-m1-call.test.ts ./qwen-live-m2-inject.test.ts ./qwen-live-m2-permission.test.ts ./qwen-live-m2-steering.test.ts ./cli/_prompt-latency-policy.test.ts ./cli/agent-loop-parity.test.ts ./cli/acp-agent-loop-parity.test.ts ./cli/hosted-harness-process.test.ts ./cli/daemon-invocation-context.test.ts ./cli/headless-workflow-skill.test.ts ./cli/workflow-ancestor-discovery.test.ts ./cli/daemon-workflow-ancestor-trust.test.ts ./cli/workflow-intent.test.ts ./cli/workflow-mode.test.ts ./cli/acp-workflow-mode.test.ts ./cli/daemon-workflow-intent.test.ts ./cli/daemon-workflow-mode.test.ts ./cli/list_directory.test.ts ./cli/tool-hook-context.test.ts ./cli/tool-hook-updated-input.test.ts ./cli/qwen-serve-routes.test.ts ./cli/qwen-serve-session-cd.test.ts ./cli/qwen-serve-streaming.test.ts ./cli/qwen-serve-standalone-concurrency.test.ts ./helpers/hosted-sse-observer.test.ts ./interactive/workflow-completion.test.ts ./interactive/workflow-intent.test.ts ./interactive/workflow-mode.test.ts ./interactive/mods-tool-middleware.test.ts ./interactive/statusline-parity.test.ts ./interactive/tui-editor-parity.test.ts ./interactive/tui-input-ownership.test.ts ./sdk-typescript/abort-and-lifecycle.test.ts ./sdk-typescript/permission-control.test.ts ./sdk-typescript/sdk-mcp-server.test.ts ./sdk-typescript/subagents.test.ts ./sdk-typescript/system-control.test.ts ./sdk-typescript/tool-control.test.ts ./sdk-typescript/workflow-intent.test.ts ./sdk-typescript/workflow-mode.test.ts", + "test:integration:no-ak:sandbox:none": "cross-env QWEN_SANDBOX=false vitest run --root ./integration-tests --poolOptions.forks.maxForks 2 ./fake-openai-server.test.ts ./test-helper.test.ts ./chat-transcript-contract.test.ts ./skill-hooks-invocation-parity.test.ts ./skill-hooks-resume.test.ts ./qwen-live-m4-acp-call.test.ts ./qwen-live-m4-acp-permission.test.ts ./qwen-live-m4-acp-steering.test.ts ./qwen-live-m4-acp-multibackend.test.ts ./qwen-live-m1-call.test.ts ./qwen-live-m2-inject.test.ts ./qwen-live-m2-permission.test.ts ./qwen-live-m2-steering.test.ts ./cli/_prompt-latency-policy.test.ts ./cli/agent-loop-parity.test.ts ./cli/acp-agent-loop-parity.test.ts ./cli/hosted-harness-process.test.ts ./cli/daemon-invocation-context.test.ts ./cli/headless-workflow-skill.test.ts ./cli/workflow-ancestor-discovery.test.ts ./cli/daemon-workflow-ancestor-trust.test.ts ./cli/workflow-intent.test.ts ./cli/workflow-mode.test.ts ./cli/acp-workflow-mode.test.ts ./cli/daemon-workflow-intent.test.ts ./cli/daemon-workflow-mode.test.ts ./cli/list_directory.test.ts ./cli/tool-hook-context.test.ts ./cli/tool-hook-updated-input.test.ts ./cli/qwen-serve-routes.test.ts ./cli/qwen-serve-session-cd.test.ts ./cli/qwen-serve-streaming.test.ts ./cli/qwen-serve-standalone-concurrency.test.ts ./helpers/hosted-sse-observer.test.ts ./interactive/workflow-completion.test.ts ./interactive/workflow-intent.test.ts ./interactive/workflow-mode.test.ts ./interactive/mods-tool-middleware.test.ts ./interactive/statusline-parity.test.ts ./interactive/tui-editor-parity.test.ts ./interactive/tui-input-ownership.test.ts ./interactive/ink-agent-loop-parity.test.ts ./sdk-typescript/abort-and-lifecycle.test.ts ./sdk-typescript/permission-control.test.ts ./sdk-typescript/sdk-mcp-server.test.ts ./sdk-typescript/subagents.test.ts ./sdk-typescript/system-control.test.ts ./sdk-typescript/tool-control.test.ts ./sdk-typescript/workflow-intent.test.ts ./sdk-typescript/workflow-mode.test.ts", "test:integration:sdk:sandbox:none": "cross-env QWEN_SANDBOX=false vitest run --root ./integration-tests --poolOptions.forks.maxForks 2 sdk-typescript", "test:integration:sdk:sandbox:docker": "cross-env QWEN_SANDBOX=docker npm run build:sandbox && QWEN_SANDBOX=docker vitest run --root ./integration-tests --poolOptions.forks.maxForks 2 sdk-typescript", "test:sdk:python": "python3 -m pytest -c packages/sdk-python/pyproject.toml packages/sdk-python/tests -q", diff --git a/packages/cli/src/config/settingsSchema.ts b/packages/cli/src/config/settingsSchema.ts index d9755eb8e97..ed903583123 100644 --- a/packages/cli/src/config/settingsSchema.ts +++ b/packages/cli/src/config/settingsSchema.ts @@ -4558,7 +4558,7 @@ const SETTINGS_SCHEMA = { requiresRestart: true, default: false, description: - 'Run headless sessions (`-p` and stream-json) and ACP sessions (Zed, VS Code, Web Shell, channels) on the shared agent loop (experimental). The interactive TUI and subagents keep their current loop for now. Can also be enabled via QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1 environment variable.', + 'Run headless sessions (`-p` and stream-json) and ACP sessions (Zed, VS Code, Web Shell, channels) on the shared agent loop (experimental). The Ink TUI sends its model requests through it, while its tool runs, OpenTUI and subagents keep their current loop for now. Can also be enabled via QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1 environment variable.', showInDialog: true, }, agentCollaboration: { diff --git a/packages/cli/src/ui/hooks/use-llm-stream.test.tsx b/packages/cli/src/ui/hooks/use-llm-stream.test.tsx index 2b26e265aa0..b92f9ff0673 100644 --- a/packages/cli/src/ui/hooks/use-llm-stream.test.tsx +++ b/packages/cli/src/ui/hooks/use-llm-stream.test.tsx @@ -202,6 +202,17 @@ vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => { }); const mockUseReactToolScheduler = useReactToolScheduler as Mock; +const mockRunModelRound = vi.hoisted(() => vi.fn()); +vi.mock( + '@qwen-code/qwen-code-core/core/agent-loop.js', + async (importOriginal) => { + const actual = + (await importOriginal()) as typeof import('@qwen-code/qwen-code-core/core/agent-loop.js'); + mockRunModelRound.mockImplementation(actual.runModelRound); + return { ...actual, runModelRound: mockRunModelRound }; + }, +); + vi.mock('./useReactToolScheduler.js', async (importOriginal) => { const actualSchedulerModule = (await importOriginal()) as any; return { @@ -24514,4 +24525,145 @@ describe('useLlmStream', () => { // turnCount = history.length - 2 = 3. expect(event.turnCount).toBe(3); }); + + describe('shared agent loop round (experimental.unifiedAgentLoop)', () => { + const toolRequest = (callId: string) => ({ + type: ServerLlmEventType.ToolCallRequest, + value: { + callId, + name: 'read_file', + args: { file_path: `/test/dir/${callId}.txt` }, + isClientInitiated: false, + prompt_id: 'prompt-shared', + }, + }); + + beforeEach(() => { + ( + mockConfig as unknown as { isUnifiedAgentLoopEnabled: () => boolean } + ).isUnifiedAgentLoopEnabled = () => true; + mockRunModelRound.mockClear(); + }); + + it('sends the round through runModelRound with the same request', async () => { + mockSendMessageStream.mockReturnValue( + (async function* () { + yield { type: ServerLlmEventType.Content, value: 'Hello.' }; + })(), + ); + const { result } = renderTestHook(); + + await act(async () => { + await result.current.submitQuery( + 'Say hello.', + SendMessageType.UserQuery, + 'prompt-shared', + ); + }); + + expect(mockRunModelRound).toHaveBeenCalledTimes(1); + const [, round, signal] = mockRunModelRound.mock.calls[0]; + expect(mockSendMessageStream).toHaveBeenCalledTimes(1); + expect(mockSendMessageStream).toHaveBeenCalledWith( + round.parts, + signal, + 'prompt-shared', + round.options, + ); + expect(round).toMatchObject({ + promptId: 'prompt-shared', + options: { type: SendMessageType.UserQuery }, + }); + expect(mockScheduleToolCalls).not.toHaveBeenCalled(); + }); + + it('schedules the calls the round collected, without those a retry replaced', async () => { + mockSendMessageStream.mockReturnValue( + (async function* () { + yield toolRequest('dropped'); + yield { type: ServerLlmEventType.Retry, isContinuation: false }; + yield toolRequest('kept_a'); + yield toolRequest('kept_b'); + })(), + ); + const { result } = renderTestHook(); + + await act(async () => { + await result.current.submitQuery('Read the files.'); + }); + + expect(mockScheduleToolCalls).toHaveBeenCalledTimes(1); + expect( + mockScheduleToolCalls.mock.calls[0][0].map( + (request: { callId: string }) => request.callId, + ), + ).toEqual(['kept_a', 'kept_b']); + }); + + it('schedules the calls the round collected, without those a model fallback replaced', async () => { + mockSendMessageStream.mockReturnValue( + (async function* () { + yield toolRequest('dropped'); + yield { + type: ServerLlmEventType.ModelFallback, + fromModel: 'primary-model', + toModel: 'fallback-model', + fallbackIndex: 1, + }; + yield toolRequest('kept'); + })(), + ); + const { result } = renderTestHook(); + + await act(async () => { + await result.current.submitQuery('Read the file.'); + }); + + expect(mockScheduleToolCalls).toHaveBeenCalledTimes(1); + expect( + mockScheduleToolCalls.mock.calls[0][0].map( + (request: { callId: string }) => request.callId, + ), + ).toEqual(['kept']); + }); + + it('schedules nothing when the round ends on a stop reason', async () => { + mockSendMessageStream.mockReturnValue( + (async function* () { + yield toolRequest('before_stop'); + yield { + type: ServerLlmEventType.MaxSessionTurns, + }; + })(), + ); + const { result } = renderTestHook(); + + await act(async () => { + await result.current.submitQuery('Read the file.'); + }); + + expect(mockRunModelRound).toHaveBeenCalledTimes(1); + expect(mockScheduleToolCalls).not.toHaveBeenCalled(); + }); + + it('ends the round on a cancel without scheduling its calls', async () => { + const after = vi.fn(); + mockSendMessageStream.mockReturnValue( + (async function* () { + yield toolRequest('cancelled'); + yield { type: ServerLlmEventType.UserCancelled }; + after(); + yield toolRequest('after_cancel'); + })(), + ); + const { result } = renderTestHook(); + + await act(async () => { + await result.current.submitQuery('Read the file.'); + }); + + expect(after).not.toHaveBeenCalled(); + expect(mockScheduleToolCalls).not.toHaveBeenCalled(); + }); + }); }); diff --git a/packages/cli/src/ui/hooks/use-llm-stream.ts b/packages/cli/src/ui/hooks/use-llm-stream.ts index b833fabd283..eff1e1d6c96 100644 --- a/packages/cli/src/ui/hooks/use-llm-stream.ts +++ b/packages/cli/src/ui/hooks/use-llm-stream.ts @@ -22,6 +22,7 @@ import { type Config, type EditorType, type LlmClient, + type SendMessageOptions, type Logger, type RetryInfo, type ServerLlmChatCompressedEvent, @@ -91,6 +92,11 @@ import { getApiHistoryPromptId, } from '@qwen-code/qwen-code-core'; import type { HumanPromptInput } from '@qwen-code/qwen-code-core/core/workflow-intent.js'; +import { + runModelRound, + type AgentLoopRound, + type AgentLoopStopReason, +} from '@qwen-code/qwen-code-core/core/agent-loop.js'; import { type Part, type PartListUnion, FinishReason } from '@google/genai'; import { getInteractiveModSession, @@ -439,6 +445,22 @@ interface StreamProcessingResult { userPromptBlocked: boolean; } +/** + * A model round on the shared agent loop (`experimental.unifiedAgentLoop`): + * `runModelRound` sends it, collects its tool calls and turns the client's + * terminal events into a stop reason. + */ +interface SharedModelRound { + client: LlmClient; + round: AgentLoopRound; + signal: AbortSignal; +} + +/** Ends a shared model round from inside its event callback. */ +class StreamProcessingEnded { + constructor(readonly result: StreamProcessingResult) {} +} + const EDIT_TOOL_NAMES = new Set([ ToolNames.EDIT, 'replace', // legacy alias, may still arrive from older providers @@ -2959,7 +2981,7 @@ export const useLlmStream = ( const processLlmStreamEvents = useCallback( async ( - stream: AsyncIterable, + source: AsyncIterable | SharedModelRound, userMessageTimestamp: number, signal: AbortSignal, submitType: SendMessageType, @@ -2985,7 +3007,12 @@ export const useLlmStream = ( : count, 0, ); - const toolCallRequests: ToolCallRequestInfo[] = []; + const sharedRound = 'round' in source; + // On the shared round, `runModelRound` collects the calls and drops + // those of an attempt a retry or fallback replaced; the processor's own + // resets of `toolCallRequests` below matter only on the current path. + let toolCallRequests: ToolCallRequestInfo[] = []; + let roundStop: AgentLoopStopReason | undefined; let streamInteractionOwner = trackInteractionOwner ? activeInteractionOwnerRef.current : undefined; @@ -3134,252 +3161,218 @@ export const useLlmStream = ( }, STREAM_UPDATE_THROTTLE_MS); }; - flushBufferedStreamEventsRef.current.add(flushBufferedStreamEvents); - dualOutput?.startAssistantMessage(); - try { - for await (const event of stream) { - if (!streamInteractionOwner && promptId) { - streamInteractionOwner = getActiveInteractionSpan(promptId); - if (trackInteractionOwner && streamInteractionOwner) { - activeInteractionOwnerRef.current ??= streamInteractionOwner; - } + const handleStreamEvent = ( + event: LlmEvent, + ): StreamProcessingResult | undefined => { + if (!streamInteractionOwner && promptId) { + streamInteractionOwner = getActiveInteractionSpan(promptId); + if (trackInteractionOwner && streamInteractionOwner) { + activeInteractionOwnerRef.current ??= streamInteractionOwner; } - dualOutput?.processEvent(event); - const modTurn = modTurnRef.current; - if (trackInteractionOwner && modTurn && modTurn.promptId === promptId) - modTurn.turn.saw(event); - switch (event.type) { - case ServerLlmEventType.Thought: - // Subject-only chunks are discrete status updates for the - // loading indicator and render immediately. Anything carrying - // streamed text (with or without a subject) goes through the - // throttled buffer so it batches with adjacent reasoning - // chunks; the flush merger preserves the subject. - if (event.value.subject && !event.value.description) { - flushBufferedStreamEvents(); - setThought(event.value); - } else { - bufferedEvents.push({ kind: 'thought', value: event.value }); - scheduleBufferedStreamFlush(); - } - break; - case ServerLlmEventType.Content: { - // Thinking is done once the answer starts streaming; reset the - // title status. On the thinking→answer transition, flush any - // buffered reasoning so the full thought is captured, then commit - // it to history (collapsed) above the answer. After that the - // condition is false, so normal content batching resumes. - if ( - pendingThoughtItemRef.current || - bufferedEvents.some((e) => e.kind === 'thought') - ) { - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp, true); - thoughtBuffer = ''; - } - setThought((prev) => (prev ? null : prev)); - const displayParts = event.parts ?? [{ text: event.value }]; - for (const part of displayParts) { - if ('text' in part) { - if (part.text.length > 0) { - bufferedEvents.push({ kind: 'content', value: part.text }); - } - } else { - const image = getInlineImageData({ - inlineData: part.inlineData, - }); - if (image) { - bufferedEvents.push({ kind: 'image', value: image }); - } - } - } + } + dualOutput?.processEvent(event); + const modTurn = modTurnRef.current; + if (trackInteractionOwner && modTurn && modTurn.promptId === promptId) + modTurn.turn.saw(event); + switch (event.type) { + case ServerLlmEventType.Thought: + // Subject-only chunks are discrete status updates for the + // loading indicator and render immediately. Anything carrying + // streamed text (with or without a subject) goes through the + // throttled buffer so it batches with adjacent reasoning + // chunks; the flush merger preserves the subject. + if (event.value.subject && !event.value.description) { + flushBufferedStreamEvents(); + setThought(event.value); + } else { + bufferedEvents.push({ kind: 'thought', value: event.value }); scheduleBufferedStreamFlush(); - break; } - case ServerLlmEventType.ToolCallRequest: - // Thinking is done once a tool call is issued; flush buffered - // reasoning then commit it to history (collapsed) above the tool - // output. + break; + case ServerLlmEventType.Content: { + // Thinking is done once the answer starts streaming; reset the + // title status. On the thinking→answer transition, flush any + // buffered reasoning so the full thought is captured, then commit + // it to history (collapsed) above the answer. After that the + // condition is false, so normal content batching resumes. + if ( + pendingThoughtItemRef.current || + bufferedEvents.some((e) => e.kind === 'thought') + ) { flushBufferedStreamEvents(); commitPendingThought(userMessageTimestamp, true); thoughtBuffer = ''; - setThought((prev) => (prev ? null : prev)); - if (event.value.goalContext && turnAdmission) { - bindGoalTurn( - event.value.goalContext, - turnAdmission.turnKey, - turnAdmission.origin, - turnAdmission.controller, - ); - } - toolCallRequests.push(event.value); - // Count tool call args JSON toward token estimation. - try { - const argsJson = JSON.stringify(event.value.args); - streamingResponseLengthRef.current += argsJson.length; - } catch { - // Best-effort — don't block on serialization errors - } - break; - case ServerLlmEventType.UserCancelled: - flushBufferedStreamEvents(); - toolCallRequests.length = 0; - handleUserCancelledEvent(userMessageTimestamp); - return { - status: StreamProcessingStatus.UserCancelled, - scheduledToolContinuation: false, - userPromptBlocked, - }; - case ServerLlmEventType.Error: - flushBufferedStreamEvents(); - handleErrorEvent( - event.value, - userMessageTimestamp, - submitType, - trackInteractionOwner && - toolContinuationOwner?.foregroundAbortController !== - undefined && - abortControllerRef.current === - toolContinuationOwner.foregroundAbortController, - ); - break; - case ServerLlmEventType.ChatCompressed: - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - handleChatCompressionEvent(event.value, userMessageTimestamp); - llmMessageBuffer = ''; - assistantOutputStarted = false; - break; - case ServerLlmEventType.ToolCallConfirmation: - case ServerLlmEventType.ToolCallResponse: - flushBufferedStreamEvents(); - break; - case ServerLlmEventType.ToolCallPreparation: - break; - case ServerLlmEventType.MaxSessionTurns: - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - if (pendingHistoryItemRef.current) { - commitItemInOrder( - pendingHistoryItemRef.current, - userMessageTimestamp, - ); - setPendingHistoryItem(null); - } - handleMaxSessionTurnsEvent(); - llmMessageBuffer = ''; - assistantOutputStarted = false; - break; - case ServerLlmEventType.SessionTokenLimitExceeded: - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - if (pendingHistoryItemRef.current) { - commitItemInOrder( - pendingHistoryItemRef.current, - userMessageTimestamp, - ); - setPendingHistoryItem(null); - } - handleSessionTokenLimitExceededEvent(event.value); - llmMessageBuffer = ''; - assistantOutputStarted = false; - break; - case ServerLlmEventType.Finished: - flushBufferedStreamEvents(); - // A thinking-only turn (no content/tool) still commits its - // reasoning so it persists collapsed in history. - commitPendingThought(userMessageTimestamp); - // Seal off this turn's UI state before the parent re-enters - // sendMessageStream for a continuation (Stop-hook block at - // client.ts:1378 or next-speaker auto-continue at 1444). Both - // paths yield* a fresh Turn through this same stream processor, - // so without this seal the next turn's first content/thought - // chunk appends to this turn's pending item — visible in the UI - // as "t" → "te" → "tes" cumulative rendering even though each - // turn is persisted as a clean, separate assistant message. - if (pendingHistoryItemRef.current) { - commitItemInOrder( - pendingHistoryItemRef.current, - userMessageTimestamp, - ); - setPendingHistoryItem(null); - } - llmMessageBuffer = ''; - thoughtBuffer = ''; - assistantOutputStarted = false; - assistantInlineImageCount = 0; - setThought(null); - handleFinishedEvent( - event as ServerLlmFinishedEvent, - userMessageTimestamp, - ); - break; - case ServerLlmEventType.Citation: - flushBufferedStreamEvents(); - if (showCitations(settings)) { - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - llmMessageBuffer = ''; - assistantOutputStarted = false; - } - handleCitationEvent(event.value, userMessageTimestamp); - break; - case ServerLlmEventType.LoopDetected: - flushBufferedStreamEvents(); - // handle later because we want to move pending history to history - // before we add loop detected message to history - loopDetectedRef.current = true; - loopStopRef.current = event.value; - break; - case ServerLlmEventType.Retry: - // On fresh restart (escalation / rate-limit / invalid stream), - // clear pending content and buffers to discard the failed attempt. - // On continuation (recovery), keep the pending gemini item AND - // buffers so the model's continuation text appends to them — - // otherwise handleContentEvent would see a null pending item, - // create a fresh one, and reset the buffer to just the new chunk, - // losing the partial text we meant to preserve. - if (!event.isContinuation) { - discardBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp, true); - setPendingAssistantItems([]); - if (pendingHistoryItemRef.current) { - setPendingHistoryItem(null); + } + setThought((prev) => (prev ? null : prev)); + const displayParts = event.parts ?? [{ text: event.value }]; + for (const part of displayParts) { + if ('text' in part) { + if (part.text.length > 0) { + bufferedEvents.push({ kind: 'content', value: part.text }); } - thoughtBuffer = ''; - setThought(null); - llmMessageBuffer = ''; - assistantOutputStarted = false; - assistantInlineImageCount = 0; } else { - flushBufferedStreamEvents(); + const image = getInlineImageData({ + inlineData: part.inlineData, + }); + if (image) { + bufferedEvents.push({ kind: 'image', value: image }); + } } - // Always discard tool call requests from the truncated/failed - // attempt to prevent duplicate execution after escalation or - // recovery. The recovery path now skips turns that already - // contain a functionCall (see llm-chat.ts), so this only - // clears stale requests from pre-RETRY accumulation. - toolCallRequests.length = 0; - // Show retry info if available (rate-limit / throttling errors) - if (event.retryInfo) { - startRetryCountdown(event.retryInfo); - } else { - // The retry attempt is starting now, so any prior retry UI is stale. - clearRetryCountdown(); - } - break; - case ServerLlmEventType.ModelFallback: { - // The primary model (or a prior fallback) exhausted its retry - // budget on a capacity/availability error and the system is - // switching to the next fallback model. Discard partial content - // from the failed attempt and show a notification. + } + scheduleBufferedStreamFlush(); + break; + } + case ServerLlmEventType.ToolCallRequest: + // Thinking is done once a tool call is issued; flush buffered + // reasoning then commit it to history (collapsed) above the tool + // output. + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp, true); + thoughtBuffer = ''; + setThought((prev) => (prev ? null : prev)); + if (event.value.goalContext && turnAdmission) { + bindGoalTurn( + event.value.goalContext, + turnAdmission.turnKey, + turnAdmission.origin, + turnAdmission.controller, + ); + } + if (!sharedRound) toolCallRequests.push(event.value); + // Count tool call args JSON toward token estimation. + try { + const argsJson = JSON.stringify(event.value.args); + streamingResponseLengthRef.current += argsJson.length; + } catch { + // Best-effort — don't block on serialization errors + } + break; + case ServerLlmEventType.UserCancelled: + flushBufferedStreamEvents(); + toolCallRequests.length = 0; + handleUserCancelledEvent(userMessageTimestamp); + return { + status: StreamProcessingStatus.UserCancelled, + scheduledToolContinuation: false, + userPromptBlocked, + }; + case ServerLlmEventType.Error: + flushBufferedStreamEvents(); + handleErrorEvent( + event.value, + userMessageTimestamp, + submitType, + trackInteractionOwner && + toolContinuationOwner?.foregroundAbortController !== + undefined && + abortControllerRef.current === + toolContinuationOwner.foregroundAbortController, + ); + break; + case ServerLlmEventType.ChatCompressed: + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + handleChatCompressionEvent(event.value, userMessageTimestamp); + llmMessageBuffer = ''; + assistantOutputStarted = false; + break; + case ServerLlmEventType.ToolCallConfirmation: + case ServerLlmEventType.ToolCallResponse: + flushBufferedStreamEvents(); + break; + case ServerLlmEventType.ToolCallPreparation: + break; + case ServerLlmEventType.MaxSessionTurns: + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + if (pendingHistoryItemRef.current) { + commitItemInOrder( + pendingHistoryItemRef.current, + userMessageTimestamp, + ); + setPendingHistoryItem(null); + } + handleMaxSessionTurnsEvent(); + llmMessageBuffer = ''; + assistantOutputStarted = false; + break; + case ServerLlmEventType.SessionTokenLimitExceeded: + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + if (pendingHistoryItemRef.current) { + commitItemInOrder( + pendingHistoryItemRef.current, + userMessageTimestamp, + ); + setPendingHistoryItem(null); + } + handleSessionTokenLimitExceededEvent(event.value); + llmMessageBuffer = ''; + assistantOutputStarted = false; + break; + case ServerLlmEventType.Finished: + flushBufferedStreamEvents(); + // A thinking-only turn (no content/tool) still commits its + // reasoning so it persists collapsed in history. + commitPendingThought(userMessageTimestamp); + // Seal off this turn's UI state before the parent re-enters + // sendMessageStream for a continuation (Stop-hook block at + // client.ts:1378 or next-speaker auto-continue at 1444). Both + // paths yield* a fresh Turn through this same stream processor, + // so without this seal the next turn's first content/thought + // chunk appends to this turn's pending item — visible in the UI + // as "t" → "te" → "tes" cumulative rendering even though each + // turn is persisted as a clean, separate assistant message. + if (pendingHistoryItemRef.current) { + commitItemInOrder( + pendingHistoryItemRef.current, + userMessageTimestamp, + ); + setPendingHistoryItem(null); + } + llmMessageBuffer = ''; + thoughtBuffer = ''; + assistantOutputStarted = false; + assistantInlineImageCount = 0; + setThought(null); + handleFinishedEvent( + event as ServerLlmFinishedEvent, + userMessageTimestamp, + ); + break; + case ServerLlmEventType.Citation: + flushBufferedStreamEvents(); + if (showCitations(settings)) { + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + llmMessageBuffer = ''; + assistantOutputStarted = false; + } + handleCitationEvent(event.value, userMessageTimestamp); + break; + case ServerLlmEventType.LoopDetected: + flushBufferedStreamEvents(); + // handle later because we want to move pending history to history + // before we add loop detected message to history + loopDetectedRef.current = true; + loopStopRef.current = event.value; + break; + case ServerLlmEventType.Retry: + // On fresh restart (escalation / rate-limit / invalid stream), + // clear pending content and buffers to discard the failed attempt. + // On continuation (recovery), keep the pending gemini item AND + // buffers so the model's continuation text appends to them — + // otherwise handleContentEvent would see a null pending item, + // create a fresh one, and reset the buffer to just the new chunk, + // losing the partial text we meant to preserve. + if (!event.isContinuation) { discardBufferedStreamEvents(); commitPendingThought(userMessageTimestamp, true); setPendingAssistantItems([]); @@ -3391,117 +3384,181 @@ export const useLlmStream = ( llmMessageBuffer = ''; assistantOutputStarted = false; assistantInlineImageCount = 0; - toolCallRequests.length = 0; + } else { + flushBufferedStreamEvents(); + } + // Always discard tool call requests from the truncated/failed + // attempt to prevent duplicate execution after escalation or + // recovery. The recovery path now skips turns that already + // contain a functionCall (see llm-chat.ts), so this only + // clears stale requests from pre-RETRY accumulation. + toolCallRequests.length = 0; + // Show retry info if available (rate-limit / throttling errors) + if (event.retryInfo) { + startRetryCountdown(event.retryInfo); + } else { + // The retry attempt is starting now, so any prior retry UI is stale. clearRetryCountdown(); - const fromModel = - sanitizeDisplayText(event.fromModel) ?? '(unknown)'; - const toModel = sanitizeDisplayText(event.toModel) ?? '(unknown)'; + } + break; + case ServerLlmEventType.ModelFallback: { + // The primary model (or a prior fallback) exhausted its retry + // budget on a capacity/availability error and the system is + // switching to the next fallback model. Discard partial content + // from the failed attempt and show a notification. + discardBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp, true); + setPendingAssistantItems([]); + if (pendingHistoryItemRef.current) { + setPendingHistoryItem(null); + } + thoughtBuffer = ''; + setThought(null); + llmMessageBuffer = ''; + assistantOutputStarted = false; + assistantInlineImageCount = 0; + toolCallRequests.length = 0; + clearRetryCountdown(); + const fromModel = + sanitizeDisplayText(event.fromModel) ?? '(unknown)'; + const toModel = sanitizeDisplayText(event.toModel) ?? '(unknown)'; + addItem( + { + type: 'notification', + text: `Model ${fromModel} unavailable, falling back to ${toModel}`, + }, + userMessageTimestamp, + ); + break; + } + case ServerLlmEventType.HookSystemMessage: + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + // Display system message from Stop hooks with "Stop says:" prefix + // First commit any pending AI response to ensure correct ordering + if (pendingHistoryItemRef.current) { + commitItemInOrder( + pendingHistoryItemRef.current, + userMessageTimestamp, + ); + setPendingHistoryItem(null); + } + addItem( + { + type: 'stop_hook_system_message', + message: event.value, + } as HistoryItemWithoutId, + userMessageTimestamp, + ); + llmMessageBuffer = ''; + assistantOutputStarted = false; + break; + case ServerLlmEventType.GoalSettlementFailed: + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + if (pendingHistoryItemRef.current) { + commitItemInOrder( + pendingHistoryItemRef.current, + userMessageTimestamp, + ); + setPendingHistoryItem(null); + } + addItem( + { type: 'warning', text: event.value }, + userMessageTimestamp, + ); + llmMessageBuffer = ''; + assistantOutputStarted = false; + break; + case ServerLlmEventType.UserPromptSubmitBlocked: + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + userPromptBlocked = true; + handleUserPromptSubmitBlockedEvent( + event.value, + userMessageTimestamp, + ); + llmMessageBuffer = ''; + assistantOutputStarted = false; + break; + case ServerLlmEventType.StopHookLoop: + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + handleStopHookLoopEvent(event.value, userMessageTimestamp); + llmMessageBuffer = ''; + assistantOutputStarted = false; + break; + case ServerLlmEventType.GoalState: + if (event.cause && shouldDisplayGoalStateCause(event.cause)) { + flushBufferedStreamEvents(); + commitPendingThought(userMessageTimestamp); + thoughtBuffer = ''; + setThought(null); + if (pendingHistoryItemRef.current) { + commitItemInOrder( + pendingHistoryItemRef.current, + userMessageTimestamp, + ); + setPendingHistoryItem(null); + } addItem( { - type: 'notification', - text: `Model ${fromModel} unavailable, falling back to ${toModel}`, + type: 'goal_state', + snapshot: event.value, + cause: event.cause, }, userMessageTimestamp, ); - break; - } - case ServerLlmEventType.HookSystemMessage: - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - // Display system message from Stop hooks with "Stop says:" prefix - // First commit any pending AI response to ensure correct ordering - if (pendingHistoryItemRef.current) { - commitItemInOrder( - pendingHistoryItemRef.current, - userMessageTimestamp, - ); - setPendingHistoryItem(null); - } - addItem( - { - type: 'stop_hook_system_message', - message: event.value, - } as HistoryItemWithoutId, - userMessageTimestamp, - ); llmMessageBuffer = ''; assistantOutputStarted = false; - break; - case ServerLlmEventType.GoalSettlementFailed: - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - if (pendingHistoryItemRef.current) { - commitItemInOrder( - pendingHistoryItemRef.current, - userMessageTimestamp, - ); - setPendingHistoryItem(null); - } - addItem( - { type: 'warning', text: event.value }, - userMessageTimestamp, - ); - llmMessageBuffer = ''; - assistantOutputStarted = false; - break; - case ServerLlmEventType.UserPromptSubmitBlocked: - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - userPromptBlocked = true; - handleUserPromptSubmitBlockedEvent( - event.value, - userMessageTimestamp, - ); - llmMessageBuffer = ''; - assistantOutputStarted = false; - break; - case ServerLlmEventType.StopHookLoop: - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - handleStopHookLoopEvent(event.value, userMessageTimestamp); - llmMessageBuffer = ''; - assistantOutputStarted = false; - break; - case ServerLlmEventType.GoalState: - if (event.cause && shouldDisplayGoalStateCause(event.cause)) { - flushBufferedStreamEvents(); - commitPendingThought(userMessageTimestamp); - thoughtBuffer = ''; - setThought(null); - if (pendingHistoryItemRef.current) { - commitItemInOrder( - pendingHistoryItemRef.current, - userMessageTimestamp, - ); - setPendingHistoryItem(null); - } - addItem( - { - type: 'goal_state', - snapshot: event.value, - cause: event.cause, - }, - userMessageTimestamp, - ); - llmMessageBuffer = ''; - assistantOutputStarted = false; - } - break; - default: { - // enforces exhaustive switch-case - const unreachable: never = event; - return unreachable; } + break; + default: { + // enforces exhaustive switch-case + const unreachable: never = event; + return unreachable; } } + return undefined; + }; + + flushBufferedStreamEventsRef.current.add(flushBufferedStreamEvents); + dualOutput?.startAssistantMessage(); + try { + if (sharedRound) { + debugLogger.debug( + `[processLlmStreamEvents] model round on the shared agent loop (prompt ${source.round.promptId})`, + ); + const { calls, stop } = await runModelRound( + source.client, + source.round, + source.signal, + { + onModelEvent: (event) => { + const ended = handleStreamEvent(event); + if (ended) throw new StreamProcessingEnded(ended); + }, + }, + ); + toolCallRequests = calls; + roundStop = stop; + } else { + for await (const event of source) { + const ended = handleStreamEvent(event); + if (ended) return ended; + } + } + } catch (error) { + if (error instanceof StreamProcessingEnded) return error.result; + throw error; } finally { flushBufferedStreamEvents(); // A thrown stream must leave interleaved output discardable on retry. @@ -3521,7 +3578,7 @@ export const useLlmStream = ( if ( toolCallRequests.length > 0 && !signal.aborted && - !loopDetectedRef.current + (sharedRound ? roundStop === undefined : !loopDetectedRef.current) ) { const executableToolCallRequests: ToolCallRequestInfo[] = []; const duplicateResponses: Array<{ @@ -4654,36 +4711,47 @@ export const useLlmStream = ( const providerSignal = inheritedToolContinuationOwner ? processingSignal : abortSignal; - const stream = llmClient.sendMessageStream( - finalQueryToSend, - providerSignal, - prompt_id!, - { - ...sendOptions, - ...(goalBinding + const roundOptions: SendMessageOptions = { + ...sendOptions, + ...(goalBinding + ? { + goalPermit: goalBinding.permit, + goalTurnKey: goalBinding.turnKey, + goalSignal: goalBinding.controller.signal, + goalOrigin: goalBinding.origin, + getQueuedGoalTurnKey: () => + goalQueueRef?.current?.peekNextUserBatchKey(true), + } + : userAdmission ? { - goalPermit: goalBinding.permit, - goalTurnKey: goalBinding.turnKey, - goalSignal: goalBinding.controller.signal, - goalOrigin: goalBinding.origin, + goalTurnKey: userAdmission.turnKey, + goalSignal: turnController!.signal, + goalOrigin: 'user' as const, getQueuedGoalTurnKey: () => goalQueueRef?.current?.peekNextUserBatchKey(true), } - : userAdmission - ? { - goalTurnKey: userAdmission.turnKey, - goalSignal: turnController!.signal, - goalOrigin: 'user' as const, - getQueuedGoalTurnKey: () => - goalQueueRef?.current?.peekNextUserBatchKey(true), - } - : {}), - }, - ); + : {}), + }; + const source = config.isUnifiedAgentLoopEnabled?.() + ? { + client: llmClient, + signal: providerSignal, + round: { + parts: finalQueryToSend, + promptId: prompt_id!, + options: roundOptions, + }, + } + : llmClient.sendMessageStream( + finalQueryToSend, + providerSignal, + prompt_id!, + roundOptions, + ); metadata?.onRequestStarted?.(); const processingResult = await processLlmStreamEvents( - stream, + source, userMessageTimestamp, processingSignal, submitType, diff --git a/packages/core/src/core/agent-loop.test.ts b/packages/core/src/core/agent-loop.test.ts index 295bf23c7ad..2d97ea2ab01 100644 --- a/packages/core/src/core/agent-loop.test.ts +++ b/packages/core/src/core/agent-loop.test.ts @@ -239,32 +239,50 @@ describe('runAgentLoop', () => { expect(raw.recordCompletedToolCall).toHaveBeenCalledWith('read_file', {}); }); - it('drops the calls of an attempt that a retry replaced', async () => { - const stale = toolRequest('stale'); - const fresh = toolRequest('fresh'); - const { client } = fakeClient([ - [requestEvent(stale), { type: LlmEventType.Retry }, requestEvent(fresh)], - [content], - ]); - const { executor, batches } = fakeExecutor(); - const discarded: Array = []; - - await runAgentLoop({ - client, - config: fakeConfig().config, - signal: new AbortController().signal, - first: first(), - executor, - host: { - onModelEvent: (event, calls) => { - if (event.type === LlmEventType.Retry) discarded.push(calls); - }, + it.each([ + ['a retry', { type: LlmEventType.Retry }], + [ + 'a model fallback', + { + type: LlmEventType.ModelFallback, + fromModel: 'primary-model', + toModel: 'fallback-model', + fallbackIndex: 1, }, - }); + ], + ] as const)( + 'drops the calls of an attempt that %s replaced', + async (_, replaced) => { + const stale = toolRequest('stale'); + const fresh = toolRequest('fresh'); + const { client } = fakeClient([ + [ + requestEvent(stale), + replaced as unknown as ServerLlmStreamEvent, + requestEvent(fresh), + ], + [content], + ]); + const { executor, batches } = fakeExecutor(); + const discarded: Array = []; - expect(discarded).toEqual([[stale]]); - expect(batches).toEqual([[fresh]]); - }); + await runAgentLoop({ + client, + config: fakeConfig().config, + signal: new AbortController().signal, + first: first(), + executor, + host: { + onModelEvent: (event, calls) => { + if (event.type === replaced.type) discarded.push(calls); + }, + }, + }); + + expect(discarded).toEqual([[stale]]); + expect(batches).toEqual([[fresh]]); + }, + ); it.each([ [LlmEventType.LoopDetected, 'loop_detected'], diff --git a/packages/vscode-ide-companion/schemas/settings.schema.json b/packages/vscode-ide-companion/schemas/settings.schema.json index d429a17d7b7..8e3c41a9056 100644 --- a/packages/vscode-ide-companion/schemas/settings.schema.json +++ b/packages/vscode-ide-companion/schemas/settings.schema.json @@ -4605,7 +4605,7 @@ "default": false }, "unifiedAgentLoop": { - "description": "Run headless sessions (`-p` and stream-json) and ACP sessions (Zed, VS Code, Web Shell, channels) on the shared agent loop (experimental). The interactive TUI and subagents keep their current loop for now. Can also be enabled via QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1 environment variable.", + "description": "Run headless sessions (`-p` and stream-json) and ACP sessions (Zed, VS Code, Web Shell, channels) on the shared agent loop (experimental). The Ink TUI sends its model requests through it, while its tool runs, OpenTUI and subagents keep their current loop for now. Can also be enabled via QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1 environment variable.", "type": "boolean", "default": false }, diff --git a/packages/web-shell/client/settings/messages.ts b/packages/web-shell/client/settings/messages.ts index d94ed34a2a7..9e6aee74d0a 100644 --- a/packages/web-shell/client/settings/messages.ts +++ b/packages/web-shell/client/settings/messages.ts @@ -433,7 +433,7 @@ export const SETTINGS_MESSAGES_ZH: Record = { '启用持久化的工作区 Agent 在共享任务线程上协作（实验性）。与 Agent Team 相互独立：两个开关互不隐含。启用只表示允许协作；把 Agent 开放给外部调用方、信任某个连接、注册 host 仍各自需要显式配置。也可通过 QWEN_CODE_ENABLE_AGENT_COLLABORATION=1 环境变量启用。', 'settings.label.experimental.unifiedAgentLoop': '统一 Agent Loop', 'settings.description.experimental.unifiedAgentLoop': - '让 headless 会话（`-p` 与 stream-json）和 ACP 会话（Zed、VS Code、Web Shell、channel）运行在共享的 agent loop 上（实验性）。交互式 TUI 和子代理暂时保留各自现有的 loop。也可通过 QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1 环境变量启用。', + '让 headless 会话（`-p` 与 stream-json）和 ACP 会话（Zed、VS Code、Web Shell、channel）运行在共享的 agent loop 上（实验性）。Ink TUI 的模型请求也经由它发送，而 Ink TUI 的工具执行、OpenTUI 和子代理暂时保留各自现有的 loop。也可通过 QWEN_CODE_ENABLE_UNIFIED_AGENT_LOOP=1 环境变量启用。', 'settings.label.experimental.artifact': '启用 Artifacts', 'settings.description.experimental.artifact': '启用 artifact 工具，默认开启。在交互式非 SDK 会话中，模型可以把自包含 HTML 页面发布为交互式 Artifact 并在浏览器中打开；非 SDK 的 daemon 会话可使用仅记录元数据的 record_artifact 工具。设为 false 或用 QWEN_CODE_DISABLE_ARTIFACT=1 可同时禁用两者。', diff --git a/scripts/tests/ink-agent-loop-parity-retry.test.ts b/scripts/tests/ink-agent-loop-parity-retry.test.ts new file mode 100644 index 00000000000..9b8e55f58a6 --- /dev/null +++ b/scripts/tests/ink-agent-loop-parity-retry.test.ts @@ -0,0 +1,31 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { readFileSync } from 'node:fs'; +import { join, resolve } from 'node:path'; +import { describe, expect, it } from 'vitest'; + +// The Ink parity suite runs two TUI sessions per scenario in the required +// no-AK lane. A green run cannot show whether a stall would be retried, so this +// pin keeps the integration root's `retry: 2` from tripling one. +const source = readFileSync( + join( + resolve(import.meta.dirname, '../..'), + 'integration-tests/interactive/ink-agent-loop-parity.test.ts', + ), + 'utf8', +); + +describe('Ink agent loop parity suite source pins', () => { + it('runs each scenario once, without the root config retries', () => { + // Both halves: the scenario table takes the options, and they keep + // `retry: 0`. The ceiling is free to change. + expect(source).toMatch( + /it\.each\(scenarios\)\(\s*'\$name',\s*PARITY_CASE_OPTIONS,/, + ); + expect(source).toMatch(/const PARITY_CASE_OPTIONS = \{[^}]*\bretry: 0\b/); + }); +}); diff --git a/scripts/tests/no-ak-integration-ci.test.js b/scripts/tests/no-ak-integration-ci.test.js index 8eed61d10b9..ed4089747aa 100644 --- a/scripts/tests/no-ak-integration-ci.test.js +++ b/scripts/tests/no-ak-integration-ci.test.js @@ -241,6 +241,7 @@ describe('no-AK integration CI wiring', () => { './interactive/statusline-parity.test.ts', './interactive/tui-editor-parity.test.ts', './interactive/tui-input-ownership.test.ts', + './interactive/ink-agent-loop-parity.test.ts', './sdk-typescript/abort-and-lifecycle.test.ts', './sdk-typescript/permission-control.test.ts', './sdk-typescript/sdk-mcp-server.test.ts',