From e0c6992426f107e98a0361efbb2e637c9d6b4c00 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:17:00 -0600 Subject: [PATCH 1/2] fix(heartbeat): prevent saved input starvation during queue reloads (#18819) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Accepted messages and task Retry requests use durable wake receipts. > - The recovery worker loads receipts in pages of 100 and checks batches of 25. > - Company activity can reload those pages before the pending batch drains. > - Reloads reset ready receipts' dispatch times and can repeatedly favor receipts already checked. > - This pull request preserves the priority of unchecked ready receipts so saved input can reach normal admission. ## Linked Issues or Issue Description **What happened?** Saved input can remain deferred after the preceding run finishes. Repeated company eligibility changes reset the dispatch times of pending receipts during paginated reconstruction. A PostgreSQL regression checks only 75 of 178 receipts on the original code despite twelve eligibility changes. **Expected behavior** Each ready receipt must reach its admission check. Existing wait deadlines, retry backoffs, authorization, and execution gates must still apply. **Steps to reproduce** 1. Seed more than one page of deferred receipts. 2. Keep those receipts pending after their checks. 3. Send company eligibility notifications between scheduler passes. 4. Check that every receipt is reconsidered, including receipts later in the backlog. **Paperclip version or commit** Reproduced on `367f29478d`. **Deployment mode** Built from source with PostgreSQL. The defect occurs before provider execution. Related: #10195 reports a similar starvation symptom in a different promotion loop. #18621 added saved-input Retry after a successful run. This change fixes fairness in the deferred-receipt worker. ## What Changed - Add an optional entry merge callback to scoped deadline reconstruction. - Use it in the deferred-receipt worker to retain unread ready receipts' priority while refreshing their state. - Place receipts whose wait has expired into the ready queue during reconstruction. - Keep real wait, claim, and dispatch deadlines authoritative. - Add a paginated starvation regression, real process dispatch coverage, and GitHub lock-collision recovery coverage. - Document the recovery fairness rule. ## Verification - The PostgreSQL starvation regression checks only 75 of 178 receipts before the fix. It checks all 178 afterward, both with ordinary receipts and with expired execution waits. - Seven focused scheduler cases pass, including moved interrupt grace, moved native error deadline, fresh claim/wait deadlines, and unchanged-gate grace. - GitHub endpoint lock-collision recovery, retained board authority, HTTP Retry, and five current admission-gate cases pass. - Real process acceptance passes on the final code: task Retry crosses the repeatedly invalidated backlog, reconstructs its persisted queued run after a worker restart, saves `receipt-proof` in run output, and has one successor for the Retry receipt. The test holds executor capacity while advancing queue time, then starts fresh workers at PostgreSQL time before releasing a real process lease. The defect is in admission scheduling before provider execution; this acceptance uses PostgreSQL and a local process adapter. - The full workspace build and typecheck (`node scripts/typecheck-low-memory.mjs`) pass locally. The affected worker suites pass 52 of 53 cases together; the existing native multi-company batch case times out in that combined run and passes in isolation without code changes. All new regression and acceptance cases pass. - Local `pnpm test:run` stops in its first chat shard on three 15-second timeouts (Teams long native HTTP picture delivery, direct final-response materialization, and contended inbound notices). All three pass in isolation without code changes. Later full-suite groups did not run locally after that shard failure. - [The full GitHub Actions matrix passes](https://github.com/paperclipai/paperclip/actions/runs/41697489374) on `ac91df2cc6db6c77529d923177b375b8c92efd45`, including all general-test, serialized-server, browser, build, and typecheck jobs. Server shard 8 initially failed an upstream process-adoption test and passed its one rerun without code changes. - [Greptile is 5/5](https://github.com/paperclipai/paperclip/pull/18819#issuecomment-6612091535) on the same current head, with no actionable findings or unresolved review threads. The separate Paperclip Review check is neutral; its previous request was bound to an older commit. Focused command: ```sh pnpm exec vitest run server/src/__tests__/deferred-execution-receipt-work.test.ts server/src/__tests__/scoped-deadline-work.test.ts ``` ## Risks - Ready receipt order changes during repeated reconstruction. Admission still reads current durable state. - A ready receipt must not inherit an obsolete priority after it acquires a future deadline. The merge applies only when both pending and loaded entries are immediate. Deadline safety tests cover that boundary. - The merge callback is optional. Other scoped workers retain their current reconstruction behavior. ## Model Used OpenAI `gpt-6.1-sol` in Codex, max reasoning effort, with tool use and local code execution. The runtime does not expose the context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass (focused regressions and acceptance; full-run timing limitations documented above) - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 2 + .../chat-channels.integration.test.ts | 39 +++++++++ .../deferred-execution-receipt-work.test.ts | 84 ++++++++++++++++++- .../heartbeat/deferred-receipt-work.ts | 22 +++-- server/src/services/scoped-deadline-work.ts | 5 +- 5 files changed, 140 insertions(+), 12 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 325148106d2..d93ae631b6c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -756,6 +756,8 @@ Normal issue-thread feedback can still wake the current assignee according to th Restart recovery revisits accepted deferred messages through the latest issue run, including historical mention queues for another agent. Ordinary messages for a former owner retain their assignment handoff. Native sources require their matching committed finalization, settled environment leases, and released finalizer and copyback ownership before recovery clears old issue claims. A successful retained warm release can satisfy the lease check. Released local bookkeeping without a provider allocation and a matching SSH bookkeeping release need no provider cleanup receipt. Plugin releases record their successful acknowledgement. Pending or failed cleanup and durable owners still hold work. A successful fenced cleanup completion clears its in-flight claim in the same database write, so a restart before retry bookkeeping cannot strand saved input. The normal release checks preserve cleanup holds, pauses, and operator Stops. Recovery checks for a competing turn under the issue lock before it promotes saved input. +Repeated eligibility notifications must not move an unchecked, ready deferred receipt behind receipts already reconsidered. Paginated reconstruction preserves pending ready receipts' dispatch priority while refreshing their current state. Receipts whose wait deadline has passed join the ready queue on reconstruction. Future wait grace periods, claimed-work deadlines, infrastructure retry backoffs, and current admission checks remain effective. + A paused recipient does not fail an accepted deferred request. Release and periodic recovery preserve its pending receipt and may process an eligible request behind it. Resuming the agent allows normal admission to reconsider the saved input. Missing, foreign, terminated, and unapproved recipients retain their rejection rules. Explicit task holds and operator Stops still prevent unauthorized execution. This does not replay historical failed requests. Accepted agent feedback must survive a child changing to `done` before its active run exits. Deferred wake promotion may reopen that completed child only for its current assignee, with explicit agent resume intent and live tracked comments from another author. Claim promotion before reopening. Cancelled tasks, deleted comments, self-authored comments, empty continuations, and agent continuations without explicit intent keep their existing suppression rules. Normal pause, ownership, authorization, and budget gates still apply. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 25d571411f2..89f7c9741d8 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -64441,6 +64441,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { await f.service.shutdown(); } }); + it("retries saved GitHub input after an endpoint admission lock collision", async () => { + const f = await successfulGitHubQueuedInputFixture(); + const heartbeat = heartbeatService(db); + const failures: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => failures.push(error) }); + let unlock!: () => void, locked!: () => void; + const acquired = new Promise(resolve => { locked = resolve; }); + const released = new Promise(resolve => { unlock = resolve; }); + const competingAdmission = db.transaction(async tx => { + await tx.select({ id: chatEndpoints.id }).from(chatEndpoints) + .where(eq(chatEndpoints.id, f.endpoint.id)).for("no key update"); + locked(); + await released; + }); + try { + await acquired; + await heartbeat.registerDeferredReceiptWork(coordinator, () => true).ready; + const priorFailures = failures.length; + const retry = await requestTaskRetry(f.input); + await expect.poll(() => failures.length, { timeout: 5_000 }).toBeGreaterThan(priorFailures); + expect((await readAgentWakeupRequestsById(f.pending!.id))[0].runId).toBeNull(); + unlock(); await competingAdmission; + await expect.poll(async () => (await readAgentWakeupRequestsById(f.pending!.id))[0].runId, + { timeout: 40_000 }).toBeTruthy(); + await expect.poll(async () => (await readAgentWakeupRequestsById(retry.requestId!))[0].status, + { timeout: 40_000 }).toBe("coalesced"); + const [saved] = await readAgentWakeupRequestsById(f.pending!.id); + expect((await readHeartbeatRunsByCompanyId(f.fixture.companyId)) + .filter(run => run.wakeupRequestId === f.pending!.id)).toHaveLength(1); + expect((await readAgentWakeupRequestsById(retry.requestId!))[0].runId).toBe(saved.runId); + expect((await readHeartbeatRunsById(f.runId))[0]).toMatchObject({ status: "succeeded" }); + } finally { + unlock(); await competingAdmission; + await coordinator.stop(); + await f.service.shutdown(); + } + }, 90_000); + it.each(["reconnected", "newer_input", "reconnected_newer_input", "snapshot_newer_input"] as const)( "admits current GitHub task Retry without replaying obsolete input: %s", async kind => { const f = await currentGitHubTaskRetryFixture(kind.includes("newer_input")); diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index e902c734429..7c028e3c2e8 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -72,14 +72,18 @@ describe("deferred receipt durable dispatch", () => { // during actual process execution; accelerated steps only settle idle work. vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); heartbeat = heartbeatService(db); - coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true, + coordinator = createCoordinator(); + }); + function createCoordinator() { + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true, onError: error => { errors.push(error); errorTimes.push(Date.now()); } }); - const register = coordinator.register.bind(coordinator); - coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const register = worker.register.bind(worker); + worker.register = (queue, task) => register(queue, { ...task, run: signal => { const pass = Promise.resolve(task.run(signal)); passes.add(pass); void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass; } }); - }); + return worker; + } afterEach(async () => { await coordinator.stop(); await heartbeat.drainActiveRunExecutions(); for (const stop of extraStops.splice(0)) await stop(); @@ -153,6 +157,78 @@ describe("deferred receipt durable dispatch", () => { source: "on_demand", status: "completed", reason: "task_retry", payload: { issueId: f.issue.id } }); await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; await quiet(); }); + it.each([false, true])("reconsiders every saved receipt during repeated company eligibility changes (expired wait: %s)", async expiredWait => { + const f = await fixture(); + const ids = Array.from({ length: 178 }, (_, index) => `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`); + await db.insert(agentWakeupRequests).values(ids.map(id => ({ id, companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution", + updatedAt: new Date(Date.now() - 60_000), + payload: { issueId: f.issue.id, _paperclipWakeContext: { wakeCommentIds: [randomUUID()] }, + ...(expiredWait ? { executionWait: { reason: "recovery" } } : {}) }, + }))); + const checked = new Set(); + await registerDeferredReceiptWork(db, coordinator, { + resume: async scope => { checked.add(scope.requestId!); }, + settleTaskChat: async () => {}, createNativeBatch: () => async () => {}, schedulingSuppressed: async () => false, + }, () => !drain).ready; + for (let pass = 0; pass < 12; pass++) { + await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId })); + await vi.advanceTimersByTimeAsync(31_000); + await Promise.allSettled([...passes]); + await immediate(); + } + expect(checked.size).toBe(ids.length); + await quiet(); + }); + it("dispatches task Retry past a repeatedly invalidated backlog into real process output", async () => { + const f = await fixture(); + // Hold executor capacity while queue time advances; reconstruct the saved + // run with fresh workers at PostgreSQL time before real process execution. + const [occupant] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "running", startedAt: new Date() }).returning(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, + wakeOnDemand: true, maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agent.id)); + const [closed] = await db.insert(issues).values({ companyId: f.companyId, title: "Closed task history", + status: "done", assigneeAgentId: f.agent.id }).returning(); + await db.insert(agentWakeupRequests).values(Array.from({ length: 178 }, (_, index) => ({ + id: `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`, + companyId: f.companyId, agentId: f.agent.id, source: "on_demand", status: "deferred_issue_execution", + payload: { issueId: closed.id, _paperclipWakeContext: { wakeCommentIds: [randomUUID()] } }, + }))); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + const retry = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId, + observedRunId: f.source.id, idempotencyKey: randomUUID() }); + const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retry.requestId!)))[0]; + for (let pass = 0; pass < 12 && !(await read()).runId; pass++) { + await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId })); + await vi.advanceTimersByTimeAsync(31_000); + await Promise.allSettled([...passes]); + await immediate(); + } + const receipt = await read(); + expect(receipt.runId).toBeTruthy(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)))[0].status).toBe("queued"); + await coordinator.stop(); + await heartbeat.drainActiveRunExecutions(); + vi.setSystemTime(vi.getRealSystemTime()); + heartbeat = heartbeatService(db); + coordinator = createCoordinator(); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, occupant.id)); + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await waitForReal(async () => { + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)); + expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded"); + expect(run.stdoutExcerpt).toContain("receipt-proof"); + }, { timeout: 15_000 }); + expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${retry.requestId}`)).toHaveLength(1); + await finish(f.issue.id); + await quiet(); + }); it("dispatches an ordinary committed task Retry into saved process output", async () => { const f = await fixture(); await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; diff --git a/server/src/services/heartbeat/deferred-receipt-work.ts b/server/src/services/heartbeat/deferred-receipt-work.ts index 9cca7528228..b799f10b0a7 100644 --- a/server/src/services/heartbeat/deferred-receipt-work.ts +++ b/server/src/services/heartbeat/deferred-receipt-work.ts @@ -6,7 +6,7 @@ import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; -type Entry = { id: string; companyId: string; dueAt: number }; +type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean }; type NativeIntentOptions = DeferredExecutionScope & { onDeferredDeadline(at: number): void; onError(error: unknown): void; @@ -39,6 +39,11 @@ export function registerDeferredReceiptWork( }, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint), matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.requestId || entry.id === scope.requestId), + // A busy company can invalidate a paginated scan before the pending + // batch drains. Keep unread ready receipts ahead of freshly reloaded + // receipts already checked; actual wait/claim deadlines still replace. + mergeEntry: (pending, loaded) => pending.immediate && loaded.immediate + ? { ...loaded, dueAt: Math.min(pending.dueAt, loaded.dueAt) } : loaded, load: async scope => { const key = scope ? JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }) : "*"; if (!enabled) { cursors.delete(key); return []; } @@ -76,12 +81,15 @@ export function registerDeferredReceiptWork( )).orderBy(asc(agentWakeupRequests.id)).limit(100); const complete = rows.length < 100; if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); - return { rows: rows.map(row => ({ id: row.id, companyId: row.companyId, - dueAt: kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000 + const now = Date.now(); + return { rows: rows.map(row => { + const deadline = kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000 : kind === "native" && row.error ? row.updatedAt.getTime() + 30_000 : kind === "deferred" && row.delayed && !row.interruptedLegacy ? row.updatedAt.getTime() + 30_000 - : Date.now(), - })), complete }; + : null; + const immediate = deadline === null || deadline <= now; + return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate }; + }), complete }; }, check: async entry => { if (!enabled) return null; @@ -101,7 +109,7 @@ export function registerDeferredReceiptWork( : kind === "deferred" && current.status === "deferred_issue_execution" && !current.interruptedLegacy && (current.payload?.queuedCommentInterrupt || current.payload?.executionWait) ? current.updatedAt.getTime() + 30_000 : null; - if (due !== null && due > Date.now()) return { ...entry, dueAt: due }; + if (due !== null && due > Date.now()) return { ...entry, dueAt: due, immediate: false }; } const errors: unknown[] = []; let next: number | null = null; @@ -117,7 +125,7 @@ export function registerDeferredReceiptWork( catch (error) { failed(error); } } if (errors.length) throw new AggregateError(errors, `Deferred receipt ${entry.id} could not settle`); - return next === null ? null : { ...entry, dueAt: next }; + return next === null ? null : { ...entry, dueAt: next, immediate: false }; }, }, canRun); } diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index cdd69a6801e..6ce95086ad0 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -18,6 +18,8 @@ export function registerScopedDeadlineWork( decodeScope(hint: string): Scope | undefined; matches(entry: Entry, scope?: Scope): boolean; load(scope?: Scope, signal?: AbortSignal): Promise; + // Owners may retain pending queue priority while refreshing eligibility. + mergeEntry?(pending: Entry, loaded: Entry): Entry; check(entry: Entry, signal?: AbortSignal): Promise; }, canRun: () => boolean, @@ -90,7 +92,8 @@ export function registerScopedDeadlineWork( const loaded = await options.load(scope, signal); const result = Array.isArray(loaded) ? { rows: loaded, complete: true } : loaded; for (const entry of result.rows ?? []) { - entries.set(entry.id, entry); + const pending = entries.get(entry.id); + entries.set(entry.id, pending && options.mergeEntry ? options.mergeEntry(pending, entry) : entry); for (const [other, progress] of scans) if (options.matches(entry, options.decodeScope(other))) progress.seen.add(entry.id); } if (!result.complete) { dirty.add(hint); continue; } From fb6554ac6208b176a4109aba103264146e56a7fe Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:17:16 -0600 Subject: [PATCH 2/2] fix(native): settle source wake with terminal projection (#18804) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Each agent wake request can bind to one heartbeat run. > - Native recovery can finish and project a run after its live executor has stopped. > - The legacy completion callback settles the wake, but native terminal projection did not. > - A succeeded run could therefore leave its exact wake claimed with no finish time. > - This pull request settles that claim in each native terminal projection transaction and reuses durable projection recovery for old committed rows. > - The benefit is consistent run and wake receipts without a provider replay or a new sweeper. ## Linked Issues or Issue Description Refs #13856 and #10109. These open PRs describe orphan claimed wakes and cleanup after terminal persistence. This change fixes the native projection owner and uses its existing durable worker. It does not modify those contributor branches. **What happened?** A native run can be accepted and committed, then reach a terminal state through recovery. Its source wake stays claimed because the live completion callback does not return to settle it. **Expected behavior** Terminal projection must settle only the same company's exact source wake bound to that agent and run. Completed, failed and cancelled outcomes must retain their meaning. Other live, queued, cancelled, unbound or newly rebound wake records must remain unchanged. **Steps to reproduce** Commit an accepted native result in the real database fixture. Leave its source wake claimed. Call native finalization with run-status projection enabled, without the live completion callback. Before this change, the run becomes succeeded and finished while the exact wake remains claimed. ## What Changed - Settle the exact source claim inside native finalization, committed recovery, and exhausted-finalization projection transactions. Fence it by wake ID, company, agent, run and current claimed state. - Lock the matching claim and register settlement notifications before the write. A failed transaction rolls back both run projection and wake settlement. - Let the existing durable projection worker discover a terminal run with an unfinished exact claim. A repaired committed row leaves the pending set. No new timer, sweep or table is added. - Add real database regressions for native recovery, all three terminal outcomes, old committed receipts, idempotence, scoped denial controls, notification ordering and rollback. Cover provider/workspace finalization exhaustion and denial when the caller does not own terminal projection. - Preserve the existing provider, accounting, workspace, permissions and result contracts. No API, schema, user command, access-map entry or documentation contract changes are needed. ## Verification - The new native coordinator/finalizer regression failed before the production change: the run was succeeded and committed, but the bound wake was still claimed. - The initial focused database run did not reach the assertion because the SDK was unbuilt. A second attempt hit sandbox loopback restrictions. A third exposed skipped package library hydration. These setup failures were corrected before the meaningful red was captured. - Real PostgreSQL projection and response suites passed 89 tests before the independent review correction. The reviewer requested notification intent registration before mutation. The correction adds a notification-order and rollback case. All 90 tests now pass across both suites. - Server typecheck passed after building the isolated checkout's Runner and SDK prerequisites. Fresh corrected-source server typecheck and production server build also pass. Full repository checks run in current-head CI. The earlier serialized-9 CI failure was the independently repaired inbox snapshot fixture in #18798 (expected nine reader calls, observed eight). A normal merge of the containing master supplies that fix; the native suites plus the repaired inbox suite now pass 96 tests together at merge head `456c92ef8d60c3fcbeaf66ff8010a7ebd3d1109d`. Server typecheck passes. That head was superseded by the reviewed retry-exhaustion correction below. The subsequent review found a third terminal write in retry exhaustion. Real database provider and workspace exhaustion regressions failed with claimed/null-finished wakes before the fix. The same fenced helper now runs inside that transaction. At final source head `9718d4c2ec6066accf549dc6fea795891f6b90ec`, all 100 focused tests, server typecheck and production server build pass, including projection-disabled denial and rollback of run/coordinator/wake on notification failure. - Final current-head CI passed: 54 success, 2 skipped, and the neutral Paperclip Review at 5/5. Greptile also reports 5/5. Required ci / verify and ci / e2e passed. All review threads are resolved. - No live task write, provider work, accounting repair, deployed replay or manual rollout was performed. ## Risks - A stale wake binding must never be settled. The update requires the exact wake ID, company, agent, run and claimed state. Tests cover unbound and newer-run bindings, foreign company and agent, queued and cancelled records, and retained live ownership. - Old claimed debt now participates in the existing projection predicate. It uses the source wake primary key and stops matching after settlement. Existing quiet-worker tests remain in the focused suite. - Wake settlement must share the projection transaction. Notification failure and outer rollback tests preserve both original rows. - Interruption, timeout, retained historical terminal-failure coordinators, and delayed response presentation keep their current owners. This change does not infer or replay provider work. ## Model Used OpenAI Codex, `gpt-6.1-sol`, Extra High reasoning, with source inspection, code execution and regression testing. Context-window capacity is not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used with version and capability details - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have linked existing public issues or described the issue in this PR - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal ticket ID - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation where needed; existing contracts are restored - [x] I have considered and documented risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before merge --------- Co-authored-by: Paperclip --- .../native-run-projection-work.test.ts | 155 +++++++++++++++++- .../agent-chat-response-recovery.test.ts | 20 +++ .../native-runtime/native-run-finalizer.ts | 4 +- .../native-runtime/native-run-projection.ts | 36 +++- 4 files changed, 210 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/native-run-projection-work.test.ts b/server/src/__tests__/native-run-projection-work.test.ts index 26f0a9d239a..f9718354a1c 100644 --- a/server/src/__tests__/native-run-projection-work.test.ts +++ b/server/src/__tests__/native-run-projection-work.test.ts @@ -1,3 +1,4 @@ +import * as wakeSettlementNotifications from "../services/external-conversation-settlement-notifications.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js"; @@ -8,14 +9,14 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { nativeRunProjectionWork } from "../services/native-runtime/native-run-projection-work.js"; import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; -import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; +import { finalizeNativeRun, recordNativeFinalizationFailure } from "../services/native-runtime/native-run-finalizer.js"; import * as statusCommitter from "../services/native-runtime/status-decision-committer.js"; import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; @@ -120,6 +121,154 @@ describe("committed native run projection lifecycle", () => { async function accepted(status: "succeeded" | "failed" | "cancelled" = "succeeded") { const binding = await seed(); await complete(binding, "control_plane", db, status); await barrier(binding); return binding; } + async function claimedWake(binding: Awaited>) { + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "issue", + status: "claimed", runId: binding.runId, claimedAt: new Date(), payload: { issueId: binding.issueId } }).returning(); + await db.update(heartbeatRuns).set({ wakeupRequestId: wake.id }).where(eq(heartbeatRuns.id, binding.runId)); + return wake; + } + const readWake = async (id: string) => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, id)))[0]; + + it.each(["succeeded", "failed", "cancelled"] as const)("settles the exact wake atomically with a fresh %s projection", async status => { + const binding = await accepted(status), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const run = await read(binding.runId), settled = await readWake(wake.id); + expect(run.status).toBe(status); + expect(settled).toMatchObject({ status: status === "succeeded" ? "completed" : status, runId: binding.runId, + companyId, agentId, finishedAt: run.finishedAt, error: run.error }); + expect(settled.finishedAt).not.toBeNull(); + await projectCommittedRun({ db, run, coordinator: await state(binding.runId) }); + expect(await readWake(wake.id)).toEqual(settled); + }); + + it.each(["provider", "workspace"] as const)("settles the exact wake when %s finalization retries exhaust", async scope => { + const binding = await accepted(), wake = await claimedWake(binding); + if (scope === "workspace") await db.update(workspaceOperations).set({ status: "failed" }) + .where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + if (scope === "provider") await db.update(nativeRunResults).set({ + resultJson: sql`jsonb_set(${nativeRunResults.resultJson}, '{terminal,runTerminalState}', '"unknown"'::jsonb)`, + }).where(eq(nativeRunResults.runId, binding.runId)); + for (let attempt = 0; attempt < 3; attempt++) { + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, binding.runId)); + if (scope === "provider") await reconcileNativeFinalizations(db, [binding.runId]); + else await recordNativeFinalizationFailure({ db, runId: binding.runId, projectRunStatus: true, + failureScope: "workspace", error: new Error("native_workspace_finalize_failed") }); + if (attempt < 2) { + expect((await read(binding.runId)).status).toBe("running"); + expect(await readWake(wake.id)).toMatchObject({ status: "claimed", finishedAt: null }); + } + } + const run = await read(binding.runId); + expect((await state(binding.runId)).phase).toBe("terminal_failure"); + expect(run).toMatchObject({ status: scope === "provider" ? "succeeded" : "failed", nativePhase: "terminal_failure" }); + expect(run.finishedAt).not.toBeNull(); + expect(await readWake(wake.id)).toMatchObject({ status: scope === "provider" ? "completed" : "failed", + runId: binding.runId, finishedAt: run.finishedAt }); + }); + + it("keeps the source wake claimed when exhausted failure does not own terminal projection", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + for (let attempt = 0; attempt < 3; attempt++) await recordNativeFinalizationFailure({ db, runId: binding.runId, + projectRunStatus: false, failureScope: "workspace", error: new Error("native_workspace_finalize_failed") }); + expect((await state(binding.runId)).phase).toBe("terminal_failure"); + expect((await read(binding.runId)).status).toBe("running"); + expect(await readWake(wake.id)).toEqual(wake); + }); + + it("rolls back exhausted projection and its claim if wake notification registration fails", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + const fail = () => recordNativeFinalizationFailure({ db, runId: binding.runId, + projectRunStatus: true, failureScope: "workspace" as const, error: new Error("native_workspace_finalize_failed") }); + await fail(); await fail(); + const beforeRun = await read(binding.runId), beforeCoordinator = await state(binding.runId); + const notify = wakeSettlementNotifications.notifyExternalConversationSettlement; + vi.spyOn(wakeSettlementNotifications, "notifyExternalConversationSettlement").mockImplementation(async (tx, scope) => { + const [current] = await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(current.status).toBe("claimed"); + await notify(tx, scope); + throw new Error("exhaustion wake notification failure"); + }); + await expect(fail()).rejects.toThrow("exhaustion wake notification failure"); + expect(await read(binding.runId)).toEqual(beforeRun); + expect(await state(binding.runId)).toEqual(beforeCoordinator); + expect(await readWake(wake.id)).toEqual(wake); + }); + + it.each(["succeeded", "failed", "cancelled"] as const)("discovers and repairs an old %s run's claimed wake once", async status => { + const binding = await accepted(status); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const run = await read(binding.runId), wake = await claimedWake(binding); + const work = nativeRunProjectionWork(db); + const [entry] = await work.load({ companyId, runId: binding.runId }); + expect(entry).toMatchObject({ id: binding.runId }); + await work.check(entry); + expect(await readWake(wake.id)).toMatchObject({ status: status === "succeeded" ? "completed" : status, + finishedAt: run.finishedAt }); + expect((await read(binding.runId)).finishedAt).toEqual(run.finishedAt); + expect(await work.load({ companyId, runId: binding.runId })).toEqual([]); + }); + + it.each(["unbound", "newer-run", "queued", "cancelled", "foreign-company", "foreign-agent", "live-owner"] as const)( + "does not settle a %s wake claim", async fault => { + const binding = await accepted(), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + if (fault === "unbound" || fault === "newer-run") await db.update(agentWakeupRequests) + .set({ runId: fault === "unbound" ? null : randomUUID() }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "queued" || fault === "cancelled") await db.update(agentWakeupRequests) + .set({ status: fault }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "foreign-company") { + const otherCompany = randomUUID(); + await db.insert(companies).values({ id: otherCompany, name: "Other wake owner", issuePrefix: `OW${randomUUID().slice(0, 6)}` }); + await db.update(agentWakeupRequests).set({ companyId: otherCompany }).where(eq(agentWakeupRequests.id, wake.id)); + } + if (fault === "foreign-agent") { + const otherAgent = randomUUID(); + await db.insert(agents).values({ id: otherAgent, companyId, name: "Other wake owner", adapterType: "paperclip_runner" }); + await db.update(agentWakeupRequests).set({ agentId: otherAgent }).where(eq(agentWakeupRequests.id, wake.id)); + } + if (fault === "live-owner") await db.update(heartbeatRuns).set({ status: "running", nativePhase: "terminal_failure", + errorCode: "native_execution_ownership_unverified" }).where(eq(heartbeatRuns.id, binding.runId)); + const before = await readWake(wake.id); + await projectCommittedRun({ db, run: await read(binding.runId), coordinator: await state(binding.runId) }); + expect(await readWake(wake.id)).toEqual(before); + if (fault === "live-owner") expect((await read(binding.runId)).status).toBe("running"); + }, + ); + + it("registers wake settlement intent before mutation and rolls back notification failure", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const before = await read(binding.runId); + const notify = wakeSettlementNotifications.notifyExternalConversationSettlement; + const notification = vi.spyOn(wakeSettlementNotifications, "notifyExternalConversationSettlement") + .mockImplementation(async (tx, scope) => { + const [current] = await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(current.status).toBe("claimed"); + await notify(tx, scope); + throw new Error("wake intent registration failure"); + }); + await expect(projectCommittedRun({ db, run: before, coordinator: await state(binding.runId) })) + .rejects.toThrow("wake intent registration failure"); + expect(notification).toHaveBeenCalledOnce(); + expect(await read(binding.runId)).toEqual(before); + expect(await readWake(wake.id)).toEqual(wake); + }); + + it("rolls back terminal projection and exact wake settlement together", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const before = await read(binding.runId); + await expect(db.transaction(async tx => { + await projectCommittedRun({ db: tx as unknown as Db, run: before, coordinator: await state(binding.runId) }); + throw new Error("projection rollback fixture"); + })).rejects.toThrow("projection rollback fixture"); + expect(await read(binding.runId)).toEqual(before); + expect(await readWake(wake.id)).toEqual(wake); + }); + it("reconstructs a committed decision with an unfinished run projection and stays quiet after repair", async () => { const binding = await accepted(); await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); diff --git a/server/src/services/native-runtime/agent-chat-response-recovery.test.ts b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts index 918653a5fd1..b0d4ce5481a 100644 --- a/server/src/services/native-runtime/agent-chat-response-recovery.test.ts +++ b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts @@ -65,6 +65,26 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => const input = { companyId, issueId, runId }; return { ...input, input, agentId, userId, wakeId, prpEvent, reply }; } + it("settles the exact claimed wake when recovery projects a committed native run", async () => { + // The provider and result are already committed. Recovery owns terminal + // projection because the live executor will not return to completeRun. + const f = await fixture(false, { succeed: false }); + const sourceWakeId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: sourceWakeId, companyId: f.companyId, + agentId: f.agentId, source: "issue", reason: "issue_assigned", status: "claimed", + runId: f.runId, claimedAt: new Date(), payload: { issueId: f.issueId } }); + await db.update(heartbeatRuns).set({ wakeupRequestId: sourceWakeId }).where(eq(heartbeatRuns.id, f.runId)); + + await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, sourceWakeId)); + expect(run).toMatchObject({ status: "succeeded", nativePhase: "committed" }); + expect(run.finishedAt).not.toBeNull(); + expect(wake).toMatchObject({ status: "completed", runId: f.runId }); + expect(wake.finishedAt).not.toBeNull(); + }); + const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId)); const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]; diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 1cbba5703cd..3a68b7d8f3b 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -3,7 +3,7 @@ import { notifyExternalAgentSettlement } from "../../modules/external-agents/ind import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; -import { projectCommittedRun, recoveredExecutionFailureMetadata } from "./native-run-projection.js"; +import { projectCommittedRun, recoveredExecutionFailureMetadata, settleNativeTerminalWake } from "./native-run-projection.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; @@ -465,6 +465,7 @@ async function recordRetryableFailure(input: { }) .where(eq(heartbeatRuns.id, input.run.id)) .returning(); + if (projectsTerminalStatus && updatedRun) await settleNativeTerminalWake(tx as unknown as Db, updatedRun); if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null; if (supersededByNewerRun) { await issueRecoveryActionService( @@ -1679,6 +1680,7 @@ export async function finalizeNativeRun(input: { await notifyExecutionRunWork(tx, { companyId: run.companyId }); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId }); const [after] = await projectRun(tx); + if (after && input.projectRunStatus) await settleNativeTerminalWake(tx, after); return after ? { before, after } : null; }); const updatedRun = projection?.after; diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 2542b8bb1e3..706f80e7366 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -1,3 +1,5 @@ +import { notifyExternalConversationSettlement } from "../external-conversation-settlement-notifications.js"; +import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; @@ -5,7 +7,7 @@ import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-wor import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, exists, inArray, isNotNull, isNull, not, or, sql, type SQL } from "drizzle-orm"; -import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -26,11 +28,41 @@ export function projectNativeTerminalRunStatus( +function claimedNativeWakeNeedsSettlement() { + return sql`exists (select 1 from ${agentWakeupRequests} where + ${agentWakeupRequests.id} = ${heartbeatRuns.wakeupRequestId} + and ${agentWakeupRequests.companyId} = ${heartbeatRuns.companyId} + and ${agentWakeupRequests.agentId} = ${heartbeatRuns.agentId} + and ${agentWakeupRequests.runId} = ${heartbeatRuns.id} + and ${agentWakeupRequests.status} = 'claimed')`; +} + +/** Settle only the exact persisted source claim inside terminal projection. */ +export async function settleNativeTerminalWake(tx: Db, run: Pick) { + if (run.runtimeMode !== "native" || !run.wakeupRequestId || !run.finishedAt + || !["succeeded", "failed", "cancelled"].includes(run.status)) return; + const claim = and(eq(agentWakeupRequests.id, run.wakeupRequestId), + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "claimed")); + const [current] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests) + .where(claim).for("update").limit(1); + if (!current) return; + await notifyExternalConversationSettlement(tx, { companyId: run.companyId }); + await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId }); + await tx.update(agentWakeupRequests).set({ + status: run.status === "succeeded" ? "completed" : run.status, + finishedAt: run.finishedAt, error: run.error, updatedAt: new Date(), + }).where(claim); +} + export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "failed" | "cancelled" | SQL) { return and( eq(heartbeatRuns.runtimeMode, "native"), or( inArray(heartbeatRuns.status, ["queued", "running", "failed"]), + and(eq(heartbeatRuns.status, projectedStatus), + inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled"]), claimedNativeWakeNeedsSettlement()), and( eq(heartbeatRuns.status, "succeeded"), or( @@ -51,6 +83,7 @@ export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "f or( sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`, + claimedNativeWakeNeedsSettlement(), isNull(heartbeatRuns.finishedAt), sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`, sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`, @@ -164,6 +197,7 @@ export async function projectCommittedRun(input: { }) .where(needsRepair) .returning(); + if (after) await settleNativeTerminalWake(tx, after); return after ? { before, after } : null; });