From 276ef23d2a0b01b15e6bfb8ff31ab13495741b19 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 15:39:40 -0500 Subject: [PATCH] fix(server): stop continuation publication sends at admission boundaries Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 8 +++ ...25-interaction-continuation-sleep-audit.md | 28 +++++++++ .../chat-channels.integration.test.ts | 58 +++++++++++++++++++ .../chat-interaction-publications.test.ts | 18 ++++-- server/src/services/chat-channels.ts | 2 +- .../interactions/continuations.ts | 4 +- .../chat-channels/publications/queue.ts | 12 ++-- 7 files changed, 117 insertions(+), 13 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 fc7f1fbc68e..23b76f07917 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1867,3 +1867,11 @@ and future-retry publication behavior. Eleven new real-database cases pass, including outer rollback/commit, historical prefixes, and concurrent response commit. Existing Telegram acceptance also passes. Broad acceptance and fresh PR gates remain pending; no merge or full-stack completion is claimed. + +Slice 33 is [#17653](https://github.com/paperclipai/paperclip/pull/17653), draft +and unmerged. Initial broad acceptance passed 1,523 tests with 14 skipped; +workspace types/build passed. Review found that publication preparation could +start another send after drain/standby/shutdown. Three regressions reproduced +it. Per-send admission/abort checks now pass all 14 new lifecycle cases, and the +previously skipped publication suite plus startup guards pass 27 cases with the +external test database. Fresh corrected-head verification remains required. diff --git a/doc/plans/2026-10-25-interaction-continuation-sleep-audit.md b/doc/plans/2026-10-25-interaction-continuation-sleep-audit.md index 51fe279f134..0b17fd5d3e0 100644 --- a/doc/plans/2026-10-25-interaction-continuation-sleep-audit.md +++ b/doc/plans/2026-10-25-interaction-continuation-sleep-audit.md @@ -103,3 +103,31 @@ new table, expiry policy, attempt cap, or host-sleep exemption was added. Full acceptance, workspace typecheck/build, and remote review/checks are still required. These local results do not prove actual hosting or database suspension. + +## Review correction: stop before each publication + +At implementation head `6d01f4c730`, workspace typecheck/build passed. Full chat +and startup acceptance passed 1,523 tests in 465.91 seconds, but the 14 standalone +publication tests skipped because their suite supported only embedded PostgreSQL. +This is not evidence those 14 tests passed. The suite now accepts an explicitly +named disposable `_test` external database and closes its own client afterward. + +Greptile review was 4/5 and found a valid stop-boundary gap: the continuation +checked cancellation/admission after preparing publications, but that preparation +could start another send after drain, standby, or shutdown began. The three new +real-queue regressions each reproduced the second publication becoming published +when it should remain pending (`/tmp/sleep33-stop-before.log`). They hold the +first provider edit in flight, change the lifecycle state, and release the edit. + +The existing queue now accepts an optional `canContinue` callback. The +continuation owner passes its current admission/abort check through preparation. +The queue checks before entry, each batch, each reservation, and each newly +started send. Already-started work is joined and settled. Deferred actions and +publications survive resume/restart. Default callers retain their current +behavior, with shutdown also checked before invoking a reserved send. + +All 14 new continuation cases pass after this correction +(`/tmp/sleep33-stop-fixed.log`). The previously skipped publication suite and +startup guards pass separately, 27 cases total +(`/tmp/sleep33-publication-startup.log`). Fresh broad acceptance and remote gates +on the corrected head remain required. No failed run has been relabelled passing. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 567cf50490f..437ea0daf5d 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -45158,6 +45158,64 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); + it.each(["drain", "standby", "shutdown"] as const)("stops starting continuation publications during %s", async mode => { + const context = await telegramConfirmationRecoveryFixture(); + await expect(context.onAction(context.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + const next = await issueThreadInteractionService(db).create( + { id: context.conversation.issueId, companyId: context.fixture.companyId }, + { kind: "request_confirmation", continuationPolicy: "wake_assignee", title: "Next decision", + payload: { version: 1, prompt: "A second decision after the accepted response?" } }, + { agentId: context.fixture.assignedAgentId }, + ); + const nextPublication = () => db.select({ state: chatPublications.state }).from(chatPublications).where( + eq(chatPublications.idempotencyKey, `interaction:${next.id}:${context.endpoint.id}`)); + expect(await nextPublication()).toEqual([{ state: "pending" }]); + const provider = (context.service.runtime as unknown as FakeChatSdkRuntime).endpoints.get(context.endpoint.id)!; + const originalThread = provider.thread.bind(provider); + let entered = false; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const threadSpy = vi.spyOn(provider, "thread").mockImplementation(threadId => { + const thread = originalThread(threadId), edit = thread.adapter.editMessage.bind(thread.adapter); + thread.adapter.editMessage = async (...args) => { entered = true; await gate; return edit(...args); }; + return thread; + }); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const ready = context.service.registerInteractionContinuationWork(coordinator, () => canRun).ready; + let stopping: Promise | undefined; + try { + await vi.waitFor(() => expect(entered).toBe(true), { timeout: 10_000 }); + if (mode === "shutdown") stopping = coordinator.stop(); else canRun = false; + release(); await ready; await stopping; + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(provider.edits).toHaveLength(1); + expect(await nextPublication()).toEqual([{ state: "pending" }]); + expect(context.wakeup).not.toHaveBeenCalled(); + if (mode === "shutdown") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + await context.service.registerInteractionContinuationWork(coordinator, () => canRun).ready; + } else { + canRun = true; + await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(context.wakeup).toHaveBeenCalledTimes(1); + expect(await nextPublication()).toEqual([{ state: "published" }]); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { + release(); await ready; await stopping; await coordinator.stop(); vi.useRealTimers(); + threadSpy.mockRestore(); await context.service.shutdown(); + } + }); + it("defers continuation dispatch until the outer response transaction commits", async () => { const context = await telegramConfirmationRecoveryFixture(); const errors: unknown[] = []; diff --git a/server/src/__tests__/chat-interaction-publications.test.ts b/server/src/__tests__/chat-interaction-publications.test.ts index d99bc1d13dc..b3482c86759 100644 --- a/server/src/__tests__/chat-interaction-publications.test.ts +++ b/server/src/__tests__/chat-interaction-publications.test.ts @@ -27,7 +27,13 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL; +if (externalTestDatabaseUrl && !new URL(externalTestDatabaseUrl).pathname.endsWith("_test")) { + throw new Error("PAPERCLIP_TEST_DATABASE_URL must name a disposable _test database"); +} +const embeddedPostgresSupport = externalTestDatabaseUrl + ? { supported: true } + : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -43,16 +49,18 @@ describeEmbeddedPostgres( beforeAll(async () => { process.env.PAPERCLIP_PUBLIC_URL = "https://paperclip.example"; - tempDb = await startEmbeddedPostgresTestDatabase( - "paperclip-terminal-chat-interaction-", - ); - db = createDb(tempDb.connectionString); + if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl); + else { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-terminal-chat-interaction-"); + db = createDb(tempDb.connectionString); + } }, 20_000); afterAll(async () => { if (previousPublicUrl === undefined) delete process.env.PAPERCLIP_PUBLIC_URL; else process.env.PAPERCLIP_PUBLIC_URL = previousPublicUrl; + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 773fa34658e..ef9dfe40f62 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -3479,7 +3479,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); const { processSelectedPublication } = publicationWorker; const { processPendingInteractionWakeups, registerInteractionContinuationWork } = createChatInteractionContinuations(db, { - preparePublications: scope => processPendingPublications(25, { maintenance: false, scope }), + preparePublications: (scope, canContinue) => processPendingPublications(25, { maintenance: false, scope, canContinue }), redactError, CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS, get heartbeat() { return options.heartbeat; }, }); const { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions, registerTerminalConfirmationWork } = createChatInteractionCallbacks(db, { diff --git a/server/src/services/chat-channels/interactions/continuations.ts b/server/src/services/chat-channels/interactions/continuations.ts index 43185370207..82d153668e7 100644 --- a/server/src/services/chat-channels/interactions/continuations.ts +++ b/server/src/services/chat-channels/interactions/continuations.ts @@ -39,7 +39,7 @@ import { issueThreadInteractionService } from "../../issue-thread-interactions.j type InteractionContinuationsOptions = { heartbeat: IssueAssignmentWakeupDeps; - preparePublications: (scope: { companyId: string; endpointId: string; conversationId: string }) => Promise; + preparePublications: (scope: { companyId: string; endpointId: string; conversationId: string }, canContinue: () => boolean) => Promise; redactError: (error: unknown) => string; CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS: number; }; @@ -86,7 +86,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC for (const candidate of candidates) { if (preparePublications && candidate.conversationId) { await options.preparePublications({ companyId: candidate.companyId, - endpointId: candidate.endpointId, conversationId: candidate.conversationId }); + endpointId: candidate.endpointId, conversationId: candidate.conversationId }, canContinue); } if (!canContinue()) break; const writeAction = (write: (tx: Parameters[0]>[0]) => PromiseLike) => diff --git a/server/src/services/chat-channels/publications/queue.ts b/server/src/services/chat-channels/publications/queue.ts index be5538fea5b..a1b3b3e7bf3 100644 --- a/server/src/services/chat-channels/publications/queue.ts +++ b/server/src/services/chat-channels/publications/queue.ts @@ -181,16 +181,17 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti async function processPendingPublications( limit = 25, - { waitForCompletion = true, maintenance = true, scope }: { - waitForCompletion?: boolean; maintenance?: boolean; + { waitForCompletion = true, maintenance = true, scope, canContinue = () => true }: { + waitForCompletion?: boolean; maintenance?: boolean; canContinue?: () => boolean; scope?: { companyId: string; endpointId: string; conversationId: string }; } = {}, ) { - if (isShuttingDown()) return 0; + if (isShuttingDown() || !canContinue()) return 0; if (maintenance) { await enqueueInboundWakeupPublications(); await reconcileTerminalConfirmationActions(limit); } + if (isShuttingDown() || !canContinue()) return 0; const scopeCondition = scope ? and(eq(chatPublications.companyId, scope.companyId), eq(chatPublications.endpointId, scope.endpointId), eq(chatPublications.conversationId, scope.conversationId)) : undefined; const pendingTasks = () => scope @@ -287,7 +288,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti let failed = false; let firstError: unknown; try { - while (attemptedIds.length < limit && !failed && !isShuttingDown()) { + while (attemptedIds.length < limit && !failed && !isShuttingDown() && canContinue()) { if (publicationEndpointTasks.size >= PUBLICATION_ENDPOINT_CONCURRENCY) { if (!waitForCompletion) break; await Promise.race(publicationEndpointTasks.values()); @@ -328,7 +329,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti publicationEndpointTasks.size >= PUBLICATION_ENDPOINT_CONCURRENCY || attemptedIds.length >= limit || failed || - isShuttingDown() + (isShuttingDown() || !canContinue()) ) break; if (publicationEndpointTasks.has(selectedPublication.endpointId)) @@ -340,6 +341,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti // endpoint-exclusive credential fence instead of timing out behind it. const task = Promise.resolve() .then(async () => { + if (isShuttingDown() || !canContinue()) return; await processSelectedPublication(selectedPublication); await settleExternalConversation(db, selectedPublication.companyId, selectedPublication.issueId).catch((err) => { logger.warn({ err, publicationId: selectedPublication.id }, "Slack conversation settlement deferred to reconciliation");