From ac6c5191ee9047230fa53acd34e2260e96905d48 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:22:39 -0500 Subject: [PATCH] Drive saved execution receipts from committed work and deadlines Co-Authored-By: Paperclip --- .../2026-10-30-deferred-execution-sleep.md | 19 ++ .../deferred-execution-receipt-work.test.ts | 226 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 5 +- server/src/index.ts | 8 +- .../modules/run-dispatch/adapters/postgres.ts | 2 + .../wake-queue/adapters/postgres.test.ts | 9 +- .../modules/wake-queue/adapters/postgres.ts | 7 + server/src/routes/issues.ts | 2 + server/src/services/agent-configuration.ts | 2 + server/src/services/agent-lifecycle.ts | 2 + server/src/services/budgets.ts | 5 + .../chat-channels/failed-run-retry.ts | 5 + .../chat-channels/task-retry-input.ts | 2 + server/src/services/companies.ts | 3 + .../deferred-execution-notifications.ts | 34 +++ .../services/delivery-work-notifications.ts | 3 + .../execution-control-notifications.ts | 7 +- server/src/services/heartbeat.ts | 9 + .../heartbeat/deferred-receipt-work.ts | 100 ++++++++ server/src/services/heartbeat/queue.ts | 38 ++- server/src/services/heartbeat/recovery.ts | 4 +- server/src/services/heartbeat/run-control.ts | 19 +- server/src/services/instance-settings.ts | 2 + server/src/services/issues.ts | 9 + .../status-decision-committer.ts | 2 + server/src/services/task-retry.ts | 6 +- 26 files changed, 505 insertions(+), 25 deletions(-) create mode 100644 doc/plans/2026-10-30-deferred-execution-sleep.md create mode 100644 server/src/__tests__/deferred-execution-receipt-work.test.ts create mode 100644 server/src/services/deferred-execution-notifications.ts create mode 100644 server/src/services/heartbeat/deferred-receipt-work.ts diff --git a/doc/plans/2026-10-30-deferred-execution-sleep.md b/doc/plans/2026-10-30-deferred-execution-sleep.md new file mode 100644 index 00000000000..0478fe79300 --- /dev/null +++ b/doc/plans/2026-10-30-deferred-execution-sleep.md @@ -0,0 +1,19 @@ +# Deferred execution receipt ownership + +This slice removes the periodic `resumeQueuedRuns` sweep after queue dispatch migrated separately. It also delegates native status wake intents out of the periodic orphan reaper. Existing queue admission, authorization, task recovery, cancellation, resource and controller checks remain the mutation authority. + +| Duty | Durable pending truth | Committing producers | Deadline and exit | +| --- | --- | --- | --- | +| Task Retry | deferred task_retry wake with certified actor/generation | requestTaskRetry through execution control; execution, issue, agent, company and budget eligibility | immediate; current admission links canonical intent to one dispatch/run, supersession or terminal policy retires it; blocked input awaits its actual gate mutation | +| Saved comments and interaction responses | deferred wake carrying saved comment IDs or interaction mutation | wake admission insert; merge only genuinely new comment IDs or interaction identity; issue/lock/assignment/dependency and execution eligibility | immediate after terminal source; original release owner revalidates comment/card authority, process/cleanup evidence and issue claim | +| Execution wait | deferred wake with executionWait | original admission/restored source receipt; actual recovery action, execution and eligibility changes | original updatedAt + 30 seconds, then current owner revalidation; unchanged blocked receipt carries no recurring scan | +| Interrupt retry | deferred wake with queuedCommentInterrupt | authorized route persists interrupt marker; cancellation lifecycle and cleanup/eligibility transitions | original updatedAt + 30 seconds; cancelled legacy run naming exact queue bypasses initial grace as before; owner retains explicit interrupt and stop evidence | +| Cancelled legacy queue | cancelled legacy run result names exact still-deferred queue | run lifecycle terminal/control commits; original interrupt intent | immediate restart reconstruction; original release/admission owner settles or leaves saved input behind its actual blocker | +| Task chat receipt settlement | deferred task_retry with taskRetry.chatActionId | failed chat retry action settlement | immediate settlement-only pass folds linked action result; it cannot restage its own processed-but-deferred chat action. Startup/actual eligibility retains normal re-admission policy | +| Native status wake intent | system/native-status-committer wake queued/claimed without run | status committer transaction; actual execution/issue/agent/company/budget eligibility | immediate queued intent; claimedAt + 60 seconds for crash recovery; error updatedAt + 30 seconds. Existing deterministic dispatcher identity/claims prevent duplicate delivery; original intent becomes linked/coalesced/skipped while a separate blocked dispatch receipt owns saved admission | + +Three scoped owners reconstruct only their actual durable receipt predicates, with keyset pages of 100. Normal signals contain normalized company/request identity. Instance scheduler settings alone may request instance-wide reconstruction. Repeated source changes coalesce reconstruction for 30 seconds; already-known stored deadlines preempt that cooldown. Pending checks read memory only. Shutdown uses the existing coordinator and physical execution tracking; task drain reconciles producer transactions and releases the settled intent hold while retaining work for resume. Warm standby and disabled registration do not execute domain work. + +The generic run-dispatch notifier and broad issue review-state notifier deliberately do not fan out deferred eligibility. Queue bookkeeping may write them while processing its own saved input. Separate eligibility commits and settlement-only chat notifications avoid a self-restaging loop. + +C2 remains separate and mandatory: active output silence deadlines, remaining orphan/process/controller/remote authority and resource debts, legacy shutdown settlements, stranded agent status, and final aggregate chain removal. Retained workspace and migrated native finalization workers remain their existing owners. diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts new file mode 100644 index 00000000000..ba0f13bed58 --- /dev/null +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -0,0 +1,226 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, createDb, heartbeatRuns, issues } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyDeferredExecutionEligibility, notifyDeferredExecutionWork, notifyNativeStatusWakeWork, notifyTaskRetrySettlement } from "../services/deferred-execution-notifications.js"; +import { registerDeferredReceiptWork } from "../services/heartbeat/deferred-receipt-work.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { requestTaskRetry } from "../services/task-retry.js"; +import { issueService } from "../services/issues.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +vi.setConfig({ testTimeout: 90_000 }); +describe("deferred receipt durable dispatch", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let heartbeat: ReturnType; + let coordinator: ReturnType; + let drain = false; + const errors: unknown[] = []; + beforeAll(async () => { + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DEFERRED_RECEIPT_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("deferred-receipts-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + }); + beforeEach(async () => { + await db.execute(sql`truncate companies cascade`); drain = false; errors.length = 0; + heartbeat = heartbeatService(db); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true, + onError: error => errors.push(error) }); + }); + afterEach(async () => { + await coordinator.stop(); await heartbeat.drainActiveRunExecutions(); + vi.useRealTimers(); vi.restoreAllMocks(); expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function fixture() { + const companyId = randomUUID(), actorId = "deferred-receipt-user"; + await db.insert(authUsers).values({ id: actorId, name: "Receipt operator", email: `${actorId}@example.test`, + createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companies).values({ id: companyId, name: "Deferred dispatch", issuePrefix: `DE${companyId.slice(0, 6)}`, + defaultResponsibleUserId: actorId, requireBoardApprovalForNewAgents: false }); + await db.insert(companyMemberships).values({ companyId, principalId: actorId, principalType: "user", status: "active", membershipRole: "owner" }); + const [agent] = await db.insert(agents).values({ companyId, name: "Receipt process", adapterType: "process", status: "idle", + adapterConfig: { command: process.execPath, args: ["-e", "console.log('receipt-proof')"], cwd: process.cwd() }, + runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } } }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Resume saved task", status: "todo", assigneeAgentId: agent.id }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, issueId: issue.id, + status: "failed", finishedAt: new Date(), responsibleUserId: actorId, errorCode: "process_failed", + contextSnapshot: { issueId: issue.id }, resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }).returning(); + return { companyId, agent, issue, source, actorId }; + } + async function quiet() { + await vi.waitFor(async () => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 }); + const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"); + const selected = reads.mock.calls.length, executed = executes.mock.calls.length; + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed); + } + it("is quiet after empty reconstruction and terminal receipt history", async () => { + const f = await fixture(); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "completed", reason: "task_retry", payload: { issueId: f.issue.id } }); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; await quiet(); + }); + it("dispatches an ordinary committed task Retry into saved process output", async () => { + const f = await fixture(); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId, + observedRunId: f.source.id, idempotencyKey: randomUUID() }); + expect(receipt.accepted).toBe(true); + let runId: string | undefined; + await vi.waitFor(async () => { + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId!; + }, { timeout: 15_000 }); + await vi.waitFor(async () => { + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); + expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded"); + expect(run.stdoutExcerpt).toContain("receipt-proof"); + }, { timeout: 15_000 }); + await issueService(db).update(f.issue.id, { status: "done" }); + await heartbeat.drainActiveRunExecutions(); + const saved = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry")); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + if (saved.length) await vi.advanceTimersByTimeAsync(Math.max(0, Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())) - Date.now()) + 1); + await vi.waitFor(async () => expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"))).toHaveLength(0)); + const linked = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.reason, "task_retry"), sql`${agentWakeupRequests.runId} is not null`)); + expect(linked).toHaveLength(2); + expect(new Set(linked.map(row => row.runId))).toEqual(new Set([runId])); + expect(linked.map(row => row.idempotencyKey)).toContain(`task-retry-dispatch:${receipt.requestId}`); + expect(linked.find(row => row.id === receipt.requestId)?.status).toBe("coalesced"); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)).toHaveLength(1); + await quiet(); + }); + it.each(["native claim", "execution wait"])("reconstructs an exact fresh %s deadline without probing early", async kind => { + const f = await fixture(), updatedAt = new Date(), native = kind === "native claim"; + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "automation", status: native ? "claimed" : "deferred_issue_execution", claimedAt: updatedAt, updatedAt, + payload: native ? {} : { executionWait: { reason: "recovery" } }, + requestedByActorType: "system", requestedByActorId: native ? "native-status-committer" : null }).returning(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const check = vi.fn(async (scope: { companyId?: string; requestId?: string }) => { + await db.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, scope.requestId!)); + }); + await registerDeferredReceiptWork(db, coordinator, { resume: check, settleTaskChat: vi.fn(), dispatchNative: check, + schedulingSuppressed: async () => false }, () => !drain).ready; + const due = updatedAt.getTime() + (native ? 60_000 : 30_000); + expect(coordinator.nextWakeAt()).toBe(due); expect(check).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(due - Date.now() - 1); expect(check).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + await vi.waitFor(() => expect(check).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId: wake.id }))); + await quiet(); + }); + it("reconciles the first drained producer and waits for resume without domain SQL", async () => { + const f = await fixture(); drain = true; + const resume = vi.fn(async () => {}), dispatchNative = vi.fn(async () => {}); + await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), dispatchNative, + schedulingSuppressed: async () => false }, () => !drain).ready; + await db.transaction(async tx => { + await notifyDeferredExecutionWork(tx, { companyId: f.companyId }); + await tx.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, source: "on_demand", + status: "deferred_issue_execution", reason: "task_retry" }); + }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + await vi.advanceTimersByTimeAsync(1000); + expect(resume).not.toHaveBeenCalled(); expect(dispatchNative).not.toHaveBeenCalled(); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + drain = false; await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(1)); await quiet(); + }); + it.each([ + { name: "saved comment", payload: { _paperclipWakeContext: { wakeCommentIds: ["saved-comment"] } } }, + { name: "interaction", payload: { mutation: "interaction" } }, + { name: "execution wait", payload: { executionWait: { reason: "recovery" } } }, + { name: "interrupt", payload: { queuedCommentInterrupt: { actorId: "operator" } } }, + ])("activates only the committed company's $name receipt after an empty start", async ({ payload }) => { + const f = await fixture(), foreign = await fixture(); + const resume = vi.fn(async () => {}); + await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), dispatchNative: vi.fn(), + schedulingSuppressed: async () => false }, () => !drain).ready; + const [other] = await db.insert(agentWakeupRequests).values({ companyId: foreign.companyId, agentId: foreign.agent.id, + source: "on_demand", status: "deferred_issue_execution", payload }).returning(); + let requestId: string; + await db.transaction(async tx => { + await tx.transaction(async nested => { + await notifyDeferredExecutionWork(nested, { companyId: f.companyId }); + const [row] = await nested.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution", payload, updatedAt: new Date(Date.now() - 31_000) }).returning(); + requestId = row.id; + }); + expect(resume).not.toHaveBeenCalled(); + }); + await vi.waitFor(() => expect(resume).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId })), { timeout: 5_000 }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, other.id)))[0].status).toBe("deferred_issue_execution"); + await quiet(); + }); + it("reconstructs a cancelled legacy run's exact deferred queue even without comment payload", async () => { + const f = await fixture(); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution" }).returning(); + await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: { queuedCommentInterruptQueueId: wake.id } }) + .where(eq(heartbeatRuns.id, f.source.id)); + const resume = vi.fn(async () => {}); + await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), dispatchNative: vi.fn(), + schedulingSuppressed: async () => false }, () => !drain).ready; + expect(resume).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId: wake.id })); + await quiet(); + }); + it("settles a processed chat receipt without restaging it until an actual eligibility commit", async () => { + const f = await fixture(); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + const [application] = await db.insert(toolApplications).values({ companyId: f.companyId, name: "Receipt chat", type: "mcp_http" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId: f.companyId, applicationId: application.id, + uid: randomUUID(), name: "Receipt chat", transport: "rest_api" }).returning(); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId: f.companyId, connectionId: connection.id, + provider: "slack", publicId: randomUUID(), assignedAgentId: f.agent.id }).returning(); + const intentId = randomUUID(), actionId = randomUUID(), previous = new Date(Date.now() - 60_000); + await db.insert(agentWakeupRequests).values([ + { id: intentId, companyId: f.companyId, agentId: f.agent.id, source: "on_demand", reason: "task_retry", + status: "deferred_issue_execution", updatedAt: previous, payload: { issueId: f.issue.id, + taskRetry: { schema: "paperclip.task-retry.v1", generationId: intentId, sourceRunId: f.source.id, + actorId: f.actorId, strategy: "evaluate", chatActionId: actionId } } }, + { id: actionId, companyId: f.companyId, agentId: f.agent.id, source: "automation", status: "deferred_issue_execution", + payload: { issueId: f.issue.id, taskRetryRequestId: intentId } }, + ]); + await db.transaction(async tx => { + await notifyTaskRetrySettlement(tx, { companyId: f.companyId, requestId: intentId }); + await tx.insert(chatActions).values({ id: actionId, companyId: f.companyId, endpointId: endpoint.id, + providerActionId: randomUUID(), kind: "failed_run_retry", status: "processed", updatedAt: new Date(), + payload: { issueId: f.issue.id, taskRetryIntentId: intentId, failedRunId: f.source.id } }); + }); + await vi.waitFor(async () => { + const [intent] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, intentId)); + expect(intent.updatedAt.getTime()).toBeGreaterThan(previous.getTime()); + }, { timeout: 5_000 }); + expect((await db.select().from(chatActions).where(eq(chatActions.id, actionId)))[0].status).toBe("processed"); + await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId })); + await vi.waitFor(async () => expect((await db.select().from(chatActions).where(eq(chatActions.id, actionId)))[0].status) + .toBe("issued"), { timeout: 5_000 }); + await quiet(); + }); + it("outer rollback and nested savepoint rollback do not dispatch receipt intent", async () => { + const f = await fixture(); const dispatchNative = vi.fn(async () => {}); + await registerDeferredReceiptWork(db, coordinator, { resume: vi.fn(), settleTaskChat: vi.fn(), dispatchNative, + schedulingSuppressed: async () => false }, () => !drain).ready; + await expect(db.transaction(async tx => { + await notifyNativeStatusWakeWork(tx, { companyId: f.companyId }); + throw new Error("outer rollback"); + })).rejects.toThrow("outer rollback"); + await db.transaction(async tx => { + await expect(tx.transaction(async nested => { + await notifyNativeStatusWakeWork(nested, { companyId: f.companyId }); throw new Error("nested rollback"); + })).rejects.toThrow("nested rollback"); + }); + await quiet(); expect(dispatchNative).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index a26d94f50d4..55d261f91c0 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -135,6 +135,7 @@ const { reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })), reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), registerIssueStateRecoveryWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerDeferredReceiptWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -930,6 +931,8 @@ describe("startServer feedback export wiring", () => { await startServer(); await settleStartupHousekeeping(); expect(heartbeatServiceMock.registerIssueStateRecoveryWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(heartbeatServiceMock.registerDeferredReceiptWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(heartbeatServiceMock.resumeQueuedRuns).not.toHaveBeenCalled(); if (enabled) { const [coordinator, admitted] = heartbeatServiceMock.registerRunDispatchWork.mock.calls[0]; const app = await createAppMock.mock.results.at(-1)!.value; @@ -1341,7 +1344,7 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled(); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipNativeStatusWakeIntents: true, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled(); expect(retiredDetector).not.toHaveBeenCalled(); diff --git a/server/src/index.ts b/server/src/index.ts index 0ebe5c3332c..fffac2a7972 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1774,7 +1774,6 @@ async function startServerWithDatabaseTeardown( } } - await timeStartupRecovery("queued_runs", () => heartbeat.resumeQueuedRuns()); const recoveredGoalActions = await timeStartupRecovery("goal_actions", () => heartbeat.recoverPendingSessionGoalActions()); if ( recoveredGoalActions.enqueued > 0 || @@ -1806,6 +1805,8 @@ async function startServerWithDatabaseTeardown( if (config.heartbeatSchedulerEnabled) await heartbeat.registerIssueStateRecoveryWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; + if (config.heartbeatSchedulerEnabled) await heartbeat.registerDeferredReceiptWork(deliveryWork, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerRunDispatchWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork, @@ -2009,10 +2010,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. runPeriodicRecovery(() => heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }) - .then(async () => { - await heartbeat.resumeQueuedRuns(); - }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipNativeStatusWakeIntents: true, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }) .then(async () => { const scanned = await heartbeat.scanSilentActiveRuns(); if (scanned.created > 0 || scanned.escalated > 0) { diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index b05f94e1a4b..5acbfcf6884 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js"; import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restore-recovery-state.js"; import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js"; import { isPreDispatchExecutionWaitVerified } from "../../../services/pre-dispatch-execution-wait.js"; @@ -996,6 +997,7 @@ export function createPostgresRunDispatchAdapter( recordedCommentId && authors.get(recordedCommentId) === actorId ? [recordedCommentId] : []; })); if (commentIds.some(id => !recordedCommentIds.has(id))) return null; + await notifyDeferredExecutionWork(tx, { companyId: run.companyId, requestId: wake.id }); const [deferred] = await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", runId: null, claimedAt: null, finishedAt: null, error: null, payload: withQueuedCommentIdsInWakePayload({ ...payload, _paperclipWakeContext: context, diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 068a21f2420..09fbb1749ac 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.js"; import { instanceSettingsService } from "../../../services/instance-settings.js"; import { randomBytes, randomUUID } from "node:crypto"; @@ -37,7 +38,8 @@ import type { TransactionScope } from "../application/ports.js"; // a foreign-company row is invisible to a read, and a deferred-status // compare-and-set that affects no row leaves no other trace. The decision // branching itself is proven against plain facts in `domain/policy.test.ts`. -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_WAKE_QUEUE_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -57,8 +59,8 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { }; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-wake-queue-postgres-adapter-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-wake-queue-postgres-adapter-"); + db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 }); }, 20_000); afterEach(async () => { @@ -78,6 +80,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 1df4420615f..e7a5f12f3f2 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js"; import { notifyRunDispatch } from "../../../services/run-dispatch-notifications.js"; import { notifyExternalConversationSettlement } from "../../../services/external-conversation-settlement-notifications.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; @@ -1010,6 +1011,11 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { )).for("update"); if (!prior) throw new Error("wake-queue: the deferred wake to merge into was not found for this company"); const priorPayload = parseObject(prior.payload), issueId = readNonEmptyString(priorPayload.issueId); + const previousComments = new Set(queuedCommentIdsFromWakePayload(priorPayload)); + const newComments = queuedCommentIdsFromWakePayload(input.mergedPayload).some(id => !previousComments.has(id)); + const newInteraction = readNonEmptyString(input.mergedPayload.interactionId); + if (newComments || newInteraction && newInteraction !== readNonEmptyString(priorPayload.interactionId)) + await notifyDeferredExecutionWork(tx, { companyId: input.companyId, requestId: prior.id }); const priorCommentId = readNonEmptyString(priorPayload.commentId), priorActor = readNonEmptyString(prior.requestedByActorId); const [priorComment] = prior.requestedByActorType === "user" && priorActor && issueId && isUuidLike(issueId) && priorCommentId && isUuidLike(priorCommentId) ? await tx.select({ id: issueComments.id, body: issueComments.body }).from(issueComments).where(and( @@ -1081,6 +1087,7 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { async insertNewDeferredWake(scope, input) { const tx = requireAdmissionTx(scope, input.companyId); + await notifyDeferredExecutionWork(tx, { companyId: input.companyId }); await tx.insert(agentWakeupRequests).values({ ...input.durableReceipt, companyId: input.companyId, diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index cf86556ce5e..efdf363975d 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionWork } from "../services/deferred-execution-notifications.js"; import { notifyNativeRecoveryNotices } from "../services/native-recovery-notice-notifications.js"; import { notifyNativeStatusReassessment } from "../services/native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../services/unsafe-workspace-export-notifications.js"; @@ -16674,6 +16675,7 @@ export function issueRoutes( } // The click is durable fresh user intent, including when the message // predates a failed run's stop. Keep its content and original attribution. + await notifyDeferredExecutionWork(tx, { companyId: issue.companyId, requestId: locked.wake.id }); await tx.update(agentWakeupRequests).set({ payload: { ...readObject(locked.wake.payload), queuedCommentInterrupt: { actorId: actor.actorId, requestedAt: new Date().toISOString(), diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index dc10efaa082..2d30da27cb7 100644 --- a/server/src/services/agent-configuration.ts +++ b/server/src/services/agent-configuration.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifySkillVersionPinChange } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -85,6 +86,7 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura ); } await notifyRunDispatch(db, { companyId: existing.companyId }); + await notifyDeferredExecutionEligibility(db, { companyId: existing.companyId }); if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) { await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); } diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 59357fcc5de..8bb46882256 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { isAgentStatusInvokable } from "@paperclipai/shared"; @@ -41,6 +42,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li async beforeStatusChange(tx, companyId, agentId, before, after) { if (before !== after) { await notifyRunDispatch(tx, { companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); } if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 186253b19bb..985197be3cd 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; @@ -289,6 +290,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (policy.scopeType === "project") { await notifyRunDispatch(db, { companyId: policy.companyId }); + await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await db .update(projects) .set({ @@ -303,6 +305,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; await notifyRunDispatch(db, { companyId: policy.companyId }); + await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); @@ -350,6 +353,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (policy.scopeType === "project") { await notifyRunDispatch(db, { companyId: policy.companyId }); + await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await db .update(projects) .set({ @@ -365,6 +369,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; await notifyRunDispatch(db, { companyId: policy.companyId }); + await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index c324a6f3c34..f8b8c3ebeab 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionWork, notifyTaskRetrySettlement } from "../deferred-execution-notifications.js"; import { notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { createFailedRetryNotices } from "./failed-retry-notices.js"; @@ -1424,6 +1425,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti // Source authorization above checked every external receipt and its owner. // Reset only this certified unstarted batch under the task's admission lock. const skippedIds = source.taskRetryInput?.skippedActionIds ?? []; + await notifyDeferredExecutionWork(tx, { companyId: input.companyId }); await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", runId: null, finishedAt: null, payload: sql`${agentWakeupRequests.payload} - 'coalescedIntoWakeupRequestId'`, updatedAt: new Date(), }).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, input.agentId), @@ -1784,6 +1786,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti const attemptCount = Number(candidate.result?.attemptCount ?? 0) + 1; const settle = async (status: string, result: Record) => { const [current] = await db.transaction(async (tx) => { + if (typeof claimed.payload.taskRetryIntentId === "string") await notifyTaskRetrySettlement(tx, { + companyId: claimed.companyId, requestId: claimed.payload.taskRetryIntentId, + }); await notifyChatActionWork(tx, "failed_run_retry"); if (status === "processed") await notifyNativeResponsePresentation(tx, { companyId: claimed.companyId, issueId: String(claimed.payload.issueId), diff --git a/server/src/services/chat-channels/task-retry-input.ts b/server/src/services/chat-channels/task-retry-input.ts index 71743f42ad4..8257a8ec079 100644 --- a/server/src/services/chat-channels/task-retry-input.ts +++ b/server/src/services/chat-channels/task-retry-input.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js"; import { z } from "zod"; import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; import { agentWakeupRequests, chatActions, chatConversations, chatEndpoints, heartbeatRuns, issues, type Db } from "@paperclipai/db"; @@ -91,6 +92,7 @@ export async function releaseUnadmittedTaskRetryInput(db: Db, input: { companyId )).for("update"); if (receipt?.runId) return; await notifyChatInboundWakeNotices(tx, { companyId: input.companyId, issueId: issue.id }); + await notifyDeferredExecutionWork(tx, { companyId: input.companyId }); await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", finishedAt: null, payload: sql`${agentWakeupRequests.payload} - 'coalescedIntoWakeupRequestId'`, updatedAt: new Date(), }).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.status, "coalesced"), diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index a9a8f59998c..f0522ee4ddb 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; @@ -334,6 +335,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyRunDispatch(tx, { companyId: id }); + await notifyDeferredExecutionEligibility(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); @@ -491,6 +493,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyRunDispatch(tx, { companyId: id }); + await notifyDeferredExecutionEligibility(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/deferred-execution-notifications.ts b/server/src/services/deferred-execution-notifications.ts new file mode 100644 index 00000000000..b82cea725a5 --- /dev/null +++ b/server/src/services/deferred-execution-notifications.ts @@ -0,0 +1,34 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type DeferredExecutionScope = { companyId: string; requestId?: string }; +export type DeferredReceiptRecoveryOptions = { + companyId?: string; + requestId?: string; + onError?: (error: unknown) => void; +}; + +function hint(scope: DeferredExecutionScope) { + return JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }); +} + +export async function notifyDeferredExecutionWork(tx: object, scope: DeferredExecutionScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.deferredExecutionReceipts, hint(scope)); +} + +export async function notifyNativeStatusWakeWork(tx: object, scope: DeferredExecutionScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusWakeIntents, hint(scope)); +} + +export async function notifyTaskRetrySettlement(tx: object, scope: DeferredExecutionScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.taskRetryReceiptSettlement, hint(scope)); +} + +export async function notifyDeferredExecutionEligibility(tx: object, scope?: { companyId: string }) { + if (scope) { + await notifyDeferredExecutionWork(tx, scope); + await notifyNativeStatusWakeWork(tx, scope); + } else { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.deferredExecutionReceipts, "*"); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusWakeIntents, "*"); + } +} diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 8a9fdbe5686..79f0f5d1d3d 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,9 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + deferredExecutionReceipts: "deferred-execution-receipts", + taskRetryReceiptSettlement: "task-retry-receipt-settlement", + nativeStatusWakeIntents: "native-status-wake-intents", orphanBlockers: "orphan-blockers", activeRecoveryActions: "active-recovery-actions", strandedAssignments: "stranded-assignments", diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 1839176769f..7d9530f5197 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js"; @@ -5,7 +6,10 @@ import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./deliv export type ExecutionWorkScope = { companyId: string; agentId?: string; runId?: string }; export async function notifyExecutionRunWork(transaction: object, scope?: ExecutionWorkScope) { - if (scope) await notifyRunDispatch(transaction, scope); + if (scope) { + await notifyRunDispatch(transaction, scope); + await notifyDeferredExecutionEligibility(transaction, scope); + } for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { await notifyDeliveryWork(transaction, queue); @@ -20,6 +24,7 @@ export function writeExecutionControl( ): Promise { return db.transaction(async transaction => { await notifyRunDispatch(transaction, scope); + if (scope) await notifyDeferredExecutionEligibility(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors); for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); return write(transaction as unknown as Db); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 068532357ea..63b836a566f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,5 @@ +import { registerDeferredReceiptWork } from "./heartbeat/deferred-receipt-work.js"; +import { reconcileTaskChatRetryReceipts } from "./task-retry.js"; import { registerIssueRecoveryWork } from "./recovery/work.js"; import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js"; import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js"; @@ -4243,6 +4245,13 @@ export function heartbeatService( computeTaskDrain, applyTaskDrain, + registerDeferredReceiptWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerDeferredReceiptWork(db, coordinator, { + resume: scope => heartbeatQueue.resumeDeferredExecutionReceipts(scope), + settleTaskChat: scope => reconcileTaskChatRetryReceipts(db, { ...scope, settleOnly: true }), + dispatchNative: dispatchPendingNativeStatusWakeups, + schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, + }, () => canRun() && readTaskDrain(new Date()) === null), registerIssueStateRecoveryWork: (coordinator: Parameters[1], canRun: () => boolean) => registerIssueRecoveryWork(db, coordinator, { stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true, skipRelatedRecovery: true }), diff --git a/server/src/services/heartbeat/deferred-receipt-work.ts b/server/src/services/heartbeat/deferred-receipt-work.ts new file mode 100644 index 00000000000..fddbce3be8d --- /dev/null +++ b/server/src/services/heartbeat/deferred-receipt-work.ts @@ -0,0 +1,100 @@ +import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; +import { agentWakeupRequests, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; + +type Entry = { id: string; companyId: string; dueAt: number }; +type NativeIntentOptions = DeferredExecutionScope & { + onDeferredDeadline(at: number): void; + onError(error: unknown): void; +}; + +export function registerDeferredReceiptWork( + db: Db, + coordinator: ReturnType, + owner: { + resume(scope: DeferredReceiptRecoveryOptions): Promise; + settleTaskChat(scope: DeferredExecutionScope): Promise; + dispatchNative(scope: NativeIntentOptions): Promise; + schedulingSuppressed(): Promise; + }, + canRun: () => boolean, +) { + function register(kind: "deferred" | "task_chat" | "native") { + const cursors = new Map(); + let enabled = false; + const queue = kind === "deferred" ? DELIVERY_QUEUES.deferredExecutionReceipts + : kind === "task_chat" ? DELIVERY_QUEUES.taskRetryReceiptSettlement : DELIVERY_QUEUES.nativeStatusWakeIntents; + return registerScopedDeadlineWork(coordinator, { + queue, + reconstructionIntervalMs: 30_000, + beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); }, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.requestId || entry.id === scope.requestId), + load: async scope => { + const key = scope ? JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }) : "*"; + if (!enabled) { cursors.delete(key); return []; } + const cursor = cursors.get(key); + const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + status: agentWakeupRequests.status, updatedAt: agentWakeupRequests.updatedAt, + claimedAt: agentWakeupRequests.claimedAt, error: agentWakeupRequests.error, + delayed: sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null + or ${agentWakeupRequests.payload}->'executionWait' is not null`, + interruptedLegacy: sql`exists (select 1 from heartbeat_runs r + where r.company_id = ${agentWakeupRequests.companyId} and r.agent_id = ${agentWakeupRequests.agentId} + and r.runtime_mode = 'legacy' and r.status = 'cancelled' + and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`, + }).from(agentWakeupRequests).where(and( + scope ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, + scope?.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, + cursor ? gt(agentWakeupRequests.id, cursor) : undefined, + kind === "native" ? and( + eq(agentWakeupRequests.requestedByActorType, "system"), + eq(agentWakeupRequests.requestedByActorId, "native-status-committer"), + inArray(agentWakeupRequests.status, ["queued", "claimed"]), isNull(agentWakeupRequests.runId), + ) : and(eq(agentWakeupRequests.status, "deferred_issue_execution"), + kind === "task_chat" ? and(eq(agentWakeupRequests.reason, "task_retry"), + sql`${agentWakeupRequests.payload}#>>'{taskRetry,chatActionId}' is not null`) : or( + eq(agentWakeupRequests.reason, "task_retry"), + sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`, + sql`${agentWakeupRequests.payload}->'executionWait' is not null`, + sql`jsonb_typeof(${agentWakeupRequests.payload}#>'{_paperclipWakeContext,wakeCommentIds}') = 'array' + and ${agentWakeupRequests.payload}#>'{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`, + sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`, + sql`exists (select 1 from heartbeat_runs r where r.company_id = ${agentWakeupRequests.companyId} + and r.runtime_mode = 'legacy' and r.status = 'cancelled' + and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`, + )), + )).orderBy(asc(agentWakeupRequests.id)).limit(100); + const complete = rows.length < 100; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + return { rows: rows.map(row => ({ id: row.id, companyId: row.companyId, + dueAt: kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000 + : kind === "native" && row.error ? row.updatedAt.getTime() + 30_000 + : kind === "deferred" && row.delayed && !row.interruptedLegacy ? row.updatedAt.getTime() + 30_000 + : Date.now(), + })), complete }; + }, + check: async entry => { + if (!enabled) return null; + const errors: unknown[] = []; + let next: number | null = null; + const scope = { companyId: entry.companyId, requestId: entry.id }; + if (kind === "native") await owner.dispatchNative({ ...scope, + onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); }, + onError: error => { errors.push(error); }, + }); + else if (kind === "task_chat") await owner.settleTaskChat(scope); + else await owner.resume({ ...scope, onError: error => { errors.push(error); } }); + if (errors.length) throw new AggregateError(errors, `Deferred receipt ${entry.id} could not settle`); + return next === null ? null : { ...entry, dueAt: next }; + }, + }, canRun); + } + const workers = [register("deferred"), register("task_chat"), register("native")]; + return { ready: Promise.all(workers.map(worker => worker.ready)), + wake() { for (const worker of workers) worker.invalidate(); }, + async stop() { await Promise.all(workers.map(worker => worker.stop())); } }; +} diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 64ca0b84ffd..ef3ecfeb7ac 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,5 @@ +import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js"; +import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { notifyRunDispatch } from "../run-dispatch-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; @@ -222,7 +224,7 @@ export interface HeartbeatQueueDependencies extends Pick void; cancelRunInternal: (runId: string, reason?: string, options?: { errorCode?: string }) => Promise; cancelActiveForAgentInternal: (agentId: string, reason: string) => Promise; - resumeExecutionWaitComments: () => Promise; + resumeExecutionWaitComments: (scope?: DeferredReceiptRecoveryOptions) => Promise; resumeQueuedCommentInterrupt: (companyId: string, queueId: string) => Promise; resumeSavedLegacyComments: (companyId: string, queueId: string) => Promise; formatIssueIdentifierLink: (identifier: string | null, fallback: string) => string; @@ -1851,13 +1853,15 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende return cancelled; } - async function resumeQueuedRuns() { + async function resumeDeferredExecutionReceipts(scope: DeferredReceiptRecoveryOptions = {}) { if ((await getSchedulingSuppression()).suppressed) return; - await resumeExecutionWaitComments(); + await resumeExecutionWaitComments(scope); const cutoff = await getWorktreeExecutionCutoff(); const pendingInterrupts = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId }) .from(agentWakeupRequests).innerJoin(companies, eq(companies.id, agentWakeupRequests.companyId)) - .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), + .where(and(scope.companyId ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, + scope.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, + eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(companies.status, "active"), sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`, lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)), @@ -1868,6 +1872,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"), )); await resumeQueuedCommentInterrupt(wake.companyId, wake.id).catch(err => { + scope.onError?.(err); logger.warn({ err, queueId: wake.id }, "failed to resume interrupted comment queue"); }); } @@ -1881,7 +1886,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende or(eq(issues.assigneeAgentId, agentWakeupRequests.agentId), sql`${agentWakeupRequests.payload} #>> '{_paperclipWakeContext,wakeReason}' = 'issue_comment_mentioned'`))) .innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active"))) - .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), + .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(issues.executionRunId), or(and( sql`jsonb_typeof(${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}') = 'array'`, @@ -1920,7 +1927,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (latest.runtimeMode !== "legacy" && latest.runtimeMode !== "native") continue; if (wake.idempotencyKey?.startsWith("chat-inbound:")) { await resumeDeferredChatWake(db, { companyId: wake.companyId, agentId: wake.agentId, - issueId: String(wake.payload?.issueId), requestId: wake.id }).catch(() => { + issueId: String(wake.payload?.issueId), requestId: wake.id }).catch(err => { + scope.onError?.(err); logger.warn({ requestId: wake.id }, "saved chat source could not resume after recovery"); }); continue; @@ -1932,6 +1940,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende latest.errorCode === "execution_reconciliation_required"; if ((latest.status !== "cancelled" || cancelledAdmission) && await getExecutionBlocker(db, wake.companyId, String(wake.payload?.issueId))) { await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => { + scope.onError?.(err); logger.warn({ err, queueId: wake.id }, "failed to deliver saved legacy comment after recovery stopped"); }); } @@ -1939,6 +1948,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true, resumeStrandedQueue: true }).then(() => { if (latest.runtimeMode === "native") releasedNativeSources.add(releaseSource); }).catch(err => { + scope.onError?.(err); logger.warn({ err, queueId: wake.id }, "failed to promote stranded comments"); }); } @@ -1956,6 +1966,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende )) .innerJoin(companies, eq(companies.id, heartbeatRuns.companyId)) .where(and( + scope.companyId ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, + scope.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(heartbeatRuns.status, "cancelled"), eq(heartbeatRuns.runtimeMode, "legacy"), @@ -1964,10 +1976,15 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende )); for (const run of interruptedQueues) { await releaseIssueExecutionAndPromote(run, { suppressImmediateRecovery: true }).catch((err) => { + scope.onError?.(err); logger.error({ err, runId: run.id }, "failed to retry interrupted comment queue"); }); } + } + + async function resumeQueuedRuns() { + await resumeDeferredExecutionReceipts(); await resumeQueuedRunsAfterDrain(); } @@ -3413,6 +3430,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (durableRequest || wakeCommentId || hasInteractionContinuationWakeContext(enrichedContextSnapshot) || readNonEmptyString(enrichedContextSnapshot.nativeStatusWakeIntentId)) { + await notifyDeferredExecutionWork(tx, { companyId: agent.companyId }); await tx.insert(agentWakeupRequests).values({ ...durableReceiptFields, companyId: agent.companyId, agentId, source, triggerDetail, reason, @@ -4030,6 +4048,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende })); } if (followUp) { + if (!followUp.consumed) await notifyDeferredExecutionWork(tx, { companyId: agent.companyId }); const assignmentId = followUp.assignmentId; const pending = followUp.consumed ? undefined : pendingFollowUps.find(wake => wake.payload?.[followUpPayloadKey] === assignmentId && @@ -4768,6 +4787,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende companyId?: string; limit?: number; staleClaimMs?: number; + requestId?: string; + onDeferredDeadline?: (at: number) => void; + onError?: (error: unknown) => void; } = {}, ) { const now = new Date(); @@ -4780,6 +4802,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende input.companyId ? eq(agentWakeupRequests.companyId, input.companyId) : undefined, + input.requestId ? eq(agentWakeupRequests.id, input.requestId) : undefined, eq(agentWakeupRequests.requestedByActorType, "system"), eq(agentWakeupRequests.requestedByActorId, "native-status-committer"), inArray(agentWakeupRequests.status, ["queued", "claimed"]), @@ -4856,6 +4879,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende candidate.claimedAt && now.getTime() - candidate.claimedAt.getTime() < staleClaimMs ) { + input.onDeferredDeadline?.(candidate.claimedAt.getTime() + staleClaimMs); deferred += 1; continue; } @@ -5030,6 +5054,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (wakeRun) dispatched += 1; else deferred += 1; } catch (error) { + input.onError?.(error); await db .update(agentWakeupRequests) .set({ @@ -5073,6 +5098,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende withChatControlRecoveryGate, trackWakeup, resumeQueuedRuns, + resumeDeferredExecutionReceipts, resumeQueuedRunsAfterDrain, parseHeartbeatPolicy, nextTimerHeartbeatAt, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 3b180d2679d..e506f0f7e62 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1979,7 +1979,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean; skipNativeEnvironmentSettlement?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipNativeStatusWakeIntents?: boolean; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean; skipNativeEnvironmentSettlement?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -2023,7 +2023,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }); if (!opts?.skipRetainedNativeSessionCleanup) scheduleRetainedNativeSessionCleanup(); await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments); - await dispatchPendingNativeStatusWakeups().catch((error) => { + if (!opts?.skipNativeStatusWakeIntents) await dispatchPendingNativeStatusWakeups().catch((error) => { logger.warn( { err: error }, "failed to dispatch persisted native status wake intents before orphan reaping", diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 820f1e50c61..691309a1d76 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { observeStoppedWorkspaceRecovery } from "../workspace-recovery.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; @@ -777,21 +778,25 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont idempotencyKey: `task-retry-dispatch:${wake.id}` }, wake.id); } - async function resumeExecutionWaitComments() { + async function resumeExecutionWaitComments(scope: DeferredReceiptRecoveryOptions = {}) { if ((await getSchedulingSuppression()).suppressed) return; - await reconcileTaskChatRetryReceipts(db); + await reconcileTaskChatRetryReceipts(db, scope); const retries = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId }) .from(agentWakeupRequests).innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId), sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`, eq(issues.assigneeAgentId, agentWakeupRequests.agentId))) .innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active"))) - .where(and(eq(agentWakeupRequests.reason, "task_retry"), eq(agentWakeupRequests.status, "deferred_issue_execution"), + .where(and(scope.companyId ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, + scope.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, + eq(agentWakeupRequests.reason, "task_retry"), eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload}->'taskRetry'->>'chatActionId' is null`, notInArray(issues.status, ["done", "cancelled"]))) .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); for (const retry of retries) { await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(eq(agentWakeupRequests.id, retry.id)); - await resumeTaskRetry(retry.companyId, retry.id).catch(err => logger.warn({ err, requestId: retry.id }, "Task retry dispatch deferred")); + await resumeTaskRetry(retry.companyId, retry.id).catch(err => { + scope.onError?.(err); logger.warn({ err, requestId: retry.id }, "Task retry dispatch deferred"); + }); } const waits = await db.select({ wake: agentWakeupRequests }) .from(agentWakeupRequests) @@ -799,7 +804,9 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`, eq(issues.assigneeAgentId, agentWakeupRequests.agentId))) .innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active"))) - .where(and(exists(db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + .where(and(scope.companyId ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, + scope.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, + exists(db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, issues.companyId), eq(issueRecoveryActions.sourceIssueId, issues.id), executionBlockerPredicate(), ))), eq(agentWakeupRequests.status, "deferred_issue_execution"), @@ -831,11 +838,13 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue; if (canContinueCancelledRun(run)) { await resumeSavedComments(wake.companyId, wake.id).catch(err => { + scope.onError?.(err); logger.warn({ err, runId: run.id }, "failed to resume saved input after provider cancellation"); }); continue; } await resumeRemoteStopComments(run, wake.id).catch(err => { + scope.onError?.(err); logger.warn({ err, runId: run.id }, "failed to resume saved execution-wait message"); }); } diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 174c39d6125..6c5f948d303 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js"; import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; @@ -632,6 +633,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableWorktreeRunExecution !== undefined) { await notifyRunDispatch(tx); + await notifyDeferredExecutionEligibility(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyExternalAgentSettlement(tx); await notifyInstructionCommit(tx); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 34234b01c98..fa47af138b9 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; @@ -7629,6 +7630,7 @@ export function issueService(db: Db) { } await notifyIssueRecovery(dbOrTx, { companyId }); + await notifyDeferredExecutionEligibility(dbOrTx, { companyId }); await dbOrTx .delete(issueRelations) .where( @@ -10612,6 +10614,7 @@ export function issueService(db: Db) { } if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: values.companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId: values.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); @@ -11503,6 +11506,7 @@ export function issueService(db: Db) { || patch.executionState !== undefined || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined || blockedByIssueIds !== undefined) { await notifyRunDispatch(tx, { companyId: receiptExisting.companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId: receiptExisting.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); } if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt && @@ -11968,6 +11972,7 @@ export function issueService(db: Db) { await notifyDecisionTargetIssueRemoval(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined); + if (owner) await notifyDeferredExecutionEligibility(tx, owner); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); removedIssue = await tx .delete(issues) @@ -12097,6 +12102,7 @@ export function issueService(db: Db) { const updated = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: issueCompany.companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) @@ -12227,6 +12233,7 @@ export function issueService(db: Db) { const adopted = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: issueCompany.companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) @@ -12483,6 +12490,7 @@ export function issueService(db: Db) { await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: existing.companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) @@ -12544,6 +12552,7 @@ export function issueService(db: Db) { } await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, { companyId: existing.companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index a43868dd25f..3ea3096b710 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,3 +1,4 @@ +import { notifyNativeStatusWakeWork } from "../deferred-execution-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { lockCurrentNativeWorkAssessmentEvidence } from "./work-assessments.js"; import { hasNativeWorkspaceFinalizationOwner } from "./native-workspace-finalization-recovery-state.js"; @@ -478,6 +479,7 @@ async function enqueueWake(input: { .limit(1) .then((rows) => rows[0] ?? null); if (existing) return existing.id; + await notifyNativeStatusWakeWork(input.tx, { companyId: input.companyId }); const inserted = await input.tx .insert(agentWakeupRequests) .values({ diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 17075c34eb9..ba5959b3eb6 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -339,8 +339,10 @@ export async function requestTaskRetry(input: { /** Chat owns admission and authorization. The task receipt follows that * durable outcome; it never launches a parallel generic board execution. */ -export async function reconcileTaskChatRetryReceipts(db: Db) { +export async function reconcileTaskChatRetryReceipts(db: Db, scope: { companyId?: string; requestId?: string; settleOnly?: boolean } = {}) { const pending = await db.select().from(agentWakeupRequests).where(and( + scope.companyId ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, + scope.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, eq(agentWakeupRequests.reason, "task_retry"), eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload}#>>'{taskRetry,chatActionId}' is not null`, )).orderBy(agentWakeupRequests.updatedAt).limit(50); @@ -356,7 +358,7 @@ export async function reconcileTaskChatRetryReceipts(db: Db) { // A deferred chat receipt is admitted input, not a completed dispatch. // Re-stage that exact action through chat's owner; it rechecks authority // and normal queue policy before updating the existing receipt. - if (chat?.status === "processed" && chat.payload.taskRetryIntentId === wake.id && + if (!scope.settleOnly && chat?.status === "processed" && chat.payload.taskRetryIntentId === wake.id && chat.payload.failedRunId === intent.sourceRunId && receipt?.status === "deferred_issue_execution" && !receipt.runId && receipt.payload?.taskRetryRequestId === wake.id) { await db.transaction(async tx => {