From 03e1b3cd7270c7438a47562b170b3a9b2dbf112e Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 19:40:31 -0500 Subject: [PATCH] fix(server): recover captured instruction saves from committed deadlines Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 22 ++ ...-10-25-instruction-recovery-sleep-audit.md | 91 +++++- .../agent-instruction-working-copies.test.ts | 286 +++++++++++++++++- .../server-startup-feedback-export.test.ts | 10 +- server/src/index.ts | 5 +- server/src/services/agent-deletion.ts | 2 + .../agent-instruction-working-copies.ts | 83 ++++- server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + server/src/services/heartbeat.ts | 8 + server/src/services/heartbeat/recovery.ts | 4 +- server/src/services/instance-settings.ts | 2 + .../instruction-commit-notifications.ts | 7 + 13 files changed, 502 insertions(+), 21 deletions(-) create mode 100644 server/src/services/instruction-commit-notifications.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index a997f5bfc10..7239844f24a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2096,3 +2096,25 @@ Slice 40 will own captured-byte commit only. See [instruction recovery audit](2026-10-25-instruction-recovery-sleep-audit.md). Existing instruction-copy and directory suites pass 20 and 87 tests respectively on separate external PostgreSQL databases before any production behavior changes. + + +## Slice 39 accepted; slice 40 captured instruction commits + +[#17667](https://github.com/paperclipai/paperclip/pull/17667), head +`ecef7776a4b38e0c7e0b86d7009fcd01ed7658e5`, has all 53 checks passing, +fresh Greptile 5/5, and no unresolved threads. The startup fixture failure and +repair remain recorded. This PR is still draft and unmerged. + +Slice 40 migrates only captured instruction commits. Existing saved bytes, +attempt counts, and retry deadlines drive a bounded worker. Local writes signal +the existing transaction boundary; the revision service still owns authorization, +conflicts, file replacement, and exact-byte replay. A race test exposed a newer +retry deadline being ignored; the attempt claim now checks that it is still due. +Manual/startup recovery shares the guarded owner. Stopped collection and post-save +directory release remain separate duties. No schema migration is needed. + +Final instruction acceptance passes 33 tests, directory/startup passes 137, and +startup/instruction-release wiring passes 64 (184 distinct tests across these +selections). Workspace types/build pass. Fresh remote gates remain required. +See the instruction recovery audit for producer coverage and all failed attempts. +Earlier stack CI/review findings and final master reconciliation remain outstanding. diff --git a/doc/plans/2026-10-25-instruction-recovery-sleep-audit.md b/doc/plans/2026-10-25-instruction-recovery-sleep-audit.md index 22b75a20b74..b931b257e26 100644 --- a/doc/plans/2026-10-25-instruction-recovery-sleep-audit.md +++ b/doc/plans/2026-10-25-instruction-recovery-sleep-audit.md @@ -1,7 +1,8 @@ -# Next instruction recovery slices +# Instruction recovery sleep audit Inspected from slice 39 head bd490ee50b303187f0ae6f8fe2aa95477d66a40f. -Master inspected 5a28f7acf8. Audit only; no implementation or acceptance claim. +Master inspected 5a28f7acf8. The following is the initial audit; implementation +and acceptance evidence for slice 40 are recorded below. The apparent two calls in heartbeat/recovery.ts hide at least three reasons: 1. recoverStopped: collect/dispose registered instruction copies after terminal run @@ -76,3 +77,89 @@ Baseline acceptance before changing production behavior: - Suites used separate dedicated PostgreSQL databases through the guarded external test URL helper. Their embedded default and temporary filesystem isolation remain. - This only establishes a baseline. No periodic instruction duty is migrated yet. + +## Slice 40 implementation: captured commit only + +Parent: slice 39 at `ecef7776a4b38e0c7e0b86d7009fcd01ed7658e5`. +Remote master remains `5a28f7acf8a8c86536c4f5c5369bfeb49a46c41a`. + +The instruction service owns eligibility, saved retry times, and save attempts. +The existing scoped delivery worker owns coalescing, company-fair batches of 20, +reconstruction, pause/resume, failure backoff, and shutdown. Inventory uses scalar +fields and keyset pages of 100. Candidate bytes are read only for an actual attempt. +No table, migration, provider-specific scheduler, or Cloud alarm contract is added. + +The pending predicate remains `state = pending_commit` and `attempts < 3`. +The actual `nextAttemptAt` is used; null means due now. Completed copies, +prepared/live copies, permission conflicts, and exhausted candidates are quiet. +The three-attempt bound and preserved-candidate policy are unchanged. A bounded +business retry is distinct from a database/inventory failure: the latter retains +shared recovery backoff rather than discarding an unconfirmed obligation. + +The existing copy mutation boundary emits intent before writes that enter or +leave pending commit. Other copy mutations keep their existing query boundary. +The notification and update share a short transaction, including a caller-owned +outer transaction. Filesystem reads, file replacement, directory merge locks, and +provider transport operations are not enclosed in that new transaction. Company +and agent deletion signal company scope before cascading run deletion. Worktree +execution settings signal reconstruction; the heartbeat facade reads that policy +fresh and gates drain/standby. Local process signals do not cover independent SQL +writers or another replica; this remains the existing supported writer model. + +A worker rereads the candidate and atomically claims an attempt using the existing +state, attempt, baseline hash, and retained-owner guards. It also requires the +saved deadline still to be due at claim time. A newer future deadline wins without +consuming an attempt. Manual/startup recovery delegates to this same owner. The +existing revision service retains permission checks, canonical-file locking, +exact-byte replay receipts, and baseline conflicts. Shutdown joins a save already +admitted; it does not discard the persisted candidate to become quiet. + +Only the commit portion of `recoverCaptured` is excluded from periodic reaping. +Stopped-copy collection and post-save directory release keep their prior recovery +path for separate slices. The existing directory release after a successful save +remains part of that foreground settlement; its independent crash-recovery sweep +is not claimed migrated. + +## Evidence and failure record + +- Final instruction suite: 33 passed, including 13 worker cases, in 10.23 seconds. + Covers quiet prepared/completed/conflict/exhausted history, real capture after a + filesystem failure, actual saved deadline, restart, drain, standby, outer commit + and rollback, existing attempt cap, permission conflict, foreground save race, + shutdown join, newer-deadline race, worktree setting reactivation, and 103 saved + candidates across multiple inventory pages. Tests verify canonical bytes and + the save receipt, including one activity receipt after concurrent exact saves. +- Existing directory and startup acceptance: 137 passed on fresh fixtures. +- Final startup and heartbeat instruction-release wiring: 64 passed. Startup is + counted in both selections; the three final suite groups cover 184 distinct tests. +- Workspace typecheck/build passed. Final server typecheck passed again + after the manual recovery path adopted the common owner. +- Existing pending-copy index plan: with 10,000 synthetic completed rows and one + eligible candidate, PostgreSQL used `agent_instruction_copies_pending_idx`, read + one candidate and filtered two exhausted rows, with five shared buffer hits and + 0.011 ms execution. The fixture transaction rolled back. This only demonstrates + that completed history need not be scanned; it is not a fleet benchmark or a + constant-cost claim for arbitrary exhausted-pending history. +- Initial worker run: 26 passed, one drain fixture failed. The fixture disabled + transaction reconciliation as well as work admission, so the intent hold correctly + remained active. Drain now permits reconciliation; a separate standby case keeps + it disabled and verifies no passive SQL while retaining the hold. +- First repeated directory run: 106 passed, 29 failed because its reused external + database retained rows naming the prior run's removed temporary directories. + Fresh database plus fresh filesystem fixtures passed all 137 tests. The production + source was not changed to hide those failures. These filesystem suites require + a fresh external database for each full run, as embedded mode already provides. +- A new race regression failed with the candidate incorrectly saved before a + newly postponed deadline. Adding the due-time predicate to the attempt claim + made the focused regression pass, followed by all 33 instruction cases. + +## Boundaries and remaining acceptance + +Related warm-directory ownership work: #14695. Its stop/retention guarantees stay +with the directory owner. No terminal status is promoted into process-stop proof. +No provider is launched to recover a captured legacy instruction candidate. + +Current PR CI and fresh review are still required. Host suspension, database +suspension, external wake delivery, the two remaining instruction duties, the +other run-reaper duties, and the complete original stack inventory remain open. +This stack has no merge authorization. diff --git a/server/src/__tests__/agent-instruction-working-copies.test.ts b/server/src/__tests__/agent-instruction-working-copies.test.ts index 81479d07fcc..cc19f73d799 100644 --- a/server/src/__tests__/agent-instruction-working-copies.test.ts +++ b/server/src/__tests__/agent-instruction-working-copies.test.ts @@ -1,3 +1,8 @@ +import { heartbeatService } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyInstructionCommit } from "../services/instruction-commit-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import fs from "node:fs/promises"; import { execFile as execFileCallback } from "node:child_process"; @@ -9,9 +14,9 @@ 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, vi } from "vitest"; -import { eq } from "drizzle-orm"; -import { agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, agentInstructionWorkingCopies, createDb } from "@paperclipai/db"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { activityLog, agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, agentInstructionWorkingCopies, createDb } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { agentInstructionRevisionService } from "../services/agent-instruction-revisions.js"; import { agentInstructionWorkingCopyService } from "../services/agent-instruction-working-copies.js"; @@ -74,6 +79,281 @@ describe("registered run instruction copies", () => { await fs.writeFile(path.join(root, entryFile), initial); }); + describe("captured instruction commit worker", () => { + afterEach(() => { vi.useRealTimers(); vi.restoreAllMocks(); }); + const clock = () => vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + async function captured(copy: NonNullable>>, dueAt = new Date(), owner = db) { + await owner.transaction(async tx => { + await notifyInstructionCommit(tx, { companyId: copy.companyId, runId: copy.runId }); + await tx.update(agentInstructionWorkingCopies).set({ state: "pending_commit", attempts: 1, + candidateBase64: Buffer.from("recovered bytes").toString("base64"), nextAttemptAt: dueAt, + processStoppedAt: new Date(), + }).where(eq(agentInstructionWorkingCopies.runId, copy.runId)); + }); + } + async function saved(runId: string) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(100); + expect((await copies.get(companyId, runId))?.state).toBe("saved"); + }, { timeout: 5000 }); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("recovered bytes"); + } + function coordinator(canRun = () => true, canReconcile = () => true) { + return createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile, onError: error => { throw error; } }); + } + it("stays quiet for prepared copies, completed history, conflicts, and exhausted candidates", async () => { + for (const state of ["prepared", "saved", "conflict", "pending_commit"]) { + const copy = await run(); + await db.update(agentInstructionWorkingCopies).set({ state, attempts: 3 }).where(eq(agentInstructionWorkingCopies.runId, copy.runId)); + } + clock(); const worker = coordinator(); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }); + it("recovers a real capture at its saved deadline after a transient save failure", async () => { + const copy = await run(); + await fs.writeFile(path.join(copy.localRoot, entryFile), "recovered bytes"); + clock(); const worker = coordinator(); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + const rename = fs.rename.bind(fs); + let fail = true; + const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => { + if (fail && to === path.join(root, entryFile)) { fail = false; throw new Error("temporary file replacement failure"); } + return rename(from, to); + }); + const pending = await copies.collectStopped({ companyId, runId: copy.runId }); + expect(pending).toMatchObject({ state: "pending_commit", attempts: 1, nextAttemptAt: expect.any(Date) }); + renameSpy.mockRestore(); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(worker.nextWakeAt()).toBe(pending!.nextAttemptAt!.getTime()); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial); + await vi.advanceTimersByTimeAsync(30_000); + await saved(copy.runId); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }); + it.each(["restart", "drain", "standby"] as const)("reconstructs a saved candidate across %s", async mode => { + const copy = await run(); + if (mode === "restart") await captured(copy); + let admitted = mode === "restart"; + clock(); const worker = coordinator(() => admitted, () => mode !== "standby"); + try { + await agentInstructionWorkingCopyService(db).registerCapturedCommitWork(worker, () => admitted, async () => true).ready; + if (mode !== "restart") { + await captured(copy); + await vi.advanceTimersByTimeAsync(1000); + if (mode !== "standby") await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); + } + await saved(copy.runId); + } finally { await worker.stop(); } + }); + it("preserves the existing attempt cap and becomes quiet with the candidate retained", async () => { + const copy = await run(); await fs.writeFile(path.join(copy.localRoot, entryFile), "recovered bytes"); + clock(); const worker = coordinator(); + const rename = fs.rename.bind(fs); + const writes = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => { + if (to === path.join(root, entryFile)) throw new Error("file replacement still unavailable"); + return rename(from, to); + }); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + expect((await copies.collectStopped({ companyId, runId: copy.runId }))?.attempts).toBe(1); + for (const attempts of [2, 3]) { + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(100); + expect((await copies.get(companyId, copy.runId))?.attempts).toBe(attempts); + }); + } + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const count = writes.mock.calls.length; + await vi.advanceTimersByTimeAsync(300_000); + expect(writes.mock.calls.length).toBe(count); + expect(worker.nextWakeAt()).toBeNull(); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial); + expect(await copies.get(companyId, copy.runId)).toMatchObject({ state: "pending_commit", attempts: 3, + candidateBase64: Buffer.from("recovered bytes").toString("base64"), nextAttemptAt: null }); + } finally { writes.mockRestore(); await worker.stop(); } + }); + it("joins an active save during shutdown and reconstructs its remaining receipt", async () => { + const copy = await run(); await captured(copy); + clock(); const worker = coordinator(); + const rename = fs.rename.bind(fs); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const reached = new Promise(resolve => { entered = resolve; }); + const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => { + if (to === path.join(root, entryFile)) { entered(); await held; } + return rename(from, to); + }); + let stopped: Promise | undefined; + try { + const registration = copies.registerCapturedCommitWork(worker, () => true, async () => true); + await reached; + let complete = false; + stopped = worker.stop().then(() => { complete = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(complete).toBe(false); + release(); await stopped; await registration.ready; + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("recovered bytes"); + expect((await copies.get(companyId, copy.runId))?.state).toBe("saved"); + const restarted = coordinator(); + try { + await copies.registerCapturedCommitWork(restarted, () => true, async () => true).ready; + expect(restarted.nextWakeAt()).toBeNull(); + } finally { await restarted.stop(); } + } finally { release(); renameSpy.mockRestore(); await stopped; await worker.stop(); } + }); + it("honors a newer retry deadline written before its attempt claim", async () => { + const copy = await run(); await captured(copy); + clock(); const worker = coordinator(); + const later = new Date(Date.now() + 60_000); + const transaction = db.transaction.bind(db); + let changed = false; + const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => { + if (!changed) { + changed = true; + await transaction(async tx => { + await notifyInstructionCommit(tx, { companyId, runId: copy.runId }); + await tx.update(agentInstructionWorkingCopies).set({ nextAttemptAt: later }) + .where(eq(agentInstructionWorkingCopies.runId, copy.runId)); + }); + } + return transaction(callback, options); + }) as typeof db.transaction); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + expect(changed).toBe(true); + expect(await copies.get(companyId, copy.runId)).toMatchObject({ state: "pending_commit", attempts: 1, nextAttemptAt: later }); + transactions.mockRestore(); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(worker.nextWakeAt()).toBe(later.getTime()); + await vi.advanceTimersByTimeAsync(60_000); await saved(copy.runId); + } finally { transactions.mockRestore(); await worker.stop(); } + }); + it("reactivates after a real worktree setting change without polling disabled settings", async () => { + const copy = await run(); await captured(copy); + const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "sleep40-instruction-activation" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const service = heartbeatService(db, { runtimeEnv }); + clock(); const worker = coordinator(); + try { + await service.registerCapturedInstructionCommitWork(worker, () => true).ready; + expect(worker.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await service.resolveSchedulingSuppression(); + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await saved(copy.runId); + } finally { await worker.stop(); await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + }); + it("reconstructs more than one page and drains the saved candidates in bounded batches", async () => { + const copy = await run(); + const ids = Array.from({ length: 103 }, () => randomUUID()); + await db.insert(heartbeatRuns).values(ids.map(id => ({ id, companyId, agentId, + invocationSource: "on_demand", responsibleUserId: userId }))); + await db.insert(agentInstructionWorkingCopies).values(ids.map(runId => ({ ...copy, runId, + state: "pending_commit", attempts: 1, nextAttemptAt: new Date(), processStoppedAt: new Date(), + candidateBase64: Buffer.from("recovered bytes").toString("base64"), + }))); + clock(); const worker = coordinator(); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + const rows = () => db.select({ state: agentInstructionWorkingCopies.state }).from(agentInstructionWorkingCopies) + .where(eq(agentInstructionWorkingCopies.companyId, companyId)); + expect((await rows()).filter(row => row.state === "saved").length).toBeLessThanOrEqual(20); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await rows()).filter(row => row.state === "saved")).toHaveLength(103); + }, { timeout: 15_000 }); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("recovered bytes"); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }, 20_000); + it("requires outer commit and ignores a nested rollback", async () => { + const copy = await run(); clock(); const worker = coordinator(); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + await expect(db.transaction(async tx => { + await captured(copy, new Date(), tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(1000); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial); + throw new Error("outer rollback"); + })).rejects.toThrow("outer rollback"); + expect((await copies.get(companyId, copy.runId))?.state).toBe("prepared"); + await db.transaction(async tx => { + await captured(copy, new Date(), tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(1000); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial); + }); + await saved(copy.runId); + } finally { await worker.stop(); } + }); + it("preserves a permission conflict instead of retrying it forever", async () => { + const copy = await run(); await captured(copy); + await db.delete(principalPermissionGrants).where(eq(principalPermissionGrants.companyId, companyId)); + clock(); const worker = coordinator(); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await copies.get(companyId, copy.runId))?.state).toBe("conflict"); + }); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial); + expect((await copies.get(companyId, copy.runId))?.candidateBase64).toBe(Buffer.from("recovered bytes").toString("base64")); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }); + it("joins a foreground save and recovers one receipt without duplicating the save", async () => { + const copy = await run(); await fs.writeFile(path.join(copy.localRoot, entryFile), "recovered bytes"); + clock(); const worker = coordinator(); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const reached = new Promise(resolve => { entered = resolve; }); + let foreground: ReturnType | undefined; + const rename = fs.rename.bind(fs); + let holding = true; + const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => { + if (holding && to === path.join(root, entryFile)) { holding = false; entered(); await held; } + return rename(from, to); + }); + try { + await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + foreground = copies.collectStopped({ companyId, runId: copy.runId }); + await reached; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await copies.get(companyId, copy.runId))?.attempts).toBe(2); + }); + release(); await foreground; await saved(copy.runId); + const events = await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), eq(activityLog.action, "agent.files_updated"))); + expect(events).toHaveLength(1); + } finally { release(); await foreground; renameSpy.mockRestore(); await worker.stop(); } + }); + }); + it("persists an atomic editor replacement, ignores repository instructions, and gives the next run exact bytes", async () => { const copy = await run(); await fs.writeFile(path.join(home, "AGENTS.md"), "Repository text must not become agent instructions"); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 377a633ca99..5ba23b6745e 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -107,6 +107,7 @@ const { reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })), reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerCapturedInstructionCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })), resumeQueuedRuns: vi.fn(async () => undefined), recoverPendingSessionGoalActions: vi.fn(async () => ({ scanned: 0, enqueued: 0, alreadyQueued: 0, invalid: 0 })), @@ -539,6 +540,13 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers captured instruction recovery only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerCapturedInstructionCommitWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerCapturedInstructionCommitWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers retry cancellation recovery only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -896,7 +904,7 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index db158e77e90..38cd7ec4217 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1773,6 +1773,9 @@ async function startServerWithDatabaseTeardown( if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + if (config.heartbeatSchedulerEnabled) await heartbeat.registerCapturedInstructionCommitWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + const runRetentionSweep = async () => { const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active")); let archived = 0; @@ -1900,7 +1903,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. trackHeartbeatSchedulerWork(heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index 54ff9a8f3bc..b2f4fe2bf0d 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -1,3 +1,4 @@ +import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js"; import { and, eq, or, sql } from "drizzle-orm"; import { agents, agentApiKeys, agentRuntimeState, agentTaskSessions, agentWakeupRequests, activityLog, @@ -44,6 +45,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str await tx.delete(issueExecutionDecisions).where(eq(issueExecutionDecisions.actorAgentId, id)); await tx.delete(issueComments).where(eq(issueComments.authorAgentId, id)); await notifyScheduledRetryCancellation(tx, { companyId: companyId }); + await notifyInstructionCommit(tx, { companyId: companyId }); await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, id)); await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, id)); await tx.delete(agentApiKeys).where(eq(agentApiKeys.agentId, id)); diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts index 6eb2c974b6e..325d7d7256c 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -1,10 +1,14 @@ +import { notifyInstructionCommit, type InstructionCommitScope } from "./instruction-commit-notifications.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; import { agentDirectoryWorkingCopyService, isAgentDirectoryCopy } from "./agent-directory-working-copies.js"; import { createHash } from "node:crypto"; import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; import fs from "node:fs/promises"; import path from "node:path"; -import { and, asc, desc, eq, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm"; +import { and, asc, desc, eq, gt, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm"; import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db"; import { prepareAdapterExecutionTargetRuntime, @@ -73,11 +77,22 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen const [row] = await db.select().from(copies).where(scope(companyId, runId)); return row ?? null; } + async function tryPatch(row: Copy, values: Partial, requireDue = false) { + const signalsCommit = row.state === "pending_commit" || values.state === "pending_commit"; + const write = async (tx: Db) => { + if (signalsCommit) + await notifyInstructionCommit(tx, { companyId: row.companyId, runId: row.runId }); + const [updated] = await tx.update(copies).set({ ...values, updatedAt: new Date() }).where(and( + scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash), + sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`, + requireDue ? or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())) : undefined, + )).returning(); + return updated; + }; + return signalsCommit ? db.transaction(tx => write(tx as unknown as Db)) : write(db); + } async function patch(row: Copy, values: Partial) { - const [updated] = await db.update(copies).set({ ...values, updatedAt: new Date() }).where(and( - scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash), - sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`, - )).returning(); + const updated = await tryPatch(row, values); // A late cleanup must not overwrite an explicit resolution or a newer // baseline. Canonical content has its own independent head CAS. return updated ?? (await get(row.companyId, row.runId))!; @@ -300,16 +315,13 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen } /** Restarts can retry persisted bytes without touching a live filesystem or launching a model. */ - async function recoverCaptured() { - const pending = await db.select().from(copies).where(and( + async function recoverCaptured(options: { skipCommit?: boolean } = {}) { + const pending = options.skipCommit ? [] : await db.select().from(copies).where(and( eq(copies.state, "pending_commit"), or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())), lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1), )).limit(20); - for (const row of pending) { - const result = await commitCandidate(await patch(row, { attempts: row.attempts + 1 })); - if (isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result); - } + for (const row of pending) await recoverCommit(commitEntry(row)!, () => true); // A crash between a durable save and release must not retain run snapshots // forever. Only discard copies with recorded stop proof and a terminal receipt. const cleanup = await db.select().from(copies).where(and( @@ -374,6 +386,53 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen return pending.length; } + type CommitEntry = { id: string; companyId: string; dueAt: number }; + function commitEntry(row: Pick | null): CommitEntry | null { + return row?.state === "pending_commit" && row.attempts < MAX_COLLECTION_ATTEMPTS + ? { id: row.runId, companyId: row.companyId, dueAt: row.nextAttemptAt?.getTime() ?? 0 } : null; + } + async function loadCommits(scope: InstructionCommitScope | undefined, admitted: () => boolean) { + const entries: CommitEntry[] = []; + let cursor: string | undefined; + while (admitted()) { + const rows = await db.select({ runId: copies.runId, companyId: copies.companyId, state: copies.state, + attempts: copies.attempts, nextAttemptAt: copies.nextAttemptAt }).from(copies).where(and( + eq(copies.state, "pending_commit"), lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1), + scope ? eq(copies.companyId, scope.companyId) : undefined, + scope?.runId ? eq(copies.runId, scope.runId) : undefined, + cursor ? gt(copies.runId, cursor) : undefined, + )).orderBy(asc(copies.runId)).limit(100); + for (const row of rows) entries.push(commitEntry(row)!); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].runId; + } + throw new Error("Instruction commit reconstruction stopped"); + } + async function recoverCommit(entry: CommitEntry, admitted: () => boolean) { + const row = await get(entry.companyId, entry.id); + const current = commitEntry(row); + if (!row || !current || current.dueAt > Date.now()) return current; + if (!admitted()) throw new Error("Instruction commit recovery stopped"); + const claimed = await tryPatch(row, { attempts: row.attempts + 1 }, true); + if (claimed) { + const result = await commitCandidate(claimed); + if (isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result); + } + return commitEntry(await get(entry.companyId, entry.id)); + } + function registerCapturedCommitWork(coordinator: ReturnType, + canRun: () => boolean, canRecover: () => Promise) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.instructionCommit, + batchSize: 20, + decodeScope: (hint: string): InstructionCommitScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: CommitEntry, scope?: InstructionCommitScope) => !scope || + entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId), + load: async (scope, signal) => await canRecover() ? loadCommits(scope, () => canRun() && !signal?.aborted) : [], + check: async (entry, signal) => await canRecover() ? recoverCommit(entry, () => canRun() && !signal?.aborted) : null, + }, canRun); + } + async function list(companyId: string, agentId: string, actor: AuthorizationActor) { await revisions.readCurrent({ companyId, agentId }, actor); const rows = await db.select().from(copies).where(and(eq(copies.companyId, companyId), eq(copies.agentId, agentId), inArray(copies.state, ["conflict", "pending_collection", "pending_commit", "unavailable"]))).orderBy(desc(copies.createdAt)).limit(50); @@ -422,5 +481,5 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen } return row; } - return { prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release }; + return { registerCapturedCommitWork, prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release }; } diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 600f9598c8d..8e0e5db3520 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js"; import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { chatCompanyDeletion } from "./chat-channels/company-deletion.js"; @@ -96,6 +97,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(activityLog).where(eq(activityLog.companyId, id)); await tx.delete(runIdentityContexts).where(eq(runIdentityContexts.companyId, id)); await notifyScheduledRetryCancellation(tx, { companyId: id }); + await notifyInstructionCommit(tx, { companyId: id }); await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, id)); await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, id)); await tx.delete(agentApiKeys).where(eq(agentApiKeys.companyId, id)); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index fc04f316134..bbee404cecc 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + instructionCommit: "instruction-commit", scheduledRetryCancellation: "scheduled-retry-cancellation", executionFinalization: "execution-finalization", executionReplacement: "execution-replacement", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c4ee17f0fa0..22bb1fdb794 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -4062,6 +4062,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerCapturedInstructionCommitWork: ( + coordinator: Parameters[0], canRun: () => boolean, + ) => instructionCopies.registerCapturedCommitWork(coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerScheduledRetryCancellationWork: ( coordinator: Parameters[0], canRun: () => boolean, ) => registerScheduledRetryCancellationWork(coordinator, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 6b70216769e..74651303c38 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1915,7 +1915,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean }) { await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1925,7 +1925,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD await instructionCopies.recoverStopped().catch(error => { logger.warn({ err: error }, "failed to recover stopped instruction copies"); }); - await instructionCopies.recoverCaptured().catch(error => { + await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit }).catch(error => { logger.warn({ err: error }, "failed to retry captured instruction revisions"); }); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 495d16cbe67..6a8894e4f45 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js"; import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; @@ -617,6 +618,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); + if (patch.enableWorktreeRunExecution !== undefined) await notifyInstructionCommit(tx); if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx); await notifyToolConnectionHealth(tx, "*"); return tx diff --git a/server/src/services/instruction-commit-notifications.ts b/server/src/services/instruction-commit-notifications.ts new file mode 100644 index 00000000000..8e75837a429 --- /dev/null +++ b/server/src/services/instruction-commit-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type InstructionCommitScope = { companyId: string; runId?: string }; + +export async function notifyInstructionCommit(transaction: object, scope?: InstructionCommitScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.instructionCommit, scope ? JSON.stringify(scope) : "*"); +}