diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md index 2b0e54fa..1c61a7b8 100644 --- a/apps/x/packages/core/docs/turn-runtime-design.md +++ b/apps/x/packages/core/docs/turn-runtime-design.md @@ -154,9 +154,18 @@ semantics after ambiguous interruptions. Tool calls may complete in any order. The next model request must always contain tool results in the original order emitted by the model. -The initial implementation may execute sync tools sequentially for simplicity. -Async tools naturally complete independently. No behavior may rely on physical -completion order. +Sync tools in one batch execute concurrently: invocation events are appended +serially in source order before any execution starts, then all sync executions +run at once, each appending its progress and result as it lands. Result order +in the log is therefore physical completion order and is not deterministic +across runs; any given log still replays deterministically. Async tools +naturally complete independently. No behavior may rely on physical completion +order. + +Durable appends are serialized through a single internal queue per invocation: +the persist → reduce → stream ritual runs to completion for one batch of +events before the next begins, so file order, in-memory order, and stream +order are identical by construction even while executions overlap. ## 5. Storage design @@ -904,7 +913,12 @@ For one completed assistant response: 2. Determine permission requirements. 3. Apply automatic permission decisions when enabled. 4. Advance each tool independently as its permission is resolved. -5. Execute allowed sync tools using a simple sequential policy initially. +5. Record invocations for allowed tools serially in source order (sync and + async alike), then execute all allowed sync tools concurrently. There is + no concurrency cap and no per-tool serialization; tools that share state + must tolerate racing (or reject stale operations, as file edits do via + their search/replace precondition). Secondary kill-path state (the abort + registry) is scoped per tool call, never per turn. 6. Expose allowed async tool requests. 7. Suspend when any permissions or async results remain outstanding. 8. Once all calls have terminal results, build the next model request with diff --git a/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts b/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts index 4ccb15a6..c92a4428 100644 --- a/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts +++ b/apps/x/packages/core/src/turns/bridges/real-tool-registry.test.ts @@ -20,7 +20,9 @@ class FakeAbortRegistry implements IAbortRegistry { this.calls.push(`create:${runId}`); return new AbortController().signal; } - registerProcess(): void {} + registerProcess(runId: string): void { + this.calls.push(`register:${runId}`); + } unregisterProcess(): void {} abort(runId: string): void { this.calls.push(`abort:${runId}`); @@ -94,8 +96,30 @@ describe("RealToolRegistry", () => { expect(calls[0].attachment).toEqual({ type: "builtin", name: "echo" }); expect(calls[0].input).toEqual({ text: "hi" }); expect(calls[0].ctx).toMatchObject({ runId: "turn-1", toolCallId: "tc-1" }); - // Abort registry bracketed per call. - expect(abortRegistry.calls).toEqual(["create:turn-1", "cleanup:turn-1"]); + // Abort registry bracketed and keyed per tool call (sync tools in a + // turn execute concurrently; a shared turn key would let one call + // tear down its siblings' force-kill scope). + expect(abortRegistry.calls).toEqual([ + "create:turn-1:tc-1", + "cleanup:turn-1:tc-1", + ]); + }); + + it("re-keys registry calls a tool makes with ctx.runId to the call scope", async () => { + // Builtins address the abort registry with ctx.runId (the turn id). + // The scoped wrapper must pin those to the per-call key, or a + // process registered by one tool would land in no scope at all. + const { registry, abortRegistry } = makeRegistry(async ({ ctx }) => { + ctx.abortRegistry.registerProcess(ctx.runId, {} as never); + return "ok"; + }); + const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool; + await tool.execute({}, makeCtx()); + expect(abortRegistry.calls).toEqual([ + "create:turn-1:tc-1", + "register:turn-1:tc-1", + "cleanup:turn-1:tc-1", + ]); }); it("normalizes undefined results to null and serializes objects", async () => { @@ -187,9 +211,9 @@ describe("RealToolRegistry", () => { const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool; await tool.execute({}, makeCtx({ signal: controller.signal })); expect(abortRegistry.calls).toEqual([ - "create:turn-1", - "abort:turn-1", - "cleanup:turn-1", + "create:turn-1:tc-1", + "abort:turn-1:tc-1", + "cleanup:turn-1:tc-1", ]); }); @@ -199,7 +223,10 @@ describe("RealToolRegistry", () => { }); const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool; await expect(tool.execute({}, makeCtx())).rejects.toThrowError("tool exploded"); - expect(abortRegistry.calls).toEqual(["create:turn-1", "cleanup:turn-1"]); + expect(abortRegistry.calls).toEqual([ + "create:turn-1:tc-1", + "cleanup:turn-1:tc-1", + ]); }); it("resolves mcp descriptors into mcp attachments (server:tool split on first colon)", async () => { diff --git a/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts b/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts index e891ccb9..d0048161 100644 --- a/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts +++ b/apps/x/packages/core/src/turns/bridges/real-tool-registry.ts @@ -1,3 +1,4 @@ +import type { ChildProcess } from "child_process"; import type { z } from "zod"; import type { ToolAttachment } from "@x/shared/dist/agent.js"; import type { JsonValue, ToolDescriptor } from "@x/shared/dist/turns.js"; @@ -22,6 +23,47 @@ export interface RealToolRegistryDeps { builtins?: typeof BuiltinTools; } +// Sync tools within a turn execute concurrently, so abort-registry state must +// be scoped per tool call: createForRun destroys any existing state under its +// key, and cleanup would otherwise tear down the force-kill scope of +// still-running siblings. Builtins address the registry with ctx.runId (the +// turn id, which keeps its meaning elsewhere), so this wrapper pins every +// operation to the call-scoped key regardless of the key the caller passes. +class CallScopedAbortRegistry implements IAbortRegistry { + constructor( + private readonly inner: IAbortRegistry, + private readonly key: string, + ) {} + + createForRun(): AbortSignal { + return this.inner.createForRun(this.key); + } + + registerProcess(_runId: string, process: ChildProcess): void { + this.inner.registerProcess(this.key, process); + } + + unregisterProcess(_runId: string, process: ChildProcess): void { + this.inner.unregisterProcess(this.key, process); + } + + abort(): void { + this.inner.abort(this.key); + } + + forceAbort(): void { + this.inner.forceAbort(this.key); + } + + isAborted(): boolean { + return this.inner.isAborted(this.key); + } + + cleanup(): void { + this.inner.cleanup(this.key); + } +} + // Bridges persisted tool descriptors to the existing dispatch: builtins via // the BuiltinTools catalog, MCP tools via execTool's MCP path. toolId encodes // the attachment: "builtin:" or "mcp::". ask-human is the @@ -89,9 +131,14 @@ export class RealToolRegistry implements IToolRegistry { execute: async (input, ctx: ToolExecutionContext) => { // AbortSignal is the primary kill path; the abort registry is // the secondary force-kill for spawned child processes, - // bracketed per call and keyed by turn. - this.abortRegistry.createForRun(ctx.turnId); - const onAbort = () => this.abortRegistry.abort(ctx.turnId); + // bracketed and keyed per tool call (sync tools in one turn + // run concurrently). + const abortRegistry: IAbortRegistry = new CallScopedAbortRegistry( + this.abortRegistry, + `${ctx.turnId}:${ctx.toolCallId}`, + ); + abortRegistry.createForRun(ctx.turnId); + const onAbort = () => abortRegistry.abort(ctx.turnId); ctx.signal.addEventListener("abort", onAbort, { once: true }); try { const value = await this.execToolImpl( @@ -101,7 +148,7 @@ export class RealToolRegistry implements IToolRegistry { runId: ctx.turnId, toolCallId: ctx.toolCallId, signal: ctx.signal, - abortRegistry: this.abortRegistry, + abortRegistry, publish: async (event) => { if (event.type === "tool-output-stream") { await ctx.reportProgress({ @@ -147,7 +194,7 @@ export class RealToolRegistry implements IToolRegistry { }; } finally { ctx.signal.removeEventListener("abort", onAbort); - this.abortRegistry.cleanup(ctx.turnId); + abortRegistry.cleanup(ctx.turnId); } }, }; diff --git a/apps/x/packages/core/src/turns/runtime.test.ts b/apps/x/packages/core/src/turns/runtime.test.ts index 335250e4..83027868 100644 --- a/apps/x/packages/core/src/turns/runtime.test.ts +++ b/apps/x/packages/core/src/turns/runtime.test.ts @@ -1757,3 +1757,317 @@ describe("tool progress", () => { ]); }); }); + +describe("concurrent sync tool execution (10.5)", () => { + const slowDescriptor: z.infer = { + toolId: "tool.slow", + name: "slow", + description: "Slow tool", + inputSchema: {}, + execution: "sync", + requiresHuman: false, + }; + const fastDescriptor: z.infer = { + toolId: "tool.fast", + name: "fast", + description: "Fast tool", + inputSchema: {}, + execution: "sync", + requiresHuman: false, + }; + const agent: z.infer = { + ...defaultAgent, + tools: [slowDescriptor, fastDescriptor], + }; + + it("executes a batch concurrently: invocations in source order, results in completion order", async () => { + // slow (first in source order) only finishes after fast has fully + // completed AND reported progress. Under the old sequential loop this + // deadlocks (fast never starts), so settling at all proves overlap. + const order: string[] = []; + let releaseSlow!: () => void; + const slowGate = new Promise((resolve) => { + releaseSlow = resolve; + }); + const tools: RuntimeTool[] = [ + syncTool(slowDescriptor, async () => { + order.push("slow:start"); + await slowGate; + order.push("slow:end"); + return { output: "slow-done", isError: false }; + }), + syncTool(fastDescriptor, async (_input, ctx) => { + order.push("fast:start"); + await ctx.reportProgress({ note: "while slow is pending" }); + order.push("fast:end"); + releaseSlow(); + return { output: "fast-done", isError: false }; + }), + ]; + const { runtime, repo, models } = makeRuntime({ + agent, + tools, + models: [ + respond( + completedResp( + assistantCalls( + toolCallPart("S", "slow"), + toolCallPart("F", "fast"), + ), + ), + ), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("completed"); + expect(order).toEqual(["slow:start", "fast:start", "fast:end", "slow:end"]); + + // The log stays legal under interleaving: both invocations precede + // any result (source order), fast's progress and result land while + // slow is still open, and the reducer accepts the whole history. + const log = await persisted(repo, turnId); + expect(typesOf(log)).toEqual([ + "turn_created", + "model_call_requested", + "model_call_completed", + "tool_invocation_requested", + "tool_invocation_requested", + "tool_progress", + "tool_result", + "tool_result", + "model_call_requested", + "model_call_completed", + "turn_completed", + ]); + const invocations = log.filter( + (e) => e.type === "tool_invocation_requested", + ); + expect(invocations.map((e) => e.toolCallId)).toEqual(["S", "F"]); + const results = log.filter((e) => e.type === "tool_result"); + expect(results.map((e) => e.toolCallId)).toEqual(["F", "S"]); + const state = reduceTurn(log); + expect(state.toolCalls.map((tc) => tc.result?.result.output)).toEqual([ + "slow-done", + "fast-done", + ]); + + // Wire ordering is insulated from completion order: the follow-up + // request references tool results in the assistant message's source + // order, and the composed payload sends them in that order. + const followUp = log.find( + (e) => e.type === "model_call_requested" && e.modelCallIndex === 1, + ); + expect(followUp).toMatchObject({ + request: { + messages: ["assistant:0", "toolResult:S", "toolResult:F"], + }, + }); + const sent = sentMessages(models.requests[1]); + expect( + sent + .filter((m) => m.role === "tool") + .map((m) => (m as { toolCallId?: string }).toolCallId), + ).toEqual(["S", "F"]); + }); + + it("one tool's failure never disturbs its concurrent siblings", async () => { + let releaseSlow!: () => void; + const slowGate = new Promise((resolve) => { + releaseSlow = resolve; + }); + const tools: RuntimeTool[] = [ + syncTool(slowDescriptor, async () => { + await slowGate; + return { output: "slow-done", isError: false }; + }), + syncTool(fastDescriptor, async () => { + releaseSlow(); + throw new Error("fast exploded"); + }), + ]; + const { runtime, repo } = makeRuntime({ + agent, + tools, + models: [ + respond( + completedResp( + assistantCalls( + toolCallPart("S", "slow"), + toolCallPart("F", "fast"), + ), + ), + ), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, turnId); + const byId = new Map( + log + .filter((e) => e.type === "tool_result") + .map((e) => [e.toolCallId, e]), + ); + expect(byId.get("F")).toMatchObject({ + result: { output: "fast exploded", isError: true }, + }); + expect(byId.get("S")).toMatchObject({ + result: { output: "slow-done", isError: false }, + }); + }); + + it("cancellation mid-batch settles every in-flight tool", async () => { + const controller = new AbortController(); + const started: string[] = []; + function hangingTool(name: string) { + return async ( + _input: unknown, + ctx: ToolExecutionContext, + ): Promise<{ output: string; isError: boolean }> => { + started.push(name); + if (started.length === 2) { + controller.abort(); + } + await new Promise((resolve) => { + if (ctx.signal.aborted) { + resolve(); + } else { + ctx.signal.addEventListener("abort", () => resolve(), { + once: true, + }); + } + }); + throw new Error("aborted"); + }; + } + const tools: RuntimeTool[] = [ + syncTool(slowDescriptor, hangingTool("slow")), + syncTool(fastDescriptor, hangingTool("fast")), + ]; + const { runtime, repo } = makeRuntime({ + agent, + tools, + models: [ + respond( + completedResp( + assistantCalls( + toolCallPart("S", "slow"), + toolCallPart("F", "fast"), + ), + ), + ), + ], + }); + const turnId = await newTurn(runtime); + const { outcome } = await advanceAndSettle(runtime, turnId, undefined, { + signal: controller.signal, + }); + expect(outcome?.status).toBe("cancelled"); + expect(started).toEqual(["slow", "fast"]); + const log = await persisted(repo, turnId); + const results = log.filter((e) => e.type === "tool_result"); + expect(results).toHaveLength(2); + for (const result of results) { + expect(result).toMatchObject({ + result: { + output: "Tool execution was cancelled.", + isError: true, + }, + }); + } + expect(typesOf(log)).toContain("turn_cancelled"); + }); + + it("recovers a crash that left multiple sync invocations open", async () => { + const SEED_ID = "2026-07-02T10-00-00Z-0000001-000"; + const repo = new InMemoryTurnRepo(); + const batch = assistantCalls( + toolCallPart("S", "slow"), + toolCallPart("F", "fast"), + ); + repo.seed([ + { + type: "turn_created", + schemaVersion: 1, + turnId: SEED_ID, + ts: TS, + sessionId: null, + agent: { requested: { agentId: "copilot" }, resolved: agent }, + context: [], + input: user("hello"), + config: { + autoPermission: false, + humanAvailable: true, + maxModelCalls: 20, + }, + }, + { + type: "model_call_requested", + turnId: SEED_ID, + ts: TS, + modelCallIndex: 0, + request: { messages: ["input"], parameters: {} }, + }, + { + type: "model_call_completed", + turnId: SEED_ID, + ts: TS, + modelCallIndex: 0, + message: batch, + finishReason: "stop", + usage: {}, + }, + { + type: "tool_invocation_requested", + turnId: SEED_ID, + ts: TS, + toolCallId: "S", + toolId: "tool.slow", + toolName: "slow", + execution: "sync", + input: {}, + }, + { + type: "tool_invocation_requested", + turnId: SEED_ID, + ts: TS, + toolCallId: "F", + toolId: "tool.fast", + toolName: "fast", + execution: "sync", + input: {}, + }, + ]); + const { runtime } = makeRuntime({ + repo, + agent, + tools: [ + syncTool(slowDescriptor, async () => ({ + output: "never", + isError: false, + })), + syncTool(fastDescriptor, async () => ({ + output: "never", + isError: false, + })), + ], + models: [respond(completedResp(assistantText("done")))], + }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, SEED_ID); + const indeterminate = log.filter( + (e) => + e.type === "tool_result" && + e.source === "runtime" && + e.result.isError === true, + ); + expect(indeterminate.map((e) => (e as { toolCallId: string }).toolCallId).sort()).toEqual([ + "F", + "S", + ]); + }); +}); diff --git a/apps/x/packages/core/src/turns/runtime.ts b/apps/x/packages/core/src/turns/runtime.ts index da2362a4..0fd4337c 100644 --- a/apps/x/packages/core/src/turns/runtime.ts +++ b/apps/x/packages/core/src/turns/runtime.ts @@ -346,8 +346,24 @@ class TurnAdvance { } // Durable barrier: persist, re-reduce (the reducer doubles as a runtime - // assertion that the appended history is legal), then stream. - private async append(...batch: TEvent[]): Promise { + // assertion that the appended history is legal), then stream. Commits are + // serialized through an internal queue so concurrently executing tools + // can never interleave the persist/reduce/stream ritual — file order, + // in-memory order, and stream order stay identical by construction. + private appendChain: Promise = Promise.resolve(); + + private append(...batch: TEvent[]): Promise { + const task = this.appendChain.then(() => this.commit(batch)); + // A failed commit rejects for its caller only; the chain stays alive + // so other in-flight tools can still record their results. + this.appendChain = task.then( + () => undefined, + () => undefined, + ); + return task; + } + + private async commit(batch: TEvent[]): Promise { await this.turnRepo.append(this.turnId, batch); this.events.push(...batch); this.state = reduceTurn(this.events); @@ -741,8 +757,13 @@ class TurnAdvance { } } - // §10.5: execute allowed sync tools sequentially and expose allowed async - // tools, in source order. Tool failures are conversational, not terminal. + // §10.5: record invocations for allowed tools serially in source order, + // then execute the sync ones concurrently (async tools are exposed by + // their invocation; results arrive through advanceTurn). Invocations are + // durable before any execution starts, and commits are serialized by + // append's internal queue, so the log prefix is deterministic while + // results land in completion order. Tool failures are conversational, + // not terminal. private async executeAllowedTools(): Promise { const executable = this.state.toolCalls.filter( (tc) => @@ -751,8 +772,11 @@ class TurnAdvance { (this.checkerAllowed.has(tc.toolCallId) || tc.permission?.resolved?.decision === "allow"), ); + const started: Array<{ tc: ToolCallState; tool: SyncRuntimeTool }> = []; for (const tc of executable) { if (this.signal.aborted) { + // Invoked-but-unexecuted calls get their cancelled results + // from cancel(), same as before this loop ran. return; } const tool = this.toolsByName.get(tc.toolName); @@ -771,52 +795,63 @@ class TurnAdvance { if (tool.descriptor.execution === "async") { continue; // exposed; the result arrives through advanceTurn } - const syncTool = tool as SyncRuntimeTool; - try { - const result = await syncTool.execute(tc.input, { - turnId: this.turnId, - toolCallId: tc.toolCallId, - signal: this.signal, - reportProgress: async (progress) => { - await this.append({ - type: "tool_progress", - turnId: this.turnId, - ts: this.now(), - toolCallId: tc.toolCallId, - source: "sync", - progress, - }); - }, - }); - await this.append({ - type: "tool_result", - turnId: this.turnId, - ts: this.now(), - toolCallId: tc.toolCallId, - toolName: tc.toolName, - source: "sync", - result: ToolResultData.parse(result), - }); - } catch (error) { - if (this.signal.aborted) { - await this.append( - runtimeResultEvent(this.turnId, this.now(), tc, { - output: "Tool execution was cancelled.", - isError: true, - }), - ); - return; - } - await this.append({ - type: "tool_result", - turnId: this.turnId, - ts: this.now(), - toolCallId: tc.toolCallId, - toolName: tc.toolName, - source: "sync", - result: { output: errorMessage(error), isError: true }, - }); + started.push({ tc, tool: tool as SyncRuntimeTool }); + } + // Each task settles its own call (result or error), so Promise.all + // never rejects and one slow tool never blocks its siblings. + await Promise.all( + started.map(({ tc, tool }) => this.executeSyncTool(tc, tool)), + ); + } + + private async executeSyncTool( + tc: ToolCallState, + syncTool: SyncRuntimeTool, + ): Promise { + try { + const result = await syncTool.execute(tc.input, { + turnId: this.turnId, + toolCallId: tc.toolCallId, + signal: this.signal, + reportProgress: async (progress) => { + await this.append({ + type: "tool_progress", + turnId: this.turnId, + ts: this.now(), + toolCallId: tc.toolCallId, + source: "sync", + progress, + }); + }, + }); + await this.append({ + type: "tool_result", + turnId: this.turnId, + ts: this.now(), + toolCallId: tc.toolCallId, + toolName: tc.toolName, + source: "sync", + result: ToolResultData.parse(result), + }); + } catch (error) { + if (this.signal.aborted) { + await this.append( + runtimeResultEvent(this.turnId, this.now(), tc, { + output: "Tool execution was cancelled.", + isError: true, + }), + ); + return; } + await this.append({ + type: "tool_result", + turnId: this.turnId, + ts: this.now(), + toolCallId: tc.toolCallId, + toolName: tc.toolName, + source: "sync", + result: { output: errorMessage(error), isError: true }, + }); } }