From 08f7ac03dc3d0e6f3b38346a06c69f371341d991 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:12:00 -0600 Subject: [PATCH] fix(recovery): keep a never-started task waiting through a temporary accounting hold (#19702) 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 > - Admission checks every wake. A short company accounting hold can refuse new work after a control-plane restart. > - When the hold refused a new task's assignment wake, stranded-work recovery found the task with no run while the hold was still active. It moved the task to `blocked` with a board recovery comment. > - A `blocked` task leaves the stranded-assignment sweep, so nothing woke it after the hold cleared. The task stayed `blocked` with no run, no interruption and no Retry. > - The hold was temporary. Tasks created a few seconds later (after the sweep) started normally once accounting settled. > - This pull request keeps a never-started task `todo` while a temporary budget hold lasts. The existing assigned-todo liveness dispatch then admits it once the hold clears. The task shows the hold it waits for. > - The benefit is that a restart no longer silently drops new tasks created during the accounting settle window. ## Linked Issues or Issue Description **What happened?** In a restart-chaos storm on master `06bbcea12f`, three new tasks were created while the company accounting hold was active. Each assignment wake got 409 "Company cannot start work while completed runs await accounting." (or "...native run recovers its unfinished accounting."). One task rested `blocked` with 0 runs, interruption `null` and `canRetry=false`. The other two ran later. **Root cause** `reconcileStrandedAssignedIssues` (`server/src/services/recovery/service.ts`, the `readInvocationBudgetBlock` branch near line 4826) escalates any assigned task to `blocked` when a budget block exists. It does this before the assigned-todo no-run branch, which already skips a budget-blocked task and dispatches it later. So a task that never started was treated as failed recovery work. The refused wake itself is stored as a `skipped` `budget.blocked` diagnostic, which is correct; the escalation removed the only path that re-admits it. **Expected behavior** A wake refused by a temporary admission hold waits and is admitted once the hold clears, with one run. A real hard stop is still a board decision. Removed assignees and cancelled tasks still do not wake. **Steps to reproduce** Create a company with a hard-stop budget and a completed run that has `cost_accounting_pending = true`. Assign a new `todo` task. The wake gets 409. Run the stranded-assignment sweep. Before this change the task becomes `blocked` and never runs. Related: #19700 bounds the accounting hold itself. This PR is independent: it handles what happens to a task whose first wake the hold refused. ## What Changed - Stranded-work recovery skips a `todo` task with no run while a temporary budget hold blocks admission. It does not escalate it to `blocked`. The existing assigned-todo liveness dispatch admits the task once the hold clears. - A hard stop (spend at the limit, or blocked unpriced usage) keeps the existing board escalation. Recovery checks this with the read-only `policyBlocks` for the policies of the blocking scope. It does not change budget hold semantics. - The task interruption projection now covers a `todo` task with no run whose latest wake was refused with `budget.blocked`. It shows "Waiting for the budget to allow new work. The task starts automatically when it does." with no Retry. A budget pause shows the existing "budget" user action. - The no-run projection applies the same person-owned gates as the run-backed projection: a budget pause shows the "budget" action, and an ordinary agent, project, company or task pause shows "resume". Only an accounting hold shows the automatic-start text. - `ExecutionBlockerNotice` shows that wait text for a `working` presentation that carries a wait reason. - `doc/execution-semantics.md` records the rule. ## Verification - New embedded-Postgres regression `server/src/__tests__/heartbeat-refused-assignment-wake.test.ts`: - hold active, new task assigned, wake refused (409, `skipped` `budget.blocked`), two recovery sweeps: the task stays `todo` with no run and shows the wait reason. Before the fix this failed with `expected 'blocked' to be 'todo'` (the CHA-9 shape). - hold cleared, two sweeps: exactly one provider turn starts, from `issue.assigned_todo_liveness_dispatch`, and the task finishes. A later sweep does not start another turn. - counter-tests: a terminated assignee and a cancelled task get no run after the hold clears. - Existing `creates a board recovery action for budget-blocked assigned work` (real hard stop) still passes unchanged. - `pnpm exec vitest run src/__tests__/heartbeat-refused-assignment-wake.test.ts src/__tests__/heartbeat-process-recovery.test.ts src/services/task-interruption-presentation.test.ts src/__tests__/heartbeat-issue-liveness-escalation.test.ts` (server) and `src/components/ExecutionBlockerNotice.test.tsx` (ui). - Server typecheck and `pnpm check:token-gates`. - Live restart-chaos storm (local, codex/claude/opencode, 6 in flight, K2 every 3-4 minutes for 20 minutes, no RC-06 mask), revision e38bb5c8f7: 17 tasks, 10 pass. - Two new tasks (CHA-16, CHA-17) were created while the company accounting hold was active. Their assignment wakes got 409 "Company cannot start work while completed runs await accounting." Both stayed `todo` (not `blocked`) and showed `state: working`, `canRetry: false`, `retryWait.reason: budget.blocked`, "Waiting for the budget to allow new work. The task starts automatically when it does." No task rested `blocked` with 0 runs. - The hold itself never cleared in this wave (`pendingRunCount: 1` at wave end), so these two tasks did not start before their deadline. That stuck hold is the separate accounting-closure defect that #19700 bounds. The other 5 failures (`R3:retry_no_new_run:B5`) are Retries refused by the same stuck hold. The re-admission after the hold clears is proved by the embedded-Postgres regression above. ## Risks - Low. The change narrows one escalation to tasks that never started, under a budget block that is not a hard stop. Those tasks now stay `todo` and are re-checked by the existing 30-second stranded-assignment sweep, which already re-checks every admission gate. - The hard-stop check reads the blocking scope's active policies once per budget-blocked never-started task in a sweep. It runs only while a hold is active. - The interruption projection adds one indexed read (company + payload issue id) for `todo` tasks in the batch. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, test execution). ## 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 - [ ] 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-refused-assignment-wake.test.ts | 151 ++++++++++++++++++ server/src/services/recovery/service.ts | 27 +++- .../task-interruption-presentation.test.ts | 30 ++++ .../task-interruption-presentation.ts | 33 +++- .../ExecutionBlockerNotice.test.tsx | 6 + ui/src/components/ExecutionBlockerNotice.tsx | 3 + 7 files changed, 249 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 34cc6a4bdd8..c451b57df87 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..c3cef9ee955 --- /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; agentId: string }) => db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.agentId, f.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.test.ts b/server/src/services/task-interruption-presentation.test.ts index 4f29d3a02d9..dde8e1c42ee 100644 --- a/server/src/services/task-interruption-presentation.test.ts +++ b/server/src/services/task-interruption-presentation.test.ts @@ -48,6 +48,36 @@ describe("queued dependency wait presentation", () => { }); }); +describe("never-started task refused by admission", () => { + const todo = { ...issue, status: "todo" }; + const admissionWait = { id: "wake", status: "skipped", reason: "budget.blocked", agentId: "agent" }; + it("names an accounting hold as an automatic wait without a Retry", () => { + expect(projectTaskInterruption(todo, { agent, admissionWait, paused: false, approval: false })).toMatchObject({ + responsibility: "system", state: "working", canRetry: false, + retryWait: { reason: "budget.blocked", message: expect.stringContaining("starts automatically") } }); + }); + it("keeps pauses and budget stops as user actions instead of promising an automatic start", () => { + for (const [state, action] of [ + [{ project: { pausedAt: new Date(), pauseReason: null } }, "resume"], + [{ company: { status: "paused", pauseReason: "manual" } }, "resume"], + [{ paused: true }, "resume"], + [{ agent: { status: "paused", pauseReason: "manual" } }, "resume"], + [{ company: { status: "paused", pauseReason: "budget" } }, "budget"], + [{ approval: true }, "approval"], + ] as const) { + expect(projectTaskInterruption(todo, { agent, admissionWait, paused: false, approval: false, ...state })) + .toMatchObject({ responsibility: "user", state: "waiting_for_user", userAction: action }); + } + }); + it("shows nothing for other refusals, started tasks or another assignee", () => { + for (const changed of [{ reason: "agent.not_invokable" }, { status: "queued" }, { agentId: "other" }]) { + expect(projectTaskInterruption(todo, { agent, admissionWait: { ...admissionWait, ...changed }, paused: false, approval: false })) + .toBeNull(); + } + expect(projectTaskInterruption({ ...todo, status: "cancelled" }, { agent, admissionWait, paused: false, approval: false })).toBeNull(); + }); +}); + describe("unadmitted chat Retry presentation", () => { const now = Date.parse("2026-01-01T12:00:00Z"); const failed = { ...run, errorCode: "native_provider_terminal_failed", resultJson: null }; diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index d4f2e1f07c9..40804d66f47 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,20 @@ 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 }; + // Budget pauses, ordinary pauses and approvals wait on a person; only an + // accounting hold settles by itself. Admission records all of them as budget.blocked. + const userAction = !issue.assigneeAgentId ? "assign_agent" : input.approval ? "approval" + : [agent?.pauseReason, project?.pauseReason, company?.pauseReason].some(reason => reason === "budget" || reason === "budget_exceeded") + ? "budget" : input.paused || project?.pausedAt || agent?.status === "paused" || company?.status === "paused" ? "resume" : null; + if (userAction) return { ...waiting, responsibility: "user", state: "waiting_for_user", userAction }; + 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 +244,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 +323,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 +420,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 +432,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 (