From 2ca12f806399cfcef1a474453301534f4b96c1ef Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 00:27:46 -0500 Subject: [PATCH] fix(server): schedule tool health from connection changes and deadlines Co-Authored-By: Paperclip --- doc/idle-tool-connection-health.md | 118 ++++++++ ...2026-10-24-recurring-server-sleep-stack.md | 22 ++ ...0-24-tool-connection-health-sleep-audit.md | 19 ++ server/src/__tests__/access-service.test.ts | 9 +- .../email-channels.integration.test.ts | 8 +- .../github-connection-events.test.ts | 9 +- .../server-startup-feedback-export.test.ts | 12 +- server/src/__tests__/smoke-lab.test.ts | 9 +- .../src/__tests__/tool-access-service.test.ts | 64 +++-- .../tool-connection-health-work.test.ts | 260 ++++++++++++++++++ server/src/index.ts | 20 +- server/src/services/access.ts | 2 + server/src/services/chat-channels.ts | 2 + .../services/chat-channels/configuration.ts | 7 + .../connection-health-notifications.ts | 8 + .../src/services/chat-channels/endpoints.ts | 3 +- .../chat-channels/interactions/callbacks.ts | 2 + .../chat-channels/provider-catalog.ts | 6 + .../chat-channels/provider-effects.ts | 2 + .../chat-channels/publications/failures.ts | 2 + .../chat-channels/resource-reconciliation.ts | 3 + .../services/chat-channels/setup-probes.ts | 2 + server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + server/src/services/email-channels.ts | 12 +- server/src/services/email-connections.ts | 5 +- .../src/services/github-connection-events.ts | 2 + server/src/services/instance-settings.ts | 2 + server/src/services/smoke-lab.ts | 84 +++--- server/src/services/tool-access.ts | 185 +++++++------ .../tool-access/catalog/health-schedule.ts | 48 ++++ .../services/tool-access/catalog/health.ts | 5 +- .../tool-access/catalog/publication.ts | 5 +- .../tool-access/connections/removal.ts | 2 + .../tool-access/connections/setup-policy.ts | 8 +- .../tool-access/oauth/access-finalization.ts | 60 ++-- .../tool-access/oauth/cloud-callback.ts | 2 + .../tool-access/oauth/connection-refresh.ts | 5 +- .../tool-access/oauth/direct-callback.ts | 3 + .../tool-access/oauth/grant-refresh.ts | 5 +- .../tool-access/oauth/vercel-callback.ts | 5 +- .../tool-connection-health-notifications.ts | 12 + .../services/tool-connection-health-work.ts | 73 +++++ server/src/services/tool-gateway.ts | 5 +- 44 files changed, 889 insertions(+), 231 deletions(-) create mode 100644 doc/idle-tool-connection-health.md create mode 100644 server/src/__tests__/tool-connection-health-work.test.ts create mode 100644 server/src/services/chat-channels/connection-health-notifications.ts create mode 100644 server/src/services/tool-access/catalog/health-schedule.ts create mode 100644 server/src/services/tool-connection-health-notifications.ts create mode 100644 server/src/services/tool-connection-health-work.ts diff --git a/doc/idle-tool-connection-health.md b/doc/idle-tool-connection-health.md new file mode 100644 index 00000000000..1fbaad7bb42 --- /dev/null +++ b/doc/idle-tool-connection-health.md @@ -0,0 +1,118 @@ +# 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. A company hint refreshes that company's cache. Startup and an +instance feature-settings change can refresh the whole instance. Refreshes are +coalesced with a 30-second minimum spacing. 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. + +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 company +refresh does not prevent another company's checks. 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. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 978687012b5..cc4d19c72d7 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1000,3 +1000,25 @@ PR-product predicate. All 124 affected worker/queue/startup/retention/attention cases pass, along with server types/build. One initial consumer command skipped six queue tests because its DB variable was missing; the corrected run executes and passes them. Fresh remote checks and review remain required after publication. + +## 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. diff --git a/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md b/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md index 8d54547faf4..04abe345156 100644 --- a/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md +++ b/doc/plans/2026-10-24-tool-connection-health-sleep-audit.md @@ -144,3 +144,22 @@ Additional owner evidence changes the implementation plan: 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. diff --git a/server/src/__tests__/access-service.test.ts b/server/src/__tests__/access-service.test.ts index 2a00221be42..5a340cbffcf 100644 --- a/server/src/__tests__/access-service.test.ts +++ b/server/src/__tests__/access-service.test.ts @@ -1,3 +1,4 @@ +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"; @@ -28,7 +29,8 @@ import { accessService } from "../services/access.js"; import { grantsForHumanRole } from "../services/company-member-roles.js"; import { backfillPrincipalAccessCompatibility } from "../services/principal-access-compatibility.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ACCESS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; async function createCompanyWithOwner(db: ReturnType) { @@ -61,8 +63,8 @@ describeEmbeddedPostgres("access service", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-access-service-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-access-service-"); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -85,6 +87,7 @@ describeEmbeddedPostgres("access service", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/email-channels.integration.test.ts b/server/src/__tests__/email-channels.integration.test.ts index bf297326a1c..b22003f5d23 100644 --- a/server/src/__tests__/email-channels.integration.test.ts +++ b/server/src/__tests__/email-channels.integration.test.ts @@ -1,3 +1,4 @@ +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"; @@ -72,6 +73,8 @@ import { toolAccessService } from "../services/tool-access.js"; import { emailSendSchema } from "@paperclipai/shared"; import { chatChannelService } from "../services/chat-channels.js"; +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EMAIL_TEST_DATABASE_URL"); + describe("AgentMail durable email pipeline", () => { let database: Awaited>; let db: ReturnType; @@ -83,8 +86,8 @@ describe("AgentMail durable email pipeline", () => { folder, "master.key", ); - database = await startEmbeddedPostgresTestDatabase("paperclip-email-"); - db = createDb(database.connectionString); + if (!externalTestDatabaseUrl) database = await startEmbeddedPostgresTestDatabase("paperclip-email-"); + db = createDb(externalTestDatabaseUrl ?? database!.connectionString); await instanceSettingsService(db).updateExperimental({ enableChatConnectors: false, }); @@ -111,6 +114,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; diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 77d34f15e6a..7ed7d130232 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { @@ -20,7 +21,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", () => { @@ -28,8 +30,8 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-github-events-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-github-events-"); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -43,6 +45,7 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index ec2a01f88ec..4f632aa71ae 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -424,6 +424,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, @@ -847,12 +848,17 @@ describe("startServer feedback export wiring", () => { 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()).toBeGreaterThan(0); - const startupCalls = count(); + expect(count()).toBe(0); + const registrations = deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.toolConnectionHealth); + expect(registrations).toHaveLength(1); + const loads = vi.mocked(toolAccessService).mock.results.map(result => result.value.scheduledHealth.load); + const loadCount = () => loads.reduce((total, load) => total + load.mock.calls.length, 0); + expect(loadCount()).toBe(1); expect(tick).toEqual(expect.any(Function)); tick!(); await new Promise(resolve => setImmediate(resolve)); tick!(); await new Promise(resolve => setImmediate(resolve)); - expect(count()).toBe(startupCalls); + expect(count()).toBe(0); + expect(loadCount()).toBe(1); } finally { intervals.mockRestore(); } }); diff --git a/server/src/__tests__/smoke-lab.test.ts b/server/src/__tests__/smoke-lab.test.ts index 297f5843ad5..941dc3093c4 100644 --- a/server/src/__tests__/smoke-lab.test.ts +++ b/server/src/__tests__/smoke-lab.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; @@ -27,7 +28,8 @@ import { smokeLabRoutes } from "../routes/smoke-lab.js"; import { SMOKE_LAB_OAUTH_SCOPE } from "../services/smoke-lab.js"; import { errorHandler } from "../middleware/index.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_SMOKE_LAB_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; type TestDb = ReturnType; @@ -104,8 +106,8 @@ describeEmbeddedPostgres("smoke lab service pack and results API", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-smoke-lab-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-smoke-lab-"); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -128,6 +130,7 @@ describeEmbeddedPostgres("smoke lab service pack and results API", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/tool-access-service.test.ts b/server/src/__tests__/tool-access-service.test.ts index 88c936be756..38c88536a13 100644 --- a/server/src/__tests__/tool-access-service.test.ts +++ b/server/src/__tests__/tool-access-service.test.ts @@ -15876,38 +15876,44 @@ describeEmbeddedPostgres("tool access service", () => { ); await service.archiveConnection(first.connectionId, company.id, actor); fetchMock.mockRejectedValue(new Error("provider unavailable")); - const runTransaction = db.transaction.bind(db); - vi.spyOn(db, "transaction") - .mockImplementationOnce(runTransaction) - .mockRejectedValueOnce(new Error("rollback unavailable")); + await db.execute(sql`create function tool_health_rollback_failure() returns trigger language plpgsql as $$ + begin raise exception 'rollback unavailable'; end $$`); + await db.execute(sql.raw(`create trigger tool_health_rollback_failure before update on tool_connections + for each row when (old.id = '${first.connectionId}'::uuid and old.credential_policy = 'per_user' and new.credential_policy = 'shared') + execute function tool_health_rollback_failure()`)); + try { - await expect( - service.connectGalleryApp( - company.id, - { - galleryKey: "github", - connectionMethodKey: "mcp-key", - grantKind: "user", - name: "GitHub rollback failure", - credentialValues: { - "credentials.authorization": "new-personal-token", + await expect( + service.connectGalleryApp( + company.id, + { + galleryKey: "github", + connectionMethodKey: "mcp-key", + grantKind: "user", + name: "GitHub rollback failure", + credentialValues: { + "credentials.authorization": "new-personal-token", + }, }, - }, - actor, - ), - ).rejects.toMatchObject({ - status: 500, - details: { code: "connection_identity_rollback_failed" }, - }); + actor, + ), + ).rejects.toMatchObject({ + status: 500, + details: { code: "connection_identity_rollback_failed" }, + }); - await expect( - service.getConnection(first.connectionId, company.id), - ).resolves.toMatchObject({ - status: "draft", - enabled: false, - healthStatus: "error", - lastError: "connection_identity_rollback_failed", - }); + await expect( + service.getConnection(first.connectionId, company.id), + ).resolves.toMatchObject({ + status: "draft", + enabled: false, + healthStatus: "error", + lastError: "connection_identity_rollback_failed", + }); + } finally { + await db.execute(sql`drop trigger tool_health_rollback_failure on tool_connections`); + await db.execute(sql`drop function tool_health_rollback_failure()`); + } }); it("automatically gives same-named connections distinct names", async () => { diff --git a/server/src/__tests__/tool-connection-health-work.test.ts b/server/src/__tests__/tool-connection-health-work.test.ts new file mode 100644 index 00000000000..da1bded5041 --- /dev/null +++ b/server/src/__tests__/tool-connection-health-work.test.ts @@ -0,0 +1,260 @@ +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 { toolAccessService } from "../services/tool-access.js"; +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { companies, createDb, 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 { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_TOOL_HEALTH_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("scheduled tool connection health", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, applicationId: string; + let coordinator: ReturnType | undefined; + let draining = false; + const errors: unknown[] = []; + const probe = vi.fn(); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("tool-health-"); + 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.companyId, async tx => { + await tx.update(toolConnections).set({ healthCheckedAt: new Date(), healthStatus: "ok" }).where(eq(toolConnections.id, id)); + }); + return {}; + }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(companies).where(eq(companies.id, companyId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + function connection(patch: Partial = {}, signal = true) { + const insert = async (writer: typeof db) => (await writer.insert(toolConnections).values({ + companyId, applicationId, name: "Check", uid: randomUUID(), transport: "mcp_remote", status: "active", enabled: true, ...patch, + }).returning())[0]; + return signal ? writeToolConnectionHealthChange(db, companyId, 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("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(); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); 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(); + await start(); await checked(30); + 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("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 scope => { if (scope === companyId) throw new Error("scope unavailable"); return load(scope); }; + 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("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, provider: "slack" }); }); + await quiet(); + await db.transaction(async tx => { + await notifyChatConnectionHealth(tx, { companyId, 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 scope => { + const rows = await load(scope); + if (scope === companyId && !inserted) { inserted = true; await connection(); } + return rows; + }; + await connection(); await checked(2); + expect(new Set(probe.mock.calls.map(([id]) => id)).size).toBe(2); + }); + it("backs off an unpersisted check without blocking another connection and recovers", async () => { + const failed = await connection(), healthy = await connection(); + const persist = probe.getMockImplementation()!; + let broken = true; + probe.mockImplementation(async id => { if (id === failed.id && broken) throw new Error("health write unavailable"); return persist(id); }); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(probe.mock.calls.some(([id]) => id === healthy.id)).toBe(true); + expect(errors.length).toBeGreaterThan(0); + }); + const before = probe.mock.calls.length; + await vi.advanceTimersByTimeAsync(10_000); expect(probe).toHaveBeenCalledTimes(before); + broken = false; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + const [stored] = await db.select().from(toolConnections).where(eq(toolConnections.id, failed.id)); + expect(stored.healthCheckedAt).not.toBeNull(); + }, { timeout: 5000 }); + errors.length = 0; + }); + it("awaits an admitted health check during shutdown and stops future work", async () => { + await start(); await quiet(); + const persist = probe.getMockImplementation()!; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + probe.mockImplementation(async id => { await gate; return persist(id); }); + await connection(); await checked(); + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(); await stopping; coordinator = undefined; + await vi.advanceTimersByTimeAsync(1_800_000); expect(probe).toHaveBeenCalledOnce(); + }); + +}); diff --git a/server/src/index.ts b/server/src/index.ts index 06ed12a86ef..88f7a00b85d 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerToolConnectionHealthWork } from "./services/tool-connection-health-work.js"; import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "./services/approval-queue-seeding-work.js"; import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "./services/join-request-retention.js"; import { approvalRetentionService, registerApprovalRetentionWork } from "./services/approval-retention.js"; @@ -1802,12 +1803,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, @@ -1896,17 +1893,6 @@ async function startServerWithDatabaseTeardown( })); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(tools - .sweepConnectionHealth() - .then((swept) => { - if (swept.failed > 0) { - logger.warn({ ...swept }, "periodic tool connection health sweep found failing connections"); - } - }) - .catch((err) => { - logger.error({ err }, "periodic tool connection health sweep failed"); - })); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { // Periodically reap orphaned runs (5-min staleness threshold) and make sure diff --git a/server/src/services/access.ts b/server/src/services/access.ts index fced6f2c698..d80bde25f99 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { and, eq, inArray, ne, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -323,6 +324,7 @@ export function accessService(db: Db) { && grant.credentialSecretRefs.length > 0 && grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)), ); + if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, companyId); await tx.update(toolConnections).set({ credentialRefs, credentialSecretRefs, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 6963c785d7f..1cc8a7aff6a 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./chat-channels/connection-health-notifications.js"; import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; import { resolveInboundConversationBinding } from "./chat-channels/inbound/conversation-binding.js"; @@ -2299,6 +2300,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { updatedAt: new Date(), }) .where(eq(chatConversations.id, conversation!.id)); + await notifyChatConnectionHealth(taskTx, taskEndpoint); await taskTx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/configuration.ts b/server/src/services/chat-channels/configuration.ts index 80fbaafd756..2d58158c7f7 100644 --- a/server/src/services/chat-channels/configuration.ts +++ b/server/src/services/chat-channels/configuration.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { preparePhotonConfiguration, createPhotonConfiguration } from "./imessage-photon/setup.js"; import { slackRuntimeConfig } from "./slack/setup.js"; import { createTelegramSetup } from "./telegram/setup.js"; @@ -193,6 +194,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() }) @@ -336,6 +338,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption }) .where(eq(chatEndpoints.id, endpoint.id)) .returning(); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ @@ -380,6 +383,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({ @@ -484,6 +488,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() }) @@ -812,6 +817,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption updatedAt, }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ @@ -873,6 +879,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption updatedAt: new Date(), }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/connection-health-notifications.ts b/server/src/services/chat-channels/connection-health-notifications.ts new file mode 100644 index 00000000000..c2709c339e0 --- /dev/null +++ b/server/src/services/chat-channels/connection-health-notifications.ts @@ -0,0 +1,8 @@ +import type { ChatProvider } from "@paperclipai/shared"; +import { notifyToolConnectionHealth } from "../tool-connection-health-notifications.js"; +import { chatConnectionTransport } from "./provider-catalog.js"; + +export async function notifyChatConnectionHealth(transaction: object, endpoint: { companyId: string; provider: ChatProvider }) { + if (chatConnectionTransport(endpoint.provider) === "chat_sdk") return; + await notifyToolConnectionHealth(transaction, endpoint.companyId); +} diff --git a/server/src/services/chat-channels/endpoints.ts b/server/src/services/chat-channels/endpoints.ts index 842860e1036..a59879cfb09 100644 --- a/server/src/services/chat-channels/endpoints.ts +++ b/server/src/services/chat-channels/endpoints.ts @@ -1,3 +1,4 @@ +import { chatConnectionTransport } from "./provider-catalog.js"; import { slackSetupState } from "./slack/setup.js"; import { assertChatExecutionDefaultsAccess } from "../chat-execution-defaults.js"; @@ -326,7 +327,7 @@ function providerSetupState( connectionKind: "managed", connectionPurpose: "channel", ownership: "customer", - transport: input.provider === "speko" ? "voice" : "chat_sdk", + transport: chatConnectionTransport(input.provider), authKind: "api_key", credentialPolicy: "shared", status: "draft", diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index 93c1e5f9f07..04c1f647523 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { createDiscordQuestionForms } from "../discord/question-forms.js"; import { createProviderEffects, type ProviderEffectPayload } from "../provider-effects.js"; @@ -420,6 +421,7 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb updatedAt: processedAt, }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/provider-catalog.ts b/server/src/services/chat-channels/provider-catalog.ts index 5beb7879400..2dc6e819180 100644 --- a/server/src/services/chat-channels/provider-catalog.ts +++ b/server/src/services/chat-channels/provider-catalog.ts @@ -106,3 +106,9 @@ const CAPABILITIES: Record = { }, }; export { PROVIDER_LABELS, CAPABILITIES }; + +export function chatConnectionTransport(provider: ChatProvider): "voice" | "rest_api" | "chat_sdk" { + if (provider === "speko") return "voice"; + if (provider === "agentmail") return "rest_api"; + return "chat_sdk"; +} diff --git a/server/src/services/chat-channels/provider-effects.ts b/server/src/services/chat-channels/provider-effects.ts index 5c663aaeb9f..4b15a1b769d 100644 --- a/server/src/services/chat-channels/provider-effects.ts +++ b/server/src/services/chat-channels/provider-effects.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { createTelegramCallbackNoticeAuthority } from "./telegram/callback-notices.js"; import { slackThreadChannelId } from "./slack/messages.js"; @@ -325,6 +326,7 @@ export function createProviderEffects(db: Db, options: ProviderEffectsOptions) { updatedAt: new Date(), }) .where(eq(chatEndpoints.id, currentEndpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index 1f701de8cb8..84184bb8633 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; import type { EndpointRow, DbTransaction, LifecycleRuntimeFence } from "../runtime-types.js"; @@ -197,6 +198,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur updatedAt: new Date(), }) .where(eq(chatEndpoints.id, currentEndpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index 66a82e76ff3..3658dbb3169 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { imessageResourceParticipants } from "./imessage-photon/resources.js"; import { createGitHubResources, reconcileGitHubRepositoryCoordinates } from "./github/resources.js"; import { baseTeamsConversationId } from "./microsoft-teams/resources.js"; @@ -522,6 +523,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({ @@ -551,6 +553,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc updatedAt: now, }) .where(eq(chatEndpoints.id, currentEndpoint.id)); + await notifyChatConnectionHealth(tx, currentEndpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/chat-channels/setup-probes.ts b/server/src/services/chat-channels/setup-probes.ts index 3aeb21f8833..d5ae65d845c 100644 --- a/server/src/services/chat-channels/setup-probes.ts +++ b/server/src/services/chat-channels/setup-probes.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import type { EndpointRow, DbOrTransaction } from "./runtime-types.js"; import { notifyChatEndpointWork } from "../chat-work-notifications.js"; @@ -326,6 +327,7 @@ export function createSetupProbes(db: Db, options: SetupProbesOptions) { code: "chat_endpoint_test_superseded", }); } + await notifyChatConnectionHealth(tx, endpoint); await tx .update(toolConnections) .set({ diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index b48a982cc22..87695be7363 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -118,6 +119,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(executionWorkspaces).where(eq(executionWorkspaces.companyId, id)); await tx.delete(projects).where(eq(projects.companyId, id)); await agentLifecycleCompanyDeletion.deleteCompanyData(tx, id); + await notifyToolConnectionHealth(tx, id); const rows = await tx .delete(companies) .where(eq(companies.id, id)) diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 6c03e6db2d4..878a0072ac0 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -16,6 +16,7 @@ export const DELIVERY_QUEUES = { decisionExpiry: "decision-expiry", decisionContinuation: "decision-continuation", decisionArchiveDelivery: "decision-archive-delivery", + toolConnectionHealth: "tool-connection-health", toolReviewProjection: "tool-review-projection", toolReviewInterruption: "tool-review-interruption", toolReviewRecovery: "tool-review-recovery", diff --git a/server/src/services/email-channels.ts b/server/src/services/email-channels.ts index 688aa2d6cda..ab01aaf96f6 100644 --- a/server/src/services/email-channels.ts +++ b/server/src/services/email-channels.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { fastResponseService, enqueueFastResponse, fastResponseSourceCurrent, fastResponseTurnQueued } from "./fast-responses.js"; @@ -913,6 +914,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { updatedAt: now, }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyToolConnectionHealth(tx, endpoint.companyId); await tx .update(toolConnections) .set({ status: "active", enabled: true, healthStatus: "ok" }) @@ -1984,7 +1986,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, endpoint.companyId, tx => tx .update(toolConnections) .set({ healthStatus: "ok", @@ -1992,7 +1994,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) { @@ -2092,7 +2094,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { }); } async function markError(endpoint: Endpoint, error: string) { - await db + await writeToolConnectionHealthChange(db, endpoint.companyId, tx => tx .update(toolConnections) .set({ healthStatus: "degraded", @@ -2100,7 +2102,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() }) @@ -2362,6 +2364,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { .update(chatEndpoints) .set({ status: "paused" }) .where(eq(chatEndpoints.id, endpoint.id)); + await notifyToolConnectionHealth(tx, endpoint.companyId); await tx .update(toolConnections) .set({ enabled: false, status: "disabled" }) @@ -2455,6 +2458,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { updatedAt: new Date(), }) .where(eq(chatEndpoints.id, id)); + await notifyToolConnectionHealth(tx, endpoint.companyId); await tx .update(toolConnections) .set({ diff --git a/server/src/services/email-connections.ts b/server/src/services/email-connections.ts index 78f38ad5058..264b920e971 100644 --- a/server/src/services/email-connections.ts +++ b/server/src/services/email-connections.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, isNull, ne, sql } from "drizzle-orm"; import { @@ -257,7 +258,7 @@ export function emailConnectionService( actorSource: actor.localImplicit ? "local_implicit" : "session", }, ); - await db + await writeToolConnectionHealthChange(db, companyId, tx => tx .update(toolConnections) .set({ uid: `agentmail-account-${input.idempotencyKey}`, @@ -265,7 +266,7 @@ export function emailConnectionService( healthMessage: "Connected", healthCheckedAt: new Date(), }) - .where(eq(toolConnections.id, connection.id)); + .where(eq(toolConnections.id, connection.id))); if (input.grantKind === "user") await db .update(connectionGrants) diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 04022c647a1..d4786c34ab8 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js"; import { chatEndpoints } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; @@ -326,6 +327,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, binding.companyId); await database.update(toolConnections).set({ healthStatus: unavailable ? "failed" : "ok", healthMessage: unavailable diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index d13920e8b9f..7808541872a 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import type { Db } from "@paperclipai/db"; import { companies, instanceSettings } from "@paperclipai/db"; @@ -613,6 +614,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); + await notifyToolConnectionHealth(tx, "*"); return tx .update(instanceSettings) .set({ diff --git a/server/src/services/smoke-lab.ts b/server/src/services/smoke-lab.ts index 931021f54dc..2ae60f5dd8b 100644 --- a/server/src/services/smoke-lab.ts +++ b/server/src/services/smoke-lab.ts @@ -1,3 +1,4 @@ +import { 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,14 @@ export function smokeLabService(db: Db, options: { )); const now = new Date(); if (existing) { - const [updated] = await db.update(toolApplications).set({ + const [updated] = await writeToolConnectionHealthChange(db, input.companyId, tx => 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(); + }).where(eq(toolApplications.id, existing.id)).returning()); return { row: updated ?? existing, created: false }; } const [created] = await db.insert(toolApplications).values({ @@ -742,41 +743,43 @@ 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({ + return writeToolConnectionHealthChange(db, input.companyId, 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({ 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 +966,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, companyId, tx => tx.update(toolConnections).set({ config: { ...asRecord(connection.config), - url: `${httpSidecar.url}/mcp`, - catalogUrl: `${httpSidecar.url}/catalog`, - toolCallUrl: `${httpSidecar.url}/tools/call`, + url: `${sidecarUrl}/mcp`, + catalogUrl: `${sidecarUrl}/catalog`, + toolCallUrl: `${sidecarUrl}/tools/call`, }, healthStatus: "ok", healthMessage: "Smoke Lab HTTP fixture sidecar is running.", healthCheckedAt: now, lastHealthAt: now, updatedAt: now, - }).where(eq(toolConnections.id, connection.id)); + }).where(eq(toolConnections.id, connection.id))); } async function startHttpSidecar(companyId?: string) { diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 42645cbf3ce..e3d4c33feb9 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -1,3 +1,5 @@ +import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; +import { connectionHealthSchedule } from "./tool-access/catalog/health-schedule.js"; import { restConnectorFor } from "./connectors/rest-registry.js"; import { createConnectionCredentialValidation } from "./tool-access/connections/credential-validation.js"; import { createConnectionGrants } from "./tool-access/connections/grants.js"; @@ -1217,7 +1219,7 @@ export function toolAccessService( safeDefault: true, }; if (existing) { - const [updated] = await db + const [updated] = await writeToolConnectionHealthChange(db, companyId, tx => tx .update(toolApplications) .set({ name: definition.applicationName, @@ -1229,7 +1231,7 @@ export function toolAccessService( updatedAt: new Date(), }) .where(eq(toolApplications.id, existing.id)) - .returning(); + .returning()); return { row: updated, created: false }; } const [created] = await db @@ -1260,11 +1262,46 @@ export function toolAccessService( quarantineNewEntries: true, }; if (existing) { - const [updated] = await db - .update(toolConnections) - .set({ + const updated = await writeToolConnectionHealthChange(db, companyId, 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, 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, @@ -1272,42 +1309,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 }; } @@ -2478,7 +2486,7 @@ export function toolAccessService( personalIdentityUserId || dedicatedAgentId ? [] : credentialSecretRefs; if (revivedConnectionPrevious) { forgetMcpHttpSessions(revivedConnectionPrevious.id); - [connectionRow] = await db + [connectionRow] = await writeToolConnectionHealthChange(db, companyId, tx => tx .update(toolConnections) .set({ name, @@ -2500,7 +2508,7 @@ export function toolAccessService( updatedAt: new Date(), }) .where(eq(toolConnections.id, revivedConnectionPrevious.id)) - .returning(); + .returning()); } else { const connectionId = randomUUID(); [connectionRow] = await db @@ -2923,6 +2931,7 @@ export function toolAccessService( connectionSetupMutationFingerprint(attemptedConnection), ); if (connectionMutationIsStillCurrent) { + await notifyToolConnectionHealth(tx, companyId); await tx .update(toolConnections) .set({ @@ -2998,7 +3007,7 @@ export function toolAccessService( // surface the restoration failure instead of returning only the // original provider error. try { - await db + await writeToolConnectionHealthChange(db, companyId, tx => tx .update(toolConnections) .set({ status: "draft", @@ -3014,7 +3023,7 @@ export function toolAccessService( eq(toolConnections.id, revivedConnectionPrevious.id), eq(toolConnections.companyId, companyId), ), - ); + )); } catch (quarantineError) { identityRollbackError = new AggregateError( [rollbackError, quarantineError], @@ -3023,9 +3032,10 @@ export function toolAccessService( } } } else if (connectionRow) { - await db + const failedConnectionId = connectionRow.id; + await writeToolConnectionHealthChange(db, companyId, tx => tx .delete(toolConnections) - .where(eq(toolConnections.id, connectionRow.id)) + .where(eq(toolConnections.id, failedConnectionId))) .catch(() => undefined); } if ( @@ -3138,6 +3148,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, companyId); 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))); @@ -3494,6 +3505,7 @@ export function toolAccessService( }, tx, ); + await notifyToolConnectionHealth(tx, companyId); const [updatedConnection] = await tx .update(toolConnections) .set({ status: "active", enabled: true, ...(input.agentInstructions !== undefined ? { agentInstructions: input.agentInstructions } : {}), updatedAt: new Date() }) @@ -4514,41 +4526,44 @@ export function toolAccessService( } 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: - binding.actorType === "user" ? binding.actorId : null, - }) - .returning(); - await ensureDefaultOrganizationGrant(row); - await syncCredentialBindings(row); - await ensureRuntimeSlot(row); + const row = await writeToolConnectionHealthChange(db, companyId, 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: + binding.actorType === "user" ? binding.actorId : null, + }) + .returning(); + await ensureDefaultOrganizationGrant(row, tx); + await syncCredentialBindings(row, [], tx); + await createToolRuntimeSlots(tx, runtimeSupervisor).ensureRuntimeSlot(row); + return row; + }); emitConnectionCreated(row, "api"); return toConnection(row); }, @@ -4633,9 +4648,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.companyId, 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); }, @@ -4685,6 +4703,7 @@ export function toolAccessService( listAppsNeedingAttention, sweepConnectionHealth, + scheduledHealth: connectionHealthSchedule(db, checkConnectionHealth), sweepGitHubConnectionContinuity, ...catalogReader, diff --git a/server/src/services/tool-access/catalog/health-schedule.ts b/server/src/services/tool-access/catalog/health-schedule.ts new file mode 100644 index 00000000000..af6ca8c92b3 --- /dev/null +++ b/server/src/services/tool-access/catalog/health-schedule.ts @@ -0,0 +1,48 @@ +import { and, eq, 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; dueAt: number }; +const HEALTH_FRESHNESS_MS = 15 * 60_000; + +export function connectionHealthSchedule(db: Db, checkHealth: ConnectionHealth["checkConnectionHealth"]) { + async function load(companyId?: string, connectionId?: string): Promise { + const rows = await db.select({ + id: toolConnections.id, + companyId: toolConnections.companyId, + checkedAt: toolConnections.healthCheckedAt, + applicationKey: toolApplications.applicationKey, + sourceTemplateKey: sql`coalesce(${toolConnections.config}->>'sourceTemplateKey', ${toolConnections.transportConfig}->>'sourceTemplateKey')`, + }).from(toolConnections).innerJoin(toolApplications, and( + eq(toolApplications.id, toolConnections.applicationId), + eq(toolApplications.companyId, toolConnections.companyId), + )).where(and( + eq(toolConnections.enabled, true), eq(toolConnections.status, "active"), + ne(toolConnections.transport, "chat_sdk"), ne(toolConnections.transport, "runtime_auth"), + ne(toolApplications.type, "paperclip_plugin"), + companyId === undefined ? undefined : eq(toolConnections.companyId, companyId), + connectionId === undefined ? undefined : eq(toolConnections.id, connectionId), + )); + const definitions = rows.map(row => ({ row, app: connectionSetupApp(row.applicationKey, row.sourceTemplateKey) })); + const settings = definitions.some(({ app }) => app?.experimentalFeatureKey) + ? await instanceSettingsService(db).getExperimental() : null; + return definitions.filter(({ app }) => !app || !settings || isConnectorFeatureEnabled(app, settings)) + .map(({ row }) => ({ id: row.id, companyId: row.companyId, + dueAt: row.checkedAt === null ? 0 : row.checkedAt.getTime() + HEALTH_FRESHNESS_MS })); + } + + async function check(id: string): Promise { + const [before] = await load(undefined, id); + if (!before || before.dueAt > Date.now()) return before ?? null; + let failure: unknown; + try { await checkHealth(id, { actorType: "system", actorId: "tool_health_sweep" }); } + catch (error) { failure = error; } + const [after] = await load(before.companyId, id); + if (failure && after && after.dueAt <= Date.now()) throw failure; + return after ?? null; + } + return { load, check }; +} diff --git a/server/src/services/tool-access/catalog/health.ts b/server/src/services/tool-access/catalog/health.ts index c008f20853e..c387560eadf 100644 --- a/server/src/services/tool-access/catalog/health.ts +++ b/server/src/services/tool-access/catalog/health.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { assertSupportedConnection } from "../connections/setup-policy.js"; import { connectionSetupUrl, connectionReconnectUrl } from "../identifiers.js"; import { createGitHubGrantContinuity } from "../github-grant-continuity.js"; @@ -46,7 +47,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.companyId, tx => tx .update(toolConnections) .set({ healthStatus: status, @@ -57,7 +58,7 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d updatedAt: now, }) .where(eq(toolConnections.id, connection.id)) - .returning(); + .returning()); if (connection.transport === "local_stdio") { await db .update(toolRuntimeSlots) diff --git a/server/src/services/tool-access/catalog/publication.ts b/server/src/services/tool-access/catalog/publication.ts index 113e227cbf5..5257cbb3bfe 100644 --- a/server/src/services/tool-access/catalog/publication.ts +++ b/server/src/services/tool-access/catalog/publication.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectorToolPolicy } from "../../connectors/tool-policy.js"; import { connectionSetupUrl, connectionReconnectUrl } from "../identifiers.js"; import { createToolRuntimeSlots } from "../runtime/slots.js"; @@ -268,7 +269,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.companyId, tx => tx .update(toolConnections) .set({ config: normalizedConfig, @@ -286,7 +287,7 @@ export function createToolCatalogPublication(db: Db, { now, audit, getConnection updatedAt: refreshedAt, }) .where(eq(toolConnections.id, connection.id)) - .returning(); + .returning()); if (connection.transport === "local_stdio") { await ensureRuntimeSlot(updatedConnection); diff --git a/server/src/services/tool-access/connections/removal.ts b/server/src/services/tool-access/connections/removal.ts index 71ab1767558..2fef00ff9ff 100644 --- a/server/src/services/tool-access/connections/removal.ts +++ b/server/src/services/tool-access/connections/removal.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { toConnection } from "./projection.js"; import { actorBinding } from "../actor.js"; import { vercelConnectResourcesFor } from "../vercel-credentials.js"; @@ -536,6 +537,7 @@ async function removeConnection( .returning({ id: connectionTokenIssuances.id }); const archived = await db.transaction(async (tx) => { + await notifyToolConnectionHealth(tx, connection.companyId); await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId))) .for("update"); diff --git a/server/src/services/tool-access/connections/setup-policy.ts b/server/src/services/tool-access/connections/setup-policy.ts index cf07e847b2f..8c4104d1a23 100644 --- a/server/src/services/tool-access/connections/setup-policy.ts +++ b/server/src/services/tool-access/connections/setup-policy.ts @@ -14,6 +14,11 @@ export function assertSupportedConnection(connection: typeof toolConnections.$in } } +export function connectionSetupApp(applicationKey: unknown, sourceTemplateKey: unknown) { + return applicationKey != null ? getAppDefinitionForApplicationKey(applicationKey) + : typeof sourceTemplateKey === "string" ? getConnectableAppDefinition(sourceTemplateKey) : null; +} + export function createConnectionSetupPolicy(db: Db) { async function assertExperimentalConnectorSetupEnabled(provider: unknown, existing = false) { const app = typeof provider === "string" ? getConnectableAppDefinition(provider) : null; @@ -36,8 +41,7 @@ export function createConnectionSetupPolicy(db: Db) { resolvedApplicationKey = application?.applicationKey; } const marker = connection.config?.sourceTemplateKey ?? connection.transportConfig?.sourceTemplateKey; - const app = resolvedApplicationKey != null ? getAppDefinitionForApplicationKey(resolvedApplicationKey) - : typeof marker === "string" ? getConnectableAppDefinition(marker) : null; + const app = connectionSetupApp(resolvedApplicationKey, marker); if (app) await assertExperimentalConnectorSetupEnabled(app.slug, connection.status !== "draft"); } return { assertExperimentalConnectorSetupEnabled, assertConnectorConnectionSetupEnabled }; diff --git a/server/src/services/tool-access/oauth/access-finalization.ts b/server/src/services/tool-access/oauth/access-finalization.ts index e27e8a67a0d..85569d99d3a 100644 --- a/server/src/services/tool-access/oauth/access-finalization.ts +++ b/server/src/services/tool-access/oauth/access-finalization.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import type { ActorInfo } from "../types.js"; import { connectionMethodForConnection } from "../connection-methods.js"; import { emitConnectionUpdated } from "../../connector-telemetry.js"; @@ -112,33 +113,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, companyId, 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 @@ -323,6 +328,7 @@ async function finalizeOAuthAccess( updatedAt: nowAt, }) .where(eq(connectionGrants.id, personalGrant.id)); + await notifyToolConnectionHealth(tx, companyId); [connection] = await tx .update(toolConnections) .set({ diff --git a/server/src/services/tool-access/oauth/cloud-callback.ts b/server/src/services/tool-access/oauth/cloud-callback.ts index 5c5a3902620..a61c3b203f3 100644 --- a/server/src/services/tool-access/oauth/cloud-callback.ts +++ b/server/src/services/tool-access/oauth/cloud-callback.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { managedConnectorProfile } from "./client-settings.js"; import type { ActorInfo } from "../types.js"; import { oauthConfig, normalizeOauthScopes } from "./metadata.js"; @@ -355,6 +356,7 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { scopes: [...profile.scopes], }, }; + await notifyToolConnectionHealth(tx, connection.companyId); [connection] = await tx .update(toolConnections) .set({ diff --git a/server/src/services/tool-access/oauth/connection-refresh.ts b/server/src/services/tool-access/oauth/connection-refresh.ts index 959d0619c00..fd3ae9b7aff 100644 --- a/server/src/services/tool-access/oauth/connection-refresh.ts +++ b/server/src/services/tool-access/oauth/connection-refresh.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import type { ActorInfo } from "../types.js"; import { createOAuthClientProtocol, storedOAuthTokenEndpointAuthMethod } from "./client.js"; import { oauthConfig, normalizeOauthScopes } from "./metadata.js"; @@ -59,7 +60,7 @@ const oauthRefreshFlights = new Map>(); reauthorizationRequiredAt: now().toISOString(), }, }; - const [updated] = await db + const [updated] = await writeToolConnectionHealthChange(db, connection.companyId, tx => tx .update(toolConnections) .set({ status: "draft", @@ -87,7 +88,7 @@ const oauthRefreshFlights = new Map>(); )`, ), ) - .returning(); + .returning()); if (updated) { await syncCredentialBindings(updated); emitConnectionUpdated(updated, connection, "credential_refresh"); diff --git a/server/src/services/tool-access/oauth/direct-callback.ts b/server/src/services/tool-access/oauth/direct-callback.ts index a95f49c8a21..cd9b0723def 100644 --- a/server/src/services/tool-access/oauth/direct-callback.ts +++ b/server/src/services/tool-access/oauth/direct-callback.ts @@ -1,3 +1,4 @@ +import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { credentialScope } from "./client-settings.js"; import { withoutOAuthRefreshLease } from "./refresh-leases.js"; import type { ActorInfo } from "../types.js"; @@ -331,6 +332,7 @@ export function createDirectOAuthCallback(db: Db, dependencies: Dependencies) { }, }, }; + await notifyToolConnectionHealth(tx, connection.companyId); const [updatedConnection] = await tx .update(toolConnections) .set({ @@ -610,6 +612,7 @@ export function createDirectOAuthCallback(db: Db, dependencies: Dependencies) { }, }, }; + await notifyToolConnectionHealth(tx, connection.companyId); const [updatedConnection] = await tx .update(toolConnections) .set({ diff --git a/server/src/services/tool-access/oauth/grant-refresh.ts b/server/src/services/tool-access/oauth/grant-refresh.ts index 4a3748f6d83..6a1657d8482 100644 --- a/server/src/services/tool-access/oauth/grant-refresh.ts +++ b/server/src/services/tool-access/oauth/grant-refresh.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import type { ActorInfo } from "../types.js"; import { createOAuthClientProtocol, storedOAuthTokenEndpointAuthMethod } from "./client.js"; import { oauthConfig, normalizeOauthScopes, resolveGrantedOauthScopes } from "./metadata.js"; @@ -393,7 +394,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo connection.id, connection.companyId, ); - const [reauthorizationRequired] = await db + const [reauthorizationRequired] = await writeToolConnectionHealthChange(db, latestConnection.companyId, tx => tx .update(toolConnections) .set({ status: "draft", @@ -417,7 +418,7 @@ export function createOAuthGrantRefresh(db: Db, { now, store, leases, currentClo eq(toolConnections.companyId, latestConnection.companyId), ), ) - .returning(); + .returning()); await syncCredentialBindings(reauthorizationRequired); emitConnectionUpdated( reauthorizationRequired, diff --git a/server/src/services/tool-access/oauth/vercel-callback.ts b/server/src/services/tool-access/oauth/vercel-callback.ts index 87164abd425..f8391eda77c 100644 --- a/server/src/services/tool-access/oauth/vercel-callback.ts +++ b/server/src/services/tool-access/oauth/vercel-callback.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { vercelConnectHttpError, vercelCredentialFor, vercelConnectResourcesFor } from "../vercel-credentials.js"; import type { ActorInfo } from "../types.js"; import { connectionMethodForConnection, projectedConnectionHeaders } from "../connection-methods.js"; @@ -183,7 +184,7 @@ export function createVercelOAuthCallback(db: Db, dependencies: Dependencies) { status: connection.status, enabled: connection.enabled, }; - [connection] = await db + [connection] = await writeToolConnectionHealthChange(db, connection.companyId, tx => tx .update(toolConnections) .set({ status: "active", @@ -198,7 +199,7 @@ export function createVercelOAuthCallback(db: Db, dependencies: Dependencies) { eq(toolConnections.companyId, connection.companyId), ), ) - .returning(); + .returning()); emitConnectionUpdated( connection, preActivationLifecycle, diff --git a/server/src/services/tool-connection-health-notifications.ts b/server/src/services/tool-connection-health-notifications.ts new file mode 100644 index 00000000000..163023d674a --- /dev/null +++ b/server/src/services/tool-connection-health-notifications.ts @@ -0,0 +1,12 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export const notifyToolConnectionHealth = (transaction: object, companyId: string) => + notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolConnectionHealth, companyId); + +export function writeToolConnectionHealthChange(db: Db, companyId: string, write: (transaction: Db) => Promise): Promise { + return db.transaction(async tx => { + await notifyToolConnectionHealth(tx, companyId); + return write(tx as unknown as Db); + }); +} diff --git a/server/src/services/tool-connection-health-work.ts b/server/src/services/tool-connection-health-work.ts new file mode 100644 index 00000000000..ab28cff9067 --- /dev/null +++ b/server/src/services/tool-connection-health-work.ts @@ -0,0 +1,73 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { ConnectionHealthDeadline } from "./tool-access/catalog/health-schedule.js"; + +export function registerToolConnectionHealthWork( + coordinator: ReturnType, + service: { + load(companyId?: string): Promise; + check(id: string): Promise; + }, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const deadlines = new Map(); + const failures = new Map(); + const retryMs = 30_000; + let earliest = 0; + function request(hints?: ReadonlySet) { + for (const hint of hints ?? ["*"]) dirty.add(hint); + } + function next() { + let result: number | null = dirty.size ? earliest : null; + for (const entry of deadlines.values()) { + const at = Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0); + 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 = new Set(); + 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; } + try { + const rows = await service.load(scope === "*" ? undefined : scope); + for (const entry of deadlines.values()) if (scope === "*" || entry.companyId === scope) deadlines.delete(entry.id); + for (const entry of rows) deadlines.set(entry.id, entry); + for (const id of failures.keys()) { + const current = deadlines.get(id); + if (!current || current.dueAt > Date.now()) failures.delete(id); + } + } catch (error) { dirty.add(scope); unavailable.add(scope); errors.push(error); } + } + const due = [...deadlines.values()].filter(entry => !unavailable.has("*") && !unavailable.has(entry.companyId) + && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) + .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)).slice(0, 25); + for (const entry of due) { + 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 (errors.length) throw new AggregateError(errors, "Tool connection health checks could not settle"); + }, + hasPending: async () => dirty.size > 0, + nextRunAt: next, + }); +} diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 9f15e0a3be0..6a0b8fb60bb 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,3 +1,4 @@ +import { writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { writeApprovalChange } from "./approval-work-notifications.js"; import { toolReviewProjectionService } from "./tool-review-projection.js"; import { toolReviewInterruptionService } from "./tool-review-interruption.js"; @@ -3804,7 +3805,7 @@ export function createToolGatewayService( message: string | null, ) { const now = new Date(); - await db + await writeToolConnectionHealthChange(db, connection.companyId, tx => tx .update(toolConnections) .set({ healthStatus: status, @@ -3814,7 +3815,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(