From 7f7927779de3140b3fa0afb560e642b72585ea24 Mon Sep 17 00:00:00 2001 From: Alexei Ciobanu Date: Sat, 11 Jul 2026 13:33:24 +1000 Subject: [PATCH] fix: serialize patch file mutations --- src/index.ts | 98 ++++++++++++++++++++++++++++++---------------- test/index.test.ts | 83 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 148 insertions(+), 33 deletions(-) diff --git a/src/index.ts b/src/index.ts index 0756bad..a10a1fb 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,6 +8,7 @@ import { getLanguageFromPath, highlightCode, type ToolDefinition, + withFileMutationQueue, } from "@earendil-works/pi-coding-agent"; import { Box, Container, Spacer, Text } from "@earendil-works/pi-tui"; import * as Diff from "diff"; @@ -1129,41 +1130,46 @@ async function applySingleHunk( hunk: ParsedPatch, ): Promise<{ summary: string; appliedFile: string; fuzz: number }> { const absolutePath = await resolvePatchPath(cwd, hunk.filePath); - if (hunk.type === "add") { - await mkdir(path.dirname(absolutePath), { recursive: true }); - await writeFileAtomic(absolutePath, hunk.content); - return { summary: `add: ${hunk.filePath}`, appliedFile: hunk.filePath, fuzz: 0 }; - } - - if (hunk.type === "delete") { - await stat(absolutePath); - await rm(absolutePath); - return { summary: `delete: ${hunk.filePath}`, appliedFile: hunk.filePath, fuzz: 0 }; - } - - const currentContent = await readFile(absolutePath, "utf-8"); - const chunkResult = - hunk.chunks.length === 0 - ? { content: currentContent, fuzz: 0 } - : replaceChunks(currentContent, hunk.filePath, hunk.chunks); - const nextContent = chunkResult.content; - - if (hunk.movePath) { - const absoluteMovePath = await resolvePatchPath(cwd, hunk.movePath); - await mkdir(path.dirname(absoluteMovePath), { recursive: true }); - await writeFileAtomic(absoluteMovePath, nextContent); - if (absoluteMovePath !== absolutePath) { + const absoluteMovePath = + hunk.type === "update" && hunk.movePath ? await resolvePatchPath(cwd, hunk.movePath) : undefined; + const mutationPaths = absoluteMovePath ? [absolutePath, absoluteMovePath] : [absolutePath]; + + return withPatchFileMutationQueues(mutationPaths, async () => { + if (hunk.type === "add") { + await mkdir(path.dirname(absolutePath), { recursive: true }); + await writeFileAtomic(absolutePath, hunk.content); + return { summary: `add: ${hunk.filePath}`, appliedFile: hunk.filePath, fuzz: 0 }; + } + + if (hunk.type === "delete") { + await stat(absolutePath); await rm(absolutePath); + return { summary: `delete: ${hunk.filePath}`, appliedFile: hunk.filePath, fuzz: 0 }; } - return { - summary: `move: ${hunk.filePath} -> ${hunk.movePath}`, - appliedFile: hunk.movePath, - fuzz: chunkResult.fuzz, - }; - } - await writeFileAtomic(absolutePath, nextContent); - return { summary: `update: ${hunk.filePath}`, appliedFile: hunk.filePath, fuzz: chunkResult.fuzz }; + const currentContent = await readFile(absolutePath, "utf-8"); + const chunkResult = + hunk.chunks.length === 0 + ? { content: currentContent, fuzz: 0 } + : replaceChunks(currentContent, hunk.filePath, hunk.chunks); + const nextContent = chunkResult.content; + + if (hunk.movePath && absoluteMovePath) { + await mkdir(path.dirname(absoluteMovePath), { recursive: true }); + await writeFileAtomic(absoluteMovePath, nextContent); + if (absoluteMovePath !== absolutePath) { + await rm(absolutePath); + } + return { + summary: `move: ${hunk.filePath} -> ${hunk.movePath}`, + appliedFile: hunk.movePath, + fuzz: chunkResult.fuzz, + }; + } + + await writeFileAtomic(absolutePath, nextContent); + return { summary: `update: ${hunk.filePath}`, appliedFile: hunk.filePath, fuzz: chunkResult.fuzz }; + }); } export async function applyPatchDetailed( @@ -1355,6 +1361,32 @@ async function resolvePatchPath(cwd: string, filePath: string): Promise return absolutePath; } +async function canonicalMutationPath(filePath: string): Promise { + try { + return await realpath(filePath); + } catch (error) { + if (hasErrorCode(error, "ENOENT")) { + return path.resolve(filePath); + } + throw error; + } +} + +async function withPatchFileMutationQueues(filePaths: string[], operation: () => Promise): Promise { + const canonicalPaths = await Promise.all(filePaths.map(canonicalMutationPath)); + const sortedPaths = [...new Set(canonicalPaths)].sort((left, right) => left.localeCompare(right)); + + const runQueued = (index: number): Promise => { + const filePath = sortedPaths[index]; + if (filePath === undefined) { + return operation(); + } + return withFileMutationQueue(filePath, () => runQueued(index + 1)); + }; + + return runQueued(0); +} + function syncToolset( pi: Pick, model: Model | undefined, @@ -1439,7 +1471,7 @@ export function createApplyPatchTool(): ApplyPatchToolDefinition { { type: "text", text: [ - "apply_patch partially failed.", + result.hasPartialSuccess ? "apply_patch partially failed." : "apply_patch failed.", `Failed: ${failed}`, `Recovery: MUST read ${mustReadText} before retrying.`, result.appliedFiles.length > 0 diff --git a/test/index.test.ts b/test/index.test.ts index 62fab9f..f97cacd 100644 --- a/test/index.test.ts +++ b/test/index.test.ts @@ -841,6 +841,89 @@ EOF`; ); }); + it("#given apply patch tool complete failure #when executed #then does not report partial failure", async () => { + // given + const directory = await createTempDirectory(); + await writeFile(path.join(directory, "broken.txt"), "line\n", "utf-8"); + const patch = `*** Begin Patch +*** Update File: broken.txt +@@ +-missing ++changed +*** End Patch`; + + // when + const result = await createApplyPatchTool().execute("apply-patch-test", { input: patch }, undefined, undefined, { + cwd: directory, + } as never); + + // then + const text = result.content.find((block) => block.type === "text")?.text ?? ""; + expect(text).toContain("apply_patch failed."); + expect(text).not.toContain("partially failed"); + expect(text).toContain("No file actions were applied."); + }); + + it("#given concurrent patches to different lines in one file #when applied #then preserves both updates", async () => { + // given + const directory = await createTempDirectory(); + await writeFile(path.join(directory, "shared.txt"), "first\nsecond\n", "utf-8"); + const firstPatch = `*** Begin Patch +*** Update File: shared.txt +@@ +-first ++FIRST +*** End Patch`; + const secondPatch = `*** Begin Patch +*** Update File: shared.txt +@@ +-second ++SECOND +*** End Patch`; + + // when + await Promise.all([applyPatch(directory, firstPatch), applyPatch(directory, secondPatch)]); + + // then + expect(await readFile(path.join(directory, "shared.txt"), "utf-8")).toBe("FIRST\nSECOND\n"); + }); + + it("#given concurrent update and move of one file #when applied #then produces a serialized outcome", async () => { + // given + const directory = await createTempDirectory(); + await writeFile(path.join(directory, "source.txt"), "first\nsecond\n", "utf-8"); + const updatePatch = `*** Begin Patch +*** Update File: source.txt +@@ +-second ++SECOND +*** End Patch`; + const movePatch = `*** Begin Patch +*** Update File: source.txt +*** Move to: destination.txt +@@ +-first ++FIRST +*** End Patch`; + + // when + const [updateResult, moveResult] = await Promise.all([ + applyPatchDetailed(directory, updatePatch), + applyPatchDetailed(directory, movePatch), + ]); + + // then + const failureCount = updateResult.failures.length + moveResult.failures.length; + expect(failureCount === 0 || failureCount === 1).toBe(true); + await expect(readFile(path.join(directory, "source.txt"), "utf-8")).rejects.toMatchObject({ code: "ENOENT" }); + const destination = await readFile(path.join(directory, "destination.txt"), "utf-8"); + if (failureCount === 0) { + expect(destination).toBe("FIRST\nSECOND\n"); + } else { + expect(destination).toBe("FIRST\nsecond\n"); + } + }); + it("#given successful patch write #when applying patch #then atomic temp files are cleaned", async () => { // given const directory = await createTempDirectory();