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