fix(server): coalesce receiver recovery without redundant stop wakes

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

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -1211,3 +1211,20 @@ 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.
+
+
+## Receiver review repair
+
+The real company-removal producer regression reproduced two inventory reads for
+one observed removal. Discovery-initiated stops now suppress their own completion
+wake; independent token loss still wakes discovery after listener retirement.
+A full reconstruction also covers its already-pending scoped hints. Hints that
+arrive during the load remain dirty, and narrower writes cannot bypass a failed
+full reconstruction's retry deadline.
+
+Both redundant-load regressions failed before repair. All 45 selected receiver,
+startup and real Discord producer tests pass after repair, including the added
+full-refresh failure/backoff case. Server typecheck and build pass. Prior broad
+consumer/workspace verification remains recorded above; these 45 overlap that
+coverage and are not an additional unique total. Publication requires fresh
+remote CI and review. The full migration inventory remains outstanding.

doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md

@@ -231,3 +231,20 @@ Master was fetched again at 0d73505098. Company prerequisite #17561 at e8febb30e
 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.
+
+
+## Receiver review repair
+
+The real company-removal producer regression reproduced two inventory reads for
+one observed removal. Discovery-initiated stops now suppress their own completion
+wake; independent token loss still wakes discovery after listener retirement.
+A full reconstruction also covers its already-pending scoped hints. Hints that
+arrive during the load remain dirty, and narrower writes cannot bypass a failed
+full reconstruction's retry deadline.
+
+Both redundant-load regressions failed before repair. All 45 selected receiver,
+startup and real Discord producer tests pass after repair, including the added
+full-refresh failure/backoff case. Server typecheck and build pass. Prior broad
+consumer/workspace verification remains recorded above; these 45 overlap that
+coverage and are not an additional unique total. Publication requires fresh
+remote CI and review. The full migration inventory remains outstanding.

server/src/__tests__/chat-channels.integration.test.ts

@@ -1,4 +1,4 @@
-import { notifyChatEndpointConfiguration } from "../services/chat-channels/configuration-work-notifications.js";
+import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } 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";
@@ -6799,6 +6799,29 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
       }
     });
 
+    it("does not rediscover a receiver stopped by a company removal refresh", 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(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+      let reads: ReturnType<typeof vi.spyOn> | undefined;
+      try {
+        await f.configure(); await f.service.registerReceiverWork(coordinator, () => true).ready;
+        await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); });
+        const owned = f.runtime.get(f.endpoint.id)!;
+        reads = vi.spyOn(db, "select");
+        await db.transaction(async tx => {
+          await notifyChatCompanyConfiguration(tx, f.fixture.companyId);
+          await tx.delete(chatEndpoints).where(eq(chatEndpoints.id, f.endpoint.id));
+        });
+        await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); });
+        expect(owned.shutdown).toHaveBeenCalledOnce();
+        expect(reads).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled();
+      } finally {
+        reads?.mockRestore(); await coordinator.stop(); vi.useRealTimers(); await f.service.shutdown();
+        await db.update(chatEndpoints).set({ status: "archived" }).where(eq(chatEndpoints.id, 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();

server/src/__tests__/chat-receiver-work.test.ts

@@ -153,6 +153,34 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     service.load = original; reconcile.mockResolvedValue("owned");
     await settled(); errors.length = 0; await quiet();
   });
+  it("coalesces a full refresh with its already pending scoped hints", async () => {
+    const service = chatReceiverSchedule(db, { reconcile, stop });
+    await start(service); await settled();
+    const load = vi.spyOn(service, "load");
+    await db.transaction(async tx => {
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*");
+      await endpoint({}, tx as unknown as typeof db);
+      await endpoint({}, tx as unknown as typeof db);
+    });
+    await settled(); expect(reconcile).toHaveBeenCalledTimes(2);
+    expect(load).toHaveBeenCalledOnce(); expect(load.mock.calls[0][0]).toBeUndefined();
+    load.mockRestore(); await quiet();
+  });
+  it("keeps full-refresh backoff when a narrower producer arrives", async () => {
+    const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load;
+    await start(service); await settled();
+    const load = vi.spyOn(service, "load").mockRejectedValue(new Error("inventory unavailable"));
+    await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*"));
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); });
+    const retryAt = coordinator!.nextWakeAt();
+    const row = await endpoint();
+    await quiet(10_000);
+    expect(load).toHaveBeenCalledOnce(); expect(coordinator!.nextWakeAt()).toBe(retryAt);
+    load.mockImplementation(original);
+    await settled(); expect(reconcile).toHaveBeenCalledWith(row.id);
+    expect(load).toHaveBeenCalledTimes(2); expect(load.mock.calls[1][0]).toBeUndefined();
+    errors.length = 0; load.mockRestore(); 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"));

