diff --git a/packages/agent/README.md b/packages/agent/README.md index be60e6d6f..687bfb91d 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -112,18 +112,35 @@ prompt("Read config.json") Tool execution mode is configurable: -- `parallel` (default): preflight tool calls sequentially, execute allowed tools concurrently, emit `tool_execution_end` as soon as each tool is finalized, then emit toolResult messages and `turn_end.toolResults` in assistant source order +- `parallel` (default): execute consecutive groups of parallel-capable calls concurrently, separated by individual sequential tool barriers - `sequential`: execute tool calls one by one, matching the historical behavior -In parallel mode, tool completion events follow tool completion order, but persisted toolResult messages still follow assistant source order. +The global mode is set through `toolExecution` in the agent config. +In parallel mode, a tool declaring `executionMode: "sequential"` executes alone at its position in the assistant message. +Calls never move across that barrier. +Tools declaring `"parallel"` or omitting `executionMode` can overlap within their consecutive group. +Global sequential mode runs every call one by one, including tools explicitly declaring `"parallel"`. -The mode can be set globally via `toolExecution` in the agent config, or per-tool via `executionMode` on `AgentTool`. If any tool call in a batch targets a tool with `executionMode: "sequential"`, the entire batch executes sequentially regardless of the global setting. +For example, `read A, read B, update_plan, read C, read D` executes as `[read A + read B]`, then `update_plan`, then `[read C + read D]` when `update_plan` declares sequential execution. -The `beforeToolCall` hook runs after `tool_execution_start` and validated argument parsing. It can block execution and attach `terminate: true` to the blocked result. The `afterToolCall` hook runs after tool execution finishes and before `tool_execution_end` and final tool result message events are emitted. +Each parallel group preflights its calls sequentially before executing permitted calls concurrently. +The `beforeToolCall` hook runs after `tool_execution_start` and argument validation, and can block execution or attach `terminate: true` to a blocked result. +The `afterToolCall` hook runs after execution and before `tool_execution_end` and result-message events. +Completion events follow finalization order within a group; result-message events, persisted results, and `turn_end.toolResults` follow assistant source order. +Execution, result hooks, completion events, and ordered result-message events all settle before the next phase begins preflight. +Async subscribers are awaited when using `Agent` or an awaited event sink with `runAgentLoop`; raw event streams remain observational. -Tools, blocked `beforeToolCall` results, and `afterToolCall` overrides can return `terminate: true` to hint that the automatic follow-up LLM call should be skipped. The loop only stops early when every finalized tool result in that batch sets `terminate: true`. Mixed batches continue normally. +Tools, blocked `beforeToolCall` results, and `afterToolCall` overrides can return `terminate: true` to skip the automatic follow-up LLM call when every finalized result in the entire assistant batch requests termination. +A terminating phase alone does not skip later phases. +Tool errors and blocked calls also allow subsequent phases to run. +Cancellation stops scheduling later phases while preserving active-phase cancellation handling. +Steering messages and the next model request remain at the batch boundary. -The `Agent` class accepts `shouldStopAfterTurn` in `AgentOptions`. Low-level loop callers can set the same hook in `AgentLoopConfig`: +The model selects independent calls; the scheduler does not infer file dependencies or shell side effects. +Existing file-mutation queues still coordinate calls within each parallel group. + +The `Agent` class accepts `shouldStopAfterTurn` in `AgentOptions`. +Low-level loop callers can set the same hook in `AgentLoopConfig`: ```typescript const stream = agentLoop( @@ -423,10 +440,10 @@ const readFileTool: AgentTool = { parameters: Type.Object({ path: Type.String({ description: "File path" }), }), - // Override execution mode for this tool (optional). - // "sequential" forces the entire batch to run one at a time. - // "parallel" allows concurrent execution with other tool calls. - // If omitted, the global toolExecution config applies. + // Scheduling metadata for global parallel mode (optional). + // "sequential" makes this call an exclusive barrier in source order. + // "parallel" or omitted metadata allows overlap within its group. + // Global sequential mode always runs calls one by one. executionMode: "sequential", execute: async (toolCallId, params, signal, onUpdate) => { const content = await fs.readFile(params.path, "utf-8"); @@ -497,7 +514,7 @@ const context: AgentContext = { const config: AgentLoopConfig = { model: getModel("openai", "gpt-4o"), convertToLlm: (msgs) => msgs.filter(m => ["user", "assistant", "toolResult"].includes(m.role)), - toolExecution: "parallel", // overridden by per-tool executionMode if set + toolExecution: "parallel", // sequential tools form ordered barriers beforeToolCall: async ({ toolCall, args, context }) => undefined, afterToolCall: async ({ toolCall, result, isError, context }) => undefined, }; diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index 241644b6d..886a6daee 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -682,13 +682,39 @@ async function executeToolCalls( emit: AgentEventSink, ): Promise { const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall"); - const hasSequentialToolCall = toolCalls.some( - (tc) => currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential", - ); - if (config.toolExecution === "sequential" || hasSequentialToolCall) { + if (config.toolExecution === "sequential") { return executeToolCallsSequential(currentContext, assistantMessage, toolCalls, config, signal, emit); } - return executeToolCallsParallel(currentContext, assistantMessage, toolCalls, config, signal, emit); + + const messages: ToolResultMessage[] = []; + let terminate = true; + for (let index = 0; index < toolCalls.length; ) { + if (signal?.aborted) break; + const sequential = + currentContext.tools?.find((tool) => tool.name === toolCalls[index].name)?.executionMode === "sequential"; + let end = index + 1; + if (!sequential) { + while ( + end < toolCalls.length && + currentContext.tools?.find((tool) => tool.name === toolCalls[end].name)?.executionMode !== "sequential" + ) { + end++; + } + } + const executePhase = sequential ? executeToolCallsSequential : executeToolCallsParallel; + const phase = await executePhase( + currentContext, + assistantMessage, + toolCalls.slice(index, end), + config, + signal, + emit, + ); + messages.push(...phase.messages); + terminate = terminate && phase.terminate; + index = end; + } + return { messages, terminate: messages.length > 0 && terminate }; } type ExecutedToolCallBatch = { diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 07fb4c5cc..dc57add0d 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -36,9 +36,11 @@ export type StreamFn = ( * Configuration for how tool calls from a single assistant message are executed. * * - "sequential": each tool call is prepared, executed, and finalized before the next one starts. - * - "parallel": tool calls are prepared sequentially, then allowed tools execute concurrently. - * `tool_execution_end` is emitted in tool completion order after each tool is finalized, - * while tool-result message artifacts are emitted later in assistant source order. + * - "parallel": consecutive parallel-capable calls form groups separated by individual + * tools declaring `executionMode: "sequential"`. Each group is prepared sequentially, + * then allowed tools execute concurrently. Completion events follow finalization order; + * tool-result message events follow assistant source order within each phase. + * Hooks and awaited events settle before the next phase is prepared. */ export type ToolExecutionMode = "sequential" | "parallel"; @@ -280,9 +282,9 @@ export interface AgentLoopConfig extends SimpleStreamOptions { /** * Tool execution mode. * - "sequential": execute tool calls one by one - * - "parallel": preflight tool calls sequentially, then execute allowed tools concurrently; - * emit `tool_execution_end` in tool completion order after each tool is finalized, - * then emit tool-result message artifacts later in assistant source order + * - "parallel": execute consecutive parallel-capable calls concurrently, separated by + * sequential tool barriers. Preflight each phase sequentially and await execution, + * result hooks, completion events, and source-ordered result events before the next phase. * * Default: "parallel" */ @@ -294,6 +296,8 @@ export interface AgentLoopConfig extends SimpleStreamOptions { * Return `{ block: true }` to prevent execution. The loop emits an error tool result instead. * A blocked result can also set `terminate: true` to participate in the batch early-termination rule. * The hook receives the agent abort signal and is responsible for honoring it. + * In parallel mode, preflight runs sequentially within the current phase, after all + * prior phases have settled. Sibling results within that phase are not yet available. */ beforeToolCall?: (context: BeforeToolCallContext, signal?: AbortSignal) => Promise; @@ -309,6 +313,8 @@ export interface AgentLoopConfig extends SimpleStreamOptions { * * Any omitted fields keep their original values. No deep merge is performed. * The hook receives the agent abort signal and is responsible for honoring it. + * Parallel siblings can finalize concurrently. All hooks and awaited result events + * in the current phase settle before preflight begins for the next phase. */ afterToolCall?: (context: AfterToolCallContext, signal?: AbortSignal) => Promise; } @@ -420,11 +426,12 @@ export interface AgentTool, ) => Promise>; /** - * Per-tool execution mode override. - * - "sequential": this tool must execute one at a time with other tool calls. - * - "parallel": this tool can execute concurrently with other tool calls. + * Per-tool scheduling metadata under global parallel execution. + * - "sequential": this call is an exclusive barrier in assistant source order. + * - "parallel": this call can execute concurrently within its consecutive group. * - * If omitted, the default execution mode applies. + * Omitted metadata is parallel-capable. Global sequential execution always runs + * calls one by one, including tools explicitly declaring "parallel". */ executionMode?: ToolExecutionMode; } diff --git a/packages/agent/test/agent-loop.test.ts b/packages/agent/test/agent-loop.test.ts index 264b90eb6..74347aa5d 100644 --- a/packages/agent/test/agent-loop.test.ts +++ b/packages/agent/test/agent-loop.test.ts @@ -1,3 +1,4 @@ +import { setImmediate } from "node:timers/promises"; import { type AssistantMessage, type AssistantMessageEvent, @@ -8,9 +9,16 @@ import { } from "@diffexai/diffex-ai"; import { Type } from "typebox"; import { describe, expect, it } from "vitest"; -import { agentLoop, agentLoopContinue } from "../src/agent-loop.ts"; +import { type AgentEventSink, agentLoop, agentLoopContinue, runAgentLoop } from "../src/agent-loop.ts"; import { setDefaultStreamFn } from "../src/index.ts"; -import type { AgentContext, AgentEvent, AgentLoopConfig, AgentMessage, AgentTool } from "../src/types.ts"; +import type { + AgentContext, + AgentEvent, + AgentLoopConfig, + AgentMessage, + AgentTool, + AgentToolCall, +} from "../src/types.ts"; // Mock stream for testing - mimics MockAssistantStream class MockAssistantStream extends EventStream { @@ -1062,7 +1070,7 @@ describe("agentLoop with AgentMessage", () => { expect(toolResultIds).toEqual(["tool-1", "tool-2"]); }); - it("should force sequential execution when one of multiple tools has executionMode=sequential", async () => { + it("should wait for a sequential tool before starting a following parallel tool", async () => { const toolSchema = Type.Object({ value: Type.String() }); const executionOrder: string[] = []; let releaseSlow: (() => void) | undefined; @@ -1113,7 +1121,7 @@ describe("agentLoop with AgentMessage", () => { const config: AgentLoopConfig = { model: createModel(), convertToLlm: identityConverter, - // parallel by default, but slowTool forces sequential + // Parallel by default, with slowTool acting as a barrier. }; let callIndex = 0; @@ -1677,6 +1685,397 @@ describe("agentLoop with AgentMessage", () => { }); }); +function gate() { + let resolve!: () => void; + const promise = new Promise((release) => { + resolve = release; + }); + return { promise, resolve }; +} + +function startToolBatch( + calls: AgentToolCall[], + tools: AgentTool[], + config: Partial = {}, + onEvent?: AgentEventSink, + signal?: AbortSignal, +) { + const events: AgentEvent[] = []; + const requests: Message[][] = []; + const result = runAgentLoop( + [createUserMessage("start")], + { systemPrompt: "", messages: [], tools }, + { model: createModel(), convertToLlm: identityConverter, ...config }, + async (event) => { + events.push(event); + await onEvent?.(event); + }, + signal, + (_model, context) => { + requests.push(structuredClone(context.messages)); + const response = new MockAssistantStream(); + const message = + requests.length === 1 + ? createAssistantMessage(calls, "toolUse") + : createAssistantMessage([{ type: "text", text: "done" }]); + queueMicrotask(() => + response.push({ type: "done", reason: message.stopReason === "toolUse" ? "toolUse" : "stop", message }), + ); + return response; + }, + ); + return { events, requests, result }; +} + +function schedulingCall(id: string, name = "parallel"): AgentToolCall { + return { type: "toolCall", id, name, arguments: { value: id } }; +} + +const schedulingSchema = Type.Object({ value: Type.String() }); + +function schedulingTool( + name: string, + execute: AgentTool["execute"], + executionMode?: AgentTool["executionMode"], +): AgentTool { + return { name, label: name, description: name, parameters: schedulingSchema, executionMode, execute }; +} + +describe("tool execution phases", () => { + it.each([ + { + name: "mixed groups", + modes: ["parallel", "parallel", "sequential", "parallel", "parallel"], + phases: [[0, 1], [2], [3, 4]], + }, + { name: "leading barrier", modes: ["sequential", "parallel", "parallel"], phases: [[0], [1, 2]] }, + { name: "trailing barrier", modes: ["parallel", "parallel", "sequential"], phases: [[0, 1], [2]] }, + { + name: "consecutive barriers", + modes: ["parallel", "sequential", "sequential", "parallel"], + phases: [[0], [1], [2], [3]], + }, + { name: "only barriers", modes: ["sequential", "sequential"], phases: [[0], [1]] }, + { name: "all parallel", modes: ["parallel", "parallel"], phases: [[0, 1]] }, + { + name: "omitted metadata", + modes: [undefined, undefined, "sequential", undefined, undefined], + phases: [[0, 1], [2], [3, 4]], + }, + { + name: "global sequential", + modes: ["parallel", "parallel", "sequential", "parallel"], + phases: [[0], [1], [2], [3]], + global: "sequential", + }, + ] as const)("schedules $name in exclusive ordered phases", async ({ modes, phases, ...scenario }) => { + const gates = modes.map(() => gate()); + const started = modes.map(() => gate()); + const executed: number[] = []; + const preflight: number[] = []; + const tools = modes.map((mode, index) => + schedulingTool( + `tool-${index}`, + async () => { + executed.push(index); + started[index].resolve(); + await gates[index].promise; + return { content: [], details: {} }; + }, + mode, + ), + ); + const calls = tools.map((tool, index) => schedulingCall(String(index), tool.name)); + const batch = startToolBatch(calls, tools, { + toolExecution: "global" in scenario ? scenario.global : undefined, + beforeToolCall: async ({ toolCall }) => { + preflight.push(Number(toolCall.id)); + return undefined; + }, + }); + try { + const expected: number[] = []; + for (const phase of phases) { + await started[phase[0]].promise; + await setImmediate(); + expected.push(...phase); + expect([...executed]).toEqual(expected); + expect([...preflight]).toEqual(expected); + for (const index of phase) gates[index].resolve(); + } + } finally { + for (const release of gates) release.resolve(); + await batch.result; + } + expect(batch.requests).toHaveLength(2); + expect( + batch.requests[1].filter((message) => message.role === "toolResult").map((message) => message.toolCallId), + ).toEqual(calls.map((call) => call.id)); + }); + + it("awaits sequential preflight, result hooks, completion listeners, and ordered message listeners before crossing barriers", async () => { + const ids = ["a", "b", "barrier", "c", "d"]; + const execution = new Map(ids.map((id) => [id, gate()])); + const started = new Map(ids.map((id) => [id, gate()])); + const stages = new Map( + [ + "preflight-b", + "hook-a", + "end-a", + "message-start-a", + "message-end-b", + "hook-barrier", + "end-barrier", + "message-end-barrier", + ].map((stage) => [stage, { entered: gate(), release: gate() }]), + ); + const waitAt = async (stage: string) => { + const control = stages.get(stage); + if (!control) return; + control.entered.resolve(); + await control.release.promise; + }; + const preflight: string[] = []; + const executed: string[] = []; + let steeringPolls = 0; + const execute: AgentTool["execute"] = async (id) => { + executed.push(id); + started.get(id)?.resolve(); + await execution.get(id)?.promise; + return { content: [{ type: "text", text: id }], details: {} }; + }; + const batch = startToolBatch( + ids.map((id) => schedulingCall(id, id === "barrier" ? "barrier" : "parallel")), + [schedulingTool("parallel", execute), schedulingTool("barrier", execute, "sequential")], + { + beforeToolCall: async ({ toolCall }) => { + preflight.push(toolCall.id); + await waitAt(`preflight-${toolCall.id}`); + return undefined; + }, + afterToolCall: async ({ toolCall }) => { + await waitAt(`hook-${toolCall.id}`); + return { content: [{ type: "text", text: `final:${toolCall.id}` }] }; + }, + getSteeringMessages: async () => { + steeringPolls++; + return steeringPolls === 2 ? [createUserMessage("steer")] : []; + }, + }, + async (event) => { + if (event.type === "tool_execution_end") await waitAt(`end-${event.toolCallId}`); + if ( + (event.type === "message_start" || event.type === "message_end") && + event.message.role === "toolResult" + ) { + await waitAt(`${event.type.replaceAll("_", "-")}-${event.message.toolCallId}`); + } + }, + ); + try { + await stages.get("preflight-b")?.entered.promise; + expect([...executed]).toEqual([]); + stages.get("preflight-b")?.release.resolve(); + await started.get("b")?.promise; + expect([...executed]).toEqual(["a", "b"]); + execution.get("b")?.resolve(); + await setImmediate(); + expect( + batch.events.filter((event) => event.type === "tool_execution_end").map((event) => event.toolCallId), + ).toEqual(["b"]); + execution.get("a")?.resolve(); + for (const stage of ["hook-a", "end-a", "message-start-a", "message-end-b"]) { + await stages.get(stage)?.entered.promise; + expect([...preflight]).toEqual(["a", "b"]); + expect(batch.requests).toHaveLength(1); + expect(steeringPolls).toBe(1); + stages.get(stage)?.release.resolve(); + } + await started.get("barrier")?.promise; + expect([...preflight]).toEqual(["a", "b", "barrier"]); + execution.get("barrier")?.resolve(); + for (const stage of ["hook-barrier", "end-barrier", "message-end-barrier"]) { + await stages.get(stage)?.entered.promise; + expect([...preflight]).toEqual(["a", "b", "barrier"]); + expect(batch.requests).toHaveLength(1); + stages.get(stage)?.release.resolve(); + } + await started.get("d")?.promise; + expect([...executed]).toEqual(ids); + execution.get("d")?.resolve(); + await setImmediate(); + execution.get("c")?.resolve(); + } finally { + for (const release of execution.values()) release.resolve(); + for (const stage of stages.values()) stage.release.resolve(); + await batch.result; + } + expect( + batch.events.filter((event) => event.type === "tool_execution_end").map((event) => event.toolCallId), + ).toEqual(["b", "a", "barrier", "d", "c"]); + for (const type of ["message_start", "message_end"] as const) { + expect( + batch.events.flatMap((event) => + event.type === type && event.message.role === "toolResult" ? [event.message.toolCallId] : [], + ), + ).toEqual(ids); + } + const messages = await batch.result; + expect(messages.filter((message) => message.role === "toolResult").map((message) => message.content)).toEqual( + ids.map((id) => [{ type: "text", text: `final:${id}` }]), + ); + expect(batch.requests[1].map((message) => message.role)).toEqual([ + "user", + "assistant", + ...ids.map(() => "toolResult"), + "user", + ]); + expect(batch.requests[1].at(-1)?.content).toBe("steer"); + }); + + it.each(["execute", "blocked", "invalid", "unknown", "result hook"])( + "continues later phases after a %s error", + async (failure) => { + const executed: string[] = []; + const finalized: string[] = []; + const execute: AgentTool["execute"] = async (id) => { + if (id === "barrier" && failure === "execute") throw new Error("execution failed"); + executed.push(id); + return { content: [], details: {} }; + }; + const barrier = schedulingCall("barrier", failure === "unknown" ? "missing" : "barrier"); + if (failure === "invalid") barrier.arguments = {}; + const batch = startToolBatch( + [schedulingCall("a"), barrier, schedulingCall("b")], + [schedulingTool("parallel", execute), schedulingTool("barrier", execute, "sequential")], + { + beforeToolCall: async ({ toolCall }) => + toolCall.id === "barrier" && failure === "blocked" + ? { block: true, reason: "blocked", terminate: true } + : undefined, + afterToolCall: async ({ toolCall }) => { + finalized.push(toolCall.id); + if (toolCall.id === "barrier" && failure === "result hook") throw new Error("result hook failed"); + return undefined; + }, + }, + ); + const messages = await batch.result; + expect(executed).toContain("b"); + expect( + messages + .filter((message) => message.role === "toolResult") + .map((message) => [message.toolCallId, message.isError]), + ).toEqual([ + ["a", false], + ["barrier", true], + ["b", false], + ]); + expect(finalized).toContain("b"); + expect(batch.requests).toHaveLength(2); + }, + ); + + it.each([ + { hints: [true, true, true], requests: 1 }, + { hints: [true, false, true], requests: 2 }, + { hints: [false, true, true], requests: 2 }, + { hints: [true, true, false], requests: 2 }, + ])("aggregates termination across every phase: $hints", async ({ hints, requests }) => { + const executed: string[] = []; + const batch = startToolBatch( + [schedulingCall("0"), schedulingCall("1", "barrier"), schedulingCall("2")], + [ + schedulingTool("parallel", async (id) => { + executed.push(id); + return { content: [], details: {}, terminate: hints[Number(id)] }; + }), + schedulingTool( + "barrier", + async () => { + throw new Error("blocked barrier must not execute"); + }, + "sequential", + ), + ], + { + beforeToolCall: async ({ toolCall }) => + toolCall.id === "1" ? { block: true, terminate: hints[1] } : undefined, + }, + ); + await batch.result; + expect(executed).toEqual(["0", "2"]); + expect(batch.requests).toHaveLength(requests); + }); + + it.each(["parallel", "barrier", "preflight", "result hook", "message listener"])( + "stops scheduling later phases when cancelled during %s", + async (when) => { + const controller = new AbortController(); + const entered = gate(); + const release = gate(); + const preflight: string[] = []; + const executed: string[] = []; + const pause = async () => { + entered.resolve(); + await release.promise; + }; + const execute: AgentTool["execute"] = async (id, _args, signal) => { + executed.push(id); + if ((when === "parallel" && id === "a") || (when === "barrier" && id === "barrier")) { + await pause(); + if (signal?.aborted) throw new Error("active call aborted"); + } + return { content: [], details: {} }; + }; + const batch = startToolBatch( + [schedulingCall("a"), schedulingCall("b"), schedulingCall("barrier", "barrier"), schedulingCall("later")], + [schedulingTool("parallel", execute), schedulingTool("barrier", execute, "sequential")], + { + beforeToolCall: async ({ toolCall }) => { + preflight.push(toolCall.id); + if (when === "preflight" && toolCall.id === "a") await pause(); + return undefined; + }, + afterToolCall: async ({ toolCall }) => { + if (when === "result hook" && toolCall.id === "a") await pause(); + return undefined; + }, + shouldStopAfterTurn: () => controller.signal.aborted, + }, + async (event) => { + if ( + when === "message listener" && + event.type === "message_end" && + event.message.role === "toolResult" && + event.message.toolCallId === "b" + ) + await pause(); + }, + controller.signal, + ); + try { + await entered.promise; + controller.abort(); + } finally { + release.resolve(); + await batch.result; + } + expect(preflight).not.toContain("later"); + expect(executed).not.toContain("later"); + if (when !== "barrier") expect(preflight).not.toContain("barrier"); + if (when === "preflight") expect(executed).toEqual([]); + const results = (await batch.result).filter((message) => message.role === "toolResult"); + expect(results.map((message) => message.toolCallId)).toEqual( + when === "preflight" ? ["a"] : when === "barrier" ? ["a", "b", "barrier"] : ["a", "b"], + ); + if (when === "parallel" || when === "barrier" || when === "preflight") + expect(results.some((message) => message.isError)).toBe(true); + expect(batch.requests).toHaveLength(1); + }, + ); +}); + describe("agentLoopContinue with AgentMessage", () => { it("should throw when context has no messages", () => { const context: AgentContext = { diff --git a/packages/coding-agent/docs/extensions.md b/packages/coding-agent/docs/extensions.md index 5486a8473..7f30b7dc9 100644 --- a/packages/coding-agent/docs/extensions.md +++ b/packages/coding-agent/docs/extensions.md @@ -625,12 +625,16 @@ diffex.on("message_end", async (event, ctx) => { Fired for tool execution lifecycle updates. -In parallel tool mode: -- `tool_execution_start` is emitted in assistant source order during the preflight phase +In parallel tool mode, consecutive parallel-capable calls form groups separated by individual sequential tool barriers: + +- `tool_execution_start` is emitted in assistant source order during each phase's preflight - `tool_execution_update` events may interleave across tools - `tool_execution_end` is emitted in tool completion order after each tool is finalized - final `toolResult` message events are still emitted later in assistant source order +Every phase awaits execution, `tool_result` handlers, completion handlers, and ordered result-message handlers before the next phase starts preflight. +Steering delivery and the next model request wait for the entire assistant batch. + ```typescript diffex.on("tool_execution_start", async (event, ctx) => { // event.toolCallId, event.toolName, event.args @@ -754,7 +758,11 @@ Fired after `tool_execution_start`, before the tool executes. **Can block.** Use Before `tool_call` runs, Diffex waits for previously emitted Agent events to finish draining through `AgentSession`. This means `ctx.sessionManager` is up to date through the current assistant tool-calling message. -In the default parallel tool execution mode, sibling tool calls from the same assistant message are preflighted sequentially, then executed concurrently. `tool_call` is not guaranteed to see sibling tool results from that same assistant message in `ctx.sessionManager`. +In the default parallel tool execution mode, calls in each consecutive parallel group are preflighted sequentially, then executed concurrently. +A tool declaring `executionMode: "sequential"` is an exclusive barrier at its position in the assistant message. +For example, `read A, read B, update_plan, read C, read D` runs as `[read A + read B]`, then `update_plan`, then `[read C + read D]`. +`update_plan` and `request_user_input` are sequential tools. +`tool_call` sees settled results from earlier phases in `ctx.sessionManager`, but is not guaranteed to see sibling results within its current parallel group. `event.input` is mutable. Mutate it in place to patch tool arguments before execution. @@ -816,7 +824,8 @@ diffex.on("tool_call", (event) => { Fired after tool execution finishes and before `tool_execution_end` plus the final tool result message events are emitted. **Can modify result.** -In parallel tool mode, `tool_result` and `tool_execution_end` may interleave in tool completion order, while final `toolResult` message events are still emitted later in assistant source order. +Within a parallel group, `tool_result` and `tool_execution_end` may interleave in tool finalization order, while final `toolResult` message events follow assistant source order. +All result handlers and awaited events settle before the next phase's `tool_call` handlers run. `tool_result` handlers chain like middleware: - Handlers run in extension load order @@ -969,7 +978,8 @@ Use this before reading project-local extension configuration that should only b Read-only access to session state. See [Session Format](session-format.md) for the full SessionManager API and entry types. -For `tool_call`, this state is synchronized through the current assistant message before handlers run. In parallel tool execution mode it is still not guaranteed to include sibling tool results from the same assistant message. +For `tool_call`, this state is synchronized through the current assistant message and all completed execution phases before handlers run. +In parallel mode, it is not guaranteed to include sibling results within the current group. ```typescript ctx.sessionManager.getEntries() // All entries @@ -1875,7 +1885,16 @@ export default function (diffex: ExtensionAPI) { ## Custom Tools -Register tools the LLM can call via `diffex.registerTool()`. Tools appear in the system prompt and can have custom rendering. +Register tools the LLM can call via `diffex.registerTool()`. +Tools appear in the system prompt and can have custom rendering. + +Set `executionMode: "sequential"` for a tool that must run alone as an ordered barrier. +Set `executionMode: "parallel"` or omit it to allow concurrency within a consecutive parallel group. +Global `toolExecution: "sequential"` runs all calls one by one regardless of tool metadata. +The model remains responsible for selecting independent calls; the scheduler does not infer file dependencies or shell side effects. +Existing file-mutation queues still coordinate calls within each parallel group. +Errors and blocked barriers allow later phases to run, and termination is evaluated across the entire batch. +Cancellation stops scheduling later phases while active calls retain their cancellation handling. Use `promptSnippet` for a short one-line entry in the `Available tools` section in the default system prompt. If omitted, custom tools are left out of that section. diff --git a/packages/coding-agent/test/branch-summary-extensions.test.ts b/packages/coding-agent/test/branch-summary-extensions.test.ts index c750682f9..756cfa12f 100644 --- a/packages/coding-agent/test/branch-summary-extensions.test.ts +++ b/packages/coding-agent/test/branch-summary-extensions.test.ts @@ -45,7 +45,11 @@ describe("Branch summary extensions", () => { expect(summaryEntry?.type).toBe("branch_summary"); expect(summaryEntry?.fromHook).toBe(true); - expect(summaryEntry?.summary).toBe("Summary provided by extension"); + expect(summaryEntry?.summary).toBe( + "Summary provided by extension\n\n" + + "The plan below is authoritative for the selected conversation branch. Plan references in the exploration summary above are historical.\n" + + 'Current task context (latest plan):\n{\n "plan": []\n}', + ); expect(summaryEntry?.usage).toEqual(usage); const stats = harness.session.getSessionStats(); diff --git a/packages/coding-agent/test/suite/agent-session-tool-scheduling.test.ts b/packages/coding-agent/test/suite/agent-session-tool-scheduling.test.ts new file mode 100644 index 000000000..a0a16dc7e --- /dev/null +++ b/packages/coding-agent/test/suite/agent-session-tool-scheduling.test.ts @@ -0,0 +1,107 @@ +import { setImmediate } from "node:timers/promises"; +import { fauxAssistantMessage, fauxToolCall } from "@diffexai/diffex-ai"; +import { Type } from "typebox"; +import { afterEach, describe, expect, it } from "vitest"; +import { resolveCurrentPlanState } from "../../src/core/current-plan-state.ts"; +import { createHarness, type Harness } from "./harness.ts"; +import { deferred } from "./subagent-harness.ts"; + +describe("AgentSession tool scheduling", () => { + let harness: Harness | undefined; + afterEach(async () => { + await harness?.cleanup(); + }); + + it("overlaps independent calls on both sides of a persisted update_plan barrier", async () => { + const ids = ["read-a", "read-b", "read-c", "read-d"]; + const gates = new Map(ids.map((id) => [id, deferred()])); + const started = new Map(ids.map((id) => [id, deferred()])); + const completed = new Map(ids.map((id) => [id, deferred()])); + const executionOrder: string[] = []; + const preflightOrder: string[] = []; + const plan = [{ id: "inspect", step: "Inspect the files", status: "in_progress" }]; + harness = await createHarness({ + extensionFactories: [ + (pi) => { + pi.registerTool({ + name: "controlled_read", + label: "Controlled read", + description: "Read a fixture through a controlled gate", + parameters: Type.Object({}), + executionMode: "parallel", + async execute(id) { + executionOrder.push(id); + started.get(id)?.resolve(); + await gates.get(id)?.promise; + return { content: [{ type: "text", text: id }], details: {} }; + }, + }); + pi.on("tool_call", async (event, ctx) => { + preflightOrder.push(event.toolCallId); + const priorResults = ctx.sessionManager + .getBranch() + .flatMap((entry) => + entry.type === "message" && entry.message.role === "toolResult" + ? [entry.message.toolCallId] + : [], + ); + if (event.toolCallId === "plan") expect(priorResults).toEqual(["read-a", "read-b"]); + if (event.toolCallId === "read-c" || event.toolCallId === "read-d") { + expect(priorResults).toEqual(["read-a", "read-b", "plan"]); + } + }); + }, + ], + }); + harness.session.subscribe((event) => { + if (event.type === "tool_execution_end") completed.get(event.toolCallId)?.resolve(); + }); + const calls = ids.map((id) => fauxToolCall("controlled_read", {}, { id })); + calls.splice(2, 0, fauxToolCall("update_plan", { plan }, { id: "plan" })); + harness.setResponses([ + fauxAssistantMessage(calls, { stopReason: "toolUse" }), + (context) => { + expect( + context.messages.filter((message) => message.role === "toolResult").map((message) => message.toolCallId), + ).toEqual(["read-a", "read-b", "plan", "read-c", "read-d"]); + return fauxAssistantMessage("done"); + }, + ]); + const prompt = harness.session.prompt("Read two files, publish the plan, then read two more files."); + try { + await started.get("read-a")?.promise; + await setImmediate(); + expect([...executionOrder]).toEqual(["read-a", "read-b"]); + expect([...preflightOrder]).toEqual(["read-a", "read-b"]); + gates.get("read-b")?.resolve(); + await completed.get("read-b")?.promise; + expect(preflightOrder).not.toContain("plan"); + gates.get("read-a")?.resolve(); + await started.get("read-c")?.promise; + await setImmediate(); + expect([...executionOrder]).toEqual(ids); + expect(resolveCurrentPlanState(harness.sessionManager)).toEqual({ status: "active", operation: { plan } }); + gates.get("read-d")?.resolve(); + await completed.get("read-d")?.promise; + gates.get("read-c")?.resolve(); + } finally { + for (const gate of gates.values()) gate.resolve(); + await prompt; + } + expect(harness.eventsOfType("tool_execution_end").map((event) => event.toolCallId)).toEqual([ + "read-b", + "read-a", + "plan", + "read-d", + "read-c", + ]); + expect( + harness.sessionManager + .getBranch() + .flatMap((entry) => + entry.type === "message" && entry.message.role === "toolResult" ? [entry.message.toolCallId] : [], + ), + ).toEqual(["read-a", "read-b", "plan", "read-c", "read-d"]); + expect(harness.faux.state.callCount).toBe(2); + }); +}); diff --git a/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts b/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts index 6ed30b74e..e60970c54 100644 --- a/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts +++ b/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts @@ -66,7 +66,9 @@ describe("issue #7253: manual compaction during an active response", () => { harness.settingsManager.setCompactionEnabled(true); const compactPromise = harness.session.compact(); - const compactExpectation = expect(compactPromise).resolves.toMatchObject({ summary: "manual summary" }); + const compactExpectation = expect(compactPromise).resolves.toMatchObject({ + summary: 'manual summary\n\nCurrent task context (latest plan):\n{\n "plan": []\n}', + }); releaseSecondResponse(); await Promise.all([promptPromise, compactExpectation]);