refactor(server): isolate inbound notice reconciliation and committed triggers

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

server/src/services/chat-channels.ts

@@ -1262,7 +1262,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown,
     get teamsFileUploadRequest() { return options.teamsFileUploadRequest; },
   });
-  const { enqueueInboundWakeupPublications, authorizeInboundWakePublication } = createInboundWakeNotices(db, { authorizeInboundWakeup, enqueueFailedChatRetryPublications, authorizeRetainedChatSourcePublication, isExternalActionAuthorizationChange, isShuttingDown: () => shuttingDown });
+  const { enqueueInboundWakeupPublications, loadInboundWakeNoticeScopes, reconcileInboundWakeNotices, authorizeInboundWakePublication } = createInboundWakeNotices(db, { authorizeInboundWakeup, enqueueFailedChatRetryPublications, authorizeRetainedChatSourcePublication, isExternalActionAuthorizationChange, isShuttingDown: () => shuttingDown });
 
   async function authorizeRetainedChatSourcePublication(
     tx: DbOrTransaction,
@@ -3554,6 +3554,8 @@ return {
     reconcileProviderRuntimes,
     processPendingPublications,
     enqueueInboundWakeupPublications,
+    loadInboundWakeNoticeScopes,
+    reconcileInboundWakeNotices,
     schedulePendingPublications,
     processPendingDeliveries,
     processQueuedDeliveries: () => processPendingDeliveries(25, undefined, false),

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

@@ -18,9 +18,13 @@ type NoticeOptions = {
   isExternalActionAuthorizationChange: (error: unknown) => boolean;
   isShuttingDown: () => boolean;
 };
+export type InboundWakeNoticeScope = { companyId: string; issueId?: string };
+type NoticeCursor = { 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: { createdAt: string; id: string } | null = null;
+  let inboundWakeNoticeCursor: NoticeCursor | null = null;
   async function currentInboundWakeNotice(
     tx: DbOrTransaction,
     action: typeof chatActions.$inferSelect,
@@ -130,11 +134,8 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
     return Boolean(link);
   }
 
-  async function enqueueInboundWakeupPublications(limit = 50): Promise<number> {
-    if (options.isShuttingDown()) return 0;
-    const retryInserted = await enqueueFailedChatRetryPublications(limit);
+  async function noticeCandidates(limit: number, cursor: NoticeCursor | null, scope?: InboundWakeNoticeScope): Promise<NoticeCandidate[]> {
     const owner = alias(agentWakeupRequests, "chat_notice_owner");
-    const cursor = inboundWakeNoticeCursor;
     // Keep UUID columns indexable. Match only canonical lowercase UUID text,
     // preserving the old comparison while malformed payloads match nothing.
     const payloadUuid = (value: SQL) =>
@@ -161,7 +162,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
           ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`;
     // A bounded keyset sweep, not an in-memory work queue. Advancing even past
     // revoked candidates prevents an inaccessible old message starving others.
-    const candidates = await db
+    return db
       .select({ action: chatActions, cursorCreatedAt: sql<string>`${chatActions.createdAt}::text` })
       .from(chatActions)
       .innerJoin(
@@ -175,6 +176,9 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
       .where(
         and(
           eq(chatActions.kind, "inbound_wakeup"),
+          sql`${payloadIssueId} is not null`,
+          scope ? eq(chatActions.companyId, scope.companyId) : undefined,
+          scope?.issueId ? eq(payloadIssueId, scope.issueId) : undefined,
           inArray(chatActions.status, ["processed", "failed"]),
           eq(owner.companyId, chatActions.companyId),
           or(
@@ -206,13 +210,21 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
       )
       .orderBy(asc(chatActions.createdAt), asc(chatActions.id))
       .limit(Math.max(1, Math.min(limit, 200)));
+  }
+
+  function continuation(candidates: NoticeCandidate[], limit: number): NoticeCursor | null {
     const last = candidates.at(-1);
-    inboundWakeNoticeCursor =
-      candidates.length >= Math.max(1, Math.min(limit, 200)) && last
-        ? { createdAt: last.cursorCreatedAt, id: last.action.id }
-        : null;
-    let inserted = retryInserted;
+    return candidates.length === limit && last
+      ? { createdAt: last.cursorCreatedAt, id: last.action.id }
+      : null;
+  }
+
+  async function projectNotices(candidates: NoticeCandidate[], signal?: AbortSignal) {
+    let inserted = 0;
+    let contended = false;
     for (const { action } of candidates) {
+      signal?.throwIfAborted();
+      if (options.isShuttingDown()) throw new Error("Inbound wake notice reconciliation stopped");
       try {
         inserted += await db.transaction(async (tx) => {
           const [receipt] = await tx
@@ -292,15 +304,58 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
           return rows.length;
         });
       } catch (error) {
-        if (
-          !isExternalActionAuthorizationChange(error) &&
-          !isExternalChatWaitAuthorizationContention(error) &&
-          !(error instanceof NativeChatReviewPresentationContentionError)
-        )
-          throw error;
+        if (isExternalChatWaitAuthorizationContention(error)
+          || error instanceof NativeChatReviewPresentationContentionError) contended = true;
+        else if (!isExternalActionAuthorizationChange(error)) throw error;
       }
     }
-    return inserted;
+    return { inserted, contended };
+  }
+
+  /** The explicit/manual API keeps its bounded cursor and combined retry-notice
+   * behavior. Background owners reconcile the two duties independently. */
+  async function enqueueInboundWakeupPublications(limit = 50): Promise<number> {
+    if (options.isShuttingDown()) return 0;
+    const retryInserted = await enqueueFailedChatRetryPublications(limit);
+    const pageSize = Math.max(1, Math.min(limit, 200));
+    const candidates = await noticeCandidates(pageSize, inboundWakeNoticeCursor);
+    inboundWakeNoticeCursor = continuation(candidates, pageSize);
+    return retryInserted + (await projectNotices(candidates)).inserted;
+  }
+
+  async function loadInboundWakeNoticeScopes(scope?: InboundWakeNoticeScope, signal?: AbortSignal) {
+    const scopes = new Map<string, { companyId: string; issueId: string }>();
+    let cursor: NoticeCursor | null = null;
+    do {
+      signal?.throwIfAborted();
+      const candidates = await noticeCandidates(200, cursor, scope);
+      for (const { action } of candidates) {
+        // The candidate SQL already requires a canonical UUID issue binding.
+        const issueId = String(action.payload.issueId);
+        scopes.set(`${action.companyId}:${issueId}`, { companyId: action.companyId, issueId });
+      }
+      cursor = continuation(candidates, 200);
+    } while (cursor);
+    return [...scopes.values()];
+  }
+
+  async function reconcileInboundWakeNotices(input: {
+    scope: InboundWakeNoticeScope; limit?: number; 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 };
   }
 
   async function authorizeInboundWakePublication(
@@ -380,5 +435,5 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
       throw error;
     }
   }
-  return { enqueueInboundWakeupPublications, authorizeInboundWakePublication };
+  return { enqueueInboundWakeupPublications, loadInboundWakeNoticeScopes, reconcileInboundWakeNotices, authorizeInboundWakePublication };
 }

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

@@ -0,0 +1,7 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+import type { InboundWakeNoticeScope } from "./chat-channels/inbound/wakeup-notices.js";
+
+export function notifyChatInboundWakeNotices(tx: object, scope: InboundWakeNoticeScope) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices,
+    JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
+}

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

@@ -0,0 +1,35 @@
+import type { ChatChannelService } from "./chat-channels.js";
+import type { 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 };
+const batchSize = 100;
+
+/** Admission receipts own queued/not-started/removed notices. The channel
+ * service retains authorization and publication policy; this owner schedules
+ * only unfinished batches, contention, and failed reconciliation. */
+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,
+    }));
+  }
+  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;
+  }
+  function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.chatInboundWakeNotices,
+      decodeScope: (hint: string): InboundWakeNoticeScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: Entry, scope?: InboundWakeNoticeScope) => !scope ||
+        (entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId)),
+      load, check,
+    }, canRun);
+  }
+  return { load, check, register };
+}

server/src/services/chat-run-milestone-notifications.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
 import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
@@ -10,4 +11,6 @@ export async function notifyChatRunMilestones(tx: object, scope: ChatRunMileston
   // Lifecycle and coarse native progress share lineage and authority. Extend
   // this eligibility boundary once rather than adding wrappers to its callers.
   await notifyChatNativeProgress(tx, scope);
+  // Admission and current chat authority also govern saved queue notices.
+  await notifyChatInboundWakeNotices(tx, scope);
 }

server/src/services/delivery-work-notifications.ts

@@ -55,6 +55,7 @@ export const DELIVERY_QUEUES = {
   chatInbound: "chat-inbound",
   chatPublications: "chat-publications",
   chatNativeProgress: "chat-native-progress",
+  chatInboundWakeNotices: "chat-inbound-wake-notices",
   chatRunMilestones: "chat-run-milestones",
   chatReceipts: "chat-slack-receipts",
   browser: "browser-use",

server/src/services/issue-tree-control.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
 import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
@@ -1178,28 +1179,31 @@ export function issueTreeControlService(db: Db) {
     const issueIds = treeIssues.map((issue) => issue.id);
     if (issueIds.length === 0) return [];
     const now = new Date();
-    return db
-      .update(agentWakeupRequests)
-      .set({
-        status: "cancelled",
-        finishedAt: now,
-        error: reason,
-        updatedAt: now,
-      })
-      .where(
-        and(
-          eq(agentWakeupRequests.companyId, companyId),
-          inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
-          isNull(agentWakeupRequests.runId),
-          inArray(sql<string | null>`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds),
-        ),
-      )
-      .returning({
-        id: agentWakeupRequests.id,
-        agentId: agentWakeupRequests.agentId,
-        reason: agentWakeupRequests.reason,
-        payload: agentWakeupRequests.payload,
-      });
+    return db.transaction(async tx => {
+      await notifyChatInboundWakeNotices(tx, { companyId });
+      return tx
+        .update(agentWakeupRequests)
+        .set({
+          status: "cancelled",
+          finishedAt: now,
+          error: reason,
+          updatedAt: now,
+        })
+        .where(
+          and(
+            eq(agentWakeupRequests.companyId, companyId),
+            inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
+            isNull(agentWakeupRequests.runId),
+            inArray(sql<string | null>`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds),
+          ),
+        )
+        .returning({
+          id: agentWakeupRequests.id,
+          agentId: agentWakeupRequests.agentId,
+          reason: agentWakeupRequests.reason,
+          payload: agentWakeupRequests.payload,
+        });
+    });
   }
 
   return {

server/src/services/issues.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
 import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
 import { notifyApprovalWork } from "./approval-work-notifications.js";
 import { notifyDecisionTargetWork } from "./decision-target-work.js";
@@ -12485,6 +12486,11 @@ export function issueService(db: Db) {
       };
 
       return db.transaction(async (tx) => {
+        const [source] = await tx.select({ companyId: issueComments.companyId, issueId: issueComments.issueId })
+          .from(issueComments).where(and(eq(issueComments.id, commentId), isNull(issueComments.deletedAt)))
+          .for("update");
+        if (!source) return null;
+        await notifyChatInboundWakeNotices(tx, source);
         const now = new Date();
         const [comment] = await tx
           .update(issueComments)