From aeaaed7b3e17e1377433165a38fef558acacb631 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 14:52:52 -0500 Subject: [PATCH] fix(server): wake terminal confirmation recovery from committed changes Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 14 ++ ...10-25-terminal-confirmation-sleep-audit.md | 108 ++++++++-- .../app-private-hostname-gate.test.ts | 2 + .../chat-channels.integration.test.ts | 204 +++++++++++++++++- server/src/app.ts | 1 + server/src/services/chat-channels.ts | 4 +- .../chat-channels/interactions/callbacks.ts | 4 +- .../telegram/confirmation-actions.ts | 55 ++++- .../chat-confirmation-notifications.ts | 8 + .../chat-run-milestone-notifications.ts | 2 + .../services/delivery-work-notifications.ts | 1 + 11 files changed, 375 insertions(+), 28 deletions(-) create mode 100644 server/src/services/chat-confirmation-notifications.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 f971fa1ff56..6170fb203df 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1821,3 +1821,17 @@ implementation or PR yet. Older #17490/#17572 CI failures, #17576 inventory amplification findings, #17580 offboarding/cleanup ownership findings, the entire remaining original list, final master reconciliation and all final-head checks remain required. All stack PRs remain unmerged. + + +## Slice 32 implementation checkpoint + +`codex/sleep-32-terminal-confirmations` implements terminal-confirmation recovery +on committed signals and the shared scoped coordinator. Its Telegram owner keeps +payload validation, settlement, continuation staging and activity attribution. +The audit is `2026-10-25-terminal-confirmation-sleep-audit.md`. Nine distinct focused +cases pass across the recorded commands, including real callback failure, outer +commit/rollback, historical waiting/invalid prefixes, and a write during an active +settlement. The new source guard excludes this duty from periodic maintenance. +Full local acceptance, remote review/checks and publication are pending here. +This supersedes the audit-only slice-32 status above. All older stack gates and +every remaining original obligation remain required; no PR is merged. diff --git a/doc/plans/2026-10-25-terminal-confirmation-sleep-audit.md b/doc/plans/2026-10-25-terminal-confirmation-sleep-audit.md index caed3ad8d3c..129e0bbfa72 100644 --- a/doc/plans/2026-10-25-terminal-confirmation-sleep-audit.md +++ b/doc/plans/2026-10-25-terminal-confirmation-sleep-audit.md @@ -3,7 +3,7 @@ Date: 2026-10-25. Branch: `codex/sleep-32-terminal-confirmations`. Parent: [#17649](https://github.com/paperclipai/paperclip/pull/17649) at `2384384eeec4f32d073242ebf597b81df2df1ccb`. -Status: owner audit only; no worker migration or PR yet. +Status: implemented; focused lifecycle acceptance passes. Full acceptance and remote gates are pending. ## Obligation and existing boundary @@ -31,20 +31,20 @@ continuation delivery and GitHub maintenance remain separate reasons/PRs. ## Producer inventory -- [ ] Acceptance callback: notify before its processing-action write inside the +- [x] Acceptance callback: notify before its processing-action write inside the existing `afterResolveInTransaction` owner. -- [ ] Rejection callback: same contract, separately exercise the real path. -- [ ] Initial action token creation in `chat-interaction-publications.ts` writes +- [x] Rejection callback: same contract, separately exercise the real path. +- [x] Initial action token creation in `chat-interaction-publications.ts` writes `issued`; a user still deciding does not owe settlement work. It must be quiet. -- [ ] Terminal/supersession publication helpers expire issued sibling tokens; +- [x] Terminal/supersession publication helpers expire issued sibling tokens; inspect their exact predicates before treating them as processing producers. -- [ ] Settlement writers move processing actions to processed/failed/expired. +- [x] Settlement writers move processing actions to processed/failed/expired. Preserve publication notifications, activity provenance and continuation order. -- [ ] Board-resolution compatibility: current integration coverage fabricates a +- [x] Board-resolution compatibility: current integration coverage fabricates a processing action before the board decision. Establish how historical pending processing rows reactivate from a later interaction change, without repeatedly scanning an unchanged human-wait state. -- [ ] Startup reconstructs processing debt. Outer rollback creates no actionable +- [x] Startup reconstructs processing debt. Outer rollback creates no actionable debt, ambiguous commit uses the existing transaction signal protocol, and first-write drain/resume retains the wake. @@ -81,21 +81,22 @@ one-use token behavior, redelivery deduplication, board-race provenance and failed continuation handling. Reuse its fixture and assertions rather than building a parallel authorization fixture. -- [ ] Reproduce missed recovery after the real postcommit callback failure when +- [x] Reproduce missed recovery after the real postcommit callback failure when periodic/manual maintenance is not invoked. -- [ ] Prove committed acceptance and rejection activate recovery from idle. -- [ ] Prove first-write drain/resume, stop/restart, failed selection/retry and +- [x] Prove committed acceptance and rejection activate recovery from idle. +- [x] Prove first-write drain/resume, stop/restart, failed selection/retry and writes during a drain/sweep. -- [ ] Verify outer rollback/commit through actual producer ownership where the +- [x] Verify outer rollback/commit through actual producer ownership where the callback service accepts a caller-owned transaction. -- [ ] Prove empty state, issued tokens awaiting a person, and completed history +- [x] Prove empty state, issued tokens awaiting a person, and completed history make no recurring SQL after settlement. -- [ ] Prove bounded progress behind pending/malformed/contended candidate prefixes. -- [ ] Retain the existing terminal edit-before-continuation behavior, duplicate +- [x] Prove bounded progress behind pending/malformed/contended candidate prefixes. +- [x] Retain the existing terminal edit-before-continuation behavior, duplicate callback handling and board-resolution attribution. -- [ ] Register production work and remove only its periodic maintenance call. -- [ ] Update startup/action-worker tests and run targeted PostgreSQL acceptance, - server types/build, required full acceptance and fresh remote review/checks. +- [x] Register production work and remove only its periodic maintenance call. +- [ ] Complete final server/workspace types/build, full chat/startup acceptance, + and fresh remote review/checks. Production wiring has a source guard; the + existing generic action-worker registry is unchanged. ## Scope remains open @@ -103,3 +104,74 @@ No Cloud wake gate, transport ownership, provider deadline, or sleep eligibility changes are implied. The supported process-local writer model remains explicit. Every remaining original recurring-server reason and older stack finding stays in the ledger; this audit does not reduce the requested scope. + +## Implementation and boundary decisions + +The Telegram owner now exposes only `registerTerminalConfirmationWork` through +its existing callback/service wiring. Shared app startup selects this owner; it +has no Telegram payload or decision policy. The owner reconstructs processing +action IDs from existing rows, using 100-row UUID-keyset pages, and registers +per-action entries with `registerScopedDeadlineWork`. The shared worker attempts +at most 25 entries per sweep, rotates companies, retains commits arriving during +an attempt, and backs off query/settlement failures from 30 seconds to five +minutes. A failed entry does not prevent later entries from progressing. + +Successful settlement, already terminal/missing actions, invalid actions that +receive a failure record, and actions still waiting on a pending interaction +remove their scheduled entry. The latter retain their durable processing row; +they wake when the interaction changes. Issued tokens and completed history do +not enter recovery. Missing conversation IDs now receive the existing +`confirmation_action_binding_invalid` failure classification instead of remaining +processing forever. Other invalid-binding and authorization policy stays with the +existing settler. + +Both real callback decision transactions await a confirmation intent before +writing processing state. The existing issue/company milestone notification +boundary also emits an identifier-only confirmation hint. This covers board +accept/reject, expiry, withdrawal, supersession and other supported interaction +changes without adding wrappers to each writer. It also means unrelated milestone +changes can cause one scoped reconstruction; an issue hint never restarts a global +inventory. The company/issue query includes orphaned conversation bindings in the +same company so invalid work can receive an explicit failure record. Company-wide +hints and startup reconstruct the full matching processing inventory in bounded +SQL pages; total enumeration and memory are proportional to that inventory, not +capped at 100 rows. No claim of constant startup cost is made. + +Only terminal-confirmation reconciliation was removed from periodic publication +maintenance. Explicit `processPendingPublications` maintenance still calls the +existing manual reconciler. Interaction continuation delivery and GitHub duties +are unchanged and remain separate stack reasons. The host sleep gate is unchanged, +including its treatment of durable processing debt. Process-local signals cover +supported writes in this process; independent SQL writers/other replicas require +an ingress protocol. Startup reconstruction remains the recovery path after a +process loss. + +## Evidence and retained failures + +- `origin/master` refreshed to `1949e726f8515127a04314e3793c427ccfc33c1e`; + full-stack reconciliation is still outstanding. The exact-obligation GitHub + search found no competing terminal-confirmation migration. +- `/tmp/sleep32-confirmation-before.log` is an invalid test invocation: the root + Vitest project filter `server` did not match a named project. No tests ran. +- `/tmp/sleep32-confirmation-before-valid.log` uses the server configuration and + reproduces the real callback gap: a committed action remains processing instead + of reaching processed when the new worker has no producer notification. +- `/tmp/sleep32-confirmation-lifecycle.log` initially passed accept, reject and + restart, but drain and failed-selection tests had not advanced to the retained + 30-second deadline. A failed spy assertion also leaked an unconsumed injected + error into the following legacy test. The tests now advance exactly to the + registered deadline and always restore the spy. The worker was not changed to + conceal those harness failures. +- `/tmp/sleep32-confirmation-lifecycle-fixed.log`: all six cases pass, including + the existing provider edit-before-continuation and duplicate callback journey. +- `/tmp/sleep32-confirmation-atomic.log`: eight focused cases pass, including the + startup guard and a real outer commit/rollback of the board-resolution owner. + The test passes 101 pending and 101 malformed actions, verifies failure evidence, + parks human-wait debt, and observes no fabricated provider-attributed activity. +- `/tmp/sleep32-confirmation-inflight.log`: a real PostgreSQL row lock holds one + settlement while a second real callback commits. Releasing it settles both + actions and returns the worker to quiet. This is a ninth distinct focused case. +- `/tmp/sleep32-types.log`: server typecheck passed before the final in-flight + concurrency test was added. A final typecheck is still required. +- Quiet assertions advance 120 seconds and count root database select/transaction + calls after startup and settlement. They do not establish provider suspension. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 1dad4ca398a..629bd1139f6 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -92,6 +92,8 @@ describe("createChatReconciliationCoordinator", () => { chat.indexOf("async function schedulePendingPublications(")); expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications("); expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications("); + expect(publicationMaintenance).not.toContain("reconcileTerminalConfirmationActions("); + expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,"); expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,"); }); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index e9adaddc0ab..6c5fd9a18eb 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -44819,7 +44819,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(stillPending.status).toBe("pending"); }); - it("renders safe Telegram confirmations as opaque one-use buttons and wakes the resolved task", async () => { + async function telegramConfirmationRecoveryFixture(decision: "accept" | "reject" = "accept") { const fixture = await seedCompany(); let injectSettlementCrash = true; const settlementBarrier = vi.fn(async () => { @@ -44977,7 +44977,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ).toBe(true); const acceptAction = confirmationActions.find( - (action) => action.payload.decision === "accept", + (action) => action.payload.decision === decision, ); if (!acceptAction) throw new Error("Telegram accept action was not persisted"); @@ -45009,7 +45009,203 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { value: undefined, }, }; - await expect(callbacks.onAction(actionEvent)).rejects.toThrow( + return { fixture, callbacks, onAction: callbacks.onAction, endpoint, service, wakeup, settlementBarrier, dm, conversation, linkedUserId, principal, interaction, publication, acceptAction, actionEvent }; + } + + it.each(["accept", "reject", "drain", "restart", "failed selection"] as const)("recovers committed terminal confirmations without polling through %s", async mode => { + const context = await telegramConfirmationRecoveryFixture(mode === "reject" ? "reject" : "accept"); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + 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 context.service.registerTerminalConfirmationWork(coordinator, () => canRun).ready; + await settle(); await quiet(); + if (["drain", "failed selection"].includes(mode)) canRun = false; + if (mode === "restart") await coordinator.stop(); + await expect(context.onAction(context.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + if (["drain", "failed selection", "restart"].includes(mode)) { + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, context.acceptAction.id))).toEqual([{ status: "processing" }]); + await quiet(); + if (mode === "failed selection") { + const select = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("temporary confirmation selection failure"); }); + try { + canRun = true; + await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(errors).toHaveLength(1); }); + canRun = false; + } finally { select.mockRestore(); } + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + await quiet(); + } + canRun = true; + if (coordinator.nextWakeAt() !== null) await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + await context.service.registerTerminalConfirmationWork(coordinator, () => canRun).ready; + } + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, context.acceptAction.id))).toEqual([{ status: "processed" }]); + }, { timeout: 10_000 }); + await settle(); await quiet(); + expect(await db.select({ id: chatActions.id }).from(chatActions).where(and( + eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"), + eq(sql`${chatActions.payload}->>'interactionId'`, context.interaction.id), + ))).toHaveLength(1); + expect(await db.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.companyId, context.fixture.companyId), + eq(sql`${activityLog.details}->>'interactionId'`, context.interaction.id), + eq(sql`${activityLog.details}->>'source'`, "external_chat"), + ))).toHaveLength(1); + expect(errors).toEqual(mode === "failed selection" ? [expect.objectContaining({ + errors: [expect.objectContaining({ message: "temporary confirmation selection failure" })], + })] : []); + } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("parks pending confirmation history, passes malformed prefixes, and wakes on the outer interaction commit", async () => { + const waiting = await telegramConfirmationRecoveryFixture(); + const terminal = await telegramConfirmationRecoveryFixture(); + const pendingIds = Array.from({ length: 101 }, (_, index) => `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}`); + const invalidIds = Array.from({ length: 101 }, (_, index) => `00000000-0000-4000-8001-${String(index + 1).padStart(12, "0")}`); + await db.insert(chatActions).values([...pendingIds, ...invalidIds].map(id => ({ + ...waiting.acceptAction, id, providerActionId: `confirmation-history:${id}`, status: "processing", + conversationId: invalidIds.includes(id) ? null : waiting.conversation.id, result: null, + }))); + await expect(terminal.onAction(terminal.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 30_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 waiting.service.registerTerminalConfirmationWork(coordinator, () => true).ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, terminal.acceptAction.id))).toEqual([{ status: "processed" }]); + }, { timeout: 10_000 }); + await settle(); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(101); + expect(await db.select({ id: chatActions.id }).from(chatActions).where(and( + inArray(chatActions.id, invalidIds), eq(chatActions.status, "failed"), + eq(sql`${chatActions.result}->>'code'`, "confirmation_action_binding_invalid"), + ))).toHaveLength(101); + await quiet(); + const [issue] = await db.select().from(issues).where(eq(issues.id, waiting.conversation.issueId)); + await expect(db.transaction(async tx => { + await issueThreadInteractionService(tx as unknown as Db).acceptInteraction(issue, waiting.interaction.id, {}, { userId: waiting.linkedUserId }); + expect(await db.select({ status: issueThreadInteractions.status }).from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, waiting.interaction.id))).toEqual([{ status: "pending" }]); + throw new Error("rollback outer confirmation decision"); + })).rejects.toThrow("rollback outer confirmation decision"); + await settle(); await quiet(); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(101); + await db.transaction(async tx => { + await issueThreadInteractionService(tx as unknown as Db).acceptInteraction(issue, waiting.interaction.id, {}, { userId: waiting.linkedUserId }); + expect(await db.select({ status: issueThreadInteractions.status }).from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, waiting.interaction.id))).toEqual([{ status: "pending" }]); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(101); + }); + await settle(); await quiet(); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(0); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processed")))).toHaveLength(1); + expect(await db.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.companyId, waiting.fixture.companyId), + eq(sql`${activityLog.details}->>'interactionId'`, waiting.interaction.id), + eq(sql`${activityLog.details}->>'source'`, "external_chat"), + ))).toHaveLength(0); + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); vi.useRealTimers(); + await db.delete(chatActions).where(inArray(chatActions.id, [...pendingIds, ...invalidIds])); + } + }); + + it("retains a confirmation committed while another settlement is in flight", async () => { + const first = await telegramConfirmationRecoveryFixture(); + const second = await telegramConfirmationRecoveryFixture(); + await expect(first.onAction(first.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + let locked!: () => void, release!: () => void; + const lockReady = new Promise(resolve => { locked = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const transaction = db.transaction(async tx => { + await tx.select({ id: chatActions.id }).from(chatActions).where(eq(chatActions.id, first.acceptAction.id)).for("update"); + locked(); await held; + }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + let ready: Promise | undefined; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await lockReady; + ready = first.service.registerTerminalConfirmationWork(coordinator, () => true).ready; + await vi.waitFor(async () => { + const blocked = await db.execute(sql`select pid from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + and query like 'update "chat_actions"%'`); + expect(blocked.length).toBeGreaterThan(0); + }, { timeout: 10_000 }); + await expect(second.onAction(second.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, second.acceptAction.id))).toEqual([{ status: "processing" }]); + release(); await transaction; await ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(inArray(chatActions.id, [first.acceptAction.id, second.acceptAction.id]))) + .toEqual([{ status: "processed" }, { status: "processed" }]); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const select = vi.spyOn(db, "select"), queryTransaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(queryTransaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); queryTransaction.mockRestore(); } + expect(errors).toEqual([]); + } finally { + release(); await transaction; await ready; await coordinator.stop(); vi.useRealTimers(); + } + }); + + it("renders safe Telegram confirmations as opaque one-use buttons and wakes the resolved task", async () => { + const { fixture, onAction, endpoint, service, wakeup, settlementBarrier, dm, conversation, linkedUserId, principal, interaction, publication, acceptAction, actionEvent } = await telegramConfirmationRecoveryFixture(); + await expect(onAction(actionEvent)).rejects.toThrow( "injected post-interaction confirmation settlement crash", ); expect(settlementBarrier).toHaveBeenCalledTimes(1); @@ -45188,7 +45384,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ) .then((rows) => rows.length); - await callbacks.onAction(actionEvent); + await onAction(actionEvent); await new Promise((resolve) => setImmediate(resolve)); expect(dm.post).toHaveBeenCalledTimes(providerPostCountBeforeRedelivery); await expect( diff --git a/server/src/app.ts b/server/src/app.ts index a9212bdd97a..db44966bbfa 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1288,6 +1288,7 @@ export async function createApp( chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 17b456ed421..645d66461b9 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -3354,7 +3354,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPublicationMaintenance(limit = 25) { scheduleGitHubMaintenance(); - await reconcileTerminalConfirmationActions(limit); await processPendingInteractionWakeups(limit); } @@ -3483,7 +3482,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { processPendingInteractionWakeups } = createChatInteractionContinuations(db, { redactError, CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS, get heartbeat() { return options.heartbeat; }, }); - const { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions } = createChatInteractionCallbacks(db, { + const { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions, registerTerminalConfirmationWork } = createChatInteractionCallbacks(db, { CAPABILITIES, MAX_ERROR_TEXT, actionThreadMatchesConversation, nonDirectDestinationAllowed, sameTeamsThreadIdentity, ensurePrincipal, endpointRecord, runtimeCallbackRecord, runtimeCallbackEndpoint, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, @@ -3506,6 +3505,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + registerTerminalConfirmationWork, loadFailedRetryNoticeScopes, reconcileFailedRetryNotices, registerTeamsAttemptRecoveryWork, diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index 69e29c0e049..71c260a637e 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -139,7 +139,7 @@ type InteractionCallbacksOptions = { export function createChatInteractionCallbacks(db: Db, options: InteractionCallbacksOptions) { const { CAPABILITIES, MAX_ERROR_TEXT, actionThreadMatchesConversation, nonDirectDestinationAllowed, sameTeamsThreadIdentity, ensurePrincipal, endpointRecord, runtimeCallbackRecord, runtimeCallbackEndpoint, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, recordCurrentMicrosoftTeamsRoute, lockTelegramCallbackNoticeAuthority, stageProviderEffect, scheduleProviderEffect, invalidateRuntime, redactError, persistence, questionResponses, scheduleMessageProcessing, processPendingPublications } = options; const { handleDiscordFormCorrection, acceptDiscordQuestionForm } = createDiscordQuestionForms(db, { requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, persistence }); - const { reconcileTerminalConfirmationActions, handleTelegramConfirmation } = createTelegramConfirmationActions(db, { + const { reconcileTerminalConfirmationActions, handleTelegramConfirmation, registerTerminalConfirmationWork } = createTelegramConfirmationActions(db, { requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, scheduleMessageProcessing, processPendingPublications, get confirmationResolutionPersistBarrier() { return options.confirmationResolutionPersistBarrier; }, }); @@ -2011,5 +2011,5 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb }); return acceptedResponse(); } - return { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions }; + return { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions, registerTerminalConfirmationWork }; } diff --git a/server/src/services/chat-channels/telegram/confirmation-actions.ts b/server/src/services/chat-channels/telegram/confirmation-actions.ts index 8e5c9971b8f..8cef42250c3 100644 --- a/server/src/services/chat-channels/telegram/confirmation-actions.ts +++ b/server/src/services/chat-channels/telegram/confirmation-actions.ts @@ -1,11 +1,19 @@ import type { DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import { notifyChatConfirmationWork, type ChatConfirmationScope } from "../../chat-confirmation-notifications.js"; + import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; import { and, asc, eq, + gt, + isNull, + or, inArray, ne, sql, @@ -152,6 +160,7 @@ export function createTelegramConfirmationActions(db: Db, options: TelegramConfi }); }, afterResolveInTransaction: async (tx, resolved) => { + await notifyChatConfirmationWork(tx, { companyId: action.companyId, issueId: conversation.issueId }); const [owned] = await tx .update(chatActions) .set({ @@ -202,6 +211,7 @@ export function createTelegramConfirmationActions(db: Db, options: TelegramConfi }); }, afterResolveInTransaction: async (tx, resolved) => { + await notifyChatConfirmationWork(tx, { companyId: action.companyId, issueId: conversation.issueId }); const [owned] = await tx .update(chatActions) .set({ @@ -293,8 +303,13 @@ export function createTelegramConfirmationActions(db: Db, options: TelegramConfi ) .then((rows) => rows[0] ?? null); if (action?.status === "processed") return true; - if (!action || action.status !== "processing" || !action.conversationId) + if (!action || action.status !== "processing") return false; + if (!action.conversationId) { + await db.update(chatActions).set({ + status: "failed", result: { code: "confirmation_action_binding_invalid" }, updatedAt: new Date(), + }).where(and(eq(chatActions.id, action.id), eq(chatActions.status, "processing"))); return false; + } const token = action.payload; if ( token.version !== 1 || @@ -652,5 +667,41 @@ export function createTelegramConfirmationActions(db: Db, options: TelegramConfi } return settled; } - return { settleTerminalConfirmationAction, reconcileTerminalConfirmationActions, handleTelegramConfirmation }; + function registerTerminalConfirmationWork( + coordinator: ReturnType, + canRun: () => boolean, + ) { + type Entry = { id: string; companyId: string; issueId: string | null; dueAt: number }; + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatTerminalConfirmations, + decodeScope: (hint: string): ChatConfirmationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: ChatConfirmationScope) => !scope || + (entry.companyId === scope.companyId && (!scope.issueId || !entry.issueId || entry.issueId === scope.issueId)), + load: async (scope?: ChatConfirmationScope, signal?: AbortSignal): Promise => { + const entries: Entry[] = []; + let cursor: string | undefined; + while (!signal?.aborted) { + const rows = await db.select({ + id: chatActions.id, companyId: chatActions.companyId, issueId: chatConversations.issueId, + }).from(chatActions).leftJoin(chatConversations, and( + eq(chatConversations.companyId, chatActions.companyId), eq(chatConversations.id, chatActions.conversationId), + )).where(and( + eq(chatActions.kind, "confirmation_response"), eq(chatActions.status, "processing"), + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? or(eq(chatConversations.issueId, scope.issueId), isNull(chatConversations.id)) : undefined, + cursor ? gt(chatActions.id, cursor) : undefined, + )).orderBy(asc(chatActions.id)).limit(100); + entries.push(...rows.map(row => ({ ...row, dueAt: 0 }))); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + throw signal?.reason ?? new Error("Confirmation recovery stopped"); + }, + check: async (entry: Entry) => { + await settleTerminalConfirmationAction(entry.id); + return null; + }, + }, canRun); + } + return { settleTerminalConfirmationAction, reconcileTerminalConfirmationActions, handleTelegramConfirmation, registerTerminalConfirmationWork }; } diff --git a/server/src/services/chat-confirmation-notifications.ts b/server/src/services/chat-confirmation-notifications.ts new file mode 100644 index 00000000000..90bfb688f49 --- /dev/null +++ b/server/src/services/chat-confirmation-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type ChatConfirmationScope = { companyId: string; issueId?: string }; + +export async function notifyChatConfirmationWork(tx: object, scope: ChatConfirmationScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTerminalConfirmations, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +} diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 6964bd6b069..972d6789bb7 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,3 +1,4 @@ +import { notifyChatConfirmationWork } from "./chat-confirmation-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -13,4 +14,5 @@ export async function notifyChatRunMilestones(tx: object, scope: ChatRunMileston await notifyChatNativeProgress(tx, scope); // Admission and current chat authority also govern saved queue notices. await notifyChatInboundWakeNotices(tx, scope); + await notifyChatConfirmationWork(tx, scope); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 6e4deaebbab..940388e4a0d 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", + chatTerminalConfirmations: "chat-terminal-confirmations", chatFailedRetryNotices: "chat-failed-retry-notices", chatRunMilestones: "chat-run-milestones", chatReceipts: "chat-slack-receipts",