fix(server): schedule tool health from committed changes and deadlines (#17554)

## Thinking Path

> - Paperclip manages AI-agent work and its connected tools.
> - Idle hosted instances need their database and network activity to
stop.
> - The maintenance tick queries tool connections even when no health
check is due.
> - Existing connection rows already contain eligibility and the last
health observation.
> - This pull request reconstructs health deadlines and refreshes them
after committed changes.
> - The shared coordinator preserves retries, drain, and shutdown
without another private timer.

## Linked Issues or Issue Description

**What happened?**

The aggregate maintenance tick scans connection health every 30 seconds.
Empty instances and connections with fresh health observations still
cause SQL activity.

**Expected behavior**

No eligible connection means no recurring health SQL. Eligible
connections retain the existing 15-minute proactive check. Configuration
changes and health observations update the schedule after commit.

**Steps to reproduce**

Start a server with heartbeat scheduling enabled and no tool
connections. Let startup finish and observe two maintenance ticks. The
old code repeats the health inventory query. The new startup regression
verifies that those ticks do not invoke the health inventory.

**Paperclip version or commit**

This branch includes merged parent #17518 and master
`96727b3cfefb61f85ffe1fbc29a9b6dd6837efe8`. The 47 changed paths belong
to the single tool-connection health duty. No database file differs from
master.

Related: #14638 covers runtime-auth exclusions. #15448, #15325, and
#11910 concern MCP failure classification. This PR preserves those
provider policies and does not replace their work.

## What Changed

- Keep the GitHub error-projection unit fixture at the health-write
boundary and verify the real transaction path in the PostgreSQL
health-worker suite.

- Add a health schedule owned by tool access. Read lean eligibility
fields and saved observation timestamps.
- Register health work with the existing coordinator. Refresh only dirty
connection/application/company scopes using bounded inventory pages,
check due connections in fair company batches, continue concrete due
backlog after one second, and retain failures with backoff.
- Notify before supported eligibility and observation writes in their
owning transactions. Cover API/gallery setup, OAuth, catalog/gateway
observations, email, voice, access removal, deletion, and settings.
- Commit connection creation with grant, binding, and runtime-slot SQL
before its wake. Keep provider calls outside the new transactions.
- Keep provider-to-transport mapping in the chat catalog. SDK-only chat
changes do not wake health work.
- Replace aggregate health scans, retain the public manual sweep, and
document the producer audit and remaining sleep limits.
- Keep metadata-only example and Smoke Lab installs scoped. Lock and
re-read application type before publishing a real eligibility change.
- Keep failed inventory reads on their own retry deadline while healthy
due batches continue.
- Apply fixture migrations before using external disposable databases in
all six producer suites.

## Verification

Published head: `8c1d293d1ae2b7db753ab20f5a2a2edec1fb0a62`. Fresh CI and
reviews are required for this exact head.

- **623 distinct targeted cases pass.** The final four complete health,
fair-batching, GitHub health recovery, and startup suites pass 80 cases
after the retry fix. Five complete producer suites pass 543 cases: tool
access 469, access 18, email 45, GitHub events four, and Smoke Lab
seven. The earlier 26 health cases are part of this total, not
additional cases.
- The six producer suites first passed against separate, initially empty
PostgreSQL 17 databases at `acbf262843`. They apply their own
migrations. The subsequent repair changes only the health worker and its
regression; its complete final consumer suites pass on that integrated
parent. Server typecheck and build pass with the final repair.
- **Installer amplification:** five real installs with 205 unrelated
connections reproduced 1,035 example rows and 1,045 Smoke Lab rows.
Repeated metadata installs now load only changed fixture connections.
The same cases restore an excluded application through each installer
and require all 205 unrelated IDs to re-enter the schedule.
- **Concurrent eligibility:** a real PostgreSQL lock-wait case requires
the installer to notify after a concurrent type change. A mutation that
retains the lock but uses the old lookup fails the settled-hint
assertion; original source was restored byte for byte.
- **Retry isolation:** the failing baseline read one unavailable company
three times while another company's 60 due checks drained. Final tests
require one initial read, at least 30 seconds between failed reads,
prompt healthy progress, eventual recovery, and SQL quiet after
settlement.
- Real-service producer coverage includes AgentMail activation/pause,
GitHub installation changes, and member access removal. Lifecycle
coverage includes outer commit/rollback, startup reconstruction,
drain/resume, failure, and shutdown.
- Failure history and exact evidence scope: [health
acceptance](https://github.com/paperclipai/paperclip/blob/8c1d293d1ae2b7db753ab20f5a2a2edec1fb0a62/doc/idle-tool-connection-health.md#acceptance-follow-up-after-the-plugin-parent-merged-october-27),
[producer
audit](https://github.com/paperclipai/paperclip/blob/8c1d293d1ae2b7db753ab20f5a2a2edec1fb0a62/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md).
Prior remote failures and earlier 706-case evidence remain historical.
Startup tests emitted listener-count warnings; all assertions passed. No
full local repository-suite, deployed sleep, fleet savings, or live
provider qualification claim is made.

## Risks

- Supported live writers must use the transaction signals. Independent
SQL writers and other replicas need their own wake integration.
- Configured eligible connections still have a real proactive health
obligation. This does not prove Cloud alarm coverage or host/database
suspension.
- Existing manual probes and catalog operations can overlap. This
migration does not introduce a credential-version CAS or change provider
failure policy.
- Voice remains eligible under the previous health policy. Its generic
probe can still report an unsupported transport; changing that behavior
is separate work.
- No database migration, new table, workflow, or lockfile change. This
PR includes its merged parent and current inspected master; later stack
PRs and all current-head acceptance gates remain separate.

## Model Used

OpenAI Codex, GPT-6 family, with repository tools, code execution, and
reasoning. The exact serving version and context-window size are not
exposed in this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-27 21:37:50 -05:00
1 parent 5cfa2bf26d
commit 94a780286f
47 files changed
+1795 -246

No files matched your search

+291
View File
@@ -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.
@@ -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.
@@ -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.
@@ -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.
+19 -6
View File
@@ -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<typeof createDb>) {
@@ -61,8 +65,9 @@ describeEmbeddedPostgres("access service", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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);
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
@@ -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<typeof vi.fn> }) => Promise<void>,
beforeSetup?: (input: { companyId: string; fetcher: typeof fetch; wakeup: ReturnType<typeof vi.fn> }) => Promise<void>,
) {
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<ReturnType<typeof startToolHealthWorker>> | 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.
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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) {
@@ -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<typeof import("../services/tool-connection-health-notifications.js")>(),
writeToolConnectionHealthChange: async <T>(db: Db, _target: unknown, write: (transaction: Db) => Promise<T>) => write(db),
}));
const SUSPENDED_INSTALLATION_URL = "https://github.com/organizations/paperclipai/settings/installations/101";
function json(value: unknown): Response {
@@ -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<typeof connectionHealthSchedule>[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(); }
},
};
}
@@ -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<typeof setInterval>;
}) 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<void>(resolve => setImmediate(resolve));
tick!(); await new Promise<void>(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,
+8 -3
View File
@@ -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<typeof createDb>;
@@ -104,8 +107,9 @@ describeEmbeddedPostgres("smoke lab service pack and results API", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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();
});
@@ -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 () => {
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>, companyId: string, applicationId: string;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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<typeof toolConnections.$inferInsert> = {}, 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<string>();
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<string>();
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<void>(resolve => { unlock = resolve; });
const admitted = new Promise<void>(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<typeof tools.installExample> | 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<void>(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([]);
});
});
+3 -18
View File
@@ -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
+2
View File
@@ -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,
+2
View File
@@ -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({
@@ -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({
@@ -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 });
}
@@ -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",
@@ -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({
@@ -106,3 +106,9 @@ const CAPABILITIES: Record<ChatProvider, ChatAdapterCapabilities> = {
},
};
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";
}
@@ -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({
@@ -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({
@@ -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({
@@ -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({
+2
View File
@@ -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))
@@ -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",
+8 -4
View File
@@ -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({
+3 -2
View File
@@ -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)
@@ -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
+2
View File
@@ -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({
+58 -46
View File
@@ -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) {
+106 -82
View File
@@ -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,
@@ -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<ConnectionHealthDeadline[]> {
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<string | null>`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<ConnectionHealthDeadline | null> {
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 };
}
@@ -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)
@@ -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);
@@ -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");
@@ -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 };
@@ -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({
@@ -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({
@@ -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<string, Promise<unknown>>();
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<string, Promise<unknown>>();
)`,
),
)
.returning();
.returning());
if (updated) {
await syncCredentialBindings(updated);
emitConnectionUpdated(updated, connection, "credential_refresh");
@@ -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({
@@ -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,
@@ -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,
@@ -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<T>(db: Db, target: HealthChange, write: (transaction: Db) => Promise<T>): Promise<T> {
return db.transaction(async tx => {
await notifyToolConnectionHealth(tx, target);
return write(tx as unknown as Db);
});
}
@@ -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<typeof createDeliveryWorkCoordinator>,
service: {
load(companyId?: string, connectionId?: string, applicationId?: string, signal?: AbortSignal): Promise<ConnectionHealthDeadline[]>;
check(id: string): Promise<ConnectionHealthDeadline | null>;
},
canRun: () => boolean,
) {
const dirty = new Set<string>(["*"]);
const deadlines = new Map<string, ConnectionHealthDeadline>();
const failures = new Map<string, { count: number; retryAt: number }>();
const refreshFailures = new Map<string, { scope: HealthScope; retryAt: number }>();
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<string>) {
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,
});
}
+3 -2
View File
@@ -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(