fix(server): bound inbound notice authorization passes

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

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

@@ -71499,9 +71499,17 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
         const second = await context.service.enqueueInboundWakeupPublications(50);
         expect(first + second).toBe(1);
       } else {
-        expect(await context.service.reconcileInboundWakeNotices({
-          scope: { companyId: context.fixture.companyId, issueId: context.conversation.issueId }, limit: 50,
-        })).toEqual({ inserted: 1, contended: false });
+        const scope = { companyId: context.fixture.companyId, issueId: context.conversation.issueId };
+        const transaction = vi.spyOn(db, "transaction");
+        try {
+          const first = await context.service.reconcileInboundWakeNotices({ scope, limit: 50 });
+          expect(transaction).toHaveBeenCalledTimes(50);
+          expect(first).toMatchObject({ inserted: 0, contended: false, nextCursor: expect.any(Object) });
+          transaction.mockClear();
+          expect(await context.service.reconcileInboundWakeNotices({ scope, limit: 50, cursor: first.nextCursor }))
+            .toEqual({ inserted: 1, contended: false, nextCursor: null });
+          expect(transaction).toHaveBeenCalledTimes(1);
+        } finally { transaction.mockRestore(); }
       }
       const rows = await readChatPublicationsByEndpointId(context.endpoint.id);
       expect(rows).toEqual([expect.objectContaining({
@@ -71515,19 +71523,135 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     }
   });
 
+  it.each(["denied", "contended"] as const)("yields a %s inbound notice backlog between bounded passes without losing work", async mode => {
+    const context = await deferredChatQueueFixture();
+    const other = await deferredChatQueueFixture();
+    const rejectedIds = Array.from({ length: 150 }, () => randomUUID());
+    let release = () => {};
+    let holder: Promise<unknown> | undefined;
+    try {
+      const [receipt] = await readAgentWakeupRequestsById(context.action.id);
+      const at = new Date(context.action.createdAt.getTime() + (mode === "denied" ? -1000 : 1000)).toISOString();
+      await db.insert(chatActions).values(rejectedIds.map(id => ({
+        ...context.action, id, providerActionId: `inbound_wakeup:${id}`,
+        payload: { ...context.action.payload, commentId: randomUUID() },
+        createdAt: sql`${at}::timestamptz + interval '0.000123 seconds'`,
+      })));
+      await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({
+        ...receipt, id, idempotencyKey: `chat-fairness-fixture:${id}`,
+      })));
+      if (mode === "contended") {
+        let locked!: () => void;
+        const ready = new Promise<void>(resolve => { locked = resolve; });
+        const hold = new Promise<void>(resolve => { release = resolve; });
+        holder = db.transaction(async tx => {
+          await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, context.action.id)).for("update");
+          locked(); await hold;
+        });
+        await ready;
+      }
+      const worker = chatInboundWakeNoticeWork(context.service);
+      const [entry] = await worker.load({ companyId: context.fixture.companyId, issueId: context.conversation.issueId });
+      const [otherEntry] = await worker.load({ companyId: other.fixture.companyId, issueId: other.conversation.issueId });
+      const transaction = vi.spyOn(db, "transaction");
+      try {
+        let remaining = await worker.check(entry!);
+        expect(transaction).toHaveBeenCalledTimes(100);
+        expect(remaining).toMatchObject({ cursor: expect.any(Object), contended: mode === "contended" });
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]);
+        // Another scope can finish while the first still has a saved cursor.
+        expect(await worker.check(otherEntry!)).toBeNull();
+        expect(await readChatPublicationsByEndpointId(other.endpoint.id)).toHaveLength(1);
+        transaction.mockClear();
+        remaining = await worker.check(remaining!);
+        expect(transaction).toHaveBeenCalledTimes(51);
+        if (mode === "contended") {
+          expect(remaining).toMatchObject({ cursor: null });
+          expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]);
+          release(); await holder;
+          remaining = await worker.check(remaining!);
+          expect(remaining).not.toBeNull();
+          remaining = await worker.check(remaining!);
+        }
+        expect(remaining).toBeNull();
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({
+          commentId: context.action.payload.commentId,
+          payload: { text: "Your follow-up is queued.", progressState: "queued" },
+        })]);
+      } finally { transaction.mockRestore(); }
+    } finally {
+      release(); await holder;
+      await context.service.shutdown(); await other.service.shutdown();
+      await db.delete(chatActions).where(inArray(chatActions.id, [...rejectedIds, context.action.id]));
+      await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, [...rejectedIds, context.action.id]));
+    }
+  });
+
+  it("restarts an inbound notice cursor when privacy is restored during a bounded pass", async () => {
+    const context = await deferredChatQueueFixture(false, undefined, true);
+    const rejectedIds = Array.from({ length: 100 }, () => randomUUID());
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    let held = false, release!: () => void;
+    const pending = new Promise<void>(resolve => { release = resolve; });
+    let ready: Promise<unknown> | undefined;
+    const reconcile = context.service.reconcileInboundWakeNotices.bind(context.service);
+    const pass = vi.spyOn(context.service, "reconcileInboundWakeNotices").mockImplementationOnce(async input => {
+      const result = await reconcile(input);
+      expect(result).toMatchObject({ inserted: 0, nextCursor: expect.any(Object) });
+      held = true; await pending; return result;
+    });
+    try {
+      await db.update(issues).set({ visibility: "private", privacyRootIssueId: context.conversation.issueId,
+        createdByUserId: null, responsibleUserId: null, assigneeUserId: null,
+      }).where(eq(issues.id, context.conversation.issueId));
+      const [receipt] = await readAgentWakeupRequestsById(context.action.id);
+      const at = new Date(context.action.createdAt.getTime() + 1000).toISOString();
+      await db.insert(chatActions).values(rejectedIds.map(id => ({
+        ...context.action, id, providerActionId: `inbound_wakeup:${id}`,
+        payload: { ...context.action.payload, commentId: randomUUID() }, createdAt: new Date(at),
+      })));
+      await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({
+        ...receipt, id, idempotencyKey: `chat-cursor-reset:${id}`,
+      })));
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+      ready = chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready;
+      await vi.waitFor(() => expect(held).toBe(true), { timeout: 10_000 });
+      await issueService(db).update(context.conversation.issueId, { visibility: "open" });
+      release(); await ready;
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+        expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
+      }, { timeout: 10_000 });
+      const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
+      } finally { select.mockRestore(); transaction.mockRestore(); }
+      expect(errors).toEqual([]);
+    } finally {
+      release(); await ready; await coordinator.stop(); pass.mockRestore(); vi.useRealTimers();
+      await context.service.shutdown();
+      await db.delete(chatActions).where(inArray(chatActions.id, [...rejectedIds, context.action.id]));
+      await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, [...rejectedIds, context.action.id]));
+    }
+  });
+
   it("isolates pre-run FIFO reconciliation scopes and exhausts published history", async () => {
     const first = await deferredChatQueueFixture();
     const second = await deferredChatQueueFixture();
     try {
       const scope = { companyId: first.fixture.companyId, issueId: first.conversation.issueId };
       expect(await first.service.loadInboundWakeNoticeScopes({ companyId: scope.companyId })).toEqual([scope]);
-      expect(await first.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 1, contended: false });
+      expect(await first.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 1, contended: false, nextCursor: null });
       expect(await readChatPublicationsByEndpointId(second.endpoint.id)).toEqual([]);
       expect(await first.service.loadInboundWakeNoticeScopes(scope)).toEqual([]);
-      expect(await first.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: false });
+      expect(await first.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: false, nextCursor: null });
       expect(await second.service.reconcileInboundWakeNotices({
         scope: { companyId: second.fixture.companyId, issueId: second.conversation.issueId },
-      })).toEqual({ inserted: 1, contended: false });
+      })).toEqual({ inserted: 1, contended: false, nextCursor: null });
     } finally {
       await first.service.shutdown(); await second.service.shutdown();
     }
