From e620ce36784df20575cc12454b22c81606bc572a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 00:24:09 -0600 Subject: [PATCH] fix(heartbeat): deliver each accepted follow-up exactly once through restarts and admission holds (#19741) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A user can comment, click Retry, or rely on automatic recovery while an agent works on a task > - Restart-chaos waves showed accepted follow-ups being dropped, or delivered twice, around restarts and temporary company admission holds (accounting still settling after a restart) > - A dropped comment means the user's request is never acted on. A duplicate reopens work that is already done > - This pull request makes accepted work wait through a temporary hold and resume when it clears. It also stops a comment from being promoted again after a completed run already received it > - The benefit is that each accepted follow-up is delivered exactly once, eventually, unless the task is cancelled or a reached hard stop applies ## Linked Issues or Issue Description Refs #19702 (never-started tasks during a temporary hold), #19700 (bounded accounting hold), #14910 (stale reopen from a deferred wake; no file overlap). **What happened?** Restart-chaos waves found these shapes: 1. **Comment wake cancelled.** A comment was saved behind a run and promoted when the run ended. A temporary company hold refused the unstarted queued run, and claim cancelled it terminally. The recovery wake that followed was cancelled the same way, and the task rested `blocked` without the comment. 2. **Dead Retry.** A Retry click during the hold stayed `deferred_issue_execution` after the hold cleared. Nothing woke it, and the issue held no execution lock. 3. **Refused assignment.** A new task's first wake was refused during the hold. Nothing woke stranded-work recovery when the hold cleared, so the task stayed `todo` with no run. 4. **Silent comment after recovery.** A comment was saved behind a run that restart recovery finalized. A company-scoped hint waited out the 30 s deferred-receipt rescan, so the task looked done with the message unread. 5. **Comment delivered twice.** A run whose provider process was lost had a saved comment. Its retry started after the comment, received it in its prompt (task history), did the work, and finished the task. The saved wake was then promoted anyway, which reopened the done task. **Expected behavior** An accepted follow-up (comment, Retry, recovery) is delivered exactly once, eventually, unless the task is cancelled. A temporary refusal never ends in a terminal cancel, skip, or block. Reaching a hard stop remains the board's decision. ## What Changed - `heartbeat/queue.ts`: on a budget block at claim, an unstarted queued run stays queued when the block is temporary. A reached hard stop or blocked unpriced usage still cancels it. - `recovery/service.ts`: stranded-work recovery waits out a temporary hold for any task. It does not escalate to `blocked`. This extends #19702 beyond never-started tasks. - `budget-block-ownership.ts`: a shared `isBoardOwnedBudgetBlock`, moved out of the recovery service. `budgets.ts` is unchanged. - `run-cost-accounting.ts`: settling a run's accounting wakes queued-run dispatch, the deferred-receipt worker, and stranded-work recovery. Held work waits passively; this is what resumes it. - `deferred-execution-notifications.ts`: new `notifyDeferredWakesForIssue`. `native-run-projection.ts` (a run reaching a terminal status) and `environments.ts` (lease release) now hint each saved wake of the issue by id, so delivery does not wait out the company rescan interval. - `wake-queue/adapters/postgres.ts`: release treats a saved comment as delivered only when a succeeded run of the wake's agent on that issue started after the comment's latest revision, and either received the comment as its own wake input (`wakeCommentIds`) or was a retry of the interrupted turn whose task history included it. In that case it cancels the saved wake with "Queued messages were already delivered to a completed run" and leaves the task as it is. Task history alone never consumes a message saved for its own turn. One query checks the whole batch. - `run-cost-accounting.ts`: settling late receipt debt also sends the hold-release notifications. - `doc/execution-semantics.md`: documents the rule. ## Verification Red on `master` and green with this change, on embedded Postgres: - `server/src/__tests__/heartbeat-held-comment-wake.test.ts`: - A promoted comment refused by the hold stays pending, with no cancelled run and no `blocked` task. It is delivered once after the hold clears through real `accountRunCost`. - A Retry clicked during the hold starts exactly once after accounting settles, which wakes its owner. - A new task whose assignment was refused during the hold starts once accounting settles, which wakes stranded-work recovery. - A comment that a later succeeded retry already received as history, or that a later run received as wake input, is consumed without reopening the task. - Guard tests (green on both): - A comment is still delivered in each of these cases: the later run never received it, it was edited after that run started, it was posted while the run was already running, or a non-retry run only saw it as history. - A task whose continuation was refused waits out the hold. - A reached hard stop is still refused. - `server/src/__tests__/native-run-projection-work.test.ts`: projecting a recovered run terminal hints the saved comment's own wake (red on `master`). - Related suites pass locally (14 files, 482 tests): wake-queue, queued comments routes, refused assignment wake, native run projection. - Live replays (local, unmasked, kept instances): - `--replay w-20261107T031005-s7501/t04` (Claude C1, K1): PASS. The comment's run was created at 03:56:24 during the hold, stayed queued, started at 04:06:24 when the hold cleared, and succeeded. It is the only run carrying the comment. - `--replay w-20261107T002911-s7401/t01` (before the per-wake hint): it reproduced shape 4. Run 592d772f was killed 250 ms after `run.result.accepted`, released and finalized at restart, and the task rested `done` with the comment still saved. Calling the existing sweep on the kept database promoted it, which confirmed that the missing piece was the trigger. ## Risks - A queued run can now sit queued for the length of a temporary hold, which #19700 bounds. It does not hold the issue's execution lock, so later comments still coalesce into it. - The delivered-comment rule depends on the run's recorded prompt context. A succeeded run that received the comment in its history counts as delivery, even if the agent did not act on it. This matches exactly-once delivery semantics. - Settling accounting now emits three extra notifications per run. They are coalesced per company. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, run as an agent in Claude Code with tool use (shell, file editing, local test execution, live restart-chaos replays). ## 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 - [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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/execution-semantics.md | 2 + .../heartbeat-held-comment-wake.test.ts | 288 ++++++++++++++++++ .../native-run-projection-work.test.ts | 24 +- .../modules/wake-queue/adapters/postgres.ts | 30 +- .../modules/wake-queue/application/ports.ts | 7 +- .../wake-queue/application/use-cases.ts | 7 +- server/src/services/budget-block-ownership.ts | 19 ++ .../deferred-execution-notifications.ts | 16 +- server/src/services/environments.ts | 10 +- server/src/services/heartbeat/queue.ts | 6 + .../native-runtime/native-run-projection.ts | 6 +- server/src/services/recovery/service.ts | 29 +- server/src/services/run-cost-accounting.ts | 20 +- 13 files changed, 429 insertions(+), 35 deletions(-) create mode 100644 server/src/__tests__/heartbeat-held-comment-wake.test.ts create mode 100644 server/src/services/budget-block-ownership.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 53662e782da..0a00ade2955 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -493,6 +493,8 @@ Repeated automatic signals for an unchanged gate share one durable skipped-wake A new task whose first assignment wake is refused by a temporary budget hold (completed runs awaiting accounting, a native run recovering its accounting, or a hold awaiting reconciliation) stays `todo`. Stranded-work recovery does not move a task with no run to `blocked` for such a hold. The task shows the hold it waits for, and the assigned-task liveness sweep admits it once the hold clears, through normal admission and with one run. A reached hard stop or blocked unpriced usage remains a board decision. Removed assignees and cancelled tasks still do not wake. +The same holds for accepted work beyond a first assignment. A temporary budget hold never terminally cancels or escalates accepted work: an unstarted queued run (a promoted comment, a Retry, or a recovery) stays queued, and stranded-work recovery leaves the task as it is instead of moving it to `blocked`. These waits are passive, so settling a run's accounting wakes queued-run dispatch, the saved-receipt worker, and stranded-work recovery for the company. The held work then starts through normal admission, once. When a native run reaches a terminal status through recovery, or releases its environment, each message saved behind it is hinted by its own receipt, so delivery does not wait out a company-wide rescan interval. + New comments received during an execution hold retain their individual deferred receipts and ordered comment ids. Release cannot drain those receipts while replay remains blocked; the next eligible wake can adopt them. Authorized external-chat requests also remain deferred with their exact durable receipt. They must use normal promotion and current authorization; a generic wake cannot adopt only their comment ids and discard their actor or session contract. A wait does not authorize replay, reset an incident retry budget, or bypass an interaction's delivery rules. The conversation groups repeated empty pre-start reconciliation cancellations into a neutral waiting notice. Started runs, actual startup failures, and run history remain inspectable. No historical run records are deleted. diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts new file mode 100644 index 00000000000..31b747e9c24 --- /dev/null +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -0,0 +1,288 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, agentWakeupRequests, authUsers, budgetPolicies, companies, companyMemberships, costEvents, createDb, heartbeatRuns, + issueComments, issues } from "@paperclipai/db"; +import { accountRunCost } from "../services/run-cost-accounting.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; +import { requestTaskRetry } from "../services/task-retry.js"; +import { queueIssueAssignmentWakeup } from "../services/issue-assignment-wakeup.js"; +import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { recoveryPostgresSupport, startRecoveryPostgres } from "./helpers/recovery-postgres.js"; + +const describePostgres = recoveryPostgresSupport.supported ? describe : describe.skip; +if (!recoveryPostgresSupport.supported) console.warn(`Skipping held comment wake tests: ${recoveryPostgresSupport.reason}`); + +/** + * RC-37: a follow-up comment saved behind a run is promoted when that run ends. + * A temporary company admission hold (accounting that is still settling after a + * restart) can refuse the promoted run before it starts. The comment must stay + * saved and be delivered exactly once after the hold clears. + */ +describePostgres("a promoted comment refused by a temporary admission hold", () => { + let temporary: Awaited>; + let db: ReturnType; + let heartbeat: ReturnType; + const adapterType = "codex_local"; + const dispatched: Array<{ runId: string; context: Record }> = []; + const oldHome = process.env.PAPERCLIP_HOME; + let home: string; + + beforeAll(async () => { + temporary = await startRecoveryPostgres("heartbeat-held-comment-"); + db = createDb(temporary.connectionString); + home = await mkdtemp(path.join(tmpdir(), "paperclip-held-comment-")); + process.env.PAPERCLIP_HOME = home; + heartbeat = heartbeatService(db); + registerServerAdapter({ + type: adapterType, supportsLocalAgentJwt: false, + execute: async input => { + dispatched.push({ runId: input.runId, context: input.context as Record }); + // The agent finishes the task, so no disposition repair follows. + await db.update(issues).set({ status: "done" }).where(eq(issues.id, String(input.context.issueId))); + return { exitCode: 0, signal: null, timedOut: false, resultJson: {} }; + }, + testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }), + }); + }, 240_000); + + afterAll(async () => { + await heartbeat?.drainActiveRunExecutions(); + unregisterServerAdapter(adapterType); + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + if (oldHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = oldHome; + if (home) await rm(home, { recursive: true, force: true }); + }); + + /** A task whose run just ended with a user comment saved behind it, in a + * company that holds new work while another run's accounting settles. */ + async function seedHeldComment(options: { hardStopReached?: boolean } = {}) { + const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(), issueId = randomUUID(); + const userId = `comment-user-${companyId}`; + await db.insert(authUsers).values({ id: userId, name: "Commenter", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companies).values({ id: companyId, name: "Held comment fixture", issuePrefix: `C${companyId.slice(0, 6)}` }); + await db.insert(companyMemberships).values({ companyId, principalId: userId, principalType: "user", status: "active", membershipRole: "member" }); + await db.insert(agents).values([ + { id: agentId, companyId, name: "Assignee", adapterType, status: "idle", adapterConfig: {} }, + { id: otherAgentId, companyId, name: "Earlier worker", adapterType, status: "idle", adapterConfig: {} }, + ]); + await db.insert(budgetPolicies).values({ companyId, scopeType: "company", scopeId: companyId, windowKind: "calendar_month_utc", + amount: options.hardStopReached ? 1 : 2_000, hardStopEnabled: true, unpricedUsagePolicy: "allow" }); + const [pending] = await db.insert(heartbeatRuns).values({ companyId, agentId: otherAgentId, invocationSource: "automation", + triggerDetail: "system", status: "succeeded", startedAt: new Date(Date.now() - 60_000), finishedAt: new Date(), + costAccountingPending: true, contextSnapshot: {}, + usageJson: { costUsd: 0.01, usageSource: "per_run", billingType: "metered_api", provider: "openai", model: "gpt" } }).returning(); + await db.insert(issues).values({ id: issueId, companyId, title: "Task with a follow-up comment", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: userId, issueNumber: 1 }); + const [source] = await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "assignment", + triggerDetail: "system", status: "succeeded", startedAt: new Date(Date.now() - 30_000), finishedAt: new Date(), + contextSnapshot: { issueId, wakeReason: "issue_assigned" } }).returning(); + const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: userId, + body: "Also include the FOLLOWUP line." }).returning(); + // The durable receipt the comment route writes while the source run is active. + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: userId, + payload: { issueId, commentId: comment.id, mutation: "comment", _paperclipWakeContext: { + issueId, taskId: issueId, commentId: comment.id, wakeCommentId: comment.id, wakeCommentIds: [comment.id], + wakeReason: "issue_commented", source: "issue.comment" } }, + }).returning(); + return { companyId, agentId, issueId, userId, pendingRunId: pending.id, sourceRunId: source.id, commentId: comment.id, wakeId: wake.id }; + } + + const issueRuns = (f: { companyId: string; issueId: string }) => db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, f.companyId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${f.issueId}`)); + const readWake = async (id: string) => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, id)))[0]; + const deliveries = (commentId: string) => dispatched.filter(turn => + Array.isArray(turn.context.wakeCommentIds) && (turn.context.wakeCommentIds as string[]).includes(commentId)); + + /** Settle the other run's accounting through the production path, and + * return which passive owners it woke. Held work waits on these wakes. */ + async function settleAccounting(f: { pendingRunId: string }) { + const woken = new Set(), owner = db; + const stops = [DELIVERY_QUEUES.queuedRunDispatch, DELIVERY_QUEUES.deferredExecutionReceipts, DELIVERY_QUEUES.strandedAssignments] + .map(queue => subscribeDeliveryWork(owner, queue, () => woken.add(queue))); + try { + expect(await accountRunCost(db, f.pendingRunId)).toBe(true); + await new Promise(resolve => setTimeout(resolve, 200)); + } finally { for (const stop of stops) stop(); } + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.pendingRunId)))[0].costAccountingPending).toBe(false); + return woken; + } + + async function settle(f: { companyId: string }) { + await heartbeat.resumeQueuedRuns(); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.drainActiveRunExecutions(); + } + + it("keeps the comment saved through the hold and delivers it exactly once after the hold clears", async () => { + const f = await seedHeldComment(); + // The source run has ended: the saved comment is promoted while the hold lasts. + await settle(f); await settle(f); + expect(deliveries(f.commentId)).toHaveLength(0); + // The comment stays pending: either saved or on an unstarted queued run. + const held = await readWake(f.wakeId); + expect(["deferred_issue_execution", "queued"]).toContain(held.status); + const pending = held.runId ? (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, held.runId)))[0] : null; + if (pending) expect(pending).toMatchObject({ status: "queued", startedAt: null }); + expect((await issueRuns(f)).filter(run => run.status === "cancelled")).toEqual([]); + const [waiting] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(waiting.status).not.toBe("blocked"); + + // Accounting settles and the hold clears. Settlement wakes the held work. + expect(await settleAccounting(f)).toEqual(new Set([DELIVERY_QUEUES.queuedRunDispatch, + DELIVERY_QUEUES.deferredExecutionReceipts, DELIVERY_QUEUES.strandedAssignments])); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) + .where(eq(agentWakeupRequests.id, f.wakeId)); + await settle(f); await settle(f); + expect(deliveries(f.commentId)).toHaveLength(1); + + // Later sweeps never deliver it again. + await settle(f); + expect(deliveries(f.commentId)).toHaveLength(1); + }, 180_000); + + it("keeps a Retry clicked during the hold pending and starts it exactly once after the hold clears", async () => { + const f = await seedHeldComment(); + // The run the user retries failed; its saved comment was already discarded. + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "adapter_failed" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + const retry = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: f.userId, + idempotencyKey: randomUUID(), observedRunId: f.sourceRunId }); + expect(retry).toMatchObject({ accepted: true }); + await settle(f); await settle(f); + const retried = () => dispatched.filter(turn => turn.context.wakeReason === "task_retry" + || (turn.context.taskRetry as { generationId?: string } | undefined)?.generationId === retry.requestId); + expect(retried()).toHaveLength(0); + const held = await readWake(retry.requestId!); + expect(["deferred_issue_execution", "queued"]).toContain(held.status); + expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.status === "cancelled")).toEqual([]); + + // Nothing polls a held Retry: settling accounting must wake its owner. + expect(await settleAccounting(f)).toContain(DELIVERY_QUEUES.deferredExecutionReceipts); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) + .where(eq(agentWakeupRequests.id, retry.requestId!)); + await settle(f); await settle(f); + const started = (await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.startedAt); + expect(started).toHaveLength(1); + await settle(f); + expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.startedAt)).toHaveLength(1); + }, 180_000); + + it("starts a new task whose assignment was refused during the hold once accounting settles", async () => { + const f = await seedHeldComment(); + // A second task is created and assigned while the hold lasts. + const [task] = await db.insert(issues).values({ companyId: f.companyId, title: "Created during the hold", status: "todo", + assigneeAgentId: f.agentId, responsibleUserId: f.userId, issueNumber: 2 }).returning(); + await expect(queueIssueAssignmentWakeup({ heartbeat, issue: { id: task.id, assigneeAgentId: f.agentId, status: "todo" }, + reason: "issue_assigned", mutation: "create", contextSource: "issue.create", + requestedByActorType: "user", requestedByActorId: f.userId, rethrowOnError: true })).rejects.toMatchObject({ status: 409 }); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + expect((await issueRuns({ companyId: f.companyId, issueId: task.id }))).toHaveLength(0); + // Nothing polls a refused assignment: settling accounting must wake its recovery. + expect(await settleAccounting(f)).toContain(DELIVERY_QUEUES.strandedAssignments); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.drainActiveRunExecutions(); + const started = (await issueRuns({ companyId: f.companyId, issueId: task.id })).filter(run => run.startedAt); + expect(started).toHaveLength(1); + }, 180_000); + + it("waits out the hold instead of blocking a task whose continuation it refused", async () => { + const f = await seedHeldComment(); + // The task's run ended without settling the task; its comment was already consumed. + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + const [waiting] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(waiting.status).toBe("in_progress"); + expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId)).toHaveLength(0); + + await settleAccounting(f); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.drainActiveRunExecutions(); + expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.startedAt)).toHaveLength(1); + }, 180_000); + + it.each([ + { delivery: "task history", context: (commentId: string) => ({ executionContinuation: { messages: [{ id: commentId }] } }) }, + { delivery: "wake input", context: (commentId: string) => ({ wakeCommentIds: [commentId] }) }, + ])("consumes a saved comment that a later completed run already received as $delivery, without reopening it", async ({ context }) => { + // RC-53: the saved comment's source run was lost; its retry started after + // the comment, received it in its prompt, finished the task, and only then + // did the saved wake drain. + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "process_lost" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation", + triggerDetail: "system", status: "succeeded", startedAt: new Date(), finishedAt: new Date(), retryOfRunId: f.sourceRunId, + contextSnapshot: { issueId: f.issueId, wakeReason: "transient_failure_retry", ...context(f.commentId) } }); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await settle(f); await settle(f); + expect(deliveries(f.commentId)).toHaveLength(0); + expect(await readWake(f.wakeId)).toMatchObject({ status: "cancelled", + error: "Queued messages were already delivered to a completed run" }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("done"); + }, 180_000); + + it.each([ + { shape: "edited after the later run started", edit: true, startedBeforeComment: false }, + { shape: "posted while the later run was already running", edit: false, startedBeforeComment: true }, + ])("still delivers a saved comment $shape", async ({ edit, startedBeforeComment }) => { + // A run's recorded context can be refreshed after it starts, and an edit + // made after it started is unread: neither proves delivery. + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "process_lost" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const startedAt = startedBeforeComment ? new Date(Date.now() - 60_000) : new Date(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation", + triggerDetail: "system", status: "succeeded", startedAt, finishedAt: new Date(), retryOfRunId: f.sourceRunId, + contextSnapshot: { issueId: f.issueId, wakeReason: "transient_failure_retry", + executionContinuation: { messages: [{ id: f.commentId }] } } }); + if (edit) await db.update(issueComments).set({ body: "Edited: also include the FOLLOWUP line.", updatedAt: new Date(Date.now() + 1_000) }) + .where(eq(issueComments.id, f.commentId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await settle(f); await settle(f); + expect(deliveries(f.commentId)).toHaveLength(1); + }, 180_000); + + it("still delivers a saved comment that the later completed run never received", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "process_lost" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation", + triggerDetail: "system", status: "succeeded", startedAt: new Date(), finishedAt: new Date(), retryOfRunId: f.sourceRunId, + contextSnapshot: { issueId: f.issueId, wakeReason: "transient_failure_retry", executionContinuation: { messages: [] } } }); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await settle(f); await settle(f); + expect(deliveries(f.commentId)).toHaveLength(1); + }, 180_000); + + it("still gives a saved comment its own turn when a later non-retry run only saw it as task history", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "process_lost" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation", + triggerDetail: "system", status: "succeeded", startedAt: new Date(), finishedAt: new Date(), + contextSnapshot: { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentIds: [randomUUID()], + executionContinuation: { messages: [{ id: f.commentId }] } } }); + await settle(f); await settle(f); + expect(deliveries(f.commentId)).toHaveLength(1); + }, 180_000); + + it("still refuses work at a reached hard stop, which the board owns", async () => { + const f = await seedHeldComment({ hardStopReached: true }); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + // The comment's wake was promoted earlier; a reached hard stop refuses it terminally. + await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agentId, provider: "openai", model: "gpt", + costCents: 50, occurredAt: new Date() }); + await settle(f); + expect(deliveries(f.commentId)).toHaveLength(0); + expect(await issueRuns(f).then(runs => runs.filter(run => run.status === "queued"))).toHaveLength(0); + }, 180_000); +}); diff --git a/server/src/__tests__/native-run-projection-work.test.ts b/server/src/__tests__/native-run-projection-work.test.ts index d4af716f0a2..52bf0fb1924 100644 --- a/server/src/__tests__/native-run-projection-work.test.ts +++ b/server/src/__tests__/native-run-projection-work.test.ts @@ -11,8 +11,9 @@ import { join } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agentWakeupRequests, agents, authUsers, companies, companyMemberships, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; +import { subscribeDatabaseWork, agentWakeupRequests, agents, authUsers, companies, companyMemberships, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { nativeRunProjectionWork } from "../services/native-runtime/native-run-projection-work.js"; import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; import { NATIVE_RUN_FAILED_UNEXPLAINED } from "../services/native-runtime/native-run-failure-explanation.js"; @@ -744,6 +745,27 @@ describe("committed native run projection lifecycle", () => { expect((await followUps(binding.issueId, binding.runId)).map(run => run.id)).toEqual([followUp.id]); }); + it("hints the saved comment's own wake when recovery projects its run, not only the company", async () => { + // A company hint waits out the deferred-receipt worker's reconstruction + // interval after the decision commit's pass, so the task looked done + // with its message still unread (RC-37, s7401 replay). + const binding = await seed(true); + const { wake } = await deferredComment(binding); + await complete(binding, "control_plane"); await barrier(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + expect((await read(binding.runId)).status).toBe("running"); + const hints: string[] = []; + const stop = subscribeDatabaseWork(db, DELIVERY_QUEUES.deferredExecutionReceipts, (event, received) => { + if (event === "settled") hints.push(...(received ?? [])); + }); + try { + await projectCommittedRun({ db, run: await read(binding.runId), coordinator: await state(binding.runId) }); + } finally { stop(); } + expect((await read(binding.runId)).status).toBe("succeeded"); + expect(hints.map(hint => JSON.parse(hint) as { requestId?: string })).toContainEqual( + expect.objectContaining({ requestId: wake.id })); + }); + it("does not wake again for a comment the recovered run already received as a steer", async () => { const binding = await seed(true); const { comments: [steered, saved], wake } = await deferredComment(binding, diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 9a2c633643c..f8c748820fd 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -314,14 +314,40 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .from(issueComments) .where(and(eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, queryableCommentIds), currentConversationCommentCondition())); const targetsFinishingRunAgent = wakeAgentId === finishingRunAgentId; - const liveNonSelfCommentIds = queryableCommentIds.filter((commentId) => { + const liveAndUnauthored = queryableCommentIds.filter((commentId) => { const row = rows.find((candidate) => candidate.id === commentId); return Boolean(row && !row.deletedAt && (!targetsFinishingRunAgent || row.createdByRunId !== finishingRunId)); }); + // A succeeded run of the wake's agent that started after the comment's + // latest revision consumed it when it received the comment as its own + // wake input, or when it was a retry of an interrupted turn whose task + // history included the comment (the retry re-reads the whole thread). + // Promoting the saved wake again would deliver it twice and can reopen + // finished work. Task history alone does not consume a message saved + // for its own turn, and a run that was already running never counts: + // its context can be refreshed after the fact, and an edit made after + // it started is still unread. + const delivered = new Set(liveAndUnauthored.length === 0 ? [] : (await tx.select({ id: issueComments.id }) + .from(issueComments).where(and( + eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), + inArray(issueComments.id, liveAndUnauthored), + sql`exists (select 1 from ${heartbeatRuns} delivered_run + where delivered_run.company_id = ${companyId} and delivered_run.agent_id = ${wakeAgentId} + and delivered_run.status = 'succeeded' + and (delivered_run.native_issue_id = ${issueId} or delivered_run.context_snapshot->>'issueId' = ${issueId}) + and delivered_run.started_at > ${issueComments.createdAt} + and delivered_run.started_at >= ${issueComments.updatedAt} + and (coalesce(delivered_run.context_snapshot->'wakeCommentIds', '[]'::jsonb) ? ${issueComments.id}::text + or (delivered_run.retry_of_run_id is not null and jsonb_path_exists(delivered_run.context_snapshot, '$.executionContinuation.messages[*] ? (@.id == $id)', + jsonb_build_object('id', ${issueComments.id}::text)) + or jsonb_path_exists(delivered_run.context_snapshot, '$.paperclipWake.messages[*] ? (@.id == $id)', + jsonb_build_object('id', ${issueComments.id}::text)))))`, + ))).map(row => row.id)); + const liveNonSelfCommentIds = liveAndUnauthored.filter(commentId => !delivered.has(commentId)); const containedSelfAuthoredComment = rows.some( (row) => targetsFinishingRunAgent && !row.deletedAt && row.createdByRunId === finishingRunId, ); - return { liveNonSelfCommentIds, containedSelfAuthoredComment }; + return { liveNonSelfCommentIds, containedSelfAuthoredComment, deliveredCommentIds: [...delivered] }; }, async cancelDeferredWake({ companyId, wakeId, agentId, reason, now }) { diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index 927498cb7bb..7cf3cc4e7c8 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -125,7 +125,12 @@ export interface WakeQueueTransaction { finishingRunId: string; finishingRunAgentId: string; queuedCommentIds: string[]; - }): Promise<{ liveNonSelfCommentIds: string[]; containedSelfAuthoredComment: boolean }>; + }): Promise<{ + liveNonSelfCommentIds: string[]; + containedSelfAuthoredComment: boolean; + /** Live comment ids a succeeded run of the wake's agent already received in its prompt. */ + deliveredCommentIds?: string[]; + }>; /** Cancels the wake with `status = 'deferred_issue_execution'` as an atomic compare-and-set guard. */ cancelDeferredWake(input: { companyId: string; diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index ddc41582fa4..10e8c617f25 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -226,7 +226,8 @@ async function runReleaseDrain( continue; } - let liveness = { liveNonSelfCommentIds: candidate.queuedCommentIds, containedSelfAuthoredComment: false }; + let liveness: { liveNonSelfCommentIds: string[]; containedSelfAuthoredComment: boolean; deliveredCommentIds?: string[] } = + { liveNonSelfCommentIds: candidate.queuedCommentIds, containedSelfAuthoredComment: false }; if ( !candidate.authorizedFailedChatRetry && candidate.queuedCommentIds.length > 0 @@ -270,7 +271,9 @@ async function runReleaseDrain( agentId: candidate.agentId, reason: commentAction.selfAuthored ? "Deferred wake contained only comments authored by the finishing run" - : "Queued messages were discarded before promotion", + : liveness.deliveredCommentIds?.length + ? "Queued messages were already delivered to a completed run" + : "Queued messages were discarded before promotion", now: input.now, }); continue; diff --git a/server/src/services/budget-block-ownership.ts b/server/src/services/budget-block-ownership.ts new file mode 100644 index 00000000000..bb89f33e94c --- /dev/null +++ b/server/src/services/budget-block-ownership.ts @@ -0,0 +1,19 @@ +import { and, eq } from "drizzle-orm"; +import { budgetPolicies, type Db } from "@paperclipai/db"; +import { policyBlocks } from "./budgets.js"; + +/** A reached hard stop (spend at the limit, or blocked unpriced usage) is a + * board decision. Any other budget block (completed runs awaiting accounting, + * a native run recovering its accounting, a hold awaiting reconciliation) + * clears by itself. Read only: the admission check already reconciled. */ +export async function isBoardOwnedBudgetBlock( + db: Db, + companyId: string, + block: { scopeType: string; scopeId: string }, +) { + const policies = await db.select().from(budgetPolicies).where(and( + eq(budgetPolicies.companyId, companyId), eq(budgetPolicies.scopeType, block.scopeType), + eq(budgetPolicies.scopeId, block.scopeId), eq(budgetPolicies.isActive, true))); + for (const policy of policies) if (await policyBlocks(db, policy)) return true; + return false; +} diff --git a/server/src/services/deferred-execution-notifications.ts b/server/src/services/deferred-execution-notifications.ts index 1207e59b63a..35878b3ec89 100644 --- a/server/src/services/deferred-execution-notifications.ts +++ b/server/src/services/deferred-execution-notifications.ts @@ -1,5 +1,5 @@ -import type { Db } from "@paperclipai/db"; -import { sql } from "drizzle-orm"; +import { agentWakeupRequests, type Db } from "@paperclipai/db"; +import { and, eq, sql } from "drizzle-orm"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type DeferredExecutionScope = { companyId: string; requestId?: string }; @@ -17,6 +17,18 @@ export async function notifyDeferredExecutionWork(tx: object, scope: DeferredExe await notifyDeliveryWork(tx, DELIVERY_QUEUES.deferredExecutionReceipts, hint(scope)); } +/** The last gate holding one issue's saved wakes cleared (its run reached a + * terminal status or released its environment). Hint each saved wake by id: a + * company hint can wait out the worker's reconstruction interval, which left + * a delivered-looking task with an unread message. */ +export async function notifyDeferredWakesForIssue(tx: object, scope: { companyId: string; issueId: string }) { + const wakes = await (tx as Pick).select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${scope.issueId}`, + )).limit(25); + for (const wake of wakes) await notifyDeferredExecutionWork(tx, { companyId: scope.companyId, requestId: wake.id }); +} + export async function notifyNativeStatusWakeWork(tx: object, scope: DeferredExecutionScope) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusWakeIntents, hint(scope)); } diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index 26c6cae9d99..d3588319726 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -4,7 +4,7 @@ import { notifyNativeEnvironmentSettlement } from "./native-environment-settleme import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; -import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js"; +import { notifyDeferredExecutionWork, notifyDeferredWakesForIssue } from "./deferred-execution-notifications.js"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { randomUUID } from "node:crypto"; @@ -1771,6 +1771,7 @@ export function environmentService(db: Db) { // Messages saved behind a finished run wait for its cleanup to // settle. Reconsider them through normal deferred admission. await notifyDeferredExecutionWork(tx, { companyId: previous.companyId }); + if (previous.issueId) await notifyDeferredWakesForIssue(tx, { companyId: previous.companyId, issueId: previous.issueId }); } if (previous.creationOperationId && previous.status === "pending_cleanup" && options?.cleanupStatus === "success" && status !== "pending_cleanup") { await tx.update(sandboxCreationOperations).set({ state: "resolved", updatedAt: now }) @@ -1905,12 +1906,15 @@ export function environmentService(db: Db) { ): Promise => { const now = new Date(); const rows = await db.transaction(async tx => { - const [run] = await tx.select({ companyId: heartbeatRuns.companyId }).from(heartbeatRuns) - .where(eq(heartbeatRuns.id, heartbeatRunId)).limit(1); + const [run] = await tx.select({ companyId: heartbeatRuns.companyId, + issueId: sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')` }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, heartbeatRunId)).limit(1); if (run) { await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: heartbeatRunId }); await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: heartbeatRunId }); await notifyDeferredExecutionWork(tx, { companyId: run.companyId }); + if (run.issueId && /^[0-9a-f-]{36}$/i.test(run.issueId)) + await notifyDeferredWakesForIssue(tx, { companyId: run.companyId, issueId: run.issueId }); } return tx.update(environmentLeases) .set({ diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index d7de7d7cda3..5cdf1ec3a79 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -186,6 +186,7 @@ import type { HeartbeatRecoveryDependencies, createHeartbeatRecovery } from "./r import { executeIssuePostCommitActions, type IssuePostCommitAction, type issueService } from "../issues.js"; import type { issueTreeControlService } from "../issue-tree-control.js"; import type { budgetService } from "../budgets.js"; +import { isBoardOwnedBudgetBlock } from "../budget-block-ownership.js"; import type { instanceSettingsService } from "../instance-settings.js"; import type { createRunDispatch } from "../../modules/run-dispatch/index.js"; import type { PostCommitEffect as WakeQueuePostCommitEffect, createWakeQueue } from "../../modules/wake-queue/index.js"; @@ -908,6 +909,11 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ); if (budgetBlock) { if (caseHandoffReceipt) return null; + // A temporary hold (accounting that is still settling) only delays + // accepted work. Cancelling the unstarted run would drop the user's + // message, Retry, or recovery it carries; keep it queued. Settling + // accounting re-notifies run dispatch. A reached hard stop is the board's. + if (!run.startedAt && !(await isBoardOwnedBudgetBlock(db, run.companyId, budgetBlock))) return null; await cancelRunInternal(run.id, budgetBlock.reason); return null; } diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 77dd186a30e..20d2eeb72b4 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -14,6 +14,7 @@ import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-pre import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; +import { notifyDeferredWakesForIssue } from "../deferred-execution-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js"; @@ -203,7 +204,10 @@ export async function projectCommittedRun(input: { }) .where(needsRepair) .returning(); - if (after) await settleNativeTerminalWake(tx, after); + if (after) { + await settleNativeTerminalWake(tx, after); + await notifyDeferredWakesForIssue(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); + } return after ? { before, after } : null; }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 8dbeede572e..b84ca31b738 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -117,7 +117,8 @@ import { } from "../activity-log.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; -import { budgetService, policyBlocks } from "../budgets.js"; +import { budgetService } from "../budgets.js"; +import { isBoardOwnedBudgetBlock } from "../budget-block-ownership.js"; import { unadmittedChatWakeupCondition } from "../durable-chat-wakeup.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { PROVIDER_LIMIT_DEFAULT_COOLDOWN_MS } from "../system-recovery-state.js"; @@ -2157,21 +2158,6 @@ export function recoveryService( return Boolean(await readInvocationBudgetBlock(issue, agentId)); } - /** A reached hard stop (spend at the limit, or blocked unpriced usage) is a - * board decision. Any other budget block (completed runs awaiting accounting, - * a native run recovering its accounting, a hold awaiting reconciliation) - * clears by itself. Read only: the admission check already reconciled. */ - async function isBoardOwnedBudgetBlock( - companyId: string, - block: { scopeType: string; scopeId: string }, - ) { - const policies = await db.select().from(budgetPolicies).where(and( - eq(budgetPolicies.companyId, companyId), eq(budgetPolicies.scopeType, block.scopeType), - eq(budgetPolicies.scopeId, block.scopeId), eq(budgetPolicies.isActive, true))); - for (const policy of policies) if (await policyBlocks(db, policy)) return true; - return false; - } - async function reconcileUnassignedBlockingIssues(opts?: { companyId?: string }) { const candidates = await db .select({ @@ -4846,12 +4832,11 @@ export function recoveryService( } } const budgetBlock = await readInvocationBudgetBlock(issue, agentId); - // A task that never started has nothing to recover: a temporary accounting - // hold only delays its first admission. Leave it `todo` so the assigned-todo - // liveness dispatch below admits it once the hold clears. A reached hard - // stop still goes to the board. - if (budgetBlock && issue.status === "todo" && !latestRun - && !(await isBoardOwnedBudgetBlock(issue.companyId, budgetBlock))) { + // A temporary hold (accounting still settling) only delays the task's + // start or continuation; it is not a reason to escalate. Leave the task + // as it is: settling accounting wakes this recovery again, which then + // admits it. A reached hard stop still goes to the board. + if (budgetBlock && !(await isBoardOwnedBudgetBlock(db, issue.companyId, budgetBlock))) { result.skipped += 1; continue; } diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 58bbf21a2ba..7afc306ab70 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -1,4 +1,7 @@ import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { usdToUnits, unitsToCents } from "@paperclipai/shared"; import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm"; import { budgetReservations, agents, environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, runUsageReceipts, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db"; @@ -117,6 +120,15 @@ const nativeProviderWorkNeverStarted = sql`${heartbeatRuns.runtimeMode} = 'nativ * grace plus a margin, a terminal run whose recovery has settled no longer * holds new company work: its debt settles as unpriced, flagged so the * blocking policy still counts it and the board sees which run it was. */ +/** A run's accounting settled: an admission hold may have cleared. Held work + * (queued runs, saved messages and Retries, and tasks whose start or + * continuation was refused) waits passively, so wake its owners. */ +async function notifyAccountingHoldRelease(tx: object, companyId: string) { + await notifyRunDispatch(tx, { companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId }); + await notifyIssueRecovery(tx, { companyId }); +} + export const ACCOUNTING_CLOSURE_TIMEOUT_MS = REMOTE_RUNNER_RECONNECT_GRACE_MS + 3 * 60_000; /** `at` is one shared cutoff, so the actionable scan and the wake query * cannot both miss a run whose bound falls between them. */ @@ -205,9 +217,13 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService let [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update"); if (!run?.costAccountingPending || !terminalStatuses.includes(run.status)) return false; if (run.costAccountedAt) { - if (object(run.usageJson).accountingLateReceiptPending === true) + if (object(run.usageJson).accountingLateReceiptPending === true) { + // Clearing late debt can release an admission hold too. + await notifyAccountingHoldRelease(tx, run.companyId); return reconcileLateRunUsageReceiptInTransaction(tx, run, publications); + } await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId }); + await notifyAccountingHoldRelease(tx, run.companyId); await tx.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, run.id)); await queueBudgetEvaluation(tx, { companyId: run.companyId, agentId: run.agentId, projectId: text(object(object(run.usageJson).ledgerScope).projectId) }); @@ -249,6 +265,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService // Acknowledgement is part of the same transaction. Budget checks inside // this transaction must not mistake this receipt for missing accounting. await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId }); + await notifyAccountingHoldRelease(tx, run.companyId); await tx.update(heartbeatRuns).set({ costAccountingPending: false, costAccountedAt: new Date() }).where(eq(heartbeatRuns.id, run.id)); await tx.update(budgetReservations).set({ state: preProviderFailure ? "released" : "settled", settledAt: new Date() }) .where(and(eq(budgetReservations.companyId, run.companyId), eq(budgetReservations.runId, run.id), eq(budgetReservations.state, "held"))); @@ -326,6 +343,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService await queueBudgetEvaluation(tx, { companyId: run.companyId, agentId: run.agentId, projectId: text(scope.projectId) }); // Late usage reported by a later run can arrive before this charge posts. const lateDebt = usage.accountingLateReceiptPending === true; + if (!lateDebt) await notifyAccountingHoldRelease(tx, run.companyId); await tx.update(heartbeatRuns).set({ costAccountingPending: lateDebt, costAccountedAt: new Date(), accountingProjectionVersion: "v2", accountingLastError: lateDebt ? "late_usage_requires_reconciliation" : null, accountingLastAttemptAt: new Date(), // A complete receipt arriving later may price this charge (see usage-receipts).