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 fb1924e10a2..4b0c0f25e8e 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2535,3 +2535,37 @@ real worker and owner acceptance passed, but startup wiring coverage was missing from the local selection. The double now exposes the method, and new assertions verify scheduler enablement and periodic-reaper exclusion. All 48 startup tests pass in 5.34 seconds. This is a test-only repair; fresh head CI/review is required. + +Slice 39 is draft [#17667](https://github.com/paperclipai/paperclip/pull/17667), +initial head `bd490ee50b303187f0ae6f8fe2aa95477d66a40f`, targeting slice 38. +Its initial review was 5/5. The startup repair is now pushed at `ecef7776a4`; +fresh CI and review are required. No merge or full-stack acceptance is claimed. + +The instruction-copy audit splits the apparent two reaper calls into three +reasons: stopped collection, captured-byte commit, and post-save directory release. +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 new file mode 100644 index 00000000000..bf91963ca71 --- /dev/null +++ b/doc/plans/2026-10-25-instruction-recovery-sleep-audit.md @@ -0,0 +1,364 @@ +# Instruction recovery sleep audit + +Inspected from slice 39 head bd490ee50b303187f0ae6f8fe2aa95477d66a40f. +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 + AND independently established local/remote process stop. Never infer stop from + terminal status. Temporary staging failure has its own pre-launch guarantee. +2. recoverCaptured pending loop: retry saving durable candidate bytes, state + pending_commit, attempts <=2, nextAttemptAt null or due. Canonical authorization + and base comparison remain in agentInstructionRevisionService.commit. +3. recoverCaptured cleanup loop: release temporary agent directories after a + terminal save receipt plus processStoppedAt, baseline/cleanupPending marker, + and due nextAttemptAt. Preserve remote identity and cleanup authority. + +Keep one reason per PR. Captured commit is the smallest independent next slice; +stopped collection and post-save release must remain on the existing reaper until +separately migrated. Do not retire recoverCaptured wholesale with only save covered. + +Durability already exists in agent_instruction_working_copies, defined in +packages/db/src/schema/agent_instruction_revisions.ts. Existing index +agent_instruction_copies_pending_idx(state,next_attempt_at) may suffice; inspect +EXPLAIN before adding an index. No new table needed. Attempts and nextAttemptAt +are business recovery state; do not reset the three-attempt bound. + +Producer map: +- agent-instruction-working-copies.ts::patch owns CAS writes with state, attempts, + baseHash, retainedByRunId predicates. Add before-write signal in short existing/ + new DB transaction, without enclosing file/network work. +- collectStopped captures bytes via patch(pending_commit) before calling commit. + commitCandidate updates saved/conflict/retry state via patch; resolve also patch. +- prepare inserts preparing rows; no captured work until patch. acknowledgeExplicitSave + only advances prepared baseline, not pending work. +- agent-directory-working-copies.ts receives patch for ordinary mutations. Its + direct inserts/adoption/deletes need explicit audit but currently create prepared + or staging rows, not pending_commit. Do not assume legacy candidate payloads + and agent-directory receipts have the same semantics. +- Run/company/agent deletion cascades can remove obligations: signal exact scope + before supported deletion to disarm stale entries. + +Concurrency risk to test before relying on events: +A pending_commit write can wake recovery while foreground commit is still active. +Revision commit serializes through its owner and explicitly deduplicates exact +candidate bytes before testing baseline equality. Copy CAS can lose to a competing +attempt counter update; a later worker must recover the receipt without falsely +reporting conflict or saving twice. Test actual producer plus held foreground +commit and verify canonical bytes, one durable activity receipt, and final state. +No transaction should contain process/file collection just to emit a signal. + +Stopped-copy producer audit is wider: +- terminal run transitions (including direct recovery writers) +- copies.processStoppedAt and retention/handoff state +- native.local_process_stopped via recordNativeLocalProcessStop in heartbeat, + native restart recovery, and native-local-process-stop itself +- later process-start/identity events invalidate local stop receipts +- environment lease remoteExecutionTermination receipts and removal/reacquisition + via environment-runtime, environments, task runtime, workspace recovery/export +- stalled unconfirmed owners are real pending obligations; preserve safe backoff + or a complete notification replacement. Never claim quiet by dropping them. + +Testing planned: +Real managed instruction file and authorization, actual collectStopped producer; +no-config/prepared/history/exhausted/conflict quiet; exact saved deadline; transient +save error; rollback/outer commit; foreground race; restart; drain/standby; shutdown; +more than one page and company fairness. Assert file bytes and save receipts, not +only callback invocation. Existing test harnesses currently require embedded DB; +use guarded external disposable URLs here as needed, maintaining embedded default. + +Remaining acceptance of 39 and older stack gates stays outstanding. No merge +is authorized. No claim about host suspension or expanded Cloud alarm coverage. + +Baseline acceptance before changing production behavior: +- 20 instruction working-copy tests passed in 6.83 seconds. +- 87 persistent agent-directory tests passed in 18.27 seconds. +- 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. +The later stack campaign authorizes an ordinary protected merge only after exact-current-head Greptile 5/5, clean CI, and required review findings are resolved. + +## Foreground ownership correction (2026-10-27) + +Review of PR #17670 found that the original foreground race assertion accepted +an extra recovery attempt while the first save still held the canonical-file +operation. Exact-byte deduplication prevented a duplicate write but did not protect +the copy receipt: an incremented attempt invalidated the foreground failure CAS +and could lose its 30-second retry deadline. + +The instruction-copy service now serializes collection and captured recovery per +company/run using the existing database work scope. Separate service instances and +dedicated connections belonging to that scope share the same in-flight owner. +Recovery joins the live operation, then rereads durable state and its deadline +before claiming an attempt. Rejected operations release ownership; no elapsed-time +lease, new table, polling loop, or transaction around filesystem work is added. +This is process-local ownership within the existing supported writer model, not a +cross-replica lock. Persisted candidates still reconstruct after process restart. + +Regression evidence against owner head d1a7c64c8b03ada0741c7c5ebb8ff9cb5190479b: +- The six-case real-save barrier regression failed in five cases before the fix: + attempts advanced from one to two while foreground file replacement was held. + The shared-service success case passed; the result is not described as six failures. +- With the fix, all 38 instruction-copy tests and 14 heartbeat helper tests pass. + Tests cover successful and failed foreground saves across shared, separate, and + dedicated-connection services, a two-minute held operation, the preserved retry + deadline, eventual canonical bytes, and exactly one file-update activity receipt. +- All 87 persistent agent-directory tests pass on a separate fresh PostgreSQL + database, covering the existing directory consumer of the collection boundary. +- Server typechecking passes with the expanded regression tests. +- Four embedded-only heartbeat integration cases were skipped by their existing + support gate; they are not included in the 139 passing cases above. This is not + evidence of full repository CI or deployed host/database suspension. + +## Retry fixture timing correction (2026-10-27) + +A later CI run exposed a test race, not a different production retry policy. +The attempt counter is written before filesystem replacement completes. The old +attempt-cap test waited for that counter, advanced a fixed 31 seconds, and could +move the clock before the pending save recorded its next retry deadline. It then +expected attempt three before that newly recorded deadline was due. + +Holding the second filesystem replacement across that clock advance reproduced +the exact CI assertion: expected three attempts, received two. The test now waits +for the completed attempt's persisted deadline and advances to that deadline. +It retains the deliberate slow-save barrier, the three-attempt cap, the original +canonical bytes, the captured candidate, and the absence of later retries. The +captured-commit group also retires its own company fixture during teardown, so a +failed test cannot leave retry debt in the next test's global reconstruction. + +All 38 instruction-copy cases pass on a fresh PostgreSQL 17 database with the +owner branch's 396 migrations. The controlled reproduction fails without the +fixture correction. No production retry code, timeout, or assertion is weakened. + + +## Current parent composition and acceptance (2026-10-28) + +The existing slice was integrated with the updated cancellation parent +`d9da519b32bd7e5ecaf2156c3de217fc47187f50` by a normal merge, yielding +`77cd93fa08310ecdc834964e9a4326f794c2be87`. All substantive owned edits remain +identical after excluding unchanged context and hunk coordinates. The literal +patch comparison initially rejected the updated context; zero-context comparison +and inspection establish that the edits themselves are unchanged. Every unrelated +upstream blob is preserved. The entire database source tree is identical to the +parent: this slice has no migration or table. + +All six complete files passed, **268 cases with no skips**, in 46.26 seconds: + +| Complete file | Passing cases | +| --- | ---: | +| Instruction working copies and captured commits | 38 | +| Persistent agent directory working copies | 87 | +| Server startup and feedback export | 53 | +| Heartbeat instruction-copy boundary | 14 | +| Heartbeat recovery | 45 | +| Heartbeat run lifecycle | 31 | + +Instruction copies, directory copies, and recovery each used a separate newly +created PostgreSQL database migrated through the current 0404 parent history +before fixture setup. The lifecycle suite retained its real embedded database. +No reused filesystem database, skipped suite, increased timeout, or narrowed +case-name selection is counted as acceptance. The earlier failures above remain +part of the record. + +Database build, production server typecheck and build, module boundaries, target +migration-order validation, and all three fresh migrations passed. File collection +and replacement stay outside the short pending-state transaction. Foreground +ownership, the persisted next deadline, attempt cap, authorization, exact-byte +receipt and drain/shutdown contracts retain their owners. No provider turn is +started to replay captured bytes. + +This is local worker and integration evidence. Final publication must carry the +actual landed parent ancestry, followed by fresh current-head CI and review. +It does not establish host or database suspension or a durable Cloud alarm. + + +## Review style correction and interrupted CI + +Greptile reviewed `cffe70b42b7b41f0b69c548dc47b8f1d5b753de5` at 5/5 but left +an actionable P2: CONTRIBUTING.md prohibits new code comments. Seven new +comment lines in the instruction-copy owner and tests are removed. The +foreground ownership, attempt-cap barrier and fixture isolation explanations +remain in this audit. TypeScript parsing verifies that both affected files have +exactly the same non-comment token stream before and after removal. All other +source bytes remain unchanged; this does not reinterpret the 268-case source +acceptance as a new whole-suite run. Fresh head CI and Greptile are required. + +The first cffe CI server shard 9 stopped after receiving a runner shutdown +signal at 23:50:45Z, then reported cancellation at 23:50:57Z. It has no final +assertion verdict and is not counted as passing. The log is retained; no old-head +retry is requested because the required style correction creates a new head. +The full new-head CI must run independently. + +## Current master and append-only parent adoption + +The parent retry-cancellation slice now carries actual #17666 squash +`7afee8f0622e273d10506b0febfb68f143d168b1`, the new Coordinator and Runner +changes, and the regenerated append-only migration 0406. A clean normal merge +preserves both complete added/removed patches and every unrelated upstream blob. +The complete database subtree equals parent +`48b3fcb83f305616df02a4337a42ec0867bb4b80`; this slice adds no migration. +The parent audit follow-up changes documentation only. + +Source `654c4694d5fad8ebfcc1219bc4a76ec94f4516b8` passes all **268 cases in six +complete files**, with no skipped cases, in 45.38 seconds. Four separate fresh +PostgreSQL 17 databases migrated through 0406 prevent fixture interference. +Instruction, directory, and recovery suites use their supported external paths. +The embedded-only lifecycle suite temporarily used its own external fixture +because this host cannot start more embedded clusters. Its fixture import was +restored byte for byte afterward. All assertions and original timing limits were +unchanged. This is qualified external-database acceptance; fresh CI still must +pass with the original embedded fixture. Database build, server typecheck/build, +module boundaries, target migration order, and all four fresh migrations pass. +No new code comments remain relative to this parent. + +Prior head `a1a0ae5b02e8a650408bf4c55ae99071c3bff59c` had passing CI, Greptile +5/5 and Ponytail 5/5. Those results become historical when the new head is +published. Keep the old runner interruption and all earlier failures in the +ledger. The current parent must merge first, and the new published head needs +fresh exact-head reviews and CI. These results do not establish deployed sleep. + +### Local validation command correction before publication + +A cloned local runner selected its first command by position, but this slice's +first command was the database build rather than migration order. It replaced +the build with a pnpm filter that matched no projects and exited zero. That was +not evidence of a database build. The migration-order command also retained old +revision arguments. The next unpublished slice exposed the error when its first +fresh migration could not find compiled migration assets. Publication stopped. +The original command receipts and logs remain; no product pass is inferred from +the no-match command or the failed migration. + +The runner now selects commands by their names. It executed the actual database +build and migration check against the exact current parent and head. The full +six-file acceptance then ran again at audit-only descendant +`b25636e6261f98f1e53ff28a1f259ea03b9c41a8`: **268 cases pass without skips in +40.97 seconds**, using four new disposable databases migrated through 0406. +Types, server build, boundaries, and the actual current migration-order check +also pass. The lifecycle fixture qualification above still applies; the file was +restored exactly. The already-published parent used a different, correct command +layout and was not affected by this runner error. + +## Bounded retry-cancellation parent acceptance + +A clean normal merge adopts published parent +`405e533394b26215e22d0452e9b0207d3f784a79`, including its bounded reconstruction +and durable browser reply check. Both complete added/removed patches and every +unrelated upstream blob are preserved. The entire database subtree still equals +that parent: this slice has no schema or migration change. + +Source `9fb75e508f06b05b740099403ce021a61e7c6603` passes all **268 cases in six +complete files**, without skips, in 41.89 seconds. Four unused fresh PostgreSQL 17 +databases migrated through 0406 isolate the fixtures. The same qualified temporary +external lifecycle fixture is restored byte for byte. Actual database build, +server types/build, module boundaries and exact-current migration order pass. +The earlier local runner failure and corrected evidence remain in the preceding +section; this run uses commands selected by name. + +Stacked publication may run this non-migration slice's independent CI while the +parent finishes. It does not allow merging into an unmerged parent. At merge, +the exact reviewed parent must have landed, the target must be master, and the +actual combined tree must preserve the complete owned patch and upstream changes. +The subsequent index-bearing child still needs the actual parent migration squash +ancestry before publication. Old reviews on this slice's prior head do not count +for the new published head. + +## Landed retry-cancellation squash ancestry + +Parent #17667 merged as `27270e9bf8b7a294cf03f8b84f4857f1b1ff05c7` after +all 58 checks and both current-head reviews passed. Its entire tree equals the +already-adopted reviewed parent `405e533394b26215e22d0452e9b0207d3f784a79`. +The ordinary merge probe still found nine overlapping text conflicts from the +squashed ancestry. A normal merge records that actual ancestry. The exact parent +was already an ancestor of the accepted descendant; restoring the accepted +versions of the inspected conflicts leaves the **entire result tree identical** +to the previously published `fc5cee9fa00923a2d6fba27386ad7a6ab0ef101e`. +This proves all automatic and manual resolutions preserve tested code and parent +changes. The follow-up here changes documentation only. The 268-case source +acceptance and database proofs remain applicable, but the new SHA needs fresh +CI and exact-head reviews. Prior stacked-head results are not reused as passes. diff --git a/server/src/__tests__/agent-directory-working-copies.test.ts b/server/src/__tests__/agent-directory-working-copies.test.ts index 69fbb1c7d0e..65353060433 100644 --- a/server/src/__tests__/agent-directory-working-copies.test.ts +++ b/server/src/__tests__/agent-directory-working-copies.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import fs from "node:fs/promises"; import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; @@ -22,6 +23,8 @@ import { AgentDirectoryReuseInvalidatedError, agentDirectoryWorkingCopyService } import { withDirectoryMergeLock } from "@paperclipai/adapter-utils/workspace-restore-merge"; import { heartbeatRunEvents } from "@paperclipai/db"; +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_DIRECTORY_TEST_DATABASE_URL"); + describe("persistent agent directories", () => { let database: Awaited>; let db: ReturnType; @@ -42,13 +45,14 @@ describe("persistent agent directories", () => { beforeAll(async () => { home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "instruction-working-copies-"))); process.env.PAPERCLIP_HOME = home; - database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-"); + db = createDb(external ?? database.connectionString); copies = agentInstructionWorkingCopyService(db); revisions = agentInstructionRevisionService(db); }, 90_000); afterAll(async () => { if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); if (home) { const writable = async (dir: string) => { diff --git a/server/src/__tests__/agent-instruction-working-copies.test.ts b/server/src/__tests__/agent-instruction-working-copies.test.ts index bd29eb5bc9b..85654fcb793 100644 --- a/server/src/__tests__/agent-instruction-working-copies.test.ts +++ b/server/src/__tests__/agent-instruction-working-copies.test.ts @@ -1,3 +1,9 @@ +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"; import { promisify } from "node:util"; @@ -8,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, withDedicatedDbConnection } 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"; @@ -18,6 +24,8 @@ import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js"; import { buildNativeRuntimeContext } from "../services/native-runtime/runtime-context.js"; +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_INSTRUCTION_COPY_TEST_DATABASE_URL"); + describe("registered run instruction copies", () => { let database: Awaited>; let db: ReturnType; @@ -38,13 +46,14 @@ describe("registered run instruction copies", () => { beforeAll(async () => { home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "instruction-working-copies-"))); process.env.PAPERCLIP_HOME = home; - database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-"); + db = createDb(external ?? database.connectionString); copies = agentInstructionWorkingCopyService(db); revisions = agentInstructionRevisionService(db); }, 90_000); afterAll(async () => { if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); if (home) { const writable = async (dir: string) => { @@ -70,6 +79,334 @@ describe("registered run instruction copies", () => { await fs.writeFile(path.join(root, entryFile), initial); }); + describe("captured instruction commit worker", () => { + afterEach(async () => { + vi.useRealTimers(); vi.restoreAllMocks(); + + await db.update(agentInstructionWorkingCopies).set({ state: "saved", nextAttemptAt: null }) + .where(eq(agentInstructionWorkingCopies.companyId, companyId)); + }); + 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); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const saving = new Promise(resolve => { entered = resolve; }); + let destinationWrites = 0; + const writes = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => { + if (to === path.join(root, entryFile)) { + if (++destinationWrites === 2) { entered(); await held; } + 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]) { + const previous = await copies.get(companyId, copy.runId); + expect(previous?.nextAttemptAt).toBeInstanceOf(Date); + await vi.advanceTimersByTimeAsync(Math.max(0, previous!.nextAttemptAt!.getTime() - Date.now()) + 1); + if (attempts === 2) { + await saving; + + + expect((await copies.get(companyId, copy.runId))?.attempts).toBe(2); + await vi.advanceTimersByTimeAsync(31_000); + release(); + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(100); + const row = await copies.get(companyId, copy.runId); + expect(row?.attempts).toBe(attempts); + if (attempts < 3) expect(row!.nextAttemptAt!.getTime()).toBeGreaterThan(Date.now()); + else expect(row?.nextAttemptAt).toBeNull(); + expect(idleWorkSnapshot().active).toBe(0); + }); + } + 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 { release(); 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.each((["shared", "separate", "dedicated"] as const).flatMap(mode => + (["saved", "retry"] as const).map(outcome => ({ mode, outcome }))))("joins foreground ownership without consuming a recovery attempt ($mode, $outcome)", async ({ mode, outcome }) => { + const copy = await run(); await fs.writeFile(path.join(copy.localRoot, entryFile), "recovered bytes"); + const exercise = async (workerDb: typeof db) => { + const recoveryCopies = mode === "shared" ? copies : agentInstructionWorkingCopyService(workerDb); + 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; + if (outcome === "retry") throw new Error("temporary foreground replacement failure"); + } + return rename(from, to); + }); + try { + await recoveryCopies.registerCapturedCommitWork(worker, () => true, async () => true).ready; + foreground = copies.collectStopped({ companyId, runId: copy.runId }); + await reached; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + }); + + + await vi.advanceTimersByTimeAsync(120_000); + expect(await copies.get(companyId, copy.runId)).toMatchObject({ state: "pending_commit", attempts: 1 }); + release(); + const result = await foreground; + if (outcome === "retry") { + expect(result).toMatchObject({ state: "pending_commit", attempts: 1, nextAttemptAt: expect.any(Date) }); + expect(result!.nextAttemptAt!.getTime()).toBeGreaterThan(Date.now() + 29_000); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(worker.nextWakeAt()).toBe(result!.nextAttemptAt!.getTime()); + await vi.advanceTimersByTimeAsync(Math.max(0, result!.nextAttemptAt!.getTime() - Date.now() - 5000)); + expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial); + expect((await copies.get(companyId, copy.runId))?.attempts).toBe(1); + await vi.advanceTimersByTimeAsync(5000); + } else expect(result).toMatchObject({ state: "saved", attempts: 1 }); + 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(); } + }; + if (mode === "dedicated") await withDedicatedDbConnection(db, exercise); + else await exercise(db); + }); + }); + 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 d4abf20fe82..a8fe2e8dbc5 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -115,6 +115,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 })), @@ -550,6 +551,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(); @@ -941,7 +949,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 4b30bfdad2d..b059c4aad85 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1820,6 +1820,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; @@ -1947,7 +1950,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. runPeriodicRecovery(() => 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..ef5b7a21ceb 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -1,11 +1,15 @@ +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 { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db"; +import { and, asc, desc, eq, gt, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm"; +import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, databaseWorkScopeKey, type Db } from "@paperclipai/db"; import { prepareAdapterExecutionTargetRuntime, runAdapterExecutionTargetShellCommand, @@ -32,6 +36,23 @@ const MAX_COLLECTION_ATTEMPTS = 3; // supply one. These handles are optional: durable captured bytes survive restart. const liveTargets = new Map(); const targetKey = (companyId: string, runId: string) => `${companyId}:${runId}`; +const copySaves = new WeakMap>>(); + +function serializeCopySave(db: Db, companyId: string, runId: string, save: () => Promise): Promise { + const scope = databaseWorkScopeKey(db); + let active = copySaves.get(scope); + if (!active) copySaves.set(scope, active = new Map()); + const key = targetKey(companyId, runId); + + + const operation = (active.get(key) ?? Promise.resolve()).then(save); + const settled = operation.then(() => undefined, () => undefined); + active.set(key, settled); + return operation.finally(() => { + if (active.get(key) === settled) active.delete(key); + }); +} + /** Finish only bounded immediate collection work; deferred ownership remains pending. */ export async function collectStoppedInstructionCopyWithRetries>(collect: () => Promise): Promise { @@ -73,11 +94,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))!; @@ -277,7 +309,10 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen } /** Caller must have joined the owned provider process before granting this proof. */ - async function collectStopped(input: { companyId: string; runId: string; target?: AdapterExecutionTarget | null }) { + function collectStopped(input: { companyId: string; runId: string; target?: AdapterExecutionTarget | null }): Promise { + return serializeCopySave(db, input.companyId, input.runId, () => collectStoppedOnce(input)); + } + async function collectStoppedOnce(input: Parameters[0]) { let row = await get(input.companyId, input.runId); if (!row) return row; if (isAgentDirectoryCopy(row)) return directories.collectStopped(row, input.target); @@ -300,16 +335,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 +406,56 @@ 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"); + } + function recoverCommit(entry: CommitEntry, admitted: () => boolean): Promise { + return serializeCopySave(db, entry.companyId, entry.id, () => recoverCommitOnce(entry, admitted)); + } + async function recoverCommitOnce(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 +504,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 8685c7599fb..b8cc5197d5a 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"; @@ -98,6 +99,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 0a41dcf3690..432fc20006b 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", workspaceRecovery: "workspace-recovery", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 8535d7946dd..e11584ebdd7 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -4084,6 +4084,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 ce3f3de2172..ba1a51ce58a 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1947,7 +1947,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(); @@ -1957,7 +1957,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) : "*"); +}