Skip to content

Commit ee95479

Browse files
committed
fix(agent-adapters): robustness fixes to pi + claude ACP adapters
Five verified fixes to the Pi and Claude ACP agent adapters (each with a unit test), found via a multi-agent review: - claude: tool-permission handler no longer auto-resolves on a 2s timer (was fail-open: ran the guest's tool without host consent); host is authoritative. - claude: partial tool input attributed by streaming content-block index, not Map insertion order. - claude: a dead query reader marks the session closed so the next prompt() fails fast instead of hanging to the ACP timeout. - pi: live session subscription is torn down on replace + conn.closed; editSnapshots cleared per turn + on cancel. - pi + claude: failed session/update writes are logged to stderr instead of silently swallowed (emit chain kept alive). Tests: registry/agent/{claude,pi}/tests/adapter.test.mjs. No auto-resolve on timeout remains in any adapter.
1 parent 3883c2c commit ee95479

6 files changed

Lines changed: 413 additions & 32 deletions

File tree

registry/agent/claude/package.json

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
"scripts": {
1919
"build": "tsc && node ./scripts/build-patched-cli.mjs",
2020
"check-types": "tsc --noEmit",
21-
"test": "pnpm build && node --test tests/*.test.mjs"
21+
"test": "pnpm build && node --test --test-force-exit tests/*.test.mjs"
2222
},
2323
"dependencies": {
2424
"@agentclientprotocol/sdk": "^0.16.1",

registry/agent/claude/src/adapter.ts

Lines changed: 47 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -326,7 +326,9 @@ type PendingTurn = {
326326
sawToolCall: boolean;
327327
};
328328

329-
class ClaudeQuerySession {
329+
// Exported for unit tests (the constructor takes an injectable `queryFactory`,
330+
// so the SDK can be faked). Not part of the package's public API.
331+
export class ClaudeQuerySession {
330332
private promptQueue = new AsyncQueue<SDKUserMessage>();
331333
private query: Query;
332334
private readyPromise: Promise<void>;
@@ -338,6 +340,11 @@ class ClaudeQuerySession {
338340
string,
339341
{ toolName: string; rawInput?: Record<string, unknown> }
340342
>();
343+
// Maps a streaming content-block `index` -> toolUseId for the current turn.
344+
// `content_block_delta` (input_json_delta) events reference the tool call by
345+
// its content-block index, NOT by tool-call ordinal — any text/thinking block
346+
// before a tool_use shifts that index. Cleared on each turn terminus.
347+
private toolUseBlockIndex = new Map<number, string>();
341348
private reader: Promise<void>;
342349
private closed = false;
343350
private cancelled = false;
@@ -688,9 +695,15 @@ class ClaudeQuerySession {
688695
}
689696
} finally {
690697
traceAdapter(`consume_finally session=${this.sessionId}`);
698+
// The reader loop has exited — the SDK query stream is done (cleanly or
699+
// via error), so this session can never produce another result. Mark it
700+
// closed so a subsequent prompt() fails fast via the guard in prompt()
701+
// instead of queueing onto a dead reader and hanging to the ACP method
702+
// timeout (a zombie session).
703+
this.closed = true;
691704
if (this.pendingTurn) {
692705
this.pendingTurn.reject(
693-
new Error("Claude query ended before producing a result"),
706+
new Error("Claude session ended before producing a result"),
694707
);
695708
this.pendingTurn = null;
696709
}
@@ -793,6 +806,12 @@ class ClaudeQuerySession {
793806
const toolName = String(block.name ?? "tool");
794807
const rawInput = isRecord(block.input) ? block.input : undefined;
795808
this.activeToolCalls.set(toolUseId, { toolName, rawInput });
809+
// Remember this tool_use block's content-block index so subsequent
810+
// input_json_delta events (which reference the block by index) are
811+
// attributed to the right tool call.
812+
if (typeof event.index === "number") {
813+
this.toolUseBlockIndex.set(Number(event.index), toolUseId);
814+
}
796815
if (this.pendingTurn) this.pendingTurn.sawToolCall = true;
797816

798817
await this.emit({
@@ -892,6 +911,7 @@ class ClaudeQuerySession {
892911
});
893912
}
894913
this.activeToolCalls.clear();
914+
this.toolUseBlockIndex.clear();
895915

896916
await this.lastEmit;
897917

@@ -921,7 +941,16 @@ class ClaudeQuerySession {
921941
update,
922942
}),
923943
)
924-
.catch(() => {});
944+
// The catch is load-bearing: lastEmit is awaited at turn end and a
945+
// rejected chain would halt all later updates and surface as a spurious
946+
// prompt failure. But never swallow silently — a dropped session/update
947+
// (host disconnect / broken pipe) must be host-visible, so write it to
948+
// stderr (the onAgentStderr channel).
949+
.catch((error) => {
950+
process.stderr.write(
951+
`[claude-acp] failed to deliver session/update: ${formatError(error)}\n`,
952+
);
953+
});
925954
return this.lastEmit;
926955
}
927956

@@ -960,26 +989,13 @@ class ClaudeQuerySession {
960989
toolCallId: options.toolUseID,
961990
},
962991
};
963-
const permissionFallbackMs = Number.parseInt(
964-
process.env.CLAUDE_CODE_PERMISSION_FALLBACK_MS ?? "2000",
965-
10,
966-
);
967-
const response = await Promise.race([
968-
this.conn.requestPermission(request),
969-
new Promise<RequestPermissionResponse>((resolve) => {
970-
setTimeout(() => {
971-
traceAdapter(
972-
`permission_request_fallback session=${this.sessionId} tool=${toolName} toolUseId=${options.toolUseID}`,
973-
);
974-
resolve({
975-
outcome: {
976-
outcome: "selected",
977-
optionId: "allow_once",
978-
},
979-
});
980-
}, Number.isFinite(permissionFallbackMs) ? permissionFallbackMs : 2000);
981-
}),
982-
]);
992+
// The host's permission handler is authoritative: never auto-resolve
993+
// the request on a timer. A timer-based fallback here would either
994+
// fail OPEN (auto-allow — the untrusted guest's tool runs without host
995+
// consent) or fail closed early (deny a legitimate slow approval). If
996+
// the host never answers, the request fails via the bounded ACP method
997+
// timeout, which surfaces to the host rather than silently granting.
998+
const response = await this.conn.requestPermission(request);
983999
traceAdapter(
9841000
`permission_request_done session=${this.sessionId} tool=${toolName} toolUseId=${options.toolUseID}`,
9851001
);
@@ -997,9 +1013,14 @@ class ClaudeQuerySession {
9971013
toolName: string;
9981014
rawInput?: Record<string, unknown>;
9991015
} | null {
1000-
const entry = [...this.activeToolCalls.entries()][index];
1001-
if (!entry) return null;
1002-
const [toolUseId, value] = entry;
1016+
// `index` is the streaming content-block index, not a tool-call ordinal —
1017+
// resolve it through the per-turn block-index map so the partial input is
1018+
// attributed to the correct tool call even when text/thinking blocks
1019+
// precede the tool_use block.
1020+
const toolUseId = this.toolUseBlockIndex.get(index);
1021+
if (!toolUseId) return null;
1022+
const value = this.activeToolCalls.get(toolUseId);
1023+
if (!value) return null;
10031024
return { toolUseId, toolName: value.toolName, rawInput: value.rawInput };
10041025
}
10051026
}
Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
import test, { after } from "node:test";
2+
import assert from "node:assert/strict";
3+
import { resolve as resolvePath } from "node:path";
4+
5+
// Stop fns for every "open" fake query, drained after the run so the dangling
6+
// consume() loops (and the process) can exit.
7+
const cleanups = [];
8+
after(() => {
9+
for (const stop of cleanups) stop();
10+
});
11+
12+
// Unit tests for the ClaudeQuerySession adapter fixes. The class is exported
13+
// for testing and its constructor takes an injectable `queryFactory`, so these
14+
// drive the translation/permission/teardown logic with a fake Claude SDK query
15+
// and a mock ACP connection — no real SDK, no VM.
16+
const packageDir = resolvePath(import.meta.dirname, "..");
17+
const { ClaudeQuerySession } = await import(
18+
resolvePath(packageDir, "dist", "adapter.js")
19+
);
20+
21+
function makeConn(overrides = {}) {
22+
let closeConn;
23+
const closed = new Promise((r) => {
24+
closeConn = r;
25+
});
26+
const updates = [];
27+
return {
28+
updates,
29+
closeConn: () => closeConn(),
30+
sessionUpdate: async (u) => {
31+
updates.push(u);
32+
},
33+
requestPermission: async () => ({
34+
outcome: { outcome: "selected", optionId: "allow_once" },
35+
}),
36+
closed,
37+
...overrides,
38+
};
39+
}
40+
41+
function makeQuery({ endImmediately = false } = {}) {
42+
let stop;
43+
const stopped = new Promise((r) => {
44+
stop = r;
45+
});
46+
if (!endImmediately) cleanups.push(stop);
47+
return {
48+
setMcpServers: async () => {},
49+
interrupt: async () => {},
50+
setPermissionMode: async () => {},
51+
async *[Symbol.asyncIterator]() {
52+
if (endImmediately) return;
53+
await stopped; // ends consume() when drained in `after`
54+
},
55+
};
56+
}
57+
58+
function makeSession({ endImmediately = false, conn } = {}) {
59+
const c = conn ?? makeConn();
60+
let capturedOptions;
61+
const queryFactory = (arg) => {
62+
capturedOptions = arg.options;
63+
return makeQuery({ endImmediately });
64+
};
65+
const sess = new ClaudeQuerySession(
66+
c,
67+
"sess-1",
68+
"/workspace",
69+
"default",
70+
{ cwd: "/workspace", mcpServers: undefined },
71+
"/usr/bin/claude",
72+
queryFactory,
73+
);
74+
return { sess, conn: c, getOptions: () => capturedOptions };
75+
}
76+
77+
// ── Fix #5: input_json_delta maps to the correct tool by content-block index ──
78+
test("claude #5: partial tool input is attributed by content-block index, not insertion order", async () => {
79+
const { sess, conn } = makeSession();
80+
sess.pendingTurn = {
81+
sawAssistantText: false,
82+
sawToolCall: false,
83+
resolve() {},
84+
reject() {},
85+
};
86+
// A text block occupies content-block index 0; the tool_use block is index 1.
87+
await sess.handleStreamEvent({
88+
event: { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "thinking..." } },
89+
});
90+
await sess.handleStreamEvent({
91+
event: {
92+
type: "content_block_start",
93+
index: 1,
94+
content_block: { type: "tool_use", id: "tool-A", name: "Bash", input: {} },
95+
},
96+
});
97+
await sess.handleStreamEvent({
98+
event: { type: "content_block_delta", index: 1, delta: { type: "input_json_delta", partial_json: '{"command":"ls"}' } },
99+
});
100+
await sess.lastEmit;
101+
102+
const toolUpdates = conn.updates
103+
.map((u) => u.update)
104+
.filter((u) => u?.sessionUpdate === "tool_call_update");
105+
// With the old insertion-order lookup, findToolCallByIndex(1) on a 1-entry
106+
// map returns null and the partial input is silently dropped (no update).
107+
assert.equal(toolUpdates.length, 1, "expected exactly one tool_call_update for the partial input");
108+
assert.equal(toolUpdates[0].toolCallId, "tool-A", "partial input must target the tool at block index 1");
109+
assert.equal(toolUpdates[0].rawInput.partial_json, '{"command":"ls"}');
110+
});
111+
112+
test("claude #5: two tool_use blocks at different indices each get their own partial input", async () => {
113+
const { sess, conn } = makeSession();
114+
sess.pendingTurn = { sawAssistantText: false, sawToolCall: false, resolve() {}, reject() {} };
115+
// index 0 = text, index 1 = toolA, index 2 = toolB
116+
await sess.handleStreamEvent({ event: { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "x" } } });
117+
await sess.handleStreamEvent({ event: { type: "content_block_start", index: 1, content_block: { type: "tool_use", id: "tool-A", name: "Bash", input: {} } } });
118+
await sess.handleStreamEvent({ event: { type: "content_block_start", index: 2, content_block: { type: "tool_use", id: "tool-B", name: "Read", input: {} } } });
119+
await sess.handleStreamEvent({ event: { type: "content_block_delta", index: 2, delta: { type: "input_json_delta", partial_json: '{"path":"/a"}' } } });
120+
await sess.handleStreamEvent({ event: { type: "content_block_delta", index: 1, delta: { type: "input_json_delta", partial_json: '{"command":"echo"}' } } });
121+
await sess.lastEmit;
122+
123+
const updates = conn.updates.map((u) => u.update).filter((u) => u?.sessionUpdate === "tool_call_update");
124+
const byTool = Object.fromEntries(updates.map((u) => [u.toolCallId, u.rawInput.partial_json]));
125+
assert.equal(byTool["tool-B"], '{"path":"/a"}', "block index 2 → tool-B");
126+
assert.equal(byTool["tool-A"], '{"command":"echo"}', "block index 1 → tool-A");
127+
});
128+
129+
// ── Fix #1: permission handler is host-authoritative; no timer auto-resolve ──
130+
test("claude #1: permission handler returns the host's deny decision", async () => {
131+
const conn = makeConn({
132+
requestPermission: async () => ({ outcome: { outcome: "selected", optionId: "reject_once" } }),
133+
});
134+
const { getOptions } = makeSession({ conn });
135+
const canUseTool = getOptions().canUseTool;
136+
const result = await canUseTool("Bash", { command: "rm -rf /" }, { toolUseID: "t1", title: "Bash", suggestions: [] });
137+
assert.equal(result.behavior, "deny", "host reject must produce a deny");
138+
});
139+
140+
test("claude #1: permission handler does NOT auto-resolve on a timer when the host is silent", async () => {
141+
let pendingResolve;
142+
const conn = makeConn({
143+
// Never settles — simulates a host that hasn't answered yet.
144+
requestPermission: () => new Promise((r) => {
145+
pendingResolve = r;
146+
}),
147+
});
148+
const { getOptions } = makeSession({ conn });
149+
const canUseTool = getOptions().canUseTool;
150+
const handlerPromise = canUseTool("Bash", {}, { toolUseID: "t1", title: "Bash", suggestions: [] });
151+
const timer = new Promise((r) => setTimeout(() => r("TIMER_WON"), 300));
152+
const winner = await Promise.race([handlerPromise.then(() => "HANDLER_WON"), timer]);
153+
assert.equal(winner, "TIMER_WON", "handler must not auto-resolve before the host answers (no fail-open timer)");
154+
// settle the pending request so the handler promise doesn't dangle
155+
pendingResolve({ outcome: { outcome: "selected", optionId: "reject_once" } });
156+
await handlerPromise;
157+
});
158+
159+
// ── Fix #2: emit logs delivery failures (host-visible) and keeps the chain alive ──
160+
test("claude #2: a failed sessionUpdate is logged to stderr and the emit chain survives", async () => {
161+
let failNext = true;
162+
const delivered = [];
163+
const conn = makeConn({
164+
sessionUpdate: async (u) => {
165+
if (failNext) {
166+
failNext = false;
167+
throw new Error("broken pipe");
168+
}
169+
delivered.push(u);
170+
},
171+
});
172+
const { sess } = makeSession({ conn });
173+
174+
const writes = [];
175+
const orig = process.stderr.write;
176+
process.stderr.write = (s) => {
177+
writes.push(String(s));
178+
return true;
179+
};
180+
try {
181+
await sess.emit({ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "a" } });
182+
await sess.emit({ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "b" } });
183+
await sess.lastEmit;
184+
} finally {
185+
process.stderr.write = orig;
186+
}
187+
assert.ok(
188+
writes.some((w) => w.includes("failed to deliver session/update")),
189+
"a delivery failure must be written to stderr, not swallowed",
190+
);
191+
assert.equal(delivered.length, 1, "the chain must survive the failure and deliver the next update");
192+
});
193+
194+
// ── Fix #4: a dead reader marks the session closed so prompt() fails fast ──
195+
test("claude #4: once the query stream ends, prompt() fails fast instead of hanging", async () => {
196+
const { sess } = makeSession({ endImmediately: true });
197+
await sess.reader; // wait for consume() to finish on the now-ended query
198+
await assert.rejects(
199+
sess.prompt({ prompt: [{ type: "text", text: "hi" }] }),
200+
/Session is closed/,
201+
"a prompt on a dead session must reject promptly, not hang",
202+
);
203+
});

registry/agent/pi/package.json

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@
1818
"scripts": {
1919
"build": "tsc && node scripts/build-snapshot-bundle.mjs",
2020
"build:snapshot": "node scripts/build-snapshot-bundle.mjs",
21-
"check-types": "tsc --noEmit"
21+
"check-types": "tsc --noEmit",
22+
"test": "pnpm build && node --test --test-force-exit tests/*.test.mjs"
2223
},
2324
"dependencies": {
2425
"@agentclientprotocol/sdk": "^0.16.1",

0 commit comments

Comments
 (0)