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(