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, + )); +}