mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
feat(x): execute sync tools concurrently within a turn batch
Sync tools in one assistant batch now run via Promise.all instead of sequentially — a tool pending on I/O no longer blocks its siblings. Three coordinated changes keep the event-sourced runtime sound: - executeAllowedTools is two-phase: invocation events are appended serially in source order (durable before any side effect, deterministic log prefix), then all sync executions run concurrently, each appending progress/results as they land. Per-call error and cancel semantics are unchanged (moved to executeSyncTool). - append() commits through an internal queue: persist → reduce → stream runs to completion per batch, so file order, in-memory order, and stream order stay identical even while executions overlap. A failed commit rejects only its caller; the chain survives for siblings. - Abort-registry state is scoped per tool call (turnId:toolCallId) via a wrapper, fixing two latent races: createForRun destroying a running sibling's tracked processes, and cleanup tearing down the turn-wide force-kill scope when the first tool finished. Wire ordering is untouched: model requests already reference tool results by the assistant message's source order, pinned by a new test. No concurrency cap and no per-tool serialization by design; tools that share state must tolerate racing (file edits already reject stale writes via their search/replace precondition). Spec §4.5/§10.5 updated. New runtime tests cover overlap (deadlock unless concurrent), progress interleaving, sibling failure isolation, mid-batch cancellation, and crash recovery with multiple open invocations. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
4c4488c3e7
commit
d2b68a4684
5 changed files with 502 additions and 65 deletions
|
|
@ -154,9 +154,18 @@ semantics after ambiguous interruptions.
|
||||||
Tool calls may complete in any order. The next model request must always contain
|
Tool calls may complete in any order. The next model request must always contain
|
||||||
tool results in the original order emitted by the model.
|
tool results in the original order emitted by the model.
|
||||||
|
|
||||||
The initial implementation may execute sync tools sequentially for simplicity.
|
Sync tools in one batch execute concurrently: invocation events are appended
|
||||||
Async tools naturally complete independently. No behavior may rely on physical
|
serially in source order before any execution starts, then all sync executions
|
||||||
completion order.
|
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
|
## 5. Storage design
|
||||||
|
|
||||||
|
|
@ -904,7 +913,12 @@ For one completed assistant response:
|
||||||
2. Determine permission requirements.
|
2. Determine permission requirements.
|
||||||
3. Apply automatic permission decisions when enabled.
|
3. Apply automatic permission decisions when enabled.
|
||||||
4. Advance each tool independently as its permission is resolved.
|
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.
|
6. Expose allowed async tool requests.
|
||||||
7. Suspend when any permissions or async results remain outstanding.
|
7. Suspend when any permissions or async results remain outstanding.
|
||||||
8. Once all calls have terminal results, build the next model request with
|
8. Once all calls have terminal results, build the next model request with
|
||||||
|
|
|
||||||
|
|
@ -20,7 +20,9 @@ class FakeAbortRegistry implements IAbortRegistry {
|
||||||
this.calls.push(`create:${runId}`);
|
this.calls.push(`create:${runId}`);
|
||||||
return new AbortController().signal;
|
return new AbortController().signal;
|
||||||
}
|
}
|
||||||
registerProcess(): void {}
|
registerProcess(runId: string): void {
|
||||||
|
this.calls.push(`register:${runId}`);
|
||||||
|
}
|
||||||
unregisterProcess(): void {}
|
unregisterProcess(): void {}
|
||||||
abort(runId: string): void {
|
abort(runId: string): void {
|
||||||
this.calls.push(`abort:${runId}`);
|
this.calls.push(`abort:${runId}`);
|
||||||
|
|
@ -94,8 +96,30 @@ describe("RealToolRegistry", () => {
|
||||||
expect(calls[0].attachment).toEqual({ type: "builtin", name: "echo" });
|
expect(calls[0].attachment).toEqual({ type: "builtin", name: "echo" });
|
||||||
expect(calls[0].input).toEqual({ text: "hi" });
|
expect(calls[0].input).toEqual({ text: "hi" });
|
||||||
expect(calls[0].ctx).toMatchObject({ runId: "turn-1", toolCallId: "tc-1" });
|
expect(calls[0].ctx).toMatchObject({ runId: "turn-1", toolCallId: "tc-1" });
|
||||||
// Abort registry bracketed per call.
|
// Abort registry bracketed and keyed per tool call (sync tools in a
|
||||||
expect(abortRegistry.calls).toEqual(["create:turn-1", "cleanup:turn-1"]);
|
// 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 () => {
|
it("normalizes undefined results to null and serializes objects", async () => {
|
||||||
|
|
@ -187,9 +211,9 @@ describe("RealToolRegistry", () => {
|
||||||
const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool;
|
const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool;
|
||||||
await tool.execute({}, makeCtx({ signal: controller.signal }));
|
await tool.execute({}, makeCtx({ signal: controller.signal }));
|
||||||
expect(abortRegistry.calls).toEqual([
|
expect(abortRegistry.calls).toEqual([
|
||||||
"create:turn-1",
|
"create:turn-1:tc-1",
|
||||||
"abort:turn-1",
|
"abort:turn-1:tc-1",
|
||||||
"cleanup:turn-1",
|
"cleanup:turn-1:tc-1",
|
||||||
]);
|
]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|
@ -199,7 +223,10 @@ describe("RealToolRegistry", () => {
|
||||||
});
|
});
|
||||||
const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool;
|
const tool = (await registry.resolve(descriptor())) as SyncRuntimeTool;
|
||||||
await expect(tool.execute({}, makeCtx())).rejects.toThrowError("tool exploded");
|
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 () => {
|
it("resolves mcp descriptors into mcp attachments (server:tool split on first colon)", async () => {
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,4 @@
|
||||||
|
import type { ChildProcess } from "child_process";
|
||||||
import type { z } from "zod";
|
import type { z } from "zod";
|
||||||
import type { ToolAttachment } from "@x/shared/dist/agent.js";
|
import type { ToolAttachment } from "@x/shared/dist/agent.js";
|
||||||
import type { JsonValue, ToolDescriptor } from "@x/shared/dist/turns.js";
|
import type { JsonValue, ToolDescriptor } from "@x/shared/dist/turns.js";
|
||||||
|
|
@ -22,6 +23,47 @@ export interface RealToolRegistryDeps {
|
||||||
builtins?: typeof BuiltinTools;
|
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
|
// Bridges persisted tool descriptors to the existing dispatch: builtins via
|
||||||
// the BuiltinTools catalog, MCP tools via execTool's MCP path. toolId encodes
|
// the BuiltinTools catalog, MCP tools via execTool's MCP path. toolId encodes
|
||||||
// the attachment: "builtin:<name>" or "mcp:<server>:<tool>". ask-human is the
|
// the attachment: "builtin:<name>" or "mcp:<server>:<tool>". ask-human is the
|
||||||
|
|
@ -89,9 +131,14 @@ export class RealToolRegistry implements IToolRegistry {
|
||||||
execute: async (input, ctx: ToolExecutionContext) => {
|
execute: async (input, ctx: ToolExecutionContext) => {
|
||||||
// AbortSignal is the primary kill path; the abort registry is
|
// AbortSignal is the primary kill path; the abort registry is
|
||||||
// the secondary force-kill for spawned child processes,
|
// the secondary force-kill for spawned child processes,
|
||||||
// bracketed per call and keyed by turn.
|
// bracketed and keyed per tool call (sync tools in one turn
|
||||||
this.abortRegistry.createForRun(ctx.turnId);
|
// run concurrently).
|
||||||
const onAbort = () => this.abortRegistry.abort(ctx.turnId);
|
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 });
|
ctx.signal.addEventListener("abort", onAbort, { once: true });
|
||||||
try {
|
try {
|
||||||
const value = await this.execToolImpl(
|
const value = await this.execToolImpl(
|
||||||
|
|
@ -101,7 +148,7 @@ export class RealToolRegistry implements IToolRegistry {
|
||||||
runId: ctx.turnId,
|
runId: ctx.turnId,
|
||||||
toolCallId: ctx.toolCallId,
|
toolCallId: ctx.toolCallId,
|
||||||
signal: ctx.signal,
|
signal: ctx.signal,
|
||||||
abortRegistry: this.abortRegistry,
|
abortRegistry,
|
||||||
publish: async (event) => {
|
publish: async (event) => {
|
||||||
if (event.type === "tool-output-stream") {
|
if (event.type === "tool-output-stream") {
|
||||||
await ctx.reportProgress({
|
await ctx.reportProgress({
|
||||||
|
|
@ -147,7 +194,7 @@ export class RealToolRegistry implements IToolRegistry {
|
||||||
};
|
};
|
||||||
} finally {
|
} finally {
|
||||||
ctx.signal.removeEventListener("abort", onAbort);
|
ctx.signal.removeEventListener("abort", onAbort);
|
||||||
this.abortRegistry.cleanup(ctx.turnId);
|
abortRegistry.cleanup(ctx.turnId);
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|
|
||||||
|
|
@ -1757,3 +1757,317 @@ describe("tool progress", () => {
|
||||||
]);
|
]);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("concurrent sync tool execution (10.5)", () => {
|
||||||
|
const slowDescriptor: z.infer<typeof ToolDescriptor> = {
|
||||||
|
toolId: "tool.slow",
|
||||||
|
name: "slow",
|
||||||
|
description: "Slow tool",
|
||||||
|
inputSchema: {},
|
||||||
|
execution: "sync",
|
||||||
|
requiresHuman: false,
|
||||||
|
};
|
||||||
|
const fastDescriptor: z.infer<typeof ToolDescriptor> = {
|
||||||
|
toolId: "tool.fast",
|
||||||
|
name: "fast",
|
||||||
|
description: "Fast tool",
|
||||||
|
inputSchema: {},
|
||||||
|
execution: "sync",
|
||||||
|
requiresHuman: false,
|
||||||
|
};
|
||||||
|
const agent: z.infer<typeof ResolvedAgent> = {
|
||||||
|
...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<void>((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<void>((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<void>((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",
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
|
||||||
|
|
@ -346,8 +346,24 @@ class TurnAdvance {
|
||||||
}
|
}
|
||||||
|
|
||||||
// Durable barrier: persist, re-reduce (the reducer doubles as a runtime
|
// Durable barrier: persist, re-reduce (the reducer doubles as a runtime
|
||||||
// assertion that the appended history is legal), then stream.
|
// assertion that the appended history is legal), then stream. Commits are
|
||||||
private async append(...batch: TEvent[]): Promise<void> {
|
// 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<void> = Promise.resolve();
|
||||||
|
|
||||||
|
private append(...batch: TEvent[]): Promise<void> {
|
||||||
|
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<void> {
|
||||||
await this.turnRepo.append(this.turnId, batch);
|
await this.turnRepo.append(this.turnId, batch);
|
||||||
this.events.push(...batch);
|
this.events.push(...batch);
|
||||||
this.state = reduceTurn(this.events);
|
this.state = reduceTurn(this.events);
|
||||||
|
|
@ -741,8 +757,13 @@ class TurnAdvance {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// §10.5: execute allowed sync tools sequentially and expose allowed async
|
// §10.5: record invocations for allowed tools serially in source order,
|
||||||
// tools, in source order. Tool failures are conversational, not terminal.
|
// 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<void> {
|
private async executeAllowedTools(): Promise<void> {
|
||||||
const executable = this.state.toolCalls.filter(
|
const executable = this.state.toolCalls.filter(
|
||||||
(tc) =>
|
(tc) =>
|
||||||
|
|
@ -751,8 +772,11 @@ class TurnAdvance {
|
||||||
(this.checkerAllowed.has(tc.toolCallId) ||
|
(this.checkerAllowed.has(tc.toolCallId) ||
|
||||||
tc.permission?.resolved?.decision === "allow"),
|
tc.permission?.resolved?.decision === "allow"),
|
||||||
);
|
);
|
||||||
|
const started: Array<{ tc: ToolCallState; tool: SyncRuntimeTool }> = [];
|
||||||
for (const tc of executable) {
|
for (const tc of executable) {
|
||||||
if (this.signal.aborted) {
|
if (this.signal.aborted) {
|
||||||
|
// Invoked-but-unexecuted calls get their cancelled results
|
||||||
|
// from cancel(), same as before this loop ran.
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const tool = this.toolsByName.get(tc.toolName);
|
const tool = this.toolsByName.get(tc.toolName);
|
||||||
|
|
@ -771,52 +795,63 @@ class TurnAdvance {
|
||||||
if (tool.descriptor.execution === "async") {
|
if (tool.descriptor.execution === "async") {
|
||||||
continue; // exposed; the result arrives through advanceTurn
|
continue; // exposed; the result arrives through advanceTurn
|
||||||
}
|
}
|
||||||
const syncTool = tool as SyncRuntimeTool;
|
started.push({ tc, tool: tool as SyncRuntimeTool });
|
||||||
try {
|
}
|
||||||
const result = await syncTool.execute(tc.input, {
|
// Each task settles its own call (result or error), so Promise.all
|
||||||
turnId: this.turnId,
|
// never rejects and one slow tool never blocks its siblings.
|
||||||
toolCallId: tc.toolCallId,
|
await Promise.all(
|
||||||
signal: this.signal,
|
started.map(({ tc, tool }) => this.executeSyncTool(tc, tool)),
|
||||||
reportProgress: async (progress) => {
|
);
|
||||||
await this.append({
|
}
|
||||||
type: "tool_progress",
|
|
||||||
turnId: this.turnId,
|
private async executeSyncTool(
|
||||||
ts: this.now(),
|
tc: ToolCallState,
|
||||||
toolCallId: tc.toolCallId,
|
syncTool: SyncRuntimeTool,
|
||||||
source: "sync",
|
): Promise<void> {
|
||||||
progress,
|
try {
|
||||||
});
|
const result = await syncTool.execute(tc.input, {
|
||||||
},
|
turnId: this.turnId,
|
||||||
});
|
toolCallId: tc.toolCallId,
|
||||||
await this.append({
|
signal: this.signal,
|
||||||
type: "tool_result",
|
reportProgress: async (progress) => {
|
||||||
turnId: this.turnId,
|
await this.append({
|
||||||
ts: this.now(),
|
type: "tool_progress",
|
||||||
toolCallId: tc.toolCallId,
|
turnId: this.turnId,
|
||||||
toolName: tc.toolName,
|
ts: this.now(),
|
||||||
source: "sync",
|
toolCallId: tc.toolCallId,
|
||||||
result: ToolResultData.parse(result),
|
source: "sync",
|
||||||
});
|
progress,
|
||||||
} catch (error) {
|
});
|
||||||
if (this.signal.aborted) {
|
},
|
||||||
await this.append(
|
});
|
||||||
runtimeResultEvent(this.turnId, this.now(), tc, {
|
await this.append({
|
||||||
output: "Tool execution was cancelled.",
|
type: "tool_result",
|
||||||
isError: true,
|
turnId: this.turnId,
|
||||||
}),
|
ts: this.now(),
|
||||||
);
|
toolCallId: tc.toolCallId,
|
||||||
return;
|
toolName: tc.toolName,
|
||||||
}
|
source: "sync",
|
||||||
await this.append({
|
result: ToolResultData.parse(result),
|
||||||
type: "tool_result",
|
});
|
||||||
turnId: this.turnId,
|
} catch (error) {
|
||||||
ts: this.now(),
|
if (this.signal.aborted) {
|
||||||
toolCallId: tc.toolCallId,
|
await this.append(
|
||||||
toolName: tc.toolName,
|
runtimeResultEvent(this.turnId, this.now(), tc, {
|
||||||
source: "sync",
|
output: "Tool execution was cancelled.",
|
||||||
result: { output: errorMessage(error), isError: true },
|
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 },
|
||||||
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue