From d4f7426bb40d273acc98449855dc496a9990d9c7 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 13:16:10 -0500 Subject: [PATCH] fix(server): wake inbound queue notices from committed changes Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 16 + ...-10-25-inbound-wake-notices-sleep-audit.md | 69 +++++ .../app-private-hostname-gate.test.ts | 5 + .../chat-channels.integration.test.ts | 285 +++++++++++++++++- server/src/app.ts | 2 + .../modules/wake-queue/adapters/postgres.ts | 3 + .../adapters/queued-comment-postgres.ts | 2 + server/src/routes/issues.ts | 21 +- server/src/services/chat-channels.ts | 4 +- .../services/chat-channels/inbound/wakeups.ts | 5 + .../chat-channels/slack/session-stop.ts | 2 + server/src/services/companies.ts | 4 + server/src/services/heartbeat/run-control.ts | 52 ++-- server/src/services/issues.ts | 7 + server/src/services/projects.ts | 13 +- 15 files changed, 432 insertions(+), 58 deletions(-) diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index fd3fbc97208..015a2bde75a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1737,3 +1737,19 @@ base-publication hook. The hook fixes a reproduced commit-order wake gap; its 16 cases at `bcf2f38ff5`; final typecheck passes too. Remote checks/review are pending. Parent #17606 now has passing checks and Greptile 5/5 on `ed76256b97`; all three review threads are resolved. This is not completion of the broader stack. + +## Current gate update: native progress and inbound notices (2026-10-25) + +#17639 at `6c40618ec1bb098e1c7fe4098133d1de71d287ee` now has all checks +passing (CI run 40218349023), a fresh Greptile 5/5 review of that commit, and +zero review threads. The circular signal-payload and over-wide atomic-event +selection failures were repaired, not bypassed by rerunning unchanged failures. +The PR remains unmerged. + +Slice 30 now registers its scoped notice worker and removes normal inbound +notices from periodic publication maintenance. Focused acceptance passes 42 +cases. Its full verification and new PR are still pending. See the dated inbound +wake notice audit for the producer inventory, regressions and fixture corrections. +This update does not clear the older Grok/runtime-discovery CI failures, voice +session recovery review, or inbound voice cleanup ownership findings, and it does +not shrink the remaining original checklist. 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 7032db01107..e35983cf042 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 @@ -220,3 +220,72 @@ fresh `paperclip_sleep30_full_chat_20261025_test` database, with output at `/tmp/sleep30-full-chat-acceptance.log`; its result must be read before claiming acceptance. Typecheck after merging the second parent repair passed in `/tmp/sleep30-after-parent-typecheck.log`. + +## Producer and lifecycle completion (2026-10-25) + +The full chat suite at `c579148977` finished successfully: 1,458 tests in +`/tmp/sleep30-full-chat-acceptance.log`. This was the baseline before the +cancellation and privacy changes below, not acceptance of later edits. + +Real service regressions reproduced stale visible queue notices after agent +cancellation, budget cancellation, and company archival. Signals now belong to +those owners' transactions, as well as Slack Stop and deferred queue cancellation +and failure. Agent cancellation now rechecks pending status and absence of a run +at update time, so it cannot cancel a receipt promoted after candidate selection. +Caller-owned transactions preserve outer commit and rollback. Company lifecycle +intentionally owns its root transaction; its test respects that contract. + +Task status/assignment changes and privacy changes can restore eligibility. +Task privacy can affect descendants, so it invalidates the company scope. Explicit +task grants and private-project membership use their existing persistence owners; +the project-member helper owns a small transaction that also nests under callers. +Project visibility changes use the existing project transaction. Shared chat +identity, endpoint, resource and membership signals already cover the other +current-authority predicates. Ordinary health refreshes and receipt cursor touches +do not signal this worker. Run-bound settlement and Dot/Muse consumption do not +create a not-started notice; their existing response/progress owners retain that +policy. Failed-run retry notices remain a separate owner. + +Failed inbound-action settlement and repaired delivery acceptance now invalidate +normal notices when a valid durable request exists. Preparing actions whose +underlying deliveries were rejected cannot satisfy the notice predicate and do +not acquire a notice obligation. Publication completion already signals the task +inside the transaction that saves the provider receipt and message link. A new +race test holds provider I/O, removes the source, observes the notice worker +settle before the receipt exists, and proves that the eventual receipt commit +reactivates the removed-source edit. + +The app now registers `chatInboundWakeNoticeWork` with the existing coordinator. +`processPublicationMaintenance` retains only the separate failed-run retry notice +pass, plus its other unmigrated duties. The explicit/manual combined +`enqueueInboundWakeupPublications` API remains available. + +Focused acceptance at the current source passes 42 cases in +`/tmp/sleep30-wired-acceptance.log`. These include actual admission after idle, +scoped/manual cursor pagination, denied history, receipt-lock contention, +transaction rollback and commit, queue rejection/failure, agent/budget/company +cancellation, privacy restoration through task grants and project membership, +signed Slack Stop, late provider receipt, restart deduplication, writes after +shutdown, first write and failed selection during drain, and a commit arriving +while an empty selection returns. Quiet assertions observe root SQL entry points +after advancing fake time by two minutes. The lifecycle tests do not prove host +or database suspension, and no Cloud sleep-safety blocker is relaxed. + +Preserved failures and fixture corrections: +- `/tmp/sleep30-cancellation-before.log`: agent cancellation missed its wake; + an attempted agent reassignment fixture was rejected by the existing external + channel binding rule. The unsupported fixture was removed, not bypassed. +- `/tmp/sleep30-budget-company-before.log`: budget cancellation missed its wake; + company lifecycle correctly rejected a caller transaction. The root-company + reproduction in `/tmp/sleep30-company-root-before.log` then showed its missing + wake. All five initial retirement cases passed after the repair. +- `/tmp/sleep30-notice-privacy-before.log`: opening a private task did not wake + its saved notice. The repair passes. Initial grant tests used an unlinked + external principal as a user; corrected fixtures establish a linked board user + through the normal chat setup before admission. +- The first signed Stop fixture had an invalid Slack channel ID and a verifying, + rather than qualified active, endpoint. Correcting both makes the signed event + reach cancellation. `/tmp/sleep30-signed-stop-qualified.log` passes. + +Workspace typecheck/build, full acceptance after startup wiring, PR publication, +and latest-head CI/review are still required before this slice is ready. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 77cf0d1e5c8..29b46cddee7 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -79,6 +79,7 @@ describe("createChatReconciliationCoordinator", () => { expect(maintenance).not.toContain("chatChannels.voice."); expect(maintenance).not.toContain("enqueueSafeNativeChatProgress"); expect(source).toContain("nativeChatProgressWork(db).register(deliveryWork,"); + expect(source).toContain("chatInboundWakeNoticeWork(chatChannels).register(deliveryWork,"); expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,"); const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8"); @@ -87,6 +88,10 @@ describe("createChatReconciliationCoordinator", () => { expect(teams).not.toContain("selectDistinct"); const chat = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8"); expect(chat).not.toContain("scheduleTeamsFileMaintenance"); + const publicationMaintenance = chat.slice(chat.indexOf("async function processPublicationMaintenance("), + chat.indexOf("async function schedulePendingPublications(")); + expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications("); + expect(publicationMaintenance).toContain("await enqueueFailedChatRetryPublications("); }); it("separates live voice recovery from the remaining periodic maintenance", () => { diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 610804851f5..e14612c4d9e 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -178,6 +178,9 @@ import { } from "../services/chat-teams-personal-recipient.js"; import * as discordQuestionForms from "../services/chat-discord-question-forms.js"; import { issueService } from "../services/issues.js"; +import { companyService } from "../services/companies.js"; +import { ensureProjectAccessMember } from "../services/projects.js"; +import { createPostgresWakeQueueAdapter } from "../modules/wake-queue/adapters/postgres.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { getExternalChannelBindingSummary } from "../services/chat-channel-binding.js"; import { PaperclipRunnerToolAuthority } from "../services/native-runtime/paperclip-runner-tool-authority.js"; @@ -71008,10 +71011,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(committed.providerRuntime.slackFilePublicationAttempts).toBe(2); }); - async function deferredChatQueueFixture(admissionFails = false, beforeAdmission?: (service: ChatChannelService) => Promise) { + async function deferredChatQueueFixture(admissionFails = false, beforeAdmission?: (service: ChatChannelService) => Promise, linkedBoardUser = false) { const fixture = await seedCompany(); let ownerId: string | null = null; const configured = await configuredSlackEndpoint(fixture, { + linkedBoardUser, wakeup: async (agentId, opts) => { if (admissionFails) throw new Error("PRIVATE admission failure token=do-not-publish"); @@ -71067,8 +71071,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }, }); const thread = makeThread({ - channelId: "C-DEFERRED-NOTICE", - id: "slack:C-DEFERRED-NOTICE:1999000.1", + channelId: "CDEFERREDNOTICE", + id: "slack:CDEFERREDNOTICE:1999000.1", }); const send = async (id = "1999000.1") => deliverMessage({ @@ -71078,6 +71082,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { message: makeMessage({ id, text: "@maya exact queued request", + userId: linkedBoardUser ? "UBOARD" : undefined, mentioned: id === "1999000.1", }), trigger: id === "1999000.1" ? "mention" : "subscribed_message", @@ -71170,8 +71175,158 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); - it.each(["tree cancellation", "source removal"] as const)("wakes a visible pre-run FIFO retirement only after %s commits", async change => { + it("reconstructs inbound admission committed after the notice worker stops", async () => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + const errors: unknown[] = []; + 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 coordinator.stop(); + }); + 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(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); } + }); + + it.each(["task visibility", "project membership", "task grant"] as const)("wakes a saved inbound notice when %s restores privacy access", async kind => { + const context = await deferredChatQueueFixture(false, undefined, true); + 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 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 }); + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + if (kind === "task visibility") { + await issueService(db).update(context.conversation.issueId, { visibility: "open" }); + } else if (kind === "project membership") { + const [project] = await db.insert(projects).values({ companyId: context.fixture.companyId, + name: "Private notice project", visibility: "private", + }).returning(); + await db.update(issues).set({ projectId: project.id }).where(eq(issues.id, context.conversation.issueId)); + await ensureProjectAccessMember(db, { companyId: context.fixture.companyId, projectId: project.id, + subjectType: "user", subjectId: String(context.action.payload.requestedByActorId), + }); + } else { + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { req.actor = { ...boardActor(context.fixture.companyId), isInstanceAdmin: true }; next(); }); + app.use("/api", issueRoutes(db, createStorageService().storage)); app.use(errorHandler); + await request(app).post(`/api/issues/${context.conversation.issueId}/access-grants`).send({ + subjectType: "user", subjectId: String(context.action.payload.requestedByActorId), + }).expect(201); + } + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it.each(["startup", "first admission", "failed selection"] as const)("retains inbound notices through %s during drain without passive SQL", async kind => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + let canRun = kind !== "startup"; + const errors: unknown[] = []; + try { + const context = await deferredChatQueueFixture(false, async channel => { + service = channel; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const selection = kind === "failed selection" + ? vi.spyOn(channel, "loadInboundWakeNoticeScopes").mockRejectedValueOnce(new Error("temporary notice selection outage")) + : undefined; + try { await chatInboundWakeNoticeWork(channel).register(coordinator, () => canRun).ready; } + finally { selection?.mockRestore(); } + canRun = false; + }); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + 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(); } + canRun = true; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toHaveLength(kind === "failed selection" ? 1 : 0); + } finally { await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); } + }); + + it("retains admission committed while an empty inbound-notice selection is returning", async () => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + let selection: ReturnType | undefined; + let ready: Promise | undefined; + let held = false, release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + const errors: unknown[] = []; + try { + const context = await deferredChatQueueFixture(false, async channel => { + service = channel; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const load = channel.loadInboundWakeNoticeScopes.bind(channel); + selection = vi.spyOn(channel, "loadInboundWakeNoticeScopes").mockImplementationOnce(async (...args) => { + const rows = await load(...args); held = true; await pending; return rows; + }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + ready = chatInboundWakeNoticeWork(channel).register(coordinator, () => true).ready; + await vi.waitFor(() => expect(held).toBe(true)); + }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + release(); await ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { release(); await ready; selection?.mockRestore(); await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); } + }); + + it.each(["tree cancellation", "source removal", "agent cancellation", "budget cancellation", "company archive", "queue rejection", "queue failure"] as const)("wakes a visible pre-run FIFO retirement only after %s commits", async change => { const context = await deferredChatQueueFixture(); + const releaseRunId = randomUUID(); + if (change === "queue rejection" || change === "queue failure") { + await db.insert(heartbeatRuns).values({ id: releaseRunId, companyId: context.fixture.companyId, + agentId: context.fixture.assignedAgentId, status: "succeeded", finishedAt: new Date(), + contextSnapshot: { issueId: context.conversation.issueId }, + }); + } const errors: unknown[] = []; const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) }); @@ -71183,26 +71338,51 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { 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), { + const retire = async (database: typeof db) => { + if (change === "source removal") return issueService(database).tombstoneComment(String(context.action.payload.commentId), { actorType: "user", userId: String(context.action.payload.requestedByActorId), - }) - : issueTreeControlService(database).cancelUnclaimedWakeupsForTree( + }); + if (change === "agent cancellation") return heartbeatService(database).cancelInvocationsForAgents( + [context.fixture.assignedAgentId], "PRIVATE cancellation", + ); + if (change === "budget cancellation") return heartbeatService(database).cancelBudgetScopeWork({ + companyId: context.fixture.companyId, scopeType: "company", scopeId: context.fixture.companyId, + }); + if (change === "company archive") return companyService(database).update(context.fixture.companyId, { status: "archived" }); + if (change === "queue rejection" || change === "queue failure") { + const adapter = createPostgresWakeQueueAdapter(database, { + resolveResponsibleUserId: async () => "owner-user", + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), + resolveSessionBeforeForWakeup: async () => null, + }); + return adapter.withIssueExecutionLock({ companyId: context.fixture.companyId, runId: releaseRunId, now: new Date() }, async (_locked, ports) => { + const target = { companyId: context.fixture.companyId, wakeId: context.action.id, now: new Date() }; + if (change === "queue rejection") await ports.transaction.cancelDeferredWake({ ...target, reason: "PRIVATE rejection" }); + else await ports.transaction.failDeferredWake(target); + return { outcome: { kind: "released" }, postCommitEffects: [] }; + }); + } + return 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); + // Company lifecycle owns its root transaction and intentionally rejects + // caller transactions. The other service producers support nesting. + if (change !== "company archive") { + 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({ @@ -71223,6 +71403,81 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } }); + it("rechecks source retirement after a queued notice's late provider receipt commits", async () => { + 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 }); + let held = false, release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + let publishing: Promise | undefined; + const provider = context.runtime.endpoints.get(context.endpoint.id)!; + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + provider.postHook = async () => { held = true; await pending; }; + publishing = context.service.processPendingPublications(100, { maintenance: false }); + await vi.waitFor(() => expect(held).toBe(true)); + await issueService(db).tombstoneComment(String(context.action.payload.commentId), { + actorType: "user", userId: "owner-user", + }); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + release(); await publishing; provider.postHook = undefined; + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: "This queued message was removed.", progressState: "failed" }, + })])); + await context.service.processPendingPublications(100, { maintenance: false }); + await settle(); + expect(provider.edits).toEqual([expect.objectContaining({ text: "This queued message was removed." })]); + expect(errors).toEqual([]); + } finally { release(); await publishing; provider.postHook = undefined; await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("wakes inbound notice retirement from a signed Slack Stop without a maintenance scan", async () => { + const context = await deferredChatQueueFixture(false, undefined, true); + await qualifySetupRoundTrip(context.service, context.endpoint.id, "UBOARD"); + await context.service.test(context.endpoint.id, "owner-user"); + 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 }); + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); await context.service.processPendingPublications(100, { maintenance: false }); await settle(); + const body = JSON.stringify({ type: "event_callback", team_id: "T-PAPERCLIP", event_id: "EvSleepQueueStop", + event: { type: "agent_session_stopped", channel: "CDEFERREDNOTICE", thread_ts: "1999000.1", + event_ts: `${Math.floor(Date.now() / 1000)}.999999`, streaming_message_ts: [], user: "UBOARD" }, + }); + const response = await context.service.handleWebhook(context.endpoint.publicId, "slack", signedSlackWebhookRequest({ + body, contentType: "application/json", url: `https://paperclip.example/api/chat-webhooks/${context.endpoint.publicId}/slack`, + })); + expect(response.status).toBe(202); + await settle(); + expect(await db.select({ status: chatActions.status, result: chatActions.result }).from(chatActions) + .where(and(eq(chatActions.endpointId, context.endpoint.id), eq(chatActions.kind, "slack_session_stop")))) + .toEqual([expect.objectContaining({ status: "processed", result: expect.objectContaining({ code: "slack_session_stop_cancelled" }) })]); + expect(await readAgentWakeupRequestsById(context.action.id)).toEqual([expect.objectContaining({ status: "cancelled" })]); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: "This follow-up was not started. Open the task in Paperclip for details.", progressState: "failed" }, + })])); + 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()); diff --git a/server/src/app.ts b/server/src/app.ts index c2813d181b9..8c3ae2daf67 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -196,6 +196,7 @@ import { toolAccessService } from "./services/tool-access.js"; import { chatChannelService } from "./services/chat-channels.js"; import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js"; import { nativeChatProgressWork } from "./services/chat-native-progress-work.js"; +import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-work.js"; import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js"; import { createCoalescedAsyncTrigger, @@ -1288,6 +1289,7 @@ export async function createApp( chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatRunMilestoneWork(db, opts.authPublicBaseUrl).register( deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(), ); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 409d2440dfd..388afd9cf61 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; @@ -312,6 +313,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }, async cancelDeferredWake({ companyId, wakeId, reason, now }) { + await notifyChatInboundWakeNotices(tx, { companyId }); const rows = await tx .update(agentWakeupRequests) .set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now }) @@ -343,6 +345,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }, async failDeferredWake({ companyId, wakeId, now }) { + await notifyChatInboundWakeNotices(tx, { companyId }); const rows = await tx .update(agentWakeupRequests) .set({ diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index e2c0e2f740f..242f9089a3d 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { and, eq, inArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -94,6 +95,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async cancelWake({ wakeId, reason, now }) { + await notifyChatInboundWakeNotices(tx, { companyId, issueId }); await tx .update(agentWakeupRequests) .set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now }) diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 69398e41a4c..f30d0e4ef16 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js"; import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; import { parseObject } from "../adapters/utils.js"; import { prepareRestrictedWorkspaceRetry } from "../services/restricted-workspace-retry.js"; @@ -13373,14 +13374,18 @@ export function issueRoutes( const subjectId = req.body.subjectId as string; await assertIssueGrantSubjectExists(issue.companyId, subjectType, subjectId); const actor = getActorInfo(req); - const result = await db.transaction((tx) => ensureActiveIssueAccessGrant(tx, { - issueId: managedIssue.id, - subjectType, - subjectId, - source: "explicit", - grantedByUserId: actor.actorType === "user" ? actor.actorId : null, - grantedByAgentId: actor.agentId, - })); + const result = await db.transaction(async (tx) => { + // Grants on an ancestor can restore current chat authority on descendants. + await notifyChatInboundWakeNotices(tx, { companyId: issue.companyId }); + return ensureActiveIssueAccessGrant(tx, { + issueId: managedIssue.id, + subjectType, + subjectId, + source: "explicit", + grantedByUserId: actor.actorType === "user" ? actor.actorId : null, + grantedByAgentId: actor.agentId, + }); + }); if (result.created) { await logActivity(db, { companyId: issue.companyId, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 49ee32f4da3..967dc54a417 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -3352,7 +3352,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPublicationMaintenance(limit = 25) { scheduleGitHubMaintenance(); - await enqueueInboundWakeupPublications(); + // Normal inbound notices are commit-driven. Failed-run retries retain their + // separate maintenance owner until its own producer audit is complete. + await enqueueFailedChatRetryPublications(); await reconcileTerminalConfirmationActions(limit); await processPendingInteractionWakeups(limit); } diff --git a/server/src/services/chat-channels/inbound/wakeups.ts b/server/src/services/chat-channels/inbound/wakeups.ts index 9dc1e221c7a..d805036c4b3 100644 --- a/server/src/services/chat-channels/inbound/wakeups.ts +++ b/server/src/services/chat-channels/inbound/wakeups.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js"; import { DELIVERY_PROCESSING_STALE_MS } from "./delivery-ledger.js"; import type { EndpointRow, DbOrTransaction } from "../runtime-types.js"; import { notifyChatRunMilestones } from "../../chat-run-milestone-notifications.js"; @@ -113,6 +114,8 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) { await notifyChatDeliveryWork(tx); if (action.status === "processed") { await notifyChatRunMilestones(tx, { companyId: action.companyId, issueId: String(action.payload.issueId) }); + } else if (action.status === "failed") { + await notifyChatInboundWakeNotices(tx, { companyId: action.companyId, issueId: String(action.payload.issueId) }); } const acceptedAt = new Date(); const accepted = await tx @@ -255,6 +258,8 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) { await notifyChatDeliveryWork(tx); if (status === "processed") { await notifyChatRunMilestones(tx, { companyId: claimed.companyId, issueId: request!.issueId }); + } else if (status === "failed" && request) { + await notifyChatInboundWakeNotices(tx, { companyId: claimed.companyId, issueId: request.issueId }); } const changed = await tx .update(chatActions) diff --git a/server/src/services/chat-channels/slack/session-stop.ts b/server/src/services/chat-channels/slack/session-stop.ts index aef5c731362..ea922411c88 100644 --- a/server/src/services/chat-channels/slack/session-stop.ts +++ b/server/src/services/chat-channels/slack/session-stop.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js"; import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js"; import { dueChatAction, PROVIDER_EFFECT_STALE_MS } from "../../chat-action-work.js"; @@ -584,6 +585,7 @@ export function createSlackSessionStop(db: Db, options: SlackSessionStopOptions) ["queued", "deferred_issue_execution"].includes(wakeup.status) && !issue.executionRunId ) { + await notifyChatInboundWakeNotices(tx, { companyId: current.companyId, issueId: issue.id }); await tx .update(agentWakeupRequests) .set({ diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 580df76b6e9..1534817fa2e 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { isRunnerApiResponseAsset } from "./asset-content-access.js"; @@ -65,6 +66,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), )); + await notifyChatInboundWakeNotices(tx, { companyId: id }); await tx .update(agentWakeupRequests) .set({ @@ -332,6 +334,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyCompanyRetentionWork(tx, id); + await notifyChatInboundWakeNotices(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); } const updated = await tx @@ -484,6 +487,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyCompanyRetentionWork(tx, id); + await notifyChatInboundWakeNotices(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); await tx .update(companies) diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index e9b21a98aa3..76e41c1e422 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { readTaskRetryIntent, reconcileTaskChatRetryReceipts } from "../task-retry.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, readNativeLocalProcessStop, recordedLocalExecutionStopped } from "../native-local-process-stop.js"; @@ -976,6 +977,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont if (wakeupIds.length === 0) return 0; + await notifyChatInboundWakeNotices(tx, { companyId: scope.companyId }); await tx .update(agentWakeupRequests) .set({ @@ -1516,35 +1518,29 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont ); if (uniqueAgentIds.length === 0) return 0; - const now = new Date(); - const wakeupIds = await db - .select({ id: agentWakeupRequests.id }) - .from(agentWakeupRequests) - .where( - and( + return db.transaction(async tx => { + const wakeups = await tx + .select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId }) + .from(agentWakeupRequests) + .where(and( inArray(agentWakeupRequests.agentId, uniqueAgentIds), - inArray(agentWakeupRequests.status, [ - "queued", - "deferred_issue_execution", - ]), - sql`${agentWakeupRequests.runId} is null`, - ), - ) - .then((rows) => rows.map((row) => row.id)); - - if (wakeupIds.length === 0) return 0; - - await db - .update(agentWakeupRequests) - .set({ - status: "cancelled", - finishedAt: now, - error: reason, - updatedAt: now, - }) - .where(inArray(agentWakeupRequests.id, wakeupIds)); - - return wakeupIds.length; + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + isNull(agentWakeupRequests.runId), + )); + if (!wakeups.length) return 0; + for (const companyId of new Set(wakeups.map(row => row.companyId))) { + await notifyChatInboundWakeNotices(tx, { companyId }); + } + const now = new Date(); + const cancelled = await tx.update(agentWakeupRequests).set({ + status: "cancelled", finishedAt: now, error: reason, updatedAt: now, + }).where(and( + inArray(agentWakeupRequests.id, wakeups.map(row => row.id)), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + isNull(agentWakeupRequests.runId), + )).returning({ id: agentWakeupRequests.id }); + return cancelled.length; + }); } async function cancelInvocationsForAgentsInternal( diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index a962a305713..7ec92cff370 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -11194,6 +11194,9 @@ export function issueService(db: Db) { }); } if (changesPrivacy) { + // Privacy inherited through task ancestry can authorize saved notices + // on descendants as well as this task. + await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId }); const nextProjectId = issueData.projectId !== undefined ? issueData.projectId : receiptExisting.projectId; const [privacyProject] = nextProjectId ? await tx.select().from(projects) .where(and(eq(projects.id, nextProjectId), eq(projects.companyId, existing.companyId))) : []; @@ -11275,6 +11278,7 @@ export function issueService(db: Db) { // Invalidate human direction at the common assignment boundary, including // plugin/service writes that do not go through HTTP run cancellation. // Keep the requester attribution for audit; cancellation revokes its use. + await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId }); await tx.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date(), updatedAt: new Date() }) .where(and(eq(agentWakeupRequests.companyId, receiptExisting.companyId), eq(agentWakeupRequests.requestedByActorType, "user"), @@ -11303,6 +11307,9 @@ export function issueService(db: Db) { if (patch.status === "done" || patch.status === "cancelled") { await notifyDecisionTargetWork(tx, `issue:${id}`); } + if (patch.status !== undefined && patch.status !== receiptExisting.status) { + await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId, issueId: id }); + } if (patch.hiddenAt === null && receiptExisting.hiddenAt !== null) { await notifyChatRunMilestones(tx, { companyId: receiptExisting.companyId, issueId: id }); } diff --git a/server/src/services/projects.ts b/server/src/services/projects.ts index 7f6795d2132..210ca148444 100644 --- a/server/src/services/projects.ts +++ b/server/src/services/projects.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { listProjectWorkspaceProviders, resolveProjectWorkspaceProvider } from "./project-workspace-provider.js"; import { retireRemovedPluginWorkspaceDependents } from "./plugin-workspaces.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -80,12 +81,11 @@ export async function ensureProjectAccessMember( subjectId: string; }, ) { - return dbOrTx - .insert(projectAccessMembers) - .values(input) - .onConflictDoNothing() - .returning() - .then((rows: Array) => rows[0] ?? null); + return dbOrTx.transaction(async (tx: Db) => { + await notifyChatInboundWakeNotices(tx, { companyId: input.companyId }); + return tx.insert(projectAccessMembers).values(input).onConflictDoNothing().returning() + .then(rows => rows[0] ?? null); + }); } /** @@ -1013,6 +1013,7 @@ export function projectService(db: Db, options: { captureWorkspaceUpdates?: bool const row = await db.transaction(async (tx) => { if (updates.visibility !== undefined) { + await notifyChatInboundWakeNotices(tx, { companyId: existingProject.companyId }); await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`issue-privacy-tree:${existingProject.companyId}`}, 0))`); } const [updated] = await tx.update(projects).set(updates).where(eq(projects.id, id)).returning();