From cbfaab918aa4000b04cf201fcddca37598b50727 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 14:10:02 -0500 Subject: [PATCH] fix(server): wake failed retry notices from receipt changes Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 13 + ...-10-25-failed-retry-notices-sleep-audit.md | 98 +++++- .../app-private-hostname-gate.test.ts | 3 +- .../chat-channels.integration.test.ts | 279 +++++++++++++++++- server/src/app.ts | 2 + .../modules/wake-queue/adapters/postgres.ts | 1 + server/src/services/chat-channels.ts | 3 - .../chat-channels/failed-retry-notices.ts | 2 - .../chat-channels/failed-run-retry.ts | 19 +- .../services/chat-failed-retry-notice-work.ts | 39 +++ .../chat-inbound-wake-notice-notifications.ts | 7 +- .../services/delivery-work-notifications.ts | 1 + server/src/services/heartbeat/queue.ts | 15 +- 13 files changed, 462 insertions(+), 20 deletions(-) create mode 100644 server/src/services/chat-failed-retry-notice-work.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 37fe3dff804..b4e1807b321 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1774,3 +1774,16 @@ voice inventory amplification findings 4371283330/4371307607, and #17580 still has unresolved company-offboarding/cleanup findings 4370459800/4370543984. These are required repairs, not exemptions. The full original migration scope, master reconciliation, and all latest-head gates remain required. + + +## Slice 31 worker and producer checkpoint (2026-10-25) + +Failed-retry notices now register a scoped worker and leave periodic publication +maintenance. Real missed wakes were reproduced for admission, agent/tree +cancellation, resumed-admission denial, early admission refusal, and requester +restoration. Their owning transactions now notify through the existing shared +boundary. Fourteen selected lifecycle cases and all 1,503 chat/startup acceptance +cases pass. Workspace typecheck and build pass. See the retry audit for retained +failures, exact commands/logs, and native source-cleanup qualification still open. +This is a draft checkpoint, not a merge-ready claim. Older parent gates and every +remaining original recurring-server duty stay in scope. No merge occurred. diff --git a/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md b/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md index 6be4dc53085..82fe5cbe584 100644 --- a/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md +++ b/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md @@ -3,9 +3,10 @@ Date: 2026-10-25. Branch: `codex/sleep-31-failed-retry-notices`. Parent: [#17647](https://github.com/paperclipai/paperclip/pull/17647), `fee07f3fa8cbc489cb9bbf657a6065c6e00748c3`. -This is slice 31 of the recurring-server-work stack. No production worker is -registered yet, and the periodic retry-notice pass remains until producer and -lifecycle acceptance is complete. No PR has been opened for this slice yet. +This is slice 31 of the recurring-server-work stack. The branch now registers +a scoped worker and removes retry notices from periodic publication maintenance. +Producer audit and full acceptance remain in progress; this is not merge-ready. +No PR has been opened for this slice yet. ## Obligation and boundary @@ -139,3 +140,94 @@ Additional producer audit findings to resolve before wiring: the separate task-intent receipt. The notice candidate uses the chat-action receipt; examine generation/source authorization before adding blanket wakes to all intent timestamp writes. + + +## Admission and cancellation migration checkpoint + +The worker now uses the shared scoped-deadline coordinator. It reconstructs +existing receipt obligations at startup, resets an issue cursor on a committed +invalidation, and retains bounded pages, real contention, and failed selection. +Settled and denied history has no recurring scan. No new table is added. + +`notifyChatInboundWakeNotices` now also notifies the retry-notice topic. This +keeps task-tree, agent, budget, company, deferred-queue, source/privacy, and +existing milestone authority signals at one boundary. The hint serializes only +company and optional issue IDs. The new worker is absent from older parents; +there it creates no new timer or database listener. + +New producer changes are narrow: + +| Owner | Obligation and notification | +| --- | --- | +| Retry admission callback | After current authorization succeeds, register intent inside the heartbeat transaction before the receipt write. | +| Retry authorization denial | Retire the exact unadmitted receipt in a transaction that registers notice intent before its conditional update. No provider I/O enters the transaction. | +| Heartbeat early refusal | A durable chat receipt can be skipped before its authorization callback. Signal its existing execution-wait transaction, or create one transaction for the previously standalone skipped-row insert. Non-chat callers retain their existing path. | +| Deferred-queue requester merge | A manual requester replacement can restore or revoke the original receipt identity. Register intent before that update. Payload-only merges do not add a notification. | + +Causal evidence retained: + +- `/tmp/sleep31-cancellation-before.log`: agent cancellation missed the notice; + the tree case initially used an invalid direct function reference. +- `/tmp/sleep31-tree-cancellation-before.log`: corrected real task-tree service + also reproduces the missing notice. The shared notification repair passes + 39 related cases in `/tmp/sleep31-cancellation-fixed.log`. +- `/tmp/sleep31-real-admission-lifecycle.log`: real heartbeat admission missed + the notice in normal, drain, and failed-selection modes. Restart alone found + it. The callback notification fixes that causal gap. +- `/tmp/sleep31-real-admission-fixed.log`: one normal-path assertion observed + an unexpected worker error without printing it. This error was not classified + as a flake. The assertion now shows full errors. The diagnostic run instead + exposed the expected failure's AggregateError wrapper; that expectation now + checks its nested cause. Retain the first observation for broad acceptance. +- `/tmp/sleep31-denial-before.log`: the actual recovery service restages a + deferred retry; a controlled admission rejection saves `failed`, but no + not-started notice appears. The conditional transaction notification fixes it. +- `/tmp/sleep31-early-refusal-before.log`: real heartbeat admission with + wake-on-demand disabled saves a skipped receipt but misses its notice. The + early-refusal writer notification fixes it. +- `/tmp/sleep31-requester-before.log`: the real deferred-queue Postgres writer + restores the original requester but misses the notice. This exercises the + producer transaction, not a synthetic notification. It does not establish + every end-to-end ordinary/retry coalescing path. +- `/tmp/sleep31-producer-lifecycle.log`: all 14 selected cases pass, including + real admission, first-write drain/resume, failed selection followed by drain, + stop/restart, agent/tree rollback and commit, admission denial, early refusal, + privacy restoration after idle and during a bounded page, and requester merge. + The quiet assertions observe no root SQL/transaction calls for 120 seconds. +- `/tmp/sleep31-workspace-typecheck.log` and + `/tmp/sleep31-workspace-build.log`: workspace typecheck and build pass. + +The full chat/startup suite passes all 1,503 cases on fresh migrated +`paperclip_sleep31_wired_chat_20261025_test`, with evidence in +`/tmp/sleep31-wired-full-chat.log` (456.47 seconds, exit 0). This includes +the existing provider authorization and publication acceptance suite. +The stopped-worker case also checks that publication maintenance no longer +projects retry notices before startup reconstruction resumes the worker. + +### Remaining producer qualifications + +`readTaskChatRetryIntent` uses saved identity, generation, source, actor and +binding. It does not use the separate task-intent status or updatedAt. Ordinary +status reconciliation therefore does not itself restore notice eligibility. +Restaging the chat action from processed to issued also stays within the set of +statuses already allowed for publication. Changes that bind a receipt to a run +remove it from this notice obligation; the projection rechecks before writing. + +`assertFailedRetryReceipt` validates stable receipt identity, not normalized +queued-comment payload. Payload-only comment normalization is not an authority +restoration for retry notices. The requester replacement is different and is +now covered at its actual writer. Existing ordinary-source notice behavior is +still included through the shared signal fanout. + +Native exact-retry source authorization needs further qualification before +readiness: it reads native finalization ownership, cleanup history, environment +lease settlement, and retained checkpoint proof. `native-session-executor.ts` +commits a settled maintenance history entry and then removes its activation +file; `environments.ts::releaseLease` uses environment-cleanup notifications. +Determine which of these can restore an already-existing retry notice and prove +the required wake after the full authority boundary. A signal before the final +filesystem handoff may be too early. Do not claim all native cleanup producers +are covered by the general milestone fanout merely because run finalizers use it. + +All older stack failures and original recurring-server duties remain required. +No merge, expanded Cloud wake coverage, or deployed host sleep is claimed. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 29b46cddee7..1dad4ca398a 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -91,7 +91,8 @@ describe("createChatReconciliationCoordinator", () => { const publicationMaintenance = chat.slice(chat.indexOf("async function processPublicationMaintenance("), chat.indexOf("async function schedulePendingPublications(")); expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications("); - expect(publicationMaintenance).toContain("await enqueueFailedChatRetryPublications("); + expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications("); + expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,"); }); it("separates live voice recovery from the remaining periodic maintenance", () => { diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 9e747646f0e..d73c1016a25 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,5 @@ +import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; +import { chatFailedRetryNoticeWork } from "../services/chat-failed-retry-notice-work.js"; import { issueTreeControlService } from "../services/issue-tree-control.js"; import { chatInboundWakeNoticeWork } from "../services/chat-inbound-wake-notice-work.js"; import { nativeChatProgressWork } from "../services/chat-native-progress-work.js"; @@ -11,7 +13,7 @@ import { configureAgentLifecycle } from "../services/agent-lifecycle.js"; import { deleteCompany } from "../services/company-deletion.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js"; -import { requestTaskRetry } from "../services/task-retry.js"; +import { requestTaskRetry, reconcileTaskChatRetryReceipts } from "../services/task-retry.js"; import { createSystemRecovery } from "../services/system-recovery-state.js"; import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { stageReceiptReactionRemovals } from "../services/chat-receipt-reactions.js"; @@ -62607,6 +62609,281 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); + async function deferredFailedRetryNoticeFixture(wakeOnDemand = true) { + const context = await failedChatRetryFixture("telegram", undefined, true); + const companyId = context.fixture.companyId, issueId = context.issue.id, leaseId = randomUUID(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, + nativePhase: "terminal_failure", runnerProfileJson: {}, + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { + schema: "paperclip.workspace-restore-recovery.v1", runId: context.runId, leaseIds: [leaseId], + } }, + }).where(eq(heartbeatRuns.id, context.runId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: context.runId, phase: "terminal_failure" }); + await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId, + heartbeatRunId: context.runId, provider: "daytona", providerLeaseId: randomUUID(), + status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "retain_on_failure", + metadata: { remoteCwd: "/home/daytona/paperclip-workspace", pendingCleanupInFlight: false }, + }); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "system", + cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), + nextAction: "Verify and preserve the retained source before continuing.", + evidence: { runId: context.runId, automaticRecovery: { replay: "blocked" }, + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] }, + historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: context.runId, leaseIds: [leaseId] }, + systemRecovery: { ...createSystemRecovery(context.runId), state: "exhausted", attempts: 2, retriesUsed: 1, + attemptId: randomUUID(), firstFailureAt: Date.now() - 1000, deadlineAt: Date.now() + 60_000, + preservationOperationIds: [randomUUID()] }, + }, + }); + const retry = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user", + actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(), + prepareChatRetry: context.service.prepareFailedChatRunRetry }); + await db.update(agents).set({ adapterType: "paperclip_runner", status: "running", + runtimeConfig: { heartbeat: { wakeOnDemand, maxConcurrentRuns: 1 } }, + }).where(eq(agents.id, context.fixture.assignedAgentId)); + await db.insert(heartbeatRuns).values({ companyId, agentId: context.fixture.assignedAgentId, + status: "running", startedAt: new Date() }); + context.wakeup.mockImplementation(heartbeatService(db).wakeup); + return { ...context, retry, admit: () => context.service.processFailedChatRunRetry(retry.chatActionId!) }; + } + + it.each(["normal", "drain", "failed selection", "stopped", "disabled"] as const)("reactivates failed retry notices from real admission after idle through %s", async mode => { + const context = await deferredFailedRetryNoticeFixture(mode !== "disabled"); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const selection = vi.spyOn(context.service, "loadFailedRetryNoticeScopes"); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const quiet = async () => { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + }; + try { + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => canRun).ready; + await settle(); await quiet(); + if (mode === "drain") canRun = false; + if (mode === "stopped") await coordinator.stop(); + if (mode === "failed selection") selection.mockRejectedValueOnce(new Error("temporary retry selection failure")); + expect(await context.admit()).toMatchObject({ status: mode === "disabled" ? "failed" : "deferred" }); + if (mode === "failed selection") { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); }); + canRun = false; + } + if (["drain", "failed selection", "stopped"].includes(mode)) { + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await quiet(); canRun = true; + if (mode === "stopped") { + await context.service.processPublicationMaintenance(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => canRun).ready; + } + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + payload: mode === "disabled" + ? { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" } + : { text: "Your retry is queued.", progressState: "queued" }, + })]); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + await quiet(); + expect(errors).toEqual(mode === "failed selection" ? [expect.objectContaining({ errors: [expect.objectContaining({ message: "temporary retry selection failure" })] })] : []); + } finally { + await coordinator.stop(); selection.mockRestore(); vi.useRealTimers(); await context.service.shutdown(); + await db.delete(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, context.fixture.companyId), + inArray(agentWakeupRequests.id, [context.retry.requestId!, context.retry.chatActionId!]), isNull(agentWakeupRequests.runId))); + } + }); + + it.each(["idle", "bounded pass", "requester merge"] as const)("reactivates denied retry notices when authority is restored during %s", async mode => { + const context = await deferredFailedRetryNoticeFixture(); + const rejectedIds = mode === "bounded pass" ? Array.from({ length: 100 }, () => randomUUID()) : []; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + let held = false, release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + let ready: Promise | undefined; + const reconcile = context.service.reconcileFailedRetryNotices.bind(context.service); + const pass = vi.spyOn(context.service, "reconcileFailedRetryNotices"); + try { + expect(await context.admit()).toMatchObject({ status: "deferred" }); + const [originalReceipt] = await readAgentWakeupRequestsById(context.retry.chatActionId!); + if (mode === "requester merge") { + expect(originalReceipt.requestedByActorType).toBe("user"); + await db.update(agentWakeupRequests).set({ requestedByActorId: "another-requester" }) + .where(eq(agentWakeupRequests.id, originalReceipt.id)); + } else { + await db.update(issues).set({ visibility: "private", privacyRootIssueId: context.issue.id, + createdByUserId: null, responsibleUserId: null, assigneeUserId: null, + }).where(eq(issues.id, context.issue.id)); + } + if (rejectedIds.length) { + const [action] = await db.select().from(chatActions).where(eq(chatActions.id, context.retry.chatActionId!)); + const [receipt] = await readAgentWakeupRequestsById(action.id); + await db.insert(chatActions).values(rejectedIds.map(id => ({ ...action, id, + providerActionId: `retry-cursor:${id}`, createdAt: new Date(action.createdAt.getTime() + 1000), + payload: { ...action.payload, sourceScopeSha256: "invalid-source" }, + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ ...receipt, id, + idempotencyKey: `retry-cursor:${id}`, + }))); + pass.mockImplementationOnce(async input => { + const result = await reconcile(input); + expect(result).toMatchObject({ inserted: 0, nextCursor: expect.any(Object) }); + held = true; await pending; return result; + }); + } + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + ready = chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready; + if (mode === "bounded pass") await vi.waitFor(() => expect(held).toBe(true), { timeout: 10_000 }); + else { + await ready; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + } + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + if (mode === "requester merge") { + await db.transaction(async tx => { + await tx.execute(sql`select id from issues where id = ${context.issue.id} for update`); + await createWakeAdmissionWriter().mergeIntoExistingDeferredWake( + createAdmissionTransactionScope(context.fixture.companyId, tx as unknown as Db), { + companyId: context.fixture.companyId, existingDeferredWakeId: originalReceipt.id, + mergedPayload: originalReceipt.payload ?? {}, nextCoalescedCount: 1, + manualUserWakeActorId: originalReceipt.requestedByActorId!, + }); + }); + } else await issueService(db).update(context.issue.id, { visibility: "open" }); + release(); await ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + expect(errors).toEqual([]); + } finally { + release(); await ready; await coordinator.stop(); pass.mockRestore(); vi.useRealTimers(); + await context.service.shutdown(); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, + [...rejectedIds, context.retry.requestId!, context.retry.chatActionId!])); + if (rejectedIds.length) await db.delete(chatActions).where(inArray(chatActions.id, rejectedIds)); + } + }); + + it("wakes a not-started retry notice after resumed admission is denied", async () => { + const context = await deferredFailedRetryNoticeFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + try { + expect(await context.admit()).toMatchObject({ status: "deferred" }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }); + await reconcileTaskChatRetryReceipts(db); + context.wakeup.mockRejectedValueOnce(Object.assign(new Error("denied at resume"), { + details: { code: "chat_failed_run_retry_not_authorized" }, + })); + expect(await context.admit()).toMatchObject({ status: "failed" }); + expect((await readAgentWakeupRequestsById(context.retry.chatActionId!))[0].status).toBe("failed"); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(2); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); + await db.delete(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, context.fixture.companyId), + inArray(agentWakeupRequests.id, [context.retry.requestId!, context.retry.chatActionId!]), isNull(agentWakeupRequests.runId))); + } + }); + + it.each(["agent", "tree"] as const)("wakes a failed retry notice only after %s cancellation commits", async kind => { + const context = await failedChatRetryFixture("telegram"); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + let stagedActionId: string | undefined; + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const cancel = (database: typeof db) => kind === "agent" + ? heartbeatService(database).cancelInvocationsForAgents([context.fixture.assignedAgentId], "Cancelled retry fixture") + : issueTreeControlService(database).cancelUnclaimedWakeupsForTree(context.fixture.companyId, context.issue.id, "Cancelled retry fixture"); + try { + const staged = await db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, { + companyId: context.fixture.companyId, issueId: context.issue.id, + agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user", + })); + stagedActionId = staged.actionId; + await context.service.processFailedChatRunRetry(staged.actionId); + await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }) + .where(eq(agentWakeupRequests.id, staged.actionId)); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + const rollback = new Error("rollback retry cancellation"); + await expect(db.transaction(async tx => { await cancel(tx as unknown as typeof db); throw rollback; })).rejects.toBe(rollback); + await settle(); + expect((await readAgentWakeupRequestsById(staged.actionId))[0].status).toBe("deferred_issue_execution"); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + await db.transaction(async tx => { await cancel(tx as unknown as typeof db); }); + await settle(); + expect((await readAgentWakeupRequestsById(staged.actionId))[0].status).toBe("cancelled"); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" }, + })])); + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); + if (stagedActionId) { + await db.delete(chatActions).where(eq(chatActions.id, stagedActionId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, stagedActionId)); + } + } + }); + it.each(["manual", "scoped"] as const)("advances failed retry notices past denied history with microsecond timestamps (%s)", async mode => { const context = await failedChatRetryFixture("telegram"); const rejectedIds = Array.from({ length: 50 }, () => randomUUID()); diff --git a/server/src/app.ts b/server/src/app.ts index 8c3ae2daf67..a9212bdd97a 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -197,6 +197,7 @@ import { chatChannelService } from "./services/chat-channels.js"; import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js"; import { nativeChatProgressWork } from "./services/chat-native-progress-work.js"; import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-work.js"; +import { chatFailedRetryNoticeWork } from "./services/chat-failed-retry-notice-work.js"; import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js"; import { createCoalescedAsyncTrigger, @@ -1290,6 +1291,7 @@ export async function createApp( chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatFailedRetryNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatRunMilestoneWork(db, opts.authPublicBaseUrl).register( deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(), ); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 388afd9cf61..1e1f5846188 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -995,6 +995,7 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { async mergeIntoExistingDeferredWake(scope, input) { const tx = requireAdmissionTx(scope, input.companyId); + if (input.manualUserWakeActorId) await notifyChatInboundWakeNotices(tx, { companyId: input.companyId }); const rows = await tx .update(agentWakeupRequests) .set({ diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 8ed25ed01bd..17b456ed421 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -3354,9 +3354,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPublicationMaintenance(limit = 25) { scheduleGitHubMaintenance(); - // Normal inbound notices are commit-driven. Failed-run retries retain their - // separate maintenance owner until its own producer audit is complete. - await enqueueFailedChatRetryPublications(50); await reconcileTerminalConfirmationActions(limit); await processPendingInteractionWakeups(limit); } diff --git a/server/src/services/chat-channels/failed-retry-notices.ts b/server/src/services/chat-channels/failed-retry-notices.ts index 493c3e016e3..cf2068079d2 100644 --- a/server/src/services/chat-channels/failed-retry-notices.ts +++ b/server/src/services/chat-channels/failed-retry-notices.ts @@ -17,8 +17,6 @@ type Options = { isShuttingDown: () => boolean; }; -/** Retry authorization belongs to the retry service. This projection owns only - * the queued/not-started notice, selected from its existing durable receipt. */ export function createFailedRetryNotices(db: Db, options: Options) { let manualCursor: FailedRetryNoticeCursor | null = null; async function candidates(limit: number, cursor: FailedRetryNoticeCursor | null, scope?: FailedRetryNoticeScope) { diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index aa434b1893e..fb219c6cbb1 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 { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { createFailedRetryNotices } from "./failed-retry-notices.js"; import { continuingTeamsFilePublication, type ContinuingTeamsFileTransfer } from "./microsoft-teams/file-transfer-source.js"; import { readTaskChatRetryIntent, taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./task-retry-authority.js"; @@ -1647,6 +1648,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti typeof claimed.payload.taskRetryIntentId === "string" ? claimed.payload.taskRetryIntentId : undefined, true, // The durable wake callback validates admission; later phases recheck containment. ); + await notifyChatInboundWakeNotices(tx, { companyId: scope.companyId, issueId: scope.issueId }); }, }), }); @@ -1684,12 +1686,17 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti receipt.payload?.taskRetryRequestId === claimed.payload.taskRetryIntentId) { // Receipt existence proves the earlier wait, not current authority. // Retire only this unadmitted wake; an admitted run keeps its history. - await db.update(agentWakeupRequests).set({ - status: "failed", error: "chat_failed_run_retry_not_authorized", - finishedAt: new Date(), updatedAt: new Date(), - }).where(and(eq(agentWakeupRequests.id, receipt.id), - eq(agentWakeupRequests.companyId, claimed.companyId), - eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId))); + await db.transaction(async tx => { + await notifyChatInboundWakeNotices(tx, { + companyId: claimed.companyId, issueId: String(claimed.payload.issueId), + }); + await tx.update(agentWakeupRequests).set({ + status: "failed", error: "chat_failed_run_retry_not_authorized", + finishedAt: new Date(), updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.id, receipt.id), + eq(agentWakeupRequests.companyId, claimed.companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId))); + }); return settle("failed", { code: "failed_chat_retry_not_authorized", attemptCount }); } return settle(["skipped", "failed", "cancelled"].includes(receipt.status) ? "failed" : "processed", { diff --git a/server/src/services/chat-failed-retry-notice-work.ts b/server/src/services/chat-failed-retry-notice-work.ts new file mode 100644 index 00000000000..c9075eb294c --- /dev/null +++ b/server/src/services/chat-failed-retry-notice-work.ts @@ -0,0 +1,39 @@ +import type { ChatChannelService } from "./chat-channels.js"; +import type { FailedRetryNoticeCursor, FailedRetryNoticeScope } from "./chat-channels/failed-retry-notices.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; + +type Entry = { + id: string; companyId: string; issueId: string; dueAt: number; + cursor: FailedRetryNoticeCursor | null; contended: boolean; +}; +const batchSize = 100; + +export function chatFailedRetryNoticeWork(service: Pick) { + async function load(scope?: FailedRetryNoticeScope, signal?: AbortSignal): Promise { + return (await service.loadFailedRetryNoticeScopes(scope, signal)).map(row => ({ + ...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0, cursor: null, contended: false, + })); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + const result = await service.reconcileFailedRetryNotices({ + scope: { companyId: entry.companyId, issueId: entry.issueId }, + limit: batchSize, cursor: entry.cursor, signal, + }); + const contended = entry.contended || result.contended; + if (result.nextCursor) return { ...entry, cursor: result.nextCursor, contended, dueAt: Date.now() }; + return contended ? { ...entry, cursor: null, contended: false, dueAt: Date.now() + 1000 } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatFailedRetryNotices, + decodeScope: (hint: string): FailedRetryNoticeScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: FailedRetryNoticeScope) => !scope || + (entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId)), + load, check, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/chat-inbound-wake-notice-notifications.ts b/server/src/services/chat-inbound-wake-notice-notifications.ts index 12e0aa08422..6389ffe8755 100644 --- a/server/src/services/chat-inbound-wake-notice-notifications.ts +++ b/server/src/services/chat-inbound-wake-notice-notifications.ts @@ -1,7 +1,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import type { InboundWakeNoticeScope } from "./chat-channels/inbound/wakeup-notices.js"; -export function notifyChatInboundWakeNotices(tx: object, scope: InboundWakeNoticeScope) { - return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices, - JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +export async function notifyChatInboundWakeNotices(tx: object, scope: InboundWakeNoticeScope) { + const hint = JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices, hint); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatFailedRetryNotices, hint); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 3ce6f0b905b..6e4deaebbab 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -56,6 +56,7 @@ export const DELIVERY_QUEUES = { chatPublications: "chat-publications", chatNativeProgress: "chat-native-progress", chatInboundWakeNotices: "chat-inbound-wake-notices", + chatFailedRetryNotices: "chat-failed-retry-notices", chatRunMilestones: "chat-run-milestones", chatReceipts: "chat-slack-receipts", browser: "browser-use", diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 68532b79765..c64049129f9 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { authorizationService } from "../authorization.js"; import { assertTaskRetryConversationAllowed, readTaskRetryIntent, retireSupersededTaskRetry } from "../task-retry.js"; @@ -2462,12 +2463,24 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const waitIssueId = issueId; return db.transaction(async (tx) => { await tx.execute(sql`select id from issues where id = ${waitIssueId} and company_id = ${agent.companyId} for update`); + if (durableRequest) await notifyChatInboundWakeNotices(tx, { + companyId: durableRequest.companyId, issueId: durableRequest.issueId, + }); return recordExecutionWait(tx as unknown as Db, { issueId: waitIssueId, request, condition: waitCondition, coalesce: coalesceExecutionWait, }); }); } - await db.insert(agentWakeupRequests).values(request); + if (durableRequest) { + await db.transaction(async tx => { + await notifyChatInboundWakeNotices(tx, { + companyId: durableRequest.companyId, issueId: durableRequest.issueId, + }); + await tx.insert(agentWakeupRequests).values(request); + }); + } else { + await db.insert(agentWakeupRequests).values(request); + } return { created: true }; }; const writeSkippedHeartbeatRequest = async (