mirror of
https://github.com/agentclientprotocol/claude-agent-acp.git
synced 2026-10-27 04:30:59 +00:00
fix: Refine tool calls from streamed input (#867)
## Why Tool-use `content_block_start` events arrive before their JSON input. This leaves pending calls with generic labels such as `Read File` or `Terminal`; long-running and concurrent calls can remain indistinguishable in clients while they execute. Anthropic specifies `input_json_delta.partial_json` as arbitrary string fragments, so delta boundaries cannot be assumed to align with fields or JSON values. ## What changed - Buffer `input_json_delta` fragments per parent message and content-block index. - Recover completed top-level fields while the overall JSON is still incomplete. - Refine the existing ACP tool call immediately when distinguishing fields such as `file_path`, `command`, `query`, `pattern`, `url`, or `description` become available. - Continue refining as later fields arrive, while keeping the early pending lifecycle and cleaning incomplete buffers at block/message boundaries. - Avoid publishing ambiguous numeric suffixes (`10` may still become `100` or `10.5`) until a delimiter arrives. ## Test coverage Parameterized coverage now includes Agent/Task, Bash, Read/Write/Edit, Glob/Grep, WebFetch/WebSearch, ReportFindings, ExitPlanMode, AskUserQuestion, Other, custom MCP calls, TodoWrite plan updates, and intentionally suppressed TaskCreate/Update/List/Get calls. Edge cases cover arbitrary field splits, unfinished strings, escaped quotes and commas, numbers, booleans, null, arrays, nested objects, and all three tool block types (`tool_use`, `server_tool_use`, `mcp_tool_use`). ## Validation - `npm run build` - `npm run test:run` (535 passed, 20 skipped) - `npm run check` --------- Co-authored-by: Ben Brandt <benjamin.j.brandt@gmail.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
nikita-ashihmin
240
src/acp-agent.ts
240
src/acp-agent.ts
@@ -510,6 +510,83 @@ export type ToolUseCache = {
|
||||
};
|
||||
};
|
||||
|
||||
type StreamedToolInput = {
|
||||
id: string;
|
||||
name: string;
|
||||
partialJson: string;
|
||||
/** Offset into `partialJson` the scanner has consumed; each delta only scans
|
||||
* the newly appended fragment, so total scan work stays linear. */
|
||||
scannedTo: number;
|
||||
inString: boolean;
|
||||
escaped: boolean;
|
||||
objectDepth: number;
|
||||
arrayDepth: number;
|
||||
/** Offset of the most recent comma at the top level of the input object
|
||||
* (-1 before the first). Everything before it is a complete field. */
|
||||
lastTopLevelComma: number;
|
||||
/** The comma offset the last emitted refinement was sliced at (-1 before the
|
||||
* first), so a field boundary only triggers one recovery attempt. */
|
||||
emittedThroughComma: number;
|
||||
};
|
||||
|
||||
export type StreamedToolInputCache = Map<string, Map<number, StreamedToolInput>>;
|
||||
|
||||
/**
|
||||
* Advance the lexer state across the fragment appended since the last delta:
|
||||
* just enough JSON awareness (string/escape, nesting depth) to spot commas
|
||||
* that sit at the top level of the input object — everything before such a
|
||||
* comma is a set of complete fields. Returns true once the input object's
|
||||
* closing brace arrives.
|
||||
*/
|
||||
function scanStreamedToolInput(state: StreamedToolInput): boolean {
|
||||
let complete = false;
|
||||
for (let index = state.scannedTo; index < state.partialJson.length; index++) {
|
||||
const character = state.partialJson[index];
|
||||
if (state.inString) {
|
||||
if (state.escaped) {
|
||||
state.escaped = false;
|
||||
} else if (character === "\\") {
|
||||
state.escaped = true;
|
||||
} else if (character === '"') {
|
||||
state.inString = false;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (character === '"') {
|
||||
state.inString = true;
|
||||
} else if (character === "{") {
|
||||
state.objectDepth++;
|
||||
} else if (character === "}") {
|
||||
state.objectDepth--;
|
||||
if (state.objectDepth === 0) {
|
||||
complete = true;
|
||||
}
|
||||
} else if (character === "[") {
|
||||
state.arrayDepth++;
|
||||
} else if (character === "]") {
|
||||
state.arrayDepth--;
|
||||
} else if (character === "," && state.objectDepth === 1 && state.arrayDepth === 0) {
|
||||
state.lastTopLevelComma = index;
|
||||
}
|
||||
}
|
||||
state.scannedTo = state.partialJson.length;
|
||||
return complete;
|
||||
}
|
||||
|
||||
/** Parse the complete top-level fields before a top-level comma by closing the
|
||||
* object at that boundary. */
|
||||
function recoveredToolInput(prefix: string): Record<string, unknown> | undefined {
|
||||
try {
|
||||
const value: unknown = JSON.parse(prefix + "}");
|
||||
return value !== null && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: undefined;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
export async function claudeCliPath(): Promise<string> {
|
||||
if (process.env.CLAUDE_CODE_EXECUTABLE) {
|
||||
return process.env.CLAUDE_CODE_EXECUTABLE;
|
||||
@@ -1271,6 +1348,13 @@ export class ClaudeAcpAgent {
|
||||
// gateways that don't carry a stable/matching id across the stream and the
|
||||
// consolidated message. Reset after each consolidated message consumes it.
|
||||
const streamedBlocks: { index: number; type: "text" | "thinking"; text: string }[] = [];
|
||||
// Tool-use blocks start streaming before their JSON input. Keep the
|
||||
// partial input per parent message and block index so completed top-level
|
||||
// fields can refine the pending tool call while it streams. Entries are
|
||||
// dropped at block/message boundaries; the whole map is swept when a turn
|
||||
// settles, since an interrupted subagent stream (keyed by a
|
||||
// parent_tool_use_id that never recurs) has no boundary event of its own.
|
||||
const streamedToolInputs: StreamedToolInputCache = new Map();
|
||||
// Stop reason accumulated for the active turn (result subtype, refusal,
|
||||
// max_tokens, …). Reset per turn; read when the turn settles at idle.
|
||||
let stopReason: StopReason = "end_turn";
|
||||
@@ -1451,6 +1535,7 @@ export class ClaudeAcpAgent {
|
||||
}
|
||||
session.turnQueue = (session.turnQueue ?? []).filter((t) => t !== turn);
|
||||
session.activeTurn = null;
|
||||
streamedToolInputs.clear();
|
||||
turn.resolve(result);
|
||||
};
|
||||
|
||||
@@ -1468,6 +1553,7 @@ export class ClaudeAcpAgent {
|
||||
turn.settled = true;
|
||||
session.turnQueue = (session.turnQueue ?? []).filter((t) => t !== turn);
|
||||
session.activeTurn = null;
|
||||
streamedToolInputs.clear();
|
||||
turn.reject(error);
|
||||
};
|
||||
|
||||
@@ -2459,6 +2545,7 @@ export class ClaudeAcpAgent {
|
||||
taskState: session.taskState,
|
||||
emittedToolCalls: session.emittedToolCalls,
|
||||
messageId: currentStreamMessageId,
|
||||
streamedToolInputs,
|
||||
},
|
||||
)) {
|
||||
await this.client.sessionUpdate(notification);
|
||||
@@ -5564,6 +5651,38 @@ function toolCallNotification(
|
||||
};
|
||||
}
|
||||
|
||||
/** Refine a pending tool call from the complete top-level fields recovered
|
||||
* from its still-streaming input. Shares `toolInfoFromToolUse` with the
|
||||
* consolidated path but never carries `content`: content built from partial
|
||||
* input is misleading (an Edit missing its `new_string` renders as a pure
|
||||
* deletion) or invalid (a Write diff without `content` lacks the required
|
||||
* `newText`), and the consolidated message supplies it moments later. */
|
||||
function streamedInputRefinement(
|
||||
toolUse: { id: string; name: string },
|
||||
input: Record<string, unknown>,
|
||||
supportsTerminalOutput: boolean,
|
||||
cwd?: string,
|
||||
): SessionNotification["update"] | undefined {
|
||||
// TodoWrite/Task* never surfaced a tool_call to refine (plan lane).
|
||||
if (!shouldEmitToolCall(toolUse.name)) {
|
||||
return undefined;
|
||||
}
|
||||
const { title, kind, locations } = toolInfoFromToolUse(
|
||||
{ ...toolUse, input },
|
||||
supportsTerminalOutput,
|
||||
cwd,
|
||||
);
|
||||
return {
|
||||
_meta: { claudeCode: { toolName: toolUse.name } } satisfies ToolUpdateMeta,
|
||||
toolCallId: toolUse.id,
|
||||
sessionUpdate: "tool_call_update",
|
||||
rawInput: input,
|
||||
title,
|
||||
kind,
|
||||
...(locations ? { locations } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert an SDKAssistantMessage (Claude) to a SessionNotification (ACP).
|
||||
* Only handles text, image, and thinking chunks for now.
|
||||
@@ -5954,28 +6073,101 @@ export function streamEventToAcpNotifications(
|
||||
taskState?: TaskState;
|
||||
emittedToolCalls?: Set<string>;
|
||||
messageId?: string;
|
||||
streamedToolInputs?: StreamedToolInputCache;
|
||||
},
|
||||
): SessionNotification[] {
|
||||
const event = message.event;
|
||||
const streamKey = message.parent_tool_use_id ?? "";
|
||||
const streamedToolInputs = options?.streamedToolInputs;
|
||||
const forwardedOptions = {
|
||||
clientCapabilities: options?.clientCapabilities,
|
||||
parentToolUseId: message.parent_tool_use_id,
|
||||
cwd: options?.cwd,
|
||||
taskState: options?.taskState,
|
||||
emittedToolCalls: options?.emittedToolCalls,
|
||||
messageId: options?.messageId,
|
||||
};
|
||||
switch (event.type) {
|
||||
case "content_block_start":
|
||||
case "content_block_start": {
|
||||
const block = event.content_block;
|
||||
if (
|
||||
streamedToolInputs &&
|
||||
(block.type === "tool_use" ||
|
||||
block.type === "server_tool_use" ||
|
||||
block.type === "mcp_tool_use")
|
||||
) {
|
||||
let inputsForMessage = streamedToolInputs.get(streamKey);
|
||||
if (!inputsForMessage) {
|
||||
inputsForMessage = new Map();
|
||||
streamedToolInputs.set(streamKey, inputsForMessage);
|
||||
}
|
||||
inputsForMessage.set(event.index, {
|
||||
id: block.id,
|
||||
name: block.name,
|
||||
partialJson: "",
|
||||
scannedTo: 0,
|
||||
inString: false,
|
||||
escaped: false,
|
||||
objectDepth: 0,
|
||||
arrayDepth: 0,
|
||||
lastTopLevelComma: -1,
|
||||
emittedThroughComma: -1,
|
||||
});
|
||||
}
|
||||
return toAcpNotifications(
|
||||
[event.content_block],
|
||||
[block],
|
||||
"assistant",
|
||||
sessionId,
|
||||
toolUseCache,
|
||||
client,
|
||||
logger,
|
||||
{
|
||||
clientCapabilities: options?.clientCapabilities,
|
||||
parentToolUseId: message.parent_tool_use_id,
|
||||
cwd: options?.cwd,
|
||||
taskState: options?.taskState,
|
||||
emittedToolCalls: options?.emittedToolCalls,
|
||||
messageId: options?.messageId,
|
||||
},
|
||||
forwardedOptions,
|
||||
);
|
||||
case "content_block_delta":
|
||||
}
|
||||
case "content_block_delta": {
|
||||
if (event.delta.type === "input_json_delta") {
|
||||
const streamedInput = streamedToolInputs?.get(streamKey)?.get(event.index);
|
||||
if (!streamedInput) return [];
|
||||
|
||||
streamedInput.partialJson += event.delta.partial_json;
|
||||
if (scanStreamedToolInput(streamedInput)) {
|
||||
// Input complete: the consolidated assistant message replays the
|
||||
// block with its full input and refines the call there; emitting
|
||||
// here too would send a duplicate identical update.
|
||||
const inputsForMessage = streamedToolInputs?.get(streamKey);
|
||||
inputsForMessage?.delete(event.index);
|
||||
if (inputsForMessage?.size === 0) streamedToolInputs?.delete(streamKey);
|
||||
return [];
|
||||
}
|
||||
if (streamedInput.lastTopLevelComma <= streamedInput.emittedThroughComma) {
|
||||
return [];
|
||||
}
|
||||
streamedInput.emittedThroughComma = streamedInput.lastTopLevelComma;
|
||||
const input = recoveredToolInput(
|
||||
streamedInput.partialJson.slice(0, streamedInput.lastTopLevelComma),
|
||||
);
|
||||
if (!input) return [];
|
||||
const supportsTerminalOutput =
|
||||
options?.clientCapabilities?._meta?.["terminal_output"] === true;
|
||||
const update = streamedInputRefinement(
|
||||
streamedInput,
|
||||
input,
|
||||
supportsTerminalOutput,
|
||||
options?.cwd,
|
||||
);
|
||||
if (!update) return [];
|
||||
if (message.parent_tool_use_id) {
|
||||
update._meta = {
|
||||
...update._meta,
|
||||
claudeCode: {
|
||||
...(update._meta?.claudeCode || {}),
|
||||
parentToolUseId: message.parent_tool_use_id,
|
||||
},
|
||||
};
|
||||
}
|
||||
applyMessageId(update, options?.messageId);
|
||||
return [{ sessionId, update }];
|
||||
}
|
||||
return toAcpNotifications(
|
||||
[event.delta],
|
||||
"assistant",
|
||||
@@ -5983,25 +6175,29 @@ export function streamEventToAcpNotifications(
|
||||
toolUseCache,
|
||||
client,
|
||||
logger,
|
||||
{
|
||||
clientCapabilities: options?.clientCapabilities,
|
||||
parentToolUseId: message.parent_tool_use_id,
|
||||
cwd: options?.cwd,
|
||||
taskState: options?.taskState,
|
||||
emittedToolCalls: options?.emittedToolCalls,
|
||||
messageId: options?.messageId,
|
||||
},
|
||||
forwardedOptions,
|
||||
);
|
||||
}
|
||||
// No content. `ping` is a Messages-API keep-alive event that the SDK's
|
||||
// `BetaRawMessageStreamEvent` union doesn't include even though the
|
||||
// wire format emits it; the `as never` cast lets us no-op it here
|
||||
// instead of letting it fall through to `unreachable`.
|
||||
case "ping" as never:
|
||||
case "message_start":
|
||||
case "message_delta":
|
||||
case "message_stop":
|
||||
case "content_block_stop":
|
||||
return [];
|
||||
// A message boundary ends every input stream on this lane: message_stop is
|
||||
// the normal end, and a message_start clears anything a prior message on
|
||||
// the lane left behind (e.g. a stream cut short mid-block).
|
||||
case "message_start":
|
||||
case "message_stop":
|
||||
streamedToolInputs?.delete(streamKey);
|
||||
return [];
|
||||
case "content_block_stop": {
|
||||
const inputsForMessage = streamedToolInputs?.get(streamKey);
|
||||
inputsForMessage?.delete(event.index);
|
||||
if (inputsForMessage?.size === 0) streamedToolInputs?.delete(streamKey);
|
||||
return [];
|
||||
}
|
||||
|
||||
default:
|
||||
unreachable(event, logger);
|
||||
|
||||
@@ -43,6 +43,7 @@ import {
|
||||
runPromptWithCancellation,
|
||||
type AcpClient,
|
||||
type SDKMessageFilter,
|
||||
type StreamedToolInputCache,
|
||||
} from "../acp-agent.js";
|
||||
import { Pushable } from "../utils.js";
|
||||
import {
|
||||
@@ -8209,6 +8210,514 @@ describe("turn abandoned by the SDK (issue #825)", () => {
|
||||
});
|
||||
|
||||
describe("streamEventToAcpNotifications", () => {
|
||||
it("refines a tool call as soon as a streamed input field is complete", () => {
|
||||
const toolUseCache = {};
|
||||
const emittedToolCalls = new Set<string>();
|
||||
const streamedToolInputs: StreamedToolInputCache = new Map();
|
||||
const options = {
|
||||
cwd: "/Users/test/project",
|
||||
emittedToolCalls,
|
||||
streamedToolInputs,
|
||||
};
|
||||
const baseMessage = {
|
||||
type: "stream_event",
|
||||
parent_tool_use_id: null,
|
||||
uuid: randomUUID(),
|
||||
session_id: "test-session",
|
||||
};
|
||||
|
||||
const start = streamEventToAcpNotifications(
|
||||
{
|
||||
...baseMessage,
|
||||
event: {
|
||||
type: "content_block_start",
|
||||
index: 0,
|
||||
content_block: {
|
||||
type: "tool_use",
|
||||
id: "toolu_read",
|
||||
name: "Read",
|
||||
input: {},
|
||||
},
|
||||
},
|
||||
} as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
|
||||
expect(start).toHaveLength(1);
|
||||
expect(start[0].update).toMatchObject({
|
||||
sessionUpdate: "tool_call",
|
||||
toolCallId: "toolu_read",
|
||||
title: "Read File",
|
||||
locations: [],
|
||||
});
|
||||
|
||||
const partial = streamEventToAcpNotifications(
|
||||
{
|
||||
...baseMessage,
|
||||
event: {
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: { type: "input_json_delta", partial_json: '{"file_' },
|
||||
},
|
||||
} as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
|
||||
expect(partial).toEqual([]);
|
||||
|
||||
const pathAvailable = streamEventToAcpNotifications(
|
||||
{
|
||||
...baseMessage,
|
||||
event: {
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: {
|
||||
type: "input_json_delta",
|
||||
partial_json: 'path":"/Users/test/project/src/ZodiacList.tsx","offset":',
|
||||
},
|
||||
},
|
||||
} as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
|
||||
// The overall JSON is still invalid, but file_path is complete and already
|
||||
// makes this pending read distinguishable from other tool calls.
|
||||
expect(pathAvailable).toHaveLength(1);
|
||||
expect(pathAvailable[0].update).toMatchObject({
|
||||
sessionUpdate: "tool_call_update",
|
||||
toolCallId: "toolu_read",
|
||||
title: "Read src/ZodiacList.tsx",
|
||||
rawInput: { file_path: "/Users/test/project/src/ZodiacList.tsx" },
|
||||
locations: [{ path: "/Users/test/project/src/ZodiacList.tsx", line: 1 }],
|
||||
});
|
||||
expect(streamedToolInputs.size).toBe(1);
|
||||
|
||||
const completed = streamEventToAcpNotifications(
|
||||
{
|
||||
...baseMessage,
|
||||
event: {
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: { type: "input_json_delta", partial_json: "10}" },
|
||||
},
|
||||
} as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
|
||||
// Completion emits nothing: the consolidated assistant message replays the
|
||||
// block with its full input and refines the call there — emitting here too
|
||||
// would send a duplicate identical update. The entry is just cleaned up.
|
||||
expect(completed).toEqual([]);
|
||||
expect(streamedToolInputs.size).toBe(0);
|
||||
});
|
||||
|
||||
describe("partial tool input coverage", () => {
|
||||
function refineFromPartialInput({
|
||||
name,
|
||||
partialJson,
|
||||
type = "tool_use",
|
||||
}: {
|
||||
name: string;
|
||||
partialJson: string;
|
||||
type?: "tool_use" | "server_tool_use" | "mcp_tool_use";
|
||||
}) {
|
||||
const toolUseCache = {};
|
||||
const emittedToolCalls = new Set<string>();
|
||||
const streamedToolInputs: StreamedToolInputCache = new Map();
|
||||
const options = {
|
||||
cwd: "/Users/test/project",
|
||||
emittedToolCalls,
|
||||
streamedToolInputs,
|
||||
};
|
||||
const baseMessage = {
|
||||
type: "stream_event",
|
||||
parent_tool_use_id: null,
|
||||
uuid: randomUUID(),
|
||||
session_id: "test-session",
|
||||
};
|
||||
|
||||
const started = streamEventToAcpNotifications(
|
||||
{
|
||||
...baseMessage,
|
||||
event: {
|
||||
type: "content_block_start",
|
||||
index: 0,
|
||||
content_block: { type, id: "toolu_partial", name, input: {} },
|
||||
},
|
||||
} as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
const refined = streamEventToAcpNotifications(
|
||||
{
|
||||
...baseMessage,
|
||||
event: {
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: { type: "input_json_delta", partial_json: partialJson },
|
||||
},
|
||||
} as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
|
||||
return { started, refined, streamedToolInputs };
|
||||
}
|
||||
|
||||
it.each([
|
||||
{
|
||||
case: "Agent description",
|
||||
name: "Agent",
|
||||
partialJson: '{"description":"Investigate issue","prompt":',
|
||||
title: "Investigate issue",
|
||||
rawInput: { description: "Investigate issue" },
|
||||
},
|
||||
{
|
||||
case: "legacy Task description",
|
||||
name: "Task",
|
||||
partialJson: '{"description":"Research dependencies","prompt":',
|
||||
title: "Research dependencies",
|
||||
rawInput: { description: "Research dependencies" },
|
||||
},
|
||||
{
|
||||
case: "Bash command with comma and escaped quotes",
|
||||
name: "Bash",
|
||||
partialJson: `{"command":${JSON.stringify('sleep 900, then echo "done"')},"timeout":`,
|
||||
title: 'sleep 900, then echo "done"',
|
||||
rawInput: { command: 'sleep 900, then echo "done"' },
|
||||
},
|
||||
{
|
||||
case: "Read file path",
|
||||
name: "Read",
|
||||
partialJson: '{"file_path":"/Users/test/project/src/read.ts","offset":',
|
||||
title: "Read src/read.ts",
|
||||
rawInput: { file_path: "/Users/test/project/src/read.ts" },
|
||||
},
|
||||
{
|
||||
case: "Write file path",
|
||||
name: "Write",
|
||||
partialJson: '{"file_path":"/Users/test/project/src/write.ts","content":',
|
||||
title: "Write src/write.ts",
|
||||
rawInput: { file_path: "/Users/test/project/src/write.ts" },
|
||||
},
|
||||
{
|
||||
case: "Edit file path",
|
||||
name: "Edit",
|
||||
partialJson: '{"file_path":"/Users/test/project/src/edit.ts","old_string":',
|
||||
title: "Edit src/edit.ts",
|
||||
rawInput: { file_path: "/Users/test/project/src/edit.ts" },
|
||||
},
|
||||
{
|
||||
case: "Glob pattern",
|
||||
name: "Glob",
|
||||
partialJson: '{"pattern":"**/*.ts","path":',
|
||||
title: "Find `**/*.ts`",
|
||||
rawInput: { pattern: "**/*.ts" },
|
||||
},
|
||||
{
|
||||
case: "Grep strings, booleans, and numbers",
|
||||
name: "Grep",
|
||||
partialJson:
|
||||
'{"pattern":"TODO, FIXME","-i":true,"-n":true,"-A":2,"-B":1,"-C":3,"output_mode":"files_with_matches","head_limit":10,"glob":"*.ts","type":"ts","multiline":true,"path":',
|
||||
title:
|
||||
'grep -i -n -A 2 -B 1 -C 3 -l | head -10 --include="*.ts" --type=ts -P "TODO, FIXME"',
|
||||
rawInput: {
|
||||
pattern: "TODO, FIXME",
|
||||
"-i": true,
|
||||
"-n": true,
|
||||
"-A": 2,
|
||||
"-B": 1,
|
||||
"-C": 3,
|
||||
output_mode: "files_with_matches",
|
||||
head_limit: 10,
|
||||
glob: "*.ts",
|
||||
type: "ts",
|
||||
multiline: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
case: "WebFetch URL",
|
||||
name: "WebFetch",
|
||||
partialJson: '{"url":"https://example.com/docs","prompt":',
|
||||
title: "Fetch https://example.com/docs",
|
||||
rawInput: { url: "https://example.com/docs" },
|
||||
},
|
||||
{
|
||||
case: "WebSearch query and domain array",
|
||||
name: "WebSearch",
|
||||
type: "server_tool_use" as const,
|
||||
partialJson:
|
||||
'{"query":"ACP tools","allowed_domains":["agentclientprotocol.com","github.com"],"blocked_domains":',
|
||||
title: '"ACP tools" (allowed: agentclientprotocol.com, github.com)',
|
||||
rawInput: {
|
||||
query: "ACP tools",
|
||||
allowed_domains: ["agentclientprotocol.com", "github.com"],
|
||||
},
|
||||
},
|
||||
{
|
||||
case: "ReportFindings nested array and object",
|
||||
name: "ReportFindings",
|
||||
partialJson:
|
||||
'{"findings":[{"file":"src/a.ts","line":7,"summary":"Broken","failure_scenario":"Fails"}],"level":',
|
||||
title: "Report 1 finding",
|
||||
rawInput: {
|
||||
findings: [
|
||||
{
|
||||
file: "src/a.ts",
|
||||
line: 7,
|
||||
summary: "Broken",
|
||||
failure_scenario: "Fails",
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
{
|
||||
case: "generic Other tool",
|
||||
name: "Other",
|
||||
partialJson: '{"query":"custom query","options":',
|
||||
title: "Other",
|
||||
rawInput: { query: "custom query" },
|
||||
},
|
||||
{
|
||||
case: "custom MCP tool",
|
||||
name: "mcp__demo__search",
|
||||
type: "mcp_tool_use" as const,
|
||||
partialJson: '{"query":"custom MCP query","options":',
|
||||
title: "mcp__demo__search",
|
||||
rawInput: { query: "custom MCP query" },
|
||||
},
|
||||
])("refines $case before the full JSON object completes", (testCase) => {
|
||||
const { refined, streamedToolInputs } = refineFromPartialInput(testCase);
|
||||
|
||||
expect(refined).toHaveLength(1);
|
||||
expect(refined[0].update).toMatchObject({
|
||||
sessionUpdate: "tool_call_update",
|
||||
toolCallId: "toolu_partial",
|
||||
title: testCase.title,
|
||||
rawInput: testCase.rawInput,
|
||||
});
|
||||
// Refinements never carry `content`: content built from partial input is
|
||||
// misleading (an Edit missing new_string renders as a deletion) or
|
||||
// invalid (a Write diff without content lacks the required newText).
|
||||
expect(refined[0].update).not.toHaveProperty("content");
|
||||
expect(streamedToolInputs.size).toBe(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
case: "ExitPlanMode plan",
|
||||
name: "ExitPlanMode",
|
||||
partialJson: '{"plan":"Implement streamed input"',
|
||||
},
|
||||
{
|
||||
case: "AskUserQuestion questions",
|
||||
name: "AskUserQuestion",
|
||||
partialJson:
|
||||
'{"questions":[{"question":"Which mode?","header":"Mode","options":[{"label":"Fast","description":"Fast mode"},{"label":"Safe","description":"Safe mode"}],"multiSelect":false}]',
|
||||
},
|
||||
])(
|
||||
"waits for the consolidated message when $case is the only field",
|
||||
({ name, partialJson }) => {
|
||||
// A single-field input has no top-level comma, so its one field only
|
||||
// completes when the whole object does — at which point the
|
||||
// consolidated assistant message refines the call. No early update.
|
||||
const { refined, streamedToolInputs } = refineFromPartialInput({ name, partialJson });
|
||||
expect(refined).toEqual([]);
|
||||
expect(streamedToolInputs.size).toBe(1);
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps streamed TodoWrite input out of the tool feed", () => {
|
||||
// TodoWrite surfaces as `plan` snapshots, not tool_calls; the snapshot is
|
||||
// emitted from the consolidated assistant message once the todos array is
|
||||
// complete, so the streamed lane stays silent.
|
||||
const { started, refined } = refineFromPartialInput({
|
||||
name: "TodoWrite",
|
||||
partialJson:
|
||||
'{"todos":[{"content":"Run tests","status":"in_progress","activeForm":"Running tests"}]',
|
||||
});
|
||||
|
||||
expect(started).toEqual([]);
|
||||
expect(refined).toEqual([]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
name: "TaskCreate",
|
||||
partialJson: '{"subject":"Create tests","description":',
|
||||
},
|
||||
{
|
||||
name: "TaskUpdate",
|
||||
partialJson: '{"taskId":"1","subject":"Update tests","status":',
|
||||
},
|
||||
{ name: "TaskList", partialJson: "{}" },
|
||||
{ name: "TaskGet", partialJson: '{"taskId":"1"}' },
|
||||
])("keeps deliberately suppressed $name calls out of the tool feed", (testCase) => {
|
||||
const { started, refined } = refineFromPartialInput(testCase);
|
||||
expect(started).toEqual([]);
|
||||
expect(refined).toEqual([]);
|
||||
});
|
||||
|
||||
it("does not publish an unfinished string", () => {
|
||||
const { refined } = refineFromPartialInput({
|
||||
name: "Bash",
|
||||
partialJson: '{"command":"sleep 900',
|
||||
});
|
||||
expect(refined).toEqual([]);
|
||||
});
|
||||
|
||||
it("does not publish an ambiguous number until its delimiter arrives", () => {
|
||||
const first = refineFromPartialInput({ name: "Bash", partialJson: '{"timeout":10' });
|
||||
expect(first.refined).toEqual([]);
|
||||
|
||||
const delimited = refineFromPartialInput({
|
||||
name: "Bash",
|
||||
partialJson: '{"timeout":100,"command":',
|
||||
});
|
||||
expect(delimited.refined).toHaveLength(1);
|
||||
expect(delimited.refined[0].update).toMatchObject({ rawInput: { timeout: 100 } });
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
label: "boolean",
|
||||
partialJson: '{"run_in_background":true,"command":',
|
||||
rawInput: { run_in_background: true },
|
||||
},
|
||||
{ label: "null", partialJson: '{"optional":null,"command":', rawInput: { optional: null } },
|
||||
{
|
||||
label: "array",
|
||||
partialJson: '{"items":[1,"two",false],"command":',
|
||||
rawInput: { items: [1, "two", false] },
|
||||
},
|
||||
{
|
||||
label: "nested object",
|
||||
partialJson: '{"options":{"limit":5,"enabled":true},"command":',
|
||||
rawInput: { options: { limit: 5, enabled: true } },
|
||||
},
|
||||
])("recovers a completed $label value at a field boundary", ({ partialJson, rawInput }) => {
|
||||
const { refined } = refineFromPartialInput({ name: "CustomTool", partialJson });
|
||||
expect(refined).toHaveLength(1);
|
||||
expect(refined[0].update).toMatchObject({ sessionUpdate: "tool_call_update", rawInput });
|
||||
});
|
||||
|
||||
it("survives a ping keep-alive arriving mid-stream", () => {
|
||||
const toolUseCache = {};
|
||||
const streamedToolInputs: StreamedToolInputCache = new Map();
|
||||
const options = { emittedToolCalls: new Set<string>(), streamedToolInputs };
|
||||
const baseMessage = {
|
||||
type: "stream_event",
|
||||
parent_tool_use_id: null,
|
||||
uuid: randomUUID(),
|
||||
session_id: "test-session",
|
||||
};
|
||||
const send = (event: unknown) =>
|
||||
streamEventToAcpNotifications(
|
||||
{ ...baseMessage, event } as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
|
||||
send({
|
||||
type: "content_block_start",
|
||||
index: 0,
|
||||
content_block: { type: "tool_use", id: "toolu_ping", name: "Bash", input: {} },
|
||||
});
|
||||
send({
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: { type: "input_json_delta", partial_json: '{"command":"sleep 900"' },
|
||||
});
|
||||
// The API interleaves ping keep-alives during long generation pauses —
|
||||
// exactly when a large input is streaming. It must not disturb the
|
||||
// in-flight buffer.
|
||||
expect(send({ type: "ping" })).toEqual([]);
|
||||
expect(streamedToolInputs.size).toBe(1);
|
||||
|
||||
const refined = send({
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: { type: "input_json_delta", partial_json: ',"timeout":' },
|
||||
});
|
||||
expect(refined).toHaveLength(1);
|
||||
expect(refined[0].update).toMatchObject({
|
||||
sessionUpdate: "tool_call_update",
|
||||
toolCallId: "toolu_ping",
|
||||
rawInput: { command: "sleep 900" },
|
||||
});
|
||||
});
|
||||
|
||||
it("drops the buffered input at content_block_stop and message boundaries", () => {
|
||||
const toolUseCache = {};
|
||||
const streamedToolInputs: StreamedToolInputCache = new Map();
|
||||
const options = { emittedToolCalls: new Set<string>(), streamedToolInputs };
|
||||
const baseMessage = {
|
||||
type: "stream_event",
|
||||
parent_tool_use_id: null,
|
||||
uuid: randomUUID(),
|
||||
session_id: "test-session",
|
||||
};
|
||||
const send = (event: unknown) =>
|
||||
streamEventToAcpNotifications(
|
||||
{ ...baseMessage, event } as Parameters<typeof streamEventToAcpNotifications>[0],
|
||||
"test-session",
|
||||
toolUseCache,
|
||||
{} as AcpClient,
|
||||
console,
|
||||
options,
|
||||
);
|
||||
|
||||
send({
|
||||
type: "content_block_start",
|
||||
index: 0,
|
||||
content_block: { type: "tool_use", id: "toolu_stop", name: "Bash", input: {} },
|
||||
});
|
||||
send({
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: { type: "input_json_delta", partial_json: '{"command":"tr' },
|
||||
});
|
||||
expect(streamedToolInputs.size).toBe(1);
|
||||
send({ type: "content_block_stop", index: 0 });
|
||||
expect(streamedToolInputs.size).toBe(0);
|
||||
|
||||
send({
|
||||
type: "content_block_start",
|
||||
index: 0,
|
||||
content_block: { type: "tool_use", id: "toolu_stale", name: "Bash", input: {} },
|
||||
});
|
||||
expect(streamedToolInputs.size).toBe(1);
|
||||
// A new message on the lane clears anything a cut-short stream left.
|
||||
send({ type: "message_start", message: {} });
|
||||
expect(streamedToolInputs.size).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
it("treats `ping` keep-alive events as no-ops without logging to stderr", () => {
|
||||
const errors: unknown[][] = [];
|
||||
const logger = {
|
||||
|
||||
Reference in New Issue
Block a user