From d6dfbb495db44b60faa37416675e8ffbb9cc10d0 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 12:30:20 -0500 Subject: [PATCH] fix(chat): preserve inbound notice cursor precision Co-Authored-By: Paperclip --- ...-10-25-inbound-wake-notices-sleep-audit.md | 16 ++++++++++ .../chat-channels.integration.test.ts | 31 +++++++++++++++++++ .../chat-channels/inbound/wakeup-notices.ts | 10 +++--- 3 files changed, 52 insertions(+), 5 deletions(-) diff --git a/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md b/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md index 7b82f92c08d..728af07dbd7 100644 --- a/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md +++ b/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md @@ -111,3 +111,19 @@ Do not turn queued work into periodic notice polling after its notice is saved. This audit changes no runtime behavior or Cloud eligibility. Exact implementation, producer coverage and acceptance remain required before a migration PR is ready. + +## Groundwork evidence + +The existing 22 FIFO/notice integration cases pass in +`/tmp/sleep30-inbound-notice-baseline.log`. A new regression then reproduces the +cursor defect: 50 authority-rejected actions share a PostgreSQL microsecond +timestamp before one valid admitted action; two bounded passes both return zero. +Evidence: `/tmp/sleep30-notice-cursor-before.log`. + +Preserving the timestamp as PostgreSQL text repairs that defect without changing +eligibility or batching. All 23 notice cases pass in +`/tmp/sleep30-notice-cursor-fixed.log` on a fresh isolated external PostgreSQL +fixture database. Production typecheck passes in +`/tmp/sleep30-cursor-typecheck.log`. The new test deletes its artificial candidates +so they cannot starve later tests. Scoped worker ownership and complete producer +coverage remain unimplemented; this is groundwork, not a polling-removal claim. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 6e8d77a89c8..d5aae73c39a 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -71126,6 +71126,37 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("advances pre-run FIFO selection past a suppressed page with microsecond timestamps", async () => { + const context = await deferredChatQueueFixture(); + const rejectedIds = Array.from({ length: 50 }, () => randomUUID()); + try { + const [receipt] = await readAgentWakeupRequestsById(context.action.id); + const before = 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: sql`${before}::timestamptz + interval '0.000123 seconds'`, + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ + ...receipt, id, idempotencyKey: `chat-cursor-fixture:${id}`, + }))); + // Rejected current authority must not keep the same full page at the + // head of every bounded pass and starve the later admitted message. + const first = await context.service.enqueueInboundWakeupPublications(50); + const second = await context.service.enqueueInboundWakeupPublications(50); + expect(first + second).toBe(1); + const rows = await readChatPublicationsByEndpointId(context.endpoint.id); + expect(rows).toEqual([expect.objectContaining({ + commentId: context.action.payload.commentId, + payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + } finally { + 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("keeps one durable pre-run FIFO notice across coalescing and restart, then reuses only its successor lane", async () => { const context = await deferredChatQueueFixture(); let restarted: ChatChannelService | null = null; diff --git a/server/src/services/chat-channels/inbound/wakeup-notices.ts b/server/src/services/chat-channels/inbound/wakeup-notices.ts index 3ded0ab4d71..2d8573f608d 100644 --- a/server/src/services/chat-channels/inbound/wakeup-notices.ts +++ b/server/src/services/chat-channels/inbound/wakeup-notices.ts @@ -20,7 +20,7 @@ type NoticeOptions = { }; export function createInboundWakeNotices(db: Db, options: NoticeOptions) { const { authorizeInboundWakeup, enqueueFailedChatRetryPublications, authorizeRetainedChatSourcePublication, isExternalActionAuthorizationChange } = options; - let inboundWakeNoticeCursor: { createdAt: Date; id: string } | null = null; + let inboundWakeNoticeCursor: { createdAt: string; id: string } | null = null; async function currentInboundWakeNotice( tx: DbOrTransaction, action: typeof chatActions.$inferSelect, @@ -162,7 +162,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { // 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 - .select({ action: chatActions }) + .select({ action: chatActions, cursorCreatedAt: sql`${chatActions.createdAt}::text` }) .from(chatActions) .innerJoin( agentWakeupRequests, @@ -193,7 +193,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { and ${agentWakeupRequests.status} not in ('cancelled','failed','skipped') and not (${sourceRemoved})) or ${hasVisibleQueue})`, cursor - ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` + ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt}::timestamptz, ${cursor.id}::uuid)` : undefined, sql`not exists (select 1 from chat_publications notice where notice.company_id = ${chatActions.companyId} @@ -206,10 +206,10 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { ) .orderBy(asc(chatActions.createdAt), asc(chatActions.id)) .limit(Math.max(1, Math.min(limit, 200))); - const last = candidates.at(-1)?.action; + const last = candidates.at(-1); inboundWakeNoticeCursor = candidates.length >= Math.max(1, Math.min(limit, 200)) && last - ? { createdAt: last.createdAt, id: last.id } + ? { createdAt: last.cursorCreatedAt, id: last.action.id } : null; let inserted = retryInserted; for (const { action } of candidates) {