From e38bb5c8f70e809dcf9d0590c50f7898caaacfad Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 6 Nov 2026 19:35:20 -0600 Subject: [PATCH] fix(recovery): keep a never-started task waiting through a temporary accounting hold A new task's assignment wake refused by a short post-restart accounting hold was turned into a dead blocked task by stranded-work recovery, with no run and no Retry. Recovery now leaves a task with no run in todo while a temporary budget hold lasts, so the existing assigned-todo liveness dispatch admits it once the hold clears. A reached hard stop stays a board decision. The task names the hold it waits for. Co-Authored-By: Claude Opus 5.5 --- doc/execution-semantics.md | 2 + .../heartbeat-refused-assignment-wake.test.ts | 151 ++++++++++++++++++ server/src/services/recovery/service.ts | 27 +++- .../task-interruption-presentation.ts | 30 +++- .../ExecutionBlockerNotice.test.tsx | 6 + ui/src/components/ExecutionBlockerNotice.tsx | 3 + 6 files changed, 216 insertions(+), 3 deletions(-) create mode 100644 server/src/__tests__/heartbeat-refused-assignment-wake.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7672ca6362f..2a3752f39df 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -491,6 +491,8 @@ and the user need not certify infrastructure state in a reconciliation form. Repeated automatic signals for an unchanged gate share one durable skipped-wake diagnostic, scoped to company, agent, issue, gate code, and condition identity. The diagnostic retains the first request and counts later observations. This applies to execution reconciliation, dependencies, pause holds, company and agent availability, budget blocks, and disabled heartbeats. These diagnostics do not consume provider attempts and are never proof that a future wake was delivered. All current gates are checked again on the next wake, including the periodic dependency reconciliation sweep. Clearing one gate does not bypass another. +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. + 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-refused-assignment-wake.test.ts b/server/src/__tests__/heartbeat-refused-assignment-wake.test.ts new file mode 100644 index 00000000000..d130de2e83b --- /dev/null +++ b/server/src/__tests__/heartbeat-refused-assignment-wake.test.ts @@ -0,0 +1,151 @@ +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 } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, agentWakeupRequests, authUsers, budgetPolicies, companies, companyMemberships, createDb, heartbeatRuns, + issues } from "@paperclipai/db"; +import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { queueIssueAssignmentWakeup } from "../services/issue-assignment-wakeup.js"; +import { taskInterruptionPresentation } from "../services/task-interruption-presentation.js"; +import { recoveryPostgresSupport, startRecoveryPostgres } from "./helpers/recovery-postgres.js"; + +const describePostgres = recoveryPostgresSupport.supported ? describe : describe.skip; +if (!recoveryPostgresSupport.supported) console.warn(`Skipping refused assignment wake tests: ${recoveryPostgresSupport.reason}`); + +/** + * A short company admission hold (for example accounting that is still settling + * after a restart) can refuse a new task's assignment wake. The task must wait + * honestly and start once the hold clears, instead of resting `blocked` with no run. + */ +describePostgres("refused assignment wake during a temporary admission hold", () => { + let temporary: Awaited>; + let db: ReturnType; + let heartbeat: ReturnType; + const adapterType = "codex_local"; + const dispatched: string[] = []; + const oldHome = process.env.PAPERCLIP_HOME; + let home: string; + + beforeAll(async () => { + temporary = await startRecoveryPostgres("heartbeat-refused-assignment-"); + db = createDb(temporary.connectionString); + home = await mkdtemp(path.join(tmpdir(), "paperclip-refused-assignment-")); + process.env.PAPERCLIP_HOME = home; + heartbeat = heartbeatService(db); + registerServerAdapter({ + type: adapterType, supportsLocalAgentJwt: false, + execute: async input => { + dispatched.push(input.runId); + 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 company with a hard-stop budget and one completed run whose accounting + * has not settled yet: admission refuses new work until it settles. */ + async function seedHeldCompany() { + const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(), issueId = randomUUID(); + const userId = `hold-user-${companyId}`; + await db.insert(authUsers).values({ id: userId, name: "Hold user", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companies).values({ id: companyId, name: "Accounting hold fixture", issuePrefix: `H${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: 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: {} }).returning(); + await db.insert(issues).values({ id: issueId, companyId, title: "New task created during the hold", status: "todo", + assigneeAgentId: agentId, responsibleUserId: userId, issueNumber: 1 }); + return { companyId, agentId, issueId, userId, pendingRunId: pending.id }; + } + + async function assign(f: Awaited>) { + return queueIssueAssignmentWakeup({ heartbeat, issue: { id: f.issueId, assigneeAgentId: f.agentId, status: "todo" }, + reason: "issue_assigned", mutation: "create", contextSource: "issue.create", + requestedByActorType: "user", requestedByActorId: f.userId, rethrowOnError: true }); + } + + const issueRuns = (f: { companyId: string; issueId: string }) => db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.agentId, (f as { agentId: string }).agentId))); + + it("keeps the task waiting during the hold and starts exactly one run once it clears", async () => { + const f = await seedHeldCompany(); + await expect(assign(f)).rejects.toMatchObject({ status: 409 }); + const [refused] = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.agentId, f.agentId))); + expect(refused).toMatchObject({ status: "skipped", reason: "budget.blocked" }); + + // Recovery sweeps while the hold lasts must not turn the wait into a dead `blocked` task. + 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("todo"); + expect(await issueRuns(f)).toHaveLength(0); + // The task names what it waits for; it does not offer a Retry that cannot start either. + expect(await taskInterruptionPresentation(db, waiting)).toMatchObject({ responsibility: "system", state: "working", + canRetry: false, retryWait: { reason: "budget.blocked", message: expect.stringContaining("starts automatically") } }); + + // Accounting settles: the hold clears. + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.drainActiveRunExecutions(); + // Exactly one provider turn starts the task. (A post-run continuation the + // adapter fixture leaves behind is cancelled unstarted once the task is done.) + const providerTurns = async () => { + const ids = new Set((await issueRuns(f)).map(run => run.id)); + return dispatched.filter(id => ids.has(id)); + }; + expect(await providerTurns()).toHaveLength(1); + const [started] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, (await providerTurns())[0])); + expect(started).toMatchObject({ status: "succeeded", contextSnapshot: expect.objectContaining({ + source: "issue.assigned_todo_liveness_dispatch" }) }); + const [finished] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(finished.status).toBe("done"); + + // A later sweep does not start a duplicate. + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.drainActiveRunExecutions(); + expect(await providerTurns()).toHaveLength(1); + }, 180_000); + + it("does not wake a task whose assignee was removed while the hold lasted", async () => { + const f = await seedHeldCompany(); + await expect(assign(f)).rejects.toMatchObject({ status: 409 }); + await db.update(agents).set({ status: "terminated" }).where(eq(agents.id, f.agentId)); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.drainActiveRunExecutions(); + expect(await issueRuns(f)).toHaveLength(0); + }, 180_000); + + it("does not wake a task cancelled while the hold lasted", async () => { + const f = await seedHeldCompany(); + await expect(assign(f)).rejects.toMatchObject({ status: 409 }); + await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + await heartbeat.drainActiveRunExecutions(); + expect(await issueRuns(f)).toHaveLength(0); + const [cancelled] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(await taskInterruptionPresentation(db, cancelled)).toBeNull(); + }, 180_000); +}); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 01c60d35d0e..8dbeede572e 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -77,6 +77,7 @@ import { chatConversations, caseHandoffs, companies, + budgetPolicies, heartbeatRunWatchdogDecisions, heartbeatRuns, issueAttachments, @@ -116,7 +117,7 @@ import { } from "../activity-log.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; -import { budgetService } from "../budgets.js"; +import { budgetService, policyBlocks } from "../budgets.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"; @@ -2156,6 +2157,21 @@ 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({ @@ -4830,6 +4846,15 @@ 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))) { + result.skipped += 1; + continue; + } if (budgetBlock) { const classification = classifyContinuationFailure(latestRun); if ( diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index d4f2e1f07c9..cf05490c6ea 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -59,6 +59,8 @@ type ProjectionState = { company?: Pick; paused: boolean; approval: boolean; + /** The latest wake for a task with no run yet, when admission refused it. */ + admissionWait?: { id: string; status: string; reason: string | null; agentId: string }; }; function boundedJsonString(value: SQL, limit: number) { @@ -89,6 +91,12 @@ const retryWaitMessages: Record = { retry_source_changed: "Retry cannot continue because its saved message changed. Send a new message to continue from saved work.", }; +// A task that never started can wait for a temporary admission hold. The +// assigned-task liveness sweep admits it once the hold clears; no Retry needed. +const admissionWaitMessages: Record = { + "budget.blocked": "Waiting for the budget to allow new work. The task starts automatically when it does.", +}; + /** Recognize old persisted failures without displaying arbitrary run error text. */ function deniedProjectId(issue: ProjectionIssue, run?: ProjectionRun): string | null | undefined { const historical = run?.errorCode === "setup_failed" && run.error === "Project is outside this low-trust boundary."; @@ -104,7 +112,17 @@ function deniedProjectId(issue: ProjectionIssue, run?: ProjectionRun): string | export function projectTaskInterruption(issue: ProjectionIssue, input: ProjectionState, now = Date.now()): TaskInterruptionPresentation | null { const { run, wait, action, agent, project, company } = input; const retryWait = input.retryWait ?? wait; - if (!run) return null; + if (!run) { + const admission = input.admissionWait; + if (!admission || admission.status !== "skipped" || !Object.hasOwn(admissionWaitMessages, admission.reason ?? "") + || issue.status !== "todo" || admission.agentId !== issue.assigneeAgentId) return null; + const waiting: TaskInterruptionPresentation = { intentId: admission.id, responsibility: "system", state: "working", + nextRetryAt: null, canRetry: false, userAction: null, sourceRunIds: [], observedRunId: null }; + // A budget pause waits on a person raising the budget; an accounting hold settles by itself. + if ([agent?.pauseReason, project?.pauseReason, company?.pauseReason].some(reason => reason === "budget" || reason === "budget_exceeded")) + return { ...waiting, responsibility: "user", state: "waiting_for_user", userAction: "budget" }; + return { ...waiting, retryWait: { reason: admission.reason!, message: admissionWaitMessages[admission.reason!] } }; + } const system = readSystemRecovery(action?.evidence); const base: TaskInterruptionPresentation = { intentId: wait?.id ?? null, responsibility: "system", state: "resolved", nextRetryAt: null, canRetry: false, userAction: null, @@ -223,7 +241,8 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const isRetryWait = sql`coalesce(jsonb_typeof(${agentWakeupRequests.payload}#>'{taskRetry,sourceRunId}') = 'string', false)`; const agentIds = selected.flatMap(issue => issue.assigneeAgentId ? [issue.assigneeAgentId] : []); const projectIds = selected.flatMap(issue => issue.projectId ? [issue.projectId] : []); - const [runs, waits, agentRows, projectRows, pauseRows, approvalRows, companyRows] = await Promise.all([ + const startingIssueIds = selected.filter(issue => issue.status === "todo" && issue.assigneeAgentId).map(issue => issue.id); + const [runs, waits, agentRows, projectRows, pauseRows, approvalRows, companyRows, admissionWaits] = await Promise.all([ db.select({ issueId: requestedIssueId, id: heartbeatRuns.id, status: heartbeatRuns.status, agentId: heartbeatRuns.agentId, companyId: heartbeatRuns.companyId, nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, startedAt: heartbeatRuns.startedAt, @@ -301,6 +320,11 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i .where(and(eq(issueApprovals.companyId, companyId), inArray(issueApprovals.issueId, ids), inArray(approvals.status, ["pending", "revision_requested"]))), db.select({ status: companies.status, pauseReason: companies.pauseReason }).from(companies).where(eq(companies.id, companyId)), + // Only consulted for tasks without a run: the latest wake says whether admission refused it. + startingIssueIds.length ? db.selectDistinctOn([wakeIssue], { issueId: wakeIssue, id: agentWakeupRequests.id, + status: agentWakeupRequests.status, reason: agentWakeupRequests.reason, agentId: agentWakeupRequests.agentId }) + .from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), inArray(wakeIssue, startingIssueIds))) + .orderBy(wakeIssue, desc(agentWakeupRequests.requestedAt), desc(agentWakeupRequests.id)) : [], ]); // Advertise a pending resume only when the scheduler would admit it, using // its own check, so the task never hides Retry behind a deferred candidate @@ -393,6 +417,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const agentMap = new Map(agentRows.map(agent => [agent.id, agent])); const projectMap = new Map(projectRows.map(project => [project.id, project])); const pauses = new Set(pauseRows.map(row => row.issueId)), approvalsSet = new Set(approvalRows.map(row => row.issueId)); + const admissionWaitMap = byIssue(admissionWaits); for (const issue of selected) { const required = requirements.get(issue.id); const denied = required?.projectIds.filter(projectId => currentAccess.get(`${issue.assigneeAgentId}:${projectId}`) !== true); @@ -404,6 +429,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i workspaceSelectionAvailable: required?.available ?? false, agent: agentMap.get(issue.assigneeAgentId ?? ""), project: projectMap.get(issue.projectId ?? ""), company: companyRows[0], paused: pauses.has(issue.id), approval: approvalsSet.has(issue.id), + admissionWait: admissionWaitMap.get(issue.id), }); result.set(issue.id, presentation); const run = runMap.get(issue.id); diff --git a/ui/src/components/ExecutionBlockerNotice.test.tsx b/ui/src/components/ExecutionBlockerNotice.test.tsx index 6386dbe7e32..214747a6050 100644 --- a/ui/src/components/ExecutionBlockerNotice.test.tsx +++ b/ui/src/components/ExecutionBlockerNotice.test.tsx @@ -41,6 +41,12 @@ describe("ordinary task interruption", () => { it.each(["working", "resolved"] as const)("does not duplicate the thread indicator for %s", async state => { await render({ ...failed, state }); expect(container.textContent).toBe(""); }); + it("names the admission hold a task that has not started waits for", async () => { + await render({ ...failed, state: "working", canRetry: false, sourceRunIds: [], observedRunId: null, + retryWait: { reason: "budget.blocked", message: "Waiting for the budget to allow new work. The task starts automatically when it does." } }); + expect(container.textContent).toBe("Waiting for the budget to allow new work. The task starts automatically when it does."); + expect(container.querySelector("button")).toBeNull(); + }); it("names the saved-work recovery step instead of a bare queue", async () => { await render({ ...failed, state: "working", canRetry: false, recoveryStep: "saving_previous_files" }); expect(container.textContent).toBe("Saving files from the previous attempt…"); diff --git a/ui/src/components/ExecutionBlockerNotice.tsx b/ui/src/components/ExecutionBlockerNotice.tsx index 583fe53bca2..7df4018b1bf 100644 --- a/ui/src/components/ExecutionBlockerNotice.tsx +++ b/ui/src/components/ExecutionBlockerNotice.tsx @@ -46,6 +46,9 @@ export function ExecutionBlockerNotice({ presentation, canRetry, onRetry, optimi } if (step && presentation?.state === "working") return
{`${step}…`}
; + // A task that has not started yet names the admission hold it waits for. + if (!optimistic && presentation?.state === "working" && presentation.retryWait) return
{presentation.retryWait.message}
; if (optimistic || !presentation || presentation.responsibility !== "system" || !["failed", "retry_scheduled"].includes(presentation.state)) return null; return (