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 d958711d44e..c8313d5acb9 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2455,3 +2455,50 @@ Slice 36 final remote gates: all 53 checks passed on 01efd3f4f7, with fresh Greptile 5/5 and no review threads. Slice 37 final scalar-query selection passed 20 tests; workspace typecheck and build passed. Full local acceptance and remote PR gates are still pending. No PR in this stack was merged. + +## Slice 37 accepted; slice 38 direct webhook recovery + +[#17664](https://github.com/paperclipai/paperclip/pull/17664), head +`0a25cae9ef10a8714801d72ac798823de335e050`, has all 53 remote checks +passing and fresh Greptile 5/5 without actionable findings. Complete local +chat/publication/startup acceptance passed 1,592 tests in 556.33 seconds. +The PR remains draft and unmerged. + +Slice 38 now routes direct GitHub missed-webhook recovery through saved checkpoint +deadlines and removes the last production chat reconciliation interval. The +required external-source cadence remains only for eligible direct integrations; +Cloud relay registrations and disabled/unconfigured states are quiet. A real +signed installation-deletion regression exposed and fixed a missing producer +notification. Forty-one selected cases pass, including 106-endpoint pagination, +five-item company-fair batches, and an existing default-batch consumer. Complete +acceptance and remote gates remain required. See +[the provider audit](2026-10-25-github-webhook-recovery-sleep-audit.md). + +Earlier #17490/#17572 CI failures, #17576/#17580 review findings, the entire +remaining original list, and final reconciliation with master stay open. The +current inspected master is `5a28f7acf8`; no stack PR has been merged. + + +## Slice 38 review repair and corrected acceptance + +Initial #17666 at `936e132c01` has 52 passing checks and a valid Greptile P1: +a receipt committed immediately before drain/stop could suppress a POST that +never started. Four regressions reproduced the loss. The owner now restores a +known-unsent claim with an exact attempt predicate while retaining uncertainty +after transport starts. Fifty-nine selected tests pass; six stop-boundary cases +also cover preservation of a newer receipt. Final verification is running. + +The initial broad command passed 1,586 tests in two files but misspelled the +standalone publication filename. The corrected three-file command is running +on a fresh database; the earlier result is not full publication acceptance. +Earlier stack CI/review issues, all remaining migrations, and master reconciliation +remain outstanding. No stack PR is merged or claimed ready as a whole. + + +Slice 38 production repair is pushed at `491a7e1eef`; fresh review is 5/5 and +the P1 thread is resolved. Workspace types/build passed. The corrected complete +three-suite run passed 1,605 tests and failed one assertion that read the timer +before worker settlement. The assertion now waits for settled idle tracking and +the actual future deadline; all 37 affected lifecycle/publication/startup cases +pass. The full failure remains in the audit; latest-head remote gates are required. +No full-run success or stack completion is inferred from that focused repair. diff --git a/doc/plans/2026-10-25-github-webhook-recovery-sleep-audit.md b/doc/plans/2026-10-25-github-webhook-recovery-sleep-audit.md new file mode 100644 index 00000000000..a1ac21b7995 --- /dev/null +++ b/doc/plans/2026-10-25-github-webhook-recovery-sleep-audit.md @@ -0,0 +1,357 @@ +# Direct GitHub webhook recovery scheduling + +Parent: slice 37 at `0a25cae9ef10a8714801d72ac798823de335e050`. +Branch: `codex/sleep-38-github-webhook-recovery`. +Inspected mainline: `5a28f7acf8` (2026-10-25). Final stack reconciliation is +still required; this slice does not incorporate every subsequent master change. + +## Obligation and stop condition + +A failed GitHub callback can leave no local inbox row. Direct App integrations +therefore require the existing bounded external history reconciliation. This is +an explicit exception to event-only dispatch, not proof that configured direct +GitHub endpoints become network-idle. Cloud relay registrations are excluded: +Cloud owns their transport retry. No eligible direct endpoints, including an +instance with only Cloud registrations or disabled endpoints, means no recurring +SQL or HTTP from this worker after reconstruction and notification settlement. + +Previously the app called the due-endpoint selector every second. This slice +reconstructs deadlines at startup, reloads affected scopes on committed changes, +and dispatches only at those deadlines. It removes the final production chat +reconciliation interval and its unused lane coordinator. Other existing chat +workers remain independently registered with the delivery coordinator. + +Durability already exists: + +- `chat_sdk_state`, key `paperclip:github-webhook-recovery:v1`, stores the + versioned runtime epoch, App identity, callback URL, floor, and `nextScanAt`. +- `chat_actions` recovery receipts store immutable source fences, attempts, + cancellation decisions, and unknown transport outcomes. +- The genuine authenticated GitHub callback owns inbox admission. Provider + history is never admitted directly or transformed into a locally signed event. + +No table, migration, Cloud alarm, or host-sleep exemption is added. + +## Ownership and policy + +State parsing, eligible endpoint selection, credential fingerprint comparison, +recovery execution, and notification scopes stay under `chat-channels/github/`. +The existing recovery owner retains signature, current authority, source-body, +resource, runtime-generation, lease, and claim-version checks. Recovery I/O stays +outside database transactions. A small provider-owned checkpoint transaction +notifies before its existing compare-and-set; generic SDK persistence does not +learn the meaning of GitHub keys. + +The per-endpoint recovery core is shared by the explicit/manual bounded API and +the worker. The worker independently reloads current eligibility and epoch before +execution. It does not call the manual selector with cached early deadlines: +that selector's legacy due predicate does not detect credential-only epoch +changes. The worker recognizes those changes immediately, even during a saved +24-hour backoff, and establishes a new floor without adopting historical events. + +Preserved policy: + +- One-minute normal external reconciliation; five-minute claim and failure + backoff; bounded provider rate-limit backoff up to one day. +- One-hour lookback, three provider-history pages, five inspected eligible + envelopes per endpoint, and three maximum redelivery attempts. +- Successful sibling deliveries suppress replay. Ambiguous redelivery outcomes + remain durable and are not blindly retried without a distinct failed attempt. +- Five endpoints per dispatch batch, fairly rotated between companies. + +The shared scoped-deadline helper gains only an optional batch limit. Its default +remains 25 for every existing caller. It still owns no provider SQL or policy. +The new owner loads at most 100 rows per query and retains all pages' deadlines; +it does not silently discard endpoints after the first page. + +Every recovery HTTP call checks dispatch admission and the worker abort signal +before starting. A call already in flight keeps the existing bounded provider +request timeout and is joined during shutdown. Its result/uncertainty settlement +may finish; stopping does not erase a committed attempt. Other registered queues +continue while a provider call is held. Drain and standby preserve dirty scopes, +failed reconstruction, and saved deadlines without passive SQL. + +## Producer inventory + +| Owner | Mutation and notification | +| --- | --- | +| `configuration.ts` | Configure/reconnect, pause, resume, removal and failure paths already call `notifyChatEndpointWork` in their owned transactions; that now includes the connection-scoped GitHub topic. | +| `setup-probes.ts` and chat task admission | Successful activation already notifies through `notifyChatEndpointWork` or endpoint configuration. | +| `credentials.ts` | Credential reference replacement and clearing notify endpoint configuration before writes. | +| `resource-reconciliation.ts`, `provider-effects.ts`, interaction callbacks and publication failures | Eligibility/status changes use existing configuration notifications. | +| `github/installation-lifecycle.ts` | New explicit endpoint notification precedes revocation, legacy identity upgrade, runtime generation/availability settlement, and failed-inventory transitions. | +| `chat-github-wizard.ts` | Setting `cloudRegistrationId` now notifies inside its short owned setup transaction; unrelated wizard metadata does not signal this worker. | +| `github/webhook-ingress.ts` | Claim and final recovery-window CAS writes notify before writing, including manual recovery. | +| Tool connection configuration/health and company deletion | Existing connection/company notification fan-out now carries this topic; removal invalidates cached deadlines. | +| `github/app-setup.ts` | Initial identity claim is restricted to incomplete setup; later credential persistence and activation are covered. Stage/management URL metadata alone does not change recovery eligibility. | +| Webhook authentication, ordinary activity, review settings, endpoint display/task settings | Do not change recovery eligibility, runtime epoch, callback origin, or recovery checkpoint; no added signal. | + +Supported writes share the database owner's process-local transaction signals. +Independent SQL writers and another server process need a separate integration; +this change does not claim they are observed. Nested savepoints still settle at +the outer transaction. Unknown transaction outcomes retain the shared intent +hold until the database protocol resolves them. + +The callback base URL comes from configured startup URLs or the live canonical +Cloud identity. `applyCloudRuntimeIdentityAssertion` allows one durable claim +and identical retries, and rejects replacement with another assertion. Warm +standby defers reconstruction until admission opens, so the worker sees the +claimed identity. Ordinary self-hosted URL configuration changes restart the +process. No permanent URL-discovery poll is introduced. + +## Evidence and retained failures + +- Initial server typecheck passed (`/tmp/sleep38-first-types.log`). +- First selection: 27 passed, four failed (`/tmp/sleep38-worker-first.log`). + One assertion confused a numeric scheduler deadline with a Date. Two lifecycle + cases created real timers before installing fake timers. The resume fixture + forced active status without complete setup. Corrected fixture/clock ownership; + no production timing change. All 31 selected cases then passed in 15.87 seconds + (`/tmp/sleep38-worker-fixed.log`). +- The first installation producer test passed while startup checkpoint signals + were still pending. Waiting for their actual future deadline exposed the missing + notification: deletion left the deadline armed. This regression failed in + `/tmp/sleep38-lifecycle-settled-regression.log` before the lifecycle writer fix. + It now passes through real signed deletion and reinstall behavior. +- Wired acceptance passed all integration cases but failed two obsolete source + assertions about the removed timer. Updated them to assert shared worker + registration and absence of the coordinator/interval. +- Final bounded selection: 41 passed in 26.55 seconds + (`/tmp/sleep38-bounded-tests.log`). Includes quiet unconfigured/Cloud/disabled + states, no SQL before a saved deadline, real callback-only admission, real + resume/pause, restart/drain/standby, nested rollback/outer commit, credential-only + epoch change, reconstruction failure across drain, held HTTP with independent + queue progress and joined shutdown, 106 deadlines across companies with an + initial five-item batch, and the prior review-projection default-batch consumer. + The backlog fixture then replaced timing-derived App IDs with random identities + to avoid a possible fixture uniqueness collision; final acceptance follows. +- Final workspace typecheck and build pass (`/tmp/sleep38-workspace-types.log` + and `/tmp/sleep38-workspace-build.log`). Voice session/retention acceptance + passes all 122 cases in 26.48 seconds (`/tmp/sleep38-voice.log`). Complete chat + acceptance and remote CI/review remain required. This is not a full-stack + readiness claim. + +## Review repair: a claimed redelivery that never reached transport + +Greptile identified a valid P1 at initial head `936e132c01`: drain/shutdown +could start after the recovery receipt committed and before the POST began. +The guard correctly prevented HTTP, but the catch marked the receipt uncertain, +and the next scan refused that same failed delivery. Four real-PostgreSQL cases +reproduced the lost callback for drain/shutdown with both a first claim and a +previous receipt (`/tmp/sleep38-unsent-regression.log`, four failures). + +The owner now distinguishes transport admission from a persisted claim. If the +request did not start, it restores the prior receipt or removes the new receipt +inside a short notifying transaction. If transport may have started, it retains +`delivery_unknown`. Settlement and restoration use the exact attempt identity +(id, processing status, count, provider delivery, and requested timestamp), so +an older attempt cannot overwrite a successor. Lease failure before HTTP follows +the same known-unsent path. Neither database transactions nor the provider POST +are replayed automatically. A crash between commit and transport, or failure to +persist restoration, still requires conservative durable uncertainty handling. + +- Server typecheck and 59 selected integration/publication/startup cases passed + after the repair (`/tmp/sleep38-unsent-types.log`, + `/tmp/sleep38-unsent-fixed.log`, 28.91 seconds). +- Two additional cases replace the receipt timestamp while preserving its + processing status and attempt count. All six stop-before-transport cases pass; + restoration leaves the newer receipt intact and restart does not replay it + (`/tmp/sleep38-unsent-cas.log`, 10.27 seconds). +- The initial full command passed 1,586 tests across two files in 571.36 seconds + (`/tmp/sleep38-full-chat.log`). It misspelled the standalone publication suite + filename, so it did not run that suite and is not three-suite acceptance. + The corrected complete command includes `chat-interaction-publications.test.ts`. +- Final workspace typecheck/build passed. Corrected three-suite acceptance on + `paperclip_sleep38_final_20261025_1837_test` passed 1,605 cases and failed one + resume timing assertion in 571.32 seconds (`/tmp/sleep38-final-chat.log`). + The test observed the newly committed checkpoint before `check` reloaded its + deadline and the coordinator scheduled it. It now waits for idle tracking to + settle and a deadline more than 30 seconds in the future, then verifies pause + disarms it. Production code is unchanged by this assertion repair. +- All 37 selected lifecycle/publication/startup cases pass with that assertion + (`/tmp/sleep38-settlement-fixed.log`, 23.10 seconds). This is focused repair + evidence, not a claim that the earlier full run passed. The latest commit still + requires all remote CI and fresh review. The production repair at `491a7e1eef` + received Greptile 5/5 and its P1 thread is resolved. + +## Remaining boundaries + +The idle-safety proof still blocks unsupported active channel obligations. A +registered in-process timer is not an acknowledged Cloud alarm and cannot wake a +suspended host. These tests establish worker behavior, not provider suspension or +fleet savings. GitHub connection discovery/replay, credential continuity, and +merged-PR confirmation are separate remaining duties. + +All earlier stack CI/review findings, remaining original inventory items, final +master reconciliation, and each latest-head PR gate remain required. No merge is +authorized or performed by this migration stack. + + +## Current-parent acceptance (October 28) + +The existing branch was merged normally with published review-projection parent +`1453a4447c0d4d33f713ae3e00734c7950f95426` at +`3d7f163bc5259b60f9bd19b7fc90b402f0a6e925`. Both complete added/removed patches +and every unrelated parent blob were preserved. This slice adds no migration; +the entire parent database tree, including migration history through 0403, is +byte-identical. The batch-size option already exists in the current parent; +this slice selects five without changing other consumers' default bound. + +At that integration, 751 distinct cases passed: + +- 274 complete GitHub review workflow cases. +- 294 other GitHub-named provider, installation, recovery, attachment, and + publication cases, including all 19 new recovery worker cases and all six + stop-before-transport variants. +- 30 complete terminal interaction publication cases. +- 145 complete voice-session integration and inbound retention cases. +- Eight current startup/hostname wiring cases. + +The first workflow filter omitted the separate new recovery group. The saved +verbose case list exposed that gap before acceptance was declared. A second +selection in another fresh database ran all remaining GitHub-named cases while +excluding the already-tested review group. The two passed-name sets have no +overlap. The first command filtered 1,504 unrelated cases; the complementary +command filtered 1,484 cases. These are focused selections, not the full chat +suite or full repository suite. The original commands and results remain in the +verification record. + +Coverage includes unconfigured/Cloud/disabled quiet, saved-deadline quiet, +authenticated callback-only admission, real installation and credential changes, +restart/drain/standby, nested commit/rollback, failed reconstruction, 106 saved +deadlines with five-item fair batches, held HTTP with independent queue progress, +joined shutdown, exact-attempt known-unsent restoration, and protection of a +newer receipt. Existing authority, signature, source, and replay checks also pass. + +Database build, server production typecheck/build, module boundaries, and target +migration order passed. Five separate fresh PostgreSQL fixtures received the +unchanged parent migrations through 0403. The audit follow-up changes only this +document; accepted source, tests, and migration hashes remain unchanged. + +The actual parent merge and fresh current-head CI/Greptile checks remain required. +Direct GitHub Apps retain their real one-minute external recovery obligation; +Cloud relay and disabled endpoints remain excluded. This does not expand Cloud +alarm coverage or prove deployed host/database suspension or fleet savings. + + +## Parent startup boundary correction + +The parent stale-review PR received a valid required review finding: its legacy +maintenance source slice used a removed end marker. The parent reproduced that +with explicit bounds, then fixed the marker and passed all 13 startup tests. +This webhook-recovery slice removes that entire coordinator and timer. During +the normal parent merge, its obsolete boundary inspection conflicts with this +slice's existing assertions that both symbols are absent. Keep those absence +assertions: there is no remaining coordinator section to slice. All other parent +changes, including its audit and unrelated onboarding smoke coverage, are carried +unchanged. The complete server, package, and UI source trees still match the +751-case accepted revision; the eight startup cases are also rerun after resolving +this overlap. This is a superseded test for removed functionality, not a waiver +of an unresolved production finding. Fresh review and CI remain required. + + +## Orphan-assignment review investigation (2026-10-28) + +Ponytail's review of `836acef0370d42323fdfdbd6d62a805b2469f1d4` required +preventing recovery from repeatedly loading an endpoint whose assigned agent +cannot resolve. Its reasoning correctly identifies that the loader does not +join agents, but the proposed orphan state conflicts with existing database +constraints. Investigation must establish that boundary before changing SQL. + +The first attempted reproduction set `assigned_agent_id` to NULL. It failed +**during fixture mutation**, with PostgreSQL 23502, before the worker ran. This +is not a failing worker regression and is not described as one. The original +schema and a freshly migrated PostgreSQL catalog confirm: + +- `chat_endpoints.assigned_agent_id` is NOT NULL. +- `chat_endpoints_assigned_agent_id_agents_id_fk` references `agents(id)` with + ON DELETE RESTRICT. +- `chat_endpoints_company_agent_fk` references `(agents.company_id, agents.id)`. +- Both foreign keys are validated and nondeferrable. + +Thus a supported writer cannot commit the missing, null, deleted or foreign-company +agent relationship described in that finding. A concurrent deletion cannot open +a committed orphan window. The application endpoint resolver joins those same +ID/company columns. Disabling constraints or corrupting a database would be a +different unsupported state; the tests do not weaken the schema to manufacture it. + +Three new real-database cases verify the boundary: NULL assignment fails 23502, +assignment to a missing UUID fails 23503, and deletion of the assigned agent fails +23503. Each proves the original assignment remains, then runs real webhook +recovery, observes one redelivery and a saved future deadline, and advances +15 seconds with no SQL reads or extra transport. All **22 recovery-scheduling +cases** pass on a fresh database, including the previous 19 cases and these +three additions. The 1759 unselected cases are not counted as current passes. + +Production source is unchanged. The earlier 751-case acceptance retains its +stated revision and scope; these 22 overlap 19 of those cases, so there are +754 distinct tested cases across the evidence, not 773. No deployed sleep claim +is made. The legacy manual recovery query remains an optional review follow-up. +The actual parent squash ancestry was merged normally with the entire result +tree identical to the previously validated branch, so already-landed parent +changes no longer inflate this PR's diff. Fresh current-head CI and review +must evaluate the constraint evidence before this slice can merge; an old +conditional verdict or neutral check is not treated as a pass. + + +## Repository comment rule follow-up + +Apply CONTRIBUTING.md's prohibition on new code comments consistently to this +slice too. Seven newly added comment lines are removed: two in the constraint +regression and five explaining GitHub pagination guards. The database-constraint +evidence above and pagination/partial-window rationale in this audit remain the +durable explanation. TypeScript parsing proves both affected files retain the +exact non-comment token stream; no recovery predicate, guard, assertion, deadline +or transport call changes. The 22-case current recovery acceptance and earlier +751-case runtime evidence keep their stated revisions and scope. Fresh head CI +and review are still required; neither an old score nor comment-only equivalence +by itself satisfies the merge gates. + + +## Current CI test corrections (2026-10-29) + +The cbe3fbea head failed two independent test boundaries. Neither changes the +production scheduler, database constraints, delivery state machine or deadlines. + +The new deleted-agent probe expected PostgreSQL 17's 23503 foreign-key error. +CI PostgreSQL 18 correctly reports 23001 (restrict violation) for the same +forbidden delete. Running the original assertion against local embedded +PostgreSQL 18 reproduced that exact failure. The assertion now accepts only +those two integrity codes for deletion; NULL still requires 23502 and a missing +UUID still requires 23503. It also checks that the original agent row and +assignment both remain before testing healthy deadline-driven recovery. All +22 recovery-scheduling cases pass on embedded PostgreSQL 18. The three integrity +probes also pass against a fresh PostgreSQL 17 database. They are the same three +cases across two engines, not six distinct cases. The invalid-orphan premise +remains ruled out by validated nondeferrable constraints, not by changing SQL. + +The attachment E2E failed waiting for a file chooser. Its screenshot and trace +showed an announcement appearing while the attachment menu was open. The global +locator handler tried to dismiss the announcement outside that modal menu, so +its own click could not complete and the original menu-item click could not run. +The unchanged failing journey passed naturally once locally; that pass did not +explain or erase the CI failure. A controlled delayed real announcement response +released after the menu opened reproduced the same 60-second filechooser and +intercepted-click failures. The handler now closes the menu through Escape, +dismisses the announcement through its visible button, and reopens the menu +so the original upload action can continue. The same first late-announcement +condition passes with this repair. The temporary fault is one-shot after that +first injection so a dismissed card is not incorrectly awaited for the second +attachment; the failing predecessor never reached a second injection. + +The temporary delay/route fault was removed before final acceptance. All **11 +complete attachment receipt journeys pass**, retries zero, in 38.7 seconds. +Every journey body, receipt/content assertion, 60-second test budget and +120-second startup budget is unchanged. The full run uses a fresh database on +the existing local PostgreSQL 17 test server and the canonical throwaway +Paperclip home, CLI onboarding, server, UI, storage and HTTP paths. An earlier +full-run attempt failed during embedded PostgreSQL initialization before any +browser test started; host semaphore usage was near capacity. That startup +failure is retained and is not counted as a test pass. No shared host resources +were deleted or unrelated databases stopped. PostgreSQL 18 coverage is provided +separately by the recovery suite and remains required in fresh CI. + +Server typecheck passes. Production source and schema are unchanged from cbe3fbea; +only the two test files and this audit change. The prior 751-case runtime scope +remains exactly as recorded. Fresh current-head CI and reviews remain required +before merge; no previous failing result is erased by these passes. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 425e53e98e4..6d86d692ade 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -1,86 +1,18 @@ import { readFileSync } from "node:fs"; -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it } from "vitest"; import { - createChatReconciliationCoordinator, shouldEnablePrivateHostnameGuard, } from "../app.ts"; -describe("createChatReconciliationCoordinator", () => { - it.each([false, true])( - "keeps later inbound sweeps independent of GitHub recovery and joins shutdown (rejects: %s)", - async (rejects) => { - let release!: () => void; - const held = new Promise((resolve) => { - release = resolve; - }); - const failure = new Error("GitHub recovery unavailable"); - const processFailedGitHubWebhookDeliveries = vi.fn(async () => { - await held; - if (rejects) throw failure; - }); - const processPendingDeliveries = vi.fn(async () => undefined); - const flushPublications = vi.fn(async () => undefined); - const onError = vi.fn(); - const input = { - processFailedGitHubWebhookDeliveries, - processPendingDeliveries, - flushPublications, - processPendingSlackFileUploadReceipts: async () => undefined, - processPendingSlackSessionSyncs: async () => undefined, - onError, - }; - const coordinator = createChatReconciliationCoordinator(input); - let draining: Promise | undefined; - try { - coordinator.reconcile(); - await vi.waitFor(() => { - expect(processFailedGitHubWebhookDeliveries).toHaveBeenCalledOnce(); - expect(processPendingDeliveries).toHaveBeenCalledTimes(1); - }); - // A second sweep must admit work arriving AFTER the first sweep while - // remote GitHub recovery remains held. Same-sweep parallelism is not enough. - coordinator.reconcile(); - await vi.waitFor(() => { - expect(processPendingDeliveries).toHaveBeenCalledTimes(2); - expect(flushPublications).toHaveBeenCalledTimes(2); - }); - expect(processFailedGitHubWebhookDeliveries).toHaveBeenCalledOnce(); - coordinator.stop(); - let drained = false; - draining = coordinator.drain().then(() => { - drained = true; - }); - await new Promise((resolve) => setImmediate(resolve)); - expect(drained).toBe(false); - coordinator.reconcile(); - release(); - await draining; - expect(drained).toBe(true); - expect(processFailedGitHubWebhookDeliveries).toHaveBeenCalledOnce(); - expect(processPendingDeliveries).toHaveBeenCalledTimes(2); - expect(onError).toHaveBeenCalledTimes(rejects ? 1 : 0); - if (rejects) - expect(onError).toHaveBeenCalledWith( - "GitHub webhook recovery", - failure, - ); - } finally { - release(); - coordinator.stop(); - await (draining ?? coordinator.drain()); - } - }, - ); - +describe("chat worker startup wiring", () => { it("keeps receiver and voice inventory out of periodic chat maintenance", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); - const maintenanceStart = source.indexOf("const chatReconciliation ="); - const maintenanceEnd = source.indexOf("let chatPublicationTimer:", maintenanceStart); - expect(maintenanceStart).toBeGreaterThanOrEqual(0); - expect(maintenanceEnd).toBeGreaterThan(maintenanceStart); - const maintenance = source.slice(maintenanceStart, maintenanceEnd); + const maintenance = source; + expect(source).not.toContain("chatPublicationTimer"); + expect(source).not.toContain("createChatReconciliationCoordinator"); + expect(source).toContain("chatChannels.registerGitHubWebhookRecoveryWork(deliveryWork,"); expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); - expect(maintenance).not.toContain("chatChannels.voice."); + expect(maintenance).not.toContain("chatChannels.voice.reconcile()"); expect(maintenance).not.toContain("enqueueSafeNativeChatProgress"); expect(source).toContain("nativeChatProgressWork(db).register(deliveryWork,"); expect(source).toContain("chatInboundWakeNoticeWork(chatChannels).register(deliveryWork,"); @@ -122,7 +54,7 @@ describe("createChatReconciliationCoordinator", () => { it("wires GitHub recovery into its independent lane", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); expect(source).toContain( - "chatChannels.processFailedGitHubWebhookDeliveries()", + "chatChannels.registerGitHubWebhookRecoveryWork(deliveryWork,", ); const service = readFileSync( new URL("../services/chat-channels.ts", import.meta.url), @@ -139,10 +71,7 @@ describe("createChatReconciliationCoordinator", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); expect(source).toContain("registerChatDeliveryWork(deliveryWork, chatChannels,"); expect(source).toContain("registerChatActionWork(deliveryWork, chatChannels,"); - const maintenance = source.slice( - source.indexOf("const chatReconciliation ="), - source.indexOf("let chatPublicationTimer:"), - ); + const maintenance = source; expect(maintenance).not.toContain("chatChannels.processPublicationMaintenance()"); expect(maintenance).not.toContain("chatChannels.schedulePendingPublications()"); expect(maintenance).not.toContain("chatChannels.processPendingPublications()"); @@ -161,127 +90,7 @@ describe("createChatReconciliationCoordinator", () => { expect(actionMaintenance).not.toContain("processPendingGitHubWebhookIngress("); }); - it("keeps slow optional recovery from suppressing later publication sweeps", async () => { - let releaseDelivery!: () => void; - let releaseSlackStatus!: () => void; - const deliveryReleased = new Promise((resolve) => { - releaseDelivery = resolve; - }); - const slackStatusReleased = new Promise((resolve) => { - releaseSlackStatus = resolve; - }); - const processPendingDeliveries = vi.fn(async () => deliveryReleased); - const flushPublications = vi.fn(async () => undefined); - const processPendingSlackFileUploadReceipts = vi.fn(async () => undefined); - const processPendingSlackSessionSyncs = vi.fn( - async () => slackStatusReleased, - ); - const onError = vi.fn(); - const coordinator = createChatReconciliationCoordinator({ - processPendingDeliveries, - flushPublications, - processPendingSlackFileUploadReceipts, - processPendingSlackSessionSyncs, - onError, - }); - coordinator.reconcile(); - await vi.waitFor(() => { - expect(processPendingDeliveries).toHaveBeenCalledTimes(1); - expect(flushPublications).toHaveBeenCalledTimes(1); - expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(1); - expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(1); - }); - coordinator.reconcile(); - await vi.waitFor(() => { - expect(flushPublications).toHaveBeenCalledTimes(2); - expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2); - }); - expect(processPendingDeliveries).toHaveBeenCalledTimes(1); - expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(1); - - releaseDelivery(); - releaseSlackStatus(); - await coordinator.drain(); - expect(onError).not.toHaveBeenCalled(); - }); - - it.each([false, true])( - "joins the independent Slack file receipt lane at shutdown (rejects: %s)", - async (rejects) => { - let releaseReceipt!: () => void; - const receiptReleased = new Promise((resolve) => { - releaseReceipt = resolve; - }); - const lookupError = new Error("receipt lookup failed"); - const processPendingSlackFileUploadReceipts = vi - .fn(async () => undefined) - .mockImplementationOnce(async () => { - await receiptReleased; - if (rejects) throw lookupError; - }); - const processPendingDeliveries = vi.fn(async () => undefined); - const flushPublications = vi.fn(async () => undefined); - const processPendingSlackSessionSyncs = vi.fn(async () => undefined); - const onError = vi.fn(); - const coordinator = createChatReconciliationCoordinator({ - processPendingDeliveries, - flushPublications, - processPendingSlackFileUploadReceipts, - processPendingSlackSessionSyncs, - onError, - }); - let draining: Promise | undefined; - try { - coordinator.reconcile(); - await vi.waitFor(() => { - expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes( - 1, - ); - expect(flushPublications).toHaveBeenCalledTimes(1); - }); - coordinator.reconcile(); - await vi.waitFor(() => { - expect(processPendingDeliveries).toHaveBeenCalledTimes(2); - expect(flushPublications).toHaveBeenCalledTimes(2); - expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(2); - }); - expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(1); - - let drained = false; - draining = coordinator.drain().then(() => { - drained = true; - }); - // Flush the promise queue, not a wall-clock delay: the held receipt - // must still be part of shutdown's joined work after other lanes end. - await new Promise((resolve) => setImmediate(resolve)); - expect(drained).toBe(false); - expect(onError).not.toHaveBeenCalled(); - - releaseReceipt(); - await draining; - expect(drained).toBe(true); - expect(onError).toHaveBeenCalledTimes(rejects ? 1 : 0); - if (rejects) { - expect(onError).toHaveBeenCalledWith( - "Slack file receipts", - lookupError, - ); - } - - // A completed or failed lookup releases only its own single-flight - // slot; the next ordinary reconciliation can recover another receipt. - coordinator.reconcile(); - await coordinator.drain(); - expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2); - expect(onError).toHaveBeenCalledTimes(rejects ? 1 : 0); - } finally { - releaseReceipt(); - await draining; - await coordinator.drain(); - } - }, - ); }); describe("shouldEnablePrivateHostnameGuard", () => { diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index c5b0118fe0d..1fba38d0cf6 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -14821,6 +14821,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }); await qualifySetupRoundTrip(service, endpoint.id); await service.test(endpoint.id, "owner-user"); + const recovery = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + onTestFinished(() => recovery.stop()); + await service.registerGitHubWebhookRecoveryWork(recovery, () => true).ready; + await expect.poll(() => (recovery.nextWakeAt() ?? 0) - Date.now(), { timeout: 5000 }).toBeGreaterThan(30_000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); const [conversation] = await readChatConversationsByEndpointId(endpoint.id); const credentialRefsBeforeDeletion = await db .select({ refs: toolConnections.credentialSecretRefs }) @@ -14847,6 +14852,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { status: "revoked", healthMessage: "GitHub App installation was removed", }); + await expect.poll(() => recovery.nextWakeAt(), { timeout: 5000 }).toBeNull(); await expect( db .select({ @@ -16113,6 +16119,347 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await configured.service.shutdown(); }); + describe("GitHub webhook recovery scheduling", () => { + it.each(["unconfigured", "cloud", "disabled"] as const)("leaves %s recovery quiet", async mode => { + const f = mode === "unconfigured" ? null : await githubPreIngressRecoveryFixture(); + const service = f?.service ?? createService().service; + if (f) { + const [endpoint] = await readChatEndpointsById(f.endpoint.id); + if (mode === "cloud") await db.update(chatEndpoints).set({ setup: { ...endpoint.setup, + github: { ...endpoint.setup.github, cloudRegistrationId: "relay-owned" } } }).where(eq(chatEndpoints.id, endpoint.id)); + else await service.configure(endpoint.id, { action: "pause" }, "owner-user"); + f.requests.length = 0; + } + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ active: 0, next: null }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); } + expect(f?.requests ?? []).toEqual([]); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["null", "missing", "deleted"] as const)("rejects a %s assigned agent before GitHub recovery can become orphaned", async mode => { + const f = await githubPreIngressRecoveryFixture(); + const [endpoint] = await readChatEndpointsById(f.endpoint.id); + await f.makeScanDue(); + + + const mutation = mode === "null" + ? db.execute(sql`update ${chatEndpoints} set assigned_agent_id = null where ${chatEndpoints.id} = ${endpoint.id}`) + : mode === "missing" + ? db.update(chatEndpoints).set({ assignedAgentId: randomUUID() }).where(eq(chatEndpoints.id, endpoint.id)) + : db.delete(agents).where(eq(agents.id, endpoint.assignedAgentId)); + const integrityCode = mode === "null" ? "23502" + : mode === "deleted" ? expect.stringMatching(/^(23001|23503)$/) : "23503"; + await expect(mutation).rejects.toMatchObject({ cause: { code: integrityCode } }); + expect((await readChatEndpointsById(endpoint.id))[0].assignedAgentId).toBe(endpoint.assignedAgentId); + expect(await db.select({ id: agents.id }).from(agents).where(eq(agents.id, endpoint.assignedAgentId))) + .toEqual([{ id: endpoint.assignedAgentId }]); + f.requests.length = 0; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, + scheduled: (coordinator.nextWakeAt() ?? 0) > Date.now() + 30_000 }), { timeout: 5000 }) + .toEqual({ active: 0, scheduled: true }); + expect(f.posts).toHaveLength(1); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"); + try { await vi.advanceTimersByTimeAsync(15_000); expect(reads).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); vi.useRealTimers(); } + expect(f.posts).toHaveLength(1); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("waits for the saved recovery deadline and requests a genuine GitHub redelivery", async () => { + const f = await githubPreIngressRecoveryFixture(); + await f.makeScanDue(); + const state = await f.checkpoint(); + const dueAt = Date.now() + 30_000; + await db.update(chatSdkState).set({ value: { ...(state.value as Record), + nextScanAt: new Date(dueAt).toISOString() } }).where(eq(chatSdkState.id, state.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBe(dueAt); + const reads = vi.spyOn(db, "select"); + try { await vi.advanceTimersByTimeAsync(15_000); expect(reads).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); } + expect(f.requests).toEqual([]); + await vi.advanceTimersByTimeAsync(16_000); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(f.posts).toHaveLength(1); }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect((await f.receipt())?.status).toBe("processed"); }); + expect(await f.ingress()).toBeUndefined(); expect(f.wakeup).not.toHaveBeenCalled(); + expect((await f.callback()).status).toBe(202); + expect(await f.ingress()).toBeDefined(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("reactivates recovery from real resume and cancels its deadline on pause", async () => { + const f = await githubPreIngressRecoveryFixture(); + const [endpoint] = await readChatEndpointsById(f.endpoint.id); + await db.update(chatEndpoints).set({ setup: { ...endpoint.setup, step: "complete" } }).where(eq(chatEndpoints.id, endpoint.id)); + await f.service.configure(f.endpoint.id, { action: "pause" }, "owner-user"); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + const oldVersion = (await f.checkpoint()).version; + await f.service.configure(f.endpoint.id, { action: "resume" }, "owner-user"); + await expect.poll(async () => (await f.checkpoint()).version, { timeout: 5000 }).toBeGreaterThan(oldVersion); + await expect.poll(() => { + const next = coordinator.nextWakeAt(); + return { active: idleWorkSnapshot().active, future: next !== null && next > Date.now() + 30_000 }; + }, { timeout: 5000 }).toEqual({ active: 0, future: true }); + await f.service.configure(f.endpoint.id, { action: "pause" }, "owner-user"); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ active: 0, next: null }); + } finally { await coordinator.stop(); } + }); + + it.each(["restart", "drain", "standby"] as const)("reconstructs due GitHub recovery through %s", async mode => { + const f = await githubPreIngressRecoveryFixture(); + await f.makeScanDue(); + let canRun = false; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + try { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => canRun).ready; + if (mode === "restart") await coordinator.stop(); + const reads = vi.spyOn(db, "select"); + try { await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); } + expect(f.requests).toEqual([]); + canRun = true; + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + } + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect((await f.receipt())?.status).toBe("processed"); }); + expect(f.posts).toHaveLength(1); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("retains GitHub recovery configuration intent through nested commit and rollback", async () => { + const f = await githubPreIngressRecoveryFixture(); + const old = await f.checkpoint(); + const oldWindow = old.value as Record; + await db.update(chatSdkState).set({ value: { ...oldWindow, + nextScanAt: new Date(Date.now() + 86_400_000).toISOString() } }).where(eq(chatSdkState.id, old.id)); + const [endpoint] = await readChatEndpointsById(f.endpoint.id); + const change = async (tx: Parameters[0]>[0]) => { + await notifyChatEndpointConfiguration(tx, endpoint); + await tx.update(chatEndpoints).set({ setup: { ...endpoint.setup, + runtimeGeneration: Number(oldWindow.generation) + 1 } as typeof endpoint.setup }).where(eq(chatEndpoints.id, endpoint.id)); + }; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + let release!: () => void; + let pending: Promise | undefined; + try { + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + await expect(db.transaction(async tx => { await tx.transaction(change); throw new Error("rollback epoch"); })).rejects.toThrow("rollback epoch"); + expect((await f.checkpoint()).version).toBe(old.version); + let entered!: () => void; + const held = new Promise(resolve => { entered = resolve; }); + const wait = new Promise(resolve => { release = resolve; }); + pending = db.transaction(async tx => { await tx.transaction(change); entered(); await wait; }); + await held; + expect((await f.checkpoint()).version).toBe(old.version); + expect(f.requests).toEqual([]); + release(); await pending; + await expect.poll(async () => (await f.checkpoint()).value, { timeout: 5000 }).toMatchObject({ + generation: Number(oldWindow.generation) + 1, outcome: "epoch_initialized" }); + expect(f.requests).toEqual([]); + expect(coordinator.nextWakeAt()!).toBeLessThan(Date.now() + 90_000); + } finally { release?.(); await pending; await coordinator.stop(); } + }); + + it("preempts a future GitHub recovery window when only credential references change", async () => { + const f = await githubPreIngressRecoveryFixture(); + const old = await f.checkpoint(); + const oldWindow = old.value as Record; + await db.update(chatSdkState).set({ value: { ...oldWindow, + nextScanAt: new Date(Date.now() + 86_400_000).toISOString() } }).where(eq(chatSdkState.id, old.id)); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.id, f.endpoint.connectionId)); + await db.transaction(async tx => { + await notifyChatEndpointConfiguration(tx, f.endpoint); + await tx.update(toolConnections).set({ credentialSecretRefs: connection.credentialSecretRefs.map(ref => ({ ...ref, + versionSelector: ref.versionSelector === "1" ? "latest" : "1" })) }).where(eq(toolConnections.id, connection.id)); + }); + await expect.poll(async () => ((await f.checkpoint()).value as Record).credentialFingerprint, + { timeout: 5000 }).not.toBe(oldWindow.credentialFingerprint); + expect((await f.checkpoint()).value).toMatchObject({ generation: oldWindow.generation, outcome: "epoch_initialized" }); + expect(f.requests).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("recovers a failed GitHub deadline reconstruction after drain without passive SQL", async () => { + const f = await githubPreIngressRecoveryFixture(); await f.makeScanDue(); + let canRun = true; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true, + onError: error => errors.push(error) }); + const reads = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("recovery inventory unavailable"); }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => canRun).ready; + expect(errors).toHaveLength(1); + reads.mockRestore(); canRun = false; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + const passive = vi.spyOn(db, "select"); + try { await vi.advanceTimersByTimeAsync(60_000); expect(passive).not.toHaveBeenCalled(); } + finally { passive.mockRestore(); } + canRun = true; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect((await f.receipt())?.status).toBe("processed"); }); + expect(errors).toHaveLength(1); + } finally { reads.mockRestore(); await coordinator.stop(); vi.useRealTimers(); } + }); + + it("reconstructs more than one page of GitHub recovery deadlines and dispatches fair batches of five", async () => { + const first = await githubPreIngressRecoveryFixture(), second = await githubPreIngressRecoveryFixture(); + const ids: string[] = []; + for (const [f, count] of [[first, 103], [second, 3]] as const) { + const [endpoint] = await readChatEndpointsById(f.endpoint.id); + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.id, endpoint.connectionId)); + const copies = Array.from({ length: count }, () => ({ id: randomUUID(), connectionId: randomUUID(), + appId: BigInt("0x" + randomUUID().replaceAll("-", "").slice(0, 12)).toString() })); + await db.insert(toolConnections).values(copies.map(copy => ({ ...connection, id: copy.connectionId, uid: copy.connectionId }))); + await db.insert(chatEndpoints).values(copies.map(copy => ({ ...endpoint, id: copy.id, connectionId: copy.connectionId, + publicId: copy.id, botExternalId: copy.appId, botUsername: "recovery-" + copy.id }))); + ids.push(...copies.map(copy => copy.id)); + } + const windows = () => db.select({ companyId: chatSdkState.companyId }).from(chatSdkState).where(and( + inArray(chatSdkState.endpointId, ids), eq(chatSdkState.stateKey, "paperclip:github-webhook-recovery:v1"))); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await first.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + const initial = await windows(); + expect(initial).toHaveLength(5); + expect(new Set(initial.map(row => row.companyId)).size).toBe(2); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await windows()).toHaveLength(106); + }, { timeout: 10_000 }); + expect(first.requests).toEqual([]); expect(second.requests).toEqual([]); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each([ + ["drain", false, false], ["shutdown", false, false], + ["drain", true, false], ["shutdown", true, false], + ["drain", false, true], ["shutdown", true, true], + ] as const)("preserves a GitHub redelivery stopped before transport (%s, prior receipt: %s, superseded: %s)", async (mode, priorReceipt, superseded) => { + const f = await githubPreIngressRecoveryFixture(); + await f.makeScanDue(); + if (priorReceipt) { + await f.service.processFailedGitHubWebhookDeliveries(5, f.endpoint.id); + const sent = (await f.receipt())!; + await db.update(chatActions).set({ result: { ...sent.result, retryAt: new Date(0).toISOString(), + requestedAt: new Date(Date.now() - 60_000).toISOString() } }).where(eq(chatActions.id, sent.id)); + f.setDeliveries([{ ...f.original, id: "9007199254740994", redelivery: true, + delivered_at: new Date(Date.now() - 15_000).toISOString() }, f.original]); + await f.makeScanDue(); + } + const previous = await f.receipt(); + let successor: Awaited>; + f.posts.length = 0; + let canRun = true, stoppedAfterClaim = false; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true, + onError: () => undefined }); + let stopping: Promise | undefined; + const transaction = db.transaction.bind(db); + const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => { + const result = await transaction(callback, options); + if (typeof result === "string" && !stoppedAfterClaim) { + const receipt = await f.receipt(); + if (receipt?.id === result && receipt.status === "processing") { + if (superseded) { + await db.update(chatActions).set({ result: { ...receipt.result, + requestedAt: new Date(Date.now() + 1000).toISOString() } }).where(eq(chatActions.id, receipt.id)); + successor = await f.receipt(); + } + stoppedAfterClaim = true; canRun = false; + if (mode === "shutdown") stopping = coordinator.stop(); + } + } + return result; + }) as Db["transaction"]); + try { + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => canRun).ready; + await stopping; + expect(stoppedAfterClaim).toBe(true); expect(f.posts).toEqual([]); + if (superseded) expect(await f.receipt()).toEqual(successor); + else if (previous) expect(await f.receipt()).toMatchObject({ status: previous.status, + result: previous.result, payload: previous.payload }); + else expect(await f.receipt()).toBeUndefined(); + transactions.mockRestore(); + await coordinator.stop(); + await f.makeScanDue(); + canRun = true; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready; + if (superseded) { + expect(f.posts).toEqual([]); + expect(await f.receipt()).toEqual(successor); + } else { + expect(f.posts).toHaveLength(1); + expect(await f.receipt()).toMatchObject({ status: "processed", result: { attempts: priorReceipt ? 2 : 1 } }); + } + } finally { transactions.mockRestore(); await coordinator.stop(); await stopping; } + }); + + it("joins held GitHub history at shutdown without starting another HTTP request", async () => { + const f = await githubPreIngressRecoveryFixture(); await f.makeScanDue(); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + f.setBeforeList(async () => { entered(); await held; }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: () => undefined }); + let deliveries = 0; + const unrelated = coordinator.register(DELIVERY_QUEUES.feedback, { retryMs: 1000, + run: async () => { deliveries++; }, hasPending: async () => false }); + await unrelated.ready; + const worker = f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true); + let stopping: Promise | undefined; + try { + await started; + unrelated.wake(); + await expect.poll(() => deliveries).toBe(2); + const count = f.requests.length; + let stopped = false; + stopping = coordinator.stop().then(() => { stopped = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(stopped).toBe(false); + release(); await worker.ready; await stopping; + expect(f.requests).toHaveLength(count); expect(f.posts).toEqual([]); + expect((await f.checkpoint()).value).toMatchObject({ outcome: "scan_failed" }); + } finally { release(); await (stopping ?? coordinator.stop()); } + }); + }); + it("leaves Cloud-managed GitHub webhook recovery to the relay", async () => { const context = await githubPreIngressRecoveryFixture(); try { diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 7effe7f6bca..ed0b95194cc 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -1,4 +1,3 @@ -import { createChatReconciliationCoordinator } from "../app.js"; import { subscribeDatabaseWork } from "@paperclipai/db"; import * as voiceReplyNotifications from "../services/voice/voice-reply-notifications.js"; import { voiceReplySchedule } from "../services/voice/voice-reply-schedule.js"; @@ -263,21 +262,12 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu } }); - it("leaves due history intact during chat maintenance and deletes it through retention", async () => { + it("deletes due history through its retention worker", async () => { const f = await inboundFixture(); const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" }); await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id)); const onError = vi.fn(); - const maintenance = createChatReconciliationCoordinator({ - processPendingDeliveries: async () => undefined, - flushPublications: async () => undefined, onError, - }); - try { - maintenance.reconcile(); await maintenance.drain(); - expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(1); - expect(onError).not.toHaveBeenCalled(); - } finally { maintenance.stop(); await maintenance.drain(); } const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); try { await f.service.inbound.registerRetentionWork(coordinator, () => true).ready; diff --git a/server/src/app.ts b/server/src/app.ts index 9bca3535255..f4efdb2165a 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -201,7 +201,6 @@ 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 } from "./services/chat-publication-reconciliation.js"; import { heartbeatService } from "./services/heartbeat.js"; import { pluginLifecycleManager } from "./services/plugin-lifecycle.js"; import { registerPluginLifecycleDelivery } from "./services/plugin-lifecycle-delivery.js"; @@ -229,7 +228,6 @@ import { createChatWebhookDiagnostics } from "./services/chat-webhook-diagnostic type UiMode = "none" | "static" | "vite-dev"; const FEEDBACK_EXPORT_FLUSH_INTERVAL_MS = 5_000; -const CHAT_PUBLICATION_FLUSH_INTERVAL_MS = 1_000; const VITE_DEV_ASSET_PREFIXES = [ "/@fs/", "/@id/", @@ -321,76 +319,6 @@ export function shouldEnablePrivateHostnameGuard(opts: { ); } -type ChatReconciliationLane = - | "deliveries" - | "GitHub webhook recovery" - | "publications" - | "Slack file receipts" - | "Slack session status"; - -/** - * Provider recovery can wait on slow external I/O. Keep each existing durable - * lane single-flight without making an optional provider effect suppress the - * next publication sweep for every endpoint. - */ -export function createChatReconciliationCoordinator(input: { - processPendingDeliveries: () => Promise; - processFailedGitHubWebhookDeliveries?: () => Promise; - flushPublications: () => Promise; - processPendingSlackFileUploadReceipts?: () => Promise; - processPendingSlackSessionSyncs?: () => Promise; - onError: (lane: ChatReconciliationLane, error: unknown) => void; -}) { - let stopped = false; - const inFlight = new Map>(); - const publicationReconciliation = createCoalescedAsyncTrigger({ - run: input.flushPublications, - onError: (error) => input.onError("publications", error), - }); - const start = ( - lane: ChatReconciliationLane, - task: () => Promise, - ) => { - if (stopped || inFlight.has(lane)) return; - const pending = Promise.resolve() - .then(task) - .then(() => undefined) - .catch((error) => input.onError(lane, error)) - .finally(() => { - if (inFlight.get(lane) === pending) inFlight.delete(lane); - }); - inFlight.set(lane, pending); - }; - return { - reconcile() { - if (stopped) return; - start("deliveries", input.processPendingDeliveries); - if (input.processFailedGitHubWebhookDeliveries) { - start( - "GitHub webhook recovery", - input.processFailedGitHubWebhookDeliveries, - ); - } - publicationReconciliation.poll(); - if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts); - if (input.processPendingSlackSessionSyncs) start("Slack session status", input.processPendingSlackSessionSyncs); - }, - notifyPublications() { - publicationReconciliation.notify(); - }, - stop() { - stopped = true; - publicationReconciliation.stop(); - }, - async drain() { - await Promise.allSettled([ - ...inFlight.values(), - ]); - await publicationReconciliation.drain(); - }, - }; -} - export function createManagedBundledPluginWorkerRecovery(input: { managedBundledPluginKeys: readonly string[]; workerManager: Pick< @@ -1301,25 +1229,7 @@ export async function createApp( deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(), ); emailChannels.start(); - const chatReconciliation = createChatReconciliationCoordinator({ - // Chat delivery, action, and Telegram discovery workers now own these - // duties. Keep the coordinator's other maintenance lanes independent. - processPendingDeliveries: async () => undefined, - processFailedGitHubWebhookDeliveries: () => - chatChannels.processFailedGitHubWebhookDeliveries(), - flushPublications: async () => undefined, - onError: (lane, err) => { - logger.error({ err, lane }, `Failed to reconcile chat ${lane}`); - }, - }); - let chatPublicationTimer: ReturnType | null = setInterval( - () => { - if (!isWarmStandby() && !isIdleTaskDrainActive()) chatReconciliation.reconcile(); - }, - CHAT_PUBLICATION_FLUSH_INTERVAL_MS, - ); - chatPublicationTimer.unref?.(); - if (!isWarmStandby() && !isIdleTaskDrainActive()) chatReconciliation.reconcile(); + chatChannels.registerGitHubWebhookRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); // Abandoned chunked-import spool sweep: hourly (plus once at startup), // deleting spool dirs whose transfer saw no activity for 24h and cancelling // their still-open ledger runs. Same setInterval + unref + shutdown-clear @@ -1446,12 +1356,6 @@ export async function createApp( jobCoordinator.stop(); resourceLifecycleDelivery.dispose(); await deliveryWork.stop(); - chatReconciliation.stop(); - if (chatPublicationTimer) { - clearInterval(chatPublicationTimer); - chatPublicationTimer = null; - } - await chatReconciliation.drain(); if (importTransferSweepTimer) { clearInterval(importTransferSweepTimer); importTransferSweepTimer = null; diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 06e74126247..804359109e5 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -877,7 +877,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { get deferWebhookProcessing() { return options.deferWebhookProcessing; }, }); const { handleTelegramSlashCommand } = createTelegramCommands({ runtime, runtimeFor, handleSdkMessage }); - const { stageGitHubWebhookIngress, githubIngressAcceptedResponse, processGitHubWebhookIngress, processFailedGitHubWebhookDeliveries, processPendingGitHubWebhookIngress } = createGitHubWebhookIngress(db, { endpointRecord, runtimeContextForRecord, resolveCredentialRefs, runtimeGeneration, credentialFingerprint, getWebhookPublicBaseUrl, redactError, handleWebhook, runtimeCallbackEndpoint, fetchImpl, persistence, withCredentialMutationLease, isShuttingDown: () => shuttingDown, get githubWebhookReplayBarrier() { return options.githubWebhookReplayBarrier; }, supportedWebhookEvents: SUPPORTED_GITHUB_WEBHOOK_EVENTS }); + const { registerWebhookRecoveryWork, stageGitHubWebhookIngress, githubIngressAcceptedResponse, processGitHubWebhookIngress, processFailedGitHubWebhookDeliveries, processPendingGitHubWebhookIngress } = createGitHubWebhookIngress(db, { endpointRecord, runtimeContextForRecord, resolveCredentialRefs, runtimeGeneration, credentialFingerprint, getWebhookPublicBaseUrl, redactError, handleWebhook, runtimeCallbackEndpoint, fetchImpl, persistence, withCredentialMutationLease, isShuttingDown: () => shuttingDown, get githubWebhookReplayBarrier() { return options.githubWebhookReplayBarrier; }, supportedWebhookEvents: SUPPORTED_GITHUB_WEBHOOK_EVENTS }); const { stageSlackSessionStop, processSlackSessionStop, processPendingSlackSessionStops, handleSlackSessionStopPayload } = createSlackSessionStop(db, { scheduleMessageProcessing, get deferWebhookProcessing() { return options.deferWebhookProcessing; }, runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, nonDirectDestinationAllowed, stageAuthorizedTaskControlPublication, getPublicBaseUrl, get heartbeat() { return options.heartbeat; }, redactError }); const { stageTelegramMaintenance, processTelegramMaintenance, processPendingTelegramMaintenance, processPendingTelegramActions, processTelegramSubscriptionDiscovery, scheduleTelegramSubscriptionDiscovery, hasPendingTelegramSubscriptionDiscovery, hasTelegramStopSubscription, recordTelegramStopSubscription } = createTelegramMaintenance(db, { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, resolveCredentials, clearCredentials, runtimeCallbackEndpoint, getWebhookPublicBaseUrl, fetchImpl, runtimeGeneration, credentialFingerprint, redactError, credentialCheckTimeoutMs: PROVIDER_CREDENTIAL_CHECK_TIMEOUT_MS }); const { reconcileGitHubWebhookRepository, reconcileProviderResourceRows, listResources, listGitHubRepositories, replaceResources, toggleAllGitHubRepositories, hasNewerProcessedProviderLifecycleEffect, applyProviderAvailability } = createChatResourceReconciliation(db, { endpointRecord, withCredentialMutationLease, runtimeGeneration, maxErrorText: MAX_ERROR_TEXT }); @@ -3583,6 +3583,7 @@ return { processFailedChatRunRetry, processFailedChatRunRetries, processPendingGitHubWebhookIngress, + registerGitHubWebhookRecoveryWork: registerWebhookRecoveryWork, processFailedGitHubWebhookDeliveries, processPendingProviderEffects, processPendingReceiptReactions, diff --git a/server/src/services/chat-channels/configuration-work-notifications.ts b/server/src/services/chat-channels/configuration-work-notifications.ts index 146b08417cd..9d3240aee12 100644 --- a/server/src/services/chat-channels/configuration-work-notifications.ts +++ b/server/src/services/chat-channels/configuration-work-notifications.ts @@ -1,3 +1,4 @@ +import { notifyGitHubWebhookRecovery } from "./github/webhook-recovery-notifications.js"; import { notifyVoiceChange } from "../voice/voice-recovery-notifications.js"; import { notifyChatReceiver, notifyChatConnectionReceiver, notifyChatCompanyReceivers } from "./runtime/receiver-work-notifications.js"; import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; @@ -6,6 +7,7 @@ import { notifyDiscordCommandRegistration } from "./discord/command-work-notific export async function notifyChatEndpointConfiguration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { if (endpoint.provider === "speko") await notifyVoiceChange(transaction, { kind: "endpoint", id: endpoint.id }); await notifyChatReceiver(transaction, endpoint); + if (endpoint.provider === "github") await notifyGitHubWebhookRecovery(transaction, { kind: "endpoint", id: endpoint.id }); if (endpoint.provider === "discord") { await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); } @@ -19,6 +21,7 @@ export async function notifyChatConnectionConfiguration(transaction: object, con export async function notifyChatConnectionObservation(transaction: object, connectionId: string, transport: ToolConnectionTransport) { await notifyChatConnectionReceiver(transaction, connectionId, transport); if (transport === "chat_sdk") { + await notifyGitHubWebhookRecovery(transaction, { kind: "connection", id: connectionId }); await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); } } @@ -26,5 +29,6 @@ export async function notifyChatConnectionObservation(transaction: object, conne export async function notifyChatCompanyConfiguration(transaction: object, companyId: string) { await notifyVoiceChange(transaction, { kind: "company", id: companyId }); await notifyChatCompanyReceivers(transaction, companyId); + await notifyGitHubWebhookRecovery(transaction, { kind: "company", id: companyId }); await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); } diff --git a/server/src/services/chat-channels/github/activity.ts b/server/src/services/chat-channels/github/activity.ts index 561e1fd7e12..e85717501d4 100644 --- a/server/src/services/chat-channels/github/activity.ts +++ b/server/src/services/chat-channels/github/activity.ts @@ -1,6 +1,6 @@ import { and, eq, sql } from "drizzle-orm"; import { chatEndpoints, chatSdkState, type Db } from "@paperclipai/db"; -import { GITHUB_RECOVERY_STATE_KEY } from "./webhook-ingress.js"; +import { GITHUB_RECOVERY_STATE_KEY } from "./webhook-recovery-state.js"; /** Direct callback recovery warnings do not describe Cloud relay delivery. */ export async function githubRecoveryActivity( diff --git a/server/src/services/chat-channels/github/installation-lifecycle.ts b/server/src/services/chat-channels/github/installation-lifecycle.ts index 15023223a91..b508a6a72df 100644 --- a/server/src/services/chat-channels/github/installation-lifecycle.ts +++ b/server/src/services/chat-channels/github/installation-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyGitHubWebhookRecovery } from "./webhook-recovery-notifications.js"; import type { VerifiedProviderIdentity } from "../runtime-types.js"; import type { EndpointRow, DbTransaction, LifecycleRuntimeFence, InternalSetupState } from "../runtime-types.js"; @@ -148,6 +149,7 @@ export function createGitHubInstallationLifecycle(db: Db, options: GitHubInstall currentInstallationRemovalReason ?? "GitHub App installation was removed", ).slice(0, MAX_ERROR_TEXT); + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: currentEndpoint.id }); await tx .update(chatEndpoints) .set({ @@ -247,6 +249,7 @@ export function createGitHubInstallationLifecycle(db: Db, options: GitHubInstall if (upgradingLegacyIdentity) { await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: currentEndpoint.id }); await tx .update(chatEndpoints) .set({ @@ -299,6 +302,7 @@ export function createGitHubInstallationLifecycle(db: Db, options: GitHubInstall await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); if (intentionallyUnavailable) { + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: currentEndpoint.id }); await tx .update(chatEndpoints) .set({ @@ -320,6 +324,7 @@ export function createGitHubInstallationLifecycle(db: Db, options: GitHubInstall canonical.endpoint.setup.step === "complete" ? "active" : "verifying"; + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: currentEndpoint.id }); await tx .update(chatEndpoints) .set({ @@ -401,6 +406,7 @@ export function createGitHubInstallationLifecycle(db: Db, options: GitHubInstall const terminal = candidate.attempts + 1 >= 5; await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: currentEndpoint.id }); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/github/webhook-ingress.ts b/server/src/services/chat-channels/github/webhook-ingress.ts index 771a486c9a9..e186b468a9a 100644 --- a/server/src/services/chat-channels/github/webhook-ingress.ts +++ b/server/src/services/chat-channels/github/webhook-ingress.ts @@ -1,3 +1,12 @@ +import { GITHUB_RECOVERY_STATE_KEY, GITHUB_RECOVERY_INTERVAL_MS, GITHUB_RECOVERY_WINDOW_MS, + GITHUB_RECOVERY_SCAN_LEASE_MS, githubRecoveryWindow, type GitHubRecoveryWindow } from "./webhook-recovery-state.js"; +import { githubWebhookRecoverySchedule } from "./webhook-recovery-schedule.js"; +import { registerGitHubWebhookRecoveryWork } from "./webhook-recovery-work.js"; +import { notifyGitHubWebhookRecovery } from "./webhook-recovery-notifications.js"; +import { createChatSdkStatePersistence } from "../sdk-state.js"; +import type { ChatSdkStateCompareAndSetInput } from "../../chat-sdk-state.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; + import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js"; import { githubRepositoryInventoryItemFromPayload } from "./resources.js"; @@ -74,38 +83,10 @@ import type { EndpointRecord } from "../runtime-types.js"; | { kind: "ignored" } | { actionId: string; kind: "staged" }; -export const GITHUB_RECOVERY_STATE_KEY = "paperclip:github-webhook-recovery:v1"; -export const GITHUB_RECOVERY_INTERVAL_MS = 60_000; -export const GITHUB_RECOVERY_WINDOW_MS = 60 * 60_000; -export const GITHUB_RECOVERY_SCAN_LEASE_MS = 5 * 60_000; -export type GitHubRecoveryWindow = LifecycleRuntimeFence & { - appId: string; - webhookUrl: string; - floor: string; - nextScanAt: string; -}; export type GitHubRecoveryReceipt = GitHubRecoveryWindow & { original: GitHubAppWebhookDeliveryDetail; }; -export function githubRecoveryWindow(value: unknown): GitHubRecoveryWindow | null { - if (!value || typeof value !== "object" || Array.isArray(value)) return null; - const item = value as GitHubRecoveryWindow; - return Number.isSafeInteger(item.generation) && - item.generation >= 0 && - typeof item.credentialFingerprint === "string" && - /^[a-f0-9]{64}$/.test(item.credentialFingerprint) && - typeof item.appId === "string" && - /^\d+$/.test(item.appId) && - typeof item.webhookUrl === "string" && - typeof item.floor === "string" && - Number.isFinite(Date.parse(item.floor)) && - typeof item.nextScanAt === "string" && - Number.isFinite(Date.parse(item.nextScanAt)) - ? item - : null; -} - export type GitHubWebhookIngressPayload = { version: 1; body: string; @@ -1073,6 +1054,7 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress async function githubRecoverySourceIsCurrent( detail: GitHubAppWebhookDeliveryDetail, credentials: Record, + sourceFetch: typeof fetch = fetchImpl, ): Promise { const comment = detail?.payload?.comment; if ( @@ -1082,7 +1064,7 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress ) return false; const current = await getGitHubRecoveryComment({ - fetch: fetchImpl, + fetch: sourceFetch, appToken: githubAppJwt(credentials.appId, credentials.privateKey), installationId: credentials.installationId, repositoryFullName: detail.payload.repositoryFullName, @@ -1161,424 +1143,468 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress ) .orderBy(asc(chatSdkState.updatedAt), asc(chatEndpoints.id)) .limit(Math.min(5, Math.max(1, limit))); - for (const { endpoint } of rows) { - const record = await endpointRecord(endpoint.id); - if (!record || !record.endpoint.botExternalId) continue; - const context = runtimeContextForRecord(record); - const webhookUrl = `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/github`; - const scope = { companyId: endpoint.companyId, endpointId: endpoint.id }; - const stored = await persistence.read(scope, GITHUB_RECOVERY_STATE_KEY); - const previous = githubRecoveryWindow(stored?.value); - const epochMatches = - previous && - previous.generation === context.generation && - previous.credentialFingerprint === context.credentialFingerprint && - previous.webhookUrl === webhookUrl && - previous.appId === record.endpoint.botExternalId; - if (epochMatches && Date.parse(previous.nextScanAt) > now.getTime()) - continue; - const window: GitHubRecoveryWindow = { - generation: context.generation, - credentialFingerprint: context.credentialFingerprint, - appId: record.endpoint.botExternalId, - webhookUrl, - floor: epochMatches ? previous.floor : now.toISOString(), - nextScanAt: new Date( - now.getTime() + GITHUB_RECOVERY_SCAN_LEASE_MS, - ).toISOString(), - }; + for (const { endpoint } of rows) await recoverGitHubWebhookEndpoint(endpoint.id); + return rows.length; + } + + async function saveRecoveryWindow(input: ChatSdkStateCompareAndSetInput) { + return db.transaction(async tx => { + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: input.endpointId }); + return createChatSdkStatePersistence(tx as unknown as Db).compareAndSet(input); + }); + } + + async function recoverGitHubWebhookEndpoint( + endpointId: string, + canRun: () => boolean = () => true, + signal?: AbortSignal, + ) { + const admitted = () => !isShuttingDown() && canRun() && !signal?.aborted; + if (!admitted() || !getWebhookPublicBaseUrl()?.startsWith("https://")) return; + const assertAdmitted = () => { + if (!admitted()) throw signal?.reason ?? new Error("GitHub webhook recovery stopped"); + }; + const recoveryFetch: typeof fetch = (input, init) => { + assertAdmitted(); + return fetchImpl(input, init); + }; + const now = new Date(); + const record = await endpointRecord(endpointId); + if (!record || !record.endpoint.botExternalId) return; + const endpoint = record.endpoint; + const context = runtimeContextForRecord(record); + const webhookUrl = `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/github`; + const scope = { companyId: endpoint.companyId, endpointId: endpoint.id }; + const stored = await persistence.read(scope, GITHUB_RECOVERY_STATE_KEY); + const previous = githubRecoveryWindow(stored?.value); + const epochMatches = + previous && + previous.generation === context.generation && + previous.credentialFingerprint === context.credentialFingerprint && + previous.webhookUrl === webhookUrl && + previous.appId === record.endpoint.botExternalId; + if (epochMatches && Date.parse(previous.nextScanAt) > now.getTime()) + return; + const window: GitHubRecoveryWindow = { + generation: context.generation, + credentialFingerprint: context.credentialFingerprint, + appId: record.endpoint.botExternalId, + webhookUrl, + floor: epochMatches ? previous.floor : now.toISOString(), + nextScanAt: new Date( + now.getTime() + GITHUB_RECOVERY_SCAN_LEASE_MS, + ).toISOString(), + }; + if ( + !(await saveRecoveryWindow({ + ...scope, + key: GITHUB_RECOVERY_STATE_KEY, + expectedVersion: stored?.version ?? null, + value: window, + expiresAt: null, + })) + ) + return; + const claimedVersion = (stored?.version ?? 0) + 1; + let nextDelay = GITHUB_RECOVERY_INTERVAL_MS; + let outcome = "scanned"; + let requested = 0; + let inspected = 0; + try { + if (!epochMatches) { + outcome = "epoch_initialized"; + return; + } + const credentials = await resolveCredentialRefs( + record.endpoint, + record.credentialSecretRefs, + ); + if (credentials.appId !== window.appId || !credentials.installationId) + return; + const appToken = githubAppJwt( + credentials.appId, + credentials.privateKey, + ); + const config = await readGitHubAppWebhookConfig({ + fetch: recoveryFetch, + appToken, + }); if ( - !(await persistence.compareAndSet({ - ...scope, - key: GITHUB_RECOVERY_STATE_KEY, - expectedVersion: stored?.version ?? null, - value: window, - expiresAt: null, - })) - ) - continue; - const claimedVersion = (stored?.version ?? 0) + 1; - let nextDelay = GITHUB_RECOVERY_INTERVAL_MS; - let outcome = "scanned"; - let requested = 0; - let inspected = 0; - try { - if (!epochMatches) { - outcome = "epoch_initialized"; - continue; - } - const credentials = await resolveCredentialRefs( - record.endpoint, - record.credentialSecretRefs, - ); - if (credentials.appId !== window.appId || !credentials.installationId) - continue; - const appToken = githubAppJwt( - credentials.appId, - credentials.privateKey, - ); - const config = await readGitHubAppWebhookConfig({ - fetch: fetchImpl, + config.url !== webhookUrl || + config.contentType !== "json" || + config.insecureSsl !== "0" + ) { + outcome = "callback_mismatch"; + nextDelay = 5 * 60_000; + return; + } + const attempts: GitHubAppWebhookDelivery[] = []; + const floor = Math.max( + Date.parse(window.floor), + now.getTime() - GITHUB_RECOVERY_WINDOW_MS, + ); + let cursor: string | undefined; + let truncated = false; + let previousTimestamp = Number.POSITIVE_INFINITY; + for (let page = 0; page < 3; page += 1) { + const result = await listGitHubAppWebhookDeliveries({ + fetch: recoveryFetch, appToken, + cursor, }); + + + + for (const attempt of result.deliveries) { + const timestamp = Date.parse(attempt.deliveredAt); + if (timestamp > previousTimestamp) + throw new Error("GitHub recovery history ordering changed"); + previousTimestamp = timestamp; + if (timestamp >= floor) attempts.push(attempt); + } + if (previousTimestamp < floor) break; + if (!result.nextCursor) break; + if (result.nextCursor === cursor) + throw new Error("GitHub recovery pagination did not advance"); + cursor = result.nextCursor; + truncated = page === 2; + } + + + if (truncated) { + outcome = "history_limit_reached"; + nextDelay = 5 * 60_000; + return; + } + const groups = new Map(); + for (const attempt of attempts) { + const group = groups.get(attempt.guid) ?? []; + group.push(attempt); + groups.set(attempt.guid, group); + } + for (const [guid, group] of groups) { if ( - config.url !== webhookUrl || - config.contentType !== "json" || - config.insecureSsl !== "0" - ) { - outcome = "callback_mismatch"; - nextDelay = 5 * 60_000; + inspected >= 5 || + isShuttingDown() || + Date.now() >= Date.parse(window.nextScanAt) + ) + break; + if ( + group.some( + (item) => + item.statusCode !== null && + item.statusCode >= 200 && + item.statusCode < 400, + ) + ) continue; - } - const attempts: GitHubAppWebhookDelivery[] = []; - const floor = Math.max( - Date.parse(window.floor), - now.getTime() - GITHUB_RECOVERY_WINDOW_MS, - ); - let cursor: string | undefined; - let truncated = false; - let previousTimestamp = Number.POSITIVE_INFINITY; - for (let page = 0; page < 3; page += 1) { - const result = await listGitHubAppWebhookDeliveries({ - fetch: fetchImpl, - appToken, - cursor, - }); - // GitHub's official recovery recipe uses newest-first pagination - // and stops at the age floor. Reject observed ordering drift rather - // than assuming unseen pages cannot contain a successful sibling. - for (const attempt of result.deliveries) { - const timestamp = Date.parse(attempt.deliveredAt); - if (timestamp > previousTimestamp) - throw new Error("GitHub recovery history ordering changed"); - previousTimestamp = timestamp; - if (timestamp >= floor) attempts.push(attempt); - } - if (previousTimestamp < floor) break; - if (!result.nextCursor) break; - if (result.nextCursor === cursor) - throw new Error("GitHub recovery pagination did not advance"); - cursor = result.nextCursor; - truncated = page === 2; - } - // All pages in this time window are needed to rule out a successful - // sibling; do not act on a partial window after hitting the work cap. - if (truncated) { - outcome = "history_limit_reached"; - nextDelay = 5 * 60_000; + const originals = group.filter((item) => !item.redelivery); + if (originals.length !== 1) continue; + const original = originals[0]!; + if ( + Date.parse(original.deliveredAt) < floor || + Date.parse(original.deliveredAt) > now.getTime() - 10_000 || + original.statusCode === null || + original.statusCode < 500 || + original.action !== "created" || + !["issue_comment", "pull_request_review_comment"].includes( + original.event, + ) || + original.installationId !== credentials.installationId || + !original.repositoryId || + !(await githubRecoveryResource(db, endpoint, original.repositoryId)) + ) continue; - } - const groups = new Map(); - for (const attempt of attempts) { - const group = groups.get(attempt.guid) ?? []; - group.push(attempt); - groups.set(attempt.guid, group); - } - for (const [guid, group] of groups) { - if ( - inspected >= 5 || - isShuttingDown() || - Date.now() >= Date.parse(window.nextScanAt) + const recoveryId = `github_webhook_recovery:${guid}`; + if (await githubRecoveryAlreadyReceived(db, endpoint, guid)) continue; + const receipt = await db + .select() + .from(chatActions) + .where( + and( + eq(chatActions.endpointId, endpoint.id), + eq(chatActions.providerActionId, recoveryId), + ), ) - break; - if ( - group.some( - (item) => - item.statusCode !== null && - item.statusCode >= 200 && - item.statusCode < 400, - ) - ) - continue; - const originals = group.filter((item) => !item.redelivery); - if (originals.length !== 1) continue; - const original = originals[0]!; - if ( - Date.parse(original.deliveredAt) < floor || - Date.parse(original.deliveredAt) > now.getTime() - 10_000 || - original.statusCode === null || - original.statusCode < 500 || - original.action !== "created" || - !["issue_comment", "pull_request_review_comment"].includes( - original.event, - ) || - original.installationId !== credentials.installationId || - !original.repositoryId || - !(await githubRecoveryResource(db, endpoint, original.repositoryId)) - ) - continue; - const recoveryId = `github_webhook_recovery:${guid}`; - if (await githubRecoveryAlreadyReceived(db, endpoint, guid)) continue; - const receipt = await db - .select() - .from(chatActions) - .where( - and( - eq(chatActions.endpointId, endpoint.id), - eq(chatActions.providerActionId, recoveryId), - ), - ) - .then((items) => items[0] ?? null); - const count = - typeof receipt?.result?.attempts === "number" - ? receipt.result.attempts - : 0; - const receiptWindow = githubRecoveryWindow(receipt?.payload); - const latest = [...group].sort( - (left, right) => - Date.parse(right.deliveredAt) - Date.parse(left.deliveredAt), - )[0]!; - if ( + .then((items) => items[0] ?? null); + const count = + typeof receipt?.result?.attempts === "number" + ? receipt.result.attempts + : 0; + const receiptWindow = githubRecoveryWindow(receipt?.payload); + const latest = [...group].sort( + (left, right) => + Date.parse(right.deliveredAt) - Date.parse(left.deliveredAt), + )[0]!; + if ( + latest.statusCode === null || + latest.statusCode < 500 || + latest.throttledAt !== null || + Date.parse(latest.deliveredAt) > now.getTime() - 10_000 + ) + continue; + if ( + receipt && + (receipt.status === "cancelled" || + !receiptWindow || + receiptWindow.generation !== window.generation || + receiptWindow.credentialFingerprint !== + window.credentialFingerprint || + receiptWindow.webhookUrl !== window.webhookUrl || + count >= 3 || + receipt.result?.latestAttemptId === latest.id || + !latest.redelivery || latest.statusCode === null || latest.statusCode < 500 || - latest.throttledAt !== null || - Date.parse(latest.deliveredAt) > now.getTime() - 10_000 - ) - continue; - if ( - receipt && - (receipt.status === "cancelled" || - !receiptWindow || - receiptWindow.generation !== window.generation || - receiptWindow.credentialFingerprint !== - window.credentialFingerprint || - receiptWindow.webhookUrl !== window.webhookUrl || - count >= 3 || - receipt.result?.latestAttemptId === latest.id || - !latest.redelivery || - latest.statusCode === null || - latest.statusCode < 500 || - Date.parse(latest.deliveredAt) < - Date.parse(String(receipt.result?.requestedAt)) || - Date.parse(String(receipt.result?.retryAt)) > now.getTime()) - ) - continue; - inspected += 1; - const detail = await getGitHubAppWebhookDelivery({ - fetch: fetchImpl, + Date.parse(latest.deliveredAt) < + Date.parse(String(receipt.result?.requestedAt)) || + Date.parse(String(receipt.result?.retryAt)) > now.getTime()) + ) + continue; + inspected += 1; + const detail = await getGitHubAppWebhookDelivery({ + fetch: recoveryFetch, + appToken, + deliveryId: original.id, + }); + const comment = detail.payload.comment; + if ( + detail.id !== original.id || + detail.guid !== guid || + detail.redelivery || + detail.url !== webhookUrl || + detail.event !== original.event || + detail.deliveredAt !== original.deliveredAt || + detail.statusCode !== original.statusCode || + detail.installationId !== original.installationId || + detail.repositoryId !== original.repositoryId || + detail.action !== "created" || + detail.payload.action !== "created" || + detail.payload.installationId !== credentials.installationId || + detail.payload.repositoryId !== original.repositoryId || + !comment || + comment.userType !== "User" || + detail.payload.senderId !== comment.userId || + Date.parse(comment.createdAt) < floor || + comment.createdAt !== comment.updatedAt || + !(await githubRecoveryResource( + db, + endpoint, + original.repositoryId, + detail.payload.repositoryFullName, + )) + ) { + await rememberGitHubRecoverySkip( + endpoint, + context, + window, + original, + "not_eligible_for_recovery", + ); + continue; + } + if (!(await githubRecoverySourceIsCurrent(detail, credentials, recoveryFetch))) { + await rememberGitHubRecoverySkip( + endpoint, + context, + window, + original, + "source_changed_or_unavailable", + ); + continue; + } + await withCredentialMutationLease(endpoint, async (lease) => { + await lease.assertOwned(); + const currentConfig = await readGitHubAppWebhookConfig({ + fetch: recoveryFetch, appToken, - deliveryId: original.id, }); - const comment = detail.payload.comment; if ( - detail.id !== original.id || - detail.guid !== guid || - detail.redelivery || - detail.url !== webhookUrl || - detail.event !== original.event || - detail.deliveredAt !== original.deliveredAt || - detail.statusCode !== original.statusCode || - detail.installationId !== original.installationId || - detail.repositoryId !== original.repositoryId || - detail.action !== "created" || - detail.payload.action !== "created" || - detail.payload.installationId !== credentials.installationId || - detail.payload.repositoryId !== original.repositoryId || - !comment || - comment.userType !== "User" || - detail.payload.senderId !== comment.userId || - Date.parse(comment.createdAt) < floor || - comment.createdAt !== comment.updatedAt || - !(await githubRecoveryResource( - db, - endpoint, - original.repositoryId, - detail.payload.repositoryFullName, - )) - ) { - await rememberGitHubRecoverySkip( - endpoint, + currentConfig.url !== webhookUrl || + currentConfig.contentType !== "json" || + currentConfig.insecureSsl !== "0" + ) + return; + const requestedAt = new Date().toISOString(); + const claimed = await db.transaction(async (tx) => { + await lease.assertOwned(tx); + const current = await runtimeCallbackEndpoint( + tx, + endpoint.id, context, - window, - original, - "not_eligible_for_recovery", + ["active"], ); - continue; - } - if (!(await githubRecoverySourceIsCurrent(detail, credentials))) { - await rememberGitHubRecoverySkip( - endpoint, - context, - window, - original, - "source_changed_or_unavailable", - ); - continue; - } - await withCredentialMutationLease(endpoint, async (lease) => { - await lease.assertOwned(); - const currentConfig = await readGitHubAppWebhookConfig({ - fetch: fetchImpl, - appToken, - }); if ( - currentConfig.url !== webhookUrl || - currentConfig.contentType !== "json" || - currentConfig.insecureSsl !== "0" - ) - return; - const claimed = await db.transaction(async (tx) => { - await lease.assertOwned(tx); - const current = await runtimeCallbackEndpoint( - tx, - endpoint.id, - context, - ["active"], - ); - if ( - !current || - !(await githubRecoveryResource( - tx, - current, - original.repositoryId!, - detail.payload.repositoryFullName, - )) - ) - return null; - const connection = await tx - .select({ id: toolConnections.id }) - .from(toolConnections) - .where( - and( - eq(toolConnections.id, current.connectionId), - eq(toolConnections.companyId, current.companyId), - eq(toolConnections.enabled, true), - eq(toolConnections.status, "active"), - ), - ) - .then((items) => items[0]); - const scan = await tx - .select({ id: chatSdkState.id }) - .from(chatSdkState) - .where( - and( - eq(chatSdkState.endpointId, endpoint.id), - eq(chatSdkState.stateKey, GITHUB_RECOVERY_STATE_KEY), - eq(chatSdkState.version, claimedVersion), - ), - ) - .then((items) => items[0]); - const local = await githubRecoveryAlreadyReceived( + !current || + !(await githubRecoveryResource( tx, current, - guid, - ); - if ( - !connection || - !scan || - local || - isShuttingDown() || - Date.now() >= Date.parse(window.nextScanAt) + original.repositoryId!, + detail.payload.repositoryFullName, + )) + ) + return null; + const connection = await tx + .select({ id: toolConnections.id }) + .from(toolConnections) + .where( + and( + eq(toolConnections.id, current.connectionId), + eq(toolConnections.companyId, current.companyId), + eq(toolConnections.enabled, true), + eq(toolConnections.status, "active"), + ), ) - return null; - const result = { - attempts: count + 1, - latestAttemptId: latest.id, - requestedAt: new Date().toISOString(), - retryAt: new Date( - Date.now() + 60_000 * 5 ** count, - ).toISOString(), - }; - const payload: GitHubRecoveryReceipt = { - ...window, - original: detail, - }; - const changes = receipt - ? await tx - .update(chatActions) - .set({ - status: "processing", - result, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, receipt.id), - eq(chatActions.updatedAt, receipt.updatedAt), - ), - ) - .returning({ id: chatActions.id }) - : await tx - .insert(chatActions) - .values({ - companyId: endpoint.companyId, - endpointId: endpoint.id, - kind: "github_webhook_recovery", - providerActionId: recoveryId, - payload, - status: "processing", - result, - }) - .onConflictDoNothing() - .returning({ id: chatActions.id }); - return changes[0]?.id ?? null; - }); - if (!claimed) return; - requested += 1; - // Commit the immutable denial fence before transport. Even an - // ambiguous POST is not reattempted without a distinct failed - // provider delivery. A stale scanner cannot overwrite a successor. + .then((items) => items[0]); + const scan = await tx + .select({ id: chatSdkState.id }) + .from(chatSdkState) + .where( + and( + eq(chatSdkState.endpointId, endpoint.id), + eq(chatSdkState.stateKey, GITHUB_RECOVERY_STATE_KEY), + eq(chatSdkState.version, claimedVersion), + ), + ) + .then((items) => items[0]); + const local = await githubRecoveryAlreadyReceived( + tx, + current, + guid, + ); + if ( + !connection || + !scan || + local || + isShuttingDown() || + Date.now() >= Date.parse(window.nextScanAt) + ) + return null; + const result = { + attempts: count + 1, + latestAttemptId: latest.id, + requestedAt, + retryAt: new Date( + Date.now() + 60_000 * 5 ** count, + ).toISOString(), + }; + const payload: GitHubRecoveryReceipt = { + ...window, + original: detail, + }; + const changes = receipt + ? await tx + .update(chatActions) + .set({ + status: "processing", + result, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, receipt.id), + eq(chatActions.updatedAt, receipt.updatedAt), + ), + ) + .returning({ id: chatActions.id }) + : await tx + .insert(chatActions) + .values({ + companyId: endpoint.companyId, + endpointId: endpoint.id, + kind: "github_webhook_recovery", + providerActionId: recoveryId, + payload, + status: "processing", + result, + }) + .onConflictDoNothing() + .returning({ id: chatActions.id }); + return changes[0]?.id ?? null; + }); + if (!claimed) return; + const ownsAttempt = and( + eq(chatActions.id, claimed), + eq(chatActions.status, "processing"), + sql`(${chatActions.result}->>'attempts')::int = ${count + 1}`, + sql`${chatActions.result}->>'latestAttemptId' = ${latest.id}`, + sql`${chatActions.result}->>'requestedAt' = ${requestedAt}`, + ); + let transportStarted = false; + try { await lease.assertOwned(); - try { - await requestGitHubAppWebhookRedelivery({ - fetch: fetchImpl, - appToken, - deliveryId: original.id, - }); - await db - .update(chatActions) - .set({ status: "processed", updatedAt: new Date() }) - .where( - and( - eq(chatActions.id, claimed), - eq(chatActions.status, "processing"), - sql`(${chatActions.result}->>'attempts')::int = ${count + 1}`, - ), - ); - } catch (error) { + await requestGitHubAppWebhookRedelivery({ + fetch: (input, init) => { + assertAdmitted(); + transportStarted = true; + requested += 1; + return fetchImpl(input, init); + }, + appToken, + deliveryId: original.id, + }); + await db + .update(chatActions) + .set({ status: "processed", updatedAt: new Date() }) + .where(ownsAttempt); + } catch (error) { + if (transportStarted) { await db .update(chatActions) .set({ status: "delivery_unknown", updatedAt: new Date() }) - .where( - and( - eq(chatActions.id, claimed), - eq(chatActions.status, "processing"), - sql`(${chatActions.result}->>'attempts')::int = ${count + 1}`, - ), - ); - throw error; + .where(ownsAttempt); + } else { + await db.transaction(async (tx) => { + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: endpoint.id }); + if (receipt) { + await tx + .update(chatActions) + .set({ status: receipt.status, result: receipt.result, updatedAt: receipt.updatedAt }) + .where(ownsAttempt); + } else { + await tx.delete(chatActions).where(ownsAttempt); + } + }); } - }); - } - } catch (error) { - outcome = "scan_failed"; - nextDelay = Math.max( - 5 * 60_000, - Math.min( - 24 * 60 * 60_000, - Number((error as { retryAfterMs?: unknown })?.retryAfterMs) || 0, - ), - ); - logger.warn( - { endpointId: endpoint.id, error: redactError(error) }, - "GitHub webhook recovery deferred", - ); - } finally { - await persistence.compareAndSet({ - ...scope, - key: GITHUB_RECOVERY_STATE_KEY, - expectedVersion: claimedVersion, - value: { - ...window, - nextScanAt: new Date(Date.now() + nextDelay).toISOString(), - outcome, - requested, - }, - expiresAt: null, + throw error; + } }); } + } catch (error) { + outcome = "scan_failed"; + nextDelay = Math.max( + 5 * 60_000, + Math.min( + 24 * 60 * 60_000, + Number((error as { retryAfterMs?: unknown })?.retryAfterMs) || 0, + ), + ); + logger.warn( + { endpointId: endpoint.id, error: redactError(error) }, + "GitHub webhook recovery deferred", + ); + } finally { + await saveRecoveryWindow({ + ...scope, + key: GITHUB_RECOVERY_STATE_KEY, + expectedVersion: claimedVersion, + value: { + ...window, + nextScanAt: new Date(Date.now() + nextDelay).toISOString(), + outcome, + requested, + }, + expiresAt: null, + }); } - return rows.length; + } + + const recoverySchedule = githubWebhookRecoverySchedule(db, { + runtimeGeneration, credentialFingerprint, getWebhookPublicBaseUrl, + recover: recoverGitHubWebhookEndpoint, + }); + function registerWebhookRecoveryWork(coordinator: ReturnType, canRun: () => boolean) { + return registerGitHubWebhookRecoveryWork(coordinator, recoverySchedule, () => canRun() && !isShuttingDown()); } async function processPendingGitHubWebhookIngress( @@ -1598,5 +1624,5 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress for (const action of actions) await processGitHubWebhookIngress(action.id); return actions.length; } - return { stageGitHubWebhookIngress, githubIngressAcceptedResponse, processGitHubWebhookIngress, processFailedGitHubWebhookDeliveries, processPendingGitHubWebhookIngress }; + return { registerWebhookRecoveryWork, stageGitHubWebhookIngress, githubIngressAcceptedResponse, processGitHubWebhookIngress, processFailedGitHubWebhookDeliveries, processPendingGitHubWebhookIngress }; } diff --git a/server/src/services/chat-channels/github/webhook-recovery-notifications.ts b/server/src/services/chat-channels/github/webhook-recovery-notifications.ts new file mode 100644 index 00000000000..11c29e06ee2 --- /dev/null +++ b/server/src/services/chat-channels/github/webhook-recovery-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; + +export type GitHubWebhookRecoveryScope = { kind: "endpoint" | "connection" | "company"; id: string }; + +export async function notifyGitHubWebhookRecovery(transaction: object, scope: GitHubWebhookRecoveryScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatGitHubWebhookRecovery, JSON.stringify(scope)); +} diff --git a/server/src/services/chat-channels/github/webhook-recovery-schedule.ts b/server/src/services/chat-channels/github/webhook-recovery-schedule.ts new file mode 100644 index 00000000000..fb0baaba3c9 --- /dev/null +++ b/server/src/services/chat-channels/github/webhook-recovery-schedule.ts @@ -0,0 +1,59 @@ +import { and, asc, eq, gt, isNotNull, sql } from "drizzle-orm"; +import { chatEndpoints, chatSdkState, toolConnections, type Db } from "@paperclipai/db"; +import type { ChatEndpointSetupState, ToolCredentialSecretRef } from "@paperclipai/shared"; +import { GITHUB_RECOVERY_STATE_KEY, githubRecoveryWindow } from "./webhook-recovery-state.js"; +import type { GitHubWebhookRecoveryScope } from "./webhook-recovery-notifications.js"; + +export type GitHubWebhookRecoveryDeadline = { id: string; companyId: string; connectionId: string; dueAt: number }; + +export function githubWebhookRecoverySchedule(db: Db, options: { + runtimeGeneration(setup: ChatEndpointSetupState): number; + credentialFingerprint(refs: ToolCredentialSecretRef[]): string; + getWebhookPublicBaseUrl(): string | null; + recover(endpointId: string, canRun: () => boolean, signal?: AbortSignal): Promise; +}) { + async function load(scope: GitHubWebhookRecoveryScope | undefined, canRun: () => boolean, signal?: AbortSignal) { + const entries: GitHubWebhookRecoveryDeadline[] = []; + const baseUrl = options.getWebhookPublicBaseUrl(); + if (!baseUrl?.startsWith("https://")) return entries; + let cursor: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await db.select({ endpoint: chatEndpoints, refs: toolConnections.credentialSecretRefs, + window: chatSdkState.value }).from(chatEndpoints).innerJoin(toolConnections, and( + eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatEndpoints.companyId), + )).leftJoin(chatSdkState, and(eq(chatSdkState.endpointId, chatEndpoints.id), + eq(chatSdkState.stateKey, GITHUB_RECOVERY_STATE_KEY))).where(and( + eq(chatEndpoints.provider, "github"), eq(chatEndpoints.status, "active"), + isNotNull(chatEndpoints.botExternalId), sql`${chatEndpoints.botExternalId} <> ''`, + sql`nullif(${chatEndpoints.setup}#>>'{github,cloudRegistrationId}', '') is null`, + eq(toolConnections.enabled, true), eq(toolConnections.status, "active"), + scope?.kind === "endpoint" ? eq(chatEndpoints.id, scope.id) : undefined, + scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined, + scope?.kind === "company" ? eq(chatEndpoints.companyId, scope.id) : undefined, + cursor ? gt(chatEndpoints.id, cursor) : undefined, + )).orderBy(asc(chatEndpoints.id)).limit(100); + for (const { endpoint, refs, window } of rows) { + const previous = githubRecoveryWindow(window); + const matches = previous && previous.generation === options.runtimeGeneration(endpoint.setup) && + previous.credentialFingerprint === options.credentialFingerprint(refs) && + previous.appId === endpoint.botExternalId && + previous.webhookUrl === `${baseUrl}/api/chat-webhooks/${endpoint.publicId}/github`; + entries.push({ id: endpoint.id, companyId: endpoint.companyId, connectionId: endpoint.connectionId, + dueAt: matches ? Date.parse(previous.nextScanAt) : 0 }); + } + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].endpoint.id; + } + throw signal?.reason ?? new Error("GitHub webhook recovery reconstruction stopped"); + } + async function check(id: string, canRun: () => boolean, signal?: AbortSignal) { + const scope = { kind: "endpoint", id } as const; + const [before] = await load(scope, canRun, signal); + if (!before || before.dueAt > Date.now()) return before ?? null; + if (!canRun() || signal?.aborted) return before; + await options.recover(id, canRun, signal); + const [after] = await load(scope, canRun, signal); + return after ?? null; + } + return { load, check }; +} diff --git a/server/src/services/chat-channels/github/webhook-recovery-state.ts b/server/src/services/chat-channels/github/webhook-recovery-state.ts new file mode 100644 index 00000000000..0bcef8cbc57 --- /dev/null +++ b/server/src/services/chat-channels/github/webhook-recovery-state.ts @@ -0,0 +1,30 @@ +import type { LifecycleRuntimeFence } from "../runtime-types.js"; + +export const GITHUB_RECOVERY_STATE_KEY = "paperclip:github-webhook-recovery:v1"; +export const GITHUB_RECOVERY_INTERVAL_MS = 60_000; +export const GITHUB_RECOVERY_WINDOW_MS = 60 * 60_000; +export const GITHUB_RECOVERY_SCAN_LEASE_MS = 5 * 60_000; +export type GitHubRecoveryWindow = LifecycleRuntimeFence & { + appId: string; + webhookUrl: string; + floor: string; + nextScanAt: string; +}; + +export function githubRecoveryWindow(value: unknown): GitHubRecoveryWindow | null { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const item = value as GitHubRecoveryWindow; + return Number.isSafeInteger(item.generation) && + item.generation >= 0 && + typeof item.credentialFingerprint === "string" && + /^[a-f0-9]{64}$/.test(item.credentialFingerprint) && + typeof item.appId === "string" && + /^\d+$/.test(item.appId) && + typeof item.webhookUrl === "string" && + typeof item.floor === "string" && + Number.isFinite(Date.parse(item.floor)) && + typeof item.nextScanAt === "string" && + Number.isFinite(Date.parse(item.nextScanAt)) + ? item + : null; +} diff --git a/server/src/services/chat-channels/github/webhook-recovery-work.ts b/server/src/services/chat-channels/github/webhook-recovery-work.ts new file mode 100644 index 00000000000..708ef8762cd --- /dev/null +++ b/server/src/services/chat-channels/github/webhook-recovery-work.ts @@ -0,0 +1,21 @@ +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; +import type { githubWebhookRecoverySchedule, GitHubWebhookRecoveryDeadline } from "./webhook-recovery-schedule.js"; +import type { GitHubWebhookRecoveryScope } from "./webhook-recovery-notifications.js"; + +export function registerGitHubWebhookRecoveryWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatGitHubWebhookRecovery, + batchSize: 5, + decodeScope: (hint: string): GitHubWebhookRecoveryScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: GitHubWebhookRecoveryDeadline, scope?: GitHubWebhookRecoveryScope) => !scope || + (scope.kind === "endpoint" ? entry.id : scope.kind === "company" ? entry.companyId : entry.connectionId) === scope.id, + load: (scope, signal) => service.load(scope, canRun, signal), + check: (entry, signal) => service.check(entry.id, canRun, signal), + }, canRun); +} diff --git a/server/src/services/chat-github-wizard.ts b/server/src/services/chat-github-wizard.ts index abe1e3a6754..cd5f38a1c75 100644 --- a/server/src/services/chat-github-wizard.ts +++ b/server/src/services/chat-github-wizard.ts @@ -1,3 +1,4 @@ +import { notifyGitHubWebhookRecovery } from "./chat-channels/github/webhook-recovery-notifications.js"; import type { AuthorizationActor } from "./authorization.js"; import { connectionManagementService, type ConnectionManagementContext } from "./connection-management.js"; import { createHash, randomBytes } from "node:crypto"; @@ -171,18 +172,22 @@ export function githubChatWizardService( bot: Endpoint, values: Partial>, ) { - await db - .update(chatEndpoints) - .set({ - setup: sql`jsonb_set(${chatEndpoints.setup}, '{github}', coalesce(${chatEndpoints.setup}->'github','{}'::jsonb) || ${JSON.stringify(values)}::jsonb)`, - updatedAt: new Date(), - }) - .where( - and( - eq(chatEndpoints.id, bot.id), - eq(chatEndpoints.companyId, bot.companyId), - ), - ); + await db.transaction(async tx => { + if (values.cloudRegistrationId !== undefined) + await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: bot.id }); + await tx + .update(chatEndpoints) + .set({ + setup: sql`jsonb_set(${chatEndpoints.setup}, '{github}', coalesce(${chatEndpoints.setup}->'github','{}'::jsonb) || ${JSON.stringify(values)}::jsonb)`, + updatedAt: new Date(), + }) + .where( + and( + eq(chatEndpoints.id, bot.id), + eq(chatEndpoints.companyId, bot.companyId), + ), + ); + }); } async function resume(bot: Endpoint) { const [company] = await db diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 4651bf27b31..2c2174eb20b 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -39,6 +39,7 @@ export const DELIVERY_QUEUES = { chatReceiptReactions: "chat-receipt-reactions", chatSessionStops: "chat-session-stops", chatProviderEffects: "chat-provider-effects", + chatGitHubWebhookRecovery: "chat-github-webhook-recovery", chatGitHubIngress: "chat-github-ingress", chatGitHubPublications: "chat-github-publications", chatGitHubReviewProjection: "chat-github-review-projection", diff --git a/tests/e2e/board-attachment-receipts.spec.ts b/tests/e2e/board-attachment-receipts.spec.ts index da87912cf87..a26baf2acf6 100644 --- a/tests/e2e/board-attachment-receipts.spec.ts +++ b/tests/e2e/board-attachment-receipts.spec.ts @@ -32,7 +32,12 @@ async function setup(page: Page, request: APIRequestContext, classic: boolean) { await page.addLocatorHandler( page.getByRole("complementary", { name: "Paperclip announcements", exact: true }) .getByRole("button", { name: "Dismiss announcement", exact: true }), - async (dismiss) => { await dismiss.click(); }, + async (dismiss) => { + const menuOpen = await page.getByRole("menu", { name: "Add to composer", exact: true }).isVisible(); + if (menuOpen) await page.keyboard.press("Escape"); + await dismiss.click(); + if (menuOpen) await page.getByRole("button", { name: "Add to composer", exact: true }).click(); + }, ); const company = await body<{ id: string; issuePrefix: string }>( await request.post("/api/companies", {