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 1e1d8913f72..d4446d4c927 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -17,7 +17,7 @@ Do not mark an item done from a timer removal or a green check alone. | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Approval retention [#17497](https://github.com/paperclipai/paperclip/pull/17497) and archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged; other source retention pending | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) merged; targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) merged | -| 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | #17554 repaired at `8c1d293d1a`; fresh CI/review in progress | +| 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | [#17554](https://github.com/paperclipai/paperclip/pull/17554) merged after all checks and both 5/5 reviews | | 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Slices 02–07 merged: [secret proposals](https://github.com/paperclipai/paperclip/pull/17446), [credentials](https://github.com/paperclipai/paperclip/pull/17450), [review expiry](https://github.com/paperclipai/paperclip/pull/17453), [approved recovery](https://github.com/paperclipai/paperclip/pull/17454), [interrupted claims](https://github.com/paperclipai/paperclip/pull/17458), [projection](https://github.com/paperclipai/paperclip/pull/17468) | | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | @@ -1483,6 +1483,42 @@ loops or changing domain backoff. All 152 selected command/coordinator/health, producer/Slack, delivery/standby/startup cases and server types/build pass. Publish for fresh CI/review. The full inventory, #17490 CI failure, and stack/master reconciliation remain open; no merge authority is implied. + + +## Slice 20 receiver implementation + +Receiver discovery now uses scoped committed configuration changes, startup +reconstruction, foreign-lease deadlines, and listener-stop completion. Voice and +all other maintenance reasons remain separate. See +[the receiver audit](2026-10-25-chat-receiver-discovery-sleep-audit.md). +A failed-scope deadline regression and a real concurrent listener-stop race were +reproduced and repaired. All 218 selected receiver/startup, Discord, Photon, +command/delivery and standby cases pass, with workspace typechecking passing. +Workspace build also passes; publication and fresh remote gates remain open. This is not +completion of the slice or the full stack. Master is refreshed at 0d73505098; +final stack/master reconciliation remains required. + +Parent Discord #17557 at d3187f76a4 and company cleanup #17561 at e8febb30eb now +have no failed or pending checks (Paperclip Review skipped). No merge was made. +The older #17490 E2E cancellation failure remains unresolved. The entire original +inventory stays in scope, with voice the next separate reason after receivers. + + +## Receiver review repair + +The real company-removal producer regression reproduced two inventory reads for +one observed removal. Discovery-initiated stops now suppress their own completion +wake; independent token loss still wakes discovery after listener retirement. +A full reconstruction also covers its already-pending scoped hints. Hints that +arrive during the load remain dirty, and narrower writes cannot bypass a failed +full reconstruction's retry deadline. + +Both redundant-load regressions failed before repair. All 45 selected receiver, +startup and real Discord producer tests pass after repair, including the added +full-refresh failure/backoff case. Server typecheck and build pass. Prior broad +consumer/workspace verification remains recorded above; these 45 overlap that +coverage and are not an additional unique total. Publication requires fresh +remote CI and review. The full migration inventory remains outstanding. ## Acceptance follow-up after the plugin parent merged (October 27) Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at @@ -1498,6 +1534,16 @@ status and test counts above are historical. The original remaining inventory is still in scope beyond the 65 published numbered slices; native backups and the remaining aggregate duties have not been migrated by this follow-up. +## Receiver follow-up with current Discord parent, October 27 + +Nineteen sleep PRs are merged through health #17554. Receiver discovery #17572 +now includes current Discord parent `cedf2030c4`, retaining the original 27-file +scope. Its existing slow-failure and stop-wake repairs have guarded mutation +proofs, 131 targeted cases pass, and both newly supported external fixtures +self-migrate. See [the owning audit](2026-10-25-chat-receiver-discovery-sleep-audit.md#current-parent-and-review-acceptance-october-27) +for exact evidence and the older CI runner-loss failure. Fresh CI/reviews and +parent landing are still required. The original remaining-duty inventory is unchanged. + ## Discord registration landing compatibility, October 27 @@ -1505,3 +1551,11 @@ Company deletion has landed. The Discord command slice retains only its original 26-file delta against the landed parent and current master at `f8563d2507`. All 141 targeted cases, server typecheck and build pass. The [owning audit](2026-10-25-discord-command-registration-sleep-audit.md#acceptance-after-the-company-deletion-parent-landed-october-27) records exact scope, fixture setup and remaining remote gates. + + +## Receiver integration with landed foundations, October 27 + +Receiver discovery keeps its 27-file scope on Discord parent `547fdf2139` and +preserves current master queue declarations. All 131 targeted cases, server +typecheck and build pass. See the [receiver audit](2026-10-25-chat-receiver-discovery-sleep-audit.md#landed-foundation-integration-acceptance-october-27). +Fresh exact-head review, CI and parent landing remain required. diff --git a/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md b/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md new file mode 100644 index 00000000000..5e037faed19 --- /dev/null +++ b/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md @@ -0,0 +1,351 @@ +# Chat receiver discovery sleep migration audit + +Parent: Discord command registration #17557, head d3187f76a4. +Master refreshed during this continuation: e6ec781a1a. +Branch: codex/sleep-20-chat-runtime-discovery. +Status: implementation and focused lifecycle verification; publication and final acceptance pending. + +## One reason for this PR + +Remove recurring inventory discovery for leased chat receivers from the aggregate +chat maintenance timer. Preserve receipt projection, credential fencing, runtime +initialization, takeover, and stop semantics. Discord command registration belongs +to the parent; voice reconciliation, publication maintenance, milestones, Teams, +and GitHub recovery remain separate migration reasons. + +`server/src/app.ts` currently invokes `chatChannels.reconcileProviderRuntimes()` +and `chatChannels.voice.reconcile()` in the same recurring maintenance lane. +The receiver method in `services/chat-channels.ts` queries every eligible Discord +or iMessage Photon endpoint, stops local ownership missing from that inventory, +then calls `runtimeFor` and `reconcileDiscordRuntime` for each eligible endpoint. +This query runs even when there are no receivers. Separating voice is necessary +before removing this callback from the maintenance lane; removing both would lose +an unmigrated obligation. + +## Durable truth and owner boundaries + +Existing `chat_endpoints`, `tool_connections`, and `chat_endpoint_leases` contain +the necessary configuration and ownership. No additional durable queue table is +needed. The receiver owner must retain the current eligibility policy: supported +leased provider; verifying/active/attention endpoint; active, enabled connection; +and exclusion of attention plus failed connection health. + +The receiver provider policy currently lives in `leasedChatProvider` in +`chat-channels.ts` and the lease-key selection inside `runtime/ownership.ts`. +Centralize that existing policy with the chat provider/receiver owner before +adding another independent Discord/Photon switch in scheduling. Shared delivery +coordination should consume normalized receiver identities and deadlines, not +provider protocols, credentials, or command-registration rules. + +`runtime/ownership.ts` owns acquisition, renewal, token fencing, and stop order. +Acquisition uses insert or expired-token CAS; background acquisition attempts once. +Foreground configuration can wait up to twenty seconds, checking every 100 ms. +That bounded active-operation wait is separate from idle inventory polling. +The current default live lease lasts fifteen seconds and renews every five +seconds while owned. These renewals are required active transport work and are +not removed in this slice. Existing inbound admission reclaims the exact token +after expiry before accepting buffered callbacks; preserve that safety boundary. + +## Proposed wake contract to verify in code + +- Reconstruct eligible receiver obligations at startup with bounded UUID pages. +- Notify before supported endpoint/connection eligibility or credential changes + in their existing transactions. Decode scoped hints only at the notification + boundary; narrow normal refreshes to an endpoint or connection. +- Keep locally owned healthy receivers in the inventory cache even when they have + no discovery deadline. Otherwise later removal cannot identify the runtime that + must stop. A cache of only future deadlines is insufficient for this duty. +- A locally healthy unchanged receiver needs no periodic rediscovery query. Its + active lease renewal and socket lifecycle remain owned by the receiver runtime. +- On a concrete foreign lease, schedule the saved expiry as the next takeover + attempt. Reload current authority at that time; a renewed foreign lease yields + its new expiry. This is a configured receiver's takeover obligation, not proof + that a standby with such an obligation can sleep indefinitely. +- Initialization/credential/database failures retain scoped retry/backoff. One + failed receiver must not starve independent companies or a healthy due backlog. +- After ownership loss, finish listener fencing/stop and lease release, then wake + discovery for that endpoint. Do not reinitialize before the old listener stops. + Avoid a callback cycle on shutdown or a permanent retry after disable/removal. +- Preserve a change arriving during refresh, drain, or stop. Use the existing + coordinator's settled hints and memory-only pending predicate; no passive SQL + during drain. Join admitted work before returning from shutdown. +- Retain the explicit/manual reconciliation entry point used by current callers, + or update those callers and their assertions deliberately. Do not leave a + second automatic periodic entry point invoking the old full inventory query. + +## Producer audit still required before implementation is accepted + +Existing helpers are starting points, not proof of complete coverage: + +1. `chat-work-notifications.ts:notifyChatEndpointWork` already covers shared + connection-origin delivery/command wakes. Inspect its callers individually. +2. `chat-channels/configuration.ts`: configure, reconnect, resume, pause, remove, + activation and rollback, including foreground acquisition and ownership waits. +3. `chat-channels/setup-probes.ts`, `provider-effects.ts`, resource reconciliation, + publication failures, and inbound activation: eligibility/health transitions. +4. Generic tool access, removal, health observations, credential/binding changes, + membership removal, and company deletion: these can change a receiver without + calling the endpoint configuration service. +5. `runtime/ownership.ts`: renewal failure, token loss, failed initialization, + runtime invalidation, lease release, and shutdown. A direct callback is suitable + for process-local completion; do not invent a SQL row for that callback. +6. Photon runtime callbacks and configuration: independently verify these paths; + the Discord parent producer audit is not evidence of Photon coverage. +7. Existing live replica/independent-writer assumptions: process-local transaction + hints cannot discover another process's arbitrary configuration writes. Retain + or explicitly integrate the supported cross-process wake contract rather than + silently removing its only recovery path. + +## Required evidence + +Start with a failing empty-instance regression proving that the aggregate chat +cadence repeatedly queries receiver inventory. Then verify: + +- empty, disabled, and retained-history states have no discovery SQL after startup; +- actual first configure/reconnect/disable/remove producers wake or cancel work; +- healthy local ownership adds no inventory scan beyond explicit mutations; +- foreign lease expiry, extension, and takeover preserve single ownership; +- renewal failure triggers bounded recovery only after fencing the old runtime; +- old buffered messages remain fenced after token or credential generation change; +- Photon and Discord preserve their existing distinct transport behavior; +- scoped changes, company deletion, nested transaction commit/rollback, uncertain + commit, page boundaries, failure isolation, concurrent changes, restart, + drain/resume, and shutdown retain obligations; +- shared coordinator behavior without this worker remains unchanged; +- the existing command-registration cases retain saved verification deadlines. + +Use real PostgreSQL for lease CAS and transaction ownership, and controlled local +runtime/provider fixtures for lifecycle effects. Existing chat integration tests +exercise manual reconciliation, foreign ownership, credential rotations, and +Photon restart. Inspect those cases before designing fixtures. SQL silence must +be measured after both tracked work and scheduler settlement; a null next-wake +value alone can occur during an admitted reload. + +## Sleep limits and remaining stack gates + +Active receiver sockets and lease renewals remain physical sleep blockers. Cloud +alarm coverage and admission/sleep-safety proof do not change. This migration +removes idle discovery activity, not the receiver's live transport obligation. +The full original recurring-server inventory remains in scope in the stack ledger. + +At audit creation, health #17554 has a passing security scan on 9ef6e4f4c8 but +other current-head checks remain running. Company cleanup #17561 is e8febb30eb. +Discord #17557 is d3187f76a4 with fresh checks required after its failure-path and +CI fixture repairs. The older #17490 E2E cancellation failure, final reconciliation +with master, and all fresh review/check gates remain outstanding. No stack PR was +merged and no new merge authority is implied. + + +Follow-up acceptance snapshot: health #17554 at 9ef6e4f4c8 now has no pending or +failed checks, passing security, and fresh Greptile 5/5. Its three reviewed findings +(fair batching, real producer coverage, inventory amplification) are implemented, +verified, and their threads resolved. This is acceptance of that current head, +not a claim that the full stack is complete or reconciled with moving master. +Company cleanup and Discord still have live current-head checks. + + +## Implementation and verification update + +The aggregate chat lane now calls voice reconciliation only. Receiver discovery +registers with the existing delivery coordinator. The receiver owner loads UUID +pages of 100, keeps owned identities with no discovery deadline, and processes +25 due receivers per fair company batch. Configuration hints refresh only their +endpoint, connection, or company. A foreign owner schedules its saved lease +expiry; a changed lease is reread before another attempt. Failures retain scoped +30-second exponential backoff capped at five minutes. Dirty scopes suppress their +cached deadlines until refresh succeeds; otherwise an old takeover deadline could +cause repeated useless dispatch while the required refresh is in backoff. + +The generic configuration notifier replaces the misleading command-only helper +name. Existing producers now notify both relevant owners. Shared delivery code +still receives only a queue topic and scope hint. Provider selection and lease +keys remain in the chat receiver owner. No schema or new per-worker table exists. + +Audited producer groups: + +| Owner | Receiver-relevant changes and integration | +| --- | --- | +| Configuration and setup probes | Existing connection helper covers configure, rotation, resume, pause, removal and completed setup before transaction writes | +| Credential persistence and removal | Configuration notifier before reference changes, including Photon | +| Inbound activation and resource reconciliation | Endpoint/connection notifier before status and health transitions | +| Publication, interaction and provider-effect failures | Endpoint configuration notifier before attention/failed-health writes | +| Discord Gateway lifecycle | Receiver hints before recovery/failure transitions; metadata-only writes do not create discovery work | +| Photon transport failure and runtime recovery | Failure callback now signals before attention/disable; existing assert-owned recovery uses the shared endpoint helper | +| Generic tool update/removal, manual health and access revocation | Existing chat connection helper, broadened configuration dispatch, and SDK transport filter | +| Company deletion | Company hint precedes endpoint removal in the owning deletion transaction | +| Ownership loss / invalidation | Local completion callback after listener stop and lease release; suppressed during shutdown | + +The additional direct-writer audit found publication-resolution, publication +success, provider-effect success, inbound filtering, and Discord root-thread +writes that only update timestamps. Endpoint display/reach preferences do not +change receiver eligibility or the runtime generation. Photon configuration +invalidates the runtime, writes allocation metadata, and proceeds through the +credential/configuration owner; its allocation metadata is not an independent +receiver identity. Slack/GitHub-only setup writers do not create leased receiver +work. Capability projection in Discord reconciliation retains its existing owner. + +A real token-loss regression found a concurrent configuration wake could acquire +another lease while the old listener was still stopping. The ownership map now +retains the stopping owner until listener shutdown and lease release complete. +Acquisition joins that owner's existing stop promise. Current-authority checks +still reject stopping owners. The test uses real timing and holds listener stop; +an earlier fake-clock observation returned before async SQL completed and did not +prove absence of overlap. The strengthened test failed before the ownership fix. + +Current passing evidence: 32 receiver/startup cases, 90 selected Discord +integration cases, all 20 Photon channel cases, and server typechecking. The +Discord selection includes configuration, credential rotation, pause/resume, +company deletion, token-loss stop ordering, and existing takeover/admission fences. +The dedicated real-DB worker takeover test also passes, as do 75 command/delivery/ +standby consumer cases and full workspace typechecking. This is 218 unique selected +passing cases. Workspace build passes. Remote review/check gates remain pending. +The full local test matrix is not claimed: embedded PostgreSQL is unavailable +on this host, so these database suites use validated disposable external URLs. +Remote CI must run the full matrix before acceptance. + +Recorded failures remain part of acceptance: initial receiver teardown omitted +agent deletion; initial Discord pause used incomplete onboarding; failed-scope +backoff initially selected the blocked cached takeover deadline; the real-timing +stop race reproduced overlapping listeners. The first two were fixture defects; +the latter two required implementation repairs. An initial backoff test expected +zero idle holds during failed intent recovery; the correct contract retains the +intent hold until reconciliation succeeds, so the test now checks the retry +deadline and SQL silence rather than weakening that hold. + +## Writer and sleep boundaries + +Notifications cover the active database owner and its transaction-sharing clients. +Warm standby does not run discovery; activation reconstructs its pending startup +inventory. An already known foreign receiver remains recoverable from lease expiry, +with real token CAS and callback fences. Arbitrary new configuration written by +an independent process or direct SQL is not discovered by process-local hints; +that deployment needs an explicit wake integration. This follows the existing +single active writer contract and must not be described as multi-writer support. +The manual reconciliation method remains available for explicit callers. + +Active sockets and lease renewals remain required work, and no Cloud admission +exemption is added. Voice, Teams, publication maintenance, milestones and other +inventory reasons remain separate PRs. The full stack is not complete. + +Master was fetched again at 0d73505098. Company prerequisite #17561 at e8febb30eb +and Discord parent #17557 at d3187f76a4 now have no failed or pending checks; +Paperclip Review is skipped. Fresh review-content reconciliation and final stack +rebasing/verification remain distinct from these check results. + + +## Receiver review repair + +The real company-removal producer regression reproduced two inventory reads for +one observed removal. Discovery-initiated stops now suppress their own completion +wake; independent token loss still wakes discovery after listener retirement. +A full reconstruction also covers its already-pending scoped hints. Hints that +arrive during the load remain dirty, and narrower writes cannot bypass a failed +full reconstruction's retry deadline. + +Both redundant-load regressions failed before repair. All 45 selected receiver, +startup and real Discord producer tests pass after repair, including the added +full-refresh failure/backoff case. Server typecheck and build pass. Prior broad +consumer/workspace verification remains recorded above; these 45 overlap that +coverage and are not an additional unique total. Publication requires fresh +remote CI and review. The full migration inventory remains outstanding. + +## Healthy receiver progress after a slow failed batch + +The receiver worker already stores a retry deadline for each failed receiver and +scope. It now supplies that deadline calculation through the coordinator's +`nextRunAtOnError` contract as well. A completed batch can take longer than the +one-second batching interval. When that batch contained an error, the coordinator +previously ignored an already-due healthy backlog and waited for its generic +30-second retry. The changed registration lets healthy due entries proceed while +the failed receiver retains its own exponential backoff. + +The existing 105-receiver PostgreSQL regression now runs with both an immediate +failure and a controlled 1,100 ms elapsed interval in the failed provider call. +The latter failed before the repair: its next wake was 30 seconds away although +healthy receivers were due. Afterward, all 105 are attempted before retrying the +failed receiver, which is attempted once during the healthy drain. Recovery then +settles and two former poll periods add no SQL. The assertion helper advances its +own fake clock; the test no longer adds an extra second on every assertion retry. + +All 20 receiver lifecycle cases and the server production typecheck pass on the +owning branch. This changes one coordinator registration and adds no provider +branch, queue, table, or independent timer. Current-head CI and combined-stack +verification remain separate gates. + +## Current-parent and review acceptance, October 27 + +The branch normally integrates Discord parent #17557 at +`cedf2030c49c3650a41b76a4767176fd78170819` through `3e957d1204`. Only the +original 27 receiver-owned paths differ from that parent. Health #17554 is now +merged; company cleanup #17561 and Discord #17557 still precede this slice. +Older stack status above describes its recorded source revision. + +The latest old review summaries still mention the delay and redundant-stop +findings, but the actual source already contains their repairs: stop-wake +suppression at `1604a9fcba` and the error-deadline registration at `ab56393acc`. +Both were revalidated on the integrated parent rather than duplicated. A guarded +mutation removing `nextRunAtOnError` fails the 1,100 ms slow-provider next-deadline +assertion. The expected error also remains visible in teardown after that early +assertion. A separate mutation reenabling discovery-initiated stop notification +fails the real company-removal producer assertion with two inventory reads +instead of one. Both source files are restored byte for byte after the checks. +Normal worker and real producer suites pass with the original fixes. + +The two newly supported external-database fixtures initially skipped migrations. +Actual worker reconstruction and Photon transport-loss cases failed on separate +empty PostgreSQL 17 databases with missing-table errors. Both fixtures now apply +pending migrations before client use. On those same initially empty databases, +the complete receiver file passes all 20 cases and the complete Photon channel +file passes all 20 cases. Existing embedded setup and safety validation remain. + +Current acceptance: five complete receiver/coordinator/batching/startup/GitHub +error-projection suites pass 99 cases. Twelve actual Discord automatic-command, +receiver configure/rotation/pause/resume/removal, ownership-loss/takeover, and Slack +held-transaction cases pass separately; 1,581 other chat integration cases are +not counted as run. With the 20 Photon cases this is 131 distinct targeted cases. +The large chat producer suite uses a separate migrated clone. Server production +typecheck and build pass on the integrated source; the later fixture-only changes +add migration setup and do not change production code. Startup listener-count +warnings and expected controlled provider failures remain in the logs. + +Required CI at the older `ab56393acc` head failed server shard 12. GitHub's +check-run annotation says the self-hosted runner lost communication with the +server; its grouped-test step is marked cancelled. The retained log ends without +a failed assertion or final test summary. That annotation does not establish the +underlying host, resource, or network cause, and the interrupted job is not a +passing result. The new parent integration and fixture repairs require a fresh +full CI run; local passes do not replace it. Fresh exact-head Greptile and +Ponytail reviews remain mandatory. No deployed sleep or full local suite is claimed. + + +## Landed-foundation integration acceptance, October 27 + +This slice normally integrates Discord parent `547fdf2139`, which contains +master `f8563d2507` and landed company-deletion fixes. The original 27-file +scope is unchanged. All owned source matches its precise original patch applied +to that parent. The shared queue catalog also preserves master's task-drain and +native-checkpoint declarations. An initial whole-file identity assertion rejected +those expected parent additions; the corrected proof checks the actual owned +patch against the integrated parent. No failed verification is counted as a pass. + +The complete five worker/shared-consumer suites pass 99 cases. Real Discord and +Slack producers pass 12 cases; the complete Photon suite passes 20. This is 131 +distinct targeted cases, with 1,581 chat cases deselected. Receiver and Photon +self-migrate separate initially empty databases. The large chat fixture uses a +separate migrated clone. Server typecheck and build pass. Only documentation +changes follow the passing source. Expected controlled provider failures and +startup signal-listener warnings remain recorded. Fresh exact-head review and +all CI must pass before this child lands after Discord. No deployed sleep or +full local suite result is inferred from these checks. + + +## Compatibility after Discord squash merge, October 27 + +Discord #17557 landed at `5bf0142d45`. The normal merge of that master caused +ancestry conflicts in overlapping producer paths. Resolution applies only the +recorded 27-file receiver patch from Discord parent `547fdf2139` to actual landed +master. Every server and package production/test file is byte-identical to the +published `6e88ca0201` source that passes 131 cases, typecheck and build. No source +verification is rerun merely for unrelated upstream UI/eval/documentation changes. +The PR-owned delta passes its whitespace check. This integration records that +proof and requires fresh exact-head review and CI before merging into master. diff --git a/doc/plans/2026-10-25-configuration-worker-drain-repair.md b/doc/plans/2026-10-25-configuration-worker-drain-repair.md index 9b2d8c64483..03501ef329e 100644 --- a/doc/plans/2026-10-25-configuration-worker-drain-repair.md +++ b/doc/plans/2026-10-25-configuration-worker-drain-repair.md @@ -43,3 +43,19 @@ followed by registration on resume without another producer. Production server typecheck passes in /tmp/sleep19-drain-typecheck.log. These tests use the migrated paperclip_sleep19_drain_20261025_test database. Remote CI/review must run on the new head; this is not evidence of provider suspension. + +## Receiver discovery — slice 20 / #17572 + +The strengthened first-receiver drain case fails with active=1 after commit in +/tmp/sleep20-drain-before.log. Apply the same owner-level pending predicate while +retaining all dirty scopes and lease/retry deadlines. All 19 receiver lifecycle +cases pass in /tmp/sleep20-drain-tests.log, including no SQL and an unchanged idle +snapshot for a minute after settlement, then receiver acquisition on resume. +Production server typecheck passes in /tmp/sleep20-drain-typecheck.log. The fresh +migrated database is paperclip_sleep20_drain_20261025_test. + +The three configuration owners now have measured drain activity as well as traffic +checks. Their retained timers remain memory-only while paused; they do not provide +external wake coverage. Fresh remote CI/review and downstream propagation remain +required. Origin/master was refreshed to e335265568; the latest change removes +idle task-run UI polling and does not replace these worker repairs. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 0dbde42a5a9..d715145e5fa 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -25,7 +25,7 @@ describe("createChatReconciliationCoordinator", () => { processFailedGitHubWebhookDeliveries, processPendingDeliveries, flushPublications, - reconcileProviderRuntimes: async () => undefined, + reconcileVoiceSessions: async () => undefined, projectRunMilestones: async () => 0, processPendingSlackFileUploadReceipts: async () => undefined, processPendingSlackSessionSyncs: async () => undefined, @@ -74,6 +74,14 @@ describe("createChatReconciliationCoordinator", () => { }, ); + it("keeps receiver inventory out of periodic chat maintenance while preserving voice", () => { + const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); + const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals")); + expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); + expect(maintenance).toContain("chatChannels.voice.reconcile()"); + expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); + }); + it("wires GitHub recovery into its independent lane", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); expect(source).toContain( @@ -135,7 +143,7 @@ describe("createChatReconciliationCoordinator", () => { const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones, - reconcileProviderRuntimes: async () => undefined, + reconcileVoiceSessions: async () => undefined, processPendingDeliveries: async () => undefined, flushPublications, processPendingSlackFileUploadReceipts: async () => undefined, @@ -198,7 +206,7 @@ describe("createChatReconciliationCoordinator", () => { const coordinator = createChatReconciliationCoordinator({ projectRunMilestones, flushPublications, - reconcileProviderRuntimes: async () => undefined, + reconcileVoiceSessions: async () => undefined, processPendingDeliveries: async () => undefined, processPendingSlackFileUploadReceipts: async () => undefined, processPendingSlackSessionSyncs: async () => undefined, @@ -238,7 +246,7 @@ describe("createChatReconciliationCoordinator", () => { const slackStatusReleased = new Promise((resolve) => { releaseSlackStatus = resolve; }); - const reconcileProviderRuntimes = vi.fn(async () => undefined); + const reconcileVoiceSessions = vi.fn(async () => undefined); const processPendingDeliveries = vi.fn(async () => deliveryReleased); const flushPublications = vi.fn(async () => undefined); const processPendingSlackFileUploadReceipts = vi.fn(async () => undefined); @@ -248,7 +256,7 @@ describe("createChatReconciliationCoordinator", () => { const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones: async () => 0, - reconcileProviderRuntimes, + reconcileVoiceSessions, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, @@ -258,7 +266,7 @@ describe("createChatReconciliationCoordinator", () => { coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(1); + expect(reconcileVoiceSessions).toHaveBeenCalledTimes(1); expect(processPendingDeliveries).toHaveBeenCalledTimes(1); expect(flushPublications).toHaveBeenCalledTimes(1); expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(1); @@ -266,7 +274,7 @@ describe("createChatReconciliationCoordinator", () => { }); coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(2); + expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2); expect(flushPublications).toHaveBeenCalledTimes(2); expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2); }); @@ -293,14 +301,14 @@ describe("createChatReconciliationCoordinator", () => { await receiptReleased; if (rejects) throw lookupError; }); - const reconcileProviderRuntimes = vi.fn(async () => undefined); + const reconcileVoiceSessions = vi.fn(async () => undefined); 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({ projectRunMilestones: async () => 0, - reconcileProviderRuntimes, + reconcileVoiceSessions, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, @@ -318,7 +326,7 @@ describe("createChatReconciliationCoordinator", () => { }); coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(2); + expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2); expect(processPendingDeliveries).toHaveBeenCalledTimes(2); expect(flushPublications).toHaveBeenCalledTimes(2); expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(2); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index d62ef466d44..ceecf407f5e 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,4 @@ +import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; import { deleteCompany } from "../services/company-deletion.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "../services/adapter-execution-control.js"; import { chatRunSourceMessageId } from "../services/chat-channels/run-origin.js"; @@ -7133,6 +7134,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { async function registrationFixture( mode: "success" | "unavailable" | "unknown" = "success", + receiverOptions: Partial> = {}, ) { const fixture = await seedCompany(); const applicationId = uniqueDiscordApplicationId(); @@ -7190,7 +7192,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return Response.json(command); }, ) as unknown as typeof globalThis.fetch; - const context = createService(new FakeChatSdkRuntime(), providerFetch); + const context = createService(new FakeChatSdkRuntime(), providerFetch, receiverOptions); const endpoint = await context.service.create( fixture.companyId, { @@ -7243,6 +7245,125 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("drives receiver discovery through real configure, rotation, pause, and resume producers", async () => { + const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 600_000, discordGatewayLeaseRenewalIntervalMs: 300_000 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const quiet = async () => { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(20_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + }; + try { + await f.service.registerReceiverWork(coordinator, () => true).ready; + await settle(); await quiet(); + await f.configure(); await settle(); + const first = f.runtime.get(f.endpoint.id); expect(first).not.toBeNull(); + await quiet(); + await f.service.configure(f.endpoint.id, { action: "reconnect", credentials: { ...f.credentials, botToken: "rotated-receiver-token" } }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).not.toBe(first); expect(first!.shutdown).toHaveBeenCalledOnce(); + // Pause requires completed onboarding; configure/reconnect intentionally + // leave this fixture waiting for its first test conversation. + const configured = await f.service.get(f.endpoint.id); + await db.update(chatEndpoints).set({ status: "active", setup: { ...configured.setup, step: "complete" } }) + .where(eq(chatEndpoints.id, f.endpoint.id)); + await f.service.configure(f.endpoint.id, { action: "pause" }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).toBeNull(); await quiet(); + await f.service.configure(f.endpoint.id, { action: "resume" }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).not.toBeNull(); await quiet(); + await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).toBeNull(); await quiet(); + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); await f.service.shutdown(); vi.useRealTimers(); + await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + + it("does not rediscover a receiver stopped by a company removal refresh", async () => { + const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 600_000, discordGatewayLeaseRenewalIntervalMs: 300_000 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + let reads: ReturnType | undefined; + try { + await f.configure(); await f.service.registerReceiverWork(coordinator, () => true).ready; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); }); + const owned = f.runtime.get(f.endpoint.id)!; + reads = vi.spyOn(db, "select"); + await db.transaction(async tx => { + await notifyChatCompanyConfiguration(tx, f.fixture.companyId); + await tx.delete(chatEndpoints).where(eq(chatEndpoints.id, f.endpoint.id)); + }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); }); + expect(owned.shutdown).toHaveBeenCalledOnce(); + expect(reads).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled(); + } finally { + reads?.mockRestore(); await coordinator.stop(); vi.useRealTimers(); await f.service.shutdown(); + await db.update(chatEndpoints).set({ status: "archived" }).where(eq(chatEndpoints.id, f.endpoint.id)); + } + }); + + it("takes over a persisted foreign receiver lease at its deadline through the worker", async () => { + const f = await registrationFixture(); + await f.configure(); await f.service.shutdown(); + const token = randomUUID(), expiresAt = new Date(Date.now() + 500); + await db.insert(chatEndpointLeases).values({ companyId: f.fixture.companyId, endpointId: f.endpoint.id, + leaseKey: "discord_gateway_runtime", token, expiresAt }); + const standby = createService(new FakeChatSdkRuntime(), f.providerFetch); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await standby.service.registerReceiverWork(coordinator, () => true).ready; + expect(standby.runtime.get(f.endpoint.id)).toBeNull(); + expect(coordinator.nextWakeAt()).toBeGreaterThanOrEqual(expiresAt.getTime()); + await vi.waitFor(() => expect(standby.runtime.get(f.endpoint.id)).not.toBeNull(), { timeout: 5000 }); + const [lease] = await db.select().from(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, f.endpoint.id)); + expect(lease.token).not.toBe(token); + expect(lease.expiresAt.getTime()).toBeGreaterThan(expiresAt.getTime()); + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); await retireRegistrationFixture(standby.service, f.endpoint.id); + } + }); + + it("restarts a receiver after token loss only after the old listener has stopped", async () => { + const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 60_000, discordGatewayLeaseRenewalIntervalMs: 50 }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + let release!: () => void; + const stopped = new Promise(resolve => { release = resolve; }); + try { + await f.configure(); + await f.service.registerReceiverWork(coordinator, () => true).ready; + const old = f.runtime.get(f.endpoint.id)!; + old.shutdown.mockImplementation(async () => { await stopped; }); + await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, f.endpoint.id)); + await vi.waitFor(() => expect(old.shutdown).toHaveBeenCalledOnce()); + await db.transaction(tx => notifyChatEndpointConfiguration(tx, f.endpoint)); + await new Promise(resolve => setTimeout(resolve, 1500)); + expect(f.runtime.get(f.endpoint.id)).toBeNull(); + release(); + await vi.waitFor(async () => { + expect(f.runtime.get(f.endpoint.id)).not.toBeNull(); + expect(f.runtime.get(f.endpoint.id)).not.toBe(old); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + expect(onError).not.toHaveBeenCalled(); + } finally { + release(); await coordinator.stop(); vi.useRealTimers(); + await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + it("learns the first configured command and cancels its deadline when the endpoint is removed", async () => { const f = await registrationFixture(); vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); diff --git a/server/src/__tests__/chat-receiver-work.test.ts b/server/src/__tests__/chat-receiver-work.test.ts new file mode 100644 index 00000000000..0beb5a612ba --- /dev/null +++ b/server/src/__tests__/chat-receiver-work.test.ts @@ -0,0 +1,245 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { applyPendingMigrations, agents, chatEndpoints, chatEndpointLeases, companies, createDb, toolApplications, toolConnections } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyDeliveryWork, DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { chatReceiverSchedule } from "../services/chat-channels/runtime/receiver-schedule.js"; +import { registerChatReceiverWork } from "../services/chat-channels/runtime/receiver-work.js"; +import { notifyChatEndpointConfiguration, notifyChatConnectionConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RECEIVER_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("chat receiver discovery", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, agentId: string, applicationId: string; + let coordinator: ReturnType | undefined; + let enabled = true, primary = true; + const errors: unknown[] = []; + const reconcile = vi.fn(async (_id: string): Promise<"owned" | "foreign"> => "owned"), stop = vi.fn(async (_id: string) => {}); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("chat-receivers-"); + if (external) await applyPendingMigrations(external); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 60_000); + beforeEach(async () => { + companyId = randomUUID(); agentId = randomUUID(); applicationId = randomUUID(); enabled = true; primary = true; errors.length = 0; + reconcile.mockReset().mockResolvedValue("owned"); stop.mockReset().mockResolvedValue(undefined); + await db.insert(companies).values({ id: companyId, name: "Receivers", issuePrefix: `R${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Receiver", adapterType: "codex_local" }); + await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Receiver", type: "chat" }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select 1`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); + await db.delete(agents).where(eq(agents.companyId, companyId)); + await db.delete(companies).where(eq(companies.id, companyId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function endpoint(patch: Partial = {}, writer = db) { + const id = patch.id ?? randomUUID(), connectionId = randomUUID(); + return writer.transaction(async tx => { + await notifyChatEndpointConfiguration(tx, { id, provider: patch.provider ?? "discord" }); + await tx.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: id, uid: id, + transport: "chat_sdk", connectionPurpose: "channel", enabled: true, status: "active" }); + return (await tx.insert(chatEndpoints).values({ id, companyId, connectionId, assignedAgentId: agentId, + publicId: id, provider: "discord", status: "active", ...patch }).returning())[0]; + }); + } + async function start(service = chatReceiverSchedule(db, { reconcile, stop })) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled && primary, canReconcile: () => primary, onError: error => errors.push(error) }); + const worker = registerChatReceiverWork(coordinator, service, () => enabled && primary); + await worker.ready; + return { worker, service }; + } + async function settled(expected: number | null = null) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBe(expected); + }, { timeout: 5000 }); + } + async function quiet(ms = 120_000) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + try { + await vi.advanceTimersByTimeAsync(ms); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); execute.mockRestore(); } + } + it("does no recurring SQL after empty reconstruction", async () => { await start(); await settled(); await quiet(); }); + it("leaves disabled, archived, and webhook-only endpoints quiet", async () => { + await endpoint({ status: "paused" }); await endpoint({ status: "archived" }); await endpoint({ provider: "slack" }); + const disabled = await endpoint(); await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, disabled.connectionId)); + const failed = await endpoint({ status: "attention" }); await db.update(toolConnections).set({ healthStatus: "failed" }).where(eq(toolConnections.id, failed.connectionId)); + await start(); await settled(); await quiet(); expect(reconcile).not.toHaveBeenCalled(); + }); + it.each(["discord", "imessage-photon"] as const)("wakes the first %s receiver and retains its identity until removal", async provider => { + await start(); await settled(); await quiet(); + const row = await endpoint({ provider }); await settled(); + expect(reconcile).toHaveBeenCalledWith(row.id); await quiet(); + await db.transaction(async tx => { await notifyChatConnectionConfiguration(tx, row.connectionId); await tx.delete(chatEndpoints).where(eq(chatEndpoints.id, row.id)); }); + await settled(); expect(stop).toHaveBeenCalledWith(row.id); await quiet(); + }); + it("waits for outer commit and discards rolled-back configuration", async () => { + await start(); await settled(); + await expect(db.transaction(async tx => { await endpoint({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(60_000); expect(reconcile).not.toHaveBeenCalled(); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await settled(); await quiet(); + let id = ""; + await db.transaction(async tx => { id = (await endpoint({}, tx as unknown as typeof db)).id; await vi.advanceTimersByTimeAsync(60_000); expect(reconcile).not.toHaveBeenCalled(); }); + await settled(); expect(reconcile).toHaveBeenCalledWith(id); + }); + it("preserves the first receiver written during drain without passive SQL", async () => { + await start(); await settled(); enabled = false; const row = await endpoint(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + }); + const drained = idleWorkSnapshot(); + await quiet(60_000); expect(idleWorkSnapshot()).toEqual(drained); expect(reconcile).not.toHaveBeenCalled(); + enabled = true; await settled(); expect(reconcile).toHaveBeenCalledWith(row.id); + }); + it("reconstructs after standby becomes primary without restarting the worker", async () => { + await endpoint(); primary = false; await start(); await quiet(60_000); + primary = true; await settled(); expect(reconcile).toHaveBeenCalledOnce(); await quiet(); + }); + it("waits for a foreign lease, observes its extension, and takes over after release", async () => { + const row = await endpoint(); let owned = false; + reconcile.mockImplementation(async () => owned ? "owned" : "foreign"); + const expiry = Date.now() + 60_000; + await db.insert(chatEndpointLeases).values({ companyId, endpointId: row.id, leaseKey: "discord_gateway_runtime", token: randomUUID(), expiresAt: new Date(expiry) }); + await start(); await settled(expiry); await quiet(10_000); expect(reconcile).toHaveBeenCalledOnce(); + const renewed = expiry + 60_000; + await db.update(chatEndpointLeases).set({ expiresAt: new Date(renewed) }).where(eq(chatEndpointLeases.endpointId, row.id)); + await vi.advanceTimersByTimeAsync(expiry - Date.now()); await settled(renewed); expect(reconcile).toHaveBeenCalledTimes(2); + await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, row.id)); owned = true; + await vi.advanceTimersByTimeAsync(renewed - Date.now()); await settled(); expect(reconcile).toHaveBeenCalledTimes(3); await quiet(); + }); + it("wakes after a local receiver stops and then returns to quiet", async () => { + const row = await endpoint(); const { worker } = await start(); await settled(); await quiet(); + worker.changed(row.id); await settled(); expect(reconcile).toHaveBeenCalledTimes(2); await quiet(); + }); + it("retains a producer arriving during its scoped refresh", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(); let inserted = false; + service.load = async (...args) => { const rows = await original(...args); if (!inserted) { inserted = true; await endpoint(); } return rows; }; + await endpoint(); await settled(); expect(reconcile).toHaveBeenCalledTimes(2); await quiet(); + }); + it("cancels all cached owned receivers after a company deletion hint", async () => { + await endpoint(); await endpoint(); await start(); await settled(); + await db.transaction(async tx => { await notifyChatCompanyConfiguration(tx, companyId); await tx.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); }); + await settled(); expect(stop).toHaveBeenCalledTimes(2); await quiet(); + }); + it("retains transient load failures and does not repeatedly scan during their backoff", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(); let broken = true; + service.load = async (...args) => { if (broken) throw new Error("load unavailable"); return original(...args); }; + await endpoint(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }); + await quiet(10_000); broken = false; await settled(); expect(reconcile).toHaveBeenCalledOnce(); errors.length = 0; + }); + it("does not schedule blocked takeover attempts before a failed scope refresh can retry", async () => { + const row = await endpoint(); + reconcile.mockResolvedValue("foreign"); + const expiry = Date.now() + 10_000; + await db.insert(chatEndpointLeases).values({ companyId, endpointId: row.id, leaseKey: "discord_gateway_runtime", token: randomUUID(), expiresAt: new Date(expiry) }); + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(expiry); + service.load = async () => { throw new Error("scoped inventory unavailable"); }; + await db.transaction(tx => notifyChatConnectionConfiguration(tx, row.connectionId)); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(errors).toHaveLength(1)); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now() + 20_000); + await quiet(10_000); + service.load = original; reconcile.mockResolvedValue("owned"); + await settled(); errors.length = 0; await quiet(); + }); + it("coalesces a full refresh with its already pending scoped hints", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }); + await start(service); await settled(); + const load = vi.spyOn(service, "load"); + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*"); + await endpoint({}, tx as unknown as typeof db); + await endpoint({}, tx as unknown as typeof db); + }); + await settled(); expect(reconcile).toHaveBeenCalledTimes(2); + expect(load).toHaveBeenCalledOnce(); expect(load.mock.calls[0][0]).toBeUndefined(); + load.mockRestore(); await quiet(); + }); + it("keeps full-refresh backoff when a narrower producer arrives", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(); + const load = vi.spyOn(service, "load").mockRejectedValue(new Error("inventory unavailable")); + await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*")); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); }); + const retryAt = coordinator!.nextWakeAt(); + const row = await endpoint(); + await quiet(10_000); + expect(load).toHaveBeenCalledOnce(); expect(coordinator!.nextWakeAt()).toBe(retryAt); + load.mockImplementation(original); + await settled(); expect(reconcile).toHaveBeenCalledWith(row.id); + expect(load).toHaveBeenCalledTimes(2); expect(load.mock.calls[1][0]).toBeUndefined(); + errors.length = 0; load.mockRestore(); await quiet(); + }); + it("recovers malformed scopes without losing a subsequent valid producer", async () => { + await start(); await settled(); + await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "unknown:old")); + await endpoint(); + await settled(); expect(reconcile).toHaveBeenCalledOnce(); expect(errors.length).toBeGreaterThan(0); errors.length = 0; await quiet(); + }); + it.each([0, 1100])("drains paged inventory despite a failed receiver after %i ms", async delay => { + for (let i = 0; i < 105; i++) await endpoint(); + const service = chatReceiverSchedule(db, { reconcile, stop }); + const reads = vi.spyOn(db, "select"); const inventory = await service.load({ kind: "company", id: companyId }); + expect(inventory).toHaveLength(105); expect(reads).toHaveBeenCalledTimes(2); reads.mockRestore(); + const failedId = inventory[0].id; + reconcile.mockImplementation(async id => { + if (id === failedId) { vi.setSystemTime(Date.now() + delay); throw new Error("bad credential"); } + return "owned"; + }); + await start(service); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + expect(coordinator!.nextWakeAt()!).toBeLessThanOrEqual(Date.now() + 1000); + await vi.waitFor(() => expect(new Set(reconcile.mock.calls.map(([id]) => id)).size).toBe(105), { timeout: 5000 }); + expect(reconcile.mock.calls.filter(([id]) => id === failedId)).toHaveLength(1); + reconcile.mockResolvedValue("owned"); await settled(); errors.length = 0; await quiet(); + }); + it("includes a second company in the first batch despite a larger earlier backlog", async () => { + for (let i = 1; i <= 30; i++) await endpoint({ id: `00000000-0000-4000-8000-${String(i).padStart(12, "0")}` }); + const firstCompany = companyId, firstAgent = agentId, firstApplication = applicationId; + const secondCompany = randomUUID(), secondAgent = randomUUID(), secondApplication = randomUUID(); + await db.insert(companies).values({ id: secondCompany, name: "Other receiver company", issuePrefix: `R${secondCompany.slice(0, 7)}` }); + await db.insert(agents).values({ id: secondAgent, companyId: secondCompany, name: "Receiver", adapterType: "codex_local" }); + await db.insert(toolApplications).values({ id: secondApplication, companyId: secondCompany, name: "Receiver", type: "chat" }); + try { + companyId = secondCompany; agentId = secondAgent; applicationId = secondApplication; + const other = await endpoint({ id: "ffffffff-ffff-4fff-8fff-ffffffffffff" }); + companyId = firstCompany; agentId = firstAgent; applicationId = firstApplication; + await start(); + expect(reconcile.mock.calls.slice(0, 25).map(([id]) => id)).toContain(other.id); + await settled(); expect(reconcile).toHaveBeenCalledTimes(31); await quiet(); + } finally { + companyId = firstCompany; agentId = firstAgent; applicationId = firstApplication; + await coordinator?.stop(); coordinator = undefined; + await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, secondCompany)); + await db.delete(agents).where(eq(agents.companyId, secondCompany)); + await db.delete(companies).where(eq(companies.id, secondCompany)); + } + }); + it("joins admitted runtime initialization during shutdown", async () => { + await start(); await settled(); let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + reconcile.mockImplementation(async () => { await gate; return "owned"; }); + await endpoint(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(reconcile).toHaveBeenCalledOnce(); }); + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(); await stopping; coordinator = undefined; errors.length = 0; + await quiet(); expect(reconcile).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/__tests__/photon/channel.integration.test.ts b/server/src/__tests__/photon/channel.integration.test.ts index 2c008782663..812ce3740e9 100644 --- a/server/src/__tests__/photon/channel.integration.test.ts +++ b/server/src/__tests__/photon/channel.integration.test.ts @@ -1,3 +1,5 @@ +import { createDeliveryWorkCoordinator } from "../../services/delivery-work-coordinator.js"; +import { resolveExternalTestDatabaseUrl } from "../helpers/external-test-database.js"; import { Readable } from "node:stream"; import type { StorageService } from "../../storage/types.js"; import { createHash, randomUUID } from "node:crypto"; @@ -21,6 +23,7 @@ import { } from "@photon-ai/advanced-imessage"; import { and, eq, sql } from "drizzle-orm"; import { + applyPendingMigrations, createDb, companies, agents, @@ -75,15 +78,18 @@ function idleStream(): TypedEventStream { ); } describe("iMessage Photon channel control plane", () => { - let database: Awaited>; + let database: Awaited> | undefined; let db: ReturnType; let secrets: string; const oldKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; const services: ChatChannelService[] = []; + const coordinators: ReturnType[] = []; const companyIds: string[] = []; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-photon-"); - db = createDb(database.connectionString); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PHOTON_TEST_DATABASE_URL"); + if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-photon-"); + if (external) await applyPendingMigrations(external); + db = createDb(external ?? database!.connectionString); secrets = await mkdtemp(path.join(tmpdir(), "photon-secrets-")); process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = path.join( secrets, @@ -91,6 +97,7 @@ describe("iMessage Photon channel control plane", () => { ); }, 30_000); afterEach(async () => { + await Promise.all(coordinators.splice(0).map(coordinator => coordinator.stop())); await Promise.all(services.splice(0).map((service) => service.shutdown())); for (const id of companyIds.splice(0)) await db @@ -100,6 +107,7 @@ describe("iMessage Photon channel control plane", () => { vi.restoreAllMocks(); }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); if (secrets) await rm(secrets, { recursive: true, force: true }); if (oldKey === undefined) @@ -364,8 +372,33 @@ describe("iMessage Photon channel control plane", () => { }, }; } + it("recovers a Photon transport loss through the receiver worker and cancels fatal recovery", async () => { + const t = await setup(); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + coordinators.push(coordinator); + await t.service.registerReceiverWork(coordinator, () => true).ready; + const first = t.service.runtime.get(t.endpoint.id); + expect(first).not.toBeNull(); + await t.callbacks().onPhotonFailure!(new Error("transport interrupted")); + await vi.waitFor(() => { + const current = t.service.runtime.get(t.endpoint.id); + expect(current).not.toBeNull(); expect(current).not.toBe(first); + }, { timeout: 5000 }); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 5000 }); + await t.callbacks().onPhotonFailure!(new PhotonError("credentials", "Rejected credentials")); + await vi.waitFor(() => expect(t.service.runtime.get(t.endpoint.id)).toBeNull(), { timeout: 5000 }); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 5000 }); + expect(await db.select().from(toolConnections).where(eq(toolConnections.id, t.endpoint.connectionId))) + .toEqual([expect.objectContaining({ enabled: false })]); + expect(onError).not.toHaveBeenCalled(); + }); + it("keeps publication authentication failures stopped across receiver reconciliation and restart", async () => { const t = await setup(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: () => {} }); + coordinators.push(coordinator); + await t.service.registerReceiverWork(coordinator, () => true).ready; const conversation = await t.start(); await t.qualify(); const callbacks = t.callbacks(); @@ -380,8 +413,8 @@ describe("iMessage Photon channel control plane", () => { expect.objectContaining({ status: "active", enabled: true, healthStatus: "failed" }), ]); await expect(callbacks.onPhotonAssertOwned!()).rejects.toThrow(); - await t.service.reconcileProviderRuntimes(); - expect(t.service.runtime.get(t.endpoint.id)).toBeFalsy(); + await vi.waitFor(() => expect(t.service.runtime.get(t.endpoint.id)).toBeFalsy(), { timeout: 5000 }); + await coordinator.stop(); const restarted = await t.restart(); expect(restarted.runtime.get(t.endpoint.id)).toBeFalsy(); expect(await restarted.get(t.endpoint.id)).toMatchObject({ status: "attention" }); diff --git a/server/src/app.ts b/server/src/app.ts index c80ff9e6a28..1c95c88a457 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -321,7 +321,7 @@ export function shouldEnablePrivateHostnameGuard(opts: { } type ChatReconciliationLane = - | "provider runtimes" + | "voice sessions" | "deliveries" | "GitHub webhook recovery" | "run milestones" @@ -335,7 +335,7 @@ type ChatReconciliationLane = * next publication sweep for every endpoint. */ export function createChatReconciliationCoordinator(input: { - reconcileProviderRuntimes: () => Promise; + reconcileVoiceSessions: () => Promise; processPendingDeliveries: () => Promise; processFailedGitHubWebhookDeliveries?: () => Promise; projectRunMilestones: () => Promise; @@ -377,7 +377,7 @@ export function createChatReconciliationCoordinator(input: { return { reconcile() { if (stopped) return; - start("provider runtimes", input.reconcileProviderRuntimes); + start("voice sessions", input.reconcileVoiceSessions); start("deliveries", input.processPendingDeliveries); if (input.processFailedGitHubWebhookDeliveries) { start( @@ -1298,16 +1298,14 @@ export async function createApp( } registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); + chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); const reconcileChatPublicationMaintenance = async () => { await chatChannels.processPublicationMaintenance(); }; const chatReconciliation = createChatReconciliationCoordinator({ - reconcileProviderRuntimes: async () => { - await chatChannels.reconcileProviderRuntimes(); - await chatChannels.voice.reconcile(); - }, + reconcileVoiceSessions: () => chatChannels.voice.reconcile(), // Chat delivery, action, and Telegram discovery workers now own these // duties. Keep the coordinator's other maintenance lanes independent. processPendingDeliveries: async () => undefined, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 324ec43b4ae..14c72cb34e3 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,4 +1,7 @@ -import { notifyChatCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { leasedChatProvider } from "./chat-channels/runtime/receiver-policy.js"; +import { chatReceiverSchedule } from "./chat-channels/runtime/receiver-schedule.js"; +import { registerChatReceiverWork } from "./chat-channels/runtime/receiver-work.js"; +import { notifyChatEndpointConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js"; import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; @@ -290,7 +293,6 @@ const ORPHAN_FOLLOW_UP_GRACE_MS = 5_000; const ORPHAN_FOLLOW_UP_MAX_ATTEMPTS = 12; const PUBLICATION_ENDPOINT_CONCURRENCY = 4; -function leasedChatProvider(provider: string): boolean { return provider === "discord" || provider === "imessage-photon"; } const DISCORD_GATEWAY_LEASE_POLL_MS = 100; const PROVIDER_CREDENTIAL_CHECK_TIMEOUT_MS = 25_000; @@ -563,7 +565,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const discordGatewayOwnerships = new Map(); const discordGatewayStopTasks = new Set>(); let shuttingDown = false; - const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } }); + let receiverWork: ReturnType | undefined; + const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, onStopped: id => receiverWork?.changed(id), get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } }); const { receiveSdkMessages } = createChatSdkMessageReceiver({ endpointRecord, processMessage, ingressOnly: (provider) => provider === "imessage-photon" @@ -1117,7 +1120,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { runtimeVersions.delete(endpoint.id); } if (discordOwnership) { - await stopDiscordGatewayOwnership(discordOwnership, false); + await stopDiscordGatewayOwnership(discordOwnership, { removeRuntime: false }); } throw error; } @@ -1134,6 +1137,30 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } } + const receiverSchedule = chatReceiverSchedule(db, { + stop: async id => { + const ownership = discordGatewayOwnerships.get(id); + // Discovery already observed removal; only independent ownership loss needs another wake. + if (ownership) await stopDiscordGatewayOwnership(ownership, { notifyStopped: false }); + }, + reconcile: async id => { + const record = await endpointRecord(id); + if (!record) return "owned"; + try { + const instance = await runtimeFor(record.endpoint, { requireDiscordOwnership: true }); + await reconcileDiscordRuntime(id, instance); + return "owned"; + } catch (error) { + if (isDiscordGatewayNotOwnedError(error)) return "foreign"; + throw error; + } + }, + }); + function registerReceiverWork(coordinator: ReturnType, canRun: () => boolean) { + receiverWork = registerChatReceiverWork(coordinator, receiverSchedule, () => !shuttingDown && canRun()); + return receiverWork; + } + async function reconcileProviderRuntimes() { if (shuttingDown) { return { eligible: 0, local: 0, ownedElsewhere: 0, failed: 0 }; @@ -2211,7 +2238,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ? "Started a conversation by mentioning the agent." : "Sent an empty message."); let comment!: Awaited>; - await notifyChatCommandRegistration(taskTx, taskEndpoint); + await notifyChatEndpointConfiguration(taskTx, taskEndpoint); await taskTx .update(chatEndpoints) .set({ @@ -3472,6 +3499,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { registerCommandWork, + registerReceiverWork, voice, authorizeFastResponse, slackRegistration, diff --git a/server/src/services/chat-channels/command-work-notifications.ts b/server/src/services/chat-channels/command-work-notifications.ts deleted file mode 100644 index e8f5812d4dd..00000000000 --- a/server/src/services/chat-channels/command-work-notifications.ts +++ /dev/null @@ -1,18 +0,0 @@ -import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; -import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js"; - -export async function notifyChatCommandRegistration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { - if (endpoint.provider === "discord") { - await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); - } -} - -export async function notifyChatConnectionCommandRegistration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { - if (transport === "chat_sdk") { - await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); - } -} - -export async function notifyChatCompanyCommandRegistration(transaction: object, companyId: string) { - await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); -} diff --git a/server/src/services/chat-channels/configuration-work-notifications.ts b/server/src/services/chat-channels/configuration-work-notifications.ts new file mode 100644 index 00000000000..5b592e896ec --- /dev/null +++ b/server/src/services/chat-channels/configuration-work-notifications.ts @@ -0,0 +1,22 @@ +import { notifyChatReceiver, notifyChatConnectionReceiver, notifyChatCompanyReceivers } from "./runtime/receiver-work-notifications.js"; +import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; +import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js"; + +export async function notifyChatEndpointConfiguration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { + await notifyChatReceiver(transaction, endpoint); + if (endpoint.provider === "discord") { + await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); + } +} + +export async function notifyChatConnectionConfiguration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { + await notifyChatConnectionReceiver(transaction, connectionId, transport); + if (transport === "chat_sdk") { + await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); + } +} + +export async function notifyChatCompanyConfiguration(transaction: object, companyId: string) { + await notifyChatCompanyReceivers(transaction, companyId); + await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); +} diff --git a/server/src/services/chat-channels/credentials.ts b/server/src/services/chat-channels/credentials.ts index a2cf2b79ed2..3d20be11237 100644 --- a/server/src/services/chat-channels/credentials.ts +++ b/server/src/services/chat-channels/credentials.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "./command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { prepareSlackInventory } from "./slack/inventory.js"; import { prepareGitHubInventory } from "./github/inventory.js"; import { prepareDiscordInventory } from "./discord/inventory.js"; @@ -352,7 +352,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { code: "chat_endpoint_credentials_changed", }); } - await notifyChatCommandRegistration(tx, endpoint); + await notifyChatEndpointConfiguration(tx, endpoint); await tx .delete(companySecretBindings) .where( @@ -464,7 +464,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { .then((rows) => rows[0] ?? null); if (!current) throw notFound("Chat connection not found"); refs = current.refs; - await notifyChatCommandRegistration(tx, endpoint); + await notifyChatEndpointConfiguration(tx, endpoint); await tx .delete(companySecretBindings) .where( diff --git a/server/src/services/chat-channels/discord/gateway-events.ts b/server/src/services/chat-channels/discord/gateway-events.ts index 47deeff7941..3ed17aa6aa3 100644 --- a/server/src/services/chat-channels/discord/gateway-events.ts +++ b/server/src/services/chat-channels/discord/gateway-events.ts @@ -1,3 +1,4 @@ +import { notifyChatReceiver } from "../runtime/receiver-work-notifications.js"; import { notifyDiscordCommandRegistration } from "./command-work-notifications.js"; import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence, InternalSetupState } from "../runtime-types.js"; @@ -130,7 +131,10 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents } const now = new Date(); if (event.type === "connecting") { - if (current.status === "attention") await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); + if (current.status === "attention") { + await notifyChatReceiver(tx, current); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); + } await tx .update(chatEndpoints) .set({ @@ -159,6 +163,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents if (current.status === "attention" && !gatewayOwnedAttention) return; const status = current.setup.step === "complete" ? "active" : "verifying"; + await notifyChatReceiver(tx, current); await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); if (status !== current.status) await notifyChatEndpointWork(tx, current.connectionId); await tx @@ -189,6 +194,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents return; } + await notifyChatReceiver(tx, current); await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); const detail = discordGatewayFailureDetail(event); const transientGuildOutage = event.type === "guild_unavailable"; diff --git a/server/src/services/chat-channels/imessage-photon/events.ts b/server/src/services/chat-channels/imessage-photon/events.ts index 14a914287fd..75f6d2cc92b 100644 --- a/server/src/services/chat-channels/imessage-photon/events.ts +++ b/server/src/services/chat-channels/imessage-photon/events.ts @@ -1,3 +1,4 @@ +import { notifyChatReceiver } from "../runtime/receiver-work-notifications.js"; import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js"; import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; @@ -104,6 +105,7 @@ export function createPhotonEvents(db: Db, options: PhotonEventsOptions) { "attention", ]); if (!endpoint) return; + await notifyChatReceiver(tx, endpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index 49e4f014a8e..d2723672dc7 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "../command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { slackQuestionFormValidationError } from "../slack/question-forms.js"; import { chatInteractionForm, chatItemVerdicts } from "../../chat-interaction-forms.js"; @@ -418,7 +418,7 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb ["active"], ); if (currentEndpoint) { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/provider-effects.ts b/server/src/services/chat-channels/provider-effects.ts index 1273c07c59b..e4a000f94ee 100644 --- a/server/src/services/chat-channels/provider-effects.ts +++ b/server/src/services/chat-channels/provider-effects.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "./command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { createTelegramCallbackNoticeAuthority } from "./telegram/callback-notices.js"; @@ -318,7 +318,7 @@ export function createProviderEffects(db: Db, options: ProviderEffectsOptions) { ); } if (input.disposition.kind === "endpoint_attention" && currentEndpoint) { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index ae2580dfb3f..b5f6648b646 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "../command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; @@ -190,7 +190,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur : null; if (disposition.kind === "endpoint_attention" && currentEndpoint) { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index 54cba206d58..b189c1cc168 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "./command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import type { ConnectionManagementContext } from "../connection-management.js"; import { imessageResourceParticipants } from "./imessage-photon/resources.js"; @@ -511,7 +511,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc : currentEndpoint.status === "attention" ? "verifying" : currentEndpoint.status; - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); if (status !== currentEndpoint.status) await notifyChatEndpointWork(tx, currentEndpoint.connectionId); await tx .update(chatEndpoints) @@ -541,7 +541,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc .where(eq(toolConnections.id, currentEndpoint.connectionId)); } } else { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/runtime/ownership.ts b/server/src/services/chat-channels/runtime/ownership.ts index 3e8bec1fbc4..4a673d41505 100644 --- a/server/src/services/chat-channels/runtime/ownership.ts +++ b/server/src/services/chat-channels/runtime/ownership.ts @@ -1,3 +1,4 @@ +import { receiverLeaseKey as leaseKeyForProvider } from "./receiver-policy.js"; import type { EndpointRow, DbTransaction, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js"; import { randomUUID } from "node:crypto"; import { and, eq, lte } from "drizzle-orm"; @@ -17,12 +18,12 @@ type OwnershipOptions = { discordGatewayOwnerships: Map; discordGatewayStopTasks: Set>; isShuttingDown: () => boolean; + onStopped?: (endpointId: string) => void; readonly discordGatewayLeaseTtlMs?: number; readonly discordGatewayLeaseRenewalIntervalMs?: number; readonly discordGatewayLeaseWaitMs?: number; readonly renewDiscordGatewayLease?: (input: { companyId: string; credentialFingerprint: string; endpointId: string; expiresAt: Date; generation: number; leaseKey: string; token: string }) => Promise; }; -const DISCORD_GATEWAY_LEASE_KEY = "discord_gateway_runtime"; const DISCORD_GATEWAY_LEASE_TTL_MS = 15_000; const DISCORD_GATEWAY_LEASE_WAIT_MS = 20_000; const DISCORD_GATEWAY_LEASE_POLL_MS = 100; @@ -73,7 +74,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { async function stopDiscordGatewayOwnership( ownership: DiscordGatewayOwnership, - removeRuntime = true, + stopOptions: { removeRuntime?: boolean; notifyStopped?: boolean } = {}, ): Promise { if (ownership.stopPromise) return ownership.stopPromise; ownership.stopping = true; @@ -81,12 +82,9 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { clearInterval(ownership.renewTimer); ownership.renewTimer = null; } - if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) { - discordGatewayOwnerships.delete(ownership.endpointId); - } ownership.stopPromise = (async () => { if ( - removeRuntime && + stopOptions.removeRuntime !== false && ownership.context.endpointRuntime && runtime.get(ownership.endpointId) === ownership.context.endpointRuntime ) { @@ -104,6 +102,10 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { // The listener is stopped before the lease is released. A standby can // therefore acquire immediately without overlapping Gateway sessions. await releaseDiscordGatewayLeaseRow(ownership); + if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) { + discordGatewayOwnerships.delete(ownership.endpointId); + } + if (!options.isShuttingDown() && stopOptions.notifyStopped !== false) options.onStopped?.(ownership.endpointId); })(); discordGatewayStopTasks.add(ownership.stopPromise); void ownership.stopPromise.finally(() => { @@ -304,7 +306,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { context: RuntimeContext, waitForOwnership: boolean, ): Promise { - const receiverLeaseKey = endpoint.provider === "imessage-photon" ? "photon_receiver_runtime" : DISCORD_GATEWAY_LEASE_KEY; + const receiverLeaseKey = leaseKeyForProvider(endpoint.provider)!; const local = discordGatewayOwnerships.get(endpoint.id); if ( local && diff --git a/server/src/services/chat-channels/runtime/receiver-policy.ts b/server/src/services/chat-channels/runtime/receiver-policy.ts new file mode 100644 index 00000000000..70983c32bbb --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-policy.ts @@ -0,0 +1,13 @@ +import type { ChatProvider } from "@paperclipai/shared"; + +export const RECEIVER_LEASE_KEYS = { + discord: "discord_gateway_runtime", + "imessage-photon": "photon_receiver_runtime", +} as const satisfies Partial>; + +export function receiverLeaseKey(provider: string) { + return Object.hasOwn(RECEIVER_LEASE_KEYS, provider) + ? RECEIVER_LEASE_KEYS[provider as keyof typeof RECEIVER_LEASE_KEYS] : undefined; +} + +export const leasedChatProvider = (provider: string) => receiverLeaseKey(provider) !== undefined; diff --git a/server/src/services/chat-channels/runtime/receiver-schedule.ts b/server/src/services/chat-channels/runtime/receiver-schedule.ts new file mode 100644 index 00000000000..4e07ec7fb95 --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-schedule.ts @@ -0,0 +1,53 @@ +import { and, eq, gt, inArray, ne, or } from "drizzle-orm"; +import { chatEndpoints, chatEndpointLeases, toolConnections, type Db } from "@paperclipai/db"; +import type { ChatProvider } from "@paperclipai/shared"; +import { RECEIVER_LEASE_KEYS, receiverLeaseKey } from "./receiver-policy.js"; +import type { ReceiverScope } from "./receiver-work-notifications.js"; + +export type ReceiverDeadline = { id: string; companyId: string; connectionId: string; dueAt: number | null }; + +export function chatReceiverSchedule(db: Db, receiver: { + reconcile: (id: string) => Promise<"owned" | "foreign">; + stop: (id: string) => Promise; +}) { + async function load(scope?: ReceiverScope, signal?: AbortSignal): Promise { + const result: ReceiverDeadline[] = []; + let cursor: string | undefined; + while (true) { + signal?.throwIfAborted(); + const rows = await db.select({ id: chatEndpoints.id, companyId: chatEndpoints.companyId, connectionId: chatEndpoints.connectionId }) + .from(chatEndpoints).innerJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatEndpoints.companyId))) + .where(and( + inArray(chatEndpoints.provider, Object.keys(RECEIVER_LEASE_KEYS) as ChatProvider[]), + inArray(chatEndpoints.status, ["verifying", "active", "attention"]), + eq(toolConnections.status, "active"), eq(toolConnections.enabled, true), + or(ne(chatEndpoints.status, "attention"), ne(toolConnections.healthStatus, "failed")), + 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(chatEndpoints.id).limit(100); + result.push(...rows.map(row => ({ ...row, dueAt: 0 }))); + if (rows.length < 100) return result; + cursor = rows[rows.length - 1].id; + } + } + async function check(id: string, signal?: AbortSignal): Promise { + const scope = { kind: "endpoint", id } as const; + const [before] = await load(scope, signal); + if (!before) { await receiver.stop(id); return null; } + signal?.throwIfAborted(); + const ownership = await receiver.reconcile(id); + const [after] = await load(scope, signal); + if (!after) { await receiver.stop(id); return null; } + if (ownership === "owned") return { ...after, dueAt: null }; + const [endpoint] = await db.select({ provider: chatEndpoints.provider }).from(chatEndpoints).where(eq(chatEndpoints.id, id)); + if (!endpoint) { await receiver.stop(id); return null; } + const key = receiverLeaseKey(endpoint.provider); + const [lease] = key ? await db.select({ expiresAt: chatEndpointLeases.expiresAt }).from(chatEndpointLeases).where(and( + eq(chatEndpointLeases.companyId, after.companyId), eq(chatEndpointLeases.endpointId, id), eq(chatEndpointLeases.leaseKey, key), + )).limit(1) : []; + return { ...after, dueAt: Math.max(Date.now() + 1000, lease?.expiresAt.getTime() ?? 0) }; + } + return { load, check, stop: receiver.stop }; +} diff --git a/server/src/services/chat-channels/runtime/receiver-work-notifications.ts b/server/src/services/chat-channels/runtime/receiver-work-notifications.ts new file mode 100644 index 00000000000..5ff7a2e2248 --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-work-notifications.ts @@ -0,0 +1,24 @@ +import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; +import { leasedChatProvider } from "./receiver-policy.js"; + +export type ReceiverScope = { kind: "endpoint" | "connection" | "company"; id: string }; +export function receiverScope(hint: string): ReceiverScope | undefined { + if (hint === "*") return undefined; + const [kind, id, extra] = hint.split(":"); + if (!id || extra !== undefined || (kind !== "endpoint" && kind !== "connection" && kind !== "company")) + throw new Error("Invalid receiver discovery scope"); + return { kind, id }; +} +export const receiverScopeHint = (scope: ReceiverScope) => `${scope.kind}:${scope.id}`; +export async function notifyChatReceiver(transaction: object, endpoint: { id: string; provider: ChatProvider }) { + if (leasedChatProvider(endpoint.provider)) + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "endpoint", id: endpoint.id })); +} +export async function notifyChatConnectionReceiver(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { + if (transport === "chat_sdk") + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "connection", id: connectionId })); +} +export async function notifyChatCompanyReceivers(transaction: object, companyId: string) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "company", id: companyId })); +} diff --git a/server/src/services/chat-channels/runtime/receiver-work.ts b/server/src/services/chat-channels/runtime/receiver-work.ts new file mode 100644 index 00000000000..c7bb909b162 --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-work.ts @@ -0,0 +1,91 @@ +import { takeCompanyWorkBatch } from "../../company-work-batch.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import type { ReceiverDeadline, chatReceiverSchedule } from "./receiver-schedule.js"; +import { receiverScope, receiverScopeHint, type ReceiverScope } from "./receiver-work-notifications.js"; + +function matches(entry: ReceiverDeadline, scope?: ReceiverScope) { + return !scope || (scope.kind === "endpoint" ? entry.id : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id; +} +export function registerChatReceiverWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const entries = new Map(); + const failures = new Map(); + const scopeFailures = new Map(); + let earliest = 0; + let lastCompany: string | null = null; + function failed(map: typeof failures, id: string) { + const count = Math.min((map.get(id)?.count ?? 0) + 1, 5); + map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }); + } + function dirtyScopes() { + return [...dirty].map(hint => { try { return receiverScope(hint); } catch { return undefined; } }); + } + function next() { + let nextAt: number | null = null; + const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; + for (const hint of dirty.has("*") ? ["*"] : dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); + const unavailable = dirtyScopes(); + for (const entry of entries.values()) if (entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope))) + take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); + return nextAt; + } + const worker = coordinator.register(DELIVERY_QUEUES.chatReceivers, { + retryMs: 30_000, + pausePassiveMaintenance: true, + onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + 1000; + const errors: unknown[] = []; + const requested = (dirty.has("*") ? ["*"] : [...dirty]).filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); + const covered = requested.includes("*") ? [...dirty] : requested; + for (const hint of covered) dirty.delete(hint); + for (let index = 0; index < requested.length; index++) { + if (signal.aborted || !canRun()) { for (const hint of requested.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; } + const hint = requested[index]; + let scope: ReceiverScope | undefined; + let decoded = false; + try { + scope = receiverScope(hint); decoded = true; + const rows = await service.load(scope, signal), retained = new Set(rows.map(row => row.id)); + for (const entry of entries.values()) if (matches(entry, scope) && !retained.has(entry.id)) { + await service.stop(entry.id); + entries.delete(entry.id); failures.delete(entry.id); + } + for (const entry of rows) entries.set(entry.id, entry); + if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint); + } catch (error) { + const retryHint = decoded ? hint : "*"; + dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error); + } + } + const unavailable = dirtyScopes(); + const due = [...entries.values()].filter(entry => entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope)) + && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) + .sort((a, b) => a.dueAt! - b.dueAt! || a.id.localeCompare(b.id)); + const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany; + for (const entry of batch.entries) { + if (signal.aborted || !canRun()) break; + try { + const current = await service.check(entry.id, signal); + if (current) entries.set(entry.id, current); else entries.delete(entry.id); + failures.delete(entry.id); + } catch (error) { failed(failures, entry.id); errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, "Chat receiver discovery could not settle"); + }, + // Preserve hints/deadlines for resume without holding settled intent during drain. + // The coordinator independently fences unresolved database transactions. + hasPending: async () => canRun() && dirty.size > 0, + nextRunAt: next, + nextRunAtOnError: next, + }); + return { ready: worker.ready, changed(id: string) { + dirty.add(receiverScopeHint({ kind: "endpoint", id })); worker.wake(); + } }; +} diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index cc4742d0b71..71b00c50cf0 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -1,4 +1,4 @@ -import { notifyChatConnectionCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { notifyChatConnectionConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export async function notifyChatDeliveryWork(tx: object): Promise { @@ -38,5 +38,5 @@ export async function notifyChatEndpointWork(tx: object, connectionId: string): for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue); await notifyChatVerificationWork(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTelegramSubscriptions, connectionId); - await notifyChatConnectionCommandRegistration(tx, connectionId); + await notifyChatConnectionConfiguration(tx, connectionId); } diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index c87d7f89a1d..10c01ed86e0 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,4 +1,4 @@ -import { notifyChatCompanyCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { chatCompanyDeletion } from "./chat-channels/company-deletion.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; @@ -65,7 +65,7 @@ export async function deleteCompany(db: Db, id: string) { const [existing] = await tx.select({ id: companies.id }).from(companies) .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; - await notifyChatCompanyCommandRegistration(tx, id); + await notifyChatCompanyConfiguration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index e95a08eb41e..8ff37d9b966 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -43,6 +43,7 @@ export const DELIVERY_QUEUES = { chatGitHubPublications: "chat-github-publications", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", + chatReceivers: "chat-receivers", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs", diff --git a/server/src/services/tool-access/catalog/health.ts b/server/src/services/tool-access/catalog/health.ts index 0ea2ab9203a..63567464a54 100644 --- a/server/src/services/tool-access/catalog/health.ts +++ b/server/src/services/tool-access/catalog/health.ts @@ -1,4 +1,4 @@ -import { notifyChatConnectionCommandRegistration } from "../../chat-channels/command-work-notifications.js"; +import { notifyChatConnectionConfiguration } from "../../chat-channels/configuration-work-notifications.js"; import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectionActorUserId } from "../actor.js"; import { assertSupportedConnection } from "../connections/setup-policy.js"; @@ -50,7 +50,7 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d ) { const now = new Date(); const [updated] = await writeToolConnectionHealthChange(db, connection, async tx => { - await notifyChatConnectionCommandRegistration(tx, connection.id, connection.transport); + await notifyChatConnectionConfiguration(tx, connection.id, connection.transport); return tx.update(toolConnections) .set({ healthStatus: status,