diff --git a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md index 78012431f27..eeb159a8e41 100644 --- a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md @@ -579,3 +579,51 @@ slice. Publication/interaction, conversation/configuration, question claim/retry and monitor/semantic receipt producers still need their own audit and proof. The production periodic projector remains; no slice publication or full-goal completion is claimed. + +## Question claim and retry checkpoint + +Question response claim state is publication lineage, not merely delivery +bookkeeping. A delivering answer can temporarily bind its source run to an +already published prompt on another conversation. Returning that claim to +pending restores the original chat lineage and can make a previously suppressed +working notice eligible. + +The question service now registers issue-scoped milestone intent before entering +that claim state and before an owned retry releases it. Retry release locks and +rechecks the existing attempt generation before signaling or updating; exhausted +claims retain ownership until the existing terminal receipt commits. Stale-claim +reset locks the actual stale rows, signals their distinct issue scopes, and resets +only those rows in the same transaction. Updates use batches of 500 IDs to avoid +an unbounded statement parameter count. This preserves the existing reset-all +semantics; it does not make the stale-row scan itself bounded. Live lease renewal, +still-owned exhausted claims and no-op claim attempts add no milestone wake. +Provider cancellation and execution remain outside these transactions. + +Verification: + +- `/tmp/sleep28-question-retry-corrected-before.log` reproduces a missing working + notice after real delivery retry. The callback captures observations for + assertions outside the service's error handler so assertions cannot be swallowed. +- `/tmp/sleep28-question-claims-before.log`: retry and stale reset fail before + the fix; a live claim remains unchanged and emits no notification. +- `/tmp/sleep28-question-final-tests.log`: 76 pass, one test expectation failed: + rollback emits settlement again when shared XID reconciliation confirms abort. + State and publication assertions passed. The corrected test checks intent, + eventual settlement, preserved claim, absent notice and subsequent idle quiet + without assuming settlement notifications are delivered exactly once. +- `/tmp/sleep28-question-final-fixed-tests.log`: all 77 cases pass: 42 milestone + worker cases plus 35 existing question-delivery cases, including lease renewal, + stale-worker fencing, retries and adapter behavior. +- `/tmp/sleep28-question-typecheck.log`: server production typecheck passed. + Test files are outside that compiler configuration. `git diff --check` passed. + +Existing question-delivery tests now support guarded +`PAPERCLIP_QUESTION_DELIVERY_TEST_DATABASE_URL`, retaining embedded mode by default +and closing their client. Their broad cleanup ran only on the dedicated migrated +`paperclip_sleep28_question_delivery_20261025_test` database. Milestone fixtures +used their separate dedicated database. + +Fresh origin/master inspected at `8c6bd5a83b`; reconciliation of the stack remains +outstanding. Publication/interaction, conversation/configuration and monitor +receipt producers remain under audit. Production registration and periodic-loop +removal are still pending, so this checkpoint is not a completed migration PR. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 84f9cf7c634..15e6bb0d3c7 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-work.test.ts @@ -597,6 +597,73 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${targetId}:queued:${f.endpointId}`); await quiet(); }); + it.each(["retry", "stale claim", "live claim", "rolled-back reset"] as const)("rechecks question prompt binding after %s", async (mode) => { + const source = await run(); + const interactions = issueThreadInteractionService(db); + const interaction = await interactions.create({ id: f.issueId, companyId: f.companyId }, { + kind: "ask_user_questions", continuationPolicy: "wake_assignee", sourceRunId: source.id, + title: "Choose", payload: { version: 1, title: "Choose", questions: [{ + id: "choice", prompt: "Continue?", selectionMode: "single", required: true, + options: [{ id: "yes", label: "Yes" }], + }] }, + }, { agentId: f.agentId, runId: source.id }); + await interactions.answerQuestions({ id: f.issueId, companyId: f.companyId, status: "in_progress" }, interaction.id, + { answers: [{ questionId: "choice", optionIds: ["yes"] }] }, { userId: f.userId }); + const other = await fixture(f.companyId); + await db.update(chatEndpoints).set({ assignedAgentId: f.agentId }).where(eq(chatEndpoints.id, other.endpointId)); + await db.update(chatConversations).set({ issueId: f.issueId }).where(eq(chatConversations.id, other.conversationId)); + await db.insert(chatPublications).values({ companyId: f.companyId, endpointId: other.endpointId, conversationId: other.conversationId, + issueId: f.issueId, idempotencyKey: `interaction:${interaction.id}:${other.endpointId}`, + payload: { text: "Continue?", interactionId: interaction.id }, state: "published", providerMessageId: "other-question", + }); + const events: string[] = []; + let claimEvents: string[] = []; + let duringClaimNotices: string[] | undefined; + const unsubscribe = subscribeDatabaseWork(db, "chat-run-milestones", event => events.push(event)); + const delivery = questionResponseDeliveryService(db, { heartbeat: { + cancelRun: async () => { + claimEvents = [...events]; + await start(); await flush(); + duringClaimNotices = (await publications()).map(row => row.idempotencyKey); + throw new Error("temporary cancellation failure"); + }, + wakeup: async () => { throw new Error("Must retain the still-active source"); }, + } }); + try { + if (mode === "retry") { + await expect(delivery.deliver(interaction.id)).resolves.toBeNull(); + } else { + await db.update(issueQuestionResponseDeliveries).set({ status: "delivering", attemptCount: 1, + lastAttemptAt: new Date(Date.now() - (mode === "live claim" ? 0 : 60_000)), + }).where(eq(issueQuestionResponseDeliveries.interactionId, interaction.id)); + await start(); await flush(); + duringClaimNotices = (await publications()).map(row => row.idempotencyKey); + if (mode === "rolled-back reset") { + await expect(db.transaction(async tx => { + await questionResponseDeliveryService(tx as unknown as typeof db, { heartbeat: {} as never }).sweepPending(0); + expect(events).toEqual(["intent"]); + throw new Error("rollback reset"); + })).rejects.toThrow("rollback reset"); + } else { + await delivery.sweepPending(0); + } + } + expect(duringClaimNotices).toBeDefined(); + expect(duringClaimNotices!.some(key => key.startsWith(`run:${source.id}:`))).toBe(false); + expect((await db.select().from(issueQuestionResponseDeliveries).where(eq(issueQuestionResponseDeliveries.interactionId, interaction.id)))[0].status).toBe(mode === "live claim" || mode === "rolled-back reset" ? "delivering" : "pending"); + await flush(); + const noticeExists = (await publications()).some(row => row.idempotencyKey === `run:${source.id}:working:${f.endpointId}`); + expect(noticeExists).toBe(mode === "retry" || mode === "stale claim"); + if (mode === "retry") expect(claimEvents).toEqual(["intent", "settled"]); + else if (mode === "live claim") expect(events).toEqual([]); + else { + expect(events[0]).toBe("intent"); + expect(events.slice(1)).toContain("settled"); + expect(events.filter(event => event === "intent")).toHaveLength(1); + } + await quiet(); + } finally { unsubscribe(); } + }); it("waits for outer commit and ignores rolled-back nested writes", async () => { await start(); await flush(); await expect(db.transaction(async tx => { diff --git a/server/src/__tests__/question-response-delivery.test.ts b/server/src/__tests__/question-response-delivery.test.ts index ad3b603ec75..9b8a013c481 100644 --- a/server/src/__tests__/question-response-delivery.test.ts +++ b/server/src/__tests__/question-response-delivery.test.ts @@ -26,6 +26,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { buildQuestionResponseDeliveryEnvelope, @@ -33,7 +34,8 @@ import { questionResponseDeliveryService, } from "../services/question-response-delivery.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_QUESTION_DELIVERY_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -64,10 +66,10 @@ describeEmbeddedPostgres("question response delivery", () => { > | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase( - "paperclip-question-delivery-", - ); - db = createDb(tempDb.connectionString); + if (!externalDatabaseUrl) { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-question-delivery-"); + } + db = createDb(externalDatabaseUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -84,6 +86,7 @@ describeEmbeddedPostgres("question response delivery", () => { }); afterAll(async () => { + await db?.$client.end(); await tempDb?.cleanup(); }); diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index b489f748ca5..6e85adee7ce 100644 --- a/server/src/services/question-response-delivery.ts +++ b/server/src/services/question-response-delivery.ts @@ -455,6 +455,9 @@ export function questionResponseDeliveryService( current.lastAttemptAt.getTime() > claimAt.getTime() - claimStaleMs ) return null; + if (current.status !== "delivering") { + await notifyChatRunMilestones(tx, { companyId: current.companyId, issueId: current.issueId }); + } return tx .update(issueQuestionResponseDeliveries) .set({ @@ -593,25 +596,28 @@ export function questionResponseDeliveryService( const nextErrorCount = delivery.errorCount + (options.bounded ? 1 : 0); const exhausted = options.bounded && nextErrorCount >= MAX_DELIVERY_ATTEMPTS; - await db - .update(issueQuestionResponseDeliveries) - .set({ - // Keep an exhausted claim owned until recordTerminal commits its outcome. - status: exhausted ? "delivering" : "pending", - ...(options.bounded ? { errorCount: nextErrorCount } : {}), - lastErrorCode: errorCode, - updatedAt: at, - }) - .where( - and( - eq(issueQuestionResponseDeliveries.id, delivery.id), - eq(issueQuestionResponseDeliveries.status, "delivering"), - eq( - issueQuestionResponseDeliveries.attemptCount, - delivery.attemptCount, - ), - ), + await db.transaction(async (tx) => { + const ownedClaim = and( + eq(issueQuestionResponseDeliveries.id, delivery.id), + eq(issueQuestionResponseDeliveries.status, "delivering"), + eq(issueQuestionResponseDeliveries.attemptCount, delivery.attemptCount), ); + const [current] = await tx.select({ id: issueQuestionResponseDeliveries.id }) + .from(issueQuestionResponseDeliveries).where(ownedClaim).for("update"); + if (!current) return; + if (!exhausted) { + await notifyChatRunMilestones(tx, { companyId: delivery.companyId, issueId: delivery.issueId }); + } + await tx.update(issueQuestionResponseDeliveries) + .set({ + // Keep an exhausted claim owned until recordTerminal commits its outcome. + status: exhausted ? "delivering" : "pending", + ...(options.bounded ? { errorCount: nextErrorCount } : {}), + lastErrorCode: errorCode, + updatedAt: at, + }) + .where(ownedClaim); + }); return exhausted; } @@ -1303,21 +1309,32 @@ export function questionResponseDeliveryService( async function sweepPending(limit = 50) { const sweepAt = now(); const staleAt = new Date(sweepAt.getTime() - claimStaleMs); - await db - .update(issueQuestionResponseDeliveries) - .set({ - status: "pending", - updatedAt: sweepAt, - }) - .where( - and( - eq(issueQuestionResponseDeliveries.status, "delivering"), - or( - isNull(issueQuestionResponseDeliveries.lastAttemptAt), - lte(issueQuestionResponseDeliveries.lastAttemptAt, staleAt), - ), + await db.transaction(async (tx) => { + const staleClaims = await tx.select({ + id: issueQuestionResponseDeliveries.id, + companyId: issueQuestionResponseDeliveries.companyId, + issueId: issueQuestionResponseDeliveries.issueId, + }).from(issueQuestionResponseDeliveries).where(and( + eq(issueQuestionResponseDeliveries.status, "delivering"), + or( + isNull(issueQuestionResponseDeliveries.lastAttemptAt), + lte(issueQuestionResponseDeliveries.lastAttemptAt, staleAt), ), - ); + )).orderBy(asc(issueQuestionResponseDeliveries.id)).for("update"); + if (staleClaims.length === 0) return; + const notified = new Set(); + for (const claim of staleClaims) { + const scope = `${claim.companyId}:${claim.issueId}`; + if (notified.has(scope)) continue; + await notifyChatRunMilestones(tx, claim); + notified.add(scope); + } + for (let offset = 0; offset < staleClaims.length; offset += 500) { + await tx.update(issueQuestionResponseDeliveries) + .set({ status: "pending", updatedAt: sweepAt }) + .where(inArray(issueQuestionResponseDeliveries.id, staleClaims.slice(offset, offset + 500).map(claim => claim.id))); + } + }); const ids = await db .select({ interactionId: issueQuestionResponseDeliveries.interactionId }) .from(issueQuestionResponseDeliveries)