From 39f866fc7386f30444fdb721870447e49d317752 Mon Sep 17 00:00:00 2001 From: can1357 Date: Mon, 15 Jun 2026 11:53:25 +0200 Subject: [PATCH] feat(agent): added interruptible tool polling for queued steering - Added an `interruptible` field to AgentTool and documented when it is honored. - Updated immediate-mode tool execution to poll steering during in-flight interruptible calls and abort them when steering is queued. - Marked the coding `job` tool as interruptible and added tests covering mid-wait aborts versus boundary-only steering drain. --- packages/agent/CHANGELOG.md | 1 + packages/agent/src/agent-loop.ts | 27 ++++- packages/agent/src/types.ts | 9 ++ packages/agent/test/agent-loop.test.ts | 143 +++++++++++++++++++++++++ packages/coding-agent/CHANGELOG.md | 1 + packages/coding-agent/src/tools/job.ts | 1 + 6 files changed, 181 insertions(+), 1 deletion(-) diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index 596a0f08c..d1ecb4620 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -3,6 +3,7 @@ ## [Unreleased] ### Added +- Added the `interruptible` tool field: when set, the agent loop may abort the tool mid-execution to deliver a queued steering message (honored only in `immediate` interrupt mode). - Added support for `gemini` and `gemma` as valid owned tool syntax values in environment configuration ## [15.13.2] - 2026-06-15 diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index 5f85d357e..e8b9eb3a0 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -74,6 +74,14 @@ const ABORTED: unique symbol = Symbol("agent-loop-aborted"); */ const MAX_PAUSED_TURN_CONTINUATIONS = 8; +/** + * Cadence (ms) for polling queued steering while an `interruptible` tool is in + * flight, so a steer cuts the wait short instead of sitting idle until the + * tool's own window elapses. A cheap synchronous queue check; latency-bounded + * at one tick. + */ +const STEERING_INTERRUPT_POLL_MS = 250; + class HarmonyLeakInterruption extends Error { constructor( readonly detection: HarmonyDetection, @@ -1797,7 +1805,24 @@ async function executeToolCalls( } } - await Promise.allSettled(tasks); + // While an interruptible tool is in flight (e.g. a `job` poll blocking on + // background work), a queued steer would otherwise wait out the tool's own + // window. Poll the steering queue and let checkSteering() abort the shared + // tool signal so the wait returns early; the boundary dequeue below then + // injects it. Gated on immediate-interrupt mode + an interruptible tool; + // checkSteering is idempotent (no-op once triggered). + const watchSteeringWhileRunning = + shouldInterruptImmediately && + (hasSteeringMessages !== undefined || getSteeringMessages !== undefined) && + records.some(r => r.tool?.interruptible === true); + const steeringWatchTimer = watchSteeringWhileRunning + ? setInterval(() => void checkSteering(), STEERING_INTERRUPT_POLL_MS) + : undefined; + try { + await Promise.allSettled(tasks); + } finally { + if (steeringWatchTimer !== undefined) clearInterval(steeringWatchTimer); + } // Yield after batch tool execution to let GC and I/O catch up, // especially when tool results are large (e.g. bash output). await yieldIfDue(); diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 56bf8b60f..04168adca 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -503,6 +503,15 @@ export interface AgentTool>) => "shared" | "exclusive"); /** If true, argument validation errors are non-fatal: raw args are passed to execute() instead of returning an error to the LLM. */ lenientArgValidation?: boolean; + /** + * If true, the agent loop may abort this tool mid-execution to deliver a + * queued steering message (instead of waiting for the tool to finish on its + * own). Set only on tools that purely *wait* and observe their abort signal + * cleanly (e.g. the `job` poll), so the abort surfaces the tool's current + * snapshot rather than corrupting a side effect. Honored only when + * `interruptMode` is "immediate". + */ + interruptible?: boolean; /** * Controls how the INTENT_FIELD (`_i`) is handled for this tool. * - `"require"` (default): `_i` is injected and required in the parameter schema. diff --git a/packages/agent/test/agent-loop.test.ts b/packages/agent/test/agent-loop.test.ts index 20dfc5929..6a67bf50f 100644 --- a/packages/agent/test/agent-loop.test.ts +++ b/packages/agent/test/agent-loop.test.ts @@ -842,6 +842,149 @@ describe("agentLoop with AgentMessage", () => { expect(sawInterruptInContext).toBe(true); }); + it("drains queued steering by aborting an interruptible tool mid-wait", async () => { + const toolSchema = z.object({}); + let steerReady = false; + let drained = false; + let observedAbort = false; + let resolvedByTimeout = false; + + const tool: AgentTool> = { + name: "wait", + label: "Wait", + description: "Blocks until aborted (mimics a job poll)", + parameters: toolSchema, + interruptible: true, + async execute(_toolCallId, _params, signal) { + steerReady = true; + const { promise, resolve } = Promise.withResolvers(); + if (signal?.aborted) { + resolve(); + } else { + const timer = setTimeout(() => { + resolvedByTimeout = true; + resolve(); + }, 2000); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(); + }, + { once: true }, + ); + } + await promise; + observedAbort = signal?.aborted === true; + return { content: [{ type: "text", text: "waited" }], details: {} }; + }, + }; + + const context: AgentContext = { systemPrompt: [""], messages: [], tools: [tool] }; + const mock = createMockModel({ + responses: [ + { content: [{ type: "toolCall", id: "tool-1", name: "wait", arguments: {} }] }, + { content: ["done"] }, + ], + }); + const config: AgentLoopConfig = { + model: mock.model, + convertToLlm: identityConverter, + interruptMode: "immediate", + hasSteeringMessages: () => steerReady && !drained, + getSteeringMessages: async () => { + if (steerReady && !drained) { + drained = true; + return [createUserMessage("interrupt")]; + } + return []; + }, + }; + + const events: AgentEvent[] = []; + for await (const event of agentLoop([createUserMessage("start")], context, config, undefined, mock.stream)) { + events.push(event); + } + + expect(observedAbort).toBe(true); + expect(resolvedByTimeout).toBe(false); + expect(drained).toBe(true); + expect( + events.some(e => e.type === "message_start" && e.message.role === "user" && e.message.content === "interrupt"), + ).toBe(true); + }); + + it("does not abort a non-interruptible tool mid-wait; steering still drains at the boundary", async () => { + const toolSchema = z.object({}); + let steerReady = false; + let drained = false; + let observedAbort = false; + let resolvedByTimeout = false; + + const tool: AgentTool> = { + name: "wait", + label: "Wait", + description: "Blocks on its own window (no interruptible flag)", + parameters: toolSchema, + async execute(_toolCallId, _params, signal) { + steerReady = true; + const { promise, resolve } = Promise.withResolvers(); + if (signal?.aborted) { + resolve(); + } else { + const timer = setTimeout(() => { + resolvedByTimeout = true; + resolve(); + }, 300); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(); + }, + { once: true }, + ); + } + await promise; + observedAbort = signal?.aborted === true; + return { content: [{ type: "text", text: "waited" }], details: {} }; + }, + }; + + const context: AgentContext = { systemPrompt: [""], messages: [], tools: [tool] }; + const mock = createMockModel({ + responses: [ + { content: [{ type: "toolCall", id: "tool-1", name: "wait", arguments: {} }] }, + { content: ["done"] }, + ], + }); + const config: AgentLoopConfig = { + model: mock.model, + convertToLlm: identityConverter, + interruptMode: "immediate", + hasSteeringMessages: () => steerReady && !drained, + getSteeringMessages: async () => { + if (steerReady && !drained) { + drained = true; + return [createUserMessage("interrupt")]; + } + return []; + }, + }; + + const events: AgentEvent[] = []; + for await (const event of agentLoop([createUserMessage("start")], context, config, undefined, mock.stream)) { + events.push(event); + } + + expect(observedAbort).toBe(false); + expect(resolvedByTimeout).toBe(true); + expect(drained).toBe(true); + expect( + events.some(e => e.type === "message_start" && e.message.role === "user" && e.message.content === "interrupt"), + ).toBe(true); + }); + it("leaves steering queued when the run is aborted while interrupted tools settle", async () => { // Regression: the mid-batch steering poll used to DEQUEUE the message into // a loop-local variable. An external abort while the in-flight tools were diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 1fbaffe37..40e51f49c 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -8,6 +8,7 @@ ### Changed +- Changed the `job` poll to return early when a steering message is queued, draining the steer immediately instead of waiting out the poll window. - Capped unexpected-stop auto-continuation to three retry attempts before giving up on repeated stops - Updated the `edit` tool's hashline prompt, grammar, and docs to recommend the `.=` inclusive range separator (`SWAP 1.=3:`); the legacy `..` form still parses. diff --git a/packages/coding-agent/src/tools/job.ts b/packages/coding-agent/src/tools/job.ts index b7b1f40c3..716bb3987 100644 --- a/packages/coding-agent/src/tools/job.ts +++ b/packages/coding-agent/src/tools/job.ts @@ -87,6 +87,7 @@ export class JobTool implements AgentTool { readonly description: string; readonly parameters = jobSchema; readonly strict = true; + readonly interruptible = true; readonly loadMode = "discoverable"; constructor(private readonly session: ToolSession) { this.description = prompt.render(jobDescription);