From fee07f3fa8cbc489cb9bbf657a6065c6e00748c3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 13:30:08 -0500 Subject: [PATCH] fix(server): bound inbound notice authorization passes Co-Authored-By: Paperclip --- ...-10-25-inbound-wake-notices-sleep-audit.md | 36 +++++ .../chat-channels.integration.test.ts | 140 +++++++++++++++++- .../chat-channels/inbound/wakeup-notices.ts | 34 ++--- .../services/chat-inbound-wake-notice-work.ts | 21 ++- 4 files changed, 197 insertions(+), 34 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 9d55b99f363..25ff4298028 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 @@ -297,3 +297,39 @@ The corrected call passes 50, preserving the old combined wrapper's default. The full chat run at that revision was deliberately interrupted (exit 130) after this compiler evidence; it is not acceptance. Its log remains at `/tmp/sleep30-wired-full-chat.log`. Fresh verification is required after repair. + + +## Bounded candidate passes after review + +At `87f5c827b0`, workspace typecheck and build passed, and the full chat plus +startup-gate selection passed all 1,486 tests in +`/tmp/sleep30-final-full-chat.log`. CI subsequently passed every code gate; +Greptile found a real fairness defect (comment 4371894496): the reconciliation +limit counted inserted notices rather than candidates authorized, so a denied +history could monopolize one pass. + +The new regression fails before repair in `/tmp/sleep30-fairness-before.log`: +a requested 50-candidate pass opens more than 50 candidate transactions. The +repair returns a precise continuation cursor after at most the requested page +size. The notice worker retains that cursor between scheduler passes. It also +retains contention from earlier pages and, after reaching the end, revisits it +from the beginning rather than silently discarding locked work. A committed +scope invalidation reconstructs the entry with a fresh cursor; no shared +scheduler or authorization policy changes are needed. + +All 44 selected integration cases pass in +`/tmp/sleep30-fairness-acceptance.log`, including a 150-row denied backlog, +an unrelated task completing between bounded passes, contention on the first +page surviving a later page, and a real privacy-restoration write while a page +is in flight. The latter revisits the formerly denied prefix, publishes exactly +once, and then performs no root SQL for two minutes of fake idle time. Manual +pagination remains separate and retains PostgreSQL microsecond precision. +Server typecheck passed after the cursor contract change. Full latest-head +remote checks and review must be revalidated after pushing the repair. + +The next owner audit must explicitly trace ordinary receipts coalesced into a +failed-run retry receipt: `findExistingDeferredWake` does not itself exclude a +retry owner. `allowRunCoalescing: false` on the retry's own admission is not +proof that a later ordinary admission cannot select it. In particular, inspect +the retry authorization-denial write before assuming it can only affect retry +notices. This is an open producer-audit question, not a verified exemption. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index e14612c4d9e..cc3801b12c1 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/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 | 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(resolve => { locked = resolve; }); + const hold = new Promise(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(resolve => { release = resolve; }); + let ready: Promise | 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(); diff --git a/server/src/services/chat-channels/inbound/wakeup-notices.ts b/server/src/services/chat-channels/inbound/wakeup-notices.ts index 015cde340c2..729ca8885d0 100644 --- a/server/src/services/chat-channels/inbound/wakeup-notices.ts +++ b/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 { + async function noticeCandidates(limit: number, cursor: InboundWakeNoticeCursor | null, scope?: InboundWakeNoticeScope): Promise { 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(); - 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( diff --git a/server/src/services/chat-inbound-wake-notice-work.ts b/server/src/services/chat-inbound-wake-notice-work.ts index c325d96b1d6..3181e409cd2 100644 --- a/server/src/services/chat-inbound-wake-notice-work.ts +++ b/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) { async function load(scope?: InboundWakeNoticeScope, signal?: AbortSignal): Promise { 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 { - 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, canRun: () => boolean) { return registerScopedDeadlineWork(coordinator, {