From 7bb1af657aa2eda4573c8b77b21afaf0fc226d9b Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:25:05 -0500 Subject: [PATCH 1/4] test: await issue recovery dispatch and cancellation settlement Co-Authored-By: Paperclip --- .../issue-recovery-deadlines.test.ts | 35 ++++++++++++++++--- 1 file changed, 31 insertions(+), 4 deletions(-) diff --git a/server/src/__tests__/issue-recovery-deadlines.test.ts b/server/src/__tests__/issue-recovery-deadlines.test.ts index b3d65f4dbcc..f5416fec571 100644 --- a/server/src/__tests__/issue-recovery-deadlines.test.ts +++ b/server/src/__tests__/issue-recovery-deadlines.test.ts @@ -1,3 +1,4 @@ +import { setImmediate as immediate } from "node:timers/promises"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; @@ -27,6 +28,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea let draining = false; const tasks = new Map number | null; hasPending: () => Promise }>(); const errors: unknown[] = []; + const passes = new Set>(); const extraStops: Array<() => Promise> = []; const extraDeadlines: Array<() => number | null> = []; beforeAll(async () => { @@ -52,7 +54,14 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea onError: error => errors.push(error) }); tasks.clear(); const register = coordinator.register.bind(coordinator); - coordinator.register = (queue, task) => { tasks.set(queue, task); return register(queue, task); }; + coordinator.register = (queue, task) => { + tasks.set(queue, task); + return register(queue, { ...task, run: signal => { + const work = task.run(signal); passes.add(work); + void work.then(() => passes.delete(work), () => passes.delete(work)); + return work; + } }); + }; worker = heartbeat.registerIssueStateRecoveryWork(coordinator, () => !draining); dispatch = heartbeat.registerRunDispatchWork(coordinator, () => !draining); cancellation = heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !draining); @@ -89,6 +98,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea } async function quiet() { await settled(); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); const reads = vi.spyOn(db, "select"); const 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"] }); @@ -97,15 +107,32 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea } async function finish(issueId: string) { await issueService(db).update(issueId, { status: "done" }); - await heartbeat.drainActiveRunExecutions(); + // Closing one blocker can legitimately start its dependent. Wait for this + // task's dispatch and execution; quiet() verifies the whole fixture later. + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(and( + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + sql`${heartbeatRuns.status} in ('queued', 'running')`, + ))).toHaveLength(0); + }, { timeout: 15_000 }); const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); - expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); if (saved.length > 0) { const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())); - await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1); + // Keep PostgreSQL I/O and coordinator settlement ahead of virtual time. + // A single leap can otherwise occur before the retry deadline is armed. + while (Date.now() <= due) { + await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1)); + await Promise.allSettled([...passes]); + await immediate(); + } await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(500); + await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); expect(remaining).toHaveLength(0); From e3fff84b6f81448c730dd165253a1a673f8d19b3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:39:21 -0500 Subject: [PATCH 2/4] fix: scope comment-triggered recovery to the affected issue Preserve company reconstruction for graph, run and eligibility changes. A real PostgreSQL producer regression verifies one assignment check among1000 open tasks, with broader eligibility reconstruction still intact. Co-Authored-By: Paperclip --- doc/plans/2026-10-30-issue-recovery-sleep.md | 6 ++- .../issue-recovery-deadlines.test.ts | 47 +++++++++++++++++++ server/src/services/issue-comment-writes.ts | 3 +- .../services/issue-recovery-notifications.ts | 11 +++-- ...ative-status-reassessment-notifications.ts | 2 +- server/src/services/recovery/work.ts | 5 +- .../services/run-dispatch-notifications.ts | 2 +- 7 files changed, 67 insertions(+), 9 deletions(-) diff --git a/doc/plans/2026-10-30-issue-recovery-sleep.md b/doc/plans/2026-10-30-issue-recovery-sleep.md index f191d5dca42..0c6cdf6b4d3 100644 --- a/doc/plans/2026-10-30-issue-recovery-sleep.md +++ b/doc/plans/2026-10-30-issue-recovery-sleep.md @@ -1,6 +1,6 @@ # Issue recovery without periodic scans -The issue recovery owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database. +The issue recovery owners reconstruct their current obligations once at startup. Ordinary comment writes invalidate only the affected issue for assignment and lock recovery. Lifecycle, graph, run, configuration and eligibility changes invalidate their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database. | Owner | Durable truth and producer boundary | Deadline or event | Exit | | --- | --- | --- | --- | @@ -11,6 +11,10 @@ The issue recovery owners reconstruct their current obligations once at startup. | Task watchdog | Active issue watchdogs and their current source subtree; normal watchdog upsert/disable; issue/comment/document/work-product and wait changes | Exact source creation plus the existing 15-second first-run grace. An incomplete scan retains the watchdog with error backoff. | Live path, open/reviewed reusable review, disabled configuration, or ineligible source. | | Stale issue lock | Existing checkout/execution run references; issue and run lifecycle changes | A committing ownership change, or the fresh owning controller's saved lease expiry. | No stale reference, a valid owner, or the existing guarded terminalization and lock clear. | +Local comment hints do not broaden into an assignment check for every open task. A real PostgreSQL regression starts with1,000 open assigned issues, writes one comment through `issueService.addComment`, and observes exactly one assignment check; a subsequent company eligibility notification still checks all1,000. Restoring the previous comment hint makes that same regression check all1,000 after the comment. The test substitutes the expensive assignment decision only to count its invocations; it uses the real transaction, notification, coordinator and paged SQL. Separate producer acceptance cases exercise real recovery decisions and process execution. + +Parent/descendant disposition repair and shared execution references make status and run changes broader than a comment. Their existing notification boundaries deliberately retain company scope. Orphan blockers, active recovery actions, resolved dependencies and watchdogs also keep their company/subtree duties; an issue comment does not drop those independent obligations. The scope serializer selects known primitive fields and never serializes an entire insert or update object. + Reconstruction uses keyset pages of at most100rows. Repeated company hints coalesce reconstruction for30seconds. Already-known entry deadlines preempt that spacing without an early probe on each comment. Scheduling suppression is read once per pass. A failed read keeps scoped backoff debt. The two subsidiary company repairs do not run inside the per-issue loop. Legacy direct aggregate API calls retain both repairs. The owner methods accept company and issue/watchdog identity. A narrowed lock check still reads other same-company references before using terminal-issue authority. A stale terminal alias cannot end an active issue's shared run. The controller lease owner supplies its fresh expiry; the existing write predicate revalidates the observed boot ID and database-clock expiry. diff --git a/server/src/__tests__/issue-recovery-deadlines.test.ts b/server/src/__tests__/issue-recovery-deadlines.test.ts index f5416fec571..e23a672c0dc 100644 --- a/server/src/__tests__/issue-recovery-deadlines.test.ts +++ b/server/src/__tests__/issue-recovery-deadlines.test.ts @@ -9,6 +9,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { heartbeatService } from "../services/heartbeat.js"; import { issueService } from "../services/issues.js"; import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { registerIssueRecoveryWork } from "../services/recovery/work.js"; import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js"; import { taskWatchdogService } from "../services/task-watchdogs.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -223,6 +224,52 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, foreignAction.id)))[0].status).toBe("active"); expect(await originalSelect().from(heartbeatRuns)).toHaveLength(0); await quiet(); }); + it("limits an ordinary comment to its issue among one thousand open assignments", async () => { + const f = await fixture(); + const rows = await db.insert(issues).values(Array.from({ length: 1000 }, (_, index) => ({ + companyId: f.companyId, title: `Open work ${index}`, status: "todo", assigneeAgentId: f.agentId, + }))).returning({ id: issues.id }); + const checked: string[] = []; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const pass = task.run(signal); passes.add(pass); + void pass.then(() => passes.delete(pass), () => passes.delete(pass)); + return pass; + } }); + // Observe the actual paged reconstruction boundary without executing one + // thousand adapters. The producer, transaction, coordinator and SQL are real; + // the other acceptance cases exercise the resulting recovery decisions. + worker = registerIssueRecoveryWork(db, coordinator, { + stranded: async scope => { checked.push(scope.issueId); }, + orphanBlockers: async () => {}, recoveryActions: async () => {}, + dependencies: async () => ({ enqueueFailed: 0 }), locks: async () => {}, + watchdogs: async () => ({ incomplete: 0 }), schedulingSuppressed: async () => false, + }, () => true); + await worker.ready; + async function settleWork() { + for (let step = 0; step < 120; step++) { + await vi.advanceTimersByTimeAsync(1000); + await Promise.allSettled([...passes]); await immediate(); + if (coordinator.nextWakeAt() === null && idleWorkSnapshot().active === 0) return; + } + throw new Error("Recovery did not settle its committed work within two virtual minutes"); + } + await settleWork(); + expect(new Set(checked).size).toBe(1000); + checked.length = 0; + await issueService(db).addComment(rows[0].id, "New evidence on this task", { userId: "recovery-user" }); + await settleWork(); + expect(checked).toEqual([rows[0].id]); + // Company eligibility changes still reconstruct all assignments. Narrowing + // comment hints must not hide unrelated work after an operator resumes it. + checked.length = 0; + await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId })); + await settleWork(); + expect(new Set(checked).size).toBe(1000); + await quiet(); + }); it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => { const f = await fixture(); await start(); await quiet(); const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId }); diff --git a/server/src/services/issue-comment-writes.ts b/server/src/services/issue-comment-writes.ts index a968ecb1fec..39d8f6f891c 100644 --- a/server/src/services/issue-comment-writes.ts +++ b/server/src/services/issue-comment-writes.ts @@ -7,7 +7,8 @@ type WriteDb = Db | Parameters[0]>[0]; export function insertIssueComment(db: WriteDb, input: PgInsertValue & { companyId: string }) { return db.transaction(async tx => { - await notifyIssueRecovery(tx, { companyId: input.companyId }); + await notifyIssueRecovery(tx, { companyId: input.companyId, + issueId: typeof input.issueId === "string" ? input.issueId : undefined }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`); return tx.insert(issueComments).values(input).returning(); }); diff --git a/server/src/services/issue-recovery-notifications.ts b/server/src/services/issue-recovery-notifications.ts index 3c974228d86..7b709960831 100644 --- a/server/src/services/issue-recovery-notifications.ts +++ b/server/src/services/issue-recovery-notifications.ts @@ -1,11 +1,16 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export type IssueRecoveryScope = { companyId: string }; +/** An issue hint is reserved for local facts such as comments. Lifecycle, + * graph, run and eligibility changes retain company scope because their + * consequences can include ancestors and shared execution references. */ +export type IssueRecoveryScope = { companyId: string; issueId?: string }; export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) { - const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + const companyHint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + const issueHint = scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*"; for (const queue of [DELIVERY_QUEUES.orphanBlockers, DELIVERY_QUEUES.activeRecoveryActions, DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies, DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) { - await notifyDeliveryWork(transaction, queue, hint); + await notifyDeliveryWork(transaction, queue, + queue === DELIVERY_QUEUES.strandedAssignments || queue === DELIVERY_QUEUES.staleIssueLocks ? issueHint : companyHint); } } diff --git a/server/src/services/native-status-reassessment-notifications.ts b/server/src/services/native-status-reassessment-notifications.ts index 4b5eac73b2f..d3b543d9914 100644 --- a/server/src/services/native-status-reassessment-notifications.ts +++ b/server/src/services/native-status-reassessment-notifications.ts @@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { - await notifyIssueRecovery(tx, scope); + await notifyIssueRecovery(tx, scope ? { companyId: scope.companyId } : undefined); return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); } diff --git a/server/src/services/recovery/work.ts b/server/src/services/recovery/work.ts index 8bf7b75c7f4..407e4f1eb6d 100644 --- a/server/src/services/recovery/work.ts +++ b/server/src/services/recovery/work.ts @@ -32,13 +32,14 @@ export function registerIssueRecoveryWork( reconstructionIntervalMs: 30_000, beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); }, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, - matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + matches: (entry, scope) => !scope || (entry.companyId === scope.companyId && (!scope.issueId || entry.id === scope.issueId)), load: async scope => { - const key = scope?.companyId ?? "*"; + const key = scope ? JSON.stringify([scope.companyId, scope.issueId]) : "*"; if (!enabled) { cursors.delete(key); return []; } const cursor = cursors.get(key); const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and( scope ? eq(issues.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(issues.id, scope.issueId) : undefined, cursor ? gt(issues.id, cursor) : undefined, statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)), statuses ? isNull(issues.hiddenAt) : undefined, diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 9e8784b7481..cba58357cf0 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type RunDispatchScope = { companyId: string }; export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { - await notifyIssueRecovery(transaction, scope); + await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined); const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); From c74f5f12bad434068fcb7209c4e22e45eed7149c Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:45:15 -0500 Subject: [PATCH 3/4] Retain moved receipt deadlines and wake released execution gates Co-Authored-By: Paperclip --- .../deferred-execution-receipt-work.test.ts | 198 +++++++++++++++++- server/src/services/access.ts | 2 + .../src/services/deferred-execution-wait.ts | 24 +++ .../heartbeat/deferred-receipt-work.ts | 18 ++ server/src/services/heartbeat/queue.ts | 23 +- server/src/services/heartbeat/run-control.ts | 8 +- server/src/services/issue-tree-control.ts | 5 + 7 files changed, 249 insertions(+), 29 deletions(-) create mode 100644 server/src/services/deferred-execution-wait.ts 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({ From dff6c00468671a088dc307f18132334e855f5d59 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:50:27 -0500 Subject: [PATCH 4/4] Join receipt acceptance passes before advancing saved deadlines Co-Authored-By: Paperclip --- .../2026-10-30-deferred-execution-sleep.md | 4 +++ .../deferred-execution-receipt-work.test.ts | 29 +++++++++++++++---- 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/doc/plans/2026-10-30-deferred-execution-sleep.md b/doc/plans/2026-10-30-deferred-execution-sleep.md index 0478fe79300..56e3611d939 100644 --- a/doc/plans/2026-10-30-deferred-execution-sleep.md +++ b/doc/plans/2026-10-30-deferred-execution-sleep.md @@ -17,3 +17,7 @@ Three scoped owners reconstruct only their actual durable receipt predicates, wi 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. + +Review controls cover queue edits that move an already loaded grace deadline. Each due check re-reads current stored grace or claim time before admission. Subtree hold release, subtree restore, and member archival publish eligibility inside their own transactions. Member archival preserves the cancelled Retry tombstone created by an earlier user reassignment. + +A resume that first records a new execution-wait reason arms its existing 30-second grace once. Writing the same closed-gate reason again does not publish another hint. A clock obligation, an eligibility change, and an infrastructure failure therefore retain separate exits: stored deadline, owning committed producer, and bounded failure backoff. Refreshing a diagnostic timestamp alone does not create permanent polling behind a deliberate pause. diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index eb60e2ed449..914f33fe609 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -1,3 +1,4 @@ +import { setImmediate as immediate } from "node:timers/promises"; import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; import { accessService } from "../services/access.js"; import { writeDeferredExecutionWait } from "../services/deferred-execution-wait.js"; @@ -27,6 +28,7 @@ describe("deferred receipt durable dispatch", () => { let coordinator: ReturnType; let drain = false; const errors: unknown[] = []; + const passes = new Set>(); beforeAll(async () => { const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DEFERRED_RECEIPT_TEST_DATABASE_URL"); if (!external) temporary = await startEmbeddedPostgresTestDatabase("deferred-receipts-"); @@ -38,6 +40,11 @@ describe("deferred receipt durable dispatch", () => { heartbeat = heartbeatService(db); coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true, onError: error => errors.push(error) }); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const pass = Promise.resolve(task.run(signal)); passes.add(pass); + void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass; + } }); }); afterEach(async () => { await coordinator.stop(); await heartbeat.drainActiveRunExecutions(); @@ -70,12 +77,24 @@ describe("deferred receipt durable dispatch", () => { } async function finish(issueId: string) { await issueService(db).update(issueId, { status: "done" }); - await heartbeat.drainActiveRunExecutions(); + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); + }, { timeout: 15_000 }); 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)); + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + if (saved.length) { + const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())); + while (Date.now() <= due) { + await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1)); + await Promise.allSettled([...passes]); await immediate(); + } + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(500); await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); + 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();