fix(instructions): retain completed receipts during retry staging

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-27 21:49:19 -05:00
1 parent 9fe6a3ac01
commit aad11ff34c
2 files changed
+48 -8

No files matched your search

@@ -8,7 +8,7 @@ const execFile = promisify(execFileCallback);
import os from "node:os";
import path from "node:path";
import { randomUUID } from "node:crypto";
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { eq } from "drizzle-orm";
import { agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, agentInstructionWorkingCopies, createDb } from "@paperclipai/db";
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
@@ -122,6 +122,36 @@ describe("registered run instruction copies", () => {
expect((await revisions.readCurrent(target(), board()))?.content).toBe("board change");
});
it.each(["before replacement", "after replacement"])("preserves the completed receipt when retry staging fails %s", async (point) => {
const copy = await run();
await fs.writeFile(path.join(copy.localRoot, entryFile), "completed turn");
const saved = (await copies.collectStopped({ companyId, runId: copy.runId }))!;
expect(saved.state).toBe("saved");
const prior = (await revisions.readCurrent(target(), board()))!;
const current = await revisions.commit({ ...target(), entryFile, baseRevisionId: prior.revision.id, content: "new board instructions", source: "api" }, board());
const originalMkdir = fs.mkdir.bind(fs);
const originalChmod = fs.chmod.bind(fs);
const mkdir = vi.spyOn(fs, "mkdir").mockImplementation(async (...args: Parameters<typeof fs.mkdir>) => {
if (point === "before replacement" && args[0] === copy.localRoot) throw new Error("retry staging failed");
return originalMkdir(...args);
});
const chmod = vi.spyOn(fs, "chmod").mockImplementation(async (...args) => {
if (point === "after replacement" && args[0] === path.join(copy.localRoot, entryFile)) throw new Error("retry staging failed");
return originalChmod(...args);
});
try {
await expect(copies.prepare({ ...target(), runId: copy.runId, cwd: home })).rejects.toThrow("retry staging failed");
} finally { mkdir.mockRestore(); chmod.mockRestore(); }
expect(await copies.get(companyId, copy.runId)).toEqual(saved);
copies = agentInstructionWorkingCopyService(db);
await copies.recoverStopped();
expect(await copies.get(companyId, copy.runId)).toEqual(saved);
expect((await revisions.readCurrent(target(), board()))?.revision.id).toBe(current.revision.id);
const retried = (await copies.prepare({ ...target(), runId: copy.runId, cwd: home }))!;
expect(retried).toMatchObject({ state: "prepared", baseRevisionId: current.revision.id, processStoppedAt: null });
expect(await fs.readFile(path.join(retried.localRoot, entryFile), "utf8")).toBe(current.content);
});
it("reads an existing canonical runtime snapshot without seeding or changing instruction bytes", async () => {
expect(await revisions.readCommittedForRuntime(target())).toBeNull();
const copy = await run();
@@ -110,6 +110,14 @@ export function agentInstructionWorkingCopyService(db: Db) {
// Refresh only bytes whose previous lifecycle is complete. A changed
// private file must keep its old CAS fence; never silently rebase edits.
refreshStoppedCopy = privateBytes === null || hash(privateBytes) === (existing.candidateHash ?? existing.baseHash);
if (!refreshStoppedCopy && privateBytes !== null) {
// A failed staging attempt can already have replaced the file before
// the completed row advances. Exact current canonical bytes are also
// accounted for; any other private edit still keeps its old fence.
const current = await revisions.readCommittedForRuntime({ companyId: input.companyId, agentId: input.agentId });
refreshStoppedCopy = current?.revision.entryFile === existing.entryFile
&& current.revision.contentHash === hash(privateBytes);
}
}
if (!refreshStoppedCopy) {
if (completed.has(existing.state) || existing.state === "unchanged_turn") {
@@ -144,12 +152,7 @@ export function agentInstructionWorkingCopyService(db: Db) {
localRoot, executionRoot, location, state: "preparing",
});
}
let row = (await get(input.companyId, input.runId))!;
if (refreshStoppedCopy) {
row = await patch(row, { state: "preparing", entryFile, baseRevisionId: baseline.revision.id,
baseHash: hash(content), candidateBase64: null, candidateHash: null, receipt: null,
processStoppedAt: null, attempts: 0, nextAttemptAt: null, errorCode: null, errorMessage: null });
}
const row = (await get(input.companyId, input.runId))!;
await assertInstructionPathSafe(localRoot, entryFile);
await execFile(process.execPath, ["-e", instructionGitExcludeProgram, workspace], { timeout: 15_000 });
await fs.mkdir(localRoot, { recursive: true, mode: 0o700 });
@@ -173,7 +176,14 @@ export function agentInstructionWorkingCopyService(db: Db) {
`node -e ${quote(instructionGitExcludeProgram)} ${quote(target.remoteCwd)}`, { cwd: target.remoteCwd, env: {}, timeoutSec: 15 });
if (excluded.exitCode !== 0 || excluded.timedOut) throw new Error("Could not exclude private instructions from Git staging");
}
const prepared = await patch(row, { state: "prepared" });
// A refresh does not supersede the completed turn until every local and
// remote staging step succeeds. Keep its receipt and stop evidence intact
// if preparation throws or the controller stops midway through staging.
const prepared = await patch(row, { state: "prepared", ...(refreshStoppedCopy ? {
entryFile, baseRevisionId: baseline.revision.id, baseHash: hash(content),
candidateBase64: null, candidateHash: null, receipt: null, processStoppedAt: null,
attempts: 0, nextAttemptAt: null, errorCode: null, errorMessage: null,
} : {}) });
if (input.target) liveTargets.set(targetKey(input.companyId, input.runId), input.target);
return prepared;
}