@@ -71547,10 +71671,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     try {
       await ready;
       const scope = { companyId: context.fixture.companyId, issueId: context.conversation.issueId };
-      expect(await context.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: true });
+      expect(await context.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: true, nextCursor: null });
       expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]);
       release(); await holder;
-      expect(await context.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: false });
+      expect(await context.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: false, nextCursor: null });
       expect(await context.service.loadInboundWakeNoticeScopes(scope)).toEqual([]);
     } finally {
       release(); await holder; await context.service.shutdown();

server/src/services/chat-channels/inbound/wakeup-notices.ts

@@ -19,12 +19,12 @@ type NoticeOptions = {
   isShuttingDown: () => boolean;
 };
 export type InboundWakeNoticeScope = { companyId: string; issueId?: string };
-type NoticeCursor = { createdAt: string; id: string };
+export type InboundWakeNoticeCursor = { createdAt: string; id: string };
 type NoticeCandidate = { action: typeof chatActions.$inferSelect; cursorCreatedAt: string };
 
 export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
   const { authorizeInboundWakeup, enqueueFailedChatRetryPublications, authorizeRetainedChatSourcePublication, isExternalActionAuthorizationChange } = options;
-  let inboundWakeNoticeCursor: NoticeCursor | null = null;
+  let inboundWakeNoticeCursor: InboundWakeNoticeCursor | null = null;
   async function currentInboundWakeNotice(
     tx: DbOrTransaction,
     action: typeof chatActions.$inferSelect,
@@ -134,7 +134,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
     return Boolean(link);
   }
 
-  async function noticeCandidates(limit: number, cursor: NoticeCursor | null, scope?: InboundWakeNoticeScope): Promise<NoticeCandidate[]> {
+  async function noticeCandidates(limit: number, cursor: InboundWakeNoticeCursor | null, scope?: InboundWakeNoticeScope): Promise<NoticeCandidate[]> {
     const owner = alias(agentWakeupRequests, "chat_notice_owner");
     // Keep UUID columns indexable. Match only canonical lowercase UUID text,
     // preserving the old comparison while malformed payloads match nothing.
@@ -212,7 +212,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
       .limit(Math.max(1, Math.min(limit, 200)));
   }
 
-  function continuation(candidates: NoticeCandidate[], limit: number): NoticeCursor | null {
+  function continuation(candidates: NoticeCandidate[], limit: number): InboundWakeNoticeCursor | null {
     const last = candidates.at(-1);
     return candidates.length === limit && last
       ? { createdAt: last.cursorCreatedAt, id: last.action.id }
@@ -325,7 +325,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
 
   async function loadInboundWakeNoticeScopes(scope?: InboundWakeNoticeScope, signal?: AbortSignal) {
     const scopes = new Map<string, { companyId: string; issueId: string }>();
-    let cursor: NoticeCursor | null = null;
+    let cursor: InboundWakeNoticeCursor | null = null;
     do {
       signal?.throwIfAborted();
       const candidates = await noticeCandidates(200, cursor, scope);
@@ -340,22 +340,16 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
   }
 
   async function reconcileInboundWakeNotices(input: {
-    scope: InboundWakeNoticeScope; limit?: number; signal?: AbortSignal;
+    scope: InboundWakeNoticeScope; limit?: number; cursor?: InboundWakeNoticeCursor | null; signal?: AbortSignal;
   }) {
-    const limit = Math.max(1, Math.min(input.limit ?? 100, 1000));
-    let inserted = 0;
-    let contended = false;
-    let cursor: NoticeCursor | null = null;
-    do {
-      input.signal?.throwIfAborted();
-      const pageSize = Math.min(200, limit - inserted);
-      const candidates = await noticeCandidates(pageSize, cursor, input.scope);
-      const result = await projectNotices(candidates, input.signal);
-      inserted += result.inserted;
-      contended ||= result.contended;
-      cursor = continuation(candidates, pageSize);
-    } while (cursor && inserted < limit);
-    return { inserted, contended };
+    // Bound authorization work, not successful publications: denied history is
+    // still a candidate check. The worker retains this precise cursor between
+    // passes and resets it when a committed change invalidates the scope.
+    input.signal?.throwIfAborted();
+    const limit = Math.max(1, Math.min(input.limit ?? 100, 200));
+    const candidates = await noticeCandidates(limit, input.cursor ?? null, input.scope);
+    const result = await projectNotices(candidates, input.signal);
+    return { ...result, nextCursor: continuation(candidates, limit) };
   }
 
   async function authorizeInboundWakePublication(

server/src/services/chat-inbound-wake-notice-work.ts

@@ -1,10 +1,13 @@
 import type { ChatChannelService } from "./chat-channels.js";
-import type { InboundWakeNoticeScope } from "./chat-channels/inbound/wakeup-notices.js";
+import type { InboundWakeNoticeCursor, InboundWakeNoticeScope } from "./chat-channels/inbound/wakeup-notices.js";
 import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
 
-type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+type Entry = {
+  id: string; companyId: string; issueId: string; dueAt: number;
+  cursor: InboundWakeNoticeCursor | null; contended: boolean;
+};
 const batchSize = 100;
 
 /** Admission receipts own queued/not-started/removed notices. The channel
@@ -14,13 +17,19 @@ export function chatInboundWakeNoticeWork(service: Pick<ChatChannelService,
   "loadInboundWakeNoticeScopes" | "reconcileInboundWakeNotices">) {
   async function load(scope?: InboundWakeNoticeScope, signal?: AbortSignal): Promise<Entry[]> {
     return (await service.loadInboundWakeNoticeScopes(scope, signal)).map(row => ({
-      ...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0,
+      ...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0, cursor: null, contended: false,
     }));
   }
   async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
-    const result = await service.reconcileInboundWakeNotices({ scope: entry, limit: batchSize, signal });
-    if (result.inserted === batchSize) return { ...entry, dueAt: Date.now() };
-    return result.contended ? { ...entry, dueAt: Date.now() + 1000 } : null;
+    const result = await service.reconcileInboundWakeNotices({
+      scope: { companyId: entry.companyId, issueId: entry.issueId },
+      limit: batchSize, cursor: entry.cursor, signal,
+    });
+    const contended = entry.contended || result.contended;
+    if (result.nextCursor) return { ...entry, cursor: result.nextCursor, contended, dueAt: Date.now() };
+    // Finish the pass before revisiting contention, so a locked prefix cannot
+    // starve later candidates. Do not forget an earlier page's unresolved lock.
+    return contended ? { ...entry, cursor: null, contended: false, dueAt: Date.now() + 1000 } : null;
   }
   function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
     return registerScopedDeadlineWork(coordinator, {