diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index ba0f13bed58..eb60e2ed449 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -1,7 +1,14 @@ +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; +import { accessService } from "../services/access.js"; +import { writeDeferredExecutionWait } from "../services/deferred-execution-wait.js"; +import { createQueuedCommentIssueLockWriter } from "../modules/wake-queue/adapters/queued-comment-postgres.js"; +import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js"; +import { issueTreeControlService } from "../services/issue-tree-control.js"; 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 { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, createDb, reconcileDatabaseWork, heartbeatRuns, issueComments, 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"; @@ -61,6 +68,15 @@ describe("deferred receipt durable dispatch", () => { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed); } + async function finish(issueId: string) { + await issueService(db).update(issueId, { 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)); + } 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, @@ -86,13 +102,7 @@ describe("deferred receipt durable dispatch", () => { 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)); + await finish(f.issue.id); 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])); @@ -208,6 +218,178 @@ describe("deferred receipt durable dispatch", () => { .toBe("issued"), { timeout: 5_000 }); await quiet(); }); + it("retains a moved interrupt grace deadline after an ordinary saved queue reorder", async () => { + const f = await fixture(), initialAt = new Date(); + const comments = await db.insert(issueComments).values(["first", "second"].map(body => ({ + companyId: f.companyId, issueId: f.issue.id, body, authorType: "user", authorUserId: f.actorId, + }))).returning(); + const payload = { issueId: f.issue.id, queuedCommentInterrupt: { actorId: f.actorId }, + _paperclipWakeContext: { wakeCommentIds: comments.map(row => row.id) } }; + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution", payload, updatedAt: initialAt }).returning(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const deps = createHeartbeatQueueTestDependencies(db), queue = createHeartbeatQueue(db, deps); + await registerDeferredReceiptWork(db, coordinator, { resume: scope => queue.resumeDeferredExecutionReceipts(scope), + settleTaskChat: async () => {}, dispatchNative: async () => {}, schedulingSuppressed: async () => false }, () => !drain).ready; + await vi.advanceTimersByTimeAsync(initialAt.getTime() + 10_000 - Date.now()); + const movedAt = new Date(); + const writer = createQueuedCommentIssueLockWriter(db, { syncCommentReferences: async () => {}, + deleteCommentReferenceSource: async () => {}, syncCommentExternalObjectsSafely: async () => {} }); + await writer.withLockedQueue({ issue: f.issue, queueId: wake.id, + actor: { actorType: "user", actorId: f.actorId, agentId: null, runId: null, agentApiKeyId: null } }, + async (_locked, tx) => { await tx.updateWakeQueuedCommentIds({ wakeId: wake.id, payload, + ids: comments.map(row => row.id).reverse(), updatedAt: movedAt }); }); + await vi.advanceTimersByTimeAsync(initialAt.getTime() + 30_000 - Date.now()); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(deps.resumeQueuedCommentInterrupt).not.toHaveBeenCalled(); + expect(coordinator.nextWakeAt()).toBe(movedAt.getTime() + 30_000); + await vi.advanceTimersByTimeAsync(movedAt.getTime() + 30_000 - Date.now()); + await vi.waitFor(() => expect(deps.resumeQueuedCommentInterrupt).toHaveBeenCalledExactlyOnceWith(f.companyId, wake.id)); + await quiet(); + }); + it("reactivates a held task Retry when the authorized subtree hold is released without wakeAgents", async () => { + const f = await fixture(), tree = issueTreeControlService(db); + const actor = { actorType: "user" as const, actorId: f.actorId, userId: f.actorId }; + const hold = await tree.createHold(f.companyId, f.issue.id, { mode: "pause", actor }); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId, + observedRunId: f.source.id, idempotencyKey: randomUUID() }); + expect(request.requestId).toBeTruthy(); + await vi.waitFor(async () => { + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)); + expect(wake.payload?.executionWait).toMatchObject({ reason: "issue_tree_hold_active" }); + expect(wake.runId).toBeNull(); + }, { timeout: 5_000 }); + await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 }); + await tree.releaseHold(f.companyId, f.issue.id, hold.hold.id, { actor, reason: "Resume saved Retry" }); + let runId: string | null = null; + await vi.waitFor(async () => { + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)); + expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId; + }, { timeout: 40_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 }); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(1); + await finish(f.issue.id); await quiet(); + }); + it("keeps a moved native dispatch error deadline instead of retrying at its old time", async () => { + const f = await fixture(), initialAt = new Date(); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "automation", status: "queued", error: "Transient dispatch failure", updatedAt: initialAt, + requestedByActorType: "system", requestedByActorId: "native-status-committer" }).returning(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const dispatchNative = vi.fn(async () => {}); + await registerDeferredReceiptWork(db, coordinator, { resume: async () => {}, settleTaskChat: async () => {}, + dispatchNative, schedulingSuppressed: async () => false }, () => !drain).ready; + await vi.advanceTimersByTimeAsync(initialAt.getTime() + 10_000 - Date.now()); + const movedAt = new Date(); + await db.update(agentWakeupRequests).set({ updatedAt: movedAt }).where(eq(agentWakeupRequests.id, wake.id)); + await vi.advanceTimersByTimeAsync(initialAt.getTime() + 30_000 - Date.now()); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(dispatchNative).not.toHaveBeenCalled(); expect(coordinator.nextWakeAt()).toBe(movedAt.getTime() + 30_000); + await vi.advanceTimersByTimeAsync(movedAt.getTime() + 30_000 - Date.now()); + await vi.waitFor(() => expect(dispatchNative).toHaveBeenCalledTimes(1)); await quiet(); + }); + it("arms one new wait reason grace and stays quiet when that same gate is still closed", async () => { + const f = await fixture(); let reason = "operator_pause"; + const resume = vi.fn(async (scope: { companyId?: string; requestId?: string }) => { + await writeDeferredExecutionWait(db, { companyId: scope.companyId!, requestId: scope.requestId! }, { reason }); + }); + await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, dispatchNative: async () => {}, + schedulingSuppressed: async () => false }, () => !drain).ready; + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution", reason: "task_retry" }).returning(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + for (const nextReason of ["operator_pause", "budget_pause"]) { + reason = nextReason; + await writeDeferredExecutionWait(db, { companyId: f.companyId, requestId: wake.id }, { reason }); + const [current] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + const due = current.updatedAt.getTime() + 30_000; + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(due)); + const checked = resume.mock.calls.length; + await vi.advanceTimersByTimeAsync(due - Date.now()); + await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(checked + 1)); + await quiet(); + } + expect(resume).toHaveBeenCalledTimes(2); + }); + it("reactivates a saved Retry through ordinary restore cancelled task", async () => { + const f = await fixture(), actor = { actorType: "user" as const, actorId: f.actorId, userId: f.actorId }; + const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId, + observedRunId: f.source.id, idempotencyKey: randomUUID() }); + const tree = issueTreeControlService(db); + const cancel = await tree.createHold(f.companyId, f.issue.id, { mode: "cancel", actor }); + await tree.cancelIssueStatusesForHold(f.companyId, f.issue.id, cancel.hold.id); + const restore = await tree.createHold(f.companyId, f.issue.id, { mode: "restore", actor }); + const commitEligibility = () => tree.restoreIssueStatusesForHold(f.companyId, f.issue.id, restore.hold.id, { actor }); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5_000 }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].runId).toBeNull(); + await commitEligibility(); + let runId: string | null = null; + await vi.waitFor(async () => { + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.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 }); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(1); + await finish(f.issue.id); await quiet(); + }); + it("member archive publishes committed eligibility while preserving a cancelled Retry tombstone", async () => { + const f = await fixture(), departing = `departing-${randomUUID()}`; + const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId, + observedRunId: f.source.id, idempotencyKey: randomUUID() }); + await db.insert(authUsers).values({ id: departing, name: "Departing member", email: `${departing}@example.test`, + createdAt: new Date(), updatedAt: new Date() }); + const [member] = await db.insert(companyMemberships).values({ companyId: f.companyId, principalId: departing, + principalType: "user", status: "active", membershipRole: "member" }).returning(); + await issueService(db).update(f.issue.id, { assigneeUserId: departing, assigneeAgentId: null }); + const readIntent = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0]; + expect(await readIntent()).toMatchObject({ status: "cancelled", runId: null }); + const deferred = vi.fn(), native = vi.fn(); + const unsubscribes = [subscribeDeliveryWork(db, DELIVERY_QUEUES.deferredExecutionReceipts, deferred), + subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeStatusWakeIntents, native)]; + try { + await expect(db.transaction(async tx => { + await accessService(tx as unknown as typeof db).archiveMember(f.companyId, member.id, + { reassignment: { assigneeAgentId: f.agent.id } }); + expect(deferred).not.toHaveBeenCalled(); expect(native).not.toHaveBeenCalled(); + throw new Error("roll back member archive"); + })).rejects.toThrow("roll back member archive"); + expect(deferred).toHaveBeenCalledTimes(1); expect(native).toHaveBeenCalledTimes(1); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].assigneeUserId).toBe(departing); + await reconcileDatabaseWork(db, DELIVERY_QUEUES.deferredExecutionReceipts); + await reconcileDatabaseWork(db, DELIVERY_QUEUES.nativeStatusWakeIntents); + deferred.mockClear(); native.mockClear(); + await db.transaction(async tx => { + await accessService(tx as unknown as typeof db).archiveMember(f.companyId, member.id, + { reassignment: { assigneeAgentId: f.agent.id } }); + expect(deferred).not.toHaveBeenCalled(); expect(native).not.toHaveBeenCalled(); + }); + expect(deferred).toHaveBeenCalledTimes(1); expect(native).toHaveBeenCalledTimes(1); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]).toMatchObject({ + assigneeUserId: null, assigneeAgentId: f.agent.id, + }); + expect((await db.select().from(companyMemberships).where(eq(companyMemberships.id, member.id)))[0].status).toBe("archived"); + expect(await readIntent()).toMatchObject({ status: "cancelled", runId: null }); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(0); + } finally { for (const unsubscribe of unsubscribes) unsubscribe(); } + 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, diff --git a/server/src/services/access.ts b/server/src/services/access.ts index aa0f9dcc828..32086736eb2 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; @@ -743,6 +744,7 @@ export function accessService(db: Db) { sql`${issues.status} not in ('done', 'cancelled')`, ); await notifyNativeStatusReassessment(tx, { companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const resetInProgress = await tx diff --git a/server/src/services/deferred-execution-wait.ts b/server/src/services/deferred-execution-wait.ts new file mode 100644 index 00000000000..c4857d5b692 --- /dev/null +++ b/server/src/services/deferred-execution-wait.ts @@ -0,0 +1,24 @@ +import { and, eq, sql } from "drizzle-orm"; +import { agentWakeupRequests, type Db } from "@paperclipai/db"; +import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js"; + +export async function writeDeferredExecutionWait(db: Db, scope: { companyId: string; requestId: string }, + condition: Record, options: { replace?: boolean } = {}) { + return db.transaction(async tx => { + const [current] = await tx.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId), + eq(agentWakeupRequests.status, "deferred_issue_execution"))).for("update"); + if (!current) return; + const previous = current.payload?.executionWait; + const previousReason = previous && typeof previous === "object" && !Array.isArray(previous) + ? (previous as Record).reason : undefined; + if (!previous || previousReason !== condition.reason) await notifyDeferredExecutionWork(tx, scope); + await tx.update(agentWakeupRequests).set({ + payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', + ${options.replace ? sql`'{}'::jsonb` : sql`coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb)`} + || ${JSON.stringify(condition)}::jsonb)`, + updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId), + eq(agentWakeupRequests.status, "deferred_issue_execution"))); + }); +} diff --git a/server/src/services/heartbeat/deferred-receipt-work.ts b/server/src/services/heartbeat/deferred-receipt-work.ts index fddbce3be8d..7a69bc3be58 100644 --- a/server/src/services/heartbeat/deferred-receipt-work.ts +++ b/server/src/services/heartbeat/deferred-receipt-work.ts @@ -79,6 +79,24 @@ export function registerDeferredReceiptWork( }, check: async entry => { if (!enabled) return null; + if (kind !== "task_chat") { + const [current] = await db.select({ status: agentWakeupRequests.status, payload: agentWakeupRequests.payload, + updatedAt: agentWakeupRequests.updatedAt, claimedAt: agentWakeupRequests.claimedAt, error: agentWakeupRequests.error, + 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(eq(agentWakeupRequests.companyId, entry.companyId), + eq(agentWakeupRequests.id, entry.id))); + if (!current) return null; + const due = kind === "native" && current.status === "claimed" && current.claimedAt + ? current.claimedAt.getTime() + 60_000 + : kind === "native" && current.status === "queued" && current.error ? current.updatedAt.getTime() + 30_000 + : kind === "deferred" && current.status === "deferred_issue_execution" && !current.interruptedLegacy + && (current.payload?.queuedCommentInterrupt || current.payload?.executionWait) + ? current.updatedAt.getTime() + 30_000 : null; + if (due !== null && due > Date.now()) return { ...entry, dueAt: due }; + } const errors: unknown[] = []; let next: number | null = null; const scope = { companyId: entry.companyId, requestId: entry.id }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index ef3ecfeb7ac..962f15f327e 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { writeDeferredExecutionWait } from "../deferred-execution-wait.js"; import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js"; import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { notifyRunDispatch } from "../run-dispatch-notifications.js"; @@ -2523,17 +2524,11 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // its one durable receipt ID on a skipped wake. if (caseRequest) return { created: false }; if (executionWaitRequestId) { - await db.update(agentWakeupRequests).set({ - payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', - coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify({ - reason: skipReason, message: patch.error ?? (skipReason === "issue_tree_hold_active" - ? "This task is paused. Resume it to send your saved message." - : "Waiting for task execution to be enabled. Your message is saved."), - })}::jsonb)`, - updatedAt: new Date(), - }).where(and(eq(agentWakeupRequests.id, executionWaitRequestId), - eq(agentWakeupRequests.companyId, agent.companyId), eq(agentWakeupRequests.agentId, agentId), - eq(agentWakeupRequests.status, "deferred_issue_execution"))); + await writeDeferredExecutionWait(db, { companyId: agent.companyId, requestId: executionWaitRequestId }, { + reason: skipReason, message: patch.error ?? (skipReason === "issue_tree_hold_active" + ? "This task is paused. Resume it to send your saved message." + : "Waiting for task execution to be enabled. Your message is saved."), + }); return { created: false }; } const request = { @@ -3421,10 +3416,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende opts.issueStateGuard?.onExecutionDeferred?.(executionBlocker.cause); const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait }; if (executionWaitRequestId) { - await tx.update(agentWakeupRequests).set({ - payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${JSON.stringify(condition)}::jsonb)`, - updatedAt: new Date(), - }).where(eq(agentWakeupRequests.id, executionWaitRequestId)); + await writeDeferredExecutionWait(tx as unknown as Db, { companyId: agent.companyId, + requestId: executionWaitRequestId }, condition, { replace: true }); return { kind: "deferred" as const, admitted: true }; } if (durableRequest || wakeCommentId || diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 691309a1d76..a410d905d6a 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import { writeDeferredExecutionWait } from "../deferred-execution-wait.js"; import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { observeStoppedWorkspaceRecovery } from "../workspace-recovery.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; @@ -617,12 +618,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont onBlocked: (reason, message) => { wait = { reason, message }; }, }); if (!admitted) { - await db.update(agentWakeupRequests).set({ - payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', - coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify(wait)}::jsonb)`, - updatedAt: new Date(), - }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId), - eq(agentWakeupRequests.status, "deferred_issue_execution"))); + await writeDeferredExecutionWait(db, { companyId: run.companyId, requestId: wake.id }, wait); continue; } } diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index ff558b81ef0..2c05df61ceb 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; @@ -760,6 +761,7 @@ export function issueTreeControlService(db: Db) { const { hold: resumeHold } = await db.transaction(async (tx) => { await notifyIssueRecovery(tx, { companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId }); const [createdHold] = await tx .insert(issueTreeHolds) .values({ @@ -839,6 +841,7 @@ export function issueTreeControlService(db: Db) { const { hold, members } = await db.transaction(async (tx) => { await notifyIssueRecovery(tx, { companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId }); const [createdHold] = await tx .insert(issueTreeHolds) .values({ @@ -1003,6 +1006,7 @@ export function issueTreeControlService(db: Db) { const now = new Date(); const releasedCancelHoldIds = activeCancelHolds.map((hold) => hold.id); const updatedIssues = await db.transaction(async (tx) => { + await notifyDeferredExecutionEligibility(tx, { companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const restored: TreeStatusUpdateResult["updatedIssues"] = []; @@ -1182,6 +1186,7 @@ export function issueTreeControlService(db: Db) { const [updated] = await db.transaction(async tx => { await notifyIssueRecovery(tx, { companyId }); + await notifyDeferredExecutionEligibility(tx, { companyId }); return tx .update(issueTreeHolds) .set({