From 94a780286fa3b9bafb4a60f705ea0465b928b3ce Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:37:50 -0500 Subject: [PATCH] fix(server): schedule tool health from committed changes and deadlines (#17554) ## Thinking Path > - Paperclip manages AI-agent work and its connected tools. > - Idle hosted instances need their database and network activity to stop. > - The maintenance tick queries tool connections even when no health check is due. > - Existing connection rows already contain eligibility and the last health observation. > - This pull request reconstructs health deadlines and refreshes them after committed changes. > - The shared coordinator preserves retries, drain, and shutdown without another private timer. ## Linked Issues or Issue Description **What happened?** The aggregate maintenance tick scans connection health every 30 seconds. Empty instances and connections with fresh health observations still cause SQL activity. **Expected behavior** No eligible connection means no recurring health SQL. Eligible connections retain the existing 15-minute proactive check. Configuration changes and health observations update the schedule after commit. **Steps to reproduce** Start a server with heartbeat scheduling enabled and no tool connections. Let startup finish and observe two maintenance ticks. The old code repeats the health inventory query. The new startup regression verifies that those ticks do not invoke the health inventory. **Paperclip version or commit** This branch includes merged parent #17518 and master `96727b3cfefb61f85ffe1fbc29a9b6dd6837efe8`. The 47 changed paths belong to the single tool-connection health duty. No database file differs from master. Related: #14638 covers runtime-auth exclusions. #15448, #15325, and #11910 concern MCP failure classification. This PR preserves those provider policies and does not replace their work. ## What Changed - Keep the GitHub error-projection unit fixture at the health-write boundary and verify the real transaction path in the PostgreSQL health-worker suite. - Add a health schedule owned by tool access. Read lean eligibility fields and saved observation timestamps. - Register health work with the existing coordinator. Refresh only dirty connection/application/company scopes using bounded inventory pages, check due connections in fair company batches, continue concrete due backlog after one second, and retain failures with backoff. - Notify before supported eligibility and observation writes in their owning transactions. Cover API/gallery setup, OAuth, catalog/gateway observations, email, voice, access removal, deletion, and settings. - Commit connection creation with grant, binding, and runtime-slot SQL before its wake. Keep provider calls outside the new transactions. - Keep provider-to-transport mapping in the chat catalog. SDK-only chat changes do not wake health work. - Replace aggregate health scans, retain the public manual sweep, and document the producer audit and remaining sleep limits. - Keep metadata-only example and Smoke Lab installs scoped. Lock and re-read application type before publishing a real eligibility change. - Keep failed inventory reads on their own retry deadline while healthy due batches continue. - Apply fixture migrations before using external disposable databases in all six producer suites. ## Verification Published head: `8c1d293d1ae2b7db753ab20f5a2a2edec1fb0a62`. Fresh CI and reviews are required for this exact head. - **623 distinct targeted cases pass.** The final four complete health, fair-batching, GitHub health recovery, and startup suites pass 80 cases after the retry fix. Five complete producer suites pass 543 cases: tool access 469, access 18, email 45, GitHub events four, and Smoke Lab seven. The earlier 26 health cases are part of this total, not additional cases. - The six producer suites first passed against separate, initially empty PostgreSQL 17 databases at `acbf262843`. They apply their own migrations. The subsequent repair changes only the health worker and its regression; its complete final consumer suites pass on that integrated parent. Server typecheck and build pass with the final repair. - **Installer amplification:** five real installs with 205 unrelated connections reproduced 1,035 example rows and 1,045 Smoke Lab rows. Repeated metadata installs now load only changed fixture connections. The same cases restore an excluded application through each installer and require all 205 unrelated IDs to re-enter the schedule. - **Concurrent eligibility:** a real PostgreSQL lock-wait case requires the installer to notify after a concurrent type change. A mutation that retains the lock but uses the old lookup fails the settled-hint assertion; original source was restored byte for byte. - **Retry isolation:** the failing baseline read one unavailable company three times while another company's 60 due checks drained. Final tests require one initial read, at least 30 seconds between failed reads, prompt healthy progress, eventual recovery, and SQL quiet after settlement. - Real-service producer coverage includes AgentMail activation/pause, GitHub installation changes, and member access removal. Lifecycle coverage includes outer commit/rollback, startup reconstruction, drain/resume, failure, and shutdown. - Failure history and exact evidence scope: [health acceptance](https://github.com/paperclipai/paperclip/blob/8c1d293d1ae2b7db753ab20f5a2a2edec1fb0a62/doc/idle-tool-connection-health.md#acceptance-follow-up-after-the-plugin-parent-merged-october-27), [producer audit](https://github.com/paperclipai/paperclip/blob/8c1d293d1ae2b7db753ab20f5a2a2edec1fb0a62/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md). Prior remote failures and earlier 706-case evidence remain historical. Startup tests emitted listener-count warnings; all assertions passed. No full local repository-suite, deployed sleep, fleet savings, or live provider qualification claim is made. ## Risks - Supported live writers must use the transaction signals. Independent SQL writers and other replicas need their own wake integration. - Configured eligible connections still have a real proactive health obligation. This does not prove Cloud alarm coverage or host/database suspension. - Existing manual probes and catalog operations can overlap. This migration does not introduce a credential-version CAS or change provider failure policy. - Voice remains eligible under the previous health policy. Its generic probe can still report an unsupported transport; changing that behavior is separate work. - No database migration, new table, workflow, or lockfile change. This PR includes its merged parent and current inspected master; later stack PRs and all current-head acceptance gates remain separate. ## Model Used OpenAI Codex, GPT-6 family, with repository tools, code execution, and reasoning. The exact serving version and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/idle-tool-connection-health.md | 291 +++++++++++ ...2026-10-24-recurring-server-sleep-stack.md | 98 ++++ ...0-24-tool-connection-health-sleep-audit.md | 196 +++++++ ...10-25-configuration-worker-drain-repair.md | 31 ++ server/src/__tests__/access-service.test.ts | 25 +- .../email-channels.integration.test.ts | 37 +- .../github-connection-events.test.ts | 30 +- .../__tests__/github-health-recovery.test.ts | 7 + .../__tests__/helpers/tool-health-worker.ts | 52 ++ .../server-startup-feedback-export.test.ts | 28 +- server/src/__tests__/smoke-lab.test.ts | 11 +- .../src/__tests__/tool-access-service.test.ts | 66 ++- .../tool-connection-health-work.test.ts | 494 ++++++++++++++++++ server/src/index.ts | 21 +- server/src/services/access.ts | 2 + server/src/services/chat-channels.ts | 2 + .../services/chat-channels/configuration.ts | 7 + .../connection-health-notifications.ts | 8 + .../src/services/chat-channels/endpoints.ts | 3 +- .../chat-channels/interactions/callbacks.ts | 2 + .../chat-channels/provider-catalog.ts | 6 + .../chat-channels/provider-effects.ts | 2 + .../chat-channels/publications/failures.ts | 2 + .../chat-channels/resource-reconciliation.ts | 3 + .../services/chat-channels/setup-probes.ts | 2 + server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 2 +- server/src/services/email-channels.ts | 12 +- server/src/services/email-connections.ts | 5 +- .../src/services/github-connection-events.ts | 2 + server/src/services/instance-settings.ts | 2 + server/src/services/smoke-lab.ts | 104 ++-- server/src/services/tool-access.ts | 188 ++++--- .../tool-access/catalog/health-schedule.ts | 62 +++ .../services/tool-access/catalog/health.ts | 5 +- .../tool-access/catalog/publication.ts | 5 +- .../tool-access/connections/removal.ts | 2 + .../tool-access/connections/setup-policy.ts | 8 +- .../tool-access/oauth/access-finalization.ts | 60 ++- .../tool-access/oauth/cloud-callback.ts | 2 + .../tool-access/oauth/connection-refresh.ts | 5 +- .../tool-access/oauth/direct-callback.ts | 3 + .../tool-access/oauth/grant-refresh.ts | 5 +- .../tool-access/oauth/vercel-callback.ts | 5 +- .../tool-connection-health-notifications.ts | 27 + .../services/tool-connection-health-work.ts | 104 ++++ server/src/services/tool-gateway.ts | 5 +- 47 files changed, 1795 insertions(+), 246 deletions(-) create mode 100644 doc/idle-tool-connection-health.md create mode 100644 doc/plans/2026-10-24-tool-connection-health-sleep-audit.md create mode 100644 doc/plans/2026-10-25-configuration-worker-drain-repair.md create mode 100644 server/src/__tests__/helpers/tool-health-worker.ts create mode 100644 server/src/__tests__/tool-connection-health-work.test.ts create mode 100644 server/src/services/chat-channels/connection-health-notifications.ts create mode 100644 server/src/services/tool-access/catalog/health-schedule.ts create mode 100644 server/src/services/tool-connection-health-notifications.ts create mode 100644 server/src/services/tool-connection-health-work.ts diff --git a/doc/idle-tool-connection-health.md b/doc/idle-tool-connection-health.md new file mode 100644 index 00000000000..484cea845ef --- /dev/null +++ b/doc/idle-tool-connection-health.md @@ -0,0 +1,291 @@ +# Tool connection health while idle + +The health worker replaces the aggregate maintenance tick's repeated connection +inventory query. It preserves the existing 15-minute proactive health interval +for eligible connections. This is a deadline migration, not removal of proactive +health checks or proof of host suspension. + +## Ownership and lifetime + +`tool-access/catalog/health-schedule.ts` owns eligibility and saved deadlines. +The existing health and catalog services still own provider behavior. The worker +uses the shared delivery coordinator for admission, wake coalescing, retry, +drain/resume, and shutdown. Startup reconstructs eligible deadlines from existing +connection rows. No new table, polling listener, or private interval is added. + +Eligible rows are active and enabled, excluding `chat_sdk`, `runtime_auth`, and +applications of type `paperclip_plugin`. Disabled experimental connectors have +no deadline until the settings producer enables them. Application status and +company status are not new filters: this migration preserves the old policy. + +An unchecked row is due immediately. Otherwise its deadline is `healthCheckedAt` +plus 15 minutes. Individual writes carry a connection ID and reload only that +connection. Application mutations refresh only that application's connections; +company deletion invalidates the company's cache. Startup and instance feature +settings can refresh the whole instance. Inventory reads use UUID keyset pages +of at most 100 rows and observe shutdown cancellation between pages. Refreshes are +coalesced with a 30-second minimum spacing when no due backlog remains. +Batches retain at most 25 checks and continue after one second while already due +rows remain, so the batch limit does not cap dispatch at 50 checks per minute. There is no scheduled refresh when +there are no eligible rows or unresolved changes. Future checks cause no SQL +before their deadline. Each pass checks at most 25 due connections, taking turns +across companies and preserving due order within each company. The company cursor +rotates across passes when more companies are due than fit in one batch. + +Before a provider check, the service reloads current eligibility and deadline. +A saved health result, including a failed health observation, owns the next +15-minute deadline. A check that cannot persist an observation retains recovery +debt with exponential backoff from 30 seconds to 15 minutes. A failed scoped +refresh does not prevent checks outside that connection, application, or company. Dirty scopes arriving during +a refresh remain pending. Drain retains them without passive database reads. +Shutdown joins admitted work; it does not abandon a provider call midway. + +## Producer coverage + +| Owner | Scheduling change | +| --- | --- | +| Tool access API and examples | Create, enable/disable, reconfigure, revive, or restore a connection; setup grants, credential bindings, and runtime slots commit before a creation wake | +| Gallery setup and failure handling | Activation, temporary quarantine, failed-identity restoration, and failed setup deletion | +| OAuth callbacks and finalization | Direct, Cloud, Vercel, and personal/shared activation; reauthorization-required disablement | +| Connection removal | Archive and disable in the owning transaction | +| Health/catalog publication | Successful or failed saved health observation moves the deadline | +| Gateway actual use | Saved remote connection health moves the deadline | +| GitHub installation events | Health observation commits with the installation-event transaction | +| Email setup/control/health | REST connection activation, pause/resume/removal, successful receipt health, and failed health observations | +| Shared chat lifecycle | Voice health activation, setup, pause/resume/removal, resource reconciliation, inbound activation, and failure observations; SDK-only chat mutations do not wake this worker | +| Access removal | Losing the last applicable credential grant disables the connection | +| Smoke Lab | Connection setup and fixture health updates | +| Company deletion | Removes cached deadlines after cascading deletion commits | +| Instance settings | Experimental connector eligibility changes use an administrative all-company hint | + +Notifications are awaited before the relevant SQL write. Existing transactions +remain the owner; standalone mutations use the same small transaction helper. +Nested calls retain the shared database client's outer-commit, rollback, and +uncertain-outcome protocol. Provider requests are outside newly added +transactions. The health worker never interprets provider protocols. + +Credential-only refresh, refresh lease acquisition/release, client registration, +OAuth authorization-start metadata, legacy credential repair, Slack-search OAuth, +aggregator account metadata, and install markers do not change health eligibility +or `healthCheckedAt`. They do not gain health notifications. A due probe reads +current credentials. AI writers use the excluded runtime-auth transport. +Application display/status updates do not change the old health eligibility. +Application deletion refuses linked connections; company cascade is covered. +New draft-only connections remain ineligible until their activation producer. + +## Limits + +The supported live writer model is this server's instrumented services. Offline +restore is reconstructed at startup. Independent SQL writers and other replicas +need their own wake integration; process-local hints do not cover them. + +Existing manual health checks and catalog operations can overlap. This migration +does not introduce a credential-version CAS or change provider failure policy. +The scheduled worker itself is non-overlapping and rechecks the current row before +probing. A naive `updatedAt` CAS would reject a health check's own catalog writes. +Voice remains eligible under the old sweep's policy even though its generic +health probe can report an unsupported transport. Changing that policy belongs +to the voice owner, separately from this scheduling migration. + +The Cloud sleep-safety proof is unchanged. An in-process deadline cannot wake +stopped compute; unsupported external wake obligations remain blockers. + +## Verification in progress + +The original startup regression showed repeated idle health sweeps. It now proves +one scheduled-worker reconstruction and no health inventory work from aggregate +ticks. Seventeen real-PostgreSQL lifecycle tests pass: empty/configured quiet, +excluded transports, first writes, nested commit/rollback, drain/resume, saved +observations, restart, page progress, real API activation/disable/archive, +company failure isolation, feature toggles, company deletion, voice activation, +concurrent producer writes, failed persistence, and admitted shutdown. + +The first consumer run passed 514 cases and failed a rollback fault fixture tied +to transaction call order. The fixture now rejects the actual restoration update +with a PostgreSQL trigger. All 515 startup/tool-access cases pass afterward. +Another 163 gateway/settings cases pass. Server typechecking passes after fixing +two mutable values captured by transaction callbacks. Additional producer/chat +consumers, query-plan inspection, build, and remote review/checks remain pending. + +An initial additional-consumer command ran before its external-database fixture +patch applied: embedded PostgreSQL failed to start and 72 tests were skipped. +That run is not passing evidence. The corrected run uses validated disposable +PostgreSQL databases; results will be recorded before PR acceptance. + +The corrected producer run first passed all 44 email cases but contaminated the +other suites with shared fixture rows, causing foreign-key cleanup failures. +Separate freshly initialized databases preserve each suite's original isolation; +all 72 access/email/GitHub/smoke-lab cases then pass. Three selected chat credential +and pause cases pass; 1,414 deselected cases are not counted as passing coverage. +The producer mapping/voice wake contract is covered by the health lifecycle test. +Server build passes. The scoped inventory SQL with 40,000 archived and 10 active +rows uses `tool_connections_company_enabled_idx`, 13 shared buffer hits, and +0.022 ms execution; the diagnostic data was rolled back. No new index is needed +for this observed query. The local passing selection totals 770 distinct tests. +Full workspace and remote acceptance gates remain required. + +Review repair: a controlled 503 probe persisted its failed health observation, +but the warning regression failed before the fix. Scheduled probe failures now +log company/connection identifiers and a fixed message, without provider prose. +The saved 15-minute deadline still controls the next check. All 64 health/startup +cases and server typechecking pass after repair; total selected coverage is now +771 unique tests. Fresh remote gates are required for this repair. + + +Review follow-up: fair round-robin company batching replaces the global oldest-25 +selection. The two-company regression failed before the repair; 19 health-worker +and three shared batching cases pass afterward. Each probe's pre/post eligibility +query already includes the connection ID, so those indexed current-state checks +remain. A scheduled observation still emits the normal transaction signal and +now refreshes only that connection following the security repair below. This +preserves uncertainty and concurrent-writer recovery without a silent-write exception. + +The real-service producer suite now proves AgentMail activation triggers its first +health probe, pause cancels the next deadline, GitHub installation events update +the cached deadline, and member removal cancels a personal connection's deadline. +The GitHub newer-access branch retains its old health deadline, as intended. +These tests use the actual coordinator, transaction notifications, schedule query, +and service mutations; AgentMail also runs the actual health service against its +controlled provider. Subsequent idle windows assert no SQL. + +The first run exposed two incorrect fixture expectations: AgentMail setup leaves +healthCheckedAt null until its first probe, and a stale GitHub installation event +does not overwrite a newer access refresh. The tests now preserve those policies. +A targeted rerun reused AgentMail's retained fixture user and failed setup with a +duplicate key; a fresh disposable database fixes that isolation issue without +changing application behavior or weakening assertions. All 66 access/email/GitHub +cases pass on the final run. Server typecheck/build pass. Cumulative selected +coverage is 776 unique tests; full-suite and fresh remote review/checks remain +acceptance gates. No whole-instance sleep or provider suspension is claimed. + + +The Discord child review also exposed fixed-batch throughput as a risk for health. +A probe-timestamp regression measured a 30.7-second gap before the 26th already +owed check. The worker now schedules a one-second continuation only when the due +selection exceeds its bounded batch. Failed work retains backoff; an empty or +fully future-dated inventory has no continuation. This is scheduling concrete +backlog, not a periodic idle scan. The initial deadline-only assertion passed +incorrectly because a pending transaction notification temporarily reported an +immediate coordinator wake; acceptance measures actual probe timestamps instead. + + +## Security review: bounded refresh scope + +Security review identified amplification in the original company hints. Five +separate real `updateConnection` calls in a company with 205 eligible connections +returned 1,025 inventory rows. A regression reproduced that result. The repaired +producer contract returns exactly five rows across those updates. Saved probe +observations use the same single-connection scope, so draining a due backlog no +longer reloads its whole company after each batch. + +The typed hint accepts an existing connection row (only company and connection +IDs are encoded), or an explicit application target. Company/global scopes remain +for company deletion and administrative feature changes. Smoke Lab allocates a +new connection ID before its insert so the transaction intent is registered before +the write. No schema, outer transaction ownership, or provider policy changes. +Application eligibility changes retain their own scoped refresh. Single-connection +refreshes remove an old cached deadline even when the row was deleted, disabled, +or moved to an excluded application; unrelated deadlines remain intact. + +Inventory queries use 100-row UUID keyset pages. A 205-row regression initially +observed one unbounded query and now requires three pages with all 205 distinct +IDs retained. Cancellation is checked between pages. Results replace a scope only +after its complete load succeeds; failed loads retain recovery and other scopes +can proceed. Malformed hint shapes request full reconstruction instead of dropping +later scopes. + +The first amplification fixture lacked its required remote endpoint and failed +validation before reaching its assertion. After adding a controlled endpoint +lookup, the baseline reproduced 1,025 versus five. The first typecheck found a +partial access-service projection without companyId; the target now supplies the +already-owned companyId explicitly. These failures are retained as evidence, not +counted as passes. Final verification and remote review are recorded below. + + +Final scoped-refresh verification: 26 health/batching cases, 63 email/access +cases, 56 GitHub/Smoke Lab/startup cases, 469 tool-access cases, and 92 gateway +cases pass (706 targeted tests on this repair). Server typecheck and build pass. +The full local test matrix was not rerun; prior remote Actions passed at +f8ad26f0ff, which still had the security finding. Fresh remote checks/security +review on the repaired head are required and are not inferred from those results. + +## GitHub health unit fixture after master integration (October 27) + +Fresh CI on #17649 failed server shard 12 in the new GitHub suspended-installation +recovery test. Its minimal database fake only implemented reads and updates; +health changes now go through the existing transactional write owner. The test +therefore received `db.transaction is not a function` instead of its expected +422 response and installation settings links. The same failure reproduced on +this owning PR before correction. + +The unit test now substitutes the health-write collaborator while preserving +all of its original error, message, connection-health, and URL assertions. This +is the boundary of that error-projection fixture, not a change to production +transactions or a substitute for transactional acceptance. The separate complete +`tool-connection-health-work` suite still uses the real write owner and a fresh +PostgreSQL database. Both complete files pass all 24 cases with no skips on owner +source based on `d2068f59a30f6f33b4543a79b539016294cbfe15`. + +The failed CI job is +[server shard 12](https://github.com/paperclipai/paperclip/actions/runs/40470614670/job/122077293740). +Fresh remote CI is still required on the follow-up head. Other review findings +on health scheduling remain separate; this fixture correction does not resolve +them or claim deployed sleep. + +## Acceptance follow-up after the plugin parent merged (October 27) + +PR #17518 is merged as `96727b3cfefb61f85ffe1fbc29a9b6dd6837efe8`. +The health branch includes that actual master through a normal merge. Its 47 +health-owned paths are the only differences from that master; it changes no +database files. All 18 preceding sleep PRs are merged. Older statements above +about an unmerged stack and prior acceptance describe their recorded revisions. + +A second security finding exposed a remaining application-wide invalidation in +example installation and Smoke Lab. With 205 unrelated eligible connections +under an installed application, five actual repeated installs loaded 1,035 rows +for the example and 1,045 rows for Smoke Lab. These are failing row-count +regressions, separate from the earlier individual-connection amplification. +Metadata-only upserts now retain connection-scoped hints from the actual fixture +writes. Application-wide refresh occurs only when its type changes eligibility. +The same tests restore an excluded application through each real installer and +require all 205 unrelated connection IDs to return to the live schedule. + +The upsert reads the current type under its owning transaction's PostgreSQL +`FOR NO KEY UPDATE` lock. This matches the non-key update and permits foreign-key +key-share locks. An earlier lookup cannot establish the current type after a +concurrent writer commits. A real lock-wait test confirms that the installer +notifies the restored application after the other writer releases its lock. +Replacing that current-type comparison with the earlier lookup reproduces a +missing settled hint; the original source is restored after that mutation check. +No provider call moves inside this transaction. + +The latest Greptile finding also reproduced: while 60 healthy due connections +drained in three bounded batches, a different company's failing inventory read +ran three times instead of once. Failed inventory scopes now retain their own +30-second retry deadline. Healthy backlog continuations cannot shorten it, and +cached connections in an unavailable scope cannot be probed from stale data. +The regression requires all healthy checks to continue promptly, failed reads +to remain at least 30 seconds apart, eventual recovery, and SQL quiet after +settlement. Successful global reconstruction clears the covered failed scopes. +This changes only the health owner; coordinator and database contracts stay intact. + +The six external-database producer suites now apply pending fixture migrations +before creating their client. All six pass against separate, initially empty +PostgreSQL 17 databases on source `acbf262843`: 26 health, 469 tool-access, +18 access, 45 email, four GitHub-event, and seven Smoke Lab cases (569 total). +Server typecheck and build also pass on that source. The subsequent health +retry repair has a separate failing baseline and passing focused test; its +complete consumer and type/build verification is recorded with publication. +These source checks do not establish deployed suspension, Cloud wake coverage, +a full local repository-suite pass, or savings. Fresh current-head reviews and +all CI gates remain mandatory before merge. + +Final retry-repair acceptance: all four complete health, company-batching, +GitHub health recovery, and server startup suites pass 80 cases. Together with +the five complete producer suites above, this covers 623 distinct cases; the +26 earlier health cases are included in that total rather than added again. +Server typecheck and build pass with the retry repair. Startup tests emit the +existing SIGINT/SIGTERM listener-count warnings; no assertion failed. The fresh +fixture results and these final worker results have different source scopes as +stated above. The failed amplification, stale-type mutation, and retry baselines +are retained; a passing acceptance does not erase those failures. 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 8d0416e6961..250d61f86e1 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1291,3 +1291,101 @@ and integrates that accepted parent. Normal ancestry reconciliation preserves its entire verified tree (`df78ca3b5e`) byte for byte; only publication docs change subsequently. All original remaining duties remain in scope, beyond the 65 published numbered slices, including native backups and remaining aggregate owners. + +## Slice 18: tool-connection health implementation + +The health migration now reconstructs per-connection deadlines and reacts to +committed company changes instead of inventory scans on every maintenance tick. +See [health lifecycle and producer coverage](../idle-tool-connection-health.md). +Existing proactive health, provider policy, voice eligibility, transaction +uncertainty, drain, and shutdown behavior remain explicit. Seventeen lifecycle, +515 startup/tool-access, and 163 gateway/settings tests pass. Additional producer +consumers and publication gates remain open. This is not a completed PR yet. + +Fresh remote inspection: approval queue #17510 at `0d3100665d` has no failed or +pending checks. Plugin #17518 at `ba0b6752cd` has no failed checks and one pending +Paperclip Review check. The full original inventory and #17490 E2E cancellation +failure remain required; no PR was merged. + +Health consumer follow-up: 72 access/email/GitHub/smoke-lab cases pass using a +separate disposable DB per suite, and three selected chat cases pass. An initial +shared-DB attempt produced cleanup foreign-key failures after email; those are +recorded, not counted as passes. Total selected coverage is 770 cases. The server +build passes. The inventory SQL uses the existing company/enabled index with +40,000 retained rows (13 buffers, 0.022 ms). Full workspace and remote gates remain. + + +## Discord publication and health fairness repair + +Discord registration is [#17557](https://github.com/paperclipai/paperclip/pull/17557), +head `3755d7caced67589de474e0ca3e1b8b5b4e659c4`, based on #17554 and attached. +It has 639 selected passing tests, workspace typecheck/build, and final server +verification. CI run **40168371442** and review run **40168371311** are confirmed +live on that head. Fresh CI and review remain required; no merge is authorized. +The source and detailed producer audit are on the Discord child branch. + +The health parent Paperclip Review completed neutral with 4/5 at `da40899606`. +It requests fair company batching and direct real-service producer coverage. +The existing probes are already scoped by connection ID, so the review's concern +about full inventory scans around each probe needs clarification rather than +removing current-state eligibility checks. Self-generated observation hints still +cause a bounded coalesced company refresh; avoid muting correctness signals merely +to reduce that bounded work without proving equivalent recovery. + +Fairness was reproduced with 30 unchecked connections in one company and one due +connection in another. The quiet company was omitted from the first batch. The +shared selector from Discord commit `e77979d4fc` was applied to the health branch +as `097f5c8aff`; the health worker now rotates companies. All 22 health/selection +tests pass (`/tmp/sleep18-fairness-after.log`), after the recorded failing baseline +(`/tmp/sleep18-fairness-before.log`). The fix is local until the additional +AgentMail activation/pause, GitHub installation event, and access-removal producer +tests are complete. Then push the owning PR, request fresh review, and merge it +into the Discord branch. Full original inventory and earlier unresolved CI remain. + + +Health producer review coverage is now implemented and verified. The actual +AgentMail setup/health/pause, GitHub installation-event, and member-removal paths +move or cancel the live coordinator deadline. All 66 owner-suite cases pass on +isolated PostgreSQL databases, alongside the prior 22 fairness/health cases and +fresh server typecheck/build. Initial fixture expectations and the retained-user +DB setup failure are preserved in the health document. Publish these changes with +the fair company selector, request fresh review, and propagate to Discord #17557. +The Discord initial head has Greptile 5/5; its CI and Paperclip review were still +running when inspected. The remaining original inventory and #17490 failure +remain open; no stack PR is merged or declared fully accepted. + + +Discord's first Paperclip review (4/5) identified a dispatch-capacity issue shared +by health: 25 checks per 30 seconds delays a large already-due backlog. Health's +actual-probe timestamp regression failed with a 30.7-second gap, then passed when +only a remaining due batch receives a one-second continuation. All 22 health and +batching cases pass; server types pass. Propagate the parent and apply the same +bounded continuation to Discord, with its own regression. The Discord review also +requests preserving queued scopes after malformed hints and broader producer +coverage. Its initial head has Greptile 5/5 but these findings remain open. + + +Health security repair: five real connection API updates in a 205-connection +company reproduced 1,025 inventory rows. Individual writers now signal their +connection IDs; application writers signal their application IDs. Genuine broad +loads use 100-row UUID pages. The regression now reads exactly five rows, with +application invalidation and unrelated-deadline preservation verified. All 706 +targeted health/producer/consumer cases and server typecheck/build pass. Publish +the repair to #17554, propagate through company-deletion #17561 into Discord +#17557, and require fresh remote gates. No PR in this stack is merged. The next +new owner remains chat runtime discovery; the full remaining ledger is unchanged. + +## Acceptance follow-up after the plugin parent merged (October 27) + +Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at +`96727b3cfefb61f85ffe1fbc29a9b6dd6837efe8`. Health #17554 now includes that +actual master. It remains the next unmerged dependency, followed by company +chat deletion #17561 and Discord registration #17557. Later children cannot be +merged into their feature parents to bypass this order. + +See [health acceptance](../idle-tool-connection-health.md#acceptance-follow-up-after-the-plugin-parent-merged-october-27) +for the remaining installer amplification and independent retry repairs, failing +baselines, real concurrency proof, and fresh-database verification. Older stack +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. diff --git a/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md b/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md new file mode 100644 index 00000000000..067a3f64f3a --- /dev/null +++ b/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md @@ -0,0 +1,196 @@ +# Tool connection health sleep migration audit + +Parent: plugin scheduling #17518, local head `faac1bf85c`. +Fresh master inspected during this continuation: `71f06d84dc`. +Branch: `codex/sleep-18-tool-connection-health`. +This is an in-progress audit and failing baseline, not a completed migration. + +## Obligation and observed blocker + +The aggregate heartbeat calls `tools.sweepConnectionHealth()` at startup and on +its recurring maintenance tick. The health owner selects due connection IDs even +when no eligible connections exist. Its default freshness threshold is fifteen +minutes and its page size is 25. The startup regression on the unchanged worker +expected one startup call after two maintenance ticks and observed three calls. +The recorded command is: + +```sh +pnpm --filter @paperclipai/server exec vitest run \ + src/__tests__/server-startup-feedback-export.test.ts \ + -t 'does not rescan idle tool health' --maxWorkers=1 +``` + +The baseline failed with `expected 3 to be 1`. This proves aggregate activation; +real database worker tests must separately prove quiet SQL, deadlines, and producer +reactivation after implementation. A removed startup callback alone is not proof. + +## Ownership and policy + +`tool-access/catalog/health.ts` owns provider health policy and delegates actual +provider behavior through existing discovery, credential, catalog, and runtime +owners. Preserve this boundary. Shared scheduling must not branch on provider +names or decode their credential payloads. + +Existing `tool_connections.health_checked_at` and connection eligibility are the +persistent source of truth. No additional health queue table is needed. Keep the +manual `checkHealth` API and normal use/refresh health updates working. Preserve +the current proactive fifteen-minute policy for eligible external checks unless +a provider's existing event/replay contract makes a check unnecessary. Clearly +name that retained external obligation: scheduling the next real check removes +empty scans but does not make a configured external source network-quiet forever. +Do not extend Cloud alarm coverage or remove active-connection sleep blockers in +this slice. + +The current SQL excludes `chat_sdk`, `runtime_auth`, and applications of type +`paperclip_plugin`. The schema couples AI purpose to `runtime_auth`, and +`checkConnectionHealth` returns immediately for AI purpose. Do not claim AI rows +are an uncovered polling case without evidence of supported legacy rows outside +those constraints. PR #14638 separately covers AI health behavior. + +Audit voice eligibility, experimental connector disablement, and exceptions before +the existing health update: a skipped check that leaves its timestamp unchanged +must not starve later candidates or cause an immediate retry loop. Keep admission +and authorization failures with their owners. Do not silently disable or delete +connections to manufacture quiet. + +## Producer inventory to finish + +This list identifies writer families; it does not yet certify complete coverage. +Classify each write by changes to eligibility, saved deadline, or relevant probe +configuration before adding a notification. Excluded transport writers need no +health wake merely because they share a table. + +- `tool-access.ts`: direct create/update, examples, gallery install/finalization, + legacy imports, application replacement/deletion, and credential configuration. +- `tool-access/connections/installs.ts` and `removal.ts`: enablement, removal, + restore paths, existing transaction ownership and downstream cleanup. +- `tool-access/catalog/health.ts`, `publication.ts`, `discovery.ts`, and + `aggregator-accounts.ts`: successful and failed probe/catalog results and + eligibility changes. Health writes must move or cancel the recorded deadline. +- `tool-access/oauth/*`: access finalization, authorization start, direct/cloud/ + Vercel callbacks, client registration, grant/connection refresh, and refresh + lease bookkeeping. Lease-only writes are not automatically health producers. +- `tool-access/token-broker/service.ts`, `tool-gateway.ts`, + `tool-oauth-legacy-backfill.ts`, and `github-connection-events.ts`: actual-use + health observations, credential repair, and access changes. +- Email connections/channels and Slack-search OAuth: determine which existing + REST/MCP connection writes remain eligible for this health owner. +- Access provisioning, agent routes, smoke-lab setup, AI connection services, + and chat provider modules: classify supported imports and shared entry points; + do not attach unnecessary wakes to excluded AI/chat transports. +- Application type changes and company deletion: retain SQL join eligibility and + remove stale deadlines after cascades. Instance experimental settings are an + intentional administrative scope, distinct from ordinary tenant mutations. + +Use existing transaction notifications before writes, including caller-owned +transactions. Ordinary company mutations must not activate cross-company scans. +Startup reconstruction is a separate instance-owned obligation. Process-local +hints are not a multi-replica or independent-SQL-writer protocol. + +## Required verification + +- Empty, disabled, archived, excluded-transport, and retained-history quietness. +- Configured eligible future check: no intervening SQL; run at the saved deadline. +- First creation/enablement and changed credentials through real service producers. +- Successful/failed manual probe and catalog/use health updates reschedule correctly. +- Disablement, deletion, and feature changes cancel or restore the proper obligation. +- Page fairness with more than 25 due rows and with an early uncheckable candidate. +- Company scope, outer commit/rollback, a commit during final reconciliation, + drain/resume, startup reconstruction, admitted shutdown, and failure recovery. +- Keep provider calls outside transactions and avoid concurrent checks that can + overwrite newer credentials or health observations with an old result. +- Actual query plan on representative retained history, current consumers, + required build/type gates, and fresh PR review/checks. + +## Related work + +Open PRs #15448, #15325, and #11910 concern MCP failure classification and continued +usability after transient tool errors. This migration must preserve that owning +policy rather than choose new disablement behavior in the scheduler. The search +also found #14638 for AI health exclusions. Inspect exact diffs before changing +those paths; titles alone do not establish duplication or compatibility. + + +## Continuation after parent review repairs + +The health branch now includes approval backoff/shared matching at `0d3100665d` +and plugin coalescing/shutdown repair at `ba0b6752cd`. The combined parent +regressions pass all 73 cases. Fresh master fetched and inspected is `a0e5166e97`; +its aggregate tick still calls the health sweep. Parent PR checks are still in +progress. The health migration is not implemented or published yet. + +Additional owner evidence changes the implementation plan: + +- `checkConnectionHealth` can call catalog discovery before writing its final + health observation. `catalog/publication.ts` itself changes health timestamps, + config, transport config, and `updatedAt`. A blanket comparison with the + connection's original `updatedAt` would reject this operation's own writes. + Concurrency fencing must distinguish a newer external configuration/observation + from the current probe's owned catalog publication; do not add a naive row-version + check that silently drops legitimate results. +- `oauth/refresh-leases.ts` changes config and `updatedAt` merely to acquire or + release its existing refresh lease. Those are active-operation bookkeeping, + not independent reasons for proactive health work. Do not notify health for + every config-column write by pattern matching. +- `oauth/connection-refresh.ts` has two different health-relevant transitions: + reauthorization failure changes status to draft and disables the connection; + successful token refresh changes credential refs/config while keeping the + connection usable. The former must cancel eligibility; the latter must refresh + the owning schedule without a global scan or a credential-refresh feedback loop. +- Experimental setup policy resolves application identity and may reject before + the current health try/catch writes a timestamp. Disabled candidates therefore + need an explicit parked state and reactivation from instance feature changes; + they must not monopolize the bounded due page or lose their configuration. + +These are audited constraints, not implemented guarantees. Complete producer +classification and prove these paths before removing the health backstop. + +## Implemented producer contract, October 25 + +The implementation and producer dispositions are now recorded in +[the health lifecycle document](../idle-tool-connection-health.md). Health +creation/setup mutations commit their grant/binding/runtime-slot SQL before the +wake. Direct/cloud/Vercel OAuth activation and refresh disablement, catalog and +actual-use observations, email control, access removal, company deletion, and +feature changes now notify the owner. Voice connections remain covered through +the chat lifecycle; the provider-to-transport mapping stays in the chat catalog, +and excluded SDK chat updates do not notify health work. + +The aggregate health scan is replaced by the shared worker. Seventeen lifecycle, +515 startup/tool-access, and 163 gateway/settings cases pass. The first consumer +run exposed an ordinal transaction mock; a real SQL restoration fault preserves +the intended regression and passes. Producer/chat consumers, query-plan evidence, +build, and remote gates remain pending. Master was refreshed to `8415f4a679`; +this uncommitted health branch remains based on the open migration stack, so +final stack reconciliation with master is still required. No merge is authorized. + + +## Security follow-up: individual refreshes + +The company-hint design amplified ordinary connection writes into full company +reads. The current contract scopes individual producers (API/example creation, +setup/rollback, OAuth, gateway, health/catalog, email, voice, GitHub events, access +removal, and Smoke Lab) by company plus connection ID. Existing application +updates in example setup and Smoke Lab use application scopes. Company deletion +and instance feature-settings changes retain broad invalidation. Startup and +broad loads use bounded UUID pages. The health owner still defines eligibility +and provider policy; the shared notification/coordinator layers remain generic. + +See `doc/idle-tool-connection-health.md` for the failing amplification baseline, +changed contract, scope-cancellation tests, and verification results. The older +company-hint observations in this audit describe the superseded implementation. + +## Acceptance follow-up after the plugin parent merged (October 27) + +The actual merged parent is master `96727b3cfefb61f85ffe1fbc29a9b6dd6837efe8`. +All 18 preceding sleep PRs are merged. Historical unmerged-state statements +above describe earlier revisions. + +The final producer and retry follow-up is recorded in +[the health lifecycle document](../idle-tool-connection-health.md#acceptance-follow-up-after-the-plugin-parent-merged-october-27). +It preserves the two real installer amplification failures, type-change lock-wait +mutation proof, independent failed-inventory retry regression, fresh database +fixture setup, and verification scope. Metadata-only installer writes do not +invalidate an entire application; actual eligibility changes still do. Inventory +failures do not inherit another company's one-second backlog continuation. +Current-head reviews and CI remain required; these are worker-level guarantees. diff --git a/doc/plans/2026-10-25-configuration-worker-drain-repair.md b/doc/plans/2026-10-25-configuration-worker-drain-repair.md new file mode 100644 index 00000000000..99ebdb41834 --- /dev/null +++ b/doc/plans/2026-10-25-configuration-worker-drain-repair.md @@ -0,0 +1,31 @@ +# Configuration worker drain repairs + +The voice drain audit exposed a pattern in three earlier independent workers: +retained dirty scopes reported as pending work keep the coordinator's intent +activity hold alive after the write has settled. No recurring SQL is necessary +to block idle admission this way. The earlier tests checked traffic but did not +check the idle counter or generation while paused. + +Each owner must retain its hints and next deadline for resume, report dirty work +as pending only while dispatch is allowed, and leave unresolved transaction +fencing to the shared coordinator. No hint is acknowledged or discarded during +drain. Business health/registration/lease deadlines and provider behavior stay +unchanged. This is an idle-work boundary repair, not provider sleep qualification. + +## Tool health — slice 18 / #17554 + +Baseline 9ef6e4f4c8 reproduces the bug: the strengthened existing drain case fails +with active=1 after commit, recorded in /tmp/sleep18-drain-before.log. The repaired +case asserts active=0, then a stable idle snapshot and no SQL/provider calls for a +minute, then the saved work runs on resume without another producer. + +All 23 lifecycle cases pass in /tmp/sleep18-drain-tests.log; production server +typecheck passes in /tmp/sleep18-drain-typecheck.log. The test database is the +fresh migrated paperclip_sleep18_drain_20261025_test. No test failures were retried +blindly. Current master bf9a8551c3 has no changes to these three worker owners +since the preceding aec86c794b inspection; full stack reconciliation remains open. + +Discord command registration and receiver discovery require the same causal +check at their own parents, then propagation through every dependent stack PR. +The full original recurring-work inventory, milestone producer audit, older E2E +failure and inbound cleanup/deletion policy finding remain outstanding. diff --git a/server/src/__tests__/access-service.test.ts b/server/src/__tests__/access-service.test.ts index 2a00221be42..0415965c955 100644 --- a/server/src/__tests__/access-service.test.ts +++ b/server/src/__tests__/access-service.test.ts @@ -1,7 +1,10 @@ +import { startToolHealthWorker } from "./helpers/tool-health-worker.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { and, eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { + applyPendingMigrations, agents, companySecretBindings, companySecrets, @@ -28,7 +31,8 @@ import { accessService } from "../services/access.js"; import { grantsForHumanRole } from "../services/company-member-roles.js"; import { backfillPrincipalAccessCompatibility } from "../services/principal-access-compatibility.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ACCESS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; async function createCompanyWithOwner(db: ReturnType) { @@ -61,8 +65,9 @@ describeEmbeddedPostgres("access service", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-access-service-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-access-service-"); + if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -85,6 +90,7 @@ describeEmbeddedPostgres("access service", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -342,6 +348,7 @@ describeEmbeddedPostgres("access service", () => { credentialPolicy: "per_user", status: "active", enabled: true, + healthCheckedAt: new Date(), }).returning().then((rows) => rows[0]!); const definition = await db.insert(userSecretDefinitions).values({ companyId: company.id, @@ -372,9 +379,15 @@ describeEmbeddedPostgres("access service", () => { agentId: agent.id, createdByUserId: member.principalId, }); - await accessService(db).archiveMember(company.id, member.id, { - reassignment: { assigneeUserId: owner.principalId }, - }); + const health = await startToolHealthWorker(db, company.id); + try { + await health.expectDeadline(connection.healthCheckedAt!.getTime() + 900_000); + await accessService(db).archiveMember(company.id, member.id, { + reassignment: { assigneeUserId: owner.principalId }, + }); + await health.expectDeadline(null); + await health.expectQuiet(); + } finally { await health.stop(); } expect(await db.select().from(connectionGrantDelegations)).toHaveLength(0); expect(await db.select().from(connectionGrantMembers)).toHaveLength(0); diff --git a/server/src/__tests__/email-channels.integration.test.ts b/server/src/__tests__/email-channels.integration.test.ts index bf297326a1c..b86338915f8 100644 --- a/server/src/__tests__/email-channels.integration.test.ts +++ b/server/src/__tests__/email-channels.integration.test.ts @@ -1,3 +1,5 @@ +import { startToolHealthWorker } from "./helpers/tool-health-worker.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { aiConnectionService } from "../services/ai-connections.js"; import { fastResponseService } from "../services/fast-responses.js"; import { fastResponseReceipt } from "../services/fast-response-provider.js"; @@ -33,6 +35,7 @@ import { } from "vitest"; import { and, eq, inArray, ne, sql } from "drizzle-orm"; import { + applyPendingMigrations, createDb, companies, agents, @@ -72,6 +75,8 @@ import { toolAccessService } from "../services/tool-access.js"; import { emailSendSchema } from "@paperclipai/shared"; import { chatChannelService } from "../services/chat-channels.js"; +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EMAIL_TEST_DATABASE_URL"); + describe("AgentMail durable email pipeline", () => { let database: Awaited>; let db: ReturnType; @@ -83,8 +88,9 @@ describe("AgentMail durable email pipeline", () => { folder, "master.key", ); - database = await startEmbeddedPostgresTestDatabase("paperclip-email-"); - db = createDb(database.connectionString); + if (!externalTestDatabaseUrl) database = await startEmbeddedPostgresTestDatabase("paperclip-email-"); + if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl); + db = createDb(externalTestDatabaseUrl ?? database!.connectionString); await instanceSettingsService(db).updateExperimental({ enableChatConnectors: false, }); @@ -111,6 +117,7 @@ describe("AgentMail durable email pipeline", () => { ); }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); if (previous === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; @@ -210,7 +217,7 @@ describe("AgentMail durable email pipeline", () => { }); async function fixture(mode: "websocket" | "webhook" = "webhook", storage?: StorageService, - beforeSetup?: (input: { fetcher: typeof fetch; wakeup: ReturnType }) => Promise, + beforeSetup?: (input: { companyId: string; fetcher: typeof fetch; wakeup: ReturnType }) => Promise, ) { const companyId = randomUUID(), agentId = randomUUID(), @@ -324,7 +331,7 @@ describe("AgentMail durable email pipeline", () => { storage, }); services.push(service); - await beforeSetup?.({ fetcher, wakeup }); + await beforeSetup?.({ companyId, fetcher, wakeup }); await service.setup( companyId, { @@ -387,6 +394,28 @@ describe("AgentMail durable email pipeline", () => { setWebhookError: (status: number) => { webhookError = status; }, }; } + it("schedules health after AgentMail activation and disarms it on pause", async () => { + let health: Awaited> | undefined; + try { + const f = await fixture("webhook", undefined, async ({ companyId, fetcher }) => { + vi.spyOn(globalThis, "fetch").mockImplementation(fetcher); + health = await startToolHealthWorker(db, companyId, new Date(), toolAccessService(db).checkHealth); + await health.expectDeadline(null); + await health.expectQuiet(); + }); + const endpoint = await f.service.getEndpoint(f.endpointId); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1_000); + expect(health!.checkHealth).toHaveBeenCalledTimes(1); + expect(health!.checkHealth.mock.settledResults[0]?.type).toBe("fulfilled"); + }, { timeout: 5_000 }); + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.id, endpoint.connectionId)); + await health!.expectDeadline(connection.healthCheckedAt!.getTime() + 900_000); + await f.service.control(f.endpointId, "pause", { userId: "email-board" }); + await health!.expectDeadline(null); + await health!.expectQuiet(); + } finally { await health?.stop(); } + }); it("activates an empty running worker from another service and drains committed inbound and outbound work", async () => { // Connector setup and tool calls construct short-lived service objects. // Their committed writes must wake the app-owned worker sharing this DB. diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 6bcd2913396..7c7b7025243 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -1,6 +1,9 @@ +import { startToolHealthWorker } from "./helpers/tool-health-worker.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { + applyPendingMigrations, activityLog, companies, connectionEventDeliveries, @@ -21,7 +24,8 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_GITHUB_EVENTS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("GitHub connection event delivery", () => { @@ -29,8 +33,9 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-github-events-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-github-events-"); + if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterEach(async () => { @@ -44,6 +49,7 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -264,6 +270,7 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { credentialPolicy: "per_user", status: "active", enabled: true, + healthCheckedAt: new Date("2026-09-04T11:59:05.000Z"), config: { sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } }, transportConfig: {}, }); @@ -329,11 +336,18 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { let currentTime = new Date("2026-09-04T12:00:05.000Z"); const service = githubConnectionEventService(db, { connector, now: () => currentTime }); - const unsubscribe = subscribeCompanyLiveEvents(companyId, () => { - throw new Error("fixture live subscriber failed"); - }); - await expect(service.pollOnce()).resolves.toMatchObject({ processed: 1, duplicate: 0, failed: 0 }); - unsubscribe(); + const health = await startToolHealthWorker(db, companyId, currentTime); + try { + await health.expectDeadline(new Date("2026-09-04T12:14:05.000Z").getTime()); + const unsubscribe = subscribeCompanyLiveEvents(companyId, () => { + throw new Error("fixture live subscriber failed"); + }); + try { + await expect(service.pollOnce()).resolves.toMatchObject({ processed: 1, duplicate: 0, failed: 0 }); + } finally { unsubscribe(); } + await health.expectDeadline(currentTime.getTime() + 900_000 - (refreshed ? 60_000 : 0)); + await health.expectQuiet(); + } finally { await health.stop(); } let [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grantId)); expect(grant?.providerTenant?.github).toMatchObject({ repositoryCount: refreshed ? 3 : 4, webhookHealth: "healthy" }); if (refreshed) { diff --git a/server/src/__tests__/github-health-recovery.test.ts b/server/src/__tests__/github-health-recovery.test.ts index 27555b23251..410d6ce7c6c 100644 --- a/server/src/__tests__/github-health-recovery.test.ts +++ b/server/src/__tests__/github-health-recovery.test.ts @@ -11,6 +11,13 @@ vi.mock("../services/tool-access/connections/projection.js", () => ({ }), })); +// This unit fixture owns GitHub error projection; the health-worker suite +// exercises the real transaction and notification boundary with PostgreSQL. +vi.mock("../services/tool-connection-health-notifications.js", async importOriginal => ({ + ...await importOriginal(), + writeToolConnectionHealthChange: async (db: Db, _target: unknown, write: (transaction: Db) => Promise) => write(db), +})); + const SUSPENDED_INSTALLATION_URL = "https://github.com/organizations/paperclipai/settings/installations/101"; function json(value: unknown): Response { diff --git a/server/src/__tests__/helpers/tool-health-worker.ts b/server/src/__tests__/helpers/tool-health-worker.ts new file mode 100644 index 00000000000..d51c18c221d --- /dev/null +++ b/server/src/__tests__/helpers/tool-health-worker.ts @@ -0,0 +1,52 @@ +import type { Db } from "@paperclipai/db"; +import { expect, vi } from "vitest"; +import { createDeliveryWorkCoordinator } from "../../services/delivery-work-coordinator.js"; +import { connectionHealthSchedule } from "../../services/tool-access/catalog/health-schedule.js"; +import { registerToolConnectionHealthWork } from "../../services/tool-connection-health-work.js"; +import { idleWorkSnapshot } from "../../services/task-admission.js"; + +export async function startToolHealthWorker(db: Db, companyId: string, now = new Date(), probe?: Parameters[1]) { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + vi.setSystemTime(now); + const errors: unknown[] = []; + const checkHealth = vi.fn(probe ?? (async () => { throw new Error("Unexpected due health probe"); })); + const schedule = connectionHealthSchedule(db, checkHealth); + const coordinator = createDeliveryWorkCoordinator({ + owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error), + }); + try { + await registerToolConnectionHealthWork(coordinator, { + load: (scope, connectionId, applicationId, signal) => schedule.load(scope ?? companyId, connectionId, applicationId, signal), check: schedule.check, + }, () => true).ready; + } catch (error) { + await coordinator.stop(); + vi.useRealTimers(); + throw error; + } + return { + checkHealth, + async expectDeadline(expected: number | null) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1_000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBe(expected); + }, { timeout: 5_000 }); + expect(errors).toEqual([]); + if (!probe) expect(checkHealth).not.toHaveBeenCalled(); + }, + async expectQuiet() { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); + expect(writes).not.toHaveBeenCalled(); + expect(executes).not.toHaveBeenCalled(); + } finally { + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + } + }, + async stop() { + try { await coordinator.stop(); } finally { vi.useRealTimers(); } + }, + }; +} diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index d0bb6ab1c33..566a45df079 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -9,7 +9,7 @@ import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import * as idleLocalWork from "../services/idle-local-work.js"; import { idleBackupWakeMarker, markIdleBackupWakeRequired } from "../services/idle-database-backup.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; -import { decisionService, decisionRetentionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; +import { decisionService, decisionRetentionService, toolAccessService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; import { companies, runDatabaseBackup, type Db } from "@paperclipai/db"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -432,6 +432,7 @@ vi.mock("../services/index.js", () => ({ routineService: routineServiceFactoryMock, statusCardService: vi.fn(() => ({})), toolAccessService: vi.fn(() => ({ + scheduledHealth: { load: vi.fn(async () => []), check: vi.fn(async () => null) }, sweepConnectionHealth: vi.fn(async () => ({ checked: 0, healthy: 0, @@ -881,6 +882,31 @@ describe("startServer feedback export wiring", () => { expect(completionSweepMock).not.toHaveBeenCalled(); }); + it("does not rescan idle tool health from the aggregate maintenance tick", async () => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true, heartbeatSchedulerIntervalMs: 30_000 })); + let tick: (() => void) | undefined; + const intervals = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, ms: number) => { + if (ms === 30_000) tick = callback; + return 1 as unknown as ReturnType; + }) as typeof setInterval); + try { + await startServer(); await settleStartupHousekeeping(); + const sweeps = vi.mocked(toolAccessService).mock.results.map(result => vi.mocked(result.value.sweepConnectionHealth)); + const count = () => sweeps.reduce((total, sweep) => total + sweep.mock.calls.length, 0); + expect(count()).toBe(0); + const registrations = deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.toolConnectionHealth); + expect(registrations).toHaveLength(1); + const loads = vi.mocked(toolAccessService).mock.results.map(result => result.value.scheduledHealth.load); + const loadCount = () => loads.reduce((total, load) => total + load.mock.calls.length, 0); + expect(loadCount()).toBe(1); + expect(tick).toEqual(expect.any(Function)); + tick!(); await new Promise(resolve => setImmediate(resolve)); + tick!(); await new Promise(resolve => setImmediate(resolve)); + expect(count()).toBe(0); + expect(loadCount()).toBe(1); + } finally { intervals.mockRestore(); } + }); + it("never invokes the retired review detector at startup or on periodic recovery", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true, diff --git a/server/src/__tests__/smoke-lab.test.ts b/server/src/__tests__/smoke-lab.test.ts index 297f5843ad5..165ca21fc51 100644 --- a/server/src/__tests__/smoke-lab.test.ts +++ b/server/src/__tests__/smoke-lab.test.ts @@ -1,8 +1,10 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { + applyPendingMigrations, activityLog, agents, authUsers, @@ -27,7 +29,8 @@ import { smokeLabRoutes } from "../routes/smoke-lab.js"; import { SMOKE_LAB_OAUTH_SCOPE } from "../services/smoke-lab.js"; import { errorHandler } from "../middleware/index.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_SMOKE_LAB_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; type TestDb = ReturnType; @@ -104,8 +107,9 @@ describeEmbeddedPostgres("smoke lab service pack and results API", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-smoke-lab-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-smoke-lab-"); + if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -128,6 +132,7 @@ describeEmbeddedPostgres("smoke lab service pack and results API", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/tool-access-service.test.ts b/server/src/__tests__/tool-access-service.test.ts index e5629ead55f..9aaf534dc1a 100644 --- a/server/src/__tests__/tool-access-service.test.ts +++ b/server/src/__tests__/tool-access-service.test.ts @@ -16,6 +16,7 @@ import { vi, } from "vitest"; import { + applyPendingMigrations, activityLog, agents, authUsers, @@ -923,6 +924,7 @@ describeEmbeddedPostgres("tool access service", () => { beforeAll(async () => { if (externalDatabaseUrl) { + await applyPendingMigrations(externalDatabaseUrl); db = createDb(externalDatabaseUrl); await instanceSettingsService(db).updateExperimental({ enableMemoryConnectors: true }); return; @@ -15884,38 +15886,44 @@ describeEmbeddedPostgres("tool access service", () => { ); await service.archiveConnection(first.connectionId, company.id, actor); fetchMock.mockRejectedValue(new Error("provider unavailable")); - const runTransaction = db.transaction.bind(db); - vi.spyOn(db, "transaction") - .mockImplementationOnce(runTransaction) - .mockRejectedValueOnce(new Error("rollback unavailable")); + await db.execute(sql`create function tool_health_rollback_failure() returns trigger language plpgsql as $$ + begin raise exception 'rollback unavailable'; end $$`); + await db.execute(sql.raw(`create trigger tool_health_rollback_failure before update on tool_connections + for each row when (old.id = '${first.connectionId}'::uuid and old.credential_policy = 'per_user' and new.credential_policy = 'shared') + execute function tool_health_rollback_failure()`)); + try { - await expect( - service.connectGalleryApp( - company.id, - { - galleryKey: "github", - connectionMethodKey: "mcp-key", - grantKind: "user", - name: "GitHub rollback failure", - credentialValues: { - "credentials.authorization": "new-personal-token", + await expect( + service.connectGalleryApp( + company.id, + { + galleryKey: "github", + connectionMethodKey: "mcp-key", + grantKind: "user", + name: "GitHub rollback failure", + credentialValues: { + "credentials.authorization": "new-personal-token", + }, }, - }, - actor, - ), - ).rejects.toMatchObject({ - status: 500, - details: { code: "connection_identity_rollback_failed" }, - }); + actor, + ), + ).rejects.toMatchObject({ + status: 500, + details: { code: "connection_identity_rollback_failed" }, + }); - await expect( - service.getConnection(first.connectionId, company.id), - ).resolves.toMatchObject({ - status: "draft", - enabled: false, - healthStatus: "error", - lastError: "connection_identity_rollback_failed", - }); + await expect( + service.getConnection(first.connectionId, company.id), + ).resolves.toMatchObject({ + status: "draft", + enabled: false, + healthStatus: "error", + lastError: "connection_identity_rollback_failed", + }); + } finally { + await db.execute(sql`drop trigger tool_health_rollback_failure on tool_connections`); + await db.execute(sql`drop function tool_health_rollback_failure()`); + } }); it("automatically gives same-named connections distinct names", async () => { diff --git a/server/src/__tests__/tool-connection-health-work.test.ts b/server/src/__tests__/tool-connection-health-work.test.ts new file mode 100644 index 00000000000..52f4afc5b20 --- /dev/null +++ b/server/src/__tests__/tool-connection-health-work.test.ts @@ -0,0 +1,494 @@ +import { logger } from "../middleware/logger.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { deleteCompany } from "../services/company-deletion.js"; +import { notifyChatConnectionHealth } from "../services/chat-channels/connection-health-notifications.js"; +import { smokeLabService } from "../services/smoke-lab.js"; +import { toolAccessService } from "../services/tool-access.js"; +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { and, eq, ne, sql } from "drizzle-orm"; +import { applyPendingMigrations, companies, createDb, subscribeDatabaseWork, toolApplications, toolConnections } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { connectionHealthSchedule } from "../services/tool-access/catalog/health-schedule.js"; +import { registerToolConnectionHealthWork } from "../services/tool-connection-health-work.js"; +import { writeToolConnectionHealthChange } from "../services/tool-connection-health-notifications.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_TOOL_HEALTH_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("scheduled tool connection health", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, applicationId: string; + let coordinator: ReturnType | undefined; + let draining = false; + const errors: unknown[] = []; + const probe = vi.fn(); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("tool-health-"); + if (external) await applyPendingMigrations(external); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + companyId = randomUUID(); draining = false; errors.length = 0; + await db.insert(companies).values({ id: companyId, name: "Tool health", issuePrefix: `H${companyId.slice(0, 6)}` }); + const [app] = await db.insert(toolApplications).values({ companyId, name: "Health fixture", type: "mcp_http" }).returning(); + applicationId = app.id; + probe.mockReset(); probe.mockImplementation(async (id: string) => { + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.id, id)); + await writeToolConnectionHealthChange(db, connection, async tx => { + await tx.update(toolConnections).set({ healthCheckedAt: new Date(), healthStatus: "ok" }).where(eq(toolConnections.id, id)); + }); + return {}; + }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(companies).where(eq(companies.id, companyId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + function connection(patch: Partial = {}, signal = true) { + const id = patch.id ?? randomUUID(); + const insert = async (writer: typeof db) => (await writer.insert(toolConnections).values({ + id, companyId, applicationId, name: "Check", uid: randomUUID(), transport: "mcp_remote", status: "active", enabled: true, ...patch, + }).returning())[0]; + return signal ? writeToolConnectionHealthChange(db, { companyId, id }, insert) : insert(db); + } + async function start(service = connectionHealthSchedule(db, probe)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); + await registerToolConnectionHealthWork(coordinator, service, () => !draining).ready; + return service; + } + async function idleAt(expected: number | null) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBe(expected); + }, { timeout: 5000 }); + } + async function quiet() { + await idleAt(null); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(180_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + } + async function checked(count = 1) { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(probe).toHaveBeenCalledTimes(count); }, { timeout: 5000 }); + } + it("stays SQL-quiet after empty startup", async () => { await start(); await quiet(); }); + it("waits without SQL until a configured check is due", async () => { + const checkedAt = new Date(); await connection({ healthCheckedAt: checkedAt }); + await start(); const due = checkedAt.getTime() + 900_000; await idleAt(due); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(180_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + await vi.advanceTimersByTimeAsync(due - Date.now()); await checked(); + }); + it("refreshes only the changed connection after real API writes in a large company", async () => { + const checkedAt = new Date(); + const rows = await db.insert(toolConnections).values(Array.from({ length: 205 }, (_, index) => ({ + companyId, applicationId, name: `Inventory ${index}`, uid: randomUUID(), transport: "mcp_remote" as const, + config: { url: "https://health.example.test/mcp" }, + status: "active" as const, enabled: true, healthCheckedAt: checkedAt, + }))).returning(); + const tools = toolAccessService(db, { remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }] }); + const service = tools.scheduledHealth, load = service.load; + let returnedRows = 0; + service.load = async (...args) => { const result = await load(...args); returnedRows += result.length; return result; }; + await start(service); await idleAt(checkedAt.getTime() + 900_000); + returnedRows = 0; + for (const row of rows.slice(0, 5)) { + await tools.updateConnection(row.id, { name: `${row.name} changed` }); + await idleAt(checkedAt.getTime() + 900_000); + } + expect(returnedRows).toBe(5); + expect(probe).not.toHaveBeenCalled(); + }); + it.each(["example", "smoke"] as const)("does not reload unrelated connections when reinstalling %s metadata", async kind => { + const settings = instanceSettingsService(db); + const enabled = (await settings.getExperimental()).enableSmokeLab; + const tools = toolAccessService(db); + const smoke = smokeLabService(db, { deploymentMode: "local_trusted", deploymentExposure: "private" }); + const install = () => kind === "example" ? tools.installExample(companyId, "safe-read-only-todo-kv") + : smoke.installFixtures(companyId); + try { + if (kind === "smoke") await settings.updateExperimental({ enableSmokeLab: true }); + await install(); + const fixtureConnections = await db.select().from(toolConnections).where(eq(toolConnections.companyId, companyId)); + const [application] = await db.select().from(toolApplications).where(and( + eq(toolApplications.companyId, companyId), ne(toolApplications.id, applicationId), + )); + const checkedAt = new Date(); + const unrelated = await db.insert(toolConnections).values(Array.from({ length: 205 }, (_, index) => ({ + companyId, applicationId: application.id, name: `Other ${index}`, uid: randomUUID(), + transport: "mcp_remote" as const, enabled: true, status: "active" as const, + healthCheckedAt: checkedAt, config: { url: "https://health.example.test/mcp" }, + }))).returning({ id: toolConnections.id }); + await db.update(toolConnections).set({ healthCheckedAt: checkedAt }).where(eq(toolConnections.companyId, companyId)); + const service = connectionHealthSchedule(db, probe), load = service.load; + let returnedRows = 0, excludedApplicationLoaded = false; + const loadedIds = new Set(); + service.load = async (...args) => { + const rows = await load(...args); returnedRows += rows.length; + for (const row of rows) loadedIds.add(row.id); + if (args[2] === application.id && rows.length === 0) excludedApplicationLoaded = true; + return rows; + }; + await start(service); await idleAt(checkedAt.getTime() + 900_000); returnedRows = 0; + for (let attempt = 0; attempt < 5; attempt++) { + await install(); await idleAt(checkedAt.getTime() + 900_000); + } + expect(returnedRows).toBeLessThanOrEqual(10 * fixtureConnections.length); + await writeToolConnectionHealthChange(db, { companyId, applicationId: application.id }, async tx => { + await tx.update(toolApplications).set({ type: "paperclip_plugin" }).where(eq(toolApplications.id, application.id)); + }); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(excludedApplicationLoaded).toBe(true)); + loadedIds.clear(); + await install(); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(unrelated.every(row => loadedIds.has(row.id))).toBe(true)); + } finally { + await coordinator?.stop(); coordinator = undefined; + if (kind === "smoke") await settings.updateExperimental({ enableSmokeLab: enabled }); + } + }); + + it("notifies restored application eligibility after waiting for a concurrent type change", async () => { + const tools = toolAccessService(db); + const installed = await tools.installExample(companyId, "safe-read-only-todo-kv"); + const hints = new Set(); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.toolConnectionHealth, (event, changed) => { + if (event === "settled") for (const hint of changed ?? []) hints.add(hint); + }); + let unlock!: () => void, acquired!: () => void; + const gate = new Promise(resolve => { unlock = resolve; }); + const admitted = new Promise(resolve => { acquired = resolve; }); + const holder = db.transaction(async tx => { + await tx.update(toolApplications).set({ type: "paperclip_plugin" }).where(eq(toolApplications.id, installed.application.id)); + acquired(); await gate; + }); + await admitted; + let installing: ReturnType | undefined; + try { + installing = tools.installExample(companyId, "safe-read-only-todo-kv"); + void installing.catch(() => undefined); + await vi.waitFor(async () => { + const waits = await db.execute(sql`select 1 from pg_stat_activity where datname = current_database() + and pid <> pg_backend_pid() and wait_event_type = 'Lock' and query like '%tool_applications%'`); + expect(waits.length).toBe(1); + }); + unlock(); await holder; await installing; + expect(hints.has(`${companyId}:application:${installed.application.id}`)).toBe(true); + expect((await db.select().from(toolApplications).where(eq(toolApplications.id, installed.application.id)))[0].type).toBe("mcp_stdio"); + } finally { + unlock(); await Promise.allSettled([holder, installing]); unsubscribe(); + } + }); + + it("pages startup inventory while retaining every eligible deadline", async () => { + const checkedAt = new Date(); + await db.insert(toolConnections).values(Array.from({ length: 205 }, (_, index) => ({ + companyId, applicationId, name: `Page ${index}`, uid: randomUUID(), transport: "mcp_remote" as const, + status: "active" as const, enabled: true, healthCheckedAt: checkedAt, + }))); + const reads = vi.spyOn(db, "select"); + const rows = await connectionHealthSchedule(db, probe).load(companyId); + expect(rows).toHaveLength(205); + expect(new Set(rows.map(row => row.id)).size).toBe(205); + expect(reads).toHaveBeenCalledTimes(3); + reads.mockRestore(); + }); + it("invalidates an application's cached deadlines without removing another application's work", async () => { + const checkedAt = new Date(); + const [otherApp] = await db.insert(toolApplications).values({ companyId, name: "Other app", type: "mcp_http" }).returning(); + await connection({ healthCheckedAt: checkedAt }); + const other = await connection({ applicationId: otherApp.id, healthCheckedAt: new Date(checkedAt.getTime() + 60_000) }); + const service = connectionHealthSchedule(db, probe), load = vi.spyOn(service, "load"); + await start(service); await idleAt(checkedAt.getTime() + 900_000); load.mockClear(); + await writeToolConnectionHealthChange(db, { companyId, applicationId }, async tx => { + await tx.update(toolApplications).set({ type: "paperclip_plugin" }).where(eq(toolApplications.id, applicationId)); + }); + await idleAt(other.healthCheckedAt!.getTime() + 900_000); + expect(load).toHaveBeenCalledWith(companyId, undefined, applicationId, expect.any(AbortSignal)); + expect(load).toHaveBeenCalledTimes(1); + await writeToolConnectionHealthChange(db, { companyId, applicationId }, async tx => { + await tx.update(toolApplications).set({ type: "mcp_http" }).where(eq(toolApplications.id, applicationId)); + }); + await idleAt(checkedAt.getTime() + 900_000); + expect(probe).not.toHaveBeenCalled(); + }); + it("keeps unrelated deadlines when a connection changes application or is deleted", async () => { + const checkedAt = new Date(); + const first = await connection({ healthCheckedAt: checkedAt }); + const other = await connection({ healthCheckedAt: new Date(checkedAt.getTime() + 60_000) }); + const [plugin] = await db.insert(toolApplications).values({ companyId, name: "Excluded app", type: "paperclip_plugin" }).returning(); + await start(); await idleAt(checkedAt.getTime() + 900_000); + await writeToolConnectionHealthChange(db, first, async tx => { + await tx.update(toolConnections).set({ applicationId: plugin.id }).where(eq(toolConnections.id, first.id)); + }); + await idleAt(other.healthCheckedAt!.getTime() + 900_000); + await writeToolConnectionHealthChange(db, other, async tx => { await tx.delete(toolConnections).where(eq(toolConnections.id, other.id)); }); + await quiet(); + }); + it("keeps disabled, archived, and excluded transports quiet", async () => { + await connection({ enabled: false }); await connection({ status: "archived" }); + await connection({ transport: "chat_sdk", connectionPurpose: "channel" }); + await connection({ transport: "runtime_auth", connectionPurpose: "ai" }); + await start(); await quiet(); expect(probe).not.toHaveBeenCalled(); + }); + it("reactivates on the first committed connection and disarms after deletion", async () => { + await start(); await quiet(); const row = await connection(); await checked(); + await writeToolConnectionHealthChange(db, companyId, async tx => { await tx.delete(toolConnections).where(eq(toolConnections.id, row.id)); }); + await quiet(); + }); + it("defers nested creation until the outer transaction commits and ignores rollback", async () => { + await start(); await quiet(); + await expect(db.transaction(async tx => { + await writeToolConnectionHealthChange(tx as unknown as typeof db, companyId, async writer => { + await writer.insert(toolConnections).values({ companyId, applicationId, name: "Rollback", uid: randomUUID(), transport: "mcp_remote", status: "active", enabled: true }); + }); + await vi.advanceTimersByTimeAsync(60_000); expect(probe).not.toHaveBeenCalled(); + throw new Error("abort"); + })).rejects.toThrow("abort"); + await quiet(); + await db.transaction(async tx => { + await writeToolConnectionHealthChange(tx as unknown as typeof db, companyId, async writer => { + await writer.insert(toolConnections).values({ companyId, applicationId, name: "Commit", uid: randomUUID(), transport: "mcp_remote", status: "active", enabled: true }); + }); + await vi.advanceTimersByTimeAsync(60_000); expect(probe).not.toHaveBeenCalled(); + }); + await checked(); + }); + it("retains first writes through drain without querying", async () => { + await start(); await quiet(); draining = true; await connection(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + const drained = idleWorkSnapshot(); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); expect(idleWorkSnapshot()).toEqual(drained); + reads.mockRestore(); expect(probe).not.toHaveBeenCalled(); + draining = false; await checked(); + }); + it("moves the deadline when an actual-use health observation commits", async () => { + const row = await connection({ healthCheckedAt: new Date() }); await start(); + const checkedAt = new Date(Date.now() + 60_000); + await writeToolConnectionHealthChange(db, companyId, async tx => { await tx.update(toolConnections).set({ healthCheckedAt: checkedAt }).where(eq(toolConnections.id, row.id)); }); + await idleAt(checkedAt.getTime() + 900_000); expect(probe).not.toHaveBeenCalled(); + }); + it("reconstructs saved future checks after restart", async () => { + const checkedAt = new Date(); await connection({ healthCheckedAt: checkedAt }); await start(); + await coordinator!.stop(); await start(); await idleAt(checkedAt.getTime() + 900_000); + await vi.advanceTimersByTimeAsync(checkedAt.getTime() + 900_000 - Date.now()); await checked(); + }); + it("drains more than one due page without repeatedly checking completed rows", async () => { + for (let i = 0; i < 30; i++) await connection(); + const probedAt: number[] = [], performProbe = probe.getMockImplementation()!; + probe.mockImplementation(async (id: string) => { probedAt.push(Date.now()); return performProbe(id); }); + await start(); await checked(30); + expect(probedAt[25] - probedAt[24]).toBeLessThan(5_000); + expect(new Set(probe.mock.calls.map(([id]) => id)).size).toBe(30); + }); + it("checks the first connection created by the real API service after setup commits", async () => { + const requests = vi.fn(async (_url: unknown, init: RequestInit) => { + const body = JSON.parse(String(init.body)); + return Response.json({ jsonrpc: "2.0", id: body.id, result: { tools: [] } }); + }); + const tools = toolAccessService(db, { + remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }], remoteHttpRequest: requests, + }); + await start(tools.scheduledHealth); await quiet(); + const row = await tools.createConnection(companyId, { applicationId, name: "Real producer", transport: "mcp_remote", status: "active", enabled: true, + config: { url: "https://health.example.test/mcp" } }); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + const [stored] = await db.select().from(toolConnections).where(eq(toolConnections.id, row.id)); + expect(stored.healthStatus).toBe("ok"); expect(stored.healthCheckedAt).not.toBeNull(); + }, { timeout: 5000 }); + expect(requests).toHaveBeenCalledOnce(); + await tools.updateConnection(row.id, { enabled: false }); await quiet(); + await tools.updateConnection(row.id, { enabled: true }); + const [stored] = await db.select().from(toolConnections).where(eq(toolConnections.id, row.id)); + await idleAt(stored.healthCheckedAt!.getTime() + 900_000); + expect(requests).toHaveBeenCalledOnce(); + await tools.archiveConnection(row.id, companyId); await quiet(); + }); + + it("checks a quiet company in the first batch despite another company's unchecked backlog", async () => { + for (let index = 0; index < 30; index++) await connection({}, false); + const otherId = randomUUID(); + await db.insert(companies).values({ id: otherId, name: "Quiet health", issuePrefix: `Q${otherId.slice(0, 6)}` }); + const [app] = await db.insert(toolApplications).values({ companyId: otherId, name: "Quiet health", type: "mcp_http" }).returning(); + try { + const [other] = await db.insert(toolConnections).values({ + companyId: otherId, applicationId: app.id, name: "Quiet", uid: randomUUID(), transport: "mcp_remote", + status: "active", enabled: true, healthCheckedAt: new Date(Date.now() - 901_000), + }).returning(); + await start(); + expect(probe).toHaveBeenCalledTimes(25); + expect(probe.mock.calls.map(([id]) => id)).toContain(other.id); + } finally { + await coordinator?.stop(); coordinator = undefined; + await db.delete(companies).where(eq(companies.id, otherId)); + } + }); + + it("continues another company's checks when one scope cannot refresh", async () => { + const service = connectionHealthSchedule(db, probe), load = service.load; + await start(service); await quiet(); + service.load = async (...args) => { if (args[0] === companyId) throw new Error("scope unavailable"); return load(...args); }; + const otherId = randomUUID(); + await db.insert(companies).values({ id: otherId, name: "Other", issuePrefix: `O${otherId.slice(0, 6)}` }); + const [app] = await db.insert(toolApplications).values({ companyId: otherId, name: "Other health", type: "mcp_http" }).returning(); + try { + await connection(); + const [other] = await writeToolConnectionHealthChange(db, otherId, tx => tx.insert(toolConnections).values({ + companyId: otherId, applicationId: app.id, name: "Other", uid: randomUUID(), transport: "mcp_remote", status: "active", enabled: true, + }).returning()); + await checked(); expect(probe.mock.calls[0][0]).toBe(other.id); + } finally { service.load = load; await db.delete(companies).where(eq(companies.id, otherId)); errors.length = 0; } + }); + it("does not accelerate failed inventory reads while another company drains multiple pages", async () => { + const service = connectionHealthSchedule(db, probe), load = service.load; + await start(service); await quiet(); + const failedAt: number[] = []; + service.load = async (...args) => { + if (args[0] === companyId) { failedAt.push(Date.now()); throw new Error("scope unavailable"); } + return load(...args); + }; + const otherId = randomUUID(); + await db.insert(companies).values({ id: otherId, name: "Backlog", issuePrefix: `B${otherId.slice(0, 6)}` }); + const [app] = await db.insert(toolApplications).values({ companyId: otherId, name: "Backlog", type: "mcp_http" }).returning(); + try { + draining = true; + await connection(); + await writeToolConnectionHealthChange(db, otherId, tx => tx.insert(toolConnections).values( + Array.from({ length: 60 }, () => ({ companyId: otherId, applicationId: app.id, name: "Due", uid: randomUUID(), + transport: "mcp_remote" as const, status: "active" as const, enabled: true })), + )); + draining = false; + await checked(60); + expect(failedAt).toHaveLength(1); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(failedAt.length).toBeGreaterThan(1)); + expect(failedAt.every((at, index) => index === 0 || at - failedAt[index - 1] >= 30_000)).toBe(true); + service.load = load; + await vi.advanceTimersByTimeAsync(30_000); await checked(61); + await vi.advanceTimersByTimeAsync(31_000); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(180_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + } finally { + draining = false; service.load = load; await coordinator?.stop(); coordinator = undefined; + await db.delete(companies).where(eq(companies.id, otherId)); errors.length = 0; + } + }); + it("cancels and restores experimental connector checks through the settings producer", async () => { + const settings = instanceSettingsService(db); + const original = (await settings.getExperimental()).enablePaperclipNotebook; + try { + await settings.updateExperimental({ enablePaperclipNotebook: false }); + await connection({ config: { sourceTemplateKey: "paperclip-notebook" } }); + await start(); await quiet(); expect(probe).not.toHaveBeenCalled(); + await settings.updateExperimental({ enablePaperclipNotebook: true }); + await checked(); + await settings.updateExperimental({ enablePaperclipNotebook: false }); + await quiet(); + } finally { + await coordinator?.stop(); coordinator = undefined; + await settings.updateExperimental({ enablePaperclipNotebook: original }); + } + }); + it("cancels a company's cached deadline when the real deletion service commits", async () => { + const checkedAt = new Date(); await connection({ healthCheckedAt: checkedAt }); + await start(); await idleAt(checkedAt.getTime() + 900_000); + await deleteCompany(db, companyId); await quiet(); expect(probe).not.toHaveBeenCalled(); + }); + it("keeps SDK-only chat mutations quiet and wakes the existing voice obligation", async () => { + const voice = await connection({ transport: "voice", connectionPurpose: "channel", status: "draft", enabled: false }); + await start(); await quiet(); + await db.transaction(async tx => { await notifyChatConnectionHealth(tx, { companyId, connectionId: voice.id, provider: "slack" }); }); + await quiet(); + await db.transaction(async tx => { + await notifyChatConnectionHealth(tx, { companyId, connectionId: voice.id, provider: "speko" }); + await tx.update(toolConnections).set({ status: "active", enabled: true }).where(eq(toolConnections.id, voice.id)); + }); + await checked(); expect(probe.mock.calls[0][0]).toBe(voice.id); + }); + it("retains a producer wake arriving while its company is being refreshed", async () => { + const service = connectionHealthSchedule(db, probe), load = service.load; + await start(service); await quiet(); + let inserted = false; + service.load = async (...args) => { + const rows = await load(...args); + if (args[0] === companyId && !inserted) { inserted = true; await connection(); } + return rows; + }; + await connection(); await checked(2); + expect(new Set(probe.mock.calls.map(([id]) => id)).size).toBe(2); + }); + it("backs off an unpersisted check without blocking another connection and recovers", async () => { + const failed = await connection(), healthy = await connection(); + const persist = probe.getMockImplementation()!; + let broken = true; + probe.mockImplementation(async id => { if (id === failed.id && broken) throw new Error("health write unavailable"); return persist(id); }); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(probe.mock.calls.some(([id]) => id === healthy.id)).toBe(true); + expect(errors.length).toBeGreaterThan(0); + }); + const before = probe.mock.calls.length; + await vi.advanceTimersByTimeAsync(10_000); expect(probe).toHaveBeenCalledTimes(before); + broken = false; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + const [stored] = await db.select().from(toolConnections).where(eq(toolConnections.id, failed.id)); + expect(stored.healthCheckedAt).not.toBeNull(); + }, { timeout: 5000 }); + errors.length = 0; + }); + it("awaits an admitted health check during shutdown and stops future work", async () => { + await start(); await quiet(); + const persist = probe.getMockImplementation()!; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + probe.mockImplementation(async id => { await gate; return persist(id); }); + await connection(); await checked(); + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(); await stopping; coordinator = undefined; + await vi.advanceTimersByTimeAsync(1_800_000); expect(probe).toHaveBeenCalledOnce(); + }); + + it("warns for a persisted failed probe without retrying before its saved deadline", async () => { + const warnings = vi.spyOn(logger, "warn").mockImplementation(() => undefined); + const requests = vi.fn(async () => new Response("private-provider-response", { status: 503 })); + const tools = toolAccessService(db, { + remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }], remoteHttpRequest: requests, + }); + const row = await connection({ config: { url: "https://health.example.test/mcp" } }); + await start(tools.scheduledHealth); + const [stored] = await db.select().from(toolConnections).where(eq(toolConnections.id, row.id)); + expect(stored.healthCheckedAt).not.toBeNull(); + expect(warnings).toHaveBeenCalledWith( + { companyId, connectionId: row.id }, "Scheduled tool connection health check failed", + ); + expect(JSON.stringify(warnings.mock.calls)).not.toContain("private-provider-response"); + const due = stored.healthCheckedAt!.getTime() + 900_000; + await idleAt(due); + const before = requests.mock.calls.length; + await vi.advanceTimersByTimeAsync(180_000); + expect(requests).toHaveBeenCalledTimes(before); + expect(errors).toEqual([]); + }); + +}); diff --git a/server/src/index.ts b/server/src/index.ts index 17680e3d035..67da19a4503 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerToolConnectionHealthWork } from "./services/tool-connection-health-work.js"; import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "./services/approval-queue-seeding-work.js"; import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "./services/join-request-retention.js"; import { approvalRetentionService, registerApprovalRetentionWork } from "./services/approval-retention.js"; @@ -1838,12 +1839,8 @@ async function startServerWithDatabaseTeardown( await step("custom_image_setup_cleanup", () => registerCustomImageCleanupWork(deliveryWork, environmentCustomImages, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, config.heartbeatSchedulerIntervalMs).ready); - await step("tool_connection_health", async () => { - const toolHealthSweep = await tools.sweepConnectionHealth(); - if (toolHealthSweep.failed > 0) { - logger.warn({ ...toolHealthSweep }, "startup tool connection health sweep found failing connections"); - } - }); + await step("tool_connection_health", () => registerToolConnectionHealthWork(deliveryWork, tools.scheduledHealth, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("decision_continuation_worker", () => registerDecisionContinuationWork(deliveryWork, decisionExecutor, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("decision_expiry_worker", () => registerDecisionExpiryWork(deliveryWork, decisionExecutor, @@ -1931,18 +1928,6 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "status-card scheduler tick failed"); })); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(tools - .sweepConnectionHealth() - .then((swept) => { - if (swept.failed > 0) { - logger.warn({ ...swept }, "periodic tool connection health sweep found failing connections"); - } - }) - .catch((err) => { - logger.error({ err }, "periodic tool connection health sweep failed"); - })); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { // Periodically reap orphaned runs (5-min staleness threshold) and make sure diff --git a/server/src/services/access.ts b/server/src/services/access.ts index d7ae99b09a7..45935dcdc86 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { and, eq, inArray, ne, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -323,6 +324,7 @@ export function accessService(db: Db) { && grant.credentialSecretRefs.length > 0 && grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)), ); + if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, { companyId, id: connection.id }); await tx.update(toolConnections).set({ credentialRefs, credentialSecretRefs, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 4613ac43484..5cc26141bc4 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./chat-channels/connection-health-notifications.js"; import { credentialFingerprint } from "./chat-channels/credential-fingerprint.js"; import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; @@ -2295,6 +2296,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { updatedAt: new Date(), }) .where(eq(chatConversations.id, conversation!.id)); + await notifyChatConnectionHealth(taskTx, taskEndpoint); await taskTx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/configuration.ts b/server/src/services/chat-channels/configuration.ts index 73c1f84dbee..b7384cc0a4a 100644 --- a/server/src/services/chat-channels/configuration.ts +++ b/server/src/services/chat-channels/configuration.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import type { ConnectionManagementContext } from "../connection-management.js"; import { CHAT_HUMAN_CREDENTIAL_FIELDS } from "./credentials.js"; import { preparePhotonConfiguration, createPhotonConfiguration } from "./imessage-photon/setup.js"; @@ -207,6 +208,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption updatedAt: pausedAt, }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ status: "disabled", enabled: false, updatedAt: new Date() }) @@ -351,6 +353,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption }) .where(eq(chatEndpoints.id, endpoint.id)) .returning(); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ @@ -396,6 +399,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption sql`coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0) = ${runtimeGeneration(activatedEndpoint.setup)}`, ), ); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ @@ -500,6 +504,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption .update(chatConversations) .set({ state: "endpoint_removed", updatedAt: new Date() }) .where(eq(chatConversations.endpointId, endpoint.id)); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ status: "archived", enabled: false, updatedAt: new Date() }) @@ -829,6 +834,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption updatedAt, }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ @@ -891,6 +897,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption updatedAt: new Date(), }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/connection-health-notifications.ts b/server/src/services/chat-channels/connection-health-notifications.ts new file mode 100644 index 00000000000..a29302ff3b8 --- /dev/null +++ b/server/src/services/chat-channels/connection-health-notifications.ts @@ -0,0 +1,8 @@ +import type { ChatProvider } from "@paperclipai/shared"; +import { notifyToolConnectionHealth } from "../tool-connection-health-notifications.js"; +import { chatConnectionTransport } from "./provider-catalog.js"; + +export async function notifyChatConnectionHealth(transaction: object, endpoint: { companyId: string; connectionId: string; provider: ChatProvider }) { + if (chatConnectionTransport(endpoint.provider) === "chat_sdk") return; + await notifyToolConnectionHealth(transaction, { companyId: endpoint.companyId, id: endpoint.connectionId }); +} diff --git a/server/src/services/chat-channels/endpoints.ts b/server/src/services/chat-channels/endpoints.ts index 4d09b7b5d85..581cc356f92 100644 --- a/server/src/services/chat-channels/endpoints.ts +++ b/server/src/services/chat-channels/endpoints.ts @@ -1,3 +1,4 @@ +import { chatConnectionTransport } from "./provider-catalog.js"; import type { ConnectionManagementContext } from "../connection-management.js"; import { slackSetupState } from "./slack/setup.js"; import { assertChatExecutionDefaultsAccess } from "../chat-execution-defaults.js"; @@ -333,7 +334,7 @@ function providerSetupState( connectionKind: "managed", connectionPurpose: "channel", ownership: "customer", - transport: input.provider === "speko" ? "voice" : "chat_sdk", + transport: chatConnectionTransport(input.provider), authKind: "api_key", credentialPolicy: "shared", status: "draft", diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index e9580a762b0..8fe0f601e36 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { slackQuestionFormValidationError } from "../slack/question-forms.js"; import { chatInteractionForm, chatItemVerdicts } from "../../chat-interaction-forms.js"; import { createDiscordQuestionForms } from "../discord/question-forms.js"; @@ -426,6 +427,7 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb updatedAt: processedAt, }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/provider-catalog.ts b/server/src/services/chat-channels/provider-catalog.ts index 5beb7879400..2dc6e819180 100644 --- a/server/src/services/chat-channels/provider-catalog.ts +++ b/server/src/services/chat-channels/provider-catalog.ts @@ -106,3 +106,9 @@ const CAPABILITIES: Record = { }, }; export { PROVIDER_LABELS, CAPABILITIES }; + +export function chatConnectionTransport(provider: ChatProvider): "voice" | "rest_api" | "chat_sdk" { + if (provider === "speko") return "voice"; + if (provider === "agentmail") return "rest_api"; + return "chat_sdk"; +} diff --git a/server/src/services/chat-channels/provider-effects.ts b/server/src/services/chat-channels/provider-effects.ts index 5c663aaeb9f..4b15a1b769d 100644 --- a/server/src/services/chat-channels/provider-effects.ts +++ b/server/src/services/chat-channels/provider-effects.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { createTelegramCallbackNoticeAuthority } from "./telegram/callback-notices.js"; import { slackThreadChannelId } from "./slack/messages.js"; @@ -325,6 +326,7 @@ export function createProviderEffects(db: Db, options: ProviderEffectsOptions) { updatedAt: new Date(), }) .where(eq(chatEndpoints.id, currentEndpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index 1f701de8cb8..84184bb8633 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; import type { EndpointRow, DbTransaction, LifecycleRuntimeFence } from "../runtime-types.js"; @@ -197,6 +198,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur updatedAt: new Date(), }) .where(eq(chatEndpoints.id, currentEndpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index 0869aff07e4..a706452196e 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import type { ConnectionManagementContext } from "../connection-management.js"; import { imessageResourceParticipants } from "./imessage-photon/resources.js"; import { createGitHubResources, reconcileGitHubRepositoryCoordinates } from "./github/resources.js"; @@ -523,6 +524,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc updatedAt: now, }) .where(eq(chatEndpoints.id, currentEndpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ @@ -552,6 +554,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc updatedAt: now, }) .where(eq(chatEndpoints.id, currentEndpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/setup-probes.ts b/server/src/services/chat-channels/setup-probes.ts index 74aec70c43e..784bf2d38bf 100644 --- a/server/src/services/chat-channels/setup-probes.ts +++ b/server/src/services/chat-channels/setup-probes.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import type { ConnectionManagementContext } from "../connection-management.js"; import type { EndpointRow, DbOrTransaction } from "./runtime-types.js"; @@ -329,6 +330,7 @@ export function createSetupProbes(db: Db, options: SetupProbesOptions) { code: "chat_endpoint_test_superseded", }); } + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index ba9d8f758ac..5a78ef30208 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -119,6 +120,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(executionWorkspaces).where(eq(executionWorkspaces.companyId, id)); await tx.delete(projects).where(eq(projects.companyId, id)); await agentLifecycleCompanyDeletion.deleteCompanyData(tx, id); + await notifyToolConnectionHealth(tx, id); const rows = await tx .delete(companies) .where(eq(companies.id, id)) diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index f4f3a3f613c..17413f951aa 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -18,7 +18,7 @@ export const DELIVERY_QUEUES = { decisionContinuation: "decision-continuation", decisionArchiveDelivery: "decision-archive-delivery", - + toolConnectionHealth: "tool-connection-health", toolReviewProjection: "tool-review-projection", toolReviewInterruption: "tool-review-interruption", toolReviewRecovery: "tool-review-recovery", diff --git a/server/src/services/email-channels.ts b/server/src/services/email-channels.ts index 782a53f10d1..b2a05c10b81 100644 --- a/server/src/services/email-channels.ts +++ b/server/src/services/email-channels.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import type { ConnectionManagementContext } from "./connection-management.js"; import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -923,6 +924,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { updatedAt: now, }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyToolConnectionHealth(tx, { companyId: endpoint.companyId, id: endpoint.connectionId }); await tx .update(toolConnections) .set({ status: "active", enabled: true, healthStatus: "ok" }) @@ -1994,7 +1996,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { .update(chatEndpoints) .set({ lastError: null, healthMessage: "Connected" }) .where(eq(chatEndpoints.id, endpoint.id)); - await db + await writeToolConnectionHealthChange(db, { companyId: endpoint.companyId, id: endpoint.connectionId }, tx => tx .update(toolConnections) .set({ healthStatus: "ok", @@ -2002,7 +2004,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { lastError: null, healthCheckedAt: new Date(), }) - .where(eq(toolConnections.id, endpoint.connectionId)); + .where(eq(toolConnections.id, endpoint.connectionId))); } } async function maintainSocket(endpoint: Endpoint) { @@ -2102,7 +2104,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { }); } async function markError(endpoint: Endpoint, error: string) { - await db + await writeToolConnectionHealthChange(db, { companyId: endpoint.companyId, id: endpoint.connectionId }, tx => tx .update(toolConnections) .set({ healthStatus: "degraded", @@ -2110,7 +2112,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { lastError: error, healthCheckedAt: new Date(), }) - .where(eq(toolConnections.id, endpoint.connectionId)); + .where(eq(toolConnections.id, endpoint.connectionId))); await db .update(chatEndpoints) .set({ lastError: error, healthMessage: error, updatedAt: new Date() }) @@ -2372,6 +2374,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { .update(chatEndpoints) .set({ status: "paused" }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyToolConnectionHealth(tx, { companyId: endpoint.companyId, id: endpoint.connectionId }); await tx .update(toolConnections) .set({ enabled: false, status: "disabled" }) @@ -2467,6 +2470,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { updatedAt: new Date(), }) .where(eq(chatEndpoints.id, id)); + await notifyToolConnectionHealth(tx, { companyId: endpoint.companyId, id: endpoint.connectionId }); await tx .update(toolConnections) .set({ diff --git a/server/src/services/email-connections.ts b/server/src/services/email-connections.ts index 2e9676d618a..dc8d4beb8a8 100644 --- a/server/src/services/email-connections.ts +++ b/server/src/services/email-connections.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, isNull, ne, sql } from "drizzle-orm"; import { @@ -268,7 +269,7 @@ export function emailConnectionService( actorSource: actor.localImplicit ? "local_implicit" : "session", }, ); - await db + await writeToolConnectionHealthChange(db, connection, tx => tx .update(toolConnections) .set({ uid: `agentmail-account-${input.idempotencyKey}`, @@ -276,7 +277,7 @@ export function emailConnectionService( healthMessage: "Connected", healthCheckedAt: new Date(), }) - .where(eq(toolConnections.id, connection.id)); + .where(eq(toolConnections.id, connection.id))); if (input.grantKind === "user") await db .update(connectionGrants) diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 6848875815f..47bdac122a2 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js"; import { chatEndpoints } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; @@ -328,6 +329,7 @@ export function githubConnectionEventService( }; await database.update(connectionGrants).set({ providerTenant, updatedAt: now() }) .where(and(eq(connectionGrants.id, binding.grantId), eq(connectionGrants.companyId, binding.companyId))); + await notifyToolConnectionHealth(database, { companyId: binding.companyId, id: binding.connectionId }); await database.update(toolConnections).set({ healthStatus: unavailable ? "failed" : "ok", healthMessage: unavailable diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index d13920e8b9f..7808541872a 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import type { Db } from "@paperclipai/db"; import { companies, instanceSettings } from "@paperclipai/db"; @@ -613,6 +614,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); + await notifyToolConnectionHealth(tx, "*"); return tx .update(instanceSettings) .set({ diff --git a/server/src/services/smoke-lab.ts b/server/src/services/smoke-lab.ts index 931021f54dc..0b0981a18be 100644 --- a/server/src/services/smoke-lab.ts +++ b/server/src/services/smoke-lab.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { spawn, type ChildProcess } from "node:child_process"; import { createHash, randomUUID } from "node:crypto"; import { existsSync } from "node:fs"; @@ -690,14 +691,20 @@ export function smokeLabService(db: Db, options: { )); const now = new Date(); if (existing) { - const [updated] = await db.update(toolApplications).set({ - name: input.name, - description: input.description, - type: input.type, - status: "active", - metadata: { smokeLab: true }, - updatedAt: now, - }).where(eq(toolApplications.id, existing.id)).returning(); + const [updated] = await db.transaction(async tx => { + const [current] = await tx.select({ type: toolApplications.type }).from(toolApplications) + .where(and(eq(toolApplications.id, existing.id), eq(toolApplications.companyId, input.companyId))) + .for("no key update"); + if (current && current.type !== input.type) await notifyToolConnectionHealth(tx, { companyId: input.companyId, applicationId: existing.id }); + return tx.update(toolApplications).set({ + name: input.name, + description: input.description, + type: input.type, + status: "active", + metadata: { smokeLab: true }, + updatedAt: now, + }).where(eq(toolApplications.id, existing.id)).returning(); + }); return { row: updated ?? existing, created: false }; } const [created] = await db.insert(toolApplications).values({ @@ -742,41 +749,45 @@ export function smokeLabService(db: Db, options: { lastHealthAt: now, updatedAt: now, }; - const ensureDefaultOrganizationGrant = async (connectionId: string) => { - const [existingGrant] = await db.select({ id: connectionGrants.id }).from(connectionGrants).where(and( - eq(connectionGrants.companyId, input.companyId), - eq(connectionGrants.connectionId, connectionId), - eq(connectionGrants.kind, "organization"), - eq(connectionGrants.isDefault, true), - )); - if (existingGrant) return; - await db.insert(connectionGrants).values({ + const connectionId = existing?.id ?? randomUUID(); + return writeToolConnectionHealthChange(db, { companyId: input.companyId, id: connectionId }, async tx => { + const ensureDefaultOrganizationGrant = async (connectionId: string) => { + const [existingGrant] = await tx.select({ id: connectionGrants.id }).from(connectionGrants).where(and( + eq(connectionGrants.companyId, input.companyId), + eq(connectionGrants.connectionId, connectionId), + eq(connectionGrants.kind, "organization"), + eq(connectionGrants.isDefault, true), + )); + if (existingGrant) return; + await tx.insert(connectionGrants).values({ + companyId: input.companyId, + connectionId, + kind: "organization", + status: "active", + isDefault: true, + credentialSecretRefs: [], + createdAt: now, + updatedAt: now, + }); + }; + if (existing) { + const [updated] = await tx.update(toolConnections).set(values).where(eq(toolConnections.id, existing.id)).returning(); + await ensureDefaultOrganizationGrant(existing.id); + return { row: updated ?? existing, created: false }; + } + const [created] = await tx.insert(toolConnections).values({ + id: connectionId, companyId: input.companyId, - connectionId, - kind: "organization", - status: "active", - isDefault: true, - credentialSecretRefs: [], + name: input.name, + uid: `smoke-lab/${input.name.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "")}`, + ...values, + createdByAgentId: input.actor?.actorType === "agent" ? input.actor.agentId : null, + createdByUserId: input.actor?.actorType === "user" ? input.actor.actorId : null, createdAt: now, - updatedAt: now, - }); - }; - if (existing) { - const [updated] = await db.update(toolConnections).set(values).where(eq(toolConnections.id, existing.id)).returning(); - await ensureDefaultOrganizationGrant(existing.id); - return { row: updated ?? existing, created: false }; - } - const [created] = await db.insert(toolConnections).values({ - companyId: input.companyId, - name: input.name, - uid: `smoke-lab/${input.name.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "")}`, - ...values, - createdByAgentId: input.actor?.actorType === "agent" ? input.actor.agentId : null, - createdByUserId: input.actor?.actorType === "user" ? input.actor.actorId : null, - createdAt: now, - }).returning(); - await ensureDefaultOrganizationGrant(created.id); - return { row: created, created: true }; + }).returning(); + await ensureDefaultOrganizationGrant(created.id); + return { row: created, created: true }; + }); } async function ensureStdioTemplate(companyId: string, actor?: SmokeLabActorInfo) { @@ -963,20 +974,21 @@ export function smokeLabService(db: Db, options: { eq(toolConnections.name, HTTP_CONNECTION_NAME), )); if (!connection || !httpSidecar) return; + const sidecarUrl = httpSidecar.url; const now = new Date(); - await db.update(toolConnections).set({ + await writeToolConnectionHealthChange(db, connection, tx => tx.update(toolConnections).set({ config: { ...asRecord(connection.config), - url: `${httpSidecar.url}/mcp`, - catalogUrl: `${httpSidecar.url}/catalog`, - toolCallUrl: `${httpSidecar.url}/tools/call`, + url: `${sidecarUrl}/mcp`, + catalogUrl: `${sidecarUrl}/catalog`, + toolCallUrl: `${sidecarUrl}/tools/call`, }, healthStatus: "ok", healthMessage: "Smoke Lab HTTP fixture sidecar is running.", healthCheckedAt: now, lastHealthAt: now, updatedAt: now, - }).where(eq(toolConnections.id, connection.id)); + }).where(eq(toolConnections.id, connection.id))); } async function startHttpSidecar(companyId?: string) { diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 6cb2220af72..8c2a82fb80c 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -1,3 +1,5 @@ +import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; +import { connectionHealthSchedule } from "./tool-access/catalog/health-schedule.js"; import type { ConnectionManagementContext } from "./connection-management.js"; import { connectionActorUserId } from "./tool-access/actor.js"; import { restConnectorFor } from "./connectors/rest-registry.js"; @@ -1219,8 +1221,12 @@ export function toolAccessService( safeDefault: true, }; if (existing) { - const [updated] = await db - .update(toolApplications) + const [updated] = await db.transaction(async tx => { + const [current] = await tx.select({ type: toolApplications.type }).from(toolApplications) + .where(and(eq(toolApplications.id, existing.id), eq(toolApplications.companyId, companyId))) + .for("no key update"); + if (current && current.type !== "mcp_stdio") await notifyToolConnectionHealth(tx, { companyId, applicationId: existing.id }); + return tx.update(toolApplications) .set({ name: definition.applicationName, description: definition.applicationDescription, @@ -1232,6 +1238,7 @@ export function toolAccessService( }) .where(eq(toolApplications.id, existing.id)) .returning(); + }); return { row: updated, created: false }; } const [created] = await db @@ -1262,11 +1269,46 @@ export function toolAccessService( quarantineNewEntries: true, }; if (existing) { - const [updated] = await db - .update(toolConnections) - .set({ + const updated = await writeToolConnectionHealthChange(db, existing, async tx => { + const [updated] = await tx + .update(toolConnections) + .set({ + applicationId, + name: definition.connectionName, + transport: "local_stdio", + status: "active", + enabled: true, + config, + transportConfig: config, + credentialRefs: [], + credentialSecretRefs: [], + updatedAt: new Date(), + }) + .where(eq(toolConnections.id, existing.id)) + .returning(); + await ensureDefaultOrganizationGrant(updated, tx); + await syncCredentialBindings(updated, [], tx); + await createToolRuntimeSlots(tx, runtimeSupervisor).ensureRuntimeSlot(updated); + return updated; + }); + emitConnectionUpdated(updated, existing, "example"); + return { row: updated, created: false }; + } + const connectionId = randomUUID(); + const created = await writeToolConnectionHealthChange(db, { companyId, id: connectionId }, async tx => { + const [created] = await tx + .insert(toolConnections) + .values({ + id: connectionId, + companyId, applicationId, name: definition.connectionName, + uid: connectionUid( + "paperclip", + definition.connectionName, + connectionId, + ), + connectionKind: "managed", transport: "local_stdio", status: "active", enabled: true, @@ -1274,42 +1316,13 @@ export function toolAccessService( transportConfig: config, credentialRefs: [], credentialSecretRefs: [], - updatedAt: new Date(), }) - .where(eq(toolConnections.id, existing.id)) .returning(); - await ensureDefaultOrganizationGrant(updated); - await syncCredentialBindings(updated); - await ensureRuntimeSlot(updated); - emitConnectionUpdated(updated, existing, "example"); - return { row: updated, created: false }; - } - const connectionId = randomUUID(); - const [created] = await db - .insert(toolConnections) - .values({ - id: connectionId, - companyId, - applicationId, - name: definition.connectionName, - uid: connectionUid( - "paperclip", - definition.connectionName, - connectionId, - ), - connectionKind: "managed", - transport: "local_stdio", - status: "active", - enabled: true, - config, - transportConfig: config, - credentialRefs: [], - credentialSecretRefs: [], - }) - .returning(); - await ensureDefaultOrganizationGrant(created); - await syncCredentialBindings(created); - await ensureRuntimeSlot(created); + await ensureDefaultOrganizationGrant(created, tx); + await syncCredentialBindings(created, [], tx); + await createToolRuntimeSlots(tx, runtimeSupervisor).ensureRuntimeSlot(created); + return created; + }); emitConnectionCreated(created, "example"); return { row: created, created: true }; } @@ -2475,7 +2488,7 @@ export function toolAccessService( personalIdentityUserId || dedicatedAgentId ? [] : credentialSecretRefs; if (revivedConnectionPrevious) { forgetMcpHttpSessions(revivedConnectionPrevious.id); - [connectionRow] = await db + [connectionRow] = await writeToolConnectionHealthChange(db, revivedConnectionPrevious, tx => tx .update(toolConnections) .set({ name, @@ -2497,7 +2510,7 @@ export function toolAccessService( updatedAt: new Date(), }) .where(eq(toolConnections.id, revivedConnectionPrevious.id)) - .returning(); + .returning()); } else { const connectionId = randomUUID(); [connectionRow] = await db @@ -2920,6 +2933,7 @@ export function toolAccessService( connectionSetupMutationFingerprint(attemptedConnection), ); if (connectionMutationIsStillCurrent) { + await notifyToolConnectionHealth(tx, revivedConnectionPrevious); await tx .update(toolConnections) .set({ @@ -2995,7 +3009,7 @@ export function toolAccessService( // surface the restoration failure instead of returning only the // original provider error. try { - await db + await writeToolConnectionHealthChange(db, revivedConnectionPrevious, tx => tx .update(toolConnections) .set({ status: "draft", @@ -3011,7 +3025,7 @@ export function toolAccessService( eq(toolConnections.id, revivedConnectionPrevious.id), eq(toolConnections.companyId, companyId), ), - ); + )); } catch (quarantineError) { identityRollbackError = new AggregateError( [rollbackError, quarantineError], @@ -3020,9 +3034,10 @@ export function toolAccessService( } } } else if (connectionRow) { - await db + const failedConnectionId = connectionRow.id; + await writeToolConnectionHealthChange(db, { companyId, id: failedConnectionId }, tx => tx .delete(toolConnections) - .where(eq(toolConnections.id, connectionRow.id)) + .where(eq(toolConnections.id, failedConnectionId))) .catch(() => undefined); } if ( @@ -3136,6 +3151,7 @@ export function toolAccessService( await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, companyId))) .for("update"); + await notifyToolConnectionHealth(tx, connection); const [row] = await tx.update(toolConnections).set({ config, transportConfig: config, status: "active", enabled: true, ...(input.agentInstructions !== undefined ? { agentInstructions: input.agentInstructions } : {}), updatedAt: new Date() }) .where(and(eq(toolConnections.id, connection.id), eq(toolConnections.companyId, companyId))).returning(); await tx.update(toolApplications).set({ status: "active", updatedAt: new Date() }).where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, companyId))); @@ -3493,6 +3509,7 @@ export function toolAccessService( }, tx, ); + await notifyToolConnectionHealth(tx, connection); const [updatedConnection] = await tx .update(toolConnections) .set({ status: "active", enabled: true, ...(input.agentInstructions !== undefined ? { agentInstructions: input.agentInstructions } : {}), updatedAt: new Date() }) @@ -4525,40 +4542,43 @@ export function toolAccessService( await actor?.managementContext?.reauthorize(); const connectionId = randomUUID(); const binding = actorBinding(actor); - const [row] = await db - .insert(toolConnections) - .values({ - id: connectionId, - companyId, - applicationId, - name: input.name, - uid: connectionUid(applicationNamespace, input.name, connectionId), - connectionKind: input.connectionKind ?? "managed", - ownership: input.ownership ?? "customer", - transport, - authKind: input.authKind ?? "none", - credentialPolicy: - input.credentialPolicy ?? - (input.authKind === "oauth" ? "per_user" : "shared"), - status: input.status ?? "draft", - enabled: input.enabled ?? false, - agentInstructions: input.agentInstructions !== undefined ? input.agentInstructions : defaultConnectionAgentInstructions( - typeof config.sourceTemplateKey === "string" ? getConnectableAppDefinition(config.sourceTemplateKey)?.agentInstructions : null, - ), - config, - transportConfig: isGoogleSheetsConnectionConfig(config) - ? config - : (input.transportConfig ?? config), - credentialRefs: input.credentialRefs ?? [], - credentialSecretRefs: input.credentialSecretRefs ?? [], - createdByAgentId: - binding.actorType === "agent" ? binding.actorId : null, - createdByUserId: connectionActorUserId(actor), - }) - .returning(); - await ensureDefaultOrganizationGrant(row); - await syncCredentialBindings(row); - await ensureRuntimeSlot(row); + const row = await writeToolConnectionHealthChange(db, { companyId, id: connectionId }, async tx => { + const [row] = await tx + .insert(toolConnections) + .values({ + id: connectionId, + companyId, + applicationId, + name: input.name, + uid: connectionUid(applicationNamespace, input.name, connectionId), + connectionKind: input.connectionKind ?? "managed", + ownership: input.ownership ?? "customer", + transport, + authKind: input.authKind ?? "none", + credentialPolicy: + input.credentialPolicy ?? + (input.authKind === "oauth" ? "per_user" : "shared"), + status: input.status ?? "draft", + enabled: input.enabled ?? false, + agentInstructions: input.agentInstructions !== undefined ? input.agentInstructions : defaultConnectionAgentInstructions( + typeof config.sourceTemplateKey === "string" ? getConnectableAppDefinition(config.sourceTemplateKey)?.agentInstructions : null, + ), + config, + transportConfig: isGoogleSheetsConnectionConfig(config) + ? config + : (input.transportConfig ?? config), + credentialRefs: input.credentialRefs ?? [], + credentialSecretRefs: input.credentialSecretRefs ?? [], + createdByAgentId: + binding.actorType === "agent" ? binding.actorId : null, + createdByUserId: connectionActorUserId(actor), + }) + .returning(); + await ensureDefaultOrganizationGrant(row, tx); + await syncCredentialBindings(row, [], tx); + await createToolRuntimeSlots(tx, runtimeSupervisor).ensureRuntimeSlot(row); + return row; + }); emitConnectionCreated(row, "api"); return toConnection(row); }, @@ -4651,9 +4671,12 @@ export function toolAccessService( await syncCredentialBindings(rows[0]!, [], tx); return rows; }) - : await write(db); - if (existing.transport !== "chat_sdk") await syncCredentialBindings(row); - await ensureRuntimeSlot(row); + : await writeToolConnectionHealthChange(db, existing, async tx => { + const rows = await write(tx); + await syncCredentialBindings(rows[0]!, [], tx); + await createToolRuntimeSlots(tx, runtimeSupervisor).ensureRuntimeSlot(rows[0]!); + return rows; + }); emitConnectionUpdated(row, existing, "api"); return toConnection(row); }, @@ -4704,6 +4727,7 @@ export function toolAccessService( listAppsNeedingAttention, sweepConnectionHealth, + scheduledHealth: connectionHealthSchedule(db, checkConnectionHealth), sweepGitHubConnectionContinuity, ...catalogReader, diff --git a/server/src/services/tool-access/catalog/health-schedule.ts b/server/src/services/tool-access/catalog/health-schedule.ts new file mode 100644 index 00000000000..130e0d0ee53 --- /dev/null +++ b/server/src/services/tool-access/catalog/health-schedule.ts @@ -0,0 +1,62 @@ +import { logger } from "../../../middleware/logger.js"; +import { and, asc, eq, gt, ne, sql } from "drizzle-orm"; +import { toolApplications, toolConnections, type Db } from "@paperclipai/db"; +import { isConnectorFeatureEnabled } from "@paperclipai/shared"; +import { instanceSettingsService } from "../../instance-settings.js"; +import { connectionSetupApp } from "../connections/setup-policy.js"; +import type { ConnectionHealth } from "./health.js"; + +export type ConnectionHealthDeadline = { id: string; companyId: string; applicationId: string; dueAt: number }; +const HEALTH_FRESHNESS_MS = 15 * 60_000; + +export function connectionHealthSchedule(db: Db, checkHealth: ConnectionHealth["checkConnectionHealth"]) { + async function load(companyId?: string, connectionId?: string, applicationId?: string, signal?: AbortSignal): Promise { + const deadlines: ConnectionHealthDeadline[] = []; + let cursor: string | undefined; + while (true) { + signal?.throwIfAborted(); + const rows = await db.select({ + id: toolConnections.id, + companyId: toolConnections.companyId, + applicationId: toolConnections.applicationId, + checkedAt: toolConnections.healthCheckedAt, + applicationKey: toolApplications.applicationKey, + sourceTemplateKey: sql`coalesce(${toolConnections.config}->>'sourceTemplateKey', ${toolConnections.transportConfig}->>'sourceTemplateKey')`, + }).from(toolConnections).innerJoin(toolApplications, and( + eq(toolApplications.id, toolConnections.applicationId), + eq(toolApplications.companyId, toolConnections.companyId), + )).where(and( + eq(toolConnections.enabled, true), eq(toolConnections.status, "active"), + ne(toolConnections.transport, "chat_sdk"), ne(toolConnections.transport, "runtime_auth"), + ne(toolApplications.type, "paperclip_plugin"), + companyId === undefined ? undefined : eq(toolConnections.companyId, companyId), + connectionId === undefined ? undefined : eq(toolConnections.id, connectionId), + applicationId === undefined ? undefined : eq(toolConnections.applicationId, applicationId), + cursor === undefined ? undefined : gt(toolConnections.id, cursor), + )).orderBy(asc(toolConnections.id)).limit(100); + const definitions = rows.map(row => ({ row, app: connectionSetupApp(row.applicationKey, row.sourceTemplateKey) })); + const settings = definitions.some(({ app }) => app?.experimentalFeatureKey) + ? await instanceSettingsService(db).getExperimental() : null; + deadlines.push(...definitions.filter(({ app }) => !app || !settings || isConnectorFeatureEnabled(app, settings)) + .map(({ row }) => ({ id: row.id, companyId: row.companyId, applicationId: row.applicationId, + dueAt: row.checkedAt === null ? 0 : row.checkedAt.getTime() + HEALTH_FRESHNESS_MS }))); + if (rows.length < 100) return deadlines; + cursor = rows[rows.length - 1].id; + } + } + + async function check(id: string): Promise { + const [before] = await load(undefined, id); + if (!before || before.dueAt > Date.now()) return before ?? null; + let failure: unknown; + try { await checkHealth(id, { actorType: "system", actorId: "tool_health_sweep" }); } + catch (error) { + failure = error; + logger.warn({ companyId: before.companyId, connectionId: id }, "Scheduled tool connection health check failed"); + } + const [after] = await load(before.companyId, id); + if (failure && after && after.dueAt <= Date.now()) throw failure; + return after ?? null; + } + return { load, check }; +} diff --git a/server/src/services/tool-access/catalog/health.ts b/server/src/services/tool-access/catalog/health.ts index 3a0eb124240..f22291cd81c 100644 --- a/server/src/services/tool-access/catalog/health.ts +++ b/server/src/services/tool-access/catalog/health.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectionActorUserId } from "../actor.js"; import { assertSupportedConnection } from "../connections/setup-policy.js"; import { connectionSetupUrl, connectionReconnectUrl } from "../identifiers.js"; @@ -47,7 +48,7 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d message: string | null, ) { const now = new Date(); - const [updated] = await db + const [updated] = await writeToolConnectionHealthChange(db, connection, tx => tx .update(toolConnections) .set({ healthStatus: status, @@ -58,7 +59,7 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d updatedAt: now, }) .where(eq(toolConnections.id, connection.id)) - .returning(); + .returning()); if (connection.transport === "local_stdio") { await db .update(toolRuntimeSlots) diff --git a/server/src/services/tool-access/catalog/publication.ts b/server/src/services/tool-access/catalog/publication.ts index 84459758207..cfb25f98386 100644 --- a/server/src/services/tool-access/catalog/publication.ts +++ b/server/src/services/tool-access/catalog/publication.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectorToolPolicy } from "../../connectors/tool-policy.js"; import { connectionSetupUrl, connectionReconnectUrl } from "../identifiers.js"; import { createToolRuntimeSlots } from "../runtime/slots.js"; @@ -269,7 +270,7 @@ export function createToolCatalogPublication(db: Db, { now, audit, getConnection : refreshOptions.enableAllByDefault ? { ...connection.transportConfig, quarantineNewEntries: false } : connection.transportConfig; - const [updatedConnection] = await db + const [updatedConnection] = await writeToolConnectionHealthChange(db, connection, tx => tx .update(toolConnections) .set({ config: normalizedConfig, @@ -287,7 +288,7 @@ export function createToolCatalogPublication(db: Db, { now, audit, getConnection updatedAt: refreshedAt, }) .where(eq(toolConnections.id, connection.id)) - .returning(); + .returning()); if (connection.transport === "local_stdio") { await ensureRuntimeSlot(updatedConnection); diff --git a/server/src/services/tool-access/connections/removal.ts b/server/src/services/tool-access/connections/removal.ts index 6679be5ced7..3a8b7073275 100644 --- a/server/src/services/tool-access/connections/removal.ts +++ b/server/src/services/tool-access/connections/removal.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { toConnection } from "./projection.js"; import { actorBinding } from "../actor.js"; import { vercelConnectResourcesFor } from "../vercel-credentials.js"; @@ -538,6 +539,7 @@ async function removeConnection( .returning({ id: connectionTokenIssuances.id }); const archived = await db.transaction(async (tx) => { + await notifyToolConnectionHealth(tx, connection); await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId))) .for("update"); diff --git a/server/src/services/tool-access/connections/setup-policy.ts b/server/src/services/tool-access/connections/setup-policy.ts index cf07e847b2f..8c4104d1a23 100644 --- a/server/src/services/tool-access/connections/setup-policy.ts +++ b/server/src/services/tool-access/connections/setup-policy.ts @@ -14,6 +14,11 @@ export function assertSupportedConnection(connection: typeof toolConnections.$in } } +export function connectionSetupApp(applicationKey: unknown, sourceTemplateKey: unknown) { + return applicationKey != null ? getAppDefinitionForApplicationKey(applicationKey) + : typeof sourceTemplateKey === "string" ? getConnectableAppDefinition(sourceTemplateKey) : null; +} + export function createConnectionSetupPolicy(db: Db) { async function assertExperimentalConnectorSetupEnabled(provider: unknown, existing = false) { const app = typeof provider === "string" ? getConnectableAppDefinition(provider) : null; @@ -36,8 +41,7 @@ export function createConnectionSetupPolicy(db: Db) { resolvedApplicationKey = application?.applicationKey; } const marker = connection.config?.sourceTemplateKey ?? connection.transportConfig?.sourceTemplateKey; - const app = resolvedApplicationKey != null ? getAppDefinitionForApplicationKey(resolvedApplicationKey) - : typeof marker === "string" ? getConnectableAppDefinition(marker) : null; + const app = connectionSetupApp(resolvedApplicationKey, marker); if (app) await assertExperimentalConnectorSetupEnabled(app.slug, connection.status !== "draft"); } return { assertExperimentalConnectorSetupEnabled, assertConnectorConnectionSetupEnabled }; diff --git a/server/src/services/tool-access/oauth/access-finalization.ts b/server/src/services/tool-access/oauth/access-finalization.ts index 2b5d2265a6a..a6ad361dde8 100644 --- a/server/src/services/tool-access/oauth/access-finalization.ts +++ b/server/src/services/tool-access/oauth/access-finalization.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectionActorUserId } from "../actor.js"; import type { ActorInfo } from "../types.js"; import { connectionMethodForConnection } from "../connection-methods.js"; @@ -114,33 +115,37 @@ async function finalizeOAuthAccess( ) { throw conflict("This connection already uses a company identity"); } - [connection] = await db - .update(toolConnections) - .set({ - credentialPolicy: "per_user", - credentialRefs: connection.credentialRefs.filter( - (ref) => ref.name !== "oauth.access_token", - ), - credentialSecretRefs: connection.credentialSecretRefs.filter( - (ref) => - ref.configPath !== "oauth.access_token" && - ref.configPath !== "oauth.refresh_token", - ), - status: "active", - enabled: true, - updatedAt: new Date(), - }) - .where( - and( - eq(toolConnections.id, connection.id), - eq(toolConnections.companyId, companyId), - ), - ) - .returning(); - await syncCredentialBindings( - connection, - personalGrant.credentialSecretRefs, - ); + connection = await writeToolConnectionHealthChange(db, connection, async tx => { + const [updated] = await tx + .update(toolConnections) + .set({ + credentialPolicy: "per_user", + credentialRefs: connection.credentialRefs.filter( + (ref) => ref.name !== "oauth.access_token", + ), + credentialSecretRefs: connection.credentialSecretRefs.filter( + (ref) => + ref.configPath !== "oauth.access_token" && + ref.configPath !== "oauth.refresh_token", + ), + status: "active", + enabled: true, + updatedAt: new Date(), + }) + .where( + and( + eq(toolConnections.id, connection.id), + eq(toolConnections.companyId, companyId), + ), + ) + .returning(); + await syncCredentialBindings( + updated, + personalGrant.credentialSecretRefs, + tx, + ); + return updated; + }); } else if ( connection.credentialPolicy !== "shared" || connection.credentialSecretRefs.length === 0 @@ -325,6 +330,7 @@ async function finalizeOAuthAccess( updatedAt: nowAt, }) .where(eq(connectionGrants.id, personalGrant.id)); + await notifyToolConnectionHealth(tx, connection); [connection] = await tx .update(toolConnections) .set({ diff --git a/server/src/services/tool-access/oauth/cloud-callback.ts b/server/src/services/tool-access/oauth/cloud-callback.ts index dd3c6feed87..af2872ce343 100644 --- a/server/src/services/tool-access/oauth/cloud-callback.ts +++ b/server/src/services/tool-access/oauth/cloud-callback.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { managedConnectorProfile } from "./client-settings.js"; import type { ActorInfo } from "../types.js"; import { oauthConfig, normalizeOauthScopes } from "./metadata.js"; @@ -356,6 +357,7 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { scopes: [...profile.scopes], }, }; + await notifyToolConnectionHealth(tx, connection); [connection] = await tx .update(toolConnections) .set({ diff --git a/server/src/services/tool-access/oauth/connection-refresh.ts b/server/src/services/tool-access/oauth/connection-refresh.ts index f6127cb9581..4cdf1e8a8e5 100644 --- a/server/src/services/tool-access/oauth/connection-refresh.ts +++ b/server/src/services/tool-access/oauth/connection-refresh.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import type { ActorInfo } from "../types.js"; import { createOAuthClientProtocol, storedOAuthTokenEndpointAuthMethod } from "./client.js"; import { oauthConfig, normalizeOauthScopes } from "./metadata.js"; @@ -59,7 +60,7 @@ const oauthRefreshFlights = new Map>(); reauthorizationRequiredAt: now().toISOString(), }, }; - const [updated] = await db + const [updated] = await writeToolConnectionHealthChange(db, connection, tx => tx .update(toolConnections) .set({ status: "draft", @@ -87,7 +88,7 @@ const oauthRefreshFlights = new Map>(); )`, ), ) - .returning(); + .returning()); if (updated) { await syncCredentialBindings(updated); emitConnectionUpdated(updated, connection, "credential_refresh"); diff --git a/server/src/services/tool-access/oauth/direct-callback.ts b/server/src/services/tool-access/oauth/direct-callback.ts index b46e6c8fd16..9029e35438c 100644 --- a/server/src/services/tool-access/oauth/direct-callback.ts +++ b/server/src/services/tool-access/oauth/direct-callback.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { credentialScope } from "./client-settings.js"; import { withoutOAuthRefreshLease } from "./refresh-leases.js"; import type { ActorInfo } from "../types.js"; @@ -332,6 +333,7 @@ export function createDirectOAuthCallback(db: Db, dependencies: Dependencies) { }, }, }; + await notifyToolConnectionHealth(tx, connection); const [updatedConnection] = await tx .update(toolConnections) .set({ @@ -611,6 +613,7 @@ export function createDirectOAuthCallback(db: Db, dependencies: Dependencies) { }, }, }; + await notifyToolConnectionHealth(tx, connection); const [updatedConnection] = await tx .update(toolConnections) .set({ diff --git a/server/src/services/tool-access/oauth/grant-refresh.ts b/server/src/services/tool-access/oauth/grant-refresh.ts index 4a3748f6d83..48e09cf27b9 100644 --- a/server/src/services/tool-access/oauth/grant-refresh.ts +++ b/server/src/services/tool-access/oauth/grant-refresh.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import type { ActorInfo } from "../types.js"; import { createOAuthClientProtocol, storedOAuthTokenEndpointAuthMethod } from "./client.js"; import { oauthConfig, normalizeOauthScopes, resolveGrantedOauthScopes } from "./metadata.js"; @@ -393,7 +394,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo connection.id, connection.companyId, ); - const [reauthorizationRequired] = await db + const [reauthorizationRequired] = await writeToolConnectionHealthChange(db, latestConnection, tx => tx .update(toolConnections) .set({ status: "draft", @@ -417,7 +418,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo eq(toolConnections.companyId, latestConnection.companyId), ), ) - .returning(); + .returning()); await syncCredentialBindings(reauthorizationRequired); emitConnectionUpdated( reauthorizationRequired, diff --git a/server/src/services/tool-access/oauth/vercel-callback.ts b/server/src/services/tool-access/oauth/vercel-callback.ts index 452497f38ea..543470a1a1d 100644 --- a/server/src/services/tool-access/oauth/vercel-callback.ts +++ b/server/src/services/tool-access/oauth/vercel-callback.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectionActorUserId } from "../actor.js"; import { vercelConnectHttpError, vercelCredentialFor, vercelConnectResourcesFor } from "../vercel-credentials.js"; import type { ActorInfo } from "../types.js"; @@ -182,7 +183,7 @@ export function createVercelOAuthCallback(db: Db, dependencies: Dependencies) { status: connection.status, enabled: connection.enabled, }; - [connection] = await db + [connection] = await writeToolConnectionHealthChange(db, connection, tx => tx .update(toolConnections) .set({ status: "active", @@ -197,7 +198,7 @@ export function createVercelOAuthCallback(db: Db, dependencies: Dependencies) { eq(toolConnections.companyId, connection.companyId), ), ) - .returning(); + .returning()); emitConnectionUpdated( connection, preActivationLifecycle, diff --git a/server/src/services/tool-connection-health-notifications.ts b/server/src/services/tool-connection-health-notifications.ts new file mode 100644 index 00000000000..075983d2972 --- /dev/null +++ b/server/src/services/tool-connection-health-notifications.ts @@ -0,0 +1,27 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +type HealthChange = string | { companyId: string; id: string } | { companyId: string; applicationId: string }; +export type HealthScope = { companyId?: string; connectionId?: string; applicationId?: string }; + +export function healthScope(hint: string): HealthScope { + if (hint === "*") return {}; + const [companyId, kind, id, extra] = hint.split(":"); + if (companyId && kind === undefined) return { companyId }; + if (companyId && id && extra === undefined) { + if (kind === "connection") return { companyId, connectionId: id }; + if (kind === "application") return { companyId, applicationId: id }; + } + throw new Error("Invalid tool connection health scope"); +} + +export const notifyToolConnectionHealth = (transaction: object, target: HealthChange) => + notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolConnectionHealth, typeof target === "string" ? target + : "id" in target ? `${target.companyId}:connection:${target.id}` : `${target.companyId}:application:${target.applicationId}`); + +export function writeToolConnectionHealthChange(db: Db, target: HealthChange, write: (transaction: Db) => Promise): Promise { + return db.transaction(async tx => { + await notifyToolConnectionHealth(tx, target); + return write(tx as unknown as Db); + }); +} diff --git a/server/src/services/tool-connection-health-work.ts b/server/src/services/tool-connection-health-work.ts new file mode 100644 index 00000000000..cc3197532d8 --- /dev/null +++ b/server/src/services/tool-connection-health-work.ts @@ -0,0 +1,104 @@ +import { healthScope, type HealthScope } from "./tool-connection-health-notifications.js"; +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 { ConnectionHealthDeadline } from "./tool-access/catalog/health-schedule.js"; + +export function registerToolConnectionHealthWork( + coordinator: ReturnType, + service: { + load(companyId?: string, connectionId?: string, applicationId?: string, signal?: AbortSignal): Promise; + check(id: string): Promise; + }, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const deadlines = new Map(); + const failures = new Map(); + const refreshFailures = new Map(); + const matches = (entry: ConnectionHealthDeadline, scope: HealthScope) => + (scope.companyId === undefined || entry.companyId === scope.companyId) + && (scope.connectionId === undefined || entry.id === scope.connectionId) + && (scope.applicationId === undefined || entry.applicationId === scope.applicationId); + const retryMs = 30_000; + let earliest = 0; + let lastCompany: string | null = null; + function request(hints?: ReadonlySet) { + for (const hint of hints ?? ["*"]) dirty.add(hint); + } + function next() { + let result: number | null = null; + for (const scope of dirty) { + const at = Math.max(earliest, refreshFailures.get(scope)?.retryAt ?? 0); + result = result === null ? at : Math.min(result, at); + } + for (const entry of deadlines.values()) { + let at = Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0); + for (const failure of refreshFailures.values()) if (matches(entry, failure.scope)) at = Math.max(at, failure.retryAt); + result = result === null ? at : Math.min(result, at); + } + return result; + } + return coordinator.register(DELIVERY_QUEUES.toolConnectionHealth, { + retryMs, + pausePassiveMaintenance: true, + onSettled: request, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + retryMs; + const scopes = dirty.has("*") ? ["*"] : [...dirty]; + dirty.clear(); + const errors: unknown[] = []; + const unavailable: HealthScope[] = []; + for (let index = 0; index < scopes.length; index++) { + const scope = scopes[index]; + if (signal.aborted || !canRun()) { for (const remaining of scopes.slice(index)) dirty.add(remaining); return; } + let target: HealthScope | undefined; + try { + target = healthScope(scope); + if ((refreshFailures.get(scope)?.retryAt ?? 0) > Date.now()) { + dirty.add(scope); unavailable.push(target); continue; + } + const rows = await service.load(target.companyId, target.connectionId, target.applicationId, signal); + if (target.connectionId) { + const current = deadlines.get(target.connectionId); + if (current && matches(current, target)) deadlines.delete(current.id); + } else for (const entry of deadlines.values()) if (matches(entry, target)) deadlines.delete(entry.id); + for (const entry of rows) deadlines.set(entry.id, entry); + if (scope === "*") refreshFailures.clear(); else refreshFailures.delete(scope); + } catch (error) { + const retryScope = target ? scope : "*"; + refreshFailures.set(retryScope, { scope: target ?? {}, retryAt: Date.now() + retryMs }); + dirty.add(retryScope); unavailable.push(target ?? {}); errors.push(error); + } + } + for (const id of failures.keys()) { + const current = deadlines.get(id); + if (!current || current.dueAt > Date.now()) failures.delete(id); + } + const due = [...deadlines.values()].filter(entry => !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); + if (current) deadlines.set(entry.id, current); else deadlines.delete(entry.id); + failures.delete(entry.id); + } catch (error) { + const count = Math.min((failures.get(entry.id)?.count ?? 0) + 1, 6); + failures.set(entry.id, { count, retryAt: Date.now() + Math.min(15 * 60_000, retryMs * 2 ** (count - 1)) }); + errors.push(error); + } + } + if (due.length > batch.entries.length) earliest = Date.now() + 1_000; + if (errors.length) throw new AggregateError(errors, "Tool connection health checks could not settle"); + }, + // Keep the hints/deadlines for resume, but release settled intent while + // paused. The coordinator independently fences unresolved transactions. + hasPending: async () => canRun() && dirty.size > 0, + nextRunAt: next, + }); +} diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index c5c9cdb07b6..bdfb0f8b0f4 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { writeApprovalChange } from "./approval-work-notifications.js"; import { toolReviewProjectionService } from "./tool-review-projection.js"; import { toolReviewInterruptionService } from "./tool-review-interruption.js"; @@ -3882,7 +3883,7 @@ export function createToolGatewayService( message: string | null, ) { const now = new Date(); - await db + await writeToolConnectionHealthChange(db, connection, tx => tx .update(toolConnections) .set({ healthStatus: status, @@ -3892,7 +3893,7 @@ export function createToolGatewayService( lastError: status === "ok" ? null : message, updatedAt: now, }) - .where(eq(toolConnections.id, connection.id)); + .where(eq(toolConnections.id, connection.id))); } function grantRefForCredential(