server/src/services/chat-channels.ts

@@ -1125,7 +1125,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
             runtimeVersions.delete(endpoint.id);
           }
           if (discordOwnership) {
-            await stopDiscordGatewayOwnership(discordOwnership, false);
+            await stopDiscordGatewayOwnership(discordOwnership, { removeRuntime: false });
           }
           throw error;
         }
@@ -1145,7 +1145,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
   const receiverSchedule = chatReceiverSchedule(db, {
     stop: async id => {
       const ownership = discordGatewayOwnerships.get(id);
-      if (ownership) await stopDiscordGatewayOwnership(ownership);
+      // Discovery already observed removal; only independent ownership loss needs another wake.
+      if (ownership) await stopDiscordGatewayOwnership(ownership, { notifyStopped: false });
     },
     reconcile: async id => {
       const record = await endpointRecord(id);

server/src/services/chat-channels/runtime/ownership.ts

@@ -74,7 +74,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
 
   async function stopDiscordGatewayOwnership(
     ownership: DiscordGatewayOwnership,
-    removeRuntime = true,
+    stopOptions: { removeRuntime?: boolean; notifyStopped?: boolean } = {},
   ): Promise<void> {
     if (ownership.stopPromise) return ownership.stopPromise;
     ownership.stopping = true;
@@ -84,7 +84,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
     }
     ownership.stopPromise = (async () => {
       if (
-        removeRuntime &&
+        stopOptions.removeRuntime !== false &&
         ownership.context.endpointRuntime &&
         runtime.get(ownership.endpointId) === ownership.context.endpointRuntime
       ) {
@@ -105,7 +105,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
       if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) {
         discordGatewayOwnerships.delete(ownership.endpointId);
       }
-      if (!options.isShuttingDown()) options.onStopped?.(ownership.endpointId);
+      if (!options.isShuttingDown() && stopOptions.notifyStopped !== false) options.onStopped?.(ownership.endpointId);
     })();
     discordGatewayStopTasks.add(ownership.stopPromise);
     void ownership.stopPromise.finally(() => {

server/src/services/chat-channels/runtime/receiver-work.ts

@@ -28,7 +28,7 @@ export function registerChatReceiverWork(
   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));
+    for (const hint of dirty.has("*") ? ["*"] : 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));
@@ -42,10 +42,11 @@ export function registerChatReceiverWork(
       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);
+      const requested = (dirty.has("*") ? ["*"] : [...dirty]).filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
+      const covered = requested.includes("*") ? [...dirty] : requested;
+      for (const hint of covered) 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; }
+        if (signal.aborted || !canRun()) { for (const hint of requested.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; }
         const hint = requested[index];
         let scope: ReceiverScope | undefined;
         let decoded = false;
@@ -57,7 +58,7 @@ export function registerChatReceiverWork(
             entries.delete(entry.id); failures.delete(entry.id);
           }
           for (const entry of rows) entries.set(entry.id, entry);
-          scopeFailures.delete(hint);
+          if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);
         } catch (error) {
           const retryHint = decoded ? hint : "*";
           dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error);