fix(server): wake chat receiver discovery from configuration and lease deadlines

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.

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.

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<void>((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);

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<Pick<ChatChannelServiceOptions, "discordGatewayLeaseTtlMs" | "discordGatewayLeaseRenewalIntervalMs">> = {},
     ) {
       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<void>(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 });

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>, companyId: string, agentId: string, applicationId: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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<typeof chatEndpoints.$inferInsert> = {}, 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<void>(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();
+  });
+});

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<T>(): TypedEventStream<T> {
   );
 }
 describe("iMessage Photon channel control plane", () => {
-  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   let db: ReturnType<typeof createDb>;
   let secrets: string;
   const oldKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE;
   const services: ChatChannelService[] = [];
+  const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
   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" });

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<unknown>;
+  reconcileVoiceSessions: () => Promise<unknown>;
   processPendingDeliveries: () => Promise<unknown>;
   processFailedGitHubWebhookDeliveries?: () => Promise<unknown>;
   projectRunMilestones: () => Promise<number>;
@@ -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,

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<string, DiscordGatewayOwnership>();
   const discordGatewayStopTasks = new Set<Promise<void>>();
   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<typeof registerChatReceiverWork> | 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<typeof createDeliveryWorkCoordinator>, 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<ReturnType<typeof issuesSvc.addComment>>;
-        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,

server/src/services/chat-channels/command-work-notifications.ts

@@ -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 });
-}

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 });
+}

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(

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";

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({

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({

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({

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({

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({

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<string, DiscordGatewayOwnership>;
   discordGatewayStopTasks: Set<Promise<void>>;
   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<boolean>;
 };
-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<DiscordGatewayOwnership | null> {
-    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 &&

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<Record<ChatProvider, string>>;
+
+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;

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<void>;
+}) {
+  async function load(scope?: ReceiverScope, signal?: AbortSignal): Promise<ReceiverDeadline[]> {
+    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<ReceiverDeadline | null> {
+    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 };
+}

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 }));
+}

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<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof chatReceiverSchedule>,
+  canRun: () => boolean,
+) {
+  const dirty = new Set<string>(["*"]);
+  const entries = new Map<string, ReceiverDeadline>();
+  const failures = new Map<string, { count: number; retryAt: number }>();
+  const scopeFailures = new Map<string, { count: number; retryAt: number }>();
+  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();
+  } };
+}

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<void> {
@@ -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);
 }

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);

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",

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,