From d33b3ac2632442acb816a1ebe551258561e79d27 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:43:17 -0600 Subject: [PATCH] fix(wake-queue): save temporarily refused work intents and promote them through one path (#19956) 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 > - Users send work to agents as assignments, comments and Retries. The wake queue delivers each one as a run > - When an admission check refused a wake for a temporary reason, the wake often ended as a terminal skip, so the work was lost > - Restart-chaos waves found dropped comments and Retries (RC-37, s7532) and double delivery (RC-53). Each fix covered one hold type > - The durable ownership plan (section E) asks for one outbox rule: a closed set of terminal outcomes, a saved deferral for every temporary refusal, and one promotion path > - This pull request adds that rule. It saves messages refused by a pause, promotes deferred intents from every hold-release event through one function, and adds a reconciler backstop > - The benefit is that an accepted message is delivered exactly once after a pause lifts, and the reconciler reports any intent that is stuck or dropped ## Linked Issues or Issue Description Refs #19831 (plan, section E), #19838 (inventory, wake-queue rows and section 1b), #19927 (Phase 2 data: the I4 false positive), #19702, #19741, #19829, #19826. Builds on #19944 (section D, merged). **What happened?** - A user commented while the agent was paused, or while the company was paused. Admission refused the wake. Nothing saved the message as pending, so when the pause lifted, nothing delivered it. - Hold-release events woke different subsets of workers. A saved intent then waited out a 30-second refusal backoff after its hold cleared. - The I4 reconciler check matched free text with a regular expression. Phase 2 found a false positive: a delivered retry whose error said "retrying once". **Expected behavior** An accepted work intent ends only as delivered, consumed by a later run, task cancelled, or hard stop. Every temporary refusal saves the intent as `deferred` with its reason and its release trigger. Every hold-release event promotes deferred intents through one function (invariant 4). **Steps to reproduce** Pause an agent. Post a user comment on its task. Resume the agent. On `master` the comment is never delivered as a wake. ## What Changed - New `server/src/services/work-intent-outbox.ts`: - `WORK_INTENT_TERMINAL_OUTCOMES`: the closed set of terminal outcomes. - `TEMPORARY_REFUSAL_TRIGGERS`: maps each temporary refusal code to its release trigger. - `promoteDeferredWorkIntents`: the one promotion function. It removes the 30-second refusal backoff from the matching deferred intents. It then wakes run dispatch, the saved-receipt worker and stranded-work recovery, and hints each saved intent of the issue. - `heartbeat/queue.ts`: the #19829 save-as-deferred block is now a shared helper. A user's message refused by a paused agent, a paused company, or a budget pause that is not a reached hard stop is saved with `executionWait: { reason, trigger, message }`. The caller still gets a 409. The message is saved only when the task's latest run is terminal, because the release drain promotes a saved message behind that run. A task that never ran keeps the message in the task, and stranded-work recovery starts the task once the pause lifts. A `pending_approval` agent and a task-tree hold are not changed here: the release drain still fails a promoted wake for a pending approval, and tree holds have their own resume path. - Hold-release events call `promoteDeferredWorkIntents`: - accounting settled (`run-cost-accounting.ts`); - environment lease released (`environments.ts`, two sites); - native run projected terminal (`native-run-projection.ts`); - budget pause lifted (`budgets.ts`); - company reactivated (`companies.ts`); - task-tree hold released (`issue-tree-control.ts`); - agent lifecycle ready (`agent-lifecycle-driver.ts`). - Reconciler (`recovery/ownership-registry.ts`, `ownership-invariant-report.ts`, `native-finalization-reconciler.ts`): - registers three new kinds: `wake_request_deferred`, `queued_run` and `scheduled_retry_run`; - I4 now reads the closed refusal code (the wake `reason`) of accepted intents, not free text; - a new I4 check reports a deferred pause refusal whose pause has lifted for more than 5 minutes. The worker then promotes it as a backstop; - a lifted pause inside its 5-minute grace window returns its deadline through `nextDueAt`, and the worker also wakes on agent-lifecycle signals (company and agent pause changes). So a missed release event is still checked. - `doc/SPEC-implementation.md`: adds the rule "Accepted work intents are delivered exactly once". ## Verification Each focused file ran alone in one vitest process (`--maxWorkers=1`) on embedded Postgres. Results at `2875c85fd2`: | Question | Loop | Elapsed | Result | |---|---|---|---| | Is a comment refused by a pause saved, then delivered once after the pause lifts? | `heartbeat-held-comment-wake.test.ts -t "pause lifts"` | 8 s | 2 pass. With `queue.ts` from `master`, both fail: no intent is saved | | Does I4 use the closed code, and does it catch a stuck deferral? | `ownership-reconciliation-report.test.ts` | 45 s | 20 pass. This includes the Phase 2 "retrying once" false positive and the case "paused, then lifted" | | No regression on the paths that changed | held-comment (15), archived-company guard (8), undispatched execution wait (32), deferred receipt work (45), recovery pause durability (16), native run projection work (55), tree-control routes and services (19, 1, 7) | 8–334 s each | All pass | - Exactly-once: each pause test runs two promotions (a release event, then the reconciler backstop) and many dispatch passes. It asserts one delivery. The existing consumption tests (RC-53) still pass. - Server package typecheck passes. - Speed: nothing new runs before `turn.start`. The new admission work happens only on a refusal. Promotion runs in the release transaction and in the background reconciler. - Not run here: chaos waves. The coordinating session schedules them on the shared harness. ## Risks - `promoteDeferredWorkIntents` moves `updated_at` back for deferred intents that were refused in the last 30 seconds. A busy company with many release events can retry refused intents more often. Each retry re-runs normal admission, and the refusal resets the backoff. - More release sites now wake run dispatch and stranded-work recovery. These notifications are coalesced per company. - Lifting a pause is now the operator's explicit authorization to send what the pause refused, as Send is after a Stop. When the pause stopped the task's run, the lifted pause issues the task's next explicit wake, which adopts the saved message once. Other messages queued behind an acknowledged Stop keep the existing rule. If Stop recovery still holds the task, the message waits for that recovery. - An agent pause's stop sweep keeps the messages that the pause deferred (`preservePauseDeferred`). Terminate and cleanup stops still end them. - The backstop check reads only pause refusals (`company.inactive`, `agent.not_invokable`), because those holds can be checked exactly. Other deferred kinds keep their existing promotion paths. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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/SPEC-implementation.md | 12 + doc/execution-semantics.md | 2 + .../heartbeat-held-comment-wake.test.ts | 207 ++++++++++++++++++ .../ownership-reconciliation-report.test.ts | 54 ++++- .../__tests__/driver-scope.test.ts | 3 +- .../modules/wake-queue/adapters/postgres.ts | 30 +-- server/src/services/agent-lifecycle-driver.ts | 5 +- server/src/services/budgets.ts | 3 + server/src/services/companies.ts | 2 + server/src/services/environments.ts | 9 +- server/src/services/heartbeat.ts | 5 +- server/src/services/heartbeat/queue.ts | 155 ++++++++++--- server/src/services/heartbeat/run-control.ts | 12 + server/src/services/issue-tree-control.ts | 4 +- .../native-finalization-reconciler.ts | 18 ++ .../native-runtime/native-run-projection.ts | 4 +- .../src/services/queued-comment-delivery.ts | 36 +++ .../recovery/ownership-invariant-report.ts | 73 +++++- .../services/recovery/ownership-registry.ts | 25 +++ server/src/services/run-cost-accounting.ts | 8 +- server/src/services/work-intent-outbox.ts | 80 +++++++ 21 files changed, 668 insertions(+), 79 deletions(-) create mode 100644 server/src/services/queued-comment-delivery.ts create mode 100644 server/src/services/work-intent-outbox.ts diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 12ebce5d86d..acb0353706d 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1793,6 +1793,18 @@ per task, ends when the owner releases them, and wakes the successor through the work-intent promotion path. A predecessor run row that no longer exists owns nothing and never holds its successor. +**Accepted work intents are delivered exactly once.** An accepted work intent +(an assignment, a comment, a Retry or a recovery) ends in one of four terminal +outcomes only: delivered, consumed by a later run that received it, task +cancelled, or a reached hard stop. Every other refusal is temporary. The intent +is saved as a deferred wake whose `executionWait` names the refusal and the +event that clears it (a lifted pause, a released lock, settled accounting, a +released task-tree hold, or resumed scheduling). Each such event promotes +deferred intents through one promotion function, and the ownership reconciler +promotes any intent whose hold cleared without that event. Promotion re-runs +normal admission. Consumption is recorded per comment or intent id, so a +promotion never delivers an intent that a later run already received. + Accepted cumulative per-run usage checkpoints contribute observed running spend before the final ledger charge. Admission counts each unfinished operation's maximum of its reservation and observed spend; hard stops count actual ledger diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 516d2888527..bca2cb3d3fa 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -512,6 +512,8 @@ Repeated automatic signals for an unchanged gate share one durable skipped-wake Another run's unfinished accounting or cleanup never refuses new work (see `doc/SPEC-implementation.md`, "Admission is independent of cleanup and accounting"). A new task whose first assignment wake is refused by a temporary condition (an explicit pause) stays `todo`. Stranded-work recovery does not move a task with no run to `blocked` for such a condition. The task shows what it waits for, and the assigned-task liveness sweep admits it once the condition 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. +A user's message refused by a pause (a paused agent, a paused company, or a budget pause that is not a reached hard stop) is saved as a deferred intent that names the refusal and its release trigger; it is never written as a terminal skip. The pause's own stop keeps it. Lifting the pause is the operator's explicit authorization to send what the pause refused, as Send is after a Stop: when the pause stopped the task's run, the lifted pause issues the task's next explicit wake, which adopts the saved message once. A task that never ran keeps the message in the task, and stranded-work recovery starts it. + The same holds for accepted work beyond a first assignment. A temporary refusal 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. diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index 2e0efced5a5..2ee971fd4fb 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -12,6 +12,7 @@ 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 { promoteDeferredWorkIntents, TEMPORARY_REFUSAL_TRIGGERS, WORK_INTENT_TERMINAL_OUTCOMES } from "../services/work-intent-outbox.js"; import { recoveryPostgresSupport, startRecoveryPostgres } from "./helpers/recovery-postgres.js"; const describePostgres = recoveryPostgresSupport.supported ? describe : describe.skip; @@ -193,6 +194,212 @@ describePostgres("saved intents while another run's accounting is pending", () = expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).not.toBe("blocked"); }, 180_000); + // Section E (invariant 4): a pause is the temporary hold left after Phase 3. + // A message refused by it is saved with its refusal and trigger, promoted + // by the one hold-release function, and delivered exactly once. + it.each([ + { hold: "agent pause", refusal: "agent.not_invokable", + apply: (f: { agentId: string }) => db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)), + lift: (f: { agentId: string }) => db.update(agents).set({ status: "idle" }).where(eq(agents.id, f.agentId)) }, + { hold: "company pause", refusal: "company.inactive", + apply: (f: { companyId: string }) => db.update(companies).set({ status: "paused", pauseReason: "manual", pausedAt: new Date() }) + .where(eq(companies.id, f.companyId)), + lift: (f: { companyId: string }) => db.update(companies).set({ status: "active", pauseReason: null, pausedAt: null }) + .where(eq(companies.id, f.companyId)) }, + ])("saves a message refused by a $hold and delivers it once when the pause lifts", async ({ refusal, apply, lift }) => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + await apply(f as never); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "Posted during the pause." }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toMatchObject({ status: 409 }); + const saved = (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) + .filter(wake => (wake.payload as { commentId?: string }).commentId === comment.id); + // Never a terminal skip: one deferred intent naming its refusal and trigger. + expect(saved).toEqual([expect.objectContaining({ status: "deferred_issue_execution", runId: null, + payload: expect.objectContaining({ executionWait: expect.objectContaining({ reason: refusal, trigger: "pause_lifted" }) }) })]); + await settle(f); + expect(deliveries(comment.id)).toHaveLength(0); + // The pause's own stop sweep (agent lifecycle "pausing") keeps the message + // it deferred; a terminate or cleanup stop would end it. + expect(await heartbeat.stopInvocationsForAgents([f.agentId], "Agent lifecycle stop requested", + { preservePauseDeferred: true })).toBe(true); + expect((await readWake(saved[0].id)).status).toBe("deferred_issue_execution"); + + await lift(f as never); + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: f.companyId, trigger: "pause_lifted", issueId: f.issueId })); + // Promotion skips the 30 s refusal backoff; it does not wait it out. + expect((await readWake(saved[0].id)).updatedAt.getTime()).toBeLessThanOrEqual(Date.now() - 29_000); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + // A repeated release event, or the reconciler backstop, never delivers it twice. + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: f.companyId, trigger: "reconciler_backstop" })); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + }, 180_000); + + // A company pause stops runs asynchronously. A message posted while the + // task's run is still stopping is saved behind that run, and delivered once + // after the run ends and the pause lifts. + it("saves a message posted while the paused company's run is still stopping, and delivers it once", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(companies).set({ status: "paused", pauseReason: "manual", pausedAt: new Date() }).where(eq(companies.id, f.companyId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "Posted while the run was stopping." }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toMatchObject({ status: 409 }); + const saved = (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) + .filter(wake => (wake.payload as { commentId?: string }).commentId === comment.id); + expect(saved).toEqual([expect.objectContaining({ status: "deferred_issue_execution", + payload: expect.objectContaining({ executionWait: expect.objectContaining({ reason: "company.inactive" }) }) })]); + // The run stops without reading the message; then the pause lifts. + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(companies).set({ status: "active", pauseReason: null, pausedAt: null }).where(eq(companies.id, f.companyId)); + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: f.companyId, trigger: "pause_lifted", issueId: f.issueId })); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + }, 180_000); + + // A pause stops the task's run with a clean, acknowledged cancellation. The + // release drain never promotes queued work behind an acknowledged stop, so + // the lifted pause issues the task's next explicit wake, which adopts the + // saved message once. The pause's own stop sweep keeps the message. + it("delivers a message saved behind a run the pause stopped, once, after the pause lifts", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issues).set({ executionRunId: f.sourceRunId }).where(eq(issues.id, f.issueId)); + await db.update(companies).set({ status: "paused", pauseReason: "manual", pausedAt: new Date() }).where(eq(companies.id, f.companyId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "Posted while the pause stopped the run." }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toMatchObject({ status: 409 }); + // The pause's stop is acknowledged: the run ends cancelled with a clean stop + // record, and the stop releases the task's execution lock. + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), errorCode: "cancelled", + resultJson: { executionCancellation: { state: "acknowledged" } } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId)); + expect(await heartbeat.stopInvocationsForAgents([f.agentId], "Agent lifecycle stop requested", + { preservePauseDeferred: true })).toBeTypeOf("boolean"); + expect(deliveries(comment.id)).toHaveLength(0); + await db.update(companies).set({ status: "active", pauseReason: null, pausedAt: null }).where(eq(companies.id, f.companyId)); + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: f.companyId, trigger: "pause_lifted", issueId: f.issueId })); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + }, 180_000); + + // Consumption is per comment: a retry that already received the paused + // message as task history consumed it, so lifting the pause never sends it + // again (RC-53 shape through the pause-lift path). + it("does not replay a paused message that a completed retry already received", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + await db.update(companies).set({ status: "paused", pauseReason: "manual", pausedAt: new Date() }).where(eq(companies.id, f.companyId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "Posted during the pause.", createdAt: new Date(Date.now() - 5_000), + updatedAt: new Date(Date.now() - 5_000) }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toMatchObject({ status: 409 }); + // A second message, posted later in the pause, that no run has read. + const [later] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "Also posted during the pause." }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: later.id, mutation: "comment" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: later.id, wakeCommentId: later.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toMatchObject({ status: 409 }); + // A retry of the source run read the whole thread, the first comment included. + 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: [{ id: comment.id }] } } }); + await db.update(companies).set({ status: "active", pauseReason: null, pausedAt: null }).where(eq(companies.id, f.companyId)); + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: f.companyId, trigger: "pause_lifted", issueId: f.issueId })); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(0); + const saved = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + sql`${agentWakeupRequests.payload}->>'commentId' = ${comment.id}`)); + expect(saved).toEqual([expect.objectContaining({ status: "cancelled", + error: "Queued messages were already delivered to a completed run" })]); + // The wake that adopts the task's saved messages delivers only the unread one. + expect(deliveries(later.id)).toHaveLength(1); + }, 180_000); + + // External chat input resumes only through its live inbound authorization, + // never through the generic pause-lift wake. + it("never replays a chat-inbound receipt through the pause-lift path", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "From an external chat." }).returning(); + const [receipt] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "automation", triggerDetail: "system", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: f.userId, idempotencyKey: `chat-inbound:${randomUUID()}`, + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment", + executionWait: { reason: "agent.not_invokable", trigger: "pause_lifted", message: "Agent is paused." }, + _paperclipWakeContext: { issueId: f.issueId, taskId: f.issueId, wakeCommentId: comment.id, wakeCommentIds: [comment.id], + wakeReason: "issue_commented", source: "chat:slack" } } }).returning(); + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: f.companyId, trigger: "pause_lifted", issueId: f.issueId })); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(0); + expect((await readWake(receipt.id)).runId).toBeNull(); + }, 180_000); + + // A task that never ran has no terminal run for the release drain to promote + // a saved message behind. Its message stays in the task, and stranded-work + // recovery starts the task once the pause lifts. + it("starts a never-run task commented during an agent pause once the pause lifts", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + const [task] = await db.insert(issues).values({ companyId: f.companyId, title: "Never ran", status: "todo", + assigneeAgentId: f.agentId, responsibleUserId: f.userId, issueNumber: 3 }).returning(); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: task.id, + authorUserId: f.userId, body: "Please start with the FOLLOWUP line." }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: task.id, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { issueId: task.id, taskId: task.id, commentId: comment.id, wakeCommentId: comment.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toMatchObject({ status: 409 }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) + .filter(wake => wake.status === "deferred_issue_execution" && (wake.payload as { issueId?: string }).issueId === task.id)).toEqual([]); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, f.agentId)); + 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("keeps the terminal outcomes closed and every temporary refusal mapped to a release trigger", () => { + expect([...WORK_INTENT_TERMINAL_OUTCOMES]).toEqual(["delivered", "consumed", "task_cancelled", "hard_stop"]); + for (const refusal of ["budget.blocked", "company.inactive", "agent.not_invokable", "issue_tree_hold_active", + "heartbeat.scheduling_suppressed"]) expect(TEMPORARY_REFUSAL_TRIGGERS[refusal]).toBeTruthy(); + }); + it.each([ { delivery: "task history", context: (commentId: string) => ({ executionContinuation: { messages: [{ id: commentId }] } }) }, { delivery: "wake input", context: (commentId: string) => ({ wakeCommentIds: [commentId] }) }, diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 0e76b9eceb3..3ab9a86ace6 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -116,7 +116,7 @@ describe("report-only ownership reconciliation", () => { processStartedAt: "2026-11-01T00:00:00.000Z" } } }); // A wake request ended terminally on a temporary refusal. await db.insert(agentWakeupRequests).values({ id: ids.wake, companyId, agentId, source: "assignment", - status: "skipped", error: "Company budget hold: admission paused" }); + status: "skipped", reason: "agent.not_invokable", requestedByActorType: "user", error: "Agent is paused" }); // A work intent refused for another run's unfinished accounting (the gate // Phase 3 removed). Saved, not dropped, so it is not an I4 violation. await db.insert(agentWakeupRequests).values({ id: ids.heldWake, companyId, agentId, source: "automation", @@ -310,7 +310,7 @@ describe("report-only ownership reconciliation", () => { source: "assignment", status: "failed", error: "adapter exited with code 1" }))); const refused = Array.from({ length: 5 }, () => randomUUID()); await db.insert(agentWakeupRequests).values(refused.map(id => ({ id, companyId, agentId, source: "assignment", - status: "skipped", error: "issue lock busy" }))); + status: "skipped", reason: "company.inactive", requestedByActorType: "user", error: "Company is not active" }))); const report = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); expect(report.violations.filter(violation => violation.rule === "I4_exactly_once_intent") .map(violation => violation.recordId).sort()).toEqual([...refused].sort()); @@ -318,6 +318,56 @@ describe("report-only ownership reconciliation", () => { expect(again.violations.filter(violation => violation.rule === "I4_exactly_once_intent")).toEqual([]); }); + // Section E: I4 reads the closed refusal code, not free text. Phase 2's + // false positive was a delivered retry whose error said "retrying once". + it("does not report a wake whose error text only mentions a retry", async () => { + const state = createOwnershipReportState(); + await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "assignment", status: "failed", + reason: "process_lost", requestedByActorType: "user", error: "Process lost; retrying once" }); + const report = await runOwnershipReconciliationPass(db, { state, record: false }); + expect(report.violations.filter(violation => violation.rule === "I4_exactly_once_intent")).toEqual([]); + }); + + it("reports a saved message whose pause lifted without promotion, and not one still paused", async () => { + const id = randomUUID(); + await db.insert(agentWakeupRequests).values({ id, companyId, agentId, source: "automation", status: "deferred_issue_execution", + requestedByActorType: "user", updatedAt: minutesAgo(10), + payload: { executionWait: { reason: "agent.not_invokable", trigger: "pause_lifted" } } }); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId)); + const paused = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(paused.violations.filter(violation => violation.kind === "wake_request_deferred")).toEqual([]); + expect(kindReport(paused, "wake_request_deferred")).toMatchObject({ inProgress: 1, verdicts: { alive: 1 } }); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, agentId)); + const lifted = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(lifted.violations.filter(violation => violation.kind === "wake_request_deferred")).toEqual([ + expect.objectContaining({ rule: "I4_exactly_once_intent", recordId: id, reason: "deferred_after_hold_cleared" })]); + }); + + it("does not report a hard-stop refusal or a message saved behind a live run", async () => { + await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "assignment", status: "skipped", + reason: "budget.blocked", requestedByActorType: "user", error: "Company is paused because its budget hard-stop was reached." }); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Busy", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ companyId, agentId, issueId, status: "running", invocationSource: "manual", + contextSnapshot: { issueId } }); + await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution", + requestedByActorType: "user", updatedAt: minutesAgo(10), + payload: { issueId, executionWait: { reason: "agent.not_invokable", trigger: "pause_lifted" } } }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.rule === "I4_exactly_once_intent")).toEqual([]); + }); + + it("keeps a backstop deadline for a lifted pause still inside its grace window", async () => { + const updatedAt = minutesAgo(1); + await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution", + requestedByActorType: "user", updatedAt, + payload: { executionWait: { reason: "agent.not_invokable", trigger: "pause_lifted" } } }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.kind === "wake_request_deferred")).toEqual([]); + // Without a release event the worker still checks again when grace ends. + expect(report.nextDueAt?.getTime()).toBe(updatedAt.getTime() + 5 * 60_000); + }); + it("attributes spool files to their company", async () => { await mkdir(orphanSpool, { recursive: true }); const runId = randomUUID(); diff --git a/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts b/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts index 408335ab410..2f1999565b7 100644 --- a/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts @@ -12,7 +12,8 @@ describe("lifecycle host stop scope", () => { const driver = createLifecycleDriver(db, {} as never); expect(await driver.runHost({ id: "target", companyId: "company", lifecycleState: "pausing", lifecycleVersion: 2, lifecycleRequiredPluginIds: [], lifecycleError: null, lifecycleOperation: null })).toBe("complete"); - expect(stop).toHaveBeenCalledWith(["target"], "Agent lifecycle stop requested"); + // A pause keeps the messages it deferred (work-intent outbox). + expect(stop).toHaveBeenCalledWith(["target"], "Agent lifecycle stop requested", { preservePauseDeferred: true }); expect(db.select).not.toHaveBeenCalled(); }); }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 100e56b3143..c3bbf5219a8 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -18,6 +18,7 @@ import { nativeManualAdmissionHoldsExecution } from "../../../services/native-ru import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import { readReleaseRun, type ReleaseRun } from "./release-run-read.js"; import type { Db } from "@paperclipai/db"; +import { deliveredQueuedCommentIds } from "../../../services/queued-comment-delivery.js"; import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared"; import { activityLog, @@ -318,31 +319,10 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru 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)); + // Promoting a saved wake whose comments a completed run already + // received would deliver them twice and can reopen finished work. + const delivered = await deliveredQueuedCommentIds(tx, { companyId, issueId, agentId: wakeAgentId, + commentIds: liveAndUnauthored }); const liveNonSelfCommentIds = liveAndUnauthored.filter(commentId => !delivered.has(commentId)); const containedSelfAuthoredComment = rows.some( (row) => targetsFinishingRunAgent && !row.deletedAt && row.createdByRunId === finishingRunId, diff --git a/server/src/services/agent-lifecycle-driver.ts b/server/src/services/agent-lifecycle-driver.ts index b92ea79238a..e02c9e9472a 100644 --- a/server/src/services/agent-lifecycle-driver.ts +++ b/server/src/services/agent-lifecycle-driver.ts @@ -1,4 +1,5 @@ import { and, eq, inArray } from "drizzle-orm"; +import { promoteDeferredWorkIntents } from "./work-intent-outbox.js"; import { agents, environments, plugins, pluginCompanySettings, type Db } from "@paperclipai/db"; import { instanceSettingsService } from "./instance-settings.js"; import { pluginHasAgentEnvironments, pluginServesAgentEnvironment } from "./plugin-environment-driver.js"; @@ -14,6 +15,7 @@ export function createLifecycleDriver(db: Db, manager: PluginWorkerManager): Lif const heartbeat = heartbeatService(db, { pluginWorkerManager: manager }); return { async onReady(agent) { + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: agent.companyId, trigger: "pause_lifted" })); await heartbeat.resumeQueuedRunsForAgent(agent.companyId, agent.id); }, async requiredPluginIds(agent) { @@ -60,7 +62,8 @@ export function createLifecycleDriver(db: Db, manager: PluginWorkerManager): Lif await db.select().from(agents).where(eq(agents.companyId, agent.companyId)))]; const reason = agent.lifecycleHolds?.includes("company_archived") ? "Cancelled because the company was archived" : "Agent lifecycle stop requested"; - if (!await heartbeat.stopInvocationsForAgents(ids, reason)) return "pending"; + if (!await heartbeat.stopInvocationsForAgents(ids, reason, + { preservePauseDeferred: agent.lifecycleState === "pausing" })) return "pending"; } return "complete"; }, diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index dde87350c09..135b62e96a2 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,4 +1,5 @@ import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; +import { promoteDeferredWorkIntents } from "./work-intent-outbox.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; @@ -415,6 +416,8 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica )); const observations = await computeObservedSpends(db, policies, now); if (policies.some((candidate, index) => hardStopReached(candidate, observations[index]))) return; + // Lifting a budget pause is a hold-release event for deferred work intents. + await promoteDeferredWorkIntents(db, { companyId: policy.companyId, trigger: "pause_lifted" }); if (policy.scopeType === "agent") { const { applyAgentBudgetHoldInTransaction } = await import("./agent-lifecycle.js"); await applyAgentBudgetHoldInTransaction(db, policy.companyId, policy.scopeId, false); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 768525e50aa..1a7265f6dce 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,4 +1,5 @@ import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; +import { promoteDeferredWorkIntents } from "./work-intent-outbox.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; @@ -347,6 +348,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { await notifyRunDispatch(tx, { companyId: id }); await notifyGitHubCompanyChange(tx, id); await notifyDeferredExecutionEligibility(tx, { companyId: id }); + if (companyPatch.status === "active") await promoteDeferredWorkIntents(tx, { companyId: id, trigger: "pause_lifted" }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyExternalObjectRefresh(tx, id); diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index 295921d2cbb..72590988116 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1,10 +1,11 @@ import { createReusableLeaseHandoff, preserveReusableLeaseHandoff, withoutReusableLeaseHandoff } from "./environment-lease-handoff.js"; +import { promoteDeferredWorkIntents } from "./work-intent-outbox.js"; import { clearRemoteWorkspaceLocationForReleasedLease } from "./remote-workspace-snapshot-retention.js"; import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; 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, notifyDeferredWakesForIssue } from "./deferred-execution-notifications.js"; +import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { withEnvironmentLeaseController } from "./environment-lease-controller.js"; import { randomUUID } from "node:crypto"; @@ -1774,7 +1775,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 }); + await promoteDeferredWorkIntents(tx, { companyId: previous.companyId, trigger: "lock_released", 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 }) @@ -1916,8 +1917,8 @@ export function environmentService(db: Db) { 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 }); + await promoteDeferredWorkIntents(tx, { companyId: run.companyId, trigger: "lock_released", + issueId: run.issueId && /^[0-9a-f-]{36}$/i.test(run.issueId) ? run.issueId : null }); } return tx.update(environmentLeases) .set({ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index a9fd5f8837d..ad1a131686d 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,4 +1,5 @@ import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js"; +import type { StopInvocationOptions } from "./heartbeat/run-control.js"; import { isContextCompactionRun } from "./context-compaction.js"; import { isUnstartedTaskQuestionAnswer } from "./unstarted-task-question-answer.js"; import { registerDeferredReceiptWork } from "./heartbeat/deferred-receipt-work.js"; @@ -4152,8 +4153,8 @@ export function heartbeatService( } } - async function stopInvocationsForAgents(agentIds: string[], reason: string) { - await cancelInvocationsForAgentsInternal(agentIds, reason); + async function stopInvocationsForAgents(agentIds: string[], reason: string, options: StopInvocationOptions = {}) { + await cancelInvocationsForAgentsInternal(agentIds, reason, options); const runs = await db.select().from(heartbeatRuns).where(inArray(heartbeatRuns.agentId, agentIds)); const leasesToRelease = await db.select({ runId: environmentLeases.heartbeatRunId }).from(environmentLeases) .innerJoin(heartbeatRuns, eq(heartbeatRuns.id, environmentLeases.heartbeatRunId)) diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 3fd4eef82a8..73dec05433f 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,4 +1,6 @@ import { CONTEXT_COMPACTION_WAKE_REASON, ContextCompactionError, isContextCompactionRun, readContextCompactionRequest } from "../context-compaction.js"; +import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; +import { deferredWorkIntentWait } from "../work-intent-outbox.js"; import { createHeartbeatTimerPolicy } from "./timer-policy.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js"; @@ -88,6 +90,7 @@ import { isNull, lte, ne, + notInArray, or, sql, } from "drizzle-orm"; @@ -1753,6 +1756,85 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende logger.warn({ err, queueId: wake.id }, "failed to resume interrupted comment queue"); }); } + // Work-intent outbox: a message refused by a pause waits for the pause to + // lift, not behind the run. Lifting the pause is the operator's explicit + // authorization to send what the pause refused, as Send is after a Stop: + // when the pause stopped the task's run, the release drain never promotes + // queued work behind that acknowledged stop, so the lifted pause issues the + // task's next explicit wake here. It reconsiders the exact saved receipt + // through normal admission (a refusal rewrites the receipt in place) and + // adopts it atomically, so it is delivered once. It runs before the + // stranded-queue pass, which touches each saved wake. + const pauseLifted = await db.select({ wake: agentWakeupRequests }) + .from(agentWakeupRequests) + .innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId), + sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`, + eq(issues.assigneeAgentId, agentWakeupRequests.agentId), isNull(issues.executionRunId), + notInArray(issues.status, ["done", "cancelled"]))) + .innerJoin(companies, and(eq(companies.id, agentWakeupRequests.companyId), eq(companies.status, "active"))) + .where(and(scope.companyId ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, + scope.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, + eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId), + // Only messages the pause path saved (never a chat-inbound receipt, + // which resumes through its own live authorization), and only pauses + // this query can see lifted. A refusal by any other gate rewrites the + // receipt's reason, so it never retries in a loop here. + sql`${agentWakeupRequests.payload}->'executionWait'->>'trigger' = 'pause_lifted'`, + sql`${agentWakeupRequests.payload}->'executionWait'->>'reason' in ('company.inactive', 'agent.not_invokable')`, + eq(agentWakeupRequests.requestedByActorType, "user"), + sql`coalesce(${agentWakeupRequests.idempotencyKey}, '') not like 'chat-inbound:%'`, + sql`exists (select 1 from agents pause_agent where pause_agent.id = ${agentWakeupRequests.agentId} + and pause_agent.status not in ('paused', 'terminated', 'pending_approval'))`, + sql`NOT EXISTS (SELECT 1 FROM heartbeat_runs live_issue_run + WHERE live_issue_run.company_id = ${issues.companyId} + AND (live_issue_run.context_snapshot->>'issueId' = ${issues.id}::text + OR live_issue_run.native_issue_id = ${issues.id}) + AND live_issue_run.status IN ('queued', 'running', 'scheduled_retry'))`, + cutoff ? gte(agentWakeupRequests.requestedAt, cutoff) : undefined)) + .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); + const pauseLiftedIssues = new Set(); + for (const { wake } of pauseLifted) { + const issueKey = `${wake.companyId}:${String(wake.payload?.issueId)}`; + if (pauseLiftedIssues.has(issueKey)) continue; + pauseLiftedIssues.add(issueKey); + // Consumption is per comment. The wake adopts every saved message of + // the task for this agent, so each receipt drops the comments a + // completed run already received; a receipt left empty is consumed. + const issueKeyId = String(wake.payload?.issueId); + const saved = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, wake.companyId), eq(agentWakeupRequests.agentId, wake.agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueKeyId}`)); + for (const receipt of saved) { + const commentIds = queuedCommentIdsFromWakePayload(receipt.payload); + const delivered = await deliveredQueuedCommentIds(db, { companyId: wake.companyId, issueId: issueKeyId, + agentId: wake.agentId, commentIds: commentIds.filter(isUuidLike) }); + if (delivered.size === 0) continue; + const remaining = commentIds.filter(commentId => !delivered.has(commentId)); + const now = new Date(); + await db.update(agentWakeupRequests).set(remaining.length === 0 + ? { status: "cancelled", finishedAt: now, updatedAt: now, error: "Queued messages were already delivered to a completed run" } + : { payload: withQueuedCommentIdsInWakePayload(parseObject(receipt.payload), remaining), updatedAt: now }) + .where(and(eq(agentWakeupRequests.id, receipt.id), eq(agentWakeupRequests.status, "deferred_issue_execution"))); + } + const [current] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.id, wake.id), + eq(agentWakeupRequests.status, "deferred_issue_execution"))); + if (!current) continue; + wake.payload = current.payload; + const context = parseObject(parseObject(wake.payload)[DEFERRED_WAKE_CONTEXT_KEY]); + const { executionWait: _wait, [DEFERRED_WAKE_CONTEXT_KEY]: _context, ...payload } = parseObject(wake.payload); + await enqueueWakeup(wake.agentId, { + source: wake.source as WakeupOptions["source"], triggerDetail: wake.triggerDetail as WakeupOptions["triggerDetail"], + reason: wake.reason ?? undefined, payload, contextSnapshot: context, + requestedByActorType: wake.requestedByActorType as WakeupOptions["requestedByActorType"], + requestedByActorId: wake.requestedByActorId ?? undefined, + }, wake.id).catch(err => { + if (err instanceof PersistedExecutionWaitRefusal) return; + scope.onError?.(err); + logger.warn({ err, requestId: wake.id }, "a message saved during a pause could not resume"); + }); + } + // A server restart or a message/cleanup race can leave a deferred wake // after its owner has released the issue lock. Revisit it through the same // release admission, so recovery holds and operator Stops still apply. @@ -2512,6 +2594,42 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }); }; + // Work-intent outbox (invariant 4): a user's message refused by a + // temporary hold is saved as a deferred intent that names the refusal and + // the hold-release event that promotes it, never written as a terminal + // skip. Promotion runs normal admission again, once. + // The release drain promotes a saved message behind the task's latest run, + // once that run is terminal; a run still stopping after a pause commits is + // that run. A task that never ran has none: its message stays in the task, + // and stranded-work recovery starts the task once the hold lifts. + const canSaveDeferredMessage = async () => { + if (executionWaitRequestId || caseRequest || durableRequest || !issueId || !isUuidLike(issueId) || + opts.requestedByActorType !== "user" || !wakeCommentId) return false; + const [latest] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, agent.companyId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)) + .limit(1); + return Boolean(latest); + }; + const saveDeferredMessage = async (refusal: string, message: string, details: Record = {}) => { + const savedIssueId = issueId!; + const condition = deferredWorkIntentWait(refusal, `${message} Your message is saved.`, details); + await db.transaction(async tx => { + await tx.execute(sql`select id from issues where id = ${savedIssueId} and company_id = ${agent.companyId} for update`); + await notifyDeferredExecutionWork(tx, { companyId: agent.companyId }); + await tx.insert(agentWakeupRequests).values({ + companyId: agent.companyId, agentId, source, triggerDetail, reason, + payload: withQueuedCommentIdsInWakePayload({ + ...payload, issueId: savedIssueId, [DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot, executionWait: condition, + }, [...new Set([...queuedCommentIdsFromRunContext(enrichedContextSnapshot), wakeCommentId!])]), + status: "deferred_issue_execution", + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + }); + }); + return new PersistedExecutionWaitRefusal(message, details); + }; + const schedulingSuppression = await getSchedulingSuppression(); // A task drain holds ADMISSION, not the request. The drain is a // process-local pre-restart hold, so a wake that arrives while it is @@ -2547,6 +2665,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (caseRequest) return null; const companyStatus = company?.status ?? "missing"; if (opts.requestedByActorType === "user") { + // A paused company is a temporary hold; an archived one is not. + if (companyStatus === "paused" && (await canSaveDeferredMessage())) + throw await saveDeferredMessage("company.inactive", "Company is not active", { status: companyStatus }); throw conflict("Company is not active", { status: companyStatus }); } await writeSkippedRequest("company.inactive", { @@ -2731,31 +2852,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }, ); if (budgetBlock) { - // A user's message refused by a temporary hold (accounting still settling) - // is saved as a deferred receipt, not skipped. It is promoted through - // normal admission once the run it waits behind ends or the hold clears. - if (!executionWaitRequestId && !caseRequest && !durableRequest && issueId && isUuidLike(issueId) && - opts.requestedByActorType === "user" && wakeCommentId && - !(await isBoardOwnedBudgetBlock(db, agent.companyId, budgetBlock))) { - const savedIssueId = issueId; - const condition = { reason: "budget.blocked", message: `${budgetBlock.reason} Your message is saved.`, - scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId }; - await db.transaction(async tx => { - await tx.execute(sql`select id from issues where id = ${savedIssueId} and company_id = ${agent.companyId} for update`); - await notifyDeferredExecutionWork(tx, { companyId: agent.companyId }); - await tx.insert(agentWakeupRequests).values({ - companyId: agent.companyId, agentId, source, triggerDetail, reason, - payload: withQueuedCommentIdsInWakePayload({ - ...payload, issueId: savedIssueId, [DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot, executionWait: condition, - }, [...new Set([...queuedCommentIdsFromRunContext(enrichedContextSnapshot), wakeCommentId])]), - status: "deferred_issue_execution", - requestedByActorType: opts.requestedByActorType ?? null, - requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, - }); - }); - throw new PersistedExecutionWaitRefusal(budgetBlock.reason, { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId }); - } + // A budget block that is not a reached hard stop is an explicit pause. + if ((await canSaveDeferredMessage()) && !(await isBoardOwnedBudgetBlock(db, agent.companyId, budgetBlock))) + throw await saveDeferredMessage("budget.blocked", budgetBlock.reason, + { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId }); const wait = await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId, }); @@ -2769,6 +2869,11 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const invokability = await getAgentInvokability(isAgentAwaitingSetup(agent) ? { ...agent, status: "idle" } : agent); if (!invokability.invokable) { if (caseRequest) return null; + // Only a pause defers through the release drain (wake-queue policy); a + // pending approval still fails a promoted wake, so it is not saved here. + if (invokability.reason === "paused" && (await canSaveDeferredMessage())) + throw await saveDeferredMessage("agent.not_invokable", invokability.message, + { status: agent.status, reason: invokability.reason }); let waitSaved = false; if (opts.requestedByActorType !== "user" || executionWaitRequestId) { const wait = await writeSkippedRequest("agent.not_invokable", { diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 7d070e67c31..fe5bb4940a6 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -177,7 +177,12 @@ async function withBudgetStopDelivery(db: Db, work: () => Promise): Promis } } +/** `preservePauseDeferred`: the stop comes from a pause, which keeps the + * messages it deferred for delivery once the pause lifts. */ +export type StopInvocationOptions = { preservePauseDeferred?: boolean }; + /** Cancellation shares executor ownership and delegates lifecycle and admission effects. */ + export interface HeartbeatRunControlDependencies extends Pick, "getRun" | "getAgent" | "getRunCleanupState" | "getRunExecutionWait">, Pick, @@ -1699,6 +1704,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont async function cancelPendingWakeupsForAgentsInternal( agentIds: string[], reason: string, + options: StopInvocationOptions = {}, ) { const uniqueAgentIds = [...new Set(agentIds)].filter( (agentId) => agentId.length > 0, @@ -1713,6 +1719,10 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont inArray(agentWakeupRequests.agentId, uniqueAgentIds), inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), isNull(agentWakeupRequests.runId), + // A pause is temporary: a message it deferred waits for the pause to + // lift (work-intent outbox), so the pause's own stop keeps it. + options.preservePauseDeferred ? sql`not (${agentWakeupRequests.status} = 'deferred_issue_execution' + and coalesce(${agentWakeupRequests.payload}->'executionWait'->>'trigger', '') = 'pause_lifted')` : undefined, )); if (!wakeups.length) return 0; for (const companyId of new Set(wakeups.map(row => row.companyId))) { @@ -1733,6 +1743,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont async function cancelInvocationsForAgentsInternal( agentIds: string[], reason: string, + options: StopInvocationOptions = {}, ) { const uniqueAgentIds = [...new Set(agentIds)].filter( (agentId) => agentId.length > 0, @@ -1744,6 +1755,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont const wakeupsCancelled = await cancelPendingWakeupsForAgentsInternal( uniqueAgentIds, reason, + options, ); return { agentIds: uniqueAgentIds, diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 0c6cd97cd1e..350c7684613 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,4 +1,5 @@ import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; +import { promoteDeferredWorkIntents } from "./work-intent-outbox.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; @@ -1209,8 +1210,7 @@ export function issueTreeControlService(db: Db) { } const [updated] = await db.transaction(async tx => { - await notifyIssueRecovery(tx, { companyId }); - await notifyDeferredExecutionEligibility(tx, { companyId }); + await promoteDeferredWorkIntents(tx, { companyId, trigger: "tree_hold_released" }); return tx .update(issueTreeHolds) .set({ diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index b5408f2d476..45e15f3feb7 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -17,11 +17,13 @@ import { DELIVERY_QUEUES, notifyDeliveryWork, subscribeDeliveryWork } from "../d import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { createOwnershipReportState, + DEFERRED_INTENT_NOT_PROMOTED, ownershipReportFingerprint, recheckSettlingKinds, runOwnershipReconciliationPass, type OwnershipReport, } from "../recovery/ownership-invariant-report.js"; +import { promoteDeferredWorkIntents } from "../work-intent-outbox.js"; import { createHash, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, not, notInArray, or, sql, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -791,6 +793,9 @@ const OWNERSHIP_REPORT_WAKE_QUEUES = [ DELIVERY_QUEUES.staleIssueLocks, DELIVERY_QUEUES.queuedRunDispatch, DELIVERY_QUEUES.accounting, + // Company and agent pause changes: the hold-release watch of the deferred + // work-intent backstop (section E). + DELIVERY_QUEUES.agentLifecycle, ] as const; /** Interval between report passes while in-progress records keep changing. */ @@ -816,6 +821,18 @@ export const OWNERSHIP_REPORT_INTERVAL_MS = 30_000; * deadline: an idle instance runs no SQL here. The pass runs in the shared * scheduler only, never on the path before `turn.start`. */ +/** Section E backstop: a deferred intent whose hold cleared without a + * release event is promoted through the one promotion function. Promotion + * re-runs normal admission, so a wrong report never bypasses a gate. */ +async function promoteMissedDeferredIntents(db: Db, report: OwnershipReport) { + const companies = new Set(report.violations.filter(violation => violation.reason === DEFERRED_INTENT_NOT_PROMOTED) + .flatMap(violation => violation.companyId ? [violation.companyId] : [])); + for (const companyId of companies) { + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId, trigger: "reconciler_backstop" })) + .catch(error => logger.warn({ err: error, companyId }, "deferred intent backstop promotion failed")); + } +} + export function registerOwnershipReconciliationWork( db: Db, coordinator: ReturnType, @@ -878,6 +895,7 @@ export function registerOwnershipReconciliationWork( lastPassAt = Date.now(); pendingWake = false; const report = await runOwnershipReconciliationPass(db, { state }); + await promoteMissedDeferredIntents(db, report); const fingerprint = ownershipReportFingerprint(report); const changing = report.inProgress > 0 && fingerprint !== lastFingerprint; // A settling kind's live owner can die without writing a signal, so diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 20d2eeb72b4..92efb555d18 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -1,4 +1,5 @@ import { notifyExternalConversationSettlement } from "../external-conversation-settlement-notifications.js"; +import { promoteDeferredWorkIntents } from "../work-intent-outbox.js"; import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; @@ -14,7 +15,6 @@ 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"; @@ -206,7 +206,7 @@ export async function projectCommittedRun(input: { .returning(); if (after) { await settleNativeTerminalWake(tx, after); - await notifyDeferredWakesForIssue(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); + await promoteDeferredWorkIntents(tx, { companyId: input.run.companyId, trigger: "lock_released", issueId: input.coordinator.issueId }); } return after ? { before, after } : null; }); diff --git a/server/src/services/queued-comment-delivery.ts b/server/src/services/queued-comment-delivery.ts new file mode 100644 index 00000000000..557a8b0b283 --- /dev/null +++ b/server/src/services/queued-comment-delivery.ts @@ -0,0 +1,36 @@ +import { and, eq, inArray, sql } from "drizzle-orm"; +import { heartbeatRuns, issueComments, type Db } from "@paperclipai/db"; + +/** + * Per-comment consumption (work-intent outbox, invariant 4). 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). 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. Every path that replays a saved message reads + * this before delivering it. + */ +export async function deliveredQueuedCommentIds(tx: Pick, input: { + companyId: string; issueId: string; agentId: string; commentIds: string[]; +}): Promise> { + if (input.commentIds.length === 0) return new Set(); + const { companyId, issueId } = input; + const rows = await tx.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), + inArray(issueComments.id, input.commentIds), + sql`exists (select 1 from ${heartbeatRuns} delivered_run + where delivered_run.company_id = ${companyId} and delivered_run.agent_id = ${input.agentId} + 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)))))`, + )); + return new Set(rows.map(row => row.id)); +} diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 74b293430de..7f68031765f 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -6,6 +6,7 @@ import { issues, type Db, } from "@paperclipai/db"; +import { TEMPORARY_REFUSAL_TRIGGERS } from "../work-intent-outbox.js"; import { logger } from "../../middleware/logger.js"; import { persistActivity } from "../activity-log.js"; import { proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js"; @@ -136,9 +137,6 @@ const MAX_RECORDS_PER_KIND = 2_000; const MAX_WRITE_ATTEMPTS = 5; /** First I4 scan after boot looks back this far. */ const INTENT_LOOKBACK_MS = 60 * 60_000; -/** Refusal texts that name a temporary condition (invariant 4), as a - * PostgreSQL case-insensitive regular expression. */ -const TEMPORARY_REFUSAL_PATTERN = "budget|paus|lock|busy|unavailable|retry|admission|hold|drain|restart|capacity|rate.?limit|timeout"; type Classification = { verdict: OwnerProof["verdict"]; reason: string; legacyHostFallback?: true; localExecutionsKnown?: true }; @@ -321,10 +319,56 @@ async function checkAdmissionIndependence(db: Db, state: OwnershipReportState, n return violations; } -/** A wake request that ended terminally on a refusal that names a temporary - * condition. Heuristic until section E gives wake requests closed reasons. The - * filter runs in SQL and the scan pages by (updated_at, id), so a busy window - * is read in full across passes and unrelated failures never hide a match. */ +/** A deferred intent still waiting this long after its hold can have + * cleared missed its release event (invariant 4). */ +const DEFERRED_PROMOTION_GRACE_MS = 5 * 60_000; + +/** Deferred pause refusals whose pause has lifted (company active, agent not paused) + * but that no release event promoted. The reconciliation worker promotes them + * through `promoteDeferredWorkIntents` as a backstop. */ +async function checkDeferredIntentsPromoted(db: Db, now: Date, limit: number): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> { + // Only messages the pause path saved, whose hold this check can read + // exactly. Each read is capped like every other check in this report. + const pauseRefusal = sql`${agentWakeupRequests.payload}->'executionWait'->>'trigger' = 'pause_lifted' + and ${agentWakeupRequests.payload}->'executionWait'->>'reason' in ('company.inactive', 'agent.not_invokable')`; + const holdCleared = sql`exists (select 1 from companies c where c.id = ${agentWakeupRequests.companyId} and c.status = 'active') + and exists (select 1 from agents a where a.id = ${agentWakeupRequests.agentId} and a.status not in ('paused', 'terminated', 'pending_approval')) + -- A message saved behind a live run or a held lock still has an owner. + and not exists (select 1 from issues i where i.id::text = ${agentWakeupRequests.payload}->>'issueId' + and i.company_id = ${agentWakeupRequests.companyId} and i.execution_run_id is not null) + and not exists (select 1 from heartbeat_runs live_run where live_run.company_id = ${agentWakeupRequests.companyId} + and (live_run.context_snapshot->>'issueId' = ${agentWakeupRequests.payload}->>'issueId' + or live_run.native_issue_id::text = ${agentWakeupRequests.payload}->>'issueId') + and live_run.status in ('queued', 'running', 'scheduled_retry'))`; + const graceStart = new Date(now.getTime() - DEFERRED_PROMOTION_GRACE_MS); + const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + issueId: sql`${agentWakeupRequests.payload}->>'issueId'` }).from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), pauseRefusal, + lt(agentWakeupRequests.updatedAt, graceStart), holdCleared)) + .orderBy(asc(agentWakeupRequests.id)).limit(Math.min(limit, MAX_RECORDS_PER_KIND)); + // A cleared hold inside its grace window is a saved deadline: the worker + // checks again when it ends. A hold still in force needs no deadline; its + // release (company or agent lifecycle change) wakes the worker. + const pending = await db.select({ at: agentWakeupRequests.updatedAt }).from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), pauseRefusal, + sql`${agentWakeupRequests.updatedAt} >= ${graceStart.toISOString()}::timestamptz`, holdCleared)) + .limit(MAX_RECORDS_PER_KIND); + const due = pending.length ? { at: new Date(Math.min(...pending.map(row => new Date(row.at).getTime()))) } : null; + return { + violations: rows.map(row => ({ rule: "I4_exactly_once_intent" as const, kind: "wake_request_deferred", recordId: row.id, + companyId: row.companyId, runId: null, issueId: row.issueId, reason: DEFERRED_INTENT_NOT_PROMOTED })), + nextDueAt: due?.at ? new Date(new Date(due.at).getTime() + DEFERRED_PROMOTION_GRACE_MS) : null, + }; +} + +export const DEFERRED_INTENT_NOT_PROMOTED = "deferred_after_hold_cleared"; + +/** Invariant 4 with section E's closed outcomes: an accepted intent (a user's + * request or a message) ended terminally on a refusal that + * `TEMPORARY_REFUSAL_TRIGGERS` lists as temporary. The refusal code is the + * wake's `reason`, never its free-text error, so a retry that delivered the + * intent is not a false positive. The scan pages by (updated_at, id), so a + * busy window is read in full across passes. */ async function checkExactlyOnceIntents(db: Db, state: OwnershipReportState, now: Date, limit: number): Promise { // The cursor keeps PostgreSQL's microsecond text form: a JS Date would round // it down and read the boundary row again. @@ -337,7 +381,13 @@ async function checkExactlyOnceIntents(db: Db, state: OwnershipReportState, now: runId: agentWakeupRequests.runId, updatedAt: sql`${agentWakeupRequests.updatedAt}::text` }) .from(agentWakeupRequests).where(and(inArray(agentWakeupRequests.status, ["skipped", "failed"]), sql`(${agentWakeupRequests.updatedAt}, ${agentWakeupRequests.id}) > (${position.at}::timestamptz, ${position.id}::uuid)`, - sql`(coalesce(${agentWakeupRequests.error}, '') || ' ' || coalesce(${agentWakeupRequests.reason}, '')) ~* ${TEMPORARY_REFUSAL_PATTERN}`)) + inArray(agentWakeupRequests.reason, Object.keys(TEMPORARY_REFUSAL_TRIGGERS)), + // A reached hard stop is a terminal outcome that shares the budget code. + sql`not (${agentWakeupRequests.reason} = 'budget.blocked' + and coalesce(${agentWakeupRequests.error}, '') ~* 'hard-stop|no reliable price')`, + or(eq(agentWakeupRequests.requestedByActorType, "user"), + sql`${agentWakeupRequests.payload} ? 'commentId'`, + sql`jsonb_typeof(${agentWakeupRequests.payload}#>'{_paperclipWakeContext,wakeCommentIds}') = 'array'`))) .orderBy(asc(agentWakeupRequests.updatedAt), asc(agentWakeupRequests.id)).limit(limit); for (const row of rows) { violations.push({ rule: "I4_exactly_once_intent", kind: "wake_request", recordId: row.id, companyId: row.companyId, @@ -471,9 +521,14 @@ export async function runOwnershipReconciliationPass(db: Db, options: { for (const check of [ () => checkAdmissionIndependence(db, state, now, limit), () => checkExactlyOnceIntents(db, state, now, limit), + async () => { + const result = await checkDeferredIntentsPromoted(db, now, limit); + if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt; + return result.violations; + }, async () => { const result = await checkVisibleNextAction(db, now, limit); - nextDueAt = result.nextDueAt; + if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt; return result.violations; }, ]) { diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index fa76be3bbe1..7d06d442344 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -847,6 +847,31 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ and(inArray(issueRecoveryActions.status, ["active", "escalated"]), inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES])), row => ({ kind: "operator", ownerType: row.ownerType }), [3, 7]), + // Work-intent outbox (section E). Each pending intent is owned by the event + // that releases it: a hold-release trigger, run dispatch, or its retry due + // time. The I4 check reports a deferred intent left after its hold cleared, + // and the reconciliation worker promotes it as a backstop. + { + id: "wake_request_deferred", title: "Wake request (deferred)", obligation: "ownership", storage: "database", + settle: null, invariantChecks: [4], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + issueId: sql`${agentWakeupRequests.payload}->>'issueId'`, + trigger: sql`${agentWakeupRequests.payload}->'executionWait'->>'trigger'`, + updatedAt: agentWakeupRequests.updatedAt }).from(agentWakeupRequests) + .where(eq(agentWakeupRequests.status, "deferred_issue_execution")).orderBy(asc(agentWakeupRequests.id)) + .limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: null, issueId: row.issueId, + owner: { kind: "operator" as const, ownerType: `hold_release:${row.trigger ?? "issue_execution"}` }, updatedAt: row.updatedAt, + })) }; + }, + }, + runKind("queued_run", "Queued run", "ownership", eq(heartbeatRuns.status, "queued"), + () => ({ kind: "operator", ownerType: "run_dispatch" }), [4]), + runKind("scheduled_retry_run", "Scheduled retry run", "ownership", eq(heartbeatRuns.status, "scheduled_retry"), + () => ({ kind: "operator", ownerType: "retry_schedule" }), [4]), ]; /** Resolve the current holder of fences that live on another row (the diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 3240d034f82..305ba884b36 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -1,7 +1,5 @@ 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 { promoteDeferredWorkIntents } from "./work-intent-outbox.js"; import { centsToUnits, centsToUsd, 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"; @@ -128,9 +126,7 @@ const nativeProviderWorkNeverStarted = sql`${heartbeatRuns.runtimeMode} = 'nativ * (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 }); + await promoteDeferredWorkIntents(tx, { companyId, trigger: "accounting_settled" }); } export const ACCOUNTING_CLOSURE_TIMEOUT_MS = REMOTE_RUNNER_RECONNECT_GRACE_MS + 3 * 60_000; diff --git a/server/src/services/work-intent-outbox.ts b/server/src/services/work-intent-outbox.ts new file mode 100644 index 00000000000..22c5b763635 --- /dev/null +++ b/server/src/services/work-intent-outbox.ts @@ -0,0 +1,80 @@ +import { and, eq, sql } from "drizzle-orm"; +import { agentWakeupRequests, type Db } from "@paperclipai/db"; +import { notifyDeferredExecutionEligibility, notifyDeferredWakesForIssue } from "./deferred-execution-notifications.js"; +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; + +/** + * Work-intent outbox (durable ownership plan, section E; invariant 4). + * + * An accepted work intent (assignment, comment, Retry, recovery) ends in one + * of a closed set of terminal outcomes. Every other refusal is temporary: the + * intent is saved as `deferred_issue_execution` with an `executionWait` that + * names the refusal and the hold-release event that clears it. Each such event + * calls `promoteDeferredWorkIntents`, and the ownership reconciler calls it as + * a backstop for a missed event. Promotion re-runs normal admission; it never + * bypasses a gate, and the existing compare-and-swap promotion plus the + * per-comment consumption check keep delivery exactly once. + */ + +/** The only terminal outcomes of an accepted work intent. */ +export const WORK_INTENT_TERMINAL_OUTCOMES = ["delivered", "consumed", "task_cancelled", "hard_stop"] as const; +export type WorkIntentTerminalOutcome = (typeof WORK_INTENT_TERMINAL_OUTCOMES)[number]; + +/** Hold-release events. Each promotes the intents its hold deferred. */ +export const WORK_INTENT_HOLD_TRIGGERS = [ + "accounting_settled", "lock_released", "pause_lifted", "tree_hold_released", "scheduling_resumed", "reconciler_backstop", +] as const; +export type WorkIntentHoldTrigger = (typeof WORK_INTENT_HOLD_TRIGGERS)[number]; + +/** Temporary refusals of an accepted intent, and the event that clears each. + * A refusal not listed here is either a terminal outcome (a reached hard + * stop, a cancelled or reassigned task) or a signal that was never accepted + * work (a disabled timer, no actionable work). */ +export const TEMPORARY_REFUSAL_TRIGGERS: Readonly> = { + "budget.blocked": "pause_lifted", + "company.inactive": "pause_lifted", + "agent.not_invokable": "pause_lifted", + "issue_tree_hold_active": "tree_hold_released", + "heartbeat.scheduling_suppressed": "scheduling_resumed", +}; + +export function temporaryRefusalTrigger(reason: string): WorkIntentHoldTrigger | null { + return TEMPORARY_REFUSAL_TRIGGERS[reason] ?? null; +} + +/** The `executionWait` saved on a deferred intent. */ +export function deferredWorkIntentWait(reason: string, message: string, details: Record = {}) { + return { ...details, reason, trigger: temporaryRefusalTrigger(reason), message }; +} + +/** Saved intents back off 30 s after each refusal (deferred-receipt-work.ts). + * A hold-release event makes the intents it may clear due at once. */ +const DEFERRED_RETRY_BACKOFF = sql`interval '30 seconds'`; + +/** + * The one promotion path for deferred work intents. Call it inside a + * transaction, normally the one that releases a hold (accounting settled, a lock released, a + * pause lifted). It wakes queued-run dispatch, the saved-receipt worker and + * stranded-work recovery for the company, and hints each saved intent of the + * issue by id. All notifications publish on commit. + */ +export async function promoteDeferredWorkIntents(tx: object, input: { + companyId: string; trigger: WorkIntentHoldTrigger; issueId?: string | null; +}) { + // Register the work signals before the write they announce. + await notifyRunDispatch(tx, { companyId: input.companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId: input.companyId }); + await notifyIssueRecovery(tx, { companyId: input.companyId }); + if (input.issueId) await notifyDeferredWakesForIssue(tx, { companyId: input.companyId, issueId: input.issueId }); + const db = tx as Pick; + await db.update(agentWakeupRequests).set({ + updatedAt: sql`least(${agentWakeupRequests.updatedAt}, now() - ${DEFERRED_RETRY_BACKOFF})`, + }).where(and( + eq(agentWakeupRequests.companyId, input.companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload}->'executionWait' is not null`, + sql`${agentWakeupRequests.updatedAt} > now() - ${DEFERRED_RETRY_BACKOFF}`, + input.issueId ? sql`${agentWakeupRequests.payload}->>'issueId' = ${input.issueId}` : undefined, + )); +}