From f45969454c86ce14a501ac7fd8357b9993c03cdf Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 02:44:00 -0500 Subject: [PATCH] fix(server): wake chat receiver discovery from configuration and lease deadlines Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 19 ++ ...-25-chat-receiver-discovery-sleep-audit.md | 91 +++++++- .../app-private-hostname-gate.test.ts | 28 ++- .../chat-channels.integration.test.ts | 100 ++++++++- .../src/__tests__/chat-receiver-work.test.ts | 206 ++++++++++++++++++ .../photon/channel.integration.test.ts | 41 +++- server/src/app.ts | 12 +- server/src/services/chat-channels.ts | 35 ++- .../command-work-notifications.ts | 18 -- .../configuration-work-notifications.ts | 22 ++ .../src/services/chat-channels/credentials.ts | 6 +- .../chat-channels/discord/gateway-events.ts | 8 +- .../chat-channels/imessage-photon/events.ts | 2 + .../chat-channels/interactions/callbacks.ts | 4 +- .../chat-channels/provider-effects.ts | 4 +- .../chat-channels/publications/failures.ts | 4 +- .../chat-channels/resource-reconciliation.ts | 6 +- .../chat-channels/runtime/ownership.ts | 12 +- .../chat-channels/runtime/receiver-policy.ts | 13 ++ .../runtime/receiver-schedule.ts | 53 +++++ .../runtime/receiver-work-notifications.ts | 24 ++ .../chat-channels/runtime/receiver-work.ts | 87 ++++++++ .../src/services/chat-work-notifications.ts | 4 +- server/src/services/company-deletion.ts | 4 +- .../services/delivery-work-notifications.ts | 1 + .../services/tool-access/catalog/health.ts | 4 +- 26 files changed, 738 insertions(+), 70 deletions(-) create mode 100644 server/src/__tests__/chat-receiver-work.test.ts delete mode 100644 server/src/services/chat-channels/command-work-notifications.ts create mode 100644 server/src/services/chat-channels/configuration-work-notifications.ts create mode 100644 server/src/services/chat-channels/runtime/receiver-policy.ts create mode 100644 server/src/services/chat-channels/runtime/receiver-schedule.ts create mode 100644 server/src/services/chat-channels/runtime/receiver-work-notifications.ts create mode 100644 server/src/services/chat-channels/runtime/receiver-work.ts 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 924d1c21203..46000b314e2 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1192,3 +1192,22 @@ loops or changing domain backoff. All 152 selected command/coordinator/health, producer/Slack, delivery/standby/startup cases and server types/build pass. Publish for fresh CI/review. The full inventory, #17490 CI failure, and stack/master reconciliation remain open; no merge authority is implied. + + +## Slice 20 receiver implementation + +Receiver discovery now uses scoped committed configuration changes, startup +reconstruction, foreign-lease deadlines, and listener-stop completion. Voice and +all other maintenance reasons remain separate. See +[the receiver audit](2026-10-25-chat-receiver-discovery-sleep-audit.md). +A failed-scope deadline regression and a real concurrent listener-stop race were +reproduced and repaired. All 218 selected receiver/startup, Discord, Photon, +command/delivery and standby cases pass, with workspace typechecking passing. +Workspace build, publication and fresh remote gates remain open. This is not +completion of the slice or the full stack. Master is refreshed at 0d73505098; +final stack/master reconciliation remains required. + +Parent Discord #17557 at d3187f76a4 and company cleanup #17561 at e8febb30eb now +have no failed or pending checks (Paperclip Review skipped). No merge was made. +The older #17490 E2E cancellation failure remains unresolved. The entire original +inventory stays in scope, with voice the next separate reason after receivers. diff --git a/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md b/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md index 2b758fc030d..c6f6af5fdbf 100644 --- a/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md @@ -3,7 +3,7 @@ Parent: Discord command registration #17557, head d3187f76a4. Master refreshed during this continuation: e6ec781a1a. Branch: codex/sleep-20-chat-runtime-discovery. -Status: owner audit; no implementation or passing migration claim yet. +Status: implementation and focused lifecycle verification; publication and final acceptance pending. ## One reason for this PR @@ -142,3 +142,92 @@ failed checks, passing security, and fresh Greptile 5/5. Its three reviewed find verified, and their threads resolved. This is acceptance of that current head, not a claim that the full stack is complete or reconciled with moving master. Company cleanup and Discord still have live current-head checks. + + +## Implementation and verification update + +The aggregate chat lane now calls voice reconciliation only. Receiver discovery +registers with the existing delivery coordinator. The receiver owner loads UUID +pages of 100, keeps owned identities with no discovery deadline, and processes +25 due receivers per fair company batch. Configuration hints refresh only their +endpoint, connection, or company. A foreign owner schedules its saved lease +expiry; a changed lease is reread before another attempt. Failures retain scoped +30-second exponential backoff capped at five minutes. Dirty scopes suppress their +cached deadlines until refresh succeeds; otherwise an old takeover deadline could +cause repeated useless dispatch while the required refresh is in backoff. + +The generic configuration notifier replaces the misleading command-only helper +name. Existing producers now notify both relevant owners. Shared delivery code +still receives only a queue topic and scope hint. Provider selection and lease +keys remain in the chat receiver owner. No schema or new per-worker table exists. + +Audited producer groups: + +| Owner | Receiver-relevant changes and integration | +| --- | --- | +| Configuration and setup probes | Existing connection helper covers configure, rotation, resume, pause, removal and completed setup before transaction writes | +| Credential persistence and removal | Configuration notifier before reference changes, including Photon | +| Inbound activation and resource reconciliation | Endpoint/connection notifier before status and health transitions | +| Publication, interaction and provider-effect failures | Endpoint configuration notifier before attention/failed-health writes | +| Discord Gateway lifecycle | Receiver hints before recovery/failure transitions; metadata-only writes do not create discovery work | +| Photon transport failure and runtime recovery | Failure callback now signals before attention/disable; existing assert-owned recovery uses the shared endpoint helper | +| Generic tool update/removal, manual health and access revocation | Existing chat connection helper, broadened configuration dispatch, and SDK transport filter | +| Company deletion | Company hint precedes endpoint removal in the owning deletion transaction | +| Ownership loss / invalidation | Local completion callback after listener stop and lease release; suppressed during shutdown | + +The additional direct-writer audit found publication-resolution, publication +success, provider-effect success, inbound filtering, and Discord root-thread +writes that only update timestamps. Endpoint display/reach preferences do not +change receiver eligibility or the runtime generation. Photon configuration +invalidates the runtime, writes allocation metadata, and proceeds through the +credential/configuration owner; its allocation metadata is not an independent +receiver identity. Slack/GitHub-only setup writers do not create leased receiver +work. Capability projection in Discord reconciliation retains its existing owner. + +A real token-loss regression found a concurrent configuration wake could acquire +another lease while the old listener was still stopping. The ownership map now +retains the stopping owner until listener shutdown and lease release complete. +Acquisition joins that owner's existing stop promise. Current-authority checks +still reject stopping owners. The test uses real timing and holds listener stop; +an earlier fake-clock observation returned before async SQL completed and did not +prove absence of overlap. The strengthened test failed before the ownership fix. + +Current passing evidence: 32 receiver/startup cases, 90 selected Discord +integration cases, all 20 Photon channel cases, and server typechecking. The +Discord selection includes configuration, credential rotation, pause/resume, +company deletion, token-loss stop ordering, and existing takeover/admission fences. +The dedicated real-DB worker takeover test also passes, as do 75 command/delivery/ +standby consumer cases and full workspace typechecking. This is 218 unique selected +passing cases. Workspace build and remote review/check gates remain pending. +The full local test matrix is not claimed: embedded PostgreSQL is unavailable +on this host, so these database suites use validated disposable external URLs. +Remote CI must run the full matrix before acceptance. + +Recorded failures remain part of acceptance: initial receiver teardown omitted +agent deletion; initial Discord pause used incomplete onboarding; failed-scope +backoff initially selected the blocked cached takeover deadline; the real-timing +stop race reproduced overlapping listeners. The first two were fixture defects; +the latter two required implementation repairs. An initial backoff test expected +zero idle holds during failed intent recovery; the correct contract retains the +intent hold until reconciliation succeeds, so the test now checks the retry +deadline and SQL silence rather than weakening that hold. + +## Writer and sleep boundaries + +Notifications cover the active database owner and its transaction-sharing clients. +Warm standby does not run discovery; activation reconstructs its pending startup +inventory. An already known foreign receiver remains recoverable from lease expiry, +with real token CAS and callback fences. Arbitrary new configuration written by +an independent process or direct SQL is not discovered by process-local hints; +that deployment needs an explicit wake integration. This follows the existing +single active writer contract and must not be described as multi-writer support. +The manual reconciliation method remains available for explicit callers. + +Active sockets and lease renewals remain required work, and no Cloud admission +exemption is added. Voice, Teams, publication maintenance, milestones and other +inventory reasons remain separate PRs. The full stack is not complete. + +Master was fetched again at 0d73505098. Company prerequisite #17561 at e8febb30eb +and Discord parent #17557 at d3187f76a4 now have no failed or pending checks; +Paperclip Review is skipped. Fresh review-content reconciliation and final stack +rebasing/verification remain distinct from these check results. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 0dbde42a5a9..d715145e5fa 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -25,7 +25,7 @@ describe("createChatReconciliationCoordinator", () => { processFailedGitHubWebhookDeliveries, processPendingDeliveries, flushPublications, - reconcileProviderRuntimes: async () => undefined, + reconcileVoiceSessions: async () => undefined, projectRunMilestones: async () => 0, processPendingSlackFileUploadReceipts: async () => undefined, processPendingSlackSessionSyncs: async () => undefined, @@ -74,6 +74,14 @@ describe("createChatReconciliationCoordinator", () => { }, ); + it("keeps receiver inventory out of periodic chat maintenance while preserving voice", () => { + const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); + const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals")); + expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); + expect(maintenance).toContain("chatChannels.voice.reconcile()"); + expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); + }); + it("wires GitHub recovery into its independent lane", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); expect(source).toContain( @@ -135,7 +143,7 @@ describe("createChatReconciliationCoordinator", () => { const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones, - reconcileProviderRuntimes: async () => undefined, + reconcileVoiceSessions: async () => undefined, processPendingDeliveries: async () => undefined, flushPublications, processPendingSlackFileUploadReceipts: async () => undefined, @@ -198,7 +206,7 @@ describe("createChatReconciliationCoordinator", () => { const coordinator = createChatReconciliationCoordinator({ projectRunMilestones, flushPublications, - reconcileProviderRuntimes: async () => undefined, + reconcileVoiceSessions: async () => undefined, processPendingDeliveries: async () => undefined, processPendingSlackFileUploadReceipts: async () => undefined, processPendingSlackSessionSyncs: async () => undefined, @@ -238,7 +246,7 @@ describe("createChatReconciliationCoordinator", () => { const slackStatusReleased = new Promise((resolve) => { releaseSlackStatus = resolve; }); - const reconcileProviderRuntimes = vi.fn(async () => undefined); + const reconcileVoiceSessions = vi.fn(async () => undefined); const processPendingDeliveries = vi.fn(async () => deliveryReleased); const flushPublications = vi.fn(async () => undefined); const processPendingSlackFileUploadReceipts = vi.fn(async () => undefined); @@ -248,7 +256,7 @@ describe("createChatReconciliationCoordinator", () => { const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones: async () => 0, - reconcileProviderRuntimes, + reconcileVoiceSessions, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, @@ -258,7 +266,7 @@ describe("createChatReconciliationCoordinator", () => { coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(1); + expect(reconcileVoiceSessions).toHaveBeenCalledTimes(1); expect(processPendingDeliveries).toHaveBeenCalledTimes(1); expect(flushPublications).toHaveBeenCalledTimes(1); expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(1); @@ -266,7 +274,7 @@ describe("createChatReconciliationCoordinator", () => { }); coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(2); + expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2); expect(flushPublications).toHaveBeenCalledTimes(2); expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2); }); @@ -293,14 +301,14 @@ describe("createChatReconciliationCoordinator", () => { await receiptReleased; if (rejects) throw lookupError; }); - const reconcileProviderRuntimes = vi.fn(async () => undefined); + const reconcileVoiceSessions = vi.fn(async () => undefined); const processPendingDeliveries = vi.fn(async () => undefined); const flushPublications = vi.fn(async () => undefined); const processPendingSlackSessionSyncs = vi.fn(async () => undefined); const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones: async () => 0, - reconcileProviderRuntimes, + reconcileVoiceSessions, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, @@ -318,7 +326,7 @@ describe("createChatReconciliationCoordinator", () => { }); coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(2); + expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2); expect(processPendingDeliveries).toHaveBeenCalledTimes(2); expect(flushPublications).toHaveBeenCalledTimes(2); expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(2); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 89764384250..65c7d4a44fd 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,4 @@ +import { notifyChatEndpointConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; import { configureAgentLifecycle } from "../services/agent-lifecycle.js"; import { deleteCompany } from "../services/company-deletion.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -6644,6 +6645,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { async function registrationFixture( mode: "success" | "unavailable" | "unknown" = "success", + receiverOptions: Partial> = {}, ) { const fixture = await seedCompany(); const applicationId = uniqueDiscordApplicationId(); @@ -6701,7 +6703,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return Response.json(command); }, ) as unknown as typeof globalThis.fetch; - const context = createService(new FakeChatSdkRuntime(), providerFetch); + const context = createService(new FakeChatSdkRuntime(), providerFetch, receiverOptions); const endpoint = await context.service.create( fixture.companyId, { @@ -6754,6 +6756,102 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("drives receiver discovery through real configure, rotation, pause, and resume producers", async () => { + const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 600_000, discordGatewayLeaseRenewalIntervalMs: 300_000 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const quiet = async () => { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(20_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + }; + try { + await f.service.registerReceiverWork(coordinator, () => true).ready; + await settle(); await quiet(); + await f.configure(); await settle(); + const first = f.runtime.get(f.endpoint.id); expect(first).not.toBeNull(); + await quiet(); + await f.service.configure(f.endpoint.id, { action: "reconnect", credentials: { ...f.credentials, botToken: "rotated-receiver-token" } }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).not.toBe(first); expect(first!.shutdown).toHaveBeenCalledOnce(); + // Pause requires completed onboarding; configure/reconnect intentionally + // leave this fixture waiting for its first test conversation. + const configured = await f.service.get(f.endpoint.id); + await db.update(chatEndpoints).set({ status: "active", setup: { ...configured.setup, step: "complete" } }) + .where(eq(chatEndpoints.id, f.endpoint.id)); + await f.service.configure(f.endpoint.id, { action: "pause" }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).toBeNull(); await quiet(); + await f.service.configure(f.endpoint.id, { action: "resume" }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).not.toBeNull(); await quiet(); + await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user"); + await settle(); expect(f.runtime.get(f.endpoint.id)).toBeNull(); await quiet(); + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); await f.service.shutdown(); vi.useRealTimers(); + await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + + it("takes over a persisted foreign receiver lease at its deadline through the worker", async () => { + const f = await registrationFixture(); + await f.configure(); await f.service.shutdown(); + const token = randomUUID(), expiresAt = new Date(Date.now() + 500); + await db.insert(chatEndpointLeases).values({ companyId: f.fixture.companyId, endpointId: f.endpoint.id, + leaseKey: "discord_gateway_runtime", token, expiresAt }); + const standby = createService(new FakeChatSdkRuntime(), f.providerFetch); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await standby.service.registerReceiverWork(coordinator, () => true).ready; + expect(standby.runtime.get(f.endpoint.id)).toBeNull(); + expect(coordinator.nextWakeAt()).toBeGreaterThanOrEqual(expiresAt.getTime()); + await vi.waitFor(() => expect(standby.runtime.get(f.endpoint.id)).not.toBeNull(), { timeout: 5000 }); + const [lease] = await db.select().from(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, f.endpoint.id)); + expect(lease.token).not.toBe(token); + expect(lease.expiresAt.getTime()).toBeGreaterThan(expiresAt.getTime()); + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); await retireRegistrationFixture(standby.service, f.endpoint.id); + } + }); + + it("restarts a receiver after token loss only after the old listener has stopped", async () => { + const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 60_000, discordGatewayLeaseRenewalIntervalMs: 50 }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + let release!: () => void; + const stopped = new Promise(resolve => { release = resolve; }); + try { + await f.configure(); + await f.service.registerReceiverWork(coordinator, () => true).ready; + const old = f.runtime.get(f.endpoint.id)!; + old.shutdown.mockImplementation(async () => { await stopped; }); + await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, f.endpoint.id)); + await vi.waitFor(() => expect(old.shutdown).toHaveBeenCalledOnce()); + await db.transaction(tx => notifyChatEndpointConfiguration(tx, f.endpoint)); + await new Promise(resolve => setTimeout(resolve, 1500)); + expect(f.runtime.get(f.endpoint.id)).toBeNull(); + release(); + await vi.waitFor(async () => { + expect(f.runtime.get(f.endpoint.id)).not.toBeNull(); + expect(f.runtime.get(f.endpoint.id)).not.toBe(old); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + expect(onError).not.toHaveBeenCalled(); + } finally { + release(); await coordinator.stop(); vi.useRealTimers(); + await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + it("learns the first configured command and cancels its deadline when the endpoint is removed", async () => { const f = await registrationFixture(); vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); diff --git a/server/src/__tests__/chat-receiver-work.test.ts b/server/src/__tests__/chat-receiver-work.test.ts new file mode 100644 index 00000000000..b895a4f9a50 --- /dev/null +++ b/server/src/__tests__/chat-receiver-work.test.ts @@ -0,0 +1,206 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { agents, chatEndpoints, chatEndpointLeases, companies, createDb, toolApplications, toolConnections } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyDeliveryWork, DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { chatReceiverSchedule } from "../services/chat-channels/runtime/receiver-schedule.js"; +import { registerChatReceiverWork } from "../services/chat-channels/runtime/receiver-work.js"; +import { notifyChatEndpointConfiguration, notifyChatConnectionConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RECEIVER_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("chat receiver discovery", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, agentId: string, applicationId: string; + let coordinator: ReturnType | undefined; + let enabled = true, primary = true; + const errors: unknown[] = []; + const reconcile = vi.fn(async (_id: string): Promise<"owned" | "foreign"> => "owned"), stop = vi.fn(async (_id: string) => {}); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("chat-receivers-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 60_000); + beforeEach(async () => { + companyId = randomUUID(); agentId = randomUUID(); applicationId = randomUUID(); enabled = true; primary = true; errors.length = 0; + reconcile.mockReset().mockResolvedValue("owned"); stop.mockReset().mockResolvedValue(undefined); + await db.insert(companies).values({ id: companyId, name: "Receivers", issuePrefix: `R${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Receiver", adapterType: "codex_local" }); + await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Receiver", type: "chat" }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select 1`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); + await db.delete(agents).where(eq(agents.companyId, companyId)); + await db.delete(companies).where(eq(companies.id, companyId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function endpoint(patch: Partial = {}, writer = db) { + const id = patch.id ?? randomUUID(), connectionId = randomUUID(); + return writer.transaction(async tx => { + await notifyChatEndpointConfiguration(tx, { id, provider: patch.provider ?? "discord" }); + await tx.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: id, uid: id, + transport: "chat_sdk", connectionPurpose: "channel", enabled: true, status: "active" }); + return (await tx.insert(chatEndpoints).values({ id, companyId, connectionId, assignedAgentId: agentId, + publicId: id, provider: "discord", status: "active", ...patch }).returning())[0]; + }); + } + async function start(service = chatReceiverSchedule(db, { reconcile, stop })) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled && primary, canReconcile: () => primary, onError: error => errors.push(error) }); + const worker = registerChatReceiverWork(coordinator, service, () => enabled && primary); + await worker.ready; + return { worker, service }; + } + async function settled(expected: number | null = null) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBe(expected); + }, { timeout: 5000 }); + } + async function quiet(ms = 120_000) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + try { + await vi.advanceTimersByTimeAsync(ms); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); execute.mockRestore(); } + } + it("does no recurring SQL after empty reconstruction", async () => { await start(); await settled(); await quiet(); }); + it("leaves disabled, archived, and webhook-only endpoints quiet", async () => { + await endpoint({ status: "paused" }); await endpoint({ status: "archived" }); await endpoint({ provider: "slack" }); + const disabled = await endpoint(); await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, disabled.connectionId)); + const failed = await endpoint({ status: "attention" }); await db.update(toolConnections).set({ healthStatus: "failed" }).where(eq(toolConnections.id, failed.connectionId)); + await start(); await settled(); await quiet(); expect(reconcile).not.toHaveBeenCalled(); + }); + it.each(["discord", "imessage-photon"] as const)("wakes the first %s receiver and retains its identity until removal", async provider => { + await start(); await settled(); await quiet(); + const row = await endpoint({ provider }); await settled(); + expect(reconcile).toHaveBeenCalledWith(row.id); await quiet(); + await db.transaction(async tx => { await notifyChatConnectionConfiguration(tx, row.connectionId); await tx.delete(chatEndpoints).where(eq(chatEndpoints.id, row.id)); }); + await settled(); expect(stop).toHaveBeenCalledWith(row.id); await quiet(); + }); + it("waits for outer commit and discards rolled-back configuration", async () => { + await start(); await settled(); + await expect(db.transaction(async tx => { await endpoint({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(60_000); expect(reconcile).not.toHaveBeenCalled(); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await settled(); await quiet(); + let id = ""; + await db.transaction(async tx => { id = (await endpoint({}, tx as unknown as typeof db)).id; await vi.advanceTimersByTimeAsync(60_000); expect(reconcile).not.toHaveBeenCalled(); }); + await settled(); expect(reconcile).toHaveBeenCalledWith(id); + }); + it("preserves the first receiver written during drain without passive SQL", async () => { + await start(); await settled(); enabled = false; const row = await endpoint(); + await quiet(60_000); expect(reconcile).not.toHaveBeenCalled(); + enabled = true; await settled(); expect(reconcile).toHaveBeenCalledWith(row.id); + }); + it("reconstructs after standby becomes primary without restarting the worker", async () => { + await endpoint(); primary = false; await start(); await quiet(60_000); + primary = true; await settled(); expect(reconcile).toHaveBeenCalledOnce(); await quiet(); + }); + it("waits for a foreign lease, observes its extension, and takes over after release", async () => { + const row = await endpoint(); let owned = false; + reconcile.mockImplementation(async () => owned ? "owned" : "foreign"); + const expiry = Date.now() + 60_000; + await db.insert(chatEndpointLeases).values({ companyId, endpointId: row.id, leaseKey: "discord_gateway_runtime", token: randomUUID(), expiresAt: new Date(expiry) }); + await start(); await settled(expiry); await quiet(10_000); expect(reconcile).toHaveBeenCalledOnce(); + const renewed = expiry + 60_000; + await db.update(chatEndpointLeases).set({ expiresAt: new Date(renewed) }).where(eq(chatEndpointLeases.endpointId, row.id)); + await vi.advanceTimersByTimeAsync(expiry - Date.now()); await settled(renewed); expect(reconcile).toHaveBeenCalledTimes(2); + await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, row.id)); owned = true; + await vi.advanceTimersByTimeAsync(renewed - Date.now()); await settled(); expect(reconcile).toHaveBeenCalledTimes(3); await quiet(); + }); + it("wakes after a local receiver stops and then returns to quiet", async () => { + const row = await endpoint(); const { worker } = await start(); await settled(); await quiet(); + worker.changed(row.id); await settled(); expect(reconcile).toHaveBeenCalledTimes(2); await quiet(); + }); + it("retains a producer arriving during its scoped refresh", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(); let inserted = false; + service.load = async (...args) => { const rows = await original(...args); if (!inserted) { inserted = true; await endpoint(); } return rows; }; + await endpoint(); await settled(); expect(reconcile).toHaveBeenCalledTimes(2); await quiet(); + }); + it("cancels all cached owned receivers after a company deletion hint", async () => { + await endpoint(); await endpoint(); await start(); await settled(); + await db.transaction(async tx => { await notifyChatCompanyConfiguration(tx, companyId); await tx.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); }); + await settled(); expect(stop).toHaveBeenCalledTimes(2); await quiet(); + }); + it("retains transient load failures and does not repeatedly scan during their backoff", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(); let broken = true; + service.load = async (...args) => { if (broken) throw new Error("load unavailable"); return original(...args); }; + await endpoint(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }); + await quiet(10_000); broken = false; await settled(); expect(reconcile).toHaveBeenCalledOnce(); errors.length = 0; + }); + it("does not schedule blocked takeover attempts before a failed scope refresh can retry", async () => { + const row = await endpoint(); + reconcile.mockResolvedValue("foreign"); + const expiry = Date.now() + 10_000; + await db.insert(chatEndpointLeases).values({ companyId, endpointId: row.id, leaseKey: "discord_gateway_runtime", token: randomUUID(), expiresAt: new Date(expiry) }); + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(expiry); + service.load = async () => { throw new Error("scoped inventory unavailable"); }; + await db.transaction(tx => notifyChatConnectionConfiguration(tx, row.connectionId)); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(errors).toHaveLength(1)); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now() + 20_000); + await quiet(10_000); + service.load = original; reconcile.mockResolvedValue("owned"); + await settled(); errors.length = 0; await quiet(); + }); + it("recovers malformed scopes without losing a subsequent valid producer", async () => { + await start(); await settled(); + await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "unknown:old")); + await endpoint(); + await settled(); expect(reconcile).toHaveBeenCalledOnce(); expect(errors.length).toBeGreaterThan(0); errors.length = 0; await quiet(); + }); + it("loads inventory in bounded pages and drains a backlog despite one failed receiver", async () => { + for (let i = 0; i < 105; i++) await endpoint(); + const service = chatReceiverSchedule(db, { reconcile, stop }); + const reads = vi.spyOn(db, "select"); const inventory = await service.load({ kind: "company", id: companyId }); + expect(inventory).toHaveLength(105); expect(reads).toHaveBeenCalledTimes(2); reads.mockRestore(); + const failedId = inventory[0].id; + reconcile.mockImplementation(async id => { if (id === failedId) throw new Error("bad credential"); return "owned"; }); + await start(service); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(new Set(reconcile.mock.calls.map(([id]) => id)).size).toBe(105); }, { timeout: 5000 }); + expect(reconcile.mock.calls.filter(([id]) => id === failedId)).toHaveLength(1); + reconcile.mockResolvedValue("owned"); await settled(); errors.length = 0; await quiet(); + }); + it("includes a second company in the first batch despite a larger earlier backlog", async () => { + for (let i = 1; i <= 30; i++) await endpoint({ id: `00000000-0000-4000-8000-${String(i).padStart(12, "0")}` }); + const firstCompany = companyId, firstAgent = agentId, firstApplication = applicationId; + const secondCompany = randomUUID(), secondAgent = randomUUID(), secondApplication = randomUUID(); + await db.insert(companies).values({ id: secondCompany, name: "Other receiver company", issuePrefix: `R${secondCompany.slice(0, 7)}` }); + await db.insert(agents).values({ id: secondAgent, companyId: secondCompany, name: "Receiver", adapterType: "codex_local" }); + await db.insert(toolApplications).values({ id: secondApplication, companyId: secondCompany, name: "Receiver", type: "chat" }); + try { + companyId = secondCompany; agentId = secondAgent; applicationId = secondApplication; + const other = await endpoint({ id: "ffffffff-ffff-4fff-8fff-ffffffffffff" }); + companyId = firstCompany; agentId = firstAgent; applicationId = firstApplication; + await start(); + expect(reconcile.mock.calls.slice(0, 25).map(([id]) => id)).toContain(other.id); + await settled(); expect(reconcile).toHaveBeenCalledTimes(31); await quiet(); + } finally { + companyId = firstCompany; agentId = firstAgent; applicationId = firstApplication; + await coordinator?.stop(); coordinator = undefined; + await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, secondCompany)); + await db.delete(agents).where(eq(agents.companyId, secondCompany)); + await db.delete(companies).where(eq(companies.id, secondCompany)); + } + }); + it("joins admitted runtime initialization during shutdown", async () => { + await start(); await settled(); let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + reconcile.mockImplementation(async () => { await gate; return "owned"; }); + await endpoint(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(reconcile).toHaveBeenCalledOnce(); }); + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(); await stopping; coordinator = undefined; errors.length = 0; + await quiet(); expect(reconcile).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/__tests__/photon/channel.integration.test.ts b/server/src/__tests__/photon/channel.integration.test.ts index 2c008782663..35e9942d192 100644 --- a/server/src/__tests__/photon/channel.integration.test.ts +++ b/server/src/__tests__/photon/channel.integration.test.ts @@ -1,3 +1,5 @@ +import { createDeliveryWorkCoordinator } from "../../services/delivery-work-coordinator.js"; +import { resolveExternalTestDatabaseUrl } from "../helpers/external-test-database.js"; import { Readable } from "node:stream"; import type { StorageService } from "../../storage/types.js"; import { createHash, randomUUID } from "node:crypto"; @@ -75,15 +77,17 @@ function idleStream(): TypedEventStream { ); } describe("iMessage Photon channel control plane", () => { - let database: Awaited>; + let database: Awaited> | undefined; let db: ReturnType; let secrets: string; const oldKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; const services: ChatChannelService[] = []; + const coordinators: ReturnType[] = []; const companyIds: string[] = []; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-photon-"); - db = createDb(database.connectionString); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PHOTON_TEST_DATABASE_URL"); + if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-photon-"); + db = createDb(external ?? database!.connectionString); secrets = await mkdtemp(path.join(tmpdir(), "photon-secrets-")); process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = path.join( secrets, @@ -91,6 +95,7 @@ describe("iMessage Photon channel control plane", () => { ); }, 30_000); afterEach(async () => { + await Promise.all(coordinators.splice(0).map(coordinator => coordinator.stop())); await Promise.all(services.splice(0).map((service) => service.shutdown())); for (const id of companyIds.splice(0)) await db @@ -100,6 +105,7 @@ describe("iMessage Photon channel control plane", () => { vi.restoreAllMocks(); }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); if (secrets) await rm(secrets, { recursive: true, force: true }); if (oldKey === undefined) @@ -364,8 +370,33 @@ describe("iMessage Photon channel control plane", () => { }, }; } + it("recovers a Photon transport loss through the receiver worker and cancels fatal recovery", async () => { + const t = await setup(); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + coordinators.push(coordinator); + await t.service.registerReceiverWork(coordinator, () => true).ready; + const first = t.service.runtime.get(t.endpoint.id); + expect(first).not.toBeNull(); + await t.callbacks().onPhotonFailure!(new Error("transport interrupted")); + await vi.waitFor(() => { + const current = t.service.runtime.get(t.endpoint.id); + expect(current).not.toBeNull(); expect(current).not.toBe(first); + }, { timeout: 5000 }); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 5000 }); + await t.callbacks().onPhotonFailure!(new PhotonError("credentials", "Rejected credentials")); + await vi.waitFor(() => expect(t.service.runtime.get(t.endpoint.id)).toBeNull(), { timeout: 5000 }); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 5000 }); + expect(await db.select().from(toolConnections).where(eq(toolConnections.id, t.endpoint.connectionId))) + .toEqual([expect.objectContaining({ enabled: false })]); + expect(onError).not.toHaveBeenCalled(); + }); + it("keeps publication authentication failures stopped across receiver reconciliation and restart", async () => { const t = await setup(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: () => {} }); + coordinators.push(coordinator); + await t.service.registerReceiverWork(coordinator, () => true).ready; const conversation = await t.start(); await t.qualify(); const callbacks = t.callbacks(); @@ -380,8 +411,8 @@ describe("iMessage Photon channel control plane", () => { expect.objectContaining({ status: "active", enabled: true, healthStatus: "failed" }), ]); await expect(callbacks.onPhotonAssertOwned!()).rejects.toThrow(); - await t.service.reconcileProviderRuntimes(); - expect(t.service.runtime.get(t.endpoint.id)).toBeFalsy(); + await vi.waitFor(() => expect(t.service.runtime.get(t.endpoint.id)).toBeFalsy(), { timeout: 5000 }); + await coordinator.stop(); const restarted = await t.restart(); expect(restarted.runtime.get(t.endpoint.id)).toBeFalsy(); expect(await restarted.get(t.endpoint.id)).toMatchObject({ status: "attention" }); diff --git a/server/src/app.ts b/server/src/app.ts index ac1666f56d2..a38f146dce3 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -321,7 +321,7 @@ export function shouldEnablePrivateHostnameGuard(opts: { } type ChatReconciliationLane = - | "provider runtimes" + | "voice sessions" | "deliveries" | "GitHub webhook recovery" | "run milestones" @@ -335,7 +335,7 @@ type ChatReconciliationLane = * next publication sweep for every endpoint. */ export function createChatReconciliationCoordinator(input: { - reconcileProviderRuntimes: () => Promise; + reconcileVoiceSessions: () => Promise; processPendingDeliveries: () => Promise; processFailedGitHubWebhookDeliveries?: () => Promise; projectRunMilestones: () => Promise; @@ -377,7 +377,7 @@ export function createChatReconciliationCoordinator(input: { return { reconcile() { if (stopped) return; - start("provider runtimes", input.reconcileProviderRuntimes); + start("voice sessions", input.reconcileVoiceSessions); start("deliveries", input.processPendingDeliveries); if (input.processFailedGitHubWebhookDeliveries) { start( @@ -1298,16 +1298,14 @@ export async function createApp( } registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); + chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); const reconcileChatPublicationMaintenance = async () => { await chatChannels.processPublicationMaintenance(); }; const chatReconciliation = createChatReconciliationCoordinator({ - reconcileProviderRuntimes: async () => { - await chatChannels.reconcileProviderRuntimes(); - await chatChannels.voice.reconcile(); - }, + reconcileVoiceSessions: () => chatChannels.voice.reconcile(), // Chat delivery, action, and Telegram discovery workers now own these // duties. Keep the coordinator's other maintenance lanes independent. processPendingDeliveries: async () => undefined, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 5059e0dddec..9fbcb10ab08 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,4 +1,7 @@ -import { notifyChatCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { leasedChatProvider } from "./chat-channels/runtime/receiver-policy.js"; +import { chatReceiverSchedule } from "./chat-channels/runtime/receiver-schedule.js"; +import { registerChatReceiverWork } from "./chat-channels/runtime/receiver-work.js"; +import { notifyChatEndpointConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js"; import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; @@ -284,7 +287,6 @@ const ORPHAN_FOLLOW_UP_GRACE_MS = 5_000; const ORPHAN_FOLLOW_UP_MAX_ATTEMPTS = 12; const PUBLICATION_ENDPOINT_CONCURRENCY = 4; -function leasedChatProvider(provider: string): boolean { return provider === "discord" || provider === "imessage-photon"; } const DISCORD_GATEWAY_LEASE_POLL_MS = 100; const PROVIDER_CREDENTIAL_CHECK_TIMEOUT_MS = 25_000; @@ -568,7 +570,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const discordGatewayOwnerships = new Map(); const discordGatewayStopTasks = new Set>(); let shuttingDown = false; - const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } }); + let receiverWork: ReturnType | undefined; + const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, onStopped: id => receiverWork?.changed(id), get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } }); const { receiveSdkMessages } = createChatSdkMessageReceiver({ endpointRecord, processMessage, ingressOnly: (provider) => provider === "imessage-photon" @@ -1139,6 +1142,29 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } } + const receiverSchedule = chatReceiverSchedule(db, { + stop: async id => { + const ownership = discordGatewayOwnerships.get(id); + if (ownership) await stopDiscordGatewayOwnership(ownership); + }, + reconcile: async id => { + const record = await endpointRecord(id); + if (!record) return "owned"; + try { + const instance = await runtimeFor(record.endpoint, { requireDiscordOwnership: true }); + await reconcileDiscordRuntime(id, instance); + return "owned"; + } catch (error) { + if (isDiscordGatewayNotOwnedError(error)) return "foreign"; + throw error; + } + }, + }); + function registerReceiverWork(coordinator: ReturnType, canRun: () => boolean) { + receiverWork = registerChatReceiverWork(coordinator, receiverSchedule, () => !shuttingDown && canRun()); + return receiverWork; + } + async function reconcileProviderRuntimes() { if (shuttingDown) { return { eligible: 0, local: 0, ownedElsewhere: 0, failed: 0 }; @@ -2215,7 +2241,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ? "Started a conversation by mentioning the agent." : "Sent an empty message."); let comment!: Awaited>; - await notifyChatCommandRegistration(taskTx, taskEndpoint); + await notifyChatEndpointConfiguration(taskTx, taskEndpoint); await taskTx .update(chatEndpoints) .set({ @@ -3475,6 +3501,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { registerCommandWork, + registerReceiverWork, voice, authorizeFastResponse, slackRegistration, diff --git a/server/src/services/chat-channels/command-work-notifications.ts b/server/src/services/chat-channels/command-work-notifications.ts deleted file mode 100644 index e8f5812d4dd..00000000000 --- a/server/src/services/chat-channels/command-work-notifications.ts +++ /dev/null @@ -1,18 +0,0 @@ -import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; -import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js"; - -export async function notifyChatCommandRegistration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { - if (endpoint.provider === "discord") { - await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); - } -} - -export async function notifyChatConnectionCommandRegistration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { - if (transport === "chat_sdk") { - await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); - } -} - -export async function notifyChatCompanyCommandRegistration(transaction: object, companyId: string) { - await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); -} diff --git a/server/src/services/chat-channels/configuration-work-notifications.ts b/server/src/services/chat-channels/configuration-work-notifications.ts new file mode 100644 index 00000000000..5b592e896ec --- /dev/null +++ b/server/src/services/chat-channels/configuration-work-notifications.ts @@ -0,0 +1,22 @@ +import { notifyChatReceiver, notifyChatConnectionReceiver, notifyChatCompanyReceivers } from "./runtime/receiver-work-notifications.js"; +import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; +import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js"; + +export async function notifyChatEndpointConfiguration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { + await notifyChatReceiver(transaction, endpoint); + if (endpoint.provider === "discord") { + await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); + } +} + +export async function notifyChatConnectionConfiguration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { + await notifyChatConnectionReceiver(transaction, connectionId, transport); + if (transport === "chat_sdk") { + await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); + } +} + +export async function notifyChatCompanyConfiguration(transaction: object, companyId: string) { + await notifyChatCompanyReceivers(transaction, companyId); + await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); +} diff --git a/server/src/services/chat-channels/credentials.ts b/server/src/services/chat-channels/credentials.ts index 667593066b9..cd24c3c72cc 100644 --- a/server/src/services/chat-channels/credentials.ts +++ b/server/src/services/chat-channels/credentials.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "./command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { prepareSlackInventory } from "./slack/inventory.js"; import { prepareGitHubInventory } from "./github/inventory.js"; import { prepareDiscordInventory } from "./discord/inventory.js"; @@ -316,7 +316,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { code: "chat_endpoint_credentials_changed", }); } - await notifyChatCommandRegistration(tx, endpoint); + await notifyChatEndpointConfiguration(tx, endpoint); await tx .delete(companySecretBindings) .where( @@ -413,7 +413,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { .then((rows) => rows[0] ?? null); if (!current) throw notFound("Chat connection not found"); refs = current.refs; - await notifyChatCommandRegistration(tx, endpoint); + await notifyChatEndpointConfiguration(tx, endpoint); await tx .delete(companySecretBindings) .where( diff --git a/server/src/services/chat-channels/discord/gateway-events.ts b/server/src/services/chat-channels/discord/gateway-events.ts index 47deeff7941..3ed17aa6aa3 100644 --- a/server/src/services/chat-channels/discord/gateway-events.ts +++ b/server/src/services/chat-channels/discord/gateway-events.ts @@ -1,3 +1,4 @@ +import { notifyChatReceiver } from "../runtime/receiver-work-notifications.js"; import { notifyDiscordCommandRegistration } from "./command-work-notifications.js"; import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence, InternalSetupState } from "../runtime-types.js"; @@ -130,7 +131,10 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents } const now = new Date(); if (event.type === "connecting") { - if (current.status === "attention") await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); + if (current.status === "attention") { + await notifyChatReceiver(tx, current); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); + } await tx .update(chatEndpoints) .set({ @@ -159,6 +163,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents if (current.status === "attention" && !gatewayOwnedAttention) return; const status = current.setup.step === "complete" ? "active" : "verifying"; + await notifyChatReceiver(tx, current); await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); if (status !== current.status) await notifyChatEndpointWork(tx, current.connectionId); await tx @@ -189,6 +194,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents return; } + await notifyChatReceiver(tx, current); await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); const detail = discordGatewayFailureDetail(event); const transientGuildOutage = event.type === "guild_unavailable"; diff --git a/server/src/services/chat-channels/imessage-photon/events.ts b/server/src/services/chat-channels/imessage-photon/events.ts index 14a914287fd..75f6d2cc92b 100644 --- a/server/src/services/chat-channels/imessage-photon/events.ts +++ b/server/src/services/chat-channels/imessage-photon/events.ts @@ -1,3 +1,4 @@ +import { notifyChatReceiver } from "../runtime/receiver-work-notifications.js"; import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js"; import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; @@ -104,6 +105,7 @@ export function createPhotonEvents(db: Db, options: PhotonEventsOptions) { "attention", ]); if (!endpoint) return; + await notifyChatReceiver(tx, endpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index 7eb531ae29b..157c48a5406 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "../command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { createDiscordQuestionForms } from "../discord/question-forms.js"; @@ -412,7 +412,7 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb ["active"], ); if (currentEndpoint) { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/provider-effects.ts b/server/src/services/chat-channels/provider-effects.ts index 1273c07c59b..e4a000f94ee 100644 --- a/server/src/services/chat-channels/provider-effects.ts +++ b/server/src/services/chat-channels/provider-effects.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "./command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { createTelegramCallbackNoticeAuthority } from "./telegram/callback-notices.js"; @@ -318,7 +318,7 @@ export function createProviderEffects(db: Db, options: ProviderEffectsOptions) { ); } if (input.disposition.kind === "endpoint_attention" && currentEndpoint) { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index ae2580dfb3f..b5f6648b646 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "../command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; @@ -190,7 +190,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur : null; if (disposition.kind === "endpoint_attention" && currentEndpoint) { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index 98bfb5822d4..0b52e8e7c69 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,4 +1,4 @@ -import { notifyChatCommandRegistration } from "./command-work-notifications.js"; +import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { imessageResourceParticipants } from "./imessage-photon/resources.js"; import { createGitHubResources, reconcileGitHubRepositoryCoordinates } from "./github/resources.js"; @@ -510,7 +510,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc : currentEndpoint.status === "attention" ? "verifying" : currentEndpoint.status; - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); if (status !== currentEndpoint.status) await notifyChatEndpointWork(tx, currentEndpoint.connectionId); await tx .update(chatEndpoints) @@ -540,7 +540,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc .where(eq(toolConnections.id, currentEndpoint.connectionId)); } } else { - await notifyChatCommandRegistration(tx, currentEndpoint); + await notifyChatEndpointConfiguration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/runtime/ownership.ts b/server/src/services/chat-channels/runtime/ownership.ts index 3e8bec1fbc4..4f2860eac30 100644 --- a/server/src/services/chat-channels/runtime/ownership.ts +++ b/server/src/services/chat-channels/runtime/ownership.ts @@ -1,3 +1,4 @@ +import { receiverLeaseKey as leaseKeyForProvider } from "./receiver-policy.js"; import type { EndpointRow, DbTransaction, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js"; import { randomUUID } from "node:crypto"; import { and, eq, lte } from "drizzle-orm"; @@ -17,12 +18,12 @@ type OwnershipOptions = { discordGatewayOwnerships: Map; discordGatewayStopTasks: Set>; isShuttingDown: () => boolean; + onStopped?: (endpointId: string) => void; readonly discordGatewayLeaseTtlMs?: number; readonly discordGatewayLeaseRenewalIntervalMs?: number; readonly discordGatewayLeaseWaitMs?: number; readonly renewDiscordGatewayLease?: (input: { companyId: string; credentialFingerprint: string; endpointId: string; expiresAt: Date; generation: number; leaseKey: string; token: string }) => Promise; }; -const DISCORD_GATEWAY_LEASE_KEY = "discord_gateway_runtime"; const DISCORD_GATEWAY_LEASE_TTL_MS = 15_000; const DISCORD_GATEWAY_LEASE_WAIT_MS = 20_000; const DISCORD_GATEWAY_LEASE_POLL_MS = 100; @@ -81,9 +82,6 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { clearInterval(ownership.renewTimer); ownership.renewTimer = null; } - if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) { - discordGatewayOwnerships.delete(ownership.endpointId); - } ownership.stopPromise = (async () => { if ( removeRuntime && @@ -104,6 +102,10 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { // The listener is stopped before the lease is released. A standby can // therefore acquire immediately without overlapping Gateway sessions. await releaseDiscordGatewayLeaseRow(ownership); + if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) { + discordGatewayOwnerships.delete(ownership.endpointId); + } + if (!options.isShuttingDown()) options.onStopped?.(ownership.endpointId); })(); discordGatewayStopTasks.add(ownership.stopPromise); void ownership.stopPromise.finally(() => { @@ -304,7 +306,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { context: RuntimeContext, waitForOwnership: boolean, ): Promise { - const receiverLeaseKey = endpoint.provider === "imessage-photon" ? "photon_receiver_runtime" : DISCORD_GATEWAY_LEASE_KEY; + const receiverLeaseKey = leaseKeyForProvider(endpoint.provider)!; const local = discordGatewayOwnerships.get(endpoint.id); if ( local && diff --git a/server/src/services/chat-channels/runtime/receiver-policy.ts b/server/src/services/chat-channels/runtime/receiver-policy.ts new file mode 100644 index 00000000000..70983c32bbb --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-policy.ts @@ -0,0 +1,13 @@ +import type { ChatProvider } from "@paperclipai/shared"; + +export const RECEIVER_LEASE_KEYS = { + discord: "discord_gateway_runtime", + "imessage-photon": "photon_receiver_runtime", +} as const satisfies Partial>; + +export function receiverLeaseKey(provider: string) { + return Object.hasOwn(RECEIVER_LEASE_KEYS, provider) + ? RECEIVER_LEASE_KEYS[provider as keyof typeof RECEIVER_LEASE_KEYS] : undefined; +} + +export const leasedChatProvider = (provider: string) => receiverLeaseKey(provider) !== undefined; diff --git a/server/src/services/chat-channels/runtime/receiver-schedule.ts b/server/src/services/chat-channels/runtime/receiver-schedule.ts new file mode 100644 index 00000000000..4e07ec7fb95 --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-schedule.ts @@ -0,0 +1,53 @@ +import { and, eq, gt, inArray, ne, or } from "drizzle-orm"; +import { chatEndpoints, chatEndpointLeases, toolConnections, type Db } from "@paperclipai/db"; +import type { ChatProvider } from "@paperclipai/shared"; +import { RECEIVER_LEASE_KEYS, receiverLeaseKey } from "./receiver-policy.js"; +import type { ReceiverScope } from "./receiver-work-notifications.js"; + +export type ReceiverDeadline = { id: string; companyId: string; connectionId: string; dueAt: number | null }; + +export function chatReceiverSchedule(db: Db, receiver: { + reconcile: (id: string) => Promise<"owned" | "foreign">; + stop: (id: string) => Promise; +}) { + async function load(scope?: ReceiverScope, signal?: AbortSignal): Promise { + const result: ReceiverDeadline[] = []; + let cursor: string | undefined; + while (true) { + signal?.throwIfAborted(); + const rows = await db.select({ id: chatEndpoints.id, companyId: chatEndpoints.companyId, connectionId: chatEndpoints.connectionId }) + .from(chatEndpoints).innerJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatEndpoints.companyId))) + .where(and( + inArray(chatEndpoints.provider, Object.keys(RECEIVER_LEASE_KEYS) as ChatProvider[]), + inArray(chatEndpoints.status, ["verifying", "active", "attention"]), + eq(toolConnections.status, "active"), eq(toolConnections.enabled, true), + or(ne(chatEndpoints.status, "attention"), ne(toolConnections.healthStatus, "failed")), + scope?.kind === "endpoint" ? eq(chatEndpoints.id, scope.id) : undefined, + scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined, + scope?.kind === "company" ? eq(chatEndpoints.companyId, scope.id) : undefined, + cursor ? gt(chatEndpoints.id, cursor) : undefined, + )).orderBy(chatEndpoints.id).limit(100); + result.push(...rows.map(row => ({ ...row, dueAt: 0 }))); + if (rows.length < 100) return result; + cursor = rows[rows.length - 1].id; + } + } + async function check(id: string, signal?: AbortSignal): Promise { + const scope = { kind: "endpoint", id } as const; + const [before] = await load(scope, signal); + if (!before) { await receiver.stop(id); return null; } + signal?.throwIfAborted(); + const ownership = await receiver.reconcile(id); + const [after] = await load(scope, signal); + if (!after) { await receiver.stop(id); return null; } + if (ownership === "owned") return { ...after, dueAt: null }; + const [endpoint] = await db.select({ provider: chatEndpoints.provider }).from(chatEndpoints).where(eq(chatEndpoints.id, id)); + if (!endpoint) { await receiver.stop(id); return null; } + const key = receiverLeaseKey(endpoint.provider); + const [lease] = key ? await db.select({ expiresAt: chatEndpointLeases.expiresAt }).from(chatEndpointLeases).where(and( + eq(chatEndpointLeases.companyId, after.companyId), eq(chatEndpointLeases.endpointId, id), eq(chatEndpointLeases.leaseKey, key), + )).limit(1) : []; + return { ...after, dueAt: Math.max(Date.now() + 1000, lease?.expiresAt.getTime() ?? 0) }; + } + return { load, check, stop: receiver.stop }; +} diff --git a/server/src/services/chat-channels/runtime/receiver-work-notifications.ts b/server/src/services/chat-channels/runtime/receiver-work-notifications.ts new file mode 100644 index 00000000000..5ff7a2e2248 --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-work-notifications.ts @@ -0,0 +1,24 @@ +import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; +import { leasedChatProvider } from "./receiver-policy.js"; + +export type ReceiverScope = { kind: "endpoint" | "connection" | "company"; id: string }; +export function receiverScope(hint: string): ReceiverScope | undefined { + if (hint === "*") return undefined; + const [kind, id, extra] = hint.split(":"); + if (!id || extra !== undefined || (kind !== "endpoint" && kind !== "connection" && kind !== "company")) + throw new Error("Invalid receiver discovery scope"); + return { kind, id }; +} +export const receiverScopeHint = (scope: ReceiverScope) => `${scope.kind}:${scope.id}`; +export async function notifyChatReceiver(transaction: object, endpoint: { id: string; provider: ChatProvider }) { + if (leasedChatProvider(endpoint.provider)) + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "endpoint", id: endpoint.id })); +} +export async function notifyChatConnectionReceiver(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { + if (transport === "chat_sdk") + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "connection", id: connectionId })); +} +export async function notifyChatCompanyReceivers(transaction: object, companyId: string) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "company", id: companyId })); +} diff --git a/server/src/services/chat-channels/runtime/receiver-work.ts b/server/src/services/chat-channels/runtime/receiver-work.ts new file mode 100644 index 00000000000..49b4cb509d4 --- /dev/null +++ b/server/src/services/chat-channels/runtime/receiver-work.ts @@ -0,0 +1,87 @@ +import { takeCompanyWorkBatch } from "../../company-work-batch.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import type { ReceiverDeadline, chatReceiverSchedule } from "./receiver-schedule.js"; +import { receiverScope, receiverScopeHint, type ReceiverScope } from "./receiver-work-notifications.js"; + +function matches(entry: ReceiverDeadline, scope?: ReceiverScope) { + return !scope || (scope.kind === "endpoint" ? entry.id : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id; +} +export function registerChatReceiverWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const entries = new Map(); + const failures = new Map(); + const scopeFailures = new Map(); + let earliest = 0; + let lastCompany: string | null = null; + function failed(map: typeof failures, id: string) { + const count = Math.min((map.get(id)?.count ?? 0) + 1, 5); + map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }); + } + function dirtyScopes() { + return [...dirty].map(hint => { try { return receiverScope(hint); } catch { return undefined; } }); + } + function next() { + let nextAt: number | null = null; + const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; + for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); + const unavailable = dirtyScopes(); + for (const entry of entries.values()) if (entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope))) + take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); + return nextAt; + } + const worker = coordinator.register(DELIVERY_QUEUES.chatReceivers, { + retryMs: 30_000, + pausePassiveMaintenance: true, + onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + 1000; + const errors: unknown[] = []; + const requested = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); + for (const hint of requested) dirty.delete(hint); + for (let index = 0; index < requested.length; index++) { + if (signal.aborted || !canRun()) { for (const hint of requested.slice(index)) dirty.add(hint); return; } + const hint = requested[index]; + let scope: ReceiverScope | undefined; + let decoded = false; + try { + scope = receiverScope(hint); decoded = true; + const rows = await service.load(scope, signal), retained = new Set(rows.map(row => row.id)); + for (const entry of entries.values()) if (matches(entry, scope) && !retained.has(entry.id)) { + await service.stop(entry.id); + entries.delete(entry.id); failures.delete(entry.id); + } + for (const entry of rows) entries.set(entry.id, entry); + scopeFailures.delete(hint); + } catch (error) { + const retryHint = decoded ? hint : "*"; + dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error); + } + } + const unavailable = dirtyScopes(); + const due = [...entries.values()].filter(entry => entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope)) + && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) + .sort((a, b) => a.dueAt! - b.dueAt! || a.id.localeCompare(b.id)); + const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany; + for (const entry of batch.entries) { + if (signal.aborted || !canRun()) break; + try { + const current = await service.check(entry.id, signal); + if (current) entries.set(entry.id, current); else entries.delete(entry.id); + failures.delete(entry.id); + } catch (error) { failed(failures, entry.id); errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, "Chat receiver discovery could not settle"); + }, + hasPending: async () => dirty.size > 0, + nextRunAt: next, + }); + return { ready: worker.ready, changed(id: string) { + dirty.add(receiverScopeHint({ kind: "endpoint", id })); worker.wake(); + } }; +} diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index cc4742d0b71..71b00c50cf0 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -1,4 +1,4 @@ -import { notifyChatConnectionCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { notifyChatConnectionConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export async function notifyChatDeliveryWork(tx: object): Promise { @@ -38,5 +38,5 @@ export async function notifyChatEndpointWork(tx: object, connectionId: string): for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue); await notifyChatVerificationWork(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTelegramSubscriptions, connectionId); - await notifyChatConnectionCommandRegistration(tx, connectionId); + await notifyChatConnectionConfiguration(tx, connectionId); } diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index db5b6e8bede..6d2453bf888 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,4 +1,4 @@ -import { notifyChatCompanyCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { chatCompanyDeletion } from "./chat-channels/company-deletion.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; @@ -65,7 +65,7 @@ export async function deleteCompany(db: Db, id: string) { const [existing] = await tx.select({ id: companies.id }).from(companies) .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; - await notifyChatCompanyCommandRegistration(tx, id); + await notifyChatCompanyConfiguration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await agentObservabilityService(tx as unknown as Db).removeForCompanyDeletion(id); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 017e83651c6..07525867d79 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -40,6 +40,7 @@ export const DELIVERY_QUEUES = { chatGitHubPublications: "chat-github-publications", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", + chatReceivers: "chat-receivers", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs", diff --git a/server/src/services/tool-access/catalog/health.ts b/server/src/services/tool-access/catalog/health.ts index 86ef894a54b..5a5998986a0 100644 --- a/server/src/services/tool-access/catalog/health.ts +++ b/server/src/services/tool-access/catalog/health.ts @@ -1,4 +1,4 @@ -import { notifyChatConnectionCommandRegistration } from "../../chat-channels/command-work-notifications.js"; +import { notifyChatConnectionConfiguration } from "../../chat-channels/configuration-work-notifications.js"; import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { assertSupportedConnection } from "../connections/setup-policy.js"; import { connectionSetupUrl, connectionReconnectUrl } from "../identifiers.js"; @@ -49,7 +49,7 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d ) { const now = new Date(); const [updated] = await writeToolConnectionHealthChange(db, connection, async tx => { - await notifyChatConnectionCommandRegistration(tx, connection.id, connection.transport); + await notifyChatConnectionConfiguration(tx, connection.id, connection.transport); return tx.update(toolConnections) .set({ healthStatus: status,