fix(worker): mark bot sends dropped by the quota gate as failed

apps/worker/__tests__/chat-worker-quota-gate.test.ts

@@ -7,6 +7,7 @@ const mocks = vi.hoisted(() => ({
   isBlockedWorkspace: vi.fn(),
   isBotMessageQuotaReached: vi.fn(),
   processJob: undefined as undefined | ((job: unknown) => Promise<void>),
+  recordQuotaSkippedSend: vi.fn(),
   resolveWorkspaceId: vi.fn(),
   sendFlowStep: vi.fn(),
   sendMessageToChannel: vi.fn(),
@@ -103,6 +104,9 @@ vi.mock("../src/chat/handlers/send-message", () => ({
   sendMessageToChannel: mocks.sendMessageToChannel,
   sendTypingToChannel: mocks.sendTypingToChannel,
 }))
+vi.mock("../src/chat/handlers/record-quota-skipped-send", () => ({
+  recordQuotaSkippedSend: mocks.recordQuotaSkippedSend,
+}))
 vi.mock("../src/chat/handlers/send-messenger-template", () => ({
   sendMessengerTemplateMessage: vi.fn(),
 }))
@@ -222,6 +226,27 @@ describe("chat worker bot message quota gate", () => {
     expect(mocks.sendMessageToChannel).not.toHaveBeenCalled()
   })
 
+  test("records the skipped send so it is not left looking delivered", async () => {
+    mocks.isBotMessageQuotaReached.mockResolvedValue(true)
+    const data = {
+      type: "sendChannelMessage",
+      data: { message: { senderType: "bot" } },
+    }
+
+    await mocks.processJob?.({ id: "job-1", data })
+
+    expect(mocks.recordQuotaSkippedSend).toHaveBeenCalledWith(data)
+  })
+
+  test("records nothing when the send goes ahead below the quota", async () => {
+    await mocks.processJob?.({
+      id: "job-1",
+      data: { type: "sendFlowMessage", data: {} },
+    })
+
+    expect(mocks.recordQuotaSkippedSend).not.toHaveBeenCalled()
+  })
+
   test("invokes a user channel-message handler even when the quota is reached", async () => {
     mocks.isBotMessageQuotaReached.mockResolvedValue(true)
 

apps/worker/__tests__/record-quota-skipped-send.test.ts

@@ -0,0 +1,164 @@
+import { beforeEach, describe, expect, test, vi } from "vitest"
+
+const mocks = vi.hoisted(() => ({
+  recordMessageSendError: vi.fn(),
+  settleCommentAutomationFailure: vi.fn(),
+  loggerError: vi.fn(),
+}))
+
+vi.mock("../src/chat/handlers/send-message", () => ({
+  recordMessageSendError: mocks.recordMessageSendError,
+}))
+vi.mock("../src/lib/comment-automation-anchor", () => ({
+  settleCommentAutomationFailure: mocks.settleCommentAutomationFailure,
+}))
+vi.mock("../src/lib/logger", () => ({
+  logger: { error: mocks.loggerError, info: vi.fn(), warn: vi.fn() },
+}))
+
+const { BOT_MESSAGE_QUOTA_REACHED_ERROR, recordQuotaSkippedSend } =
+  await import("../src/chat/handlers/record-quota-skipped-send")
+
+type JobData = Parameters<typeof recordQuotaSkippedSend>[0]
+
+const createdAt = "2026-11-14T00:00:00.000Z"
+const commentReplyAttributes = {
+  replyToCommentId: "comment-1",
+  commentAutomation: { automationId: "automation-1", replyChannel: "public" },
+}
+
+function channelMessageJob(
+  message: Record<string, unknown>,
+  extra: Record<string, unknown> = {},
+): JobData {
+  return {
+    type: "sendChannelMessage",
+    data: {
+      conversation: { id: "conversation-1", workspaceId: "workspace-1" },
+      contactInbox: { id: "contact-inbox-1" },
+      message: {
+        id: "message-1",
+        clientId: "client-1",
+        createdAt,
+        senderType: "bot",
+        ...message,
+      },
+      ...extra,
+    },
+  } as unknown as JobData
+}
+
+function flowMessageJob(commentAnchor?: Record<string, unknown>): JobData {
+  return {
+    type: "sendFlowMessage",
+    data: { commentAnchor },
+  } as unknown as JobData
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+})
+
+describe("recordQuotaSkippedSend", () => {
+  test("marks an existing channel message failed with the quota reason", async () => {
+    await recordQuotaSkippedSend(
+      channelMessageJob({ contentAttributes: commentReplyAttributes }),
+    )
+
+    expect(mocks.recordMessageSendError).toHaveBeenCalledWith(
+      "message-1",
+      "client-1",
+      "workspace-1",
+      new Date(createdAt),
+      BOT_MESSAGE_QUOTA_REACHED_ERROR,
+      { conversationId: "conversation-1", silent: false },
+    )
+    expect(mocks.settleCommentAutomationFailure).toHaveBeenCalledWith({
+      contentAttributes: commentReplyAttributes,
+      errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR,
+    })
+  })
+
+  test("keeps a bulk broadcast message out of realtime", async () => {
+    await recordQuotaSkippedSend(
+      channelMessageJob(
+        {},
+        { isBulkBroadcast: true, metadata: { type: "broadcast" } },
+      ),
+    )
+
+    expect(mocks.recordMessageSendError).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      BOT_MESSAGE_QUOTA_REACHED_ERROR,
+      { conversationId: "conversation-1", silent: true },
+    )
+  })
+
+  test("settles a comment-automation flow reply failed", async () => {
+    await recordQuotaSkippedSend(
+      flowMessageJob({
+        commentId: "comment-1",
+        replyChannel: "private",
+        automationId: "automation-1",
+      }),
+    )
+
+    expect(mocks.recordMessageSendError).not.toHaveBeenCalled()
+    expect(mocks.settleCommentAutomationFailure).toHaveBeenCalledWith({
+      contentAttributes: {
+        replyToCommentId: "comment-1",
+        commentAutomation: {
+          automationId: "automation-1",
+          replyChannel: "private",
+        },
+      },
+      errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR,
+    })
+  })
+
+  test.each([
+    ["no anchor", undefined],
+    [
+      "an anchor without an automation",
+      { commentId: "comment-1", replyChannel: "private" },
+    ],
+    [
+      "a spent anchor (first DM already delivered)",
+      {
+        commentId: "comment-1",
+        replyChannel: "private",
+        automationId: "automation-1",
+        spent: true,
+      },
+    ],
+  ])("leaves a flow message with %s untouched", async (_name, anchor) => {
+    await recordQuotaSkippedSend(flowMessageJob(anchor))
+
+    expect(mocks.settleCommentAutomationFailure).not.toHaveBeenCalled()
+  })
+
+  test("does nothing for a send whose message is created by the handler", async () => {
+    await recordQuotaSkippedSend({
+      type: "sendChatMessage",
+      data: {},
+    } as unknown as JobData)
+
+    expect(mocks.recordMessageSendError).not.toHaveBeenCalled()
+    expect(mocks.settleCommentAutomationFailure).not.toHaveBeenCalled()
+  })
+
+  test("swallows its own failures so the job is still dropped", async () => {
+    mocks.recordMessageSendError.mockRejectedValueOnce(new Error("db down"))
+
+    await expect(
+      recordQuotaSkippedSend(channelMessageJob({})),
+    ).resolves.toBeUndefined()
+    expect(mocks.loggerError).toHaveBeenCalledWith(
+      expect.objectContaining({ err: expect.any(Error) }),
+      expect.any(String),
+    )
+  })
+})

apps/worker/src/chat/handlers/record-quota-skipped-send.ts

@@ -0,0 +1,92 @@
+import { isBulkOutboundMetadata } from "@chatbotx.io/flow-config"
+import { ChatJobAction, type ChatJobData } from "@chatbotx.io/worker-config"
+import { settleCommentAutomationFailure } from "../../lib/comment-automation-anchor"
+import { logger } from "../../lib/logger"
+import { recordMessageSendError } from "./send-message"
+
+/** The `sendError` / Error Logs detail for a send the bot quota dropped. */
+export const BOT_MESSAGE_QUOTA_REACHED_ERROR = "Bot message limit reached"
+
+/**
+ * Records a bot send that the chat worker's quota gate dropped, so it does not
+ * keep looking delivered.
+ *
+ * - `sendChannelMessage`: the outgoing `Message` row already exists (a public
+ *   comment reply is written to the inbox before it is enqueued), so it gets a
+ *   `sendError` and a realtime `messageFailed`, and its comment-automation event
+ *   flips from the optimistic `sent` to `failed`.
+ * - `sendFlowMessage` carrying a comment-automation anchor: no message exists
+ *   yet, so only the automation event is settled `failed`. A `spent` anchor
+ *   means the comment's first DM already went out, so that event stays as is.
+ * - Every other bot send creates its message inside the handler, so there is
+ *   nothing to mark.
+ *
+ * Never throws: the gate must still drop the job without a retry.
+ */
+export async function recordQuotaSkippedSend(
+  jobData: ChatJobData,
+): Promise<void> {
+  try {
+    if (jobData.type === ChatJobAction.sendChannelMessage) {
+      await recordSkippedChannelMessage(jobData.data)
+      return
+    }
+    if (jobData.type === ChatJobAction.sendFlowMessage) {
+      await recordSkippedFlowMessage(jobData.data)
+    }
+  } catch (err) {
+    logger.error(
+      { err, jobType: jobData.type },
+      "Failed to record a send skipped by the bot message quota",
+    )
+  }
+}
+
+async function recordSkippedChannelMessage(
+  data: Extract<
+    ChatJobData,
+    { type: typeof ChatJobAction.sendChannelMessage }
+  >["data"],
+): Promise<void> {
+  const { conversation, message, metadata, isBulkBroadcast } = data
+  const isBulkOutbound =
+    isBulkOutboundMetadata(metadata, isBulkBroadcast) ||
+    isBulkOutboundMetadata(message.contentAttributes?.metadata, isBulkBroadcast)
+
+  await recordMessageSendError(
+    message.id,
+    message.clientId,
+    conversation.workspaceId,
+    message.createdAt ? new Date(message.createdAt) : undefined,
+    BOT_MESSAGE_QUOTA_REACHED_ERROR,
+    { conversationId: conversation.id, silent: isBulkOutbound },
+  )
+  await settleCommentAutomationFailure({
+    contentAttributes: message.contentAttributes,
+    errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR,
+  })
+}
+
+async function recordSkippedFlowMessage(
+  data: Extract<
+    ChatJobData,
+    { type: typeof ChatJobAction.sendFlowMessage }
+  >["data"],
+): Promise<void> {
+  const anchor = data.commentAnchor
+  if (!anchor?.automationId || anchor.spent) {
+    return
+  }
+
+  // The shape `readCommentAutomationAnchor` reads, as `sendFlowStep` stamps it.
+  await settleCommentAutomationFailure({
+    contentAttributes: {
+      replyToCommentId: anchor.commentId,
+      commentAutomation: {
+        automationId: anchor.automationId,
+        replyChannel: anchor.replyChannel,
+      },
+    },
+    errorDetail: BOT_MESSAGE_QUOTA_REACHED_ERROR,
+  })
+}

apps/worker/src/chat/worker.ts

@@ -15,6 +15,7 @@ import { resolveWorkspaceId } from "../lib/resolve-workspace-id"
 import { runJobWithAuditContext } from "../lib/run-job-with-audit-context"
 import { runWorker } from "../lib/shutdown"
 import { checkOutboundAutomatedResponse } from "./handlers/outbound-automated-response"
+import { recordQuotaSkippedSend } from "./handlers/record-quota-skipped-send"
 import { sendChatMessage, sendFlowStep } from "./handlers/send-flow-step"
 import {
   changeMessageStateOnChannel,
@@ -76,6 +77,7 @@ function startChatWorker() {
           { jobId: job.id, workspaceId },
           "Skipping bot send — quota reached",
         )
+        await recordQuotaSkippedSend(job.data)
         return
       }