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 728af07dbd7..66efff5c073 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 @@ -127,3 +127,47 @@ 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. + +## Scoped owner and first producers (2026-10-25) + +Parent notification repair `3447973f6f` is merged into this local branch. +The owner now exposes scoped candidate reconstruction and normal-notice-only +reconciliation. Each scoped pass has its own microsecond-precise keyset cursor, +advances past suppressed pages, and reports actual lock contention separately +from settled authorization denial. The manual API retains its bounded cursor +and combined failed-run retry behavior. The new scoped-deadline worker is +implemented but **not yet wired into production startup**; the periodic backstop +remains until all producer and lifecycle coverage is complete. + +The existing lifecycle/authority notification boundary now also invalidates +normal inbound notices. Real chat admission wakes the worker after a quiet +startup, and settled/published history becomes quiet again. Removing that +notification makes the acceptance case fail (no queued publication), proving +the case exercises the producer rather than a manual scan. + +Task-tree cancellation and source-comment tombstoning each reproduced a missed +retirement wake. Tree cancellation now registers intent inside its transaction; +caller-owned transactions retain outer commit/rollback semantics. Tombstoning +reads and locks the still-live comment's company/task binding before registering +intent and updating it; already deleted/missing comments remain no-ops. These +changes preserve existing notice text and current authorization. Tests deliver +the queued publication with maintenance disabled, perform real service mutations +inside a rolled-back outer transaction and then a committed transaction, and +observe the existing provider message being edited once. Two minutes of fake +idle time cause no further root select/transaction calls. + +29 scoped/manual/admission/retirement integration cases passed in +`/tmp/sleep30-notice-producer-acceptance.log`. Producer regression evidence: +`/tmp/sleep30-notice-worker-before-producer.log`, +`/tmp/sleep30-notice-tree-cancel-before.log`, and +`/tmp/sleep30-notice-source-removal-before.log`. Initial worker-test failures +were missing `canRun` in its coordinator fixture and a wait that checked only +the timer while a sweep was still running; the corrected wait also requires +idle-work settlement. These fixture failures are not production defects. +Server typecheck passed in `/tmp/sleep30-notice-producer-typecheck.log`. + +Still required: remaining receipt/cancellation writers (including budget, agent, +company and Slack stop paths), admission-failure settlement, authority changes +not already covered by shared notifications, batching/scoping/lifecycle tests, +production registration, and removal of only this duty from the maintenance +loop. Fresh PR creation and all checks remain outstanding for slice 30. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index d5aae73c39a..610804851f5 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,5 @@ +import { issueTreeControlService } from "../services/issue-tree-control.js"; +import { chatInboundWakeNoticeWork } from "../services/chat-inbound-wake-notice-work.js"; import { nativeChatProgressWork } from "../services/chat-native-progress-work.js"; import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "../middleware/auth.js"; import { cloudTenantPrimaryCompanyId } from "../services/cloud-instance.js"; @@ -71006,7 +71008,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(committed.providerRuntime.slackFilePublicationAttempts).toBe(2); }); - async function deferredChatQueueFixture(admissionFails = false) { + async function deferredChatQueueFixture(admissionFails = false, beforeAdmission?: (service: ChatChannelService) => Promise) { const fixture = await seedCompany(); let ownerId: string | null = null; const configured = await configuredSlackEndpoint(fixture, { @@ -71080,6 +71082,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }), trigger: id === "1999000.1" ? "mention" : "subscribed_message", }); + await beforeAdmission?.(configured.service); await send(); const [conversation] = await readChatConversationsByEndpointId(configured.endpoint.id); const [action] = await db @@ -71126,7 +71129,101 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } - it("advances pre-run FIFO selection past a suppressed page with microsecond timestamps", async () => { + it("wakes pre-run FIFO notices from real admission after quiet and becomes quiet again", async () => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + const errors: unknown[] = []; + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const quiet = async () => { + 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(); } + }; + try { + const context = await deferredChatQueueFixture(false, async channel => { + service = channel; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) }); + await chatInboundWakeNoticeWork(channel).register(coordinator, () => true).ready; + await settle(); await quiet(); + }); + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + commentId: context.action.payload.commentId, + payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + await quiet(); + await coordinator!.stop(); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) }); + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); await quiet(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(errors).toEqual([]); + } finally { + await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); + } + }); + + it.each(["tree cancellation", "source removal"] as const)("wakes a visible pre-run FIFO retirement only after %s commits", async change => { + const context = await deferredChatQueueFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const notices = () => readChatPublicationsByEndpointId(context.endpoint.id); + const expectedText = change === "source removal" ? "This queued message was removed." + : "This follow-up was not started. Open the task in Paperclip for details."; + const retire = (database: typeof db) => change === "source removal" + ? issueService(database).tombstoneComment(String(context.action.payload.commentId), { + actorType: "user", userId: String(context.action.payload.requestedByActorId), + }) + : issueTreeControlService(database).cancelUnclaimedWakeupsForTree( + context.fixture.companyId, context.conversation.issueId, "PRIVATE cancellation", + ); + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + await context.service.processPendingPublications(100, { maintenance: false }); + await settle(); + expect(await notices()).toEqual([expect.objectContaining({ state: "published" })]); + await expect(db.transaction(async tx => { + await retire(tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(1000); + expect(await notices()).toHaveLength(1); + throw new Error("outer cancellation rollback"); + })).rejects.toThrow("outer cancellation rollback"); + await settle(); expect(await notices()).toHaveLength(1); + await retire(db); + await settle(); + expect(await notices()).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: expectedText, progressState: "failed" }, + })])); + expect(await notices()).toHaveLength(2); + await context.service.processPendingPublications(100, { maintenance: false }); + await settle(); + expect(context.runtime.endpoints.get(context.endpoint.id)!.edits).toEqual([expect.objectContaining({ + text: expectedText, + })]); + 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 { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it.each(["manual", "scoped"] as const)("advances pre-run FIFO selection past a suppressed page with microsecond timestamps (%s)", async mode => { const context = await deferredChatQueueFixture(); const rejectedIds = Array.from({ length: 50 }, () => randomUUID()); try { @@ -71142,9 +71239,15 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }))); // 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); + if (mode === "manual") { + const first = await context.service.enqueueInboundWakeupPublications(50); + 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 rows = await readChatPublicationsByEndpointId(context.endpoint.id); expect(rows).toEqual([expect.objectContaining({ commentId: context.action.payload.commentId, @@ -71157,6 +71260,48 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); + 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 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 second.service.reconcileInboundWakeNotices({ + scope: { companyId: second.fixture.companyId, issueId: second.conversation.issueId }, + })).toEqual({ inserted: 1, contended: false }); + } finally { + await first.service.shutdown(); await second.service.shutdown(); + } + }); + + it("retains a contended pre-run FIFO projection and settles cancelled input without a visible notice", async () => { + const context = await deferredChatQueueFixture(); + let release!: () => void; + let locked!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { locked = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, context.action.id)).for("update"); + locked(); await hold; + await tx.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, context.action.id)); + }); + 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 readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]); + release(); await holder; + expect(await context.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: false }); + expect(await context.service.loadInboundWakeNoticeScopes(scope)).toEqual([]); + } finally { + release(); await holder; await context.service.shutdown(); + } + }); + 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.ts b/server/src/services/chat-channels.ts index 58fb4bd8e49..49ee32f4da3 100644 --- a/server/src/services/chat-channels.ts +++ b/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), diff --git a/server/src/services/chat-channels/inbound/wakeup-notices.ts b/server/src/services/chat-channels/inbound/wakeup-notices.ts index 2d8573f608d..015cde340c2 100644 --- a/server/src/services/chat-channels/inbound/wakeup-notices.ts +++ b/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 { - if (options.isShuttingDown()) return 0; - const retryInserted = await enqueueFailedChatRetryPublications(limit); + async function noticeCandidates(limit: number, cursor: NoticeCursor | null, scope?: InboundWakeNoticeScope): Promise { 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`${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 { + 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(); + 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 }; } diff --git a/server/src/services/chat-inbound-wake-notice-notifications.ts b/server/src/services/chat-inbound-wake-notice-notifications.ts new file mode 100644 index 00000000000..12e0aa08422 --- /dev/null +++ b/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 })); +} diff --git a/server/src/services/chat-inbound-wake-notice-work.ts b/server/src/services/chat-inbound-wake-notice-work.ts new file mode 100644 index 00000000000..c325d96b1d6 --- /dev/null +++ b/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) { + async function load(scope?: InboundWakeNoticeScope, signal?: AbortSignal): Promise { + return (await service.loadInboundWakeNoticeScopes(scope, signal)).map(row => ({ + ...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0, + })); + } + 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; + } + function register(coordinator: ReturnType, 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 }; +} diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index d3f38e52d81..6964bd6b069 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/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); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 951672618b3..3ce6f0b905b 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/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", diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index ecf8a07c6fd..d63984d5e10 100644 --- a/server/src/services/issue-tree-control.ts +++ b/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`${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`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds), + ), + ) + .returning({ + id: agentWakeupRequests.id, + agentId: agentWakeupRequests.agentId, + reason: agentWakeupRequests.reason, + payload: agentWakeupRequests.payload, + }); + }); } return { diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index f864bdf0c57..a962a305713 100644 --- a/server/src/services/issues.ts +++ b/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)