agent/pi
2
0
Fork 0
mirror of https://github.com/earendil-works/pi.git synced 2026-06-18 15:54:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(coding-agent): preserve mutation queue on abort

Browse Source
This commit is contained in:
Armin Ronacher
2026-05-20 11:24:08 +02:00
Unverified
parent ba33b10392
commit fa656dd06e
3 changed files with 191 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/coding-agent/src/core/tools/edit.ts
+50 -100
View File
@@ -310,112 +310,62 @@ export function createEditToolDefinition(
const { path, edits } = validateEditInput(input);
const absolutePath = resolveToCwd(path, cwd);
return withFileMutationQueue(
absolutePath,
() =>
new Promise<{
content: Array<{ type: "text"; text: string }>;
details: EditToolDetails | undefined;
}>((resolve, reject) => {
// Check if already aborted.
if (signal?.aborted) {
reject(new Error("Operation aborted"));
return;
}
return withFileMutationQueue(absolutePath, async () => {
let aborted = signal?.aborted ?? false;
const onAbort = () => {
aborted = true;
};
const throwIfAborted = (): void => {
if (aborted || signal?.aborted) {
throw new Error("Operation aborted");
}
};
let aborted = false;
signal?.addEventListener("abort", onAbort, { once: true });
try {
throwIfAborted();
// Set up abort handler.
const onAbort = () => {
aborted = true;
reject(new Error("Operation aborted"));
};
// Check if file exists.
try {
await ops.access(absolutePath);
} catch (error: unknown) {
throwIfAborted();
const errorMessage =
error instanceof Error && "code" in error ? `Error code: ${error.code}` : String(error);
throw new Error(`Could not edit file: ${path}. ${errorMessage}.`);
}
throwIfAborted();
if (signal) {
signal.addEventListener("abort", onAbort, { once: true });
}
// Read the file.
const buffer = await ops.readFile(absolutePath);
throwIfAborted();
// Perform the edit operation.
void (async () => {
try {
// Check if file exists.
try {
await ops.access(absolutePath);
} catch (error: unknown) {
const errorMessage =
error instanceof Error && "code" in error ? `Error code: ${error.code}` : String(error);
if (signal) {
signal.removeEventListener("abort", onAbort);
}
reject(new Error(`Could not edit file: ${path}. ${errorMessage}.`));
return;
}
// Strip BOM before matching. The model will not include an invisible BOM in oldText.
const rawContent = buffer.toString("utf-8");
const { bom, text: content } = stripBom(rawContent);
const originalEnding = detectLineEnding(content);
const normalizedContent = normalizeToLF(content);
const { baseContent, newContent } = applyEditsToNormalizedContent(normalizedContent, edits, path);
throwIfAborted();
// Check if aborted before reading.
if (aborted) {
return;
}
const finalContent = bom + restoreLineEndings(newContent, originalEnding);
await ops.writeFile(absolutePath, finalContent);
throwIfAborted();
// Read the file.
const buffer = await ops.readFile(absolutePath);
const rawContent = buffer.toString("utf-8");
// Check if aborted after reading.
if (aborted) {
return;
}
// Strip BOM before matching. The model will not include an invisible BOM in oldText.
const { bom, text: content } = stripBom(rawContent);
const originalEnding = detectLineEnding(content);
const normalizedContent = normalizeToLF(content);
const { baseContent, newContent } = applyEditsToNormalizedContent(
normalizedContent,
edits,
path,
);
// Check if aborted before writing.
if (aborted) {
return;
}
const finalContent = bom + restoreLineEndings(newContent, originalEnding);
await ops.writeFile(absolutePath, finalContent);
// Check if aborted after writing.
if (aborted) {
return;
}
// Clean up abort handler.
if (signal) {
signal.removeEventListener("abort", onAbort);
}
const diffResult = generateDiffString(baseContent, newContent);
resolve({
content: [
{
type: "text",
text: `Successfully replaced ${edits.length} block(s) in ${path}.`,
},
],
details: { diff: diffResult.diff, firstChangedLine: diffResult.firstChangedLine },
});
} catch (error: unknown) {
// Clean up abort handler.
if (signal) {
signal.removeEventListener("abort", onAbort);
}
if (!aborted) {
reject(error instanceof Error ? error : new Error(String(error)));
}
}
})();
}),
);
const diffResult = generateDiffString(baseContent, newContent);
return {
content: [
{
type: "text",
text: `Successfully replaced ${edits.length} block(s) in ${path}.`,
},
],
details: { diff: diffResult.diff, firstChangedLine: diffResult.firstChangedLine },
};
} finally {
signal?.removeEventListener("abort", onAbort);
}
});
},
renderCall(args, theme, context) {
const component = getEditCallRenderComponent(context.state, context.lastComponent);
packages/coding-agent/src/core/tools/write.ts
+30 -38
View File
@@ -200,44 +200,36 @@ export function createWriteToolDefinition(
) {
const absolutePath = resolveToCwd(path, cwd);
const dir = dirname(absolutePath);
return withFileMutationQueue(
absolutePath,
() =>
new Promise<{ content: Array<{ type: "text"; text: string }>; details: undefined }>(
(resolve, reject) => {
if (signal?.aborted) {
reject(new Error("Operation aborted"));
return;
}
let aborted = false;
const onAbort = () => {
aborted = true;
reject(new Error("Operation aborted"));
};
signal?.addEventListener("abort", onAbort, { once: true });
(async () => {
try {
// Create parent directories if needed.
await ops.mkdir(dir);
if (aborted) return;
// Write the file contents.
await ops.writeFile(absolutePath, content);
if (aborted) return;
signal?.removeEventListener("abort", onAbort);
resolve({
content: [
{ type: "text", text: `Successfully wrote ${content.length} bytes to ${path}` },
],
details: undefined,
});
} catch (error: any) {
signal?.removeEventListener("abort", onAbort);
if (!aborted) reject(error);
}
})();
},
),
);
return withFileMutationQueue(absolutePath, async () => {
let aborted = signal?.aborted ?? false;
const onAbort = () => {
aborted = true;
};
const throwIfAborted = (): void => {
if (aborted || signal?.aborted) {
throw new Error("Operation aborted");
}
};
signal?.addEventListener("abort", onAbort, { once: true });
try {
throwIfAborted();
// Create parent directories if needed.
await ops.mkdir(dir);
throwIfAborted();
// Write the file contents.
await ops.writeFile(absolutePath, content);
throwIfAborted();
return {
content: [{ type: "text", text: `Successfully wrote ${content.length} bytes to ${path}` }],
details: undefined,
};
} finally {
signal?.removeEventListener("abort", onAbort);
}
});
},
renderCall(args, theme, context) {
const renderArgs = args as { path?: string; file_path?: string; content?: string } | undefined;
packages/coding-agent/test/file-mutation-queue.test.ts
+111
View File
@@ -10,6 +10,18 @@ function delay(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function createDeferred(): { promise: Promise<void>; resolve: () => void } {
let resolve!: () => void;
const promise = new Promise<void>((promiseResolve) => {
resolve = promiseResolve;
});
return { promise, resolve };
}
async function resolvesWithin(promise: Promise<unknown>, ms: number): Promise<boolean> {
return Promise.race([promise.then(() => true), delay(ms).then(() => false)]);
}
const tempDirs: string[] = [];
async function createTempDir(): Promise<string> {
@@ -160,4 +172,103 @@ describe("built-in edit and write tools", () => {
const content = await readFile(filePath, "utf8");
expect(content).toBe("replacement\n");
});
it("keeps write queue locked while an aborted write is still in flight", async () => {
const dir = await createTempDir();
const filePath = join(dir, "abort-write.txt");
const firstWriteStarted = createDeferred();
const finishFirstWrite = createDeferred();
const secondWriteStarted = createDeferred();
let firstWriteSettled = false;
const writeTool = createWriteTool(dir, {
operations: {
mkdir: async () => {},
writeFile: async (path, content) => {
if (content === "first\n") {
firstWriteStarted.resolve();
await finishFirstWrite.promise;
await writeFile(path, content, "utf8");
firstWriteSettled = true;
return;
}
if (content === "second\n") {
expect(firstWriteSettled).toBe(true);
secondWriteStarted.resolve();
}
await writeFile(path, content, "utf8");
},
},
});
const controller = new AbortController();
const firstWrite = writeTool.execute("call-1", { path: filePath, content: "first\n" }, controller.signal);
await firstWriteStarted.promise;
controller.abort();
const secondWrite = writeTool.execute("call-2", { path: filePath, content: "second\n" });
expect(await resolvesWithin(secondWriteStarted.promise, 20)).toBe(false);
finishFirstWrite.resolve();
await expect(firstWrite).rejects.toThrow("Operation aborted");
await secondWrite;
const content = await readFile(filePath, "utf8");
expect(content).toBe("second\n");
});
it("keeps edit queue locked while an aborted edit write is still in flight", async () => {
const dir = await createTempDir();
const filePath = join(dir, "abort-edit.txt");
await writeFile(filePath, "alpha\nbeta\n", "utf8");
const firstWriteStarted = createDeferred();
const finishFirstWrite = createDeferred();
const secondWriteStarted = createDeferred();
let firstWriteSettled = false;
const editTool = createEditTool(dir, {
operations: {
access,
readFile,
writeFile: async (path, content) => {
if (content === "ALPHA\nbeta\n") {
firstWriteStarted.resolve();
await finishFirstWrite.promise;
await writeFile(path, content, "utf8");
firstWriteSettled = true;
return;
}
if (content === "ALPHA\nBETA\n" || content === "alpha\nBETA\n") {
expect(firstWriteSettled).toBe(true);
secondWriteStarted.resolve();
}
await writeFile(path, content, "utf8");
},
},
});
const controller = new AbortController();
const firstEdit = editTool.execute(
"call-1",
{ path: filePath, edits: [{ oldText: "alpha", newText: "ALPHA" }] },
controller.signal,
);
await firstWriteStarted.promise;
controller.abort();
const secondEdit = editTool.execute("call-2", {
path: filePath,
edits: [{ oldText: "beta", newText: "BETA" }],
});
expect(await resolvesWithin(secondWriteStarted.promise, 20)).toBe(false);
finishFirstWrite.resolve();
await expect(firstEdit).rejects.toThrow("Operation aborted");
await secondEdit;
const content = await readFile(filePath, "utf8");
expect(content).toBe("ALPHA\nBETA\n");
});
});