fix(tools): the abort cascade — kill running tool calls, honest activity labels
Root cause of the live-test bugs (2026-09-19): pi passes every run's
AbortSignal to tool execute, but toPiTool dropped it — agent.abort()
only landed at the next step boundary, so a long local grep held its
turn hostage until the dev server's memory restart killed the run
mid-tool, freezing the record at its write-ahead state (the 'went
back in time' rollback).
- lh-tool: execute ctx gains signal (parent + timeout combined); every
call races a 30s system-level timeout with a 250ms cooperative grace —
abort kills a stuck tool but lets a poll-the-signal tool deliver its
partial results. Pre-aborted signals never fire listeners: checked
upfront. toolTimeoutMs is the test seam (pdfModelOverride idiom).
- fs-shared: walkFiles returns {timedOut, aborted} instead of throwing —
abort yields partial results.
- grep: the scan phase was unbounded (the walk's 10s deadline only
bounded collection) — same budget now checked per file; find/grep
check the signal and return partials with a notice.
- pending/chat: streamingToolActivity names the ACTUAL running tool —
a part sits input-streaming through its whole execution, so the old
shimmer called minutes-long greps 'Writing a question…'.
Tests: toPiTool guard (forward, abort, timeout, grace, question
bypass, ToolInputError passthrough), walk mid-abort, find/grep
pre-abort + mid-run abort, activity labels. Gates: 617/617.
This commit is contained in:
amosblomqvist
+86
-3
@@ -31,8 +31,14 @@ export interface LhTool<TInput = any, TResult = any> {
|
||||
/**
|
||||
* Server-side execution. Absent + clientQuestion = the client answers.
|
||||
* Throws on validation failure — the loop converts to an error result.
|
||||
*
|
||||
* ctx.signal (2026-09-19, the abort-cascade fix): the run's abort signal
|
||||
* COMBINED with the system-level per-call timeout — aborting the run
|
||||
* kills a running tool, the timeout is the net so no tool can hold a
|
||||
* turn hostage (the 2.7-minute grep lesson). Tools with loops check it
|
||||
* and return partial results; fast single-op tools may ignore it.
|
||||
*/
|
||||
execute?: (input: TInput, ctx: { toolCallId: string }) => TResult | Promise<TResult>;
|
||||
execute?: (input: TInput, ctx: { toolCallId: string; signal?: AbortSignal }) => TResult | Promise<TResult>;
|
||||
/** Client-answered question tool (quiz/ask): a call ends the turn. */
|
||||
clientQuestion?: boolean;
|
||||
}
|
||||
@@ -81,6 +87,83 @@ export class ToolInputError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* System-level per-call timeout (2026-09-19, the abort-cascade fix): pi
|
||||
* passes every run's AbortSignal to tool execute, but the bridge used to
|
||||
* drop it — so agent.abort() only landed at the NEXT step boundary and a
|
||||
* long-running local tool (the grep that outlived its turn by minutes)
|
||||
* held the turn hostage until the dev server's memory restart killed the
|
||||
* run mid-tool, freezing the record at its write-ahead state (the "went
|
||||
* back in time" rollback). The guard now races every execute against (a)
|
||||
* the run's signal and (b) this timeout, rejecting with pi's own wording
|
||||
* ("Operation aborted") on abort and a clear timeout message otherwise —
|
||||
* pi turns the throw into an error tool result the model can see.
|
||||
*/
|
||||
const TOOL_TIMEOUT_MS = 30_000;
|
||||
/** Test seam (runner.ts's pdfModelOverride idiom). */
|
||||
export const toolTimeoutMs = { current: TOOL_TIMEOUT_MS };
|
||||
|
||||
/** How long an aborted/timed-out call may take to settle on its own before
|
||||
* the hard reject lands — cooperative tools poll the signal and return
|
||||
* partial results within a tick or two (the fs tools check per file). */
|
||||
const TOOL_ABORT_GRACE_MS = 250;
|
||||
|
||||
function runWithAbortGuard<T>(
|
||||
toolName: string,
|
||||
run: (signal: AbortSignal) => T | Promise<T>,
|
||||
parent: AbortSignal | undefined,
|
||||
): Promise<T> {
|
||||
const controller = new AbortController();
|
||||
const forwardAbort = () => controller.abort();
|
||||
if (parent) {
|
||||
if (parent.aborted) controller.abort();
|
||||
else parent.addEventListener("abort", forwardAbort, { once: true });
|
||||
}
|
||||
const timeoutMs = toolTimeoutMs.current;
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
const humanTimeout = timeoutMs >= 1_000 ? `${Math.round(timeoutMs / 1000)}s` : `${timeoutMs}ms`;
|
||||
let graceTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
const cleanup = () => {
|
||||
clearTimeout(timer);
|
||||
if (graceTimer) clearTimeout(graceTimer);
|
||||
parent?.removeEventListener("abort", forwardAbort);
|
||||
};
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const settle = (fn: () => void) => {
|
||||
controller.signal.removeEventListener("abort", onAbort);
|
||||
if (graceTimer) clearTimeout(graceTimer);
|
||||
fn();
|
||||
};
|
||||
const onAbort = () => {
|
||||
// Cooperative tools poll the signal and settle with partial
|
||||
// results within a tick or two (the fs tools check per file) —
|
||||
// give them a brief grace to deliver those before the hard
|
||||
// reject, so abort KILLS a stuck tool without racing away a
|
||||
// graceful one's partial result.
|
||||
graceTimer = setTimeout(
|
||||
() =>
|
||||
reject(
|
||||
new Error(
|
||||
parent?.aborted
|
||||
? "Operation aborted"
|
||||
: `Tool "${toolName}" timed out after ${humanTimeout} — the result was discarded`,
|
||||
),
|
||||
),
|
||||
TOOL_ABORT_GRACE_MS,
|
||||
);
|
||||
};
|
||||
controller.signal.addEventListener("abort", onAbort, { once: true });
|
||||
if (controller.signal.aborted) onAbort(); // an already-aborted signal never fires listeners
|
||||
Promise.resolve()
|
||||
.then(() => run(controller.signal))
|
||||
.then(
|
||||
(v) => settle(() => resolve(v)),
|
||||
(e) => settle(() => reject(e)),
|
||||
);
|
||||
})
|
||||
.finally(cleanup);
|
||||
}
|
||||
|
||||
/** Convert one lh tool to a pi AgentTool. */
|
||||
export function toPiTool(def: LhTool<any, any>): AgentTool<any, any> {
|
||||
return {
|
||||
@@ -97,7 +180,7 @@ export function toPiTool(def: LhTool<any, any>): AgentTool<any, any> {
|
||||
// dock's subject can never flip mid-turn between two parallel
|
||||
// questions. Question executes are instant; no latency is lost.
|
||||
executionMode: def.clientQuestion ? "sequential" : "parallel",
|
||||
execute: async (toolCallId, args) => {
|
||||
execute: async (toolCallId, args, signal) => {
|
||||
// The authoring check: structural JSON Schema validation already
|
||||
// ran in the loop's prepare step; this re-run adds the refinements
|
||||
// (superRefine rules) the JSON Schema can't carry.
|
||||
@@ -111,7 +194,7 @@ export function toPiTool(def: LhTool<any, any>): AgentTool<any, any> {
|
||||
details: null,
|
||||
};
|
||||
}
|
||||
const result = await def.execute(parsed.data, { toolCallId });
|
||||
const result = await runWithAbortGuard(def.name, (toolSignal) => def.execute!(parsed.data, { toolCallId, signal: toolSignal }), signal);
|
||||
// Rich image result: model-facing content parts carry the images
|
||||
// (pi's ToolResultMessage.content is (TextContent | ImageContent)[]).
|
||||
if (isImageResult(result)) {
|
||||
|
||||
@@ -36,7 +36,8 @@ export const findTool = lhTool({
|
||||
name: "find",
|
||||
description: `Search for files by glob pattern. Returns matching file paths relative to the search directory. Ignores node_modules, .git, and Library directories. Output is truncated to ${DEFAULT_LIMIT} results or ${formatSize(MAX_OUTPUT_BYTES)} (whichever is hit first).`,
|
||||
schema: findInputSchema,
|
||||
async execute(input) {
|
||||
async execute(input, ctx) {
|
||||
if (ctx?.signal?.aborted) return "Operation aborted";
|
||||
const searchPath = resolveSourcePath(input.path);
|
||||
let stat;
|
||||
try {
|
||||
@@ -49,7 +50,7 @@ export const findTool = lhTool({
|
||||
const effectiveLimit = input.limit ?? DEFAULT_LIMIT;
|
||||
const matches: string[] = [];
|
||||
let limitReached = false;
|
||||
const { timedOut } = await walkFiles(
|
||||
const { timedOut, aborted } = await walkFiles(
|
||||
searchPath,
|
||||
(filePath) => {
|
||||
const rel = path.relative(searchPath, filePath).split(path.sep).join("/");
|
||||
@@ -61,8 +62,14 @@ export const findTool = lhTool({
|
||||
matches.push(rel);
|
||||
return false;
|
||||
},
|
||||
ctx?.signal,
|
||||
);
|
||||
|
||||
if (aborted) {
|
||||
return matches.length > 0
|
||||
? `${matches.join("\n")}\n\n[search aborted — partial results]`
|
||||
: "Search aborted before any results.";
|
||||
}
|
||||
if (matches.length === 0 && !limitReached) {
|
||||
return timedOut
|
||||
? `Search timed out after 10s with no matches yet. Narrow the search to a subdirectory.`
|
||||
|
||||
@@ -136,16 +136,21 @@ export async function walkFiles(
|
||||
root: string,
|
||||
visit: (filePath: string) => boolean | Promise<boolean>,
|
||||
signal?: AbortSignal,
|
||||
): Promise<{ timedOut: boolean }> {
|
||||
): Promise<{ timedOut: boolean; aborted: boolean }> {
|
||||
const deadline = Date.now() + WALK_TIMEOUT_MS;
|
||||
let timedOut = false;
|
||||
let aborted = false;
|
||||
if (signal?.aborted) return { timedOut: false, aborted: true };
|
||||
|
||||
const walkDir = async (dir: string): Promise<boolean> => {
|
||||
if (Date.now() > deadline) {
|
||||
timedOut = true;
|
||||
return true; // stop
|
||||
}
|
||||
if (signal?.aborted) throw new Error("Operation aborted");
|
||||
if (signal?.aborted) {
|
||||
aborted = true;
|
||||
return true; // stop — abort yields partial results, not a throw
|
||||
}
|
||||
let entries: import("node:fs").Dirent[];
|
||||
try {
|
||||
entries = await fs.readdir(dir, { withFileTypes: true });
|
||||
@@ -157,6 +162,10 @@ export async function walkFiles(
|
||||
timedOut = true;
|
||||
return true;
|
||||
}
|
||||
if (signal?.aborted) {
|
||||
aborted = true;
|
||||
return true;
|
||||
}
|
||||
const full = path.join(dir, entry.name);
|
||||
if (entry.isDirectory()) {
|
||||
if (EXCLUDED_DIR_NAMES.has(entry.name)) continue;
|
||||
@@ -170,7 +179,7 @@ export async function walkFiles(
|
||||
};
|
||||
|
||||
await walkDir(root);
|
||||
return { timedOut };
|
||||
return { timedOut, aborted };
|
||||
}
|
||||
|
||||
/** True if the first 8KB chunk contains a NUL byte — pi/rg's binary skip. */
|
||||
|
||||
+27
-3
@@ -12,6 +12,7 @@ import {
|
||||
resolveSourcePath,
|
||||
truncateGrepLine,
|
||||
walkFiles,
|
||||
WALK_TIMEOUT_MS,
|
||||
} from "./fs-shared.ts";
|
||||
|
||||
/**
|
||||
@@ -45,7 +46,8 @@ export const grepTool = lhTool({
|
||||
name: "grep",
|
||||
description: `Search file contents for a pattern. Returns matching lines with file paths and line numbers. Ignores node_modules, .git, and Library directories. Output is truncated to ${DEFAULT_LIMIT} matches or ${formatSize(MAX_OUTPUT_BYTES)} (whichever is hit first). Long lines are truncated to ${GREP_MAX_LINE_LENGTH} chars.`,
|
||||
schema: grepInputSchema,
|
||||
async execute(input) {
|
||||
async execute(input, ctx) {
|
||||
if (ctx?.signal?.aborted) return "Operation aborted";
|
||||
const searchPath = resolveSourcePath(input.path);
|
||||
|
||||
let stat;
|
||||
@@ -66,6 +68,7 @@ export const grepTool = lhTool({
|
||||
// Files to scan: one explicit file, or the walked tree (glob-filtered).
|
||||
const files: string[] = [];
|
||||
let timedOut = false;
|
||||
let aborted = false;
|
||||
if (!isDirectory) {
|
||||
files.push(searchPath);
|
||||
} else {
|
||||
@@ -79,16 +82,32 @@ export const grepTool = lhTool({
|
||||
files.push(filePath);
|
||||
return false;
|
||||
},
|
||||
ctx?.signal,
|
||||
);
|
||||
timedOut = result.timedOut;
|
||||
aborted = result.aborted;
|
||||
}
|
||||
|
||||
const outLines: string[] = [];
|
||||
let matchCount = 0;
|
||||
let limitReached = false;
|
||||
let linesTruncated = false;
|
||||
// The scan phase is bounded too (2026-09-19): the walk's 10s deadline
|
||||
// bounds COLLECTION, but reading every collected file was unbounded —
|
||||
// the grep that outlived its turn by minutes. Same budget, checked per
|
||||
// file; abort lands mid-scan with partial results.
|
||||
const scanDeadline = Date.now() + WALK_TIMEOUT_MS;
|
||||
let scanTimedOut = false;
|
||||
|
||||
for (const filePath of files) {
|
||||
if (ctx?.signal?.aborted) {
|
||||
aborted = true;
|
||||
break;
|
||||
}
|
||||
if (Date.now() > scanDeadline) {
|
||||
scanTimedOut = true;
|
||||
break;
|
||||
}
|
||||
if (matchCount >= effectiveLimit) {
|
||||
limitReached = true;
|
||||
break;
|
||||
@@ -133,8 +152,13 @@ export const grepTool = lhTool({
|
||||
if (limitReached) break;
|
||||
}
|
||||
|
||||
if (aborted) {
|
||||
return outLines.length > 0
|
||||
? `${capOutput(outLines).content}\n\n[search aborted — partial results]`
|
||||
: "Search aborted before any results.";
|
||||
}
|
||||
if (outLines.length === 0) {
|
||||
return timedOut
|
||||
return timedOut || scanTimedOut
|
||||
? `Search timed out after 10s with no matches yet. Narrow the search to a subdirectory.`
|
||||
: "No matches found";
|
||||
}
|
||||
@@ -143,7 +167,7 @@ export const grepTool = lhTool({
|
||||
if (limitReached || matchCount >= effectiveLimit) notices.push(`${effectiveLimit} matches limit reached`);
|
||||
if (linesTruncated) notices.push(`Some lines truncated to ${GREP_MAX_LINE_LENGTH} chars. Use read tool to see full lines`);
|
||||
if (truncated) notices.push(`${formatSize(MAX_OUTPUT_BYTES)} limit reached`);
|
||||
if (timedOut) notices.push("search timed out after 10s — partial results, narrow the directory");
|
||||
if (timedOut || scanTimedOut) notices.push("search timed out after 10s — partial results, narrow the directory");
|
||||
return notices.length > 0 ? `${content}\n\n[${notices.join(". ")}]` : content;
|
||||
},
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user