diff --git a/.changeset/steer-background-waits.md b/.changeset/steer-background-waits.md new file mode 100644 index 00000000000..a72f764f97a --- /dev/null +++ b/.changeset/steer-background-waits.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Allow steering messages to interrupt waits for background tasks. diff --git a/docs/en/guides/interaction.md b/docs/en/guides/interaction.md index cd17ae92f16..1372c5113ed 100644 --- a/docs/en/guides/interaction.md +++ b/docs/en/guides/interaction.md @@ -126,6 +126,8 @@ The input box remains usable while the agent is thinking or calling tools, and s - **`Esc` / `Ctrl-C`**: interrupt the current turn - **`Ctrl-O`**: globally toggle the collapsed/expanded state of tool output and compaction summaries +When the agent is waiting for background tasks through `WaitFor`, pressing `Ctrl-S` ends that wait early. Background tasks keep running and existing tool results are preserved. If other foreground tools remain in the same batch, the agent processes your message after they return. + ## External editor Press `Ctrl-G` to send the current input content to an external editor. When you save and close, the text is written back into the input box; if you close without saving, the original content is preserved. This is handy when you need to enter large blocks of text or content with complex formatting. diff --git a/docs/en/reference/tools.md b/docs/en/reference/tools.md index df56d7cb15a..d2d16365c9d 100644 --- a/docs/en/reference/tools.md +++ b/docs/en/reference/tools.md @@ -131,7 +131,7 @@ Background task tools manage tasks started via `Bash`, `Agent`, or `AskUserQuest **`TaskStop`** accepts a `task_id` and optional `reason` (defaults to `Stopped by TaskStop`). Safe to call on tasks that are already in a terminal state. -**`WaitFor`** suspends the current turn until a background task finishes or the timeout elapses. Parameters: `timeout` (required, in seconds, max 600) and optional `task_id`. Without `task_id`, the wait ends as soon as any background task that was running at call time finishes; when no background tasks are running, it returns immediately. A timeout is not an error — the result lists the tasks still running, and the Agent can wait again or do other work meanwhile. A task whose result was reported by `WaitFor` does not also produce an automatic completion notification. +**`WaitFor`** suspends the current turn until a background task finishes, the timeout elapses, or a steer message arrives. Parameters: `timeout` (required, in seconds, max 600) and optional `task_id`. Without `task_id`, the wait ends as soon as any background task that was running at call time finishes; when no background tasks are running, it returns immediately. A timeout is not an error — the result lists the tasks still running, and the Agent can wait again or do other work meanwhile. Steering (`Ctrl-S` in the terminal) ends the wait early; background tasks keep running and still notify the agent on completion. A task whose result was reported by `WaitFor` does not also produce an automatic completion notification. ## Scheduled Tasks diff --git a/docs/zh/guides/interaction.md b/docs/zh/guides/interaction.md index 75b56d3026a..28c183f5e8a 100644 --- a/docs/zh/guides/interaction.md +++ b/docs/zh/guides/interaction.md @@ -126,6 +126,8 @@ Agent 思考或调用工具时，输入框仍然可用，支持以下额外操 - **`Esc` / `Ctrl-C`**：中断当前轮次 - **`Ctrl-O`**：全局切换工具输出和压缩摘要的折叠状态 +Agent 正通过 `WaitFor` 等待后台任务时，按 `Ctrl-S` 会提前结束本次等待。后台任务继续运行，已有工具结果保留；如果同批还有其他前台工具，Agent 会在它们返回后处理新消息。 + ## 外部编辑器 按 `Ctrl-G` 把当前输入内容发给外部编辑器，保存后回填到输入框，不保存则保持原样。适合需要输入大段文本或带格式内容的场景。 diff --git a/docs/zh/reference/tools.md b/docs/zh/reference/tools.md index f7faf9cda5c..2fc2d967b29 100644 --- a/docs/zh/reference/tools.md +++ b/docs/zh/reference/tools.md @@ -131,7 +131,7 @@ Plan 模式是一种受约束的工作状态：进入后 `Write` 与 `Edit` 只 **`TaskStop`** 接受 `task_id` 和可选的 `reason`（默认 `Stopped by TaskStop`）。对已处于终止状态的任务也能安全调用。 -**`WaitFor`** 把当前轮次挂起，直到后台任务结束或超时。参数：`timeout`（必填，单位秒，上限 600）和可选的 `task_id`。不传 `task_id` 时，调用时刻运行中的任意一个后台任务结束即返回；当前没有运行中的后台任务时立即返回。超时不是错误——结果会列出仍在运行的任务，Agent 可以再次等待，也可以先处理其他工作。已通过 `WaitFor` 汇报结果的任务不会再推送自动完成通知。 +**`WaitFor`** 把当前轮次挂起，直到后台任务结束、超时或收到 steer 消息。参数：`timeout`（必填，单位秒，上限 600）和可选的 `task_id`。不传 `task_id` 时，调用时刻运行中的任意一个后台任务结束即返回；当前没有运行中的后台任务时立即返回。超时不是错误——结果会列出仍在运行的任务，Agent 可以再次等待，也可以先处理其他工作。Steer（终端中按 `Ctrl-S`）会提前结束本次等待，后台任务继续运行，完成后仍会自动通知。已通过 `WaitFor` 汇报结果的任务不会再推送自动完成通知。 ## 定时任务 diff --git a/packages/agent-core-v2/src/agent/loop/loopService.ts b/packages/agent-core-v2/src/agent/loop/loopService.ts index 0b4ec99c1a2..d356783ce01 100644 --- a/packages/agent-core-v2/src/agent/loop/loopService.ts +++ b/packages/agent-core-v2/src/agent/loop/loopService.ts @@ -163,6 +163,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { initialTurnId: this.states.get(turnKey).nextTurnId, trace: () => this.activeRequestTrace, toolTurnId: () => this.active?.id, + steerSignal: () => this.active?.steerController.signal, source: () => this.active === undefined ? undefined @@ -218,11 +219,13 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { const id = prompt.promptId ?? randomUUID(); this.nudges.push({ contextMessage: message, + steer: true, bypassMaxSteps: false, turnScoped: false, onConsume: prompt.onMaterialize, onDrop: undefined, }); + active.steerController.abort(abortError('Steered by new input')); this.machineEngine().submit({ id, message: machineUserMessage(message) }); this.machineEngine().steer(id); return active.turn; @@ -503,6 +506,12 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { if (turn.stopRequested) return { type: 'fail' }; if (turn.failedStep !== undefined) return { type: 'fail' }; const consumed = this.mirrorConsumedNudges(turn); + if ( + turn.steerController.signal.aborted && + !this.nudges.slice(this.nudgeCursor).some((nudge) => nudge.steer && !nudge.dropped) + ) { + turn.steerController = new AbortController(); + } if (turn.toolStopRequested && consumed.live === 0) return { type: 'fail' }; const stepOrdinal = Math.max(this.engine?.currentStep() ?? 0, turn.steps + 1); const maxSteps = this.config.get(LOOP_CONTROL_SECTION)?.maxStepsPerTurn; @@ -652,6 +661,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { id, reservation, controller: reservation.controller, + steerController: new AbortController(), turn, startedAt: Date.now(), steps: 0, @@ -1528,6 +1538,7 @@ interface TurnReservation { interface Nudge { readonly contextMessage?: ContextMessage; + readonly steer?: boolean; readonly bypassMaxSteps: boolean; readonly turnScoped: boolean; readonly onConsume?: () => void; @@ -1566,6 +1577,7 @@ interface ActiveTurn { readonly id: number; readonly reservation: TurnReservation; readonly controller: AbortController; + steerController: AbortController; readonly turn: MutableTurn; readonly startedAt: number; steps: number; diff --git a/packages/agent-core-v2/src/agent/loop/machine/engine.ts b/packages/agent-core-v2/src/agent/loop/machine/engine.ts index 629f2af3ce3..1287783b40a 100644 --- a/packages/agent-core-v2/src/agent/loop/machine/engine.ts +++ b/packages/agent-core-v2/src/agent/loop/machine/engine.ts @@ -114,6 +114,7 @@ export interface CreateMachineEngineOptions { readonly trace?: () => LLMRequestTrace | undefined; readonly source?: () => AgentLLMRequestSource | undefined; readonly toolTurnId?: () => number | undefined; + readonly steerSignal?: () => AbortSignal | undefined; readonly gate?: (signal: AbortSignal) => Promise; readonly onTrace?: (trace: LLMRequestTrace) => void; readonly onEvent?: (event: MachineEngineEvent) => void; @@ -253,6 +254,7 @@ export function createMachineEngine(options: CreateMachineEngineOptions): Machin toolExecutor: options.toolExecutor, toolInfos: options.toolInfos, turnId: () => options.toolTurnId?.() ?? 0, + steerSignal: options.steerSignal, trace: options.trace, onToolCall: (payload) => { publish({ diff --git a/packages/agent-core-v2/src/agent/loop/machine/tools.ts b/packages/agent-core-v2/src/agent/loop/machine/tools.ts index 42042810228..aafebbc29cb 100644 --- a/packages/agent-core-v2/src/agent/loop/machine/tools.ts +++ b/packages/agent-core-v2/src/agent/loop/machine/tools.ts @@ -34,6 +34,7 @@ export interface CreateMachineToolsOptions { readonly toolExecutor: IAgentToolExecutorService; readonly toolInfos: readonly ToolInfo[]; readonly turnId: () => number; + readonly steerSignal?: () => AbortSignal | undefined; readonly trace?: () => LLMRequestTrace | undefined; readonly onToolCall?: (payload: ToolCallStartedPayload) => void; readonly onToolResult?: (toolCallId: string, result: AgentToolResult) => void; @@ -104,6 +105,7 @@ export function createMachineTools(options: CreateMachineToolsOptions): MachineT try { const stream = options.toolExecutor.execute(calls, { signal, + steerSignal: options.steerSignal?.(), turnId: options.turnId(), trace: options.trace?.(), onToolCall: options.onToolCall, diff --git a/packages/agent-core-v2/src/agent/toolExecutor/toolExecutor.ts b/packages/agent-core-v2/src/agent/toolExecutor/toolExecutor.ts index d88df865283..c1cf3349d0e 100644 --- a/packages/agent-core-v2/src/agent/toolExecutor/toolExecutor.ts +++ b/packages/agent-core-v2/src/agent/toolExecutor/toolExecutor.ts @@ -22,6 +22,7 @@ export interface ToolCallStartedPayload { export interface ToolExecutorExecuteOptions { readonly signal: AbortSignal; + readonly steerSignal?: AbortSignal; readonly turnId: number; readonly trace?: LLMRequestTrace; readonly onToolCall?: (payload: ToolCallStartedPayload) => void; diff --git a/packages/agent-core-v2/src/agent/toolExecutor/toolExecutorService.ts b/packages/agent-core-v2/src/agent/toolExecutor/toolExecutorService.ts index 8d23456a2f5..0432fe51ae6 100644 --- a/packages/agent-core-v2/src/agent/toolExecutor/toolExecutorService.ts +++ b/packages/agent-core-v2/src/agent/toolExecutor/toolExecutorService.ts @@ -527,6 +527,7 @@ export class AgentToolExecutorService implements IAgentToolExecutorService { trace: options.trace, metadata, signal, + steerSignal: options.steerSignal, onUpdate: (update) => { if (signal.aborted) return; this.dispatchToolProgress(call, update, options); diff --git a/packages/agent-core-v2/src/agent/tools/task/task-wait/taskWaitTool.ts b/packages/agent-core-v2/src/agent/tools/task/task-wait/taskWaitTool.ts index 498054b24f0..9856173d31d 100644 --- a/packages/agent-core-v2/src/agent/tools/task/task-wait/taskWaitTool.ts +++ b/packages/agent-core-v2/src/agent/tools/task/task-wait/taskWaitTool.ts @@ -15,7 +15,7 @@ import { formatPlainObject } from '#/agent/task/tools/format'; import { formatTaskList } from '#/agent/tools/task/task-list/taskListTool'; import { IFlagService } from '#/app/flag/flag'; import { ITelemetryService } from '#/app/telemetry/telemetry'; -import { abortError, linkAbortSignal } from '#/_base/utils/abort'; +import { abortError, isAbortError, linkAbortSignal } from '#/_base/utils/abort'; import { WAIT_FOR_FLAG_ID } from './flag'; import { IWaitForTool, WaitForInputSchema, type WaitForInput } from './task-wait'; import WAIT_FOR_DESCRIPTION from './task-wait.md?raw'; @@ -26,7 +26,7 @@ const PAGING_HINT_LINES = 300; const PROGRESS_INTERVAL_MS = 1_000; -type WaitForOutcome = 'completed' | 'timed_out' | 'task_not_found' | 'aborted'; +type WaitForOutcome = 'completed' | 'timed_out' | 'task_not_found' | 'aborted' | 'interrupted'; function terminalReason(info: AgentTaskInfo): 'timed_out' | 'stopped' | 'failed' | undefined { if (info.status === 'timed_out') return 'timed_out'; @@ -166,13 +166,23 @@ export class WaitForTool implements IWaitForTool { } let waited: AgentTaskInfo | undefined; + const signal = ctx.steerSignal === undefined + ? ctx.signal + : AbortSignal.any([ctx.signal, ctx.steerSignal]); const progress = startWaitProgress(args, this.tasks, ctx.onUpdate, startedAt); try { waited = args.task_id === undefined - ? await this.waitAny(runningAtStart, timeoutMs, ctx.signal) - : await this.tasks.wait(args.task_id, timeoutMs, ctx.signal); + ? await this.waitAny(runningAtStart, timeoutMs, signal) + : await this.tasks.wait(args.task_id, timeoutMs, signal); } catch (error) { + if ( + !ctx.signal.aborted && ctx.steerSignal?.aborted && + (error === ctx.steerSignal.reason || isAbortError(error)) + ) { + this.track(args, startedAt, timeoutMs, 'interrupted', 0); + return { output: this.formatInterrupted(args, startedAt, timeoutMs), isError: false }; + } this.track(args, startedAt, timeoutMs, 'aborted', 0); throw error; } finally { @@ -254,6 +264,24 @@ export class WaitForTool implements IWaitForTool { return lines.join('\n'); } + private formatInterrupted(args: WaitForInput, startedAt: number, timeoutMs: number): string { + const lines = [ + formatPlainObject({ + waitStatus: 'interrupted', + reason: 'steer', + taskId: args.task_id, + waitedMs: Date.now() - startedAt, + timeoutMs, + }), + 'New input ended this wait early. Read the new input before deciding what to do next. Background tasks have not been stopped; completion still arrives via automatic notification.', + ]; + const running = this.tasks.list(true); + if (running.length > 0) { + lines.push('', '[still_running]', formatTaskList(running, true)); + } + return lines.join('\n'); + } + private async formatCompleted( finished: AgentTaskInfo, extras: readonly AgentTaskInfo[], diff --git a/packages/agent-core-v2/src/app/telemetry/events.ts b/packages/agent-core-v2/src/app/telemetry/events.ts index 465cce28e37..6ebd9a3c98a 100644 --- a/packages/agent-core-v2/src/app/telemetry/events.ts +++ b/packages/agent-core-v2/src/app/telemetry/events.ts @@ -254,7 +254,7 @@ export interface BackgroundTaskCompletedEvent { } export interface WaitForCompletedEvent { - outcome: 'completed' | 'timed_out' | 'task_not_found' | 'aborted'; + outcome: 'completed' | 'timed_out' | 'task_not_found' | 'aborted' | 'interrupted'; timeout_ms: number; waited_ms: number; has_task_id: boolean; diff --git a/packages/agent-core-v2/src/tool/toolContract.ts b/packages/agent-core-v2/src/tool/toolContract.ts index af8db5f7417..e319c4fa1c8 100644 --- a/packages/agent-core-v2/src/tool/toolContract.ts +++ b/packages/agent-core-v2/src/tool/toolContract.ts @@ -71,6 +71,7 @@ export interface ExecutableToolContext { readonly trace?: LLMRequestTrace; readonly metadata?: unknown; readonly signal: AbortSignal; + readonly steerSignal?: AbortSignal; readonly onUpdate?: ((update: ToolUpdate) => void) | undefined; readonly onForegroundTaskStart?: ((taskId: string) => void) | undefined; } diff --git a/packages/agent-core-v2/test/agent/task/tools/task-tools.test.ts b/packages/agent-core-v2/test/agent/task/tools/task-tools.test.ts index 2d64cbfb930..7584a8348d7 100644 --- a/packages/agent-core-v2/test/agent/task/tools/task-tools.test.ts +++ b/packages/agent-core-v2/test/agent/task/tools/task-tools.test.ts @@ -1,5 +1,6 @@ import { PassThrough, Readable, type Writable } from 'node:stream'; +import { createControlledPromise } from '@antfu/utils'; import { describe, expect, it, vi } from 'vitest'; import { @@ -32,6 +33,10 @@ import { SubagentTask } from '#/agent/tools/agent/subagent-task'; import type { SubagentTaskInfo } from '#/agent/tools/agent/subagent-task'; import { IWaitForTool } from '#/agent/tools/task/task-wait/task-wait'; import { IAgentLoopService } from '#/agent/loop/loop'; +import { IAgentProfileService } from '#/agent/profile/profile'; +import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; +import { ToolProgress } from '#/agent/toolExecutor/toolExecutorEvents'; +import { IEventBus } from '#/app/event/eventBus'; import { executeTool } from '../../../tools/fixtures/execute-tool'; import { recordingTelemetry, type TelemetryRecord } from '../../../app/telemetry/stubs'; import { stubFlag } from '../../../app/flag/stubs'; @@ -1134,6 +1139,192 @@ describe('WaitForTool (harness)', () => { throw new Error(`Timed out waiting for task to terminate: ${taskId}`); } + it.each(['specific', 'any'] as const)('steers out of a running %s wait without losing tool history or stopping the background task', async (target) => { + const ctx = createTestAgent(); + const slow = controllableProcess(); + try { + await ctx.restorePersisted(); + ctx.get(IAgentProfileService).update({ activeToolNames: ['TaskList', 'WaitFor'] }); + const tasks = ctx.get(IAgentTaskService); + const taskId = tasks.registerTask(new ProcessTask(slow.proc, 'sleep 60', 'background work')); + ctx.mockNextResponse( + { type: 'function', id: 'list-before-wait', name: 'TaskList', arguments: '{}' }, + { type: 'function', id: 'wait-for-task', name: 'WaitFor', arguments: JSON.stringify({ timeout: 600, task_id: target === 'specific' ? taskId : undefined }) }, + ); + ctx.mockNextResponse({ type: 'text', text: 'Handling the new request.' }); + + const waiting = ctx.once('tool.progress'); + await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Wait for the background work.' }] }); + await waiting; + await ctx.rpc.steer({ input: [{ type: 'text', text: 'Handle this new request first.' }] }); + + await vi.waitFor(() => { + expect(ctx.llmCalls).toHaveLength(2); + }, { timeout: 1_000 }); + const history = ctx.llmCalls[1]!.history; + expect(history.filter((message) => message.role === 'tool')).toMatchObject([ + { toolCallId: 'list-before-wait', content: [{ type: 'text', text: expect.stringContaining(taskId) }] }, + { toolCallId: 'wait-for-task', content: [{ type: 'text', text: expect.stringContaining('wait_status: interrupted') }] }, + ]); + expect(history.at(-1)).toMatchObject({ + role: 'user', + content: [{ type: 'text', text: 'Handle this new request first.' }], + }); + expect(ctx.allEvents).not.toContainEqual(expect.objectContaining({ + event: 'tool.result', + args: expect.objectContaining({ toolCallId: 'wait-for-task', isError: true }), + })); + expect(tasks.getTask(taskId)?.status).toBe('running'); + expect(slow.proc.kill).not.toHaveBeenCalled(); + await ctx.get(IAgentLoopService).settled(); + ctx.mockNextResponse({ type: 'text', text: 'The background work has finished.' }); + const notified = ctx.once('task.notified'); + slow.resolveWait(0); + await notified; + await ctx.get(IAgentLoopService).settled(); + expect(tasks.getTask(taskId)?.status).toBe('completed'); + expect(ctx.allEvents.filter((event) => event.event === 'task.notified')).toHaveLength(1); + await ctx.expectResumeMatches(); + } finally { + slow.resolveWait(0); + await ctx.dispose(); + } + }); + + it.each(['before-request', 'before-tool'] as const)('interrupts every wait after %s steering and can wait again after consuming the new input', async (timing) => { + const ctx = createTestAgent(); + const slow = controllableProcess(); + try { + await ctx.restorePersisted(); + ctx.get(IAgentProfileService).update({ activeToolNames: ['WaitFor'] }); + const tasks = ctx.get(IAgentTaskService); + const taskId = tasks.registerTask(new ProcessTask(slow.proc, 'sleep 60', 'background work')); + ctx.mockNextResponse( + { type: 'function', id: 'wait-specific', name: 'WaitFor', arguments: JSON.stringify({ timeout: 600, task_id: taskId }) }, + { type: 'function', id: 'wait-any', name: 'WaitFor', arguments: '{"timeout":600}' }, + ); + ctx.mockNextResponse({ + type: 'function', id: 'wait-again', name: 'WaitFor', + arguments: JSON.stringify({ timeout: 600, task_id: taskId }), + }); + ctx.mockNextResponse({ type: 'text', text: 'The background work has finished.' }); + const steer = async () => { + await ctx.rpc.steer({ input: [{ type: 'text', text: 'Check this message before waiting again.' }] }); + await ctx.rpc.steer({ input: [{ type: 'text', text: 'Keep the background task running.' }] }); + }; + if (timing === 'before-request') { + ctx.get(IAgentLoopService).hooks.onWillBeginStep.register('steer-before-request', async (event, next) => { + if (event.step === 1) await steer(); + await next(); + }); + } else { + ctx.get(IAgentToolExecutorService).onWillExecuteTool((event) => { + if (event.toolCall.id === 'wait-specific') event.waitUntil(steer()); + }); + } + const waitingAgain = createControlledPromise(); + ctx.get(IEventBus).subscribe(ToolProgress, (event) => { + if (event.toolCallId === 'wait-again') waitingAgain.resolve(); + }); + + await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Wait for the background work.' }] }); + await waitingAgain; + + expect(ctx.llmCalls[1]?.history.filter((message) => message.role === 'tool')).toMatchObject([ + { toolCallId: 'wait-specific', content: [{ text: expect.stringContaining('wait_status: interrupted') }] }, + { toolCallId: 'wait-any', content: [{ text: expect.stringContaining('wait_status: interrupted') }] }, + ]); + expect(ctx.llmCalls[1]?.history.at(-1)).toMatchObject({ + role: 'user', + content: [{ text: 'Check this message before waiting again.\n\nKeep the background task running.' }], + }); + expect(tasks.getTask(taskId)?.status).toBe('running'); + slow.resolveWait(0); + await ctx.get(IAgentLoopService).settled(); + expect(ctx.llmCalls).toHaveLength(3); + expect(ctx.llmCalls[2]?.history.find((message) => message.toolCallId === 'wait-again')).toMatchObject({ + content: [{ text: expect.stringContaining('wait_status: completed') }], + }); + expect(ctx.allEvents.filter((event) => event.event === 'task.notified')).toHaveLength(0); + await ctx.expectResumeMatches(); + } finally { + slow.resolveWait(0); + await ctx.dispose(); + } + }); + + it.each(['steer-first', 'completion-first'] as const)('reports task completion once when it races with steering (%s)', async (order) => { + const ctx = createTestAgent(); + const slow = controllableProcess(); + try { + await ctx.restorePersisted(); + ctx.get(IAgentProfileService).update({ activeToolNames: ['WaitFor'] }); + const tasks = ctx.get(IAgentTaskService); + const taskId = tasks.registerTask(new ProcessTask(slow.proc, 'sleep 60', 'background work')); + ctx.mockNextResponse({ + type: 'function', id: 'racing-wait', name: 'WaitFor', + arguments: JSON.stringify({ timeout: 600, task_id: taskId }), + }); + ctx.mockNextResponse({ type: 'text', text: 'Handling the new request.' }); + ctx.mockNextResponse({ type: 'text', text: 'The background work has finished.' }); + const waiting = ctx.once('tool.progress'); + await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Wait for the background work.' }] }); + await waiting; + + slow.pushOutput('BACKGROUND-RESULT'); + if (order === 'completion-first') slow.resolveWait(0); + const steered = ctx.rpc.steer({ input: [{ type: 'text', text: 'Handle the new request too.' }] }); + if (order === 'steer-first') slow.resolveWait(0); + await steered; + await waitForTerminal(tasks, taskId); + await vi.waitFor(() => { + const deliveries = ctx.context.get().filter((message) => + (message.origin?.kind === 'task' && message.origin.taskId === taskId) || + (message.toolCallId === 'racing-wait' && message.content.some((part) => + part.type === 'text' && part.text.includes('wait_status: completed'), + )), + ); + expect(deliveries).toHaveLength(1); + }); + await ctx.get(IAgentLoopService).settled(); + + const history = ctx.context.get(); + expect(await tasks.readOutput(taskId)).toBe('BACKGROUND-RESULT'); + expect(history.filter((message) => message.content.some((part) => + part.type === 'text' && part.text === 'Handle the new request too.', + ))).toHaveLength(1); + expect(tasks.getTask(taskId)?.status).toBe('completed'); + expect(slow.proc.kill).not.toHaveBeenCalled(); + await ctx.expectResumeMatches(); + } finally { + slow.resolveWait(0); + await ctx.dispose(); + } + }); + + it('still cancels a wait when execution is aborted together with steering', async () => { + const ctx = createTestAgent(); + const slow = controllableProcess(); + try { + const tasks = ctx.get(IAgentTaskService); + const taskId = tasks.registerTask(new ProcessTask(slow.proc, 'sleep 60', 'background work')); + const cancelled = new AbortController(); + const steered = new AbortController(); + const pending = executeTool(ctx.get(IWaitForTool), { + ...context('cancelled-wait', { timeout: 600, task_id: taskId }, cancelled.signal), + steerSignal: steered.signal, + }); + steered.abort(); + cancelled.abort(); + + await expect(pending).rejects.toThrow('Aborted'); + expect(tasks.getTask(taskId)?.status).toBe('running'); + } finally { + slow.resolveWait(0); + await ctx.dispose(); + } + }); + it('waits for a real registered task end-to-end and suppresses its notification', async () => { const records: TelemetryRecord[] = []; const loop = stubLoopWithHooks();