fix(agent): apply steering before unstarted tools (#120470)

* fix(agent): apply steering before unstarted tools

Restore steering checkpoints before sequential tool launches and before parallel batch launch. Preserve paired synthetic tool results, async callback compatibility, and Code Mode outcome handling.

* fix(agent): delay tool loop admission commits

Commit loop-detection history only for calls crossing the final launch checkpoint. Release steering-skipped markers, add repeated-steer coverage, and align remaining steering contract text.

* fix(agent): keep tool admission lifecycle internal

Attach delayed admission callbacks through the private internal-hooks seam so steering history remains correct without widening the public Agent Core or Plugin SDK contract.

* fix(agent): preserve steering API contracts

Keep public steering callbacks Promise-based and protocol error kinds unchanged. Use private synchronous draining and structured skip details to retain launch-boundary behavior without API or generated protocol drift.

* test(gateway): use canonical steering fixture config

Use keyed agent entries in the real gateway steering harness so current main does not migrate the fixture during startup.

* fix(agent): remove unused lifecycle re-export

* fix(agent): gate tool launch after wrapper preflight

Split OpenClaw tool execution into private prepare and launch phases so steering is checked after policy, approval, validation, and reconciliation but before the original side effect. Preserve final arguments, voice grants, loop admission, context wrappers, and direct tool execution.

* fix(agent): preserve steering callback receiver

Invoke public steering callbacks with their AgentLoopConfig receiver and cover method-style implementations that read config-owned queue state.
This commit is contained in:
Peter Steinberger
2026-08-08 09:52:10 -07:00
committed by GitHub
parent 6f06fb2949
commit 7fd723b515
30 changed files with 2981 additions and 426 deletions
+4 -3
View File
@@ -106,9 +106,10 @@ OpenClaw. OpenClaw does not read session folders from other tools.
## Steering while streaming
Inbound prompts that arrive mid-run are steered into the current run by default.
Steering is delivered **after the current assistant turn finishes executing its
tool calls**, before the next LLM call, and no longer skips remaining tool calls
from the current assistant message.
The OpenClaw runtime checks for steering before unstarted tool launches and the
next model call. A running tool continues; unstarted sequential calls are skipped,
while parallel calls continue after their batch crosses its launch checkpoint.
Skipped calls receive synthetic paired results before the model sees the steer.
`/queue steer` is the default active-run behavior. `/queue followup` and
`/queue collect` make messages wait for a later turn instead of steering.
+18 -16
View File
@@ -2,7 +2,7 @@
summary: "How active-run steering queues messages at runtime boundaries"
read_when:
- Explaining how steer behaves while an agent is using tools
- Explaining why steering does not cancel an in-flight tool-call batch
- Explaining why steering does not cancel an already-running tool
- Changing active-run queue behavior or runtime steering integration
- Comparing steering with followup, collect, and interrupt queue modes
title: "Steering queue"
@@ -14,29 +14,31 @@ This page covers queue-mode steering for normal inbound messages in `steer` mode
## Runtime boundary
Steering does not interrupt a tool call that is already running. OpenClaw checks for queued steering messages at model boundaries:
Steering does not interrupt a tool call that is already running. The OpenClaw runtime checks at tool-launch boundaries as well as model boundaries:
1. The assistant asks for tool calls.
2. OpenClaw executes the current assistant message's tool-call batch.
3. OpenClaw emits the turn end event.
4. OpenClaw drains queued steering messages.
5. OpenClaw appends those messages as user messages before the next LLM call.
2. In sequential mode, OpenClaw checks immediately before each call starts, including after asynchronous resolution, validation, and pre-execution hooks.
3. A running call finishes. If a steer is waiting afterward, the unstarted sequential tail is skipped.
4. In parallel mode, OpenClaw prepares calls first, then checks once immediately before launching the prepared calls. Calls that have crossed that checkpoint continue together.
5. Every skipped call receives paired tool start/end events and a synthetic error result (`Skipped due to queued user message.`), in assistant source order.
6. OpenClaw appends the exact drained steering message before the next LLM call.
This keeps tool results paired with the assistant message that requested them, then lets the next model call see the latest user input.
This keeps every requested tool call paired with a result while ensuring accepted steering is model-visible before any later tool can start.
The native Codex app-server harness exposes `turn/steer` instead of OpenClaw runtime's internal steering queue. OpenClaw batches queued prompts for the configured quiet window, then sends a single `turn/steer` request with all collected user input in arrival order.
The native Codex app-server harness exposes `turn/steer` instead of OpenClaw runtime's internal steering queue. OpenClaw batches queued prompts for the configured quiet window, then sends a single `turn/steer` request with all collected user input in arrival order. Codex's upstream turn scheduler owns its tool scheduling and consumes accepted steering at the next model boundary; OpenClaw does not add per-tool preemption to that runtime.
Codex review and manual compaction turns reject same-turn steering. When a runtime cannot accept steering in `steer` mode, OpenClaw waits for the active run to finish before starting the prompt.
## Why steering waits for the current batch
## Tool launch boundaries
Steering applies corrections at the next model step instead of cancelling tool calls the assistant already requested. This is a deliberate design decision, not a missing feature:
OpenClaw distinguishes started work from requested work:
- A tool-call batch is one unit of work. When the model requests several tool calls in one assistant message, they usually depend on each other, for example edits across multiple files. Cancelling the not-yet-started calls leaves that work half applied, and the next model step typically has to redo the whole batch to get back to a consistent state.
- Every tool call keeps a real result. Dropping requested calls means fabricating aborted results for them, and models routinely misread synthetic failures as real ones, then retry or route around tools that never actually failed.
- The context stays append-only. Steered messages are appended at the tail, so nothing already sent to the model is rewritten and provider prompt caches stay valid.
- A sequential call that is already running completes. Later calls have not started, so OpenClaw returns synthetic skipped results for them and lets the model reconsider with the steer visible.
- A parallel batch has one atomic launch checkpoint. A steer present before it suppresses all prepared calls; a steer arriving after it does not recall any of them.
- Validation or policy outcomes finalized before the parallel checkpoint remain truthful. Only executable calls that did not start receive the steering skip result.
- The transcript stays append-only and structurally paired: assistant tool calls, real or synthetic tool results, then the steering user message.
The wait is bounded by the current tool-call batch, not by the run: a steered correction is visible to the model at its next reasoning step. Stopping the current work is a different intent than redirecting it; use `/queue interrupt` (or `/stop`) when the newest message should abort the active run instead of steering it.
Stopping already-running work is a different intent from redirecting future work. Use `/queue interrupt` (or `/stop`) when the newest message should abort the active run instead of steering it.
## Modes
@@ -51,7 +53,7 @@ The wait is bounded by the current tool-call batch, not by the run: a steered co
If four users send messages while the agent is executing a tool call:
- With default behavior, the active runtime receives all four messages in arrival order before its next model decision. OpenClaw drains them at the next model boundary; Codex receives them as one batched `turn/steer`.
- OpenClaw preserves the runtime's configured steering drain mode and FIFO order. One-at-a-time consumers keep later messages for later boundaries; `all` consumers inject the queued FIFO batch together. Codex receives messages collected during its quiet window as one batched `turn/steer`.
- With `/queue collect`, OpenClaw does not steer. It waits until the active run ends, then creates a followup turn with compatible queued messages after the debounce window.
- With `/queue interrupt`, OpenClaw aborts the active run and starts the newest message instead of steering.
@@ -63,7 +65,7 @@ Use `followup` or `collect` when you want messages to queue by default instead o
## Debounce
The built-in queue debounce applies to queued `followup` and `collect` delivery. In `steer` mode with the native Codex harness, it also sets the quiet window before sending batched `turn/steer`. For OpenClaw, active steering itself does not use the debounce timer because OpenClaw naturally batches messages until the next model boundary.
The built-in queue debounce applies to queued `followup` and `collect` delivery. In `steer` mode with the native Codex harness, it also sets the quiet window before sending batched `turn/steer`. OpenClaw active steering does not use the debounce timer; at tool-launch and model boundaries it drains FIFO according to the runtime's configured steering drain mode.
## Related
+2 -2
View File
@@ -36,7 +36,7 @@ Same-turn steering is the default. A prompt that arrives mid-run is injected int
`/queue` controls what normal inbound messages do while a session already has an active run:
- `steer`: inject messages into the active runtime. OpenClaw delivers all pending steering messages **after the current assistant turn finishes executing its tool calls**, before the next LLM call; Codex app-server receives one batched `turn/steer`. If the run is not actively streaming or steering is unavailable, OpenClaw waits until the active run ends before starting the prompt.
- `steer`: inject messages into the active runtime. OpenClaw lets an already-running tool finish, skips sequential calls that have not started, and makes the steer visible before the next tool launch or model decision. Parallel calls continue once their batch has crossed its launch checkpoint. Codex app-server receives one batched `turn/steer` and applies it at the next model boundary. If the run is not actively streaming or steering is unavailable, OpenClaw waits until the active run ends before starting the prompt.
- `followup`: do not steer. Enqueue each message for a later agent turn after the current run ends.
- `collect`: do not steer. Coalesce queued messages into a **single** followup turn after the quiet window. If messages target different channels/threads, they drain individually to preserve routing.
- `interrupt`: abort the active run for that session, then run the newest message.
@@ -79,7 +79,7 @@ When channel streaming is `partial` or `block`, steering can look like several s
- `block`: draft-sized blocks can create the same sequential appearance.
- Without streaming, steering falls back to a followup after the active run when the runtime cannot accept same-turn steering.
`steer` does not abort in-flight tools. Use `/queue interrupt` when the newest message should abort the current run.
`steer` does not abort in-flight tools. Skipped OpenClaw tool calls receive synthetic paired error results so the transcript remains valid. Use `/queue interrupt` when the newest message should abort the current run.
## Precedence
+805 -1
View File
@@ -5,7 +5,13 @@ import { describe, expect, it, vi } from "vitest";
import { agentLoop, agentLoopContinue, runAgentLoop, runAgentLoopContinue } from "./agent-loop.js";
import { Agent } from "./agent.js";
import { TRANSCRIPT_NOT_CONTINUABLE_ERROR_CODE, TranscriptNotContinuableError } from "./errors.js";
import { setInternalBeforeToolBatch } from "./internal-hooks.js";
import {
attachInternalSyncSteeringGetter,
attachInternalToolBatchLifecycle,
attachInternalToolExecutionPreparer,
setInternalBeforeToolBatch,
takeInternalToolBatchLifecycle,
} from "./internal-hooks.js";
import {
type AssistantMessage,
createAssistantMessageEventStream,
@@ -18,6 +24,7 @@ import {
type AgentToolExecutionContext,
} from "./tool-execution-context.js";
import type {
AfterToolOutcomeContext,
AgentContext,
AgentEvent,
AgentLoopConfig,
@@ -76,6 +83,21 @@ function expectTerminalFailure(events: AgentEvent[], result: AgentMessage[]): vo
});
}
describe("internal tool batch lifecycle", () => {
it("binds lifecycle state to one exact admission result and consumes it once", () => {
const result = {};
const lifecycle = {
commitReadyCalls: vi.fn(),
releaseSkippedCalls: vi.fn(),
};
expect(attachInternalToolBatchLifecycle(result, lifecycle)).toBe(result);
expect(takeInternalToolBatchLifecycle(result)).toBe(lifecycle);
expect(takeInternalToolBatchLifecycle(result)).toBeUndefined();
expect(takeInternalToolBatchLifecycle({})).toBeUndefined();
});
});
describe("agentLoop EventStream failures", () => {
it("ends the public stream when a new prompt run rejects", async () => {
const stream = agentLoop(
@@ -1003,6 +1025,782 @@ describe("agentLoop tool termination", () => {
};
}
function createDeferred() {
let resolve!: () => void;
const promise = new Promise<void>((done) => {
resolve = done;
});
return { promise, resolve };
}
function createTurnSequenceStream(
turns: AssistantMessage["content"][],
requestMessages: Message[][],
): StreamFn {
let turnIndex = 0;
return (_activeModel, context) => {
requestMessages.push(context.messages.slice());
const content = turns[turnIndex];
turnIndex += 1;
if (!content) {
throw new Error(`unexpected provider request ${turnIndex}`);
}
const stream = createAssistantMessageEventStream();
queueMicrotask(() => {
const message = makeAssistantMessage(content);
stream.push({
type: "done",
reason: message.stopReason === "toolUse" ? "toolUse" : "stop",
message,
});
stream.end();
});
return stream;
};
}
it("makes a queued steer visible before the next sequential tool starts", async () => {
const firstReleased = createDeferred();
const firstStarted = createDeferred();
const firstExecute = vi.fn(async () => {
firstStarted.resolve();
await firstReleased.promise;
return { content: [{ type: "text" as const, text: "first result" }], details: {} };
});
const secondExecute = vi.fn(async () => ({
content: [{ type: "text" as const, text: "second result" }],
details: {},
}));
const requestMessages: Message[][] = [];
const streamFn = createTurnSequenceStream(
[
[
{ type: "toolCall", id: "call-first", name: "first", arguments: {} },
{ type: "toolCall", id: "call-second", name: "second", arguments: {} },
],
[{ type: "text", text: "handled steer 1" }],
[{ type: "text", text: "handled steer 2" }],
],
requestMessages,
);
const afterToolOutcome = vi.fn(async (_context: AfterToolOutcomeContext) => undefined);
const commitReadyCalls = vi.fn();
const releaseSkippedCalls = vi.fn();
const events: AgentEvent[] = [];
const agent = new Agent({
initialState: {
model,
tools: [
{
...makeTool("first", []),
execute: firstExecute,
},
{
...makeTool("second", []),
execute: secondExecute,
resultContentSource: "network",
},
],
},
streamFn,
toolExecution: "sequential",
afterToolOutcome,
});
setInternalBeforeToolBatch(agent, async () =>
attachInternalToolBatchLifecycle({}, { commitReadyCalls, releaseSkippedCalls }),
);
agent.subscribe((event) => {
events.push(event);
});
const firstSteer = { role: "user" as const, content: "steer one", timestamp: 2 };
const secondSteer = { role: "user" as const, content: "steer two", timestamp: 3 };
const run = agent.prompt("start");
await firstStarted.promise;
agent.steer(firstSteer);
agent.steer(secondSteer);
firstReleased.resolve();
await run;
expect(firstExecute).toHaveBeenCalledOnce();
expect(secondExecute).not.toHaveBeenCalled();
expect(commitReadyCalls).toHaveBeenCalledExactlyOnceWith([
{ toolCallId: "call-first", args: {} },
]);
expect(releaseSkippedCalls).toHaveBeenCalledWith(["call-second"]);
expect(requestMessages).toHaveLength(3);
expect(agent.state.messages.slice(1, 5)).toMatchObject([
{ role: "assistant", stopReason: "toolUse" },
{ role: "toolResult", toolCallId: "call-first", isError: false },
{ role: "toolResult", toolCallId: "call-second", isError: true },
firstSteer,
]);
expect(requestMessages[1]?.slice(-4)).toMatchObject([
{ role: "assistant", stopReason: "toolUse" },
{ role: "toolResult", toolCallId: "call-first", isError: false },
{
role: "toolResult",
toolCallId: "call-second",
isError: true,
content: [{ type: "text", text: "Skipped due to queued user message." }],
details: { status: "skipped", deniedReason: "steering" },
},
firstSteer,
]);
expect(requestMessages[1]?.at(-1)).toBe(firstSteer);
expect(requestMessages[1]).not.toContain(secondSteer);
expect(requestMessages[2]?.at(-1)).toBe(secondSteer);
expect(
requestMessages[1]?.find((message) => message.role === "toolResult" && message.isError),
).not.toHaveProperty("__openclaw");
expect(afterToolOutcome).toHaveBeenCalledWith(
expect.objectContaining({
toolCall: expect.objectContaining({ id: "call-second" }),
isError: true,
executionStarted: false,
result: expect.objectContaining({
details: { status: "skipped", deniedReason: "steering" },
}),
}),
expect.any(AbortSignal),
);
const skippedOutcome = afterToolOutcome.mock.calls.find(
([outcome]) => outcome.toolCall.id === "call-second",
)?.[0];
expect(skippedOutcome).not.toHaveProperty("errorKind");
expect(
events
.filter((event) => event.type === "tool_execution_start")
.map((event) => event.toolCallId),
).toEqual(["call-first", "call-second"]);
expect(
events
.filter((event) => event.type === "tool_execution_end")
.map((event) => ({ id: event.toolCallId, started: event.executionStarted })),
).toEqual([
{ id: "call-first", started: true },
{ id: "call-second", started: false },
]);
const skippedEnd = events.find(
(event) => event.type === "tool_execution_end" && event.toolCallId === "call-second",
);
expect(skippedEnd).toMatchObject({
result: { details: { status: "skipped", deniedReason: "steering" } },
});
expect(skippedEnd).not.toHaveProperty("errorKind");
});
it("uses a private synchronous steer at the scheduler without invoking the public fallback", async () => {
const steer = { role: "user" as const, content: "redirect", timestamp: 2 };
let steerReady = false;
let steerDrained = false;
const firstExecute = vi.fn(async () => {
steerReady = true;
return { content: [], details: {} };
});
const secondExecute = vi.fn(async () => ({ content: [], details: {} }));
const publicGetter = vi.fn(async (): Promise<AgentMessage[]> => {
throw new Error("public steering fallback should not run");
});
const syncGetter = vi.fn((): AgentMessage[] => {
if (!steerReady || steerDrained) {
return [];
}
steerDrained = true;
return [steer];
});
const getSteeringMessages = attachInternalSyncSteeringGetter(publicGetter, syncGetter);
const requestMessages: Message[][] = [];
await runAgentLoop(
[{ role: "user", content: "start", timestamp: 1 }],
{
systemPrompt: "",
messages: [],
tools: [
{ ...makeTool("first", []), execute: firstExecute },
{ ...makeTool("second", []), execute: secondExecute },
],
},
{ ...config, getSteeringMessages, toolExecution: "sequential" },
() => {},
undefined,
createTurnSequenceStream(
[
[
{ type: "toolCall", id: "sync-first", name: "first", arguments: {} },
{ type: "toolCall", id: "sync-second", name: "second", arguments: {} },
],
[{ type: "text", text: "done" }],
],
requestMessages,
),
);
expect(firstExecute).toHaveBeenCalledOnce();
expect(secondExecute).not.toHaveBeenCalled();
expect(requestMessages[1]?.at(-1)).toBe(steer);
expect(syncGetter).toHaveBeenCalled();
expect(publicGetter).not.toHaveBeenCalled();
});
it("preserves the config receiver for public steering callbacks", async () => {
const steer = { role: "user" as const, content: "method steer", timestamp: 2 };
const requestMessages: Message[][] = [];
const methodConfig = {
...config,
queuedSteering: [steer] as AgentMessage[],
async getSteeringMessages() {
return this.queuedSteering.splice(0, 1);
},
} satisfies AgentLoopConfig & { queuedSteering: AgentMessage[] };
await runAgentLoop(
[{ role: "user", content: "start", timestamp: 1 }],
{ systemPrompt: "", messages: [] },
methodConfig,
() => {},
undefined,
createTurnSequenceStream([[{ type: "text", text: "done" }]], requestMessages),
);
expect(requestMessages[0]?.at(-1)).toBe(steer);
expect(methodConfig.queuedSteering).toEqual([]);
});
it("suppresses a tool when steering arrives during private execution preflight", async () => {
const preflightStarted = createDeferred();
const releasePreflight = createDeferred();
const execute = vi.fn(async () => ({ content: [], details: { executed: true } }));
const dispose = vi.fn();
const tool = attachInternalToolExecutionPreparer(
{ ...makeTool("delayed", []), execute },
async () => {
preflightStarted.resolve();
await releasePreflight.promise;
const finalArgs = { rewritten: true };
return {
kind: "ready",
args: finalArgs,
execute: async (onImplementationStart) => {
onImplementationStart?.();
return await execute();
},
dispose,
};
},
);
const requestMessages: Message[][] = [];
const afterToolOutcome = vi.fn(async () => undefined);
const commitReadyCalls = vi.fn();
const releaseSkippedCalls = vi.fn();
const agent = new Agent({
initialState: { model, tools: [tool] },
streamFn: createTurnSequenceStream(
[
[{ type: "toolCall", id: "delayed-call", name: "delayed", arguments: {} }],
[{ type: "text", text: "redirected" }],
],
requestMessages,
),
toolExecution: "sequential",
afterToolOutcome,
});
setInternalBeforeToolBatch(agent, async () =>
attachInternalToolBatchLifecycle({}, { commitReadyCalls, releaseSkippedCalls }),
);
const steer = { role: "user" as const, content: "redirect", timestamp: 2 };
const run = agent.prompt("start");
await preflightStarted.promise;
agent.steer(steer);
releasePreflight.resolve();
await run;
expect(execute).not.toHaveBeenCalled();
expect(commitReadyCalls).not.toHaveBeenCalled();
expect(dispose).toHaveBeenCalledOnce();
expect(requestMessages[1]?.slice(-3)).toMatchObject([
{ role: "assistant", stopReason: "toolUse" },
{
role: "toolResult",
toolCallId: "delayed-call",
isError: true,
details: { status: "skipped", deniedReason: "steering" },
},
steer,
]);
expect(afterToolOutcome).toHaveBeenCalledWith(
expect.objectContaining({
toolCall: expect.objectContaining({ id: "delayed-call" }),
args: { rewritten: true },
executionStarted: false,
}),
expect.any(AbortSignal),
);
});
it.each(["sequential", "parallel"] as const)(
"uses private final args for %s launch facts and hooks",
async (toolExecution) => {
const finalArgs = { rewritten: true };
const execute = vi.fn(async () => ({ content: [], details: { executed: true } }));
const tool = attachInternalToolExecutionPreparer(
{ ...makeTool("rewritten", []), execute },
async () => ({
kind: "ready",
args: finalArgs,
execute: async (start) => {
start?.();
return await execute();
},
dispose: vi.fn(),
}),
);
const afterToolCall = vi.fn(async () => undefined);
const afterToolOutcome = vi.fn(async () => undefined);
const commitReadyCalls = vi.fn();
const agent = new Agent({
initialState: { model, tools: [tool] },
streamFn: createTurnSequenceStream(
[
[{ type: "toolCall", id: "rewritten-call", name: "rewritten", arguments: {} }],
[{ type: "text", text: "done" }],
],
[],
),
toolExecution,
afterToolCall,
afterToolOutcome,
});
setInternalBeforeToolBatch(agent, async () =>
attachInternalToolBatchLifecycle(
{},
{
commitReadyCalls,
releaseSkippedCalls: vi.fn(),
},
),
);
await agent.prompt("start");
expect(commitReadyCalls).toHaveBeenCalledExactlyOnceWith([
{ toolCallId: "rewritten-call", args: finalArgs },
]);
expect(afterToolCall).toHaveBeenCalledWith(
expect.objectContaining({ args: finalArgs }),
expect.any(AbortSignal),
);
expect(afterToolOutcome).toHaveBeenCalledWith(
expect.objectContaining({ args: finalArgs, executionStarted: true }),
expect.any(AbortSignal),
);
expect(execute).toHaveBeenCalledOnce();
expect(
agent.state.messages.find(
(message) => message.role === "assistant" && message.stopReason === "toolUse",
),
).toMatchObject({
content: [expect.objectContaining({ id: "rewritten-call", arguments: {} })],
});
},
);
it("disposes private preflight when the steering checkpoint throws", async () => {
const execute = vi.fn(async () => ({ content: [], details: {} }));
const dispose = vi.fn();
const tool = attachInternalToolExecutionPreparer(
{ ...makeTool("cleanup", []), execute },
async ({ args }) => ({
kind: "ready",
args,
execute: async (onImplementationStart) => {
onImplementationStart?.();
return await execute();
},
dispose,
}),
);
const getSteeringMessages = vi
.fn<() => Promise<AgentMessage[]>>()
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockRejectedValueOnce(new Error("steering checkpoint failed"));
await expect(
runAgentLoop(
[{ role: "user", content: "start", timestamp: 1 }],
{ systemPrompt: "", messages: [], tools: [tool] },
{ ...config, toolExecution: "sequential", getSteeringMessages },
() => {},
undefined,
createTurnSequenceStream(
[[{ type: "toolCall", id: "cleanup-call", name: "cleanup", arguments: {} }]],
[],
),
),
).rejects.toThrow("steering checkpoint failed");
expect(execute).not.toHaveBeenCalled();
expect(dispose).toHaveBeenCalledOnce();
});
it("delivers async steering between tools before shouldStopAfterTurn", async () => {
const steer = { role: "user" as const, content: "keep going", timestamp: 2 };
const queued: AgentMessage[] = [];
const secondExecute = vi.fn(async () => ({ content: [], details: {} }));
const requestMessages: Message[][] = [];
const streamFn = createTurnSequenceStream(
[
[
{ type: "toolCall", id: "stop-first", name: "first", arguments: {} },
{ type: "toolCall", id: "stop-second", name: "second", arguments: {} },
],
[{ type: "text", text: "continued" }],
],
requestMessages,
);
const shouldStopAfterTurn = vi.fn(() => true);
const getSteeringMessages = vi.fn(async () => queued.splice(0, 1));
await runAgentLoop(
[{ role: "user", content: "start", timestamp: 1 }],
{
systemPrompt: "",
messages: [],
tools: [
{
...makeTool("first", []),
execute: async () => {
queued.push(steer);
return { content: [{ type: "text", text: "first result" }], details: {} };
},
},
{ ...makeTool("second", []), execute: secondExecute },
],
},
{
...config,
toolExecution: "sequential",
getSteeringMessages,
shouldStopAfterTurn,
},
() => {},
undefined,
streamFn,
);
expect(requestMessages).toHaveLength(2);
expect(requestMessages[1]?.at(-1)).toBe(steer);
expect(secondExecute).not.toHaveBeenCalled();
expect(shouldStopAfterTurn).toHaveBeenCalledOnce();
expect(getSteeringMessages).toHaveBeenCalled();
});
it("suppresses sequential tools when steering arrives from awaited message_end", async () => {
const execute = vi.fn(async () => ({ content: [], details: {} }));
const requestMessages: Message[][] = [];
const streamFn = createTurnSequenceStream(
[
[
{ type: "toolCall", id: "before-first", name: "first", arguments: {} },
{ type: "toolCall", id: "before-second", name: "second", arguments: {} },
],
[{ type: "text", text: "steer handled" }],
],
requestMessages,
);
const agent = new Agent({
initialState: {
model,
tools: [
{ ...makeTool("first", []), execute },
{ ...makeTool("second", []), execute },
],
},
streamFn,
toolExecution: "sequential",
});
const commitReadyCalls = vi.fn();
const releaseSkippedCalls = vi.fn();
setInternalBeforeToolBatch(agent, async () =>
attachInternalToolBatchLifecycle({}, { commitReadyCalls, releaseSkippedCalls }),
);
const events: AgentEvent[] = [];
const steer = { role: "user" as const, content: "before tools", timestamp: 2 };
agent.subscribe(async (event) => {
events.push(event);
if (event.type === "message_end" && event.message.role === "assistant") {
if (event.message.stopReason === "toolUse") {
await Promise.resolve();
agent.steer(steer);
}
}
});
await agent.prompt("start");
expect(execute).not.toHaveBeenCalled();
expect(commitReadyCalls).not.toHaveBeenCalled();
expect(releaseSkippedCalls).toHaveBeenCalledExactlyOnceWith(["before-first", "before-second"]);
expect(requestMessages[1]?.slice(-4)).toMatchObject([
{ role: "assistant", stopReason: "toolUse" },
{ role: "toolResult", toolCallId: "before-first", isError: true },
{ role: "toolResult", toolCallId: "before-second", isError: true },
steer,
]);
expect(requestMessages[1]?.at(-1)).toBe(steer);
expect(
events
.filter((event) => event.type === "tool_execution_end")
.map((event) => ({ id: event.toolCallId, started: event.executionStarted })),
).toEqual([
{ id: "before-first", started: false },
{ id: "before-second", started: false },
]);
});
it("releases only admitted sequential calls when steering suppresses a mixed tail", async () => {
const execute = vi.fn(async () => ({ content: [], details: {} }));
const requestMessages: Message[][] = [];
const streamFn = createTurnSequenceStream(
[
[
{ type: "toolCall", id: "invalid-tail", name: "required", arguments: {} },
{ type: "toolCall", id: "valid-tail", name: "valid", arguments: {} },
],
[{ type: "text", text: "steer handled" }],
],
requestMessages,
);
const agent = new Agent({
initialState: {
model,
tools: [
{
name: "required",
label: "required",
description: "requires input",
parameters: Type.Object({ value: Type.String() }),
execute,
},
{ ...makeTool("valid", []), execute },
],
},
streamFn,
toolExecution: "sequential",
});
const commitReadyCalls = vi.fn();
const releaseSkippedCalls = vi.fn();
setInternalBeforeToolBatch(agent, async ({ calls }) => {
expect(calls.map((call) => call.toolCall.id)).toEqual(["valid-tail"]);
return attachInternalToolBatchLifecycle({}, { commitReadyCalls, releaseSkippedCalls });
});
agent.subscribe((event) => {
if (
event.type === "message_end" &&
event.message.role === "assistant" &&
event.message.stopReason === "toolUse"
) {
agent.steer({ role: "user", content: "redirect", timestamp: 2 });
}
});
await agent.prompt("start");
expect(execute).not.toHaveBeenCalled();
expect(commitReadyCalls).not.toHaveBeenCalled();
expect(releaseSkippedCalls).toHaveBeenCalledExactlyOnceWith(["valid-tail"]);
});
it("checks steering once before launching a prepared parallel batch", async () => {
const preparationReleased = createDeferred();
const preparationBlocked = createDeferred();
const execute = vi.fn(async () => ({ content: [], details: {} }));
const requestMessages: Message[][] = [];
const streamFn = createTurnSequenceStream(
[
[
{ type: "toolCall", id: "invalid", name: "required", arguments: {} },
{ type: "toolCall", id: "prepared", name: "parallel", arguments: {} },
],
[{ type: "text", text: "steer handled" }],
],
requestMessages,
);
const agent = new Agent({
initialState: {
model,
tools: [
{
name: "required",
label: "required",
description: "requires input",
parameters: Type.Object({ value: Type.String() }),
execute,
},
{ ...makeTool("parallel", []), execute },
],
},
streamFn,
toolExecution: "parallel",
beforeToolCall: async ({ toolCall }) => {
if (toolCall.id === "prepared") {
preparationBlocked.resolve();
await preparationReleased.promise;
}
return undefined;
},
});
const commitReadyCalls = vi.fn();
const releaseSkippedCalls = vi.fn();
setInternalBeforeToolBatch(agent, async ({ calls }) => {
expect(calls.map((call) => call.toolCall.id)).toEqual(["prepared"]);
return attachInternalToolBatchLifecycle({}, { commitReadyCalls, releaseSkippedCalls });
});
const events: AgentEvent[] = [];
agent.subscribe((event) => {
events.push(event);
});
const steer = { role: "user" as const, content: "before launch", timestamp: 2 };
const run = agent.prompt("start");
await preparationBlocked.promise;
agent.steer(steer);
preparationReleased.resolve();
await run;
expect(execute).not.toHaveBeenCalled();
expect(commitReadyCalls).not.toHaveBeenCalled();
expect(releaseSkippedCalls).toHaveBeenCalledExactlyOnceWith(["prepared"]);
expect(requestMessages[1]?.slice(-4)).toMatchObject([
{ role: "assistant", stopReason: "toolUse" },
{ role: "toolResult", toolCallId: "invalid", isError: true },
{
role: "toolResult",
toolCallId: "prepared",
isError: true,
content: [{ type: "text", text: "Skipped due to queued user message." }],
details: { status: "skipped", deniedReason: "steering" },
},
steer,
]);
expect(
events
.filter((event) => event.type === "tool_execution_end")
.map((event) => ({ id: event.toolCallId, kind: event.errorKind })),
).toEqual([
{ id: "invalid", kind: "argument-validation" },
{ id: "prepared", kind: undefined },
]);
});
it("commits prepared parallel calls in assistant order at launch", async () => {
const order: string[] = [];
const requestMessages: Message[][] = [];
const streamFn = createTurnSequenceStream(
[
[
{ type: "toolCall", id: "parallel-first", name: "first", arguments: {} },
{ type: "toolCall", id: "parallel-second", name: "second", arguments: {} },
],
[{ type: "text", text: "done" }],
],
requestMessages,
);
const commitReadyCalls = vi.fn((calls: readonly { toolCallId: string; args: unknown }[]) => {
order.push(`commit:${calls.map((call) => call.toolCallId).join(",")}`);
});
const releaseSkippedCalls = vi.fn();
await runAgentLoop(
[{ role: "user", content: "run in parallel", timestamp: 1 }],
{
systemPrompt: "",
messages: [],
tools: [
{
...makeTool("first", []),
execute: async () => {
order.push("execute:parallel-first");
await Promise.resolve();
order.push("gap:parallel-first");
return { content: [], details: {} };
},
},
{
...makeTool("second", []),
execute: async () => {
order.push("execute:parallel-second");
await Promise.resolve();
order.push("gap:parallel-second");
return { content: [], details: {} };
},
},
],
},
{
...config,
toolExecution: "parallel",
beforeToolBatch: async () =>
attachInternalToolBatchLifecycle({}, { commitReadyCalls, releaseSkippedCalls }),
},
() => {},
undefined,
streamFn,
);
expect(order).toEqual([
"commit:parallel-first",
"execute:parallel-first",
"commit:parallel-second",
"execute:parallel-second",
"gap:parallel-first",
"gap:parallel-second",
]);
expect(releaseSkippedCalls).not.toHaveBeenCalled();
});
it("does not launch prepared tools when the admission commit fails", async () => {
const execute = vi.fn(async () => ({ content: [], details: {} }));
const streamFn = createTurnSequenceStream(
[[{ type: "toolCall", id: "commit-failure", name: "side-effect", arguments: {} }]],
[],
);
const commitError = new Error("admission commit failed");
const releaseSkippedCalls = vi.fn();
await expect(
runAgentLoop(
[{ role: "user", content: "run", timestamp: 1 }],
{
systemPrompt: "",
messages: [],
tools: [{ ...makeTool("side-effect", []), execute }],
},
{
...config,
toolExecution: "parallel",
beforeToolBatch: async () =>
attachInternalToolBatchLifecycle(
{},
{
commitReadyCalls: () => {
throw commitError;
},
releaseSkippedCalls,
},
),
},
() => {},
undefined,
streamFn,
),
).rejects.toBe(commitError);
expect(execute).not.toHaveBeenCalled();
expect(releaseSkippedCalls).not.toHaveBeenCalled();
});
it("gives the model one recovery turn with the normal tool catalog", async () => {
const executed: string[] = [];
const providerToolNames: string[][] = [];
@@ -2611,6 +3409,8 @@ describe("agentLoop tool termination", () => {
const controller = new AbortController();
const executed: string[] = [];
const afterToolCall = vi.fn(async () => undefined);
const commitReadyCalls = vi.fn();
const releaseSkippedCalls = vi.fn();
const streamFn: StreamFn = () => {
const stream = createAssistantMessageEventStream();
queueMicrotask(() => {
@@ -2641,6 +3441,8 @@ describe("agentLoop tool termination", () => {
{
...config,
toolExecution: "parallel",
beforeToolBatch: async () =>
attachInternalToolBatchLifecycle({}, { commitReadyCalls, releaseSkippedCalls }),
beforeToolCall: async ({ toolCall }) => {
if (toolCall.name === "gated") {
await Promise.resolve();
@@ -2664,6 +3466,8 @@ describe("agentLoop tool termination", () => {
expect(executed).toEqual([]);
expect(afterToolCall).not.toHaveBeenCalled();
expect(commitReadyCalls).not.toHaveBeenCalled();
expect(releaseSkippedCalls).not.toHaveBeenCalled();
expect(
abortedMessages
.filter((message) => message.role === "toolResult")
+523 -196
View File
@@ -11,6 +11,13 @@ import type {
import type { EventStream as SourceEventStream } from "@openclaw/llm-core";
import { TranscriptNotContinuableError } from "./errors.js";
import { uuidv7 } from "./harness/session/uuid.js";
import {
getInternalToolExecutionPreparer,
getInternalSyncSteeringGetter,
type InternalToolExecutionPreparation,
takeInternalToolBatchLifecycle,
type InternalToolBatchLifecycle,
} from "./internal-hooks.js";
import { resolveAgentReasoningOption } from "./reasoning.js";
import { type AgentCoreStreamRuntimeDeps, resolveAgentCoreStreamFn } from "./runtime-deps.js";
import {
@@ -61,6 +68,17 @@ type AssistantMessageUpdateEvent = Extract<
const TOOL_LOOP_RECOVERY_TERMINATED_MESSAGE =
"OpenClaw stopped this run because tool-loop recovery encountered another critical loop. No blocked tool action was executed.";
const STEERING_TOOL_SKIP_MESSAGE = "Skipped due to queued user message.";
function getSteeringAtCheckpoint(
config: AgentLoopConfig,
): AgentMessage[] | Promise<AgentMessage[]> {
const callback = config.getSteeringMessages;
if (!callback) {
return [];
}
return getInternalSyncSteeringGetter(callback)?.() ?? callback.call(config);
}
function appendTextDeltaToAssistantMessage(
message: AssistantMessage,
@@ -290,7 +308,10 @@ async function runLoop(
criticalToolLoopSeen: false,
};
// Check for steering messages at start (user may have typed while waiting)
let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || [];
const initialSteering = getSteeringAtCheckpoint(config);
let pendingMessages: AgentMessage[] = Array.isArray(initialSteering)
? initialSteering
: await initialSteering;
const stopIfAborted = async (): Promise<boolean> => {
if (!signal?.aborted) {
return false;
@@ -340,7 +361,9 @@ async function runLoop(
// Process pending messages (inject before next assistant response)
if (pendingMessages.length > 0) {
for (const message of pendingMessages) {
const messagesToInject = pendingMessages;
pendingMessages = [];
for (const message of messagesToInject) {
if (message.role === "user") {
turnTainted = false;
}
@@ -394,6 +417,7 @@ async function runLoop(
toolResults.push(...executedToolBatch.messages);
turnTainted ||= toolResults.some(toolResultTaintsTurn);
hasMoreToolCalls = !executedToolBatch.terminate;
pendingMessages = executedToolBatch.steeringMessages;
if (executedToolBatch.intervention) {
toolLoopRecoveryState.criticalToolLoopSeen = true;
}
@@ -459,19 +483,22 @@ async function runLoop(
return;
}
if (
await config.shouldStopAfterTurn?.({
message,
toolResults,
context: currentContext,
newMessages,
})
) {
await emit({ type: "agent_end", messages: newMessages });
return;
}
if (pendingMessages.length === 0) {
if (
await config.shouldStopAfterTurn?.({
message,
toolResults,
context: currentContext,
newMessages,
})
) {
await emit({ type: "agent_end", messages: newMessages });
return;
}
pendingMessages = (await config.getSteeringMessages?.()) || [];
const steering = getSteeringAtCheckpoint(config);
pendingMessages = Array.isArray(steering) ? steering : await steering;
}
if (await stopIfAborted()) {
return;
}
@@ -617,6 +644,7 @@ async function executeToolCalls(
const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall");
const resolvedToolCalls = new Map<AgentToolCall, ResolvedToolCallOutcome>();
const validatedToolCalls = new Map<AgentToolCall, ValidatedToolCallOutcome>();
let batchLifecycle: InternalToolBatchLifecycle | undefined;
if (config.beforeToolBatch) {
for (const toolCall of toolCalls) {
if (signal?.aborted) {
@@ -662,6 +690,7 @@ async function executeToolCalls(
terminal: criticalToolLoopSeen,
});
}
batchLifecycle = admission ? takeInternalToolBatchLifecycle(admission) : undefined;
}
}
let hasSequentialToolCall = false;
@@ -691,6 +720,7 @@ async function executeToolCalls(
toolCalls,
resolvedToolCalls,
validatedToolCalls,
batchLifecycle,
config,
signal,
emit,
@@ -702,6 +732,7 @@ async function executeToolCalls(
toolCalls,
resolvedToolCalls,
validatedToolCalls,
batchLifecycle,
config,
signal,
emit,
@@ -710,6 +741,7 @@ async function executeToolCalls(
type ExecutedToolCallBatch = {
messages: ToolResultMessage[];
steeringMessages: AgentMessage[];
terminate: boolean;
terminateRun: boolean;
intervention?: ToolLoopIntervention;
@@ -738,14 +770,30 @@ async function executeToolCallsSequential(
toolCalls: AgentToolCall[],
resolvedToolCalls: Map<AgentToolCall, ResolvedToolCallOutcome>,
validatedToolCalls: Map<AgentToolCall, ValidatedToolCallOutcome>,
batchLifecycle: InternalToolBatchLifecycle | undefined,
config: AgentLoopConfig,
signal: AbortSignal | undefined,
emit: AgentEventSink,
): Promise<ExecutedToolCallBatch> {
const finalizedCalls: FinalizedToolCallOutcome[] = [];
const messages: ToolResultMessage[] = [];
let steeringMessages: AgentMessage[] = [];
let skippedReady: { args: unknown; startEmitted: true } | undefined;
let skippedStartIndex = toolCalls.length;
for (const toolCall of toolCalls) {
for (let callIndex = 0; callIndex < toolCalls.length; callIndex++) {
const toolCall = toolCalls[callIndex];
if (!toolCall) {
continue;
}
if (!signal?.aborted) {
const steering = getSteeringAtCheckpoint(config);
steeringMessages = Array.isArray(steering) ? steering : await steering;
}
if (steeringMessages.length > 0) {
skippedStartIndex = callIndex;
break;
}
const hideFromChannelProgress = hidesToolCallFromChannelProgress(
currentContext,
toolCall,
@@ -786,20 +834,49 @@ async function executeToolCallsSequential(
signal,
);
} else {
const executed = await executePreparedToolCall(
const execution = await prepareToolCallExecution(
preparation,
{ assistantMessage, toolCall: preparation.toolCall },
signal,
emit,
);
finalized = await finalizeExecutedToolCall(
currentContext,
assistantMessage,
preparation,
executed,
config,
signal,
);
if (execution.kind === "immediate") {
finalized = await finalizeExecutedToolCall(
currentContext,
assistantMessage,
preparation,
execution.outcome,
preparation.args,
config,
signal,
);
} else {
try {
if (!signal?.aborted) {
const steering = getSteeringAtCheckpoint(config);
steeringMessages = Array.isArray(steering) ? steering : await steering;
}
if (steeringMessages.length > 0) {
skippedReady = { args: execution.args, startEmitted: true };
skippedStartIndex = callIndex;
break;
}
const executed = await execution.execute(() =>
batchLifecycle?.commitReadyCalls([{ toolCallId: toolCall.id, args: execution.args }]),
);
finalized = await finalizeExecutedToolCall(
currentContext,
assistantMessage,
preparation,
executed,
execution.args,
config,
signal,
);
} finally {
execution.dispose();
}
}
}
await emitToolExecutionEnd(finalized, emit);
@@ -809,31 +886,60 @@ async function executeToolCallsSequential(
messages.push(toolResultMessage);
if (signal?.aborted) {
// Complete the skipped tail through the normal lifecycle and outcome hook
// so the committed tool-call turn stays paired and subscriber-safe.
for (let i = finalizedCalls.length; i < toolCalls.length; i++) {
const skippedToolCall = toolCalls[i];
if (!skippedToolCall) {
continue;
}
const completed = await completeAbortedToolCall(
currentContext,
assistantMessage,
skippedToolCall,
resolvedToolCalls,
config,
signal,
emit,
);
finalizedCalls.push(completed.finalized);
messages.push(completed.message);
}
skippedStartIndex = callIndex + 1;
break;
}
}
// A steer accepted during the final call's awaited preflight or execution
// must outrank shouldStopAfterTurn even when there is no remaining tail.
if (!signal?.aborted && steeringMessages.length === 0 && skippedStartIndex === toolCalls.length) {
const steering = getSteeringAtCheckpoint(config);
steeringMessages = Array.isArray(steering) ? steering : await steering;
}
// Complete the unstarted tail through one lifecycle path so committed tool
// calls remain paired and outcome hooks observe every synthetic result.
if (steeringMessages.length > 0) {
batchLifecycle?.releaseSkippedCalls(
toolCalls
.slice(skippedStartIndex)
.filter((toolCall) => validatedToolCalls.get(toolCall)?.kind === "validated")
.map((toolCall) => toolCall.id),
);
}
for (let i = skippedStartIndex; i < toolCalls.length; i++) {
const skippedToolCall = toolCalls[i];
if (!skippedToolCall) {
continue;
}
const isSteeringSkip = steeringMessages.length > 0;
const completed = await completeUnstartedToolCall(
currentContext,
assistantMessage,
skippedToolCall,
resolvedToolCalls,
config,
signal,
emit,
{
...(i === skippedStartIndex && skippedReady ? skippedReady : {}),
...(isSteeringSkip
? {
details: { status: "skipped", deniedReason: "steering" },
message: STEERING_TOOL_SKIP_MESSAGE,
}
: {}),
},
);
await emitToolResultMessage(completed.message, emit);
finalizedCalls.push(completed.finalized);
messages.push(completed.message);
}
return {
messages,
steeringMessages,
terminate: shouldTerminateToolBatch(finalizedCalls),
terminateRun: false,
};
@@ -845,119 +951,205 @@ async function executeToolCallsParallel(
toolCalls: AgentToolCall[],
resolvedToolCalls: Map<AgentToolCall, ResolvedToolCallOutcome>,
validatedToolCalls: Map<AgentToolCall, ValidatedToolCallOutcome>,
batchLifecycle: InternalToolBatchLifecycle | undefined,
config: AgentLoopConfig,
signal: AbortSignal | undefined,
emit: AgentEventSink,
): Promise<ExecutedToolCallBatch> {
const finalizedCalls: FinalizedToolCallEntry[] = [];
const pendingExecutions = new Set<ReadyToolCallExecution>();
for (const toolCall of toolCalls) {
const hideFromChannelProgress = hidesToolCallFromChannelProgress(
currentContext,
toolCall,
resolvedToolCalls,
);
await emit({
type: "tool_execution_start",
toolCallId: toolCall.id,
toolName: toolCall.name,
args: toolCall.arguments,
...(hideFromChannelProgress ? { hideFromChannelProgress: true } : {}),
});
try {
for (const toolCall of toolCalls) {
const hideFromChannelProgress = hidesToolCallFromChannelProgress(
currentContext,
toolCall,
resolvedToolCalls,
);
await emit({
type: "tool_execution_start",
toolCallId: toolCall.id,
toolName: toolCall.name,
args: toolCall.arguments,
...(hideFromChannelProgress ? { hideFromChannelProgress: true } : {}),
});
const preparation = await prepareToolCall(
currentContext,
assistantMessage,
toolCall,
config,
signal,
resolvedToolCalls,
validatedToolCalls,
);
if (preparation.kind === "immediate") {
const finalized = await finalizeToolCallOutcome(
const preparation = await prepareToolCall(
currentContext,
assistantMessage,
{
toolCall,
result: preparation.result,
isError: preparation.isError,
executionStarted: false,
...(preparation.errorKind ? { errorKind: preparation.errorKind } : {}),
...(hideFromChannelProgress ? { hideFromChannelProgress: true } : {}),
},
toolCall.arguments,
toolCall,
config,
signal,
resolvedToolCalls,
validatedToolCalls,
);
await emitToolExecutionEnd(finalized, emit);
finalizedCalls.push(finalized);
if (signal?.aborted) {
break;
if (preparation.kind === "immediate") {
const finalized = await finalizeToolCallOutcome(
currentContext,
assistantMessage,
{
toolCall,
result: preparation.result,
isError: preparation.isError,
executionStarted: false,
...(preparation.errorKind ? { errorKind: preparation.errorKind } : {}),
...(hideFromChannelProgress ? { hideFromChannelProgress: true } : {}),
},
toolCall.arguments,
config,
signal,
);
await emitToolExecutionEnd(finalized, emit);
finalizedCalls.push(finalized);
if (signal?.aborted) {
break;
}
continue;
}
continue;
}
finalizedCalls.push(async () => {
const executed = await executePreparedToolCall(
const execution = await prepareToolCallExecution(
preparation,
{ assistantMessage, toolCall: preparation.toolCall },
signal,
emit,
);
const finalized = await finalizeExecutedToolCall(
currentContext,
assistantMessage,
preparation,
executed,
config,
signal,
);
await emitToolExecutionEnd(finalized, emit);
return finalized;
});
if (signal?.aborted) {
break;
}
}
const orderedFinalizedCalls = await Promise.all(
finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
);
const messages: ToolResultMessage[] = [];
for (const finalized of orderedFinalizedCalls) {
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit);
messages.push(toolResultMessage);
}
// Complete calls skipped before queueing through the same lifecycle contract
// as the sequential path.
if (signal?.aborted && orderedFinalizedCalls.length < toolCalls.length) {
for (let i = orderedFinalizedCalls.length; i < toolCalls.length; i++) {
const skippedToolCall = toolCalls[i];
if (!skippedToolCall) {
if (execution.kind === "immediate") {
const finalized = await finalizeExecutedToolCall(
currentContext,
assistantMessage,
preparation,
execution.outcome,
preparation.args,
config,
signal,
);
await emitToolExecutionEnd(finalized, emit);
finalizedCalls.push(finalized);
if (signal?.aborted) {
break;
}
continue;
}
const completed = await completeAbortedToolCall(
currentContext,
assistantMessage,
skippedToolCall,
resolvedToolCalls,
config,
signal,
emit,
pendingExecutions.add(execution);
finalizedCalls.push({ ...preparation, execution });
if (signal?.aborted) {
break;
}
}
const steering = signal?.aborted ? [] : getSteeringAtCheckpoint(config);
const steeringMessages = Array.isArray(steering) ? steering : await steering;
const skippedToolCallIds = [
...(steeringMessages.length > 0
? finalizedCalls.flatMap((entry) => ("kind" in entry ? [entry.toolCall.id] : []))
: []),
...(steeringMessages.length > 0
? toolCalls.slice(finalizedCalls.length).map((toolCall) => toolCall.id)
: []),
];
if (skippedToolCallIds.length > 0) {
batchLifecycle?.releaseSkippedCalls(skippedToolCallIds);
}
const orderedFinalizedCalls: FinalizedToolCallOutcome[] = [];
if (steeringMessages.length > 0) {
for (const entry of finalizedCalls) {
if (!("kind" in entry)) {
orderedFinalizedCalls.push(entry);
continue;
}
entry.execution.dispose();
pendingExecutions.delete(entry.execution);
const completed = await completeUnstartedToolCall(
currentContext,
assistantMessage,
entry.toolCall,
resolvedToolCalls,
config,
signal,
emit,
{
args: entry.execution.args,
details: { status: "skipped", deniedReason: "steering" },
message: STEERING_TOOL_SKIP_MESSAGE,
startEmitted: true,
},
);
orderedFinalizedCalls.push(completed.finalized);
}
} else {
orderedFinalizedCalls.push(
...(await Promise.all(
finalizedCalls.map(async (entry) => {
if (!("kind" in entry)) {
return entry;
}
try {
const executed = await entry.execution.execute(() =>
batchLifecycle?.commitReadyCalls([
{ toolCallId: entry.toolCall.id, args: entry.execution.args },
]),
);
const finalized = await finalizeExecutedToolCall(
currentContext,
assistantMessage,
entry,
executed,
entry.execution.args,
config,
signal,
);
await emitToolExecutionEnd(finalized, emit);
return finalized;
} finally {
entry.execution.dispose();
pendingExecutions.delete(entry.execution);
}
}),
)),
);
orderedFinalizedCalls.push(completed.finalized);
messages.push(completed.message);
}
const messages: ToolResultMessage[] = [];
for (const finalized of orderedFinalizedCalls) {
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit);
messages.push(toolResultMessage);
}
// Complete calls skipped before queueing through the same lifecycle contract
// as the sequential path.
if (signal?.aborted && orderedFinalizedCalls.length < toolCalls.length) {
for (let i = orderedFinalizedCalls.length; i < toolCalls.length; i++) {
const skippedToolCall = toolCalls[i];
if (!skippedToolCall) {
continue;
}
const completed = await completeUnstartedToolCall(
currentContext,
assistantMessage,
skippedToolCall,
resolvedToolCalls,
config,
signal,
emit,
);
await emitToolResultMessage(completed.message, emit);
orderedFinalizedCalls.push(completed.finalized);
messages.push(completed.message);
}
}
return {
messages,
steeringMessages,
terminate: shouldTerminateToolBatch(orderedFinalizedCalls),
terminateRun: false,
};
} finally {
for (const execution of pendingExecutions) {
execution.dispose();
}
}
return {
messages,
terminate: shouldTerminateToolBatch(orderedFinalizedCalls),
terminateRun: false,
};
}
type PreparedToolCall = {
@@ -985,6 +1177,19 @@ type ExecutedToolCallOutcome = {
callerCancelled?: true;
};
type ReadyToolCallExecution = {
kind: "ready";
args: unknown;
execute: (onImplementationStart?: () => void) => Promise<ExecutedToolCallOutcome>;
dispose: () => void;
};
type PreparedToolCallExecution =
| { kind: "immediate"; outcome: ExecutedToolCallOutcome }
| ReadyToolCallExecution;
type ReadyPreparedToolCall = PreparedToolCall & { execution: ReadyToolCallExecution };
type FinalizedToolCallOutcome = {
toolCall: AgentToolCall;
result: AgentToolResult<unknown>;
@@ -995,7 +1200,7 @@ type FinalizedToolCallOutcome = {
resultContentSource?: ToolResultContentSource;
};
type FinalizedToolCallEntry = FinalizedToolCallOutcome | (() => Promise<FinalizedToolCallOutcome>);
type FinalizedToolCallEntry = FinalizedToolCallOutcome | ReadyPreparedToolCall;
function shouldTerminateToolBatch(finalizedCalls: FinalizedToolCallOutcome[]): boolean {
return (
@@ -1217,67 +1422,180 @@ async function validateToolCallForBatchAdmission(
};
}
async function executePreparedToolCall(
async function prepareToolCallExecution(
prepared: PreparedToolCall,
executionContext: AgentToolExecutionContext,
signal: AbortSignal | undefined,
emit: AgentEventSink,
): Promise<ExecutedToolCallOutcome> {
// Parallel batches prepare every call first. A later preflight abort must not
// let an earlier prepared, side-effectful tool start afterward.
if (signal?.aborted) {
return {
result: createErrorToolResult("Operation aborted"),
isError: true,
executionStarted: false,
};
}
): Promise<PreparedToolCallExecution> {
const updateEvents: Promise<void>[] = [];
let acceptingUpdates = true;
try {
const result = await runWithAgentToolExecutionContext(executionContext, () =>
prepared.tool.execute(
prepared.toolCall.id,
prepared.args as never,
signal,
(partialResult) => {
if (!acceptingUpdates) {
return;
}
updateEvents.push(
Promise.resolve(
emit({
type: "tool_execution_update",
toolCallId: prepared.toolCall.id,
toolName: prepared.toolCall.name,
args: prepared.toolCall.arguments,
partialResult,
...(prepared.tool.hideFromChannelProgress === true
? { hideFromChannelProgress: true }
: {}),
}),
),
);
},
const onUpdate = (partialResult: AgentToolResult<unknown>) => {
if (!acceptingUpdates) {
return;
}
updateEvents.push(
Promise.resolve(
emit({
type: "tool_execution_update",
toolCallId: prepared.toolCall.id,
toolName: prepared.toolCall.name,
args: prepared.toolCall.arguments,
partialResult,
...(prepared.tool.hideFromChannelProgress === true
? { hideFromChannelProgress: true }
: {}),
}),
),
);
};
const finishUpdates = async () => {
acceptingUpdates = false;
await Promise.all(updateEvents);
return { result, isError: false, executionStarted: true };
} catch (error) {
acceptingUpdates = false;
await Promise.all(updateEvents);
};
const immediateError = async (error: unknown): Promise<PreparedToolCallExecution> => {
await finishUpdates();
return {
result: createErrorToolResult(error instanceof Error ? error.message : String(error)),
isError: true,
executionStarted: true,
...(signal?.aborted && error === signal.reason ? { callerCancelled: true } : {}),
kind: "immediate",
outcome: {
result: createErrorToolResult(error instanceof Error ? error.message : String(error)),
isError: true,
executionStarted: false,
},
};
} finally {
acceptingUpdates = false;
};
const readyExecution = (
args: unknown,
run: (onImplementationStart: () => void) => Promise<AgentToolResult<unknown>>,
disposeSource: () => void = () => {},
): ReadyToolCallExecution => {
let disposed = false;
const dispose = () => {
if (!disposed) {
disposed = true;
acceptingUpdates = false;
disposeSource();
}
};
return {
kind: "ready",
args,
dispose,
async execute(onImplementationStart) {
if (signal?.aborted) {
dispose();
await finishUpdates();
return {
result: createErrorToolResult("Operation aborted"),
isError: true,
executionStarted: false,
};
}
let executionStarted = false;
let implementationStartError: { error: unknown } | undefined;
try {
const result = await run(() => {
try {
onImplementationStart?.();
} catch (error) {
implementationStartError = { error };
throw error;
}
executionStarted = true;
});
if (implementationStartError) {
throw implementationStartError.error;
}
await finishUpdates();
return { result, isError: false, executionStarted };
} catch (error) {
await finishUpdates();
if (implementationStartError) {
throw implementationStartError.error;
}
return {
result: createErrorToolResult(error instanceof Error ? error.message : String(error)),
isError: true,
executionStarted,
...(executionStarted && signal?.aborted && error === signal.reason
? { callerCancelled: true }
: {}),
};
} finally {
dispose();
}
},
};
};
const preparer = getInternalToolExecutionPreparer(prepared.tool);
if (!preparer) {
return readyExecution(prepared.args, async (onImplementationStart) => {
if (signal?.aborted) {
throw signal.reason ?? new Error("Operation aborted");
}
return await runWithAgentToolExecutionContext(executionContext, () => {
onImplementationStart();
return prepared.tool.execute(
prepared.toolCall.id,
prepared.args as never,
signal,
onUpdate,
);
});
});
}
let internalPreparation: InternalToolExecutionPreparation;
try {
internalPreparation = await runWithAgentToolExecutionContext(executionContext, () =>
preparer({
toolCallId: prepared.toolCall.id,
args: prepared.args,
...(signal ? { signal } : {}),
onUpdate,
}),
);
} catch (error) {
return await immediateError(error);
}
if (internalPreparation.kind === "immediate") {
internalPreparation.dispose();
await finishUpdates();
if (internalPreparation.outcome.kind === "result") {
return {
kind: "immediate",
outcome: {
result: internalPreparation.outcome.result,
isError: internalPreparation.outcome.isError,
executionStarted: false,
},
};
}
return {
kind: "immediate",
outcome: {
result: createErrorToolResult(
internalPreparation.outcome.error instanceof Error
? internalPreparation.outcome.error.message
: String(internalPreparation.outcome.error),
),
isError: true,
executionStarted: false,
},
};
}
const readyPreparation = internalPreparation;
return readyExecution(
readyPreparation.args,
(onImplementationStart) =>
runWithAgentToolExecutionContext(executionContext, () =>
readyPreparation.execute(onImplementationStart),
),
readyPreparation.dispose,
);
}
async function finalizeExecutedToolCall(
@@ -1285,6 +1603,7 @@ async function finalizeExecutedToolCall(
assistantMessage: AssistantMessage,
prepared: PreparedToolCall,
executed: ExecutedToolCallOutcome,
finalArgs: unknown,
config: AgentLoopConfig,
signal: AbortSignal | undefined,
): Promise<FinalizedToolCallOutcome> {
@@ -1297,7 +1616,7 @@ async function finalizeExecutedToolCall(
{
assistantMessage,
toolCall: prepared.toolCall,
args: prepared.args,
args: finalArgs,
result,
isError,
context: currentContext,
@@ -1334,7 +1653,7 @@ async function finalizeExecutedToolCall(
? { resultContentSource: prepared.tool.resultContentSource }
: {}),
},
prepared.args,
finalArgs,
config,
signal,
);
@@ -1464,6 +1783,7 @@ async function completeToolLoopInterventionBatch(params: {
}
return {
messages,
steeringMessages: [],
// A later critical loop always forces termination. During first recovery,
// honor the outcome hooks: if every finalized outcome says terminate, the
// batch ends without another provider turn.
@@ -1473,7 +1793,7 @@ async function completeToolLoopInterventionBatch(params: {
};
}
async function completeAbortedToolCall(
async function completeUnstartedToolCall(
currentContext: AgentContext,
assistantMessage: AssistantMessage,
toolCall: AgentToolCall,
@@ -1481,43 +1801,50 @@ async function completeAbortedToolCall(
config: AgentLoopConfig,
signal: AbortSignal | undefined,
emit: AgentEventSink,
options: {
args?: unknown;
details?: unknown;
message?: string;
startEmitted?: boolean;
} = {},
): Promise<{ finalized: FinalizedToolCallOutcome; message: ToolResultMessage }> {
const hideFromChannelProgress = hidesToolCallFromChannelProgress(
currentContext,
toolCall,
resolvedToolCalls,
);
await emit({
type: "tool_execution_start",
toolCallId: toolCall.id,
toolName: toolCall.name,
args: toolCall.arguments,
...(hideFromChannelProgress ? { hideFromChannelProgress: true } : {}),
});
if (!options.startEmitted) {
await emit({
type: "tool_execution_start",
toolCallId: toolCall.id,
toolName: toolCall.name,
args: toolCall.arguments,
...(hideFromChannelProgress ? { hideFromChannelProgress: true } : {}),
});
}
const finalized = await finalizeToolCallOutcome(
currentContext,
assistantMessage,
{
toolCall,
result: createErrorToolResult("Operation aborted"),
result: createErrorToolResult(options.message ?? "Operation aborted", options.details),
isError: true,
executionStarted: false,
...(hideFromChannelProgress ? { hideFromChannelProgress: true } : {}),
},
toolCall.arguments,
"args" in options ? options.args : toolCall.arguments,
config,
signal,
);
await emitToolExecutionEnd(finalized, emit);
const message = createToolResultMessage(finalized);
await emitToolResultMessage(message, emit);
return { finalized, message };
}
function createErrorToolResult(message: string): AgentToolResult<unknown> {
function createErrorToolResult(message: string, details: unknown = {}): AgentToolResult<unknown> {
return {
content: [{ type: "text", text: message }],
details: {},
details,
};
}
+18 -10
View File
@@ -10,7 +10,7 @@ import type {
} from "@openclaw/llm-core";
import { runAgentLoop, runAgentLoopContinue } from "./agent-loop.js";
import { TranscriptNotContinuableError } from "./errors.js";
import { getInternalBeforeToolBatch } from "./internal-hooks.js";
import { attachInternalSyncSteeringGetter, getInternalBeforeToolBatch } from "./internal-hooks.js";
import { resolveAgentReasoningOption } from "./reasoning.js";
import { type AgentCoreStreamRuntimeDeps, resolveAgentCoreStreamFn } from "./runtime-deps.js";
import {
@@ -146,7 +146,7 @@ export interface AgentOptions {
context: PrepareNextTurnContext,
signal?: AbortSignal,
) => Promise<AgentLoopTurnUpdate | undefined> | AgentLoopTurnUpdate | undefined;
/** Queue drain mode for steering messages injected before the next assistant response. */
/** Queue drain mode for steering messages applied before the next unstarted tool or model turn. */
steeringMode?: QueueMode;
/** Queue drain mode for follow-up messages injected after the agent would otherwise stop. */
followUpMode?: QueueMode;
@@ -330,7 +330,10 @@ export class Agent {
return this.followUpQueue.mode;
}
/** Queue a message to be injected after the current assistant turn finishes. */
/**
* Queue a message for the active run. Running tools finish, while sequential
* tail calls or a parallel batch that has not launched yet are skipped.
*/
steer(message: AgentMessage): void {
this.steeringQueue.enqueue(message);
}
@@ -498,6 +501,17 @@ export class Agent {
private createLoopConfig(options: { skipInitialSteeringPoll?: boolean } = {}): AgentLoopConfig {
let skipInitialSteeringPoll = options.skipInitialSteeringPoll === true;
const drainSteeringMessages = () => {
if (skipInitialSteeringPoll) {
skipInitialSteeringPoll = false;
return [];
}
return this.steeringQueue.drain();
};
const getSteeringMessages = attachInternalSyncSteeringGetter(
async () => drainSteeringMessages(),
drainSteeringMessages,
);
return {
model: this.mutableState.model,
thinkingLevel: this.mutableState.thinkingLevel,
@@ -530,13 +544,7 @@ export class Agent {
convertToLlm: this.convertToLlm,
transformContext: this.transformContext,
getApiKey: this.getApiKey,
getSteeringMessages: async () => {
if (skipInitialSteeringPoll) {
skipInitialSteeringPoll = false;
return [];
}
return this.steeringQueue.drain();
},
getSteeringMessages,
getFollowUpMessages: async () => this.followUpQueue.drain(),
};
}
+110 -1
View File
@@ -1,4 +1,11 @@
import type { InternalBeforeToolBatchContext, InternalBeforeToolBatchResult } from "./types.js";
import type {
AgentLoopConfig,
AgentMessage,
AgentToolResult,
AgentToolUpdateCallback,
InternalBeforeToolBatchContext,
InternalBeforeToolBatchResult,
} from "./types.js";
export type InternalBeforeToolBatchHook = (
context: InternalBeforeToolBatchContext,
@@ -7,6 +14,52 @@ export type InternalBeforeToolBatchHook = (
const beforeToolBatchByAgent = new WeakMap<object, InternalBeforeToolBatchHook>();
type InternalReadyToolCall = { toolCallId: string; args: unknown };
export type InternalToolBatchLifecycle = {
/** Commit admitted calls whose tool implementations are about to start. May throw before launch. */
commitReadyCalls: (calls: readonly InternalReadyToolCall[]) => void;
/** Release admission state for admitted prepared calls suppressed by steering. */
releaseSkippedCalls: (toolCallIds: readonly string[]) => void;
};
const toolBatchLifecycleByResult = new WeakMap<
InternalBeforeToolBatchResult,
InternalToolBatchLifecycle
>();
type InternalSteeringGetter = NonNullable<AgentLoopConfig["getSteeringMessages"]>;
type InternalSyncSteeringGetter = () => AgentMessage[];
const syncSteeringGetterByCallback = new WeakMap<
InternalSteeringGetter,
InternalSyncSteeringGetter
>();
export type InternalToolExecutionPreparation =
| {
kind: "immediate";
outcome:
| { kind: "result"; result: AgentToolResult<unknown>; isError: boolean }
| { kind: "error"; error: unknown };
dispose: () => void;
}
| {
kind: "ready";
args: unknown;
execute: (onImplementationStart?: () => void) => Promise<AgentToolResult<unknown>>;
dispose: () => void;
};
export type InternalToolExecutionPreparer = (params: {
toolCallId: string;
args: unknown;
signal?: AbortSignal;
onUpdate?: AgentToolUpdateCallback;
executionArgs?: unknown[];
}) => Promise<InternalToolExecutionPreparation>;
const toolExecutionPreparerByTool = new WeakMap<object, InternalToolExecutionPreparer>();
/** Install OpenClaw-owned loop control without adding a plugin-facing Agent option. */
export function setInternalBeforeToolBatch(
agent: object,
@@ -22,3 +75,59 @@ export function setInternalBeforeToolBatch(
export function getInternalBeforeToolBatch(agent: object): InternalBeforeToolBatchHook | undefined {
return beforeToolBatchByAgent.get(agent);
}
/** Attach scheduler lifecycle ownership without widening the public admission result. */
export function attachInternalToolBatchLifecycle(
result: InternalBeforeToolBatchResult,
lifecycle: InternalToolBatchLifecycle,
): InternalBeforeToolBatchResult {
toolBatchLifecycleByResult.set(result, lifecycle);
return result;
}
export function takeInternalToolBatchLifecycle(
result: InternalBeforeToolBatchResult,
): InternalToolBatchLifecycle | undefined {
const lifecycle = toolBatchLifecycleByResult.get(result);
toolBatchLifecycleByResult.delete(result);
return lifecycle;
}
/** Attach Agent-owned synchronous draining to the exact public async callback identity. */
export function attachInternalSyncSteeringGetter(
callback: InternalSteeringGetter,
syncGetter: InternalSyncSteeringGetter,
): InternalSteeringGetter {
syncSteeringGetterByCallback.set(callback, syncGetter);
return callback;
}
export function getInternalSyncSteeringGetter(
callback: InternalSteeringGetter,
): InternalSyncSteeringGetter | undefined {
return syncSteeringGetterByCallback.get(callback);
}
/** Attach OpenClaw-owned two-phase execution without changing the public AgentTool shape. */
export function attachInternalToolExecutionPreparer<T extends object>(
tool: T,
preparer: InternalToolExecutionPreparer,
): T {
toolExecutionPreparerByTool.set(tool, preparer);
return tool;
}
export function getInternalToolExecutionPreparer(
tool: object,
): InternalToolExecutionPreparer | undefined {
return toolExecutionPreparerByTool.get(tool);
}
/** Preserve private execution ownership when an adapter replaces a tool object. */
export function copyInternalToolExecutionPreparer<T extends object>(source: object, target: T): T {
const preparer = toolExecutionPreparerByTool.get(source);
if (preparer) {
toolExecutionPreparerByTool.set(target, preparer);
}
return target;
}
+16 -9
View File
@@ -27,8 +27,8 @@ export type StreamFn = LlmStreamFn;
/**
* 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.
* - "sequential": each tool call is prepared, checked for steering, executed, and finalized before the next one starts.
* - "parallel": tool calls are prepared sequentially, checked for steering once, 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.
*/
@@ -271,7 +271,8 @@ export interface AgentLoopConfig extends SimpleStreamOptions {
* Called after each turn fully completes and `turn_end` has been emitted.
*
* If it returns true, the loop emits `agent_end` and exits before polling steering or follow-up queues,
* without starting another LLM call. The current assistant response and any tool executions finish normally.
* without starting another LLM call. Steering already drained at a tool checkpoint takes precedence,
* so this hook is deferred until that steering turn completes.
*
* Use this to request a graceful stop after the current turn, e.g. before context gets too full.
*
@@ -291,13 +292,18 @@ export interface AgentLoopConfig extends SimpleStreamOptions {
/**
* Returns steering messages to inject into the conversation mid-run.
*
* Called after the current assistant turn finishes executing its tool calls, unless `shouldStopAfterTurn` exits first.
* If messages are returned, they are added to the context before the next LLM call.
* Tool calls from the current assistant message are not skipped.
* Sequential execution checks before each tool starts, including again after
* asynchronous preparation. Parallel execution checks once after preparation
* and immediately before launching the prepared calls. A non-empty result
* skips calls that have not started and is added to context before the next
* LLM call; already-running calls continue.
*
* Once a check returns messages, the loop carries that exact result to the
* next turn without polling again. This preserves queue drain ordering.
*
* Use this for "steering" the agent while it's working.
*
* Contract: must not throw or reject. Return [] when no steering messages are available.
* Contract: must not throw or reject. Resolve to [] when no steering messages are available.
*/
getSteeringMessages?: () => Promise<AgentMessage[]>;
@@ -316,8 +322,9 @@ export interface AgentLoopConfig extends SimpleStreamOptions {
/**
* Tool execution mode.
* - "sequential": execute tool calls one by one
* - "sequential": execute tool calls one by one, checking for steering before each starts
* - "parallel": preflight tool calls sequentially, then execute allowed tools concurrently;
* steering is checked once immediately before prepared calls launch;
* emit `tool_execution_end` in tool completion order after each tool is finalized,
* then emit tool-result message artifacts later in assistant source order
*
@@ -624,7 +631,7 @@ export type AgentEvent =
toolName: string;
result: unknown;
isError: boolean;
/** False when resolution, argument preparation, validation, or policy blocked execution. */
/** False when resolution, preparation, validation, policy, or queued steering prevented execution. */
executionStarted?: boolean;
/** Typed pre-execution failure provenance for safe downstream diagnostics. */
errorKind?: "argument-validation";
+214 -90
View File
@@ -21,6 +21,10 @@ import {
finalizeBeforeToolCallExecutionParams,
prepareBeforeToolCallExecutionParams,
} from "./agent-tools.before-tool-call.wrapper.js";
import {
createInternalExecutionPreparer,
readInternalExecutionControl,
} from "./agent-tools.execution-preparer.js";
import {
copyCodeModeControlToolIdentity,
getCodeModeExecBeforeHookMetadata,
@@ -29,6 +33,10 @@ import {
import { sanitizeForConsole } from "./console-sanitize.js";
import type { ClientToolDefinition } from "./embedded-agent-runner/run/params.js";
import type { AgentTool, AgentToolResult, AgentToolUpdateCallback } from "./runtime/index.js";
import {
attachInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
} from "./runtime/internal-hooks.js";
import type { ToolDefinition } from "./sessions/index.js";
import { normalizeToolName } from "./tool-policy.js";
import { jsonResult, payloadTextResult, ToolInputError } from "./tools/common.js";
@@ -256,6 +264,49 @@ function buildToolExecutionErrorResult(params: {
});
}
async function executeAdaptedToolOperation(params: {
toolCallId: string;
normalizedToolName: string;
rawParams: unknown;
getEffectiveParams: () => unknown;
signal: AbortSignal | undefined;
run: () => Promise<unknown>;
hookContext: HookContext | undefined;
}): Promise<AgentToolResult<unknown>> {
try {
return normalizeToolExecutionResult({
toolName: params.normalizedToolName,
result: await params.run(),
});
} catch (err) {
if (params.signal?.aborted) {
throw err;
}
if (isBeforeToolCallBlockedError(err)) {
logDebug(`tools: ${params.normalizedToolName} blocked by before_tool_call: ${err.reason}`);
return buildBlockedToolResult({
reason: err.reason,
toolCallId: params.toolCallId,
runId: params.hookContext?.runId,
});
}
const described = describeToolExecutionError(err);
if (described.stack && described.stack !== described.message) {
logDebug(`tools: ${params.normalizedToolName} failed stack:\n${described.stack}`);
}
const inputPreview = describeToolFailureInputs({
toolName: params.normalizedToolName,
rawParams: params.rawParams,
effectiveParams: params.getEffectiveParams(),
});
logError(`[tools] ${params.normalizedToolName} failed: ${described.message} ${inputPreview}`);
return buildToolExecutionErrorResult({
toolName: params.normalizedToolName,
message: described.message,
});
}
}
function splitToolExecuteArgs(args: ToolExecuteArgsAny): {
toolCallId: string;
params: unknown;
@@ -280,6 +331,21 @@ function splitToolExecuteArgs(args: ToolExecuteArgsAny): {
};
}
function attachAdapterExecutionPreparer<T extends ToolDefinition>(definition: T): T {
return attachInternalToolExecutionPreparer(
definition,
createInternalExecutionPreparer((params, control) =>
definition.execute(
params.toolCallId,
params.args,
params.signal,
params.onUpdate,
control as never,
),
),
);
}
const CLIENT_TOOL_NAME_CONFLICT_PREFIX = "client tool name conflict:";
/** Find client-hosted tool names that collide with runtime or sibling tools. */
@@ -336,6 +402,7 @@ export function toToolDefinitions(
const name = tool.name || "tool";
const normalizedName = normalizeToolName(name);
const beforeHookWrapped = isToolWrappedWithBeforeToolCallHook(tool);
const sourcePreparer = getInternalToolExecutionPreparer(tool);
const definition = {
name,
label: tool.label ?? name,
@@ -347,106 +414,153 @@ export function toToolDefinitions(
executionMode: tool.executionMode,
execute: async (...args: ToolExecuteArgs): Promise<AgentToolResult<unknown>> => {
const { toolCallId, params, onUpdate, signal } = splitToolExecuteArgs(args);
const control = readInternalExecutionControl(args[4]);
recordStructuredReplayTrustForToolCall(toolCallId, tool, hookContext?.runId);
let executeParams = params;
try {
if (!beforeHookWrapped) {
const preparedParams = await prepareBeforeToolCallExecutionParams({
tool,
params,
...(toolCallId ? { toolCallId } : {}),
...(hookContext ? { ctx: hookContext } : {}),
...(signal ? { signal } : {}),
});
const hookParams = normalizeCodeModeExecBeforeHookParams({
tool,
params: preparedParams,
});
const hookMetadata = getCodeModeExecBeforeHookMetadata({
tool,
params: preparedParams,
});
const hookOutcome = await runBeforeToolCallHook({
toolName: name,
params: hookParams,
...hookMetadata,
toolCallId,
ctx: hookContext,
signal,
});
if (hookOutcome.blocked) {
if (hookOutcome.kind === "veto") {
return await executeAdaptedToolOperation({
toolCallId,
normalizedToolName: normalizedName,
rawParams: params,
getEffectiveParams: () => executeParams,
signal,
hookContext,
run: async () => {
if (!beforeHookWrapped) {
const preparedParams = await prepareBeforeToolCallExecutionParams({
tool,
params,
...(toolCallId ? { toolCallId } : {}),
...(hookContext ? { ctx: hookContext } : {}),
...(signal ? { signal } : {}),
});
const hookParams = normalizeCodeModeExecBeforeHookParams({
tool,
params: preparedParams,
});
const hookMetadata = getCodeModeExecBeforeHookMetadata({
tool,
params: preparedParams,
});
const hookOutcome = await runBeforeToolCallHook({
toolName: name,
params: hookParams,
...hookMetadata,
toolCallId,
ctx: hookContext,
signal,
});
if (hookOutcome.blocked) {
if (hookOutcome.kind === "veto") {
return buildBlockedToolResult({
reason: hookOutcome.reason,
deniedReason: hookOutcome.deniedReason,
toolCallId,
runId: hookContext?.runId,
});
}
throw new Error(hookOutcome.reason);
}
executeParams = finalizeBeforeToolCallExecutionParams({
tool,
preparedParams,
hookParams,
adjustedParams: hookOutcome.params,
finalizerMode: "adapter",
});
const decision = control ? await control.pause(executeParams) : undefined;
if (decision && !decision.launch) {
return { content: [], details: { status: "skipped" } };
}
// A voice grant binds the post-finalizer execution shape. Consuming it
// earlier would let later alias or tool-owned rewrites escape the grant.
const voiceConfirmation = consumeFinalClientVoiceToolConfirmation({
toolName: name,
params: executeParams,
ctx: hookContext,
});
if (!voiceConfirmation.allowed) {
return buildBlockedToolResult({
reason: hookOutcome.reason,
deniedReason: hookOutcome.deniedReason,
reason: voiceConfirmation.reason,
deniedReason: "client-voice-confirmation",
toolCallId,
runId: hookContext?.runId,
});
}
throw new Error(hookOutcome.reason);
decision?.start?.();
recordAdjustedParamsForToolCall(toolCallId, executeParams, hookContext?.runId);
}
executeParams = finalizeBeforeToolCallExecutionParams({
tool,
preparedParams,
hookParams,
adjustedParams: hookOutcome.params,
finalizerMode: "adapter",
});
// A voice grant binds the post-finalizer execution shape. Consuming it
// earlier would let later alias or tool-owned rewrites escape the grant.
const voiceConfirmation = consumeFinalClientVoiceToolConfirmation({
toolName: name,
params: executeParams,
ctx: hookContext,
});
if (!voiceConfirmation.allowed) {
return buildBlockedToolResult({
reason: voiceConfirmation.reason,
deniedReason: "client-voice-confirmation",
toolCallId,
runId: hookContext?.runId,
});
}
recordAdjustedParamsForToolCall(toolCallId, executeParams, hookContext?.runId);
}
const rawResult = await tool.execute(toolCallId, executeParams, signal, onUpdate);
const result = normalizeToolExecutionResult({
toolName: normalizedName,
result: rawResult,
});
return result;
} catch (err) {
if (signal?.aborted) {
throw err;
}
if (isBeforeToolCallBlockedError(err)) {
logDebug(`tools: ${normalizedName} blocked by before_tool_call: ${err.reason}`);
return buildBlockedToolResult({
reason: err.reason,
toolCallId,
runId: hookContext?.runId,
});
}
const described = describeToolExecutionError(err);
if (described.stack && described.stack !== described.message) {
logDebug(`tools: ${normalizedName} failed stack:\n${described.stack}`);
}
const inputPreview = describeToolFailureInputs({
toolName: normalizedName,
rawParams: params,
effectiveParams: executeParams,
});
logError(`[tools] ${normalizedName} failed: ${described.message} ${inputPreview}`);
return buildToolExecutionErrorResult({
toolName: normalizedName,
message: described.message,
});
}
return await tool.execute(toolCallId, executeParams, signal, onUpdate);
},
});
},
} satisfies ToolDefinition;
copyCodeModeControlToolIdentity(tool, definition);
return definition;
if (!sourcePreparer) {
return beforeHookWrapped ? definition : attachAdapterExecutionPreparer(definition);
}
return attachInternalToolExecutionPreparer(definition, async (params) => {
recordStructuredReplayTrustForToolCall(params.toolCallId, tool, hookContext?.runId);
const settle = (run: () => Promise<unknown>) =>
executeAdaptedToolOperation({
toolCallId: params.toolCallId,
normalizedToolName: normalizedName,
rawParams: params.args,
getEffectiveParams: () => params.args,
signal: params.signal,
hookContext,
run,
});
type ImmediateOutcome = Extract<
Awaited<ReturnType<typeof sourcePreparer>>,
{ kind: "immediate" }
>["outcome"];
const settleImmediate = async (outcome: ImmediateOutcome, dispose: () => void) => {
try {
const result = await settle(async () => {
if (outcome.kind === "error") {
throw outcome.error;
}
return outcome.result;
});
return {
kind: "immediate" as const,
outcome: {
kind: "result" as const,
result,
isError: outcome.kind === "result" && outcome.isError,
},
dispose,
};
} catch (error) {
return {
kind: "immediate" as const,
outcome: { kind: "error" as const, error },
dispose,
};
}
};
let prepared: Awaited<ReturnType<typeof sourcePreparer>>;
try {
prepared = await sourcePreparer({
toolCallId: params.toolCallId,
args: params.args,
...(params.signal ? { signal: params.signal } : {}),
...(params.onUpdate ? { onUpdate: params.onUpdate } : {}),
});
} catch (error) {
return await settleImmediate({ kind: "error", error }, () => {});
}
if (prepared.kind === "immediate") {
return await settleImmediate(prepared.outcome, prepared.dispose);
}
const ready = prepared;
return {
kind: "ready",
args: ready.args,
execute: (onImplementationStart) => settle(() => ready.execute(onImplementationStart)),
dispose: ready.dispose,
};
});
});
}
@@ -502,13 +616,14 @@ export function toClientToolDefinitions(
): ToolDefinition[] {
return tools.map((tool) => {
const func = tool.function;
return {
const definition = {
name: func.name,
label: func.name,
description: func.description ?? "",
parameters: func.parameters as ToolDefinition["parameters"],
execute: async (...args: ToolExecuteArgs): Promise<AgentToolResult<unknown>> => {
const { toolCallId, params, signal } = splitToolExecuteArgs(args);
const control = readInternalExecutionControl(args[4]);
if (onClientToolCall && typeof onClientToolCall !== "function") {
onClientToolCall.reserve?.(toolCallId, func.name);
}
@@ -539,6 +654,13 @@ export function toClientToolDefinitions(
const paramsRecord = coerceParamsRecord(adjustedParams, func.parameters);
// Client-hosted tools have no tool-owned finalizer, so hook reconciliation
// produces the canonical execution shape consumed here.
const decision = control ? await control.pause(paramsRecord) : undefined;
if (decision && !decision.launch) {
if (onClientToolCall && typeof onClientToolCall !== "function") {
onClientToolCall.discard?.(toolCallId, func.name);
}
return { content: [], details: { status: "skipped" } };
}
const voiceConfirmation = consumeFinalClientVoiceToolConfirmation({
toolName: func.name,
params: paramsRecord,
@@ -555,6 +677,7 @@ export function toClientToolDefinitions(
runId: hookContext?.runId,
});
}
decision?.start?.();
// Notify handler that a client tool was called.
if (onClientToolCall) {
if (typeof onClientToolCall === "function") {
@@ -586,5 +709,6 @@ export function toClientToolDefinitions(
};
},
} satisfies ToolDefinition;
return attachAdapterExecutionPreparer(definition);
});
}
+2
View File
@@ -3,6 +3,7 @@ import type { AnyAgentTool } from "./agent-tools.types.js";
import { copyBeforeToolCallHookMarker } from "./before-tool-call-metadata.js";
import { copyChannelAgentToolMeta } from "./channel-tool-metadata.js";
import { copyCodeModeControlToolIdentity } from "./code-mode-control-tools.js";
import { copyInternalToolExecutionPreparer } from "./runtime/internal-hooks.js";
import { copyToolTerminalPresentation } from "./tool-terminal-presentation.js";
/**
@@ -18,5 +19,6 @@ export function copyAgentToolMetadata<T extends AnyAgentTool>(source: AnyAgentTo
copyBeforeToolCallHookMarker(source, target);
copyToolTerminalPresentation(source, target);
copyCodeModeControlToolIdentity(source, target);
copyInternalToolExecutionPreparer(source, target);
return target;
}
+47 -1
View File
@@ -6,6 +6,10 @@ import { createAbortError } from "../infra/abort-signal.js";
*/
import { copyAgentToolMetadata } from "./agent-tool-metadata.js";
import type { AnyAgentTool } from "./agent-tools.types.js";
import {
attachInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
} from "./runtime/internal-hooks.js";
function throwAbortError(): never {
throw createAbortError("Aborted");
@@ -88,5 +92,47 @@ export function wrapToolWithAbortSignal(
);
},
};
return copyAgentToolMetadata(tool, wrappedTool);
copyAgentToolMetadata(tool, wrappedTool);
const sourcePreparer = getInternalToolExecutionPreparer(tool);
if (sourcePreparer) {
attachInternalToolExecutionPreparer(wrappedTool, async (params) => {
const combinedSignal = params.signal
? AbortSignal.any([params.signal, abortSignal])
: abortSignal;
if (combinedSignal.aborted) {
throwAbortError();
}
const yieldRunSignal = tool.name === "sessions_yield" ? abortSignal : undefined;
const sourcePreparation = sourcePreparer({ ...params, signal: combinedSignal });
let prepared;
try {
prepared = await raceWithAbortSignal(sourcePreparation, combinedSignal, yieldRunSignal);
} catch (error) {
void sourcePreparation.then(
(latePreparation) => latePreparation.dispose(),
() => undefined,
);
throw error;
}
if (prepared.kind === "immediate") {
return prepared;
}
return {
kind: "ready",
args: prepared.args,
execute: (onImplementationStart) => {
if (combinedSignal.aborted) {
throwAbortError();
}
return raceWithAbortSignal(
prepared.execute(onImplementationStart),
combinedSignal,
yieldRunSignal,
);
},
dispose: prepared.dispose,
};
});
}
return wrappedTool;
}
@@ -15,7 +15,10 @@ import {
resetDiagnosticEventsForTest,
type DiagnosticEventPayload,
} from "../infra/diagnostic-events.js";
import { resetDiagnosticSessionStateForTest } from "../logging/diagnostic-session-state.js";
import {
getDiagnosticSessionState,
resetDiagnosticSessionStateForTest,
} from "../logging/diagnostic-session-state.js";
import {
initializeGlobalHookRunner,
resetGlobalHookRunner,
@@ -55,7 +58,10 @@ import type { AnyAgentTool } from "./agent-tools.types.js";
import { markCodeModeControlTool } from "./code-mode-control-tools.js";
import { CODE_MODE_EXEC_TOOL_NAME, createCodeModeTools } from "./code-mode.js";
import { splitSdkTools } from "./embedded-agent-runner/tool-split.js";
import { getInternalToolExecutionPreparer } from "./runtime/internal-hooks.js";
import type { ExtensionContext } from "./sessions/index.js";
import { wrapToolDefinition } from "./sessions/tools/tool-definition-wrapper.js";
import { hashToolCall, recordToolCall } from "./tool-loop-detection.js";
import { setToolTerminalPresentation } from "./tool-terminal-presentation.js";
type BeforeToolCallHandlerMock = ReturnType<typeof vi.fn>;
@@ -476,6 +482,321 @@ describe("before_tool_call hook deduplication (#15502)", () => {
expect(beforeToolCallHook).toHaveBeenCalledTimes(1);
});
it("preserves private execution semantics through both session tool adapters", async () => {
const runId = "run-private-preparer-adapter";
const source = wrapToolWithBeforeToolCallHook(
asAgentTool({
name: "search",
execute: vi.fn().mockResolvedValue({ answer: 42 }),
}),
{ runId },
);
const definition = expectDefined(
toToolDefinitions([source], { runId })[0],
"wrapped search tool definition",
);
const hydrated = wrapToolDefinition(definition);
const preparer = expectDefined(
getInternalToolExecutionPreparer(hydrated),
"adapted private execution preparer",
);
const prepared = await preparer({
toolCallId: "call-private-preparer-adapter",
args: { query: "answer" },
});
expect(prepared.kind).toBe("ready");
if (prepared.kind !== "ready") {
return;
}
const result = await prepared.execute();
prepared.dispose();
expect(result.details).toEqual({ answer: 42 });
expect(
beforeToolCallTesting.structuredReplaySafeToolCallIds.has(
beforeToolCallTesting.buildAdjustedParamsKey({
runId,
toolCallId: "call-private-preparer-adapter",
}),
),
).toBe(true);
});
it("preserves adapter error and abort handling for private execution", async () => {
const failure = new Error("private execution failed");
const failedSource = wrapToolWithBeforeToolCallHook(
asAgentTool({ name: "read", execute: vi.fn().mockRejectedValue(failure) }),
);
const failedTool = wrapToolDefinition(
expectDefined(toToolDefinitions([failedSource])[0], "failed private tool definition"),
);
const failedPreparer = expectDefined(
getInternalToolExecutionPreparer(failedTool),
"failed-tool private execution preparer",
);
const failedPrepared = await failedPreparer({ toolCallId: "call-failed", args: {} });
expect(failedPrepared.kind).toBe("ready");
if (failedPrepared.kind !== "ready") {
return;
}
await expect(failedPrepared.execute()).resolves.toMatchObject({
details: { status: "error", error: failure.message },
});
failedPrepared.dispose();
const controller = new AbortController();
const abortReason = new Error("private execution aborted");
const abortedSource = wrapToolWithBeforeToolCallHook(
asAgentTool({
name: "read",
execute: vi.fn(async (_id, _params, signal?: AbortSignal) => {
signal?.throwIfAborted();
return { content: [], details: { ok: true } };
}),
}),
);
const abortedTool = wrapToolDefinition(
expectDefined(toToolDefinitions([abortedSource])[0], "aborted private tool definition"),
);
const abortedPreparer = expectDefined(
getInternalToolExecutionPreparer(abortedTool),
"aborted-tool private execution preparer",
);
const abortedPrepared = await abortedPreparer({
toolCallId: "call-aborted",
args: {},
signal: controller.signal,
});
expect(abortedPrepared.kind).toBe("ready");
if (abortedPrepared.kind !== "ready") {
return;
}
const execution = abortedPrepared.execute();
controller.abort(abortReason);
await expect(execution).rejects.toBe(abortReason);
abortedPrepared.dispose();
});
it("finalizes private policy outcomes before launch", async () => {
beforeToolCallHook = installBeforeToolCallHook({
runBeforeToolCallImpl: async () => ({ block: true, blockReason: "blocked by policy" }),
});
const execute = vi.fn().mockResolvedValue({ content: [], details: { ok: true } });
const source = wrapToolWithBeforeToolCallHook(asAgentTool({ name: "read", execute }));
const tool = wrapToolDefinition(
expectDefined(toToolDefinitions([source])[0], "policy-blocked private tool definition"),
);
const preparer = expectDefined(
getInternalToolExecutionPreparer(tool),
"policy private execution preparer",
);
const prepared = await preparer({ toolCallId: "call-policy", args: {} });
expect(prepared).toMatchObject({
kind: "immediate",
outcome: {
kind: "result",
isError: false,
result: { details: { status: "blocked", reason: "blocked by policy" } },
},
});
prepared.dispose();
expect(execute).not.toHaveBeenCalled();
});
it("commits final rewritten args immediately before private implementation", async () => {
beforeToolCallHook = installBeforeToolCallHook({
runBeforeToolCallImpl: async () => ({ params: { value: "rewritten" } }),
});
const order: string[] = [];
const execute = vi.fn(async () => {
order.push("body");
return { content: [], details: { ok: true } };
});
const source = wrapToolWithBeforeToolCallHook(asAgentTool({ name: "read", execute }));
const tool = wrapToolDefinition(
expectDefined(toToolDefinitions([source])[0], "rewritten private tool definition"),
);
const preparer = expectDefined(
getInternalToolExecutionPreparer(tool),
"rewritten private execution preparer",
);
const prepared = await preparer({
toolCallId: "call-rewritten",
args: { value: "original" },
});
expect(prepared.kind).toBe("ready");
if (prepared.kind !== "ready") {
return;
}
const onImplementationStart = vi.fn(() => {
order.push("commit");
queueMicrotask(() => order.push("gap"));
});
await prepared.execute(onImplementationStart);
prepared.dispose();
expect(prepared.args).toEqual({ value: "rewritten" });
expect(onImplementationStart).toHaveBeenCalledOnce();
expect(execute).toHaveBeenCalledWith(
"call-rewritten",
{ value: "rewritten" },
undefined,
undefined,
);
expect(order).toEqual(["commit", "body", "gap"]);
});
it("does not consume a voice grant when private execution is disposed", async () => {
const runId = "run-voice-private-dispose";
const toolParams = { action: "send", to: "target-a", message: "approved body" };
installVoiceRunBinding(runId);
approveVoiceToolParams(runId, toolParams);
const execute = vi.fn().mockResolvedValue({ content: [], details: { ok: true } });
const hookContext = { runId, agentId: "main", sessionKey: "agent:main:voice" };
const source = wrapToolWithBeforeToolCallHook(
asAgentTool({ name: "message", execute }),
hookContext,
);
const tool = wrapToolDefinition(
expectDefined(toToolDefinitions([source], hookContext)[0], "voice private tool definition"),
);
const preparer = expectDefined(
getInternalToolExecutionPreparer(tool),
"voice private execution preparer",
);
try {
const prepared = await preparer({ toolCallId: "call-voice-disposed", args: toolParams });
expect(prepared.kind).toBe("ready");
prepared.dispose();
prepared.dispose();
await Promise.resolve();
const first = await tool.execute("call-voice-retry", toolParams);
const second = await tool.execute("call-voice-consumed", toolParams);
expect(first.details).toEqual({ ok: true });
expect(second.details).toMatchObject({
status: "blocked",
deniedReason: "client-voice-confirmation",
});
expect(execute).toHaveBeenCalledOnce();
expect(consumeTrackedToolExecutionStarted("call-voice-disposed", runId)).toBeUndefined();
} finally {
resetClientVoiceConfirmationStateForTest();
vi.restoreAllMocks();
}
});
it.each(["adapter", "client"] as const)(
"suppresses the %s body after awaited hook preflight",
async (kind) => {
let releaseHook!: () => void;
let markHookStarted!: () => void;
const hookStarted = new Promise<void>((resolve) => {
markHookStarted = resolve;
});
const hookRelease = new Promise<void>((resolve) => {
releaseHook = resolve;
});
beforeToolCallHook = installBeforeToolCallHook({
runBeforeToolCallImpl: async () => {
markHookStarted();
await hookRelease;
return undefined;
},
});
const execute = vi.fn().mockResolvedValue({ content: [], details: { ok: true } });
const clientCall = vi.fn();
const definition =
kind === "adapter"
? expectDefined(
toToolDefinitions([asAgentTool({ name: "read", execute })])[0],
"unwrapped adapter definition",
)
: expectDefined(
toClientToolDefinitions(
[
{
type: "function",
function: {
name: "client_read",
description: "client read",
parameters: { type: "object", properties: {} },
},
},
],
clientCall,
)[0],
"client adapter definition",
);
const preparer = expectDefined(
getInternalToolExecutionPreparer(wrapToolDefinition(definition)),
`${kind} private execution preparer`,
);
const preparing = preparer({ toolCallId: `${kind}-call`, args: {} });
await hookStarted;
releaseHook();
const prepared = await preparing;
expect(prepared.kind).toBe("ready");
prepared.dispose();
await Promise.resolve();
expect(execute).not.toHaveBeenCalled();
expect(clientCall).not.toHaveBeenCalled();
},
);
it("finishes async reconciliation before exposing a wrapped call as ready", async () => {
beforeToolCallHook = installBeforeToolCallHook({
runBeforeToolCallImpl: async () => ({ params: { path: "/tmp/final" } }),
});
const runId = "run-reconcile-before-ready";
const sessionKey = "agent:main:reconcile-before-ready";
const state = getDiagnosticSessionState({ sessionKey, sessionId: "session-reconcile" });
recordToolCall(
state,
"read",
{ path: "/tmp/original" },
"reconcile-call",
{ enabled: true },
{ runId },
);
const execute = vi.fn().mockResolvedValue({ content: [], details: {} });
const hookContext = {
runId,
sessionKey,
sessionId: "session-reconcile",
loopDetection: { enabled: true },
};
const source = wrapToolWithBeforeToolCallHook(
asAgentTool({ name: "read", execute }),
hookContext,
);
const tool = wrapToolDefinition(
expectDefined(toToolDefinitions([source], hookContext)[0], "reconcile tool definition"),
);
const preparer = expectDefined(
getInternalToolExecutionPreparer(tool),
"reconcile private execution preparer",
);
const prepared = await preparer({
toolCallId: "reconcile-call",
args: { path: "/tmp/original" },
});
expect(prepared.kind).toBe("ready");
expect(state.toolCallHistory?.at(-1)?.argsHash).toBe(
hashToolCall("read", { path: "/tmp/final" }),
);
prepared.dispose();
expect(execute).not.toHaveBeenCalled();
});
it("passes agent context to outer code-mode exec hooks through OpenClaw custom tools", async () => {
beforeToolCallHook = installBeforeToolCallHook({
runBeforeToolCallImpl: async () => ({
@@ -102,6 +102,16 @@ export function consumeBatchAdmittedToolCall(toolCallId: string, runId?: string)
return admitted;
}
/** Release exact batch-admission markers for prepared calls suppressed by steering. */
export function releaseBatchAdmittedToolCalls(
toolCallIds: readonly string[],
runId?: string,
): void {
for (const toolCallId of toolCallIds) {
batchAdmittedToolCallIds.delete(buildAdjustedParamsKey({ runId, toolCallId }));
}
}
/** Remove unused batch-admission markers when their embedded run ends. */
export function clearBatchAdmittedToolCallsForRun(runId: string): void {
const prefix = `${runId}:`;
@@ -48,6 +48,10 @@ import type {
HookContext,
HookOutcome,
} from "./agent-tools.before-tool-call.types.js";
import {
createInternalExecutionPreparer,
readInternalExecutionControl,
} from "./agent-tools.execution-preparer.js";
import { validateToolExecutionParams } from "./agent-tools.execution-validation.js";
import {
BEFORE_TOOL_CALL_DIAGNOSTIC_OPTIONS,
@@ -62,6 +66,7 @@ import {
normalizeCodeModeExecBeforeHookParams,
reconcileCodeModeExecBeforeHookParams,
} from "./code-mode-control-tools.js";
import { attachInternalToolExecutionPreparer } from "./runtime/internal-hooks.js";
import { buildToolMutationState } from "./tool-mutation.js";
import { normalizeToolName } from "./tool-policy.js";
import {
@@ -78,6 +83,10 @@ type BeforeToolCallWrapperOptions = {
};
type ForwardedToolExecution = (...args: unknown[]) => ReturnType<AnyAgentTool["execute"]>;
const MAX_TRACKED_ADJUSTED_PARAMS = 1024;
const INTERNAL_DISPOSED_RESULT = {
content: [],
details: { status: "skipped", deniedReason: "internal-dispose" },
};
/** Run tool-owned preparation while retaining the exact prepared object. */
export async function prepareBeforeToolCallExecutionParams(params: {
@@ -283,6 +292,10 @@ export function wrapToolWithBeforeToolCallHook(
const wrappedTool: AnyAgentTool = {
...tool,
execute: async (toolCallId, params, signal, onUpdate, ...executionArgs: unknown[]) => {
const prepareControl = readInternalExecutionControl(executionArgs.at(-1));
if (prepareControl) {
executionArgs.pop();
}
const toolCallOrdinal = ctx?.allocateToolOutcomeOrdinal?.(toolCallId);
const preExecutionStartedAt = Date.now();
const normalizedToolName = normalizeToolName(toolName || "tool");
@@ -442,20 +455,6 @@ export function wrapToolWithBeforeToolCallHook(
adjustedParams: outcome.params,
finalizerMode: "wrapped",
});
// A voice grant binds the post-finalizer execution shape. Consuming it
// earlier would let later alias or tool-owned rewrites escape the grant.
const voiceConfirmation = consumeFinalClientVoiceToolConfirmation({
toolName,
params: executeParams,
ctx,
});
if (!voiceConfirmation.allowed) {
return await blockToolCall({
reason: voiceConfirmation.reason,
deniedReason: "client-voice-confirmation",
toolParams: executeParams,
});
}
// Hooks can repair or rewrite arguments; only the final execution
// shape is safe to validate, after vetoes but before side effects.
await validateToolExecutionParams(toolCallId, executeParams);
@@ -469,6 +468,29 @@ export function wrapToolWithBeforeToolCallHook(
recordPreExecutionError(error, outcome.params ?? hookParams, "tool_preparation");
throw tagBeforeToolCallFailure(error, signal);
}
let onImplementationStart: (() => void) | undefined;
if (prepareControl) {
const decision = await prepareControl.pause(executeParams);
if (!decision.launch) {
return INTERNAL_DISPOSED_RESULT;
}
onImplementationStart = decision.start;
}
// A voice grant binds the post-finalizer execution shape. Consume it only
// after steering can no longer suppress the prepared call.
const voiceConfirmation = consumeFinalClientVoiceToolConfirmation({
toolName,
params: executeParams,
ctx,
});
if (!voiceConfirmation.allowed) {
return await blockToolCall({
reason: voiceConfirmation.reason,
deniedReason: "client-voice-confirmation",
toolParams: executeParams,
});
}
onImplementationStart?.();
recordAdjustedParamsForToolCall(toolCallId, executeParams, ctx?.runId);
const eventBase = buildEventBase(executeParams);
recordToolExecutionStarted(toolCallId, ctx?.runId);
@@ -578,6 +600,23 @@ export function wrapToolWithBeforeToolCallHook(
},
};
const executeWithHooks = wrappedTool.execute;
const prepareExecution = createInternalExecutionPreparer(async (params, control) => {
recordToolExecutionTracked(params.toolCallId, ctx?.runId);
try {
return (await Reflect.apply(executeWithHooks, wrappedTool, [
params.toolCallId,
params.args,
params.signal,
params.onUpdate,
...(params.executionArgs ?? []),
control,
])) as Awaited<ReturnType<AnyAgentTool["execute"]>>;
} finally {
// Timeout observers may consume this while the call is still pending.
clearTrackedToolExecution(params.toolCallId, ctx?.runId);
}
});
attachInternalToolExecutionPreparer(wrappedTool, prepareExecution);
wrappedTool.execute = async (
toolCallId,
params,
@@ -585,20 +624,23 @@ export function wrapToolWithBeforeToolCallHook(
onUpdate,
...executionArgs: unknown[]
) => {
recordToolExecutionTracked(toolCallId, ctx?.runId);
const prepared = await prepareExecution({
toolCallId,
args: params,
signal,
onUpdate,
executionArgs,
});
try {
return await (executeWithHooks as ForwardedToolExecution)(
toolCallId,
params,
signal,
onUpdate,
...executionArgs,
);
if (prepared.kind === "immediate") {
if (prepared.outcome.kind === "error") {
throw prepared.outcome.error;
}
return prepared.outcome.result;
}
return await prepared.execute();
} finally {
// Timeout observers may consume this while the call is still pending. The
// wrapper owns final cleanup; every pre-body settle records the separate
// blocked fact, so direct callers cannot retain settled ids.
clearTrackedToolExecution(toolCallId, ctx?.runId);
prepared.dispose();
}
};
copyPluginToolMeta(tool, wrappedTool);
@@ -0,0 +1,88 @@
import type { AgentToolResult } from "./runtime/index.js";
import type { InternalToolExecutionPreparer } from "./runtime/internal-hooks.js";
const INTERNAL_EXECUTION_CONTROL = Symbol("openclawInternalExecutionControl");
type InternalExecutionControl = {
[INTERNAL_EXECUTION_CONTROL]: true;
ready: Promise<unknown>;
pause: (args: unknown) => Promise<{ launch: boolean; start?: () => void }>;
launch: (start?: () => void) => void;
dispose: () => void;
};
function createControl(): InternalExecutionControl {
let markReady!: (args: unknown) => void;
let decide!: (value: { launch: boolean; start?: () => void }) => void;
const ready = new Promise<unknown>((resolve) => {
markReady = resolve;
});
const decision = new Promise<{ launch: boolean; start?: () => void }>((resolve) => {
decide = resolve;
});
return {
[INTERNAL_EXECUTION_CONTROL]: true,
ready,
pause: (args) => {
markReady(args);
return decision;
},
launch: (start) => decide({ launch: true, start }),
dispose: () => decide({ launch: false }),
};
}
export function readInternalExecutionControl(value: unknown): InternalExecutionControl | undefined {
return value &&
typeof value === "object" &&
(value as Partial<InternalExecutionControl>)[INTERNAL_EXECUTION_CONTROL] === true
? (value as InternalExecutionControl)
: undefined;
}
export function createInternalExecutionPreparer(
startExecution: (
params: Parameters<InternalToolExecutionPreparer>[0],
control: InternalExecutionControl,
) => Promise<AgentToolResult<unknown>>,
): InternalToolExecutionPreparer {
return async (params) => {
const control = createControl();
const execution = startExecution(params, control);
const settled = await Promise.race([
control.ready.then((args) => ({ kind: "ready" as const, args })),
execution.then(
(result) => ({ kind: "result" as const, result }),
(error: unknown) => ({ kind: "error" as const, error }),
),
]);
if (settled.kind !== "ready") {
return {
kind: "immediate",
outcome:
settled.kind === "result"
? { kind: "result", result: settled.result, isError: false }
: { kind: "error", error: settled.error },
dispose() {},
};
}
let disposed = false;
return {
kind: "ready",
args: settled.args,
execute(start) {
if (!disposed) {
control.launch(start);
}
return execution;
},
dispose() {
if (!disposed) {
disposed = true;
control.dispose();
void execution.catch(() => undefined);
}
},
};
};
}
+68
View File
@@ -1,4 +1,5 @@
// Coverage for agent tool runtime execution and scoped authority.
import { expectDefined } from "@openclaw/normalization-core";
import { describe, expect, it, vi } from "vitest";
import "./test-helpers/fast-coding-tools.js";
import "./test-helpers/fast-openclaw-tools.js";
@@ -9,6 +10,10 @@ import {
runWithAgentRingZeroTools,
} from "./agent-tools.ring-zero-context.js";
import type { AnyAgentTool } from "./agent-tools.types.js";
import {
attachInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
} from "./runtime/internal-hooks.js";
import { stubTool } from "./test-helpers/fast-tool-stubs.js";
import {
getToolTerminalPresentation,
@@ -383,6 +388,69 @@ describe("wrapToolWithAbortSignal", () => {
expect(getToolTerminalPresentation(wrapped)).toBe(formatter);
});
it("does not enter private preparation when the run is already aborted", async () => {
const sourcePreparer = vi.fn(async () => ({
kind: "ready" as const,
args: {},
execute: vi.fn(async () => ({ content: [], details: {} })),
dispose: vi.fn(),
}));
const tool = attachInternalToolExecutionPreparer(
asAgentTool({ name: "prepared", execute: vi.fn() }),
sourcePreparer,
);
const runAbort = new AbortController();
runAbort.abort();
const wrapped = wrapToolWithAbortSignal(tool, runAbort.signal);
const preparer = expectDefined(
getInternalToolExecutionPreparer(wrapped),
"abort-adapted preparer",
);
await expect(preparer({ toolCallId: "already-aborted", args: {} })).rejects.toMatchObject({
name: "AbortError",
});
expect(sourcePreparer).not.toHaveBeenCalled();
});
it("disposes cancellation-ignoring private preparation after a later abort", async () => {
let release!: () => void;
let markStarted!: () => void;
const started = new Promise<void>((resolve) => {
markStarted = resolve;
});
const blocked = new Promise<void>((resolve) => {
release = resolve;
});
const dispose = vi.fn();
const body = vi.fn(async () => ({ content: [], details: {} }));
const sourcePreparer = vi.fn(async () => {
markStarted();
await blocked;
return { kind: "ready" as const, args: {}, execute: body, dispose };
});
const tool = attachInternalToolExecutionPreparer(
asAgentTool({ name: "prepared", execute: vi.fn() }),
sourcePreparer,
);
const runAbort = new AbortController();
const wrapped = wrapToolWithAbortSignal(tool, runAbort.signal);
const preparer = expectDefined(
getInternalToolExecutionPreparer(wrapped),
"abort-adapted preparer",
);
const preparing = preparer({ toolCallId: "later-aborted", args: {} });
await started;
runAbort.abort();
await expect(preparing).rejects.toMatchObject({ name: "AbortError" });
release();
await flushMicrotasks();
expect(dispose).toHaveBeenCalledOnce();
expect(body).not.toHaveBeenCalled();
});
});
vi.mock("./channel-tools.js", () => {
@@ -92,6 +92,30 @@ describe("installCodeModeRepairHook", () => {
});
});
it("leaves steering skips unchanged without spending repair", async () => {
const agent = createAgent();
const skipped = {
content: [{ type: "text" as const, text: "Skipped due to queued user message." }],
details: { status: "skipped", deniedReason: "steering" },
};
await expect(
agent.afterToolOutcome?.(
outcome({
result: skipped,
isError: true,
executionStarted: false,
}),
),
).resolves.toBeUndefined();
await expect(
agent.afterToolOutcome?.(outcome({ result: failedResult() })),
).resolves.toMatchObject({
terminate: false,
details: { repair: { allowed: true, remainingAttempts: 1 } },
});
});
it("does not spend the repair token on successful pure computation", async () => {
const agent = createAgent();
@@ -81,6 +81,16 @@ function isToolLoopRecoveryOutcome(context: AfterToolOutcomeContext): boolean {
return details.status === "blocked" && details.deniedReason === "tool-loop";
}
function isSteeringSkippedOutcome(context: AfterToolOutcomeContext): boolean {
const details = isRecord(context.result.details) ? context.result.details : {};
return (
context.isError &&
!context.executionStarted &&
details.status === "skipped" &&
details.deniedReason === "steering"
);
}
function preserveOriginalDispatchEvidence(
failure: CodeModeFailure | undefined,
original: CodeModeFailure | undefined,
@@ -235,6 +245,9 @@ export function installCodeModeRepairHook(params: { agent: Agent }): void {
if (isToolLoopRecoveryOutcome(context)) {
return prior;
}
if (isSteeringSkippedOutcome(context)) {
return prior;
}
if (signal?.aborted && !context.executionStarted) {
return prior;
}
@@ -1,3 +1,4 @@
import { expectDefined } from "@openclaw/normalization-core";
import { afterEach, describe, expect, it, vi } from "vitest";
import { getPluginToolMeta, setPluginToolMeta } from "../../../plugins/tools.js";
import {
@@ -7,6 +8,10 @@ import {
} from "../../before-tool-call-metadata.js";
import { getChannelAgentToolMeta, setChannelAgentToolMeta } from "../../channel-tool-metadata.js";
import { isCodeModeControlTool, markCodeModeControlTool } from "../../code-mode-control-tools.js";
import {
attachInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
} from "../../runtime/internal-hooks.js";
import {
getToolTerminalPresentation,
setToolTerminalPresentation,
@@ -170,4 +175,26 @@ describe("heartbeat wrapper metadata preservation", () => {
expect(isCodeModeControlTool(wrapped)).toBe(true);
});
it("applies heartbeat ownership to private preparation and execution", async () => {
const body = vi.fn(async () => ({ content: [], details: {} }));
const source = attachInternalToolExecutionPreparer(
{ name: "test-tool", execute: vi.fn() as never },
async () => ({ kind: "ready", args: {}, execute: body, dispose: vi.fn() }),
);
const wrapped = wrapEmbeddedAttemptToolWithActivity(source as never, RUN) as typeof source;
const preparer = expectDefined(
getInternalToolExecutionPreparer(wrapped),
"heartbeat-adapted preparer",
);
const prepared = await preparer({ toolCallId: "heartbeat-call", args: {} });
expect(prepared.kind).toBe("ready");
if (prepared.kind === "ready") {
await prepared.execute();
}
expect(body).toHaveBeenCalledOnce();
expect(getLastToolActivityMs(RUN)).toBeTypeOf("number");
});
});
@@ -5,6 +5,10 @@ import {
onToolActivity,
} from "../../../shared/tool-activity-heartbeat.js";
import { copyAgentToolMetadata } from "../../agent-tool-metadata.js";
import {
attachInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
} from "../../runtime/internal-hooks.js";
import type { AnyAgentTool } from "../../tools/common.js";
export { clearToolActivityRun, getLastToolActivityMs, notifyToolActivity, onToolActivity };
@@ -13,22 +17,36 @@ export function wrapEmbeddedAttemptToolWithActivity<T extends AnyAgentTool>(
tool: T,
runId: string,
): T {
const withActivity = async <R>(operation: () => Promise<R>): Promise<R> => {
const interval = setInterval(() => notifyToolActivity(runId), 60_000);
interval.unref?.();
try {
notifyToolActivity(runId);
return await operation();
} finally {
clearInterval(interval);
notifyToolActivity(runId);
}
};
const originalExecute = tool.execute;
const wrappedTool = {
...tool,
execute: (async (...args: Parameters<typeof originalExecute>) => {
// Long-running tools keep the attempt's idle watchdog alive.
const interval = setInterval(() => notifyToolActivity(runId), 60_000);
interval.unref?.();
try {
notifyToolActivity(runId);
return await originalExecute(...args);
} finally {
clearInterval(interval);
notifyToolActivity(runId);
}
}) as typeof originalExecute,
execute: ((...args: Parameters<typeof originalExecute>) =>
withActivity(() => originalExecute(...args))) as typeof originalExecute,
} as T;
// Tool metadata is identity-keyed, so object spread is insufficient.
return copyAgentToolMetadata(tool, wrappedTool);
copyAgentToolMetadata(tool, wrappedTool);
const sourcePreparer = getInternalToolExecutionPreparer(tool);
if (sourcePreparer) {
attachInternalToolExecutionPreparer(wrappedTool, async (params) => {
const prepared = await withActivity(() => sourcePreparer(params));
return prepared.kind === "ready"
? {
...prepared,
execute: (start) => withActivity(() => prepared.execute(start)),
}
: prepared;
});
}
return wrappedTool;
}
@@ -5,12 +5,34 @@ import { markCodeModeControlTool } from "../../code-mode-control-tools.js";
import type { AgentTool } from "../../runtime/index.js";
const mocks = vi.hoisted(() => ({
admitToolCallBatch: vi.fn(async (_calls: InternalToolBatchCall[]) => undefined),
attachedLifecycles: [] as Array<{
commitReadyCalls: (calls: readonly { toolCallId: string; args: unknown }[]) => void;
releaseSkippedCalls: (ids: readonly string[]) => void;
}>,
committedArgs: [] as unknown[],
releasedIds: [] as string[][],
admitToolCallBatch: vi.fn(async (_calls: InternalToolBatchCall[]) => ({
commitReadyCalls(readyCalls: readonly { toolCallId: string; args: unknown }[]) {
mocks.committedArgs.push(...readyCalls.map((call) => call.args));
},
releaseSkippedCalls(ids: readonly string[]) {
mocks.releasedIds.push([...ids]);
},
})),
}));
vi.mock("../../tool-loop-admission.js", () => ({
admitToolCallBatch: mocks.admitToolCallBatch,
}));
vi.mock("../../runtime/internal-hooks.js", () => ({
attachInternalToolBatchLifecycle: (
result: object,
lifecycle: (typeof mocks.attachedLifecycles)[number],
) => {
mocks.attachedLifecycles.push(lifecycle);
return result;
},
}));
import { createToolLoopBatchAdmission } from "./tool-loop-recovery.js";
@@ -34,6 +56,9 @@ function batchCall(id: string, args: Record<string, unknown>): InternalToolBatch
describe("tool-loop recovery batch admission", () => {
it("canonicalizes equivalent Code Mode exec aliases before loop detection", async () => {
mocks.committedArgs.length = 0;
mocks.releasedIds.length = 0;
mocks.attachedLifecycles.length = 0;
const admission = createToolLoopBatchAdmission({
sessionId: "session-1",
sessionKey: "agent:main:session-1",
@@ -44,7 +69,7 @@ describe("tool-loop recovery batch admission", () => {
throw new Error("Expected batch admission hook");
}
await admission({
const first = await admission({
assistantMessage: {
role: "assistant",
content: [],
@@ -65,7 +90,12 @@ describe("tool-loop recovery batch admission", () => {
calls: [batchCall("code-alias", { code: "return 1;" })],
context: { systemPrompt: "", messages: [] },
});
await admission({
const firstLifecycle = mocks.attachedLifecycles[0];
firstLifecycle?.commitReadyCalls([
{ toolCallId: "code-alias", args: { code: "return 1;", command: "return 1;" } },
]);
firstLifecycle?.releaseSkippedCalls([]);
const second = await admission({
assistantMessage: {
role: "assistant",
content: [],
@@ -86,11 +116,22 @@ describe("tool-loop recovery batch admission", () => {
calls: [batchCall("command-alias", { command: "return 1;" })],
context: { systemPrompt: "", messages: [] },
});
const secondLifecycle = mocks.attachedLifecycles[1];
secondLifecycle?.commitReadyCalls([
{ toolCallId: "command-alias", args: { command: "return 1;", code: "return 1;" } },
]);
secondLifecycle?.releaseSkippedCalls([]);
const admittedArgs = mocks.admitToolCallBatch.mock.calls.map(([calls]) => calls[0]?.args);
expect(admittedArgs).toEqual([
{ code: "return 1;", command: "return 1;" },
{ command: "return 1;", code: "return 1;" },
]);
expect(mocks.committedArgs).toEqual(admittedArgs);
expect(mocks.releasedIds).toEqual([[], []]);
expect(first).toEqual({});
expect(second).toEqual({});
expect(firstLifecycle?.commitReadyCalls).not.toBe(secondLifecycle?.commitReadyCalls);
expect(mocks.attachedLifecycles).toHaveLength(2);
});
});
@@ -2,7 +2,10 @@ import { clearBatchAdmittedToolCallsForRun } from "../../agent-tools.before-tool
import type { HookContext } from "../../agent-tools.before-tool-call.types.js";
import { normalizeCodeModeExecBeforeHookParams } from "../../code-mode-control-tools.js";
import type { Agent } from "../../runtime/index.js";
import type { InternalBeforeToolBatchHook } from "../../runtime/internal-hooks.js";
import {
attachInternalToolBatchLifecycle,
type InternalBeforeToolBatchHook,
} from "../../runtime/internal-hooks.js";
import { admitToolCallBatch } from "../../tool-loop-admission.js";
import { hashToolCall } from "../../tool-loop-detection.js";
import { log } from "../logger.js";
@@ -22,8 +25,14 @@ export function createToolLoopBatchAdmission(
: call.args,
}));
try {
const intervention = await admitToolCallBatch(canonicalCalls, ctx);
return intervention ? { intervention } : undefined;
const admission = await admitToolCallBatch(canonicalCalls, ctx);
const { commitReadyCalls, releaseSkippedCalls, ...result } = admission;
return commitReadyCalls && releaseSkippedCalls
? attachInternalToolBatchLifecycle(result, {
commitReadyCalls,
releaseSkippedCalls,
})
: result;
} catch (error) {
const first = canonicalCalls[0];
log.error(`tool-loop batch admission failed: ${String(error)}`);
+5
View File
@@ -1,4 +1,9 @@
export {
attachInternalToolBatchLifecycle,
attachInternalToolExecutionPreparer,
copyInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
setInternalBeforeToolBatch,
type InternalBeforeToolBatchHook,
type InternalToolExecutionPreparer,
} from "../../../packages/agent-core/src/internal-hooks.js";
@@ -324,8 +324,8 @@ export abstract class AgentSessionPrompting extends AgentSessionBase {
/**
* Queue a steering message while the agent is running.
* Delivered after the current assistant turn finishes executing its tool calls,
* before the next LLM call.
* Delivered before the next unstarted tool launch or model call. Running tools
* continue; suppressed calls receive paired synthetic results.
* Expands skill commands and prompt templates. Errors on extension commands.
* @param images Optional image attachments to include with the message
* @param userTurnTranscriptRecorder Prepared channel fields for transcript-only persistence
@@ -6,6 +6,7 @@
import type { TSchema } from "typebox";
import { copyCodeModeControlToolIdentity } from "../../code-mode-control-tools.js";
import type { AgentTool } from "../../runtime/index.js";
import { copyInternalToolExecutionPreparer } from "../../runtime/internal-hooks.js";
import type { ExtensionContext, ToolDefinition } from "../extensions/types.js";
/** Wrap a ToolDefinition into an AgentTool for the core runtime. */
@@ -33,7 +34,7 @@ export function wrapToolDefinition<
definition.execute(toolCallId, params, signal, onUpdate, ctxFactory?.() as ExtensionContext),
};
copyCodeModeControlToolIdentity(definition, tool);
return tool;
return copyInternalToolExecutionPreparer(definition, tool);
}
/** Wrap multiple ToolDefinitions into AgentTools for the core runtime. */
@@ -65,5 +66,5 @@ export function createToolDefinitionFromAgentTool(tool: AgentTool): ToolDefiniti
tool.execute(toolCallId, params, signal, onUpdate),
};
copyCodeModeControlToolIdentity(tool, definition);
return definition;
return copyInternalToolExecutionPreparer(tool, definition);
}
+102 -21
View File
@@ -1,5 +1,5 @@
import { beforeEach, describe, expect, it } from "vitest";
import { resetDiagnosticEventsForTest } from "../infra/diagnostic-events.js";
import { onDiagnosticEvent, resetDiagnosticEventsForTest } from "../infra/diagnostic-events.js";
import {
getDiagnosticSessionState,
resetDiagnosticSessionStateForTest,
@@ -63,17 +63,19 @@ describe("whole-batch tool-loop admission", () => {
const unrelatedSiblings = Array.from({ length: 20 }, (_, index) =>
call(`safe-sibling-${index}`, "write", {}),
);
const intervention = await admitToolCallBatch(
const admission = await admitToolCallBatch(
[...unrelatedSiblings, call("repeated", "process", pollArgs)],
ctx,
);
expect(intervention).toMatchObject({
kind: "critical-tool-loop",
toolCallId: "repeated",
toolName: "process",
detector: "known_poll_no_progress",
count: 20,
expect(admission).toMatchObject({
intervention: {
kind: "critical-tool-loop",
toolCallId: "repeated",
toolName: "process",
detector: "known_poll_no_progress",
count: 20,
},
});
expect(state.toolCallHistory).toHaveLength(21);
expect(state.toolCallHistory?.at(-1)).toMatchObject({
@@ -81,9 +83,12 @@ describe("whole-batch tool-loop admission", () => {
outcomeKind: "tool-loop-veto",
});
expect(consumeBatchAdmittedToolCall("safe-sibling-0", ctx.runId)).toBe(false);
await expect(
admitToolCallBatch([call("recovery-write", "write", {})], ctx),
).resolves.toBeUndefined();
await expect(admitToolCallBatch([call("recovery-write", "write", {})], ctx)).resolves.toEqual(
expect.objectContaining({
commitReadyCalls: expect.any(Function),
releaseSkippedCalls: expect.any(Function),
}),
);
});
it("blocks a batch that crosses the critical threshold within its own candidates", async () => {
@@ -110,32 +115,37 @@ describe("whole-batch tool-loop admission", () => {
});
}
const intervention = await admitToolCallBatch(
const admission = await admitToolCallBatch(
[call("candidate-20", "process", pollArgs), call("candidate-21", "process", pollArgs)],
ctx,
);
expect(intervention).toMatchObject({
kind: "critical-tool-loop",
toolCallId: "candidate-21",
detector: "known_poll_no_progress",
count: 20,
expect(admission).toMatchObject({
intervention: {
kind: "critical-tool-loop",
toolCallId: "candidate-21",
detector: "known_poll_no_progress",
count: 20,
},
});
expect(state.toolCallHistory).toHaveLength(21);
expect(consumeBatchAdmittedToolCall("candidate-20", ctx.runId)).toBe(false);
await expect(
admitToolCallBatch([call("recovery-repeat", "process", pollArgs)], ctx),
).resolves.toMatchObject({
kind: "critical-tool-loop",
toolCallId: "recovery-repeat",
detector: "known_poll_no_progress",
intervention: {
kind: "critical-tool-loop",
toolCallId: "recovery-repeat",
detector: "known_poll_no_progress",
},
});
});
it("records an admitted call once and skips only its duplicate single-call loop policy", async () => {
const admitted = call("admitted", "read", { path: "/tmp/a" });
await expect(admitToolCallBatch([admitted], ctx)).resolves.toBeUndefined();
const admission = await admitToolCallBatch([admitted], ctx);
admission.commitReadyCalls?.([{ toolCallId: admitted.toolCall.id, args: admitted.args }]);
await expect(
runBeforeToolCallHook({
toolName: admitted.toolCall.name,
@@ -161,4 +171,75 @@ describe("whole-batch tool-loop admission", () => {
expect(consumeBatchAdmittedToolCall(admitted.toolCall.id, ctx.runId)).toBe(false);
});
it("releases repeated skipped admissions without mutating bounded history", async () => {
const state = getDiagnosticSessionState({
sessionKey: ctx.sessionKey,
sessionId: ctx.sessionId,
});
for (let index = 0; index < 30; index += 1) {
recordToolCall(state, "read", { path: `/tmp/prior-${index}` }, `prior-${index}`);
}
const originalHistory = [...(state.toolCallHistory ?? [])];
const diagnosticEvents: unknown[] = [];
const unsubscribe = onDiagnosticEvent((event) => diagnosticEvents.push(event));
try {
for (let index = 0; index < 25; index += 1) {
const skipped = call(`skipped-${index}`, "write", { path: "/tmp/skipped" });
const admission = await admitToolCallBatch([skipped], ctx);
admission.releaseSkippedCalls?.([skipped.toolCall.id]);
expect(consumeBatchAdmittedToolCall(skipped.toolCall.id, ctx.runId)).toBe(false);
}
} finally {
unsubscribe();
}
expect(state.toolCallHistory).toEqual(originalHistory);
expect(diagnosticEvents).toEqual([]);
const executed = call("executed", "write", { path: "/tmp/skipped" });
const admission = await admitToolCallBatch([executed], ctx);
admission.commitReadyCalls?.([{ toolCallId: executed.toolCall.id, args: executed.args }]);
admission.releaseSkippedCalls?.([]);
expect(state.toolCallHistory).toHaveLength(30);
expect(state.toolCallHistory?.at(-1)).toMatchObject({
runId: ctx.runId,
toolCallId: executed.toolCall.id,
toolName: executed.toolCall.name,
});
expect(consumeBatchAdmittedToolCall(executed.toolCall.id, ctx.runId)).toBe(true);
expect(consumeBatchAdmittedToolCall(executed.toolCall.id, ctx.runId)).toBe(false);
});
it("commits ready siblings in assistant order and releases exact run markers", async () => {
const otherRun = { ...ctx, runId: "run-2" };
const sharedId = "shared-call";
const first = call("first", "read", { path: "/tmp/first" });
const skipped = call(sharedId, "write", { path: "/tmp/skipped" });
const last = call("last", "read", { path: "/tmp/last" });
const otherAdmission = await admitToolCallBatch(
[call(sharedId, "read", { path: "/tmp/other" })],
otherRun,
);
const admission = await admitToolCallBatch([first, skipped, last], ctx);
admission.commitReadyCalls?.([
{ toolCallId: last.toolCall.id, args: last.args },
{ toolCallId: first.toolCall.id, args: first.args },
]);
admission.releaseSkippedCalls?.([skipped.toolCall.id]);
const state = getDiagnosticSessionState({
sessionKey: ctx.sessionKey,
sessionId: ctx.sessionId,
});
expect(state.toolCallHistory?.slice(-2).map((record) => record.toolCallId)).toEqual([
first.toolCall.id,
last.toolCall.id,
]);
expect(consumeBatchAdmittedToolCall(skipped.toolCall.id, ctx.runId)).toBe(false);
expect(consumeBatchAdmittedToolCall(sharedId, otherRun.runId)).toBe(true);
otherAdmission.releaseSkippedCalls?.([sharedId]);
});
});
+82 -17
View File
@@ -1,11 +1,18 @@
import type { InternalToolBatchCall, ToolLoopIntervention } from "@openclaw/agent-core";
import type {
InternalBeforeToolBatchResult,
InternalToolBatchCall,
ToolLoopIntervention,
} from "@openclaw/agent-core";
import type { SessionState } from "../logging/diagnostic-session-state.js";
import {
beforeToolCallLog as log,
loadBeforeToolCallRuntime,
shouldEmitLoopWarning,
} from "./agent-tools.before-tool-call.diagnostics.js";
import { recordBatchAdmittedToolCall } from "./agent-tools.before-tool-call.state.js";
import {
recordBatchAdmittedToolCall,
releaseBatchAdmittedToolCalls,
} from "./agent-tools.before-tool-call.state.js";
import type { HookContext } from "./agent-tools.before-tool-call.types.js";
import { hashToolCall } from "./tool-loop-detection.js";
import { normalizeToolName } from "./tool-policy.js";
@@ -16,6 +23,11 @@ type ToolLoopCall = {
toolCallId?: string;
};
type ToolLoopBatchAdmission = InternalBeforeToolBatchResult & {
commitReadyCalls?: (calls: readonly { toolCallId: string; args: unknown }[]) => void;
releaseSkippedCalls?: (toolCallIds: readonly string[]) => void;
};
async function evaluateToolLoopCall(
call: ToolLoopCall,
ctx: HookContext,
@@ -113,17 +125,25 @@ export async function admitSingleToolCallLoop(
}
/**
* Admit an assistant tool batch atomically. Calls are only recorded after every
* sibling passes detection, so no side effect can start before a later veto.
* Admit an assistant tool batch atomically. Successful calls reserve exact
* markers here, then agent-core commits their history in assistant order at
* the final launch boundary. A later veto still records only denial evidence.
*/
export async function admitToolCallBatch(
calls: InternalToolBatchCall[],
ctx: HookContext,
): Promise<ToolLoopIntervention | undefined> {
): Promise<ToolLoopBatchAdmission> {
if (!ctx.sessionKey || ctx.loopDetection?.enabled !== true) {
return undefined;
return {};
}
const { getDiagnosticSessionState, recordToolCall } = await loadBeforeToolCallRuntime();
const {
getDiagnosticSessionState,
markDiagnosticArgumentChurnObservation,
reconcileToolCallExecutionParams,
recordToolCall,
resolveToolLoopWarningThreshold,
} = await loadBeforeToolCallRuntime();
const warningThreshold = resolveToolLoopWarningThreshold();
const sessionState = getDiagnosticSessionState({
sessionKey: ctx.sessionKey,
sessionId: ctx.sessionId,
@@ -185,21 +205,66 @@ export async function admitToolCallBatch(
recordLoopVeto(sessionState, rejectedCall);
}
}
return intervention;
return { intervention };
}
// A later sibling must assume this candidate makes no progress.
projectLoopVeto(call);
}
for (const call of calls) {
await recordToolLoopCall(
{
toolName: call.toolCall.name,
params: call.args,
toolCallId: call.toolCall.id,
},
ctx,
);
recordBatchAdmittedToolCall(call.toolCall.id, ctx.runId);
}
return undefined;
const admittedById = new Map(
calls.map((call) => [
call.toolCall.id,
{ toolName: normalizeToolName(call.toolCall.name || "tool") },
]),
);
const committedIds = new Set<string>();
const commitReadyCall = (readyCall: { toolCallId: string; args: unknown }) => {
const admitted = admittedById.get(readyCall.toolCallId);
if (!admitted || committedIds.has(readyCall.toolCallId)) {
return;
}
recordToolCall(
sessionState,
admitted.toolName,
readyCall.args,
readyCall.toolCallId,
ctx.loopDetection,
ctx.runId ? { runId: ctx.runId } : undefined,
);
const churn = reconcileToolCallExecutionParams(sessionState, {
toolName: admitted.toolName,
toolParams: readyCall.args,
toolCallId: readyCall.toolCallId,
runId: ctx.runId,
warningThreshold,
});
markDiagnosticArgumentChurnObservation({
sessionKey: ctx.sessionKey,
sessionId: ctx.sessionId,
runId: ctx.runId,
active: churn.active,
});
committedIds.add(readyCall.toolCallId);
};
return {
commitReadyCalls(readyCalls) {
if (readyCalls.length === 1 && readyCalls[0]) {
commitReadyCall(readyCalls[0]);
return;
}
const readyById = new Map(readyCalls.map((call) => [call.toolCallId, call]));
for (const call of calls) {
const readyCall = readyById.get(call.toolCall.id);
if (readyCall) {
commitReadyCall(readyCall);
}
}
},
releaseSkippedCalls(toolCallIds) {
// Agent-core only supplies admitted prepared calls suppressed at a steering checkpoint.
releaseBatchAdmittedToolCalls(toolCallIds, ctx.runId);
},
};
}
@@ -1,3 +1,4 @@
import { expectDefined } from "@openclaw/normalization-core";
import { Type } from "typebox";
import { describe, expect, it, vi } from "vitest";
import { getPluginToolMeta, setPluginToolMeta } from "../../plugins/tools.js";
@@ -6,12 +7,19 @@ import {
wrapToolWithBeforeToolCallHook,
} from "../agent-tools.before-tool-call.js";
import { getChannelAgentToolMeta, setChannelAgentToolMeta } from "../channel-tool-metadata.js";
import {
attachInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
} from "../runtime/internal-hooks.js";
import {
getToolTerminalPresentation,
setToolTerminalPresentation,
} from "../tool-terminal-presentation.js";
import type { AnyAgentTool } from "./common.js";
import { wrapToolWithGatewayCallerIdentity } from "./gateway-caller-context.js";
import {
getGatewayToolCallerIdentity,
wrapToolWithGatewayCallerIdentity,
} from "./gateway-caller-context.js";
describe("gateway caller context wrapper", () => {
it("preserves tool metadata used by policy and presentation layers", () => {
@@ -40,4 +48,43 @@ describe("gateway caller context wrapper", () => {
expect(getToolTerminalPresentation(wrapped)).toBe(getToolTerminalPresentation(tool));
expect(isToolWrappedWithBeforeToolCallHook(wrapped)).toBe(true);
});
it("applies caller identity to private preparation and execution", async () => {
const seen: unknown[] = [];
const tool = attachInternalToolExecutionPreparer(
{
name: "plugin_tool",
label: "Plugin tool",
description: "plugin tool",
parameters: Type.Object({}),
execute: vi.fn(async () => ({ content: [], details: {} })),
},
async () => {
seen.push(getGatewayToolCallerIdentity());
return {
kind: "ready",
args: {},
execute: async () => {
seen.push(getGatewayToolCallerIdentity());
return { content: [], details: {} };
},
dispose: vi.fn(),
};
},
);
const identity = { agentId: "agent-a", sessionKey: "agent-a:session" };
const wrapped = wrapToolWithGatewayCallerIdentity(tool as never, identity);
const preparer = expectDefined(
getInternalToolExecutionPreparer(wrapped),
"gateway-adapted preparer",
);
const prepared = await preparer({ toolCallId: "gateway-call", args: {} });
expect(prepared.kind).toBe("ready");
if (prepared.kind === "ready") {
await prepared.execute();
}
expect(seen).toEqual([identity, identity]);
});
});
+19 -1
View File
@@ -1,6 +1,10 @@
// Ambient trusted caller context for model-mediated Gateway tool calls.
import { AsyncLocalStorage } from "node:async_hooks";
import { copyAgentToolMetadata } from "../agent-tool-metadata.js";
import {
attachInternalToolExecutionPreparer,
getInternalToolExecutionPreparer,
} from "../runtime/internal-hooks.js";
import type { AnyAgentTool } from "./common.js";
type GatewayToolCallerIdentity = {
@@ -73,7 +77,21 @@ export function wrapToolWithGatewayCallerIdentity(
execute: async (...args) =>
await withGatewayToolCallerIdentity(identity, async () => await tool.execute?.(...args)),
};
return copyAgentToolMetadata(tool, wrapped);
copyAgentToolMetadata(tool, wrapped);
const sourcePreparer = getInternalToolExecutionPreparer(tool);
if (sourcePreparer) {
attachInternalToolExecutionPreparer(wrapped, async (params) => {
const prepared = await withGatewayToolCallerIdentity(identity, () => sourcePreparer(params));
return prepared.kind === "ready"
? {
...prepared,
execute: (start) =>
withGatewayToolCallerIdentity(identity, () => prepared.execute(start)),
}
: prepared;
});
}
return wrapped;
}
export function createGatewayToolCallerWrapper(
+254 -7
View File
@@ -1,4 +1,4 @@
import { mkdtemp, rm } from "node:fs/promises";
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import { tmpdir } from "node:os";
import path from "node:path";
@@ -14,7 +14,7 @@ import {
type OpenClawTestInstance,
} from "./helpers/openclaw-test-instance.js";
type FirstResponseKind = "final" | "tool";
type FirstResponseKind = "final" | "sequential-tools" | "tool";
type ModelRequest = { body: Record<string, unknown> };
type MockModelServer = {
baseUrl: string;
@@ -36,10 +36,20 @@ type GatewayFixture = {
chatErrors: Array<{ errorMessage?: string; runId?: string; state: "error" }>;
chatFinalRunIds: string[];
sessionKey: string;
steeringTools?: SteeringToolsFixture;
};
type SteeringToolsFixture = {
pluginDir: string;
releasePath: string;
tracePath: string;
};
const TEST_TIMEOUT_MS = 180_000;
const WAIT_OPTS = { timeout: 30_000, interval: 20 } as const;
const STEERING_PLUGIN_ID = "gateway-steering-tools";
const STEERING_GATE_TOOL = "steering_gate";
const STEERING_TAIL_TOOL = "steering_tail";
const instances: OpenClawTestInstance[] = [];
const clients: GatewayChatClient[] = [];
const diagnosticsClients: GatewayClient[] = [];
@@ -180,6 +190,52 @@ function writeToolResponse(res: ServerResponse): void {
]);
}
function writeSequentialToolsResponse(res: ServerResponse): void {
const items = [
{
type: "function_call",
id: "fc_steering_gate",
call_id: "call_steering_gate",
name: STEERING_GATE_TOOL,
arguments: "{}",
status: "completed",
},
{
type: "function_call",
id: "fc_steering_tail",
call_id: "call_steering_tail",
name: STEERING_TAIL_TOOL,
arguments: "{}",
status: "completed",
},
];
writeSse(res, [
...items.flatMap((item, outputIndex) => [
{
type: "response.output_item.added",
output_index: outputIndex,
item: { ...item, status: "in_progress", arguments: "" },
},
{
type: "response.function_call_arguments.done",
item_id: item.id,
output_index: outputIndex,
arguments: item.arguments,
},
{ type: "response.output_item.done", output_index: outputIndex, item },
]),
{
type: "response.completed",
response: {
id: "resp_steer_fifo_sequential_tools",
status: "completed",
output: items,
usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 },
},
},
]);
}
async function startMockModelServer(): Promise<MockModelServer> {
const requests: ModelRequest[] = [];
const firstResponse = createDeferred();
@@ -207,6 +263,10 @@ async function startMockModelServer(): Promise<MockModelServer> {
writeToolResponse(res);
return;
}
if (firstResponseKind === "sequential-tools") {
writeSequentialToolsResponse(res);
return;
}
}
writeTextResponse(res, requestIndex);
})().catch((error: unknown) => {
@@ -246,16 +306,113 @@ async function startMockModelServer(): Promise<MockModelServer> {
};
}
async function writeSteeringToolsPlugin(fixtureDir: string): Promise<SteeringToolsFixture> {
const pluginDir = path.join(fixtureDir, "steering-tools-plugin");
const releasePath = path.join(fixtureDir, "steering-gate.release");
const tracePath = path.join(fixtureDir, "steering-tools.trace");
await mkdir(pluginDir, { recursive: true });
await Promise.all([
writeFile(
path.join(pluginDir, "openclaw.plugin.json"),
`${JSON.stringify({
id: STEERING_PLUGIN_ID,
name: "Gateway Steering Tools",
activation: { onStartup: true },
contracts: { tools: [STEERING_GATE_TOOL, STEERING_TAIL_TOOL] },
configSchema: { type: "object", additionalProperties: false, properties: {} },
})}\n`,
"utf8",
),
writeFile(
path.join(pluginDir, "index.mjs"),
[
'import { access, appendFile } from "node:fs/promises";',
"const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));",
"async function waitForRelease() {",
" const deadline = Date.now() + 15_000;",
" while (Date.now() < deadline) {",
" try {",
` await access(${JSON.stringify(releasePath)});`,
" return;",
" } catch (error) {",
' if (error?.code !== "ENOENT") throw error;',
" }",
" await sleep(20);",
" }",
' throw new Error("steering gate release timed out");',
"}",
"export default {",
` id: ${JSON.stringify(STEERING_PLUGIN_ID)},`,
" register(api) {",
' api.on("before_tool_call", async (event) => {',
` if (event.toolName !== ${JSON.stringify(STEERING_GATE_TOOL)}) return;`,
` await appendFile(${JSON.stringify(tracePath)}, "preflight-start\\n", "utf8");`,
" await waitForRelease();",
` await appendFile(${JSON.stringify(tracePath)}, "preflight-end\\n", "utf8");`,
" });",
" api.registerTool({",
` name: ${JSON.stringify(STEERING_GATE_TOOL)},`,
' label: "Steering Gate",',
' description: "Wait for the steering gateway test release file.",',
' parameters: { type: "object", properties: {}, additionalProperties: false },',
' executionMode: "sequential",',
" async execute() {",
` await appendFile(${JSON.stringify(tracePath)}, "gate-executed\\n", "utf8");`,
' return { content: [{ type: "text", text: "steering gate completed" }], details: {} };',
" },",
" });",
" api.registerTool({",
` name: ${JSON.stringify(STEERING_TAIL_TOOL)},`,
' label: "Steering Tail",',
' description: "Record if the steering tail executes unexpectedly.",',
' parameters: { type: "object", properties: {}, additionalProperties: false },',
' executionMode: "sequential",',
" async execute() {",
` await appendFile(${JSON.stringify(tracePath)}, "tail-executed\\n", "utf8");`,
' return { content: [{ type: "text", text: "steering tail executed" }], details: {} };',
" },",
" });",
" },",
"};",
"",
].join("\n"),
"utf8",
),
]);
return { pluginDir, releasePath, tracePath };
}
async function readTrace(tracePath: string): Promise<string[]> {
try {
return (await readFile(tracePath, "utf8")).split("\n").filter(Boolean);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
return [];
}
throw error;
}
}
function createConfig(params: {
fixtureDir: string;
modelServer: MockModelServer;
steeringTools?: SteeringToolsFixture;
}): OpenClawConfig {
const provider = buildMockOpenAiResponsesProvider(
`${params.modelServer.baseUrl}/v1`,
"steer-fifo",
);
const steeringTools = params.steeringTools;
return {
plugins: { slots: { memory: "none" } },
plugins: steeringTools
? {
enabled: true,
allow: [STEERING_PLUGIN_ID],
load: { paths: [steeringTools.pluginDir] },
entries: { [STEERING_PLUGIN_ID]: { enabled: true } },
slots: { memory: "none" },
}
: { slots: { memory: "none" } },
agents: {
defaults: {
workspace: path.join(params.fixtureDir, "workspace"),
@@ -269,9 +426,13 @@ function createConfig(params: {
skills: [],
skipBootstrap: true,
},
list: [{ id: "main", default: true, model: { primary: provider.modelRef }, skills: [] }],
entries: {
main: { default: true, model: { primary: provider.modelRef }, skills: [] },
},
},
tools: { profile: "minimal" },
tools: steeringTools
? { profile: "minimal", alsoAllow: [STEERING_GATE_TOOL, STEERING_TAIL_TOOL] }
: { profile: "minimal" },
models: {
mode: "replace",
providers: {
@@ -319,15 +480,21 @@ async function connectDiagnosticsClient(instance: OpenClawTestInstance): Promise
return client;
}
async function createGatewayFixture(name: string): Promise<GatewayFixture> {
async function createGatewayFixture(
name: string,
options: { withSteeringTools?: boolean } = {},
): Promise<GatewayFixture> {
const fixtureDir = await mkdtemp(path.join(tmpdir(), `openclaw-${name}-`));
cleanupDirs.push(fixtureDir);
const steeringTools = options.withSteeringTools
? await writeSteeringToolsPlugin(fixtureDir)
: undefined;
const modelServer = await startMockModelServer();
modelServers.push(modelServer);
const instance = await createOpenClawTestInstance({
name,
gatewayToken: "steer-fifo-token",
config: createConfig({ fixtureDir, modelServer }),
config: createConfig({ fixtureDir, modelServer, steeringTools }),
env: {
OPENCLAW_LOG_LEVEL: "debug",
OPENCLAW_SKIP_PROVIDERS: undefined,
@@ -377,6 +544,7 @@ async function createGatewayFixture(name: string): Promise<GatewayFixture> {
chatErrors,
chatFinalRunIds,
sessionKey: `agent:main:${name}`,
...(steeringTools ? { steeringTools } : {}),
};
}
@@ -513,6 +681,16 @@ function contentText(content: unknown): string {
.join("\n");
}
function responseInputItems(request: ModelRequest | undefined): Array<Record<string, unknown>> {
const input = request?.body.input;
return Array.isArray(input)
? input.filter(
(item): item is Record<string, unknown> =>
item !== null && typeof item === "object" && !Array.isArray(item),
)
: [];
}
function userInputs(request: ModelRequest | undefined): string[] {
const input = request?.body.input;
if (typeof input === "string") {
@@ -596,6 +774,75 @@ describe("Gateway steer FIFO", () => {
TEST_TIMEOUT_MS,
);
it(
"suppresses sequential tools when a Gateway steer arrives during preflight",
async () => {
const fixture = await createGatewayFixture("steer-sequential-tail", {
withSteeringTools: true,
});
const steeringTools = fixture.steeringTools;
if (!steeringTools) {
throw new Error("steering tool fixture was not configured");
}
const first = await sendHeldTurn(fixture);
const steerMarker = "STEER_DURING_SEQUENTIAL_GATE";
try {
fixture.modelServer.releaseFirst("sequential-tools");
await vi.waitFor(
async () => expect(await readTrace(steeringTools.tracePath)).toEqual(["preflight-start"]),
WAIT_OPTS,
);
await queueSteer(fixture, steerMarker);
} finally {
await writeFile(steeringTools.releasePath, "release\n", "utf8");
}
await vi.waitFor(() => expect(fixture.modelServer.requests).toHaveLength(2), WAIT_OPTS);
await waitForRunTerminal(fixture, first.runId);
await vi.waitFor(
async () =>
expect(await readTrace(steeringTools.tracePath)).toEqual([
"preflight-start",
"preflight-end",
]),
WAIT_OPTS,
);
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
const nextRequest = fixture.modelServer.requests[1];
const inputItems = responseInputItems(nextRequest);
const gateOutputIndex = inputItems.findIndex(
(item) => item.type === "function_call_output" && item.call_id === "call_steering_gate",
);
const tailOutputIndex = inputItems.findIndex(
(item) => item.type === "function_call_output" && item.call_id === "call_steering_tail",
);
const steerIndex = inputItems.findIndex(
(item) => item.role === "user" && contentText(item.content).includes(steerMarker),
);
expect(gateOutputIndex).toBeGreaterThanOrEqual(0);
expect(tailOutputIndex).toBeGreaterThan(gateOutputIndex);
expect(steerIndex).toBeGreaterThan(tailOutputIndex);
expect(contentText(inputItems[gateOutputIndex]?.output)).toContain(
"Skipped due to queued user message.",
);
expect(contentText(inputItems[tailOutputIndex]?.output)).toContain(
"Skipped due to queued user message.",
);
expect(await readTrace(steeringTools.tracePath)).toEqual([
"preflight-start",
"preflight-end",
]);
expect(fixture.modelServer.requests).toHaveLength(2);
expect(fixture.chatErrors).toEqual([]);
},
TEST_TIMEOUT_MS,
);
it(
"consumes a steer at a tool control point without a fallback turn",
async () => {