From b93df4ca382f49adb195d49021ebfaac309a97af Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 20:15:13 -0600 Subject: [PATCH] fix(chat): consume a saved chat wake whose message a completed run answered (#20689) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - External chat (for example GitHub review requests) wakes the assigned agent with a saved chat message > - When the agent is already running, the chat wake is saved behind that run and released after it finishes > - The running agent often reads the new message as part of its current turn, so the message is already answered when the saved wake is released > - The release kept that chat wake anyway, removed its now-delivered message ids, and started a reviewed chat run with no message > - A reviewed chat run with no message cannot be attested, so it failed with `reviewed_chat_execution_binding_not_authorized` and the task became blocked > - This pull request treats a chat message wake like a comment wake: when a completed run already received all its messages, the saved wake is consumed > - The benefit is that a message that arrives during an active run no longer blocks the task after it was answered ## Linked Issues or Issue Description **What happened?** A GitHub chat message arrived for a task while the assigned agent's run was active. The wake was saved behind that run. The run read the message and answered it. When the run finished, the saved chat wake was still promoted. Its message ids were removed because a completed run already received them. The new run had source `chat:` and no message, so reviewed chat attestation denied it with `context_mismatch`. The run failed with `reviewed_chat_execution_binding_not_authorized`. Review-path recovery inherited the same context and failed too, and the task was blocked. **Steps to reproduce** 1. Assign a task in review to an agent that is running. 2. Deliver an external chat message for the task. The wake is saved behind the active run. 3. Let the active run receive the message as wake input and succeed. 4. Release the task's execution lock. 5. See that the saved chat wake is promoted with no message, instead of being consumed. **Expected behavior** The saved chat wake is consumed with "Queued messages were already delivered to a completed run". No new run starts. The chat does not get a "not started" notice for a message that was answered. **Paperclip version or commit** `origin/master` at 6e524d8a69. **Additional context** This was seen on a staging deployment on five runs across three tasks in one day. It is a message timing race, not a deploy or authorization problem. ## What Changed - `server/src/modules/wake-queue/adapters/postgres.ts`: a saved wake whose reason is `External chat message received` and whose payload mutation is `chat_message_received` is treated as message-only. If a completed run already received all of its messages, the release consumes it, the same as a comment wake. If any message is still unread, it is promoted as before. - `server/src/services/chat-inbound-wakeup-publications.ts`: a chat receipt consumed for that reason does not publish "This follow-up was not started". Other declines keep that notice. - `server/src/services/queued-comment-delivery.ts`: constants for the chat message wake reason and the delivered-message reason. - New test `server/src/modules/wake-queue/adapters/chat-wake-delivered.test.ts`. ## Verification - `cd server && npx vitest run src/modules/wake-queue/adapters/chat-wake-delivered.test.ts` - RED on `origin/master`: the release promotes the consumed chat wake ("Unexpected promotion of a consumed chat wake"), and the receipt resolves to a `not_started` notice. - GREEN with this change: the wake is cancelled with the delivered reason, no run starts, no "not started" notice, and a chat wake with an unread message is still promoted. - Related suites: `src/modules/wake-queue`, `heartbeat-reviewed-chat-binding.integration.test.ts`, `durable-chat-wakeup.test.ts`, `heartbeat-native-status-context.test.ts`, `external-chat-wait.integration.test.ts`, `heartbeat-held-comment-wake.test.ts`. - `npx tsc --noEmit -p server` passes. ## Risks - A chat wake that merged with another kind of wake takes the merged wake's reason and mutation. Only a wake whose current reason and mutation are both the chat message values is treated as message-only, so other merged intents keep promotion. - A visible "Your follow-up is queued." notice is not followed by a "not started" notice when the message was already answered. That matches the existing behavior when the wake is promoted. - Not changed: a separate case where messages from two different chat conversations are merged into one saved wake. That case fails with `binding_denied` and needs its own change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code with tool use (shell, file edit, read-only API access). Extended reasoning was used. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude --- .../adapters/chat-wake-delivered.test.ts | 108 ++++++++++++++++++ .../modules/wake-queue/adapters/postgres.ts | 8 +- .../wake-queue/application/use-cases.ts | 3 +- .../src/modules/wake-queue/domain/policy.ts | 3 + server/src/modules/wake-queue/index.ts | 1 + .../services/chat-channels/inbound/wakeups.ts | 7 +- .../chat-inbound-wakeup-publications.ts | 5 + server/src/services/heartbeat/queue.ts | 3 +- .../src/services/queued-comment-delivery.ts | 5 + 9 files changed, 136 insertions(+), 7 deletions(-) create mode 100644 server/src/modules/wake-queue/adapters/chat-wake-delivered.test.ts diff --git a/server/src/modules/wake-queue/adapters/chat-wake-delivered.test.ts b/server/src/modules/wake-queue/adapters/chat-wake-delivered.test.ts new file mode 100644 index 00000000000..438adc20b17 --- /dev/null +++ b/server/src/modules/wake-queue/adapters/chat-wake-delivered.test.ts @@ -0,0 +1,108 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { createReleaseIssueExecution } from "../application/use-cases.js"; +import { createPostgresWakeQueueAdapter } from "./postgres.js"; +import { resolveInboundWakeReceipt } from "../../../services/chat-inbound-wakeup-publications.js"; +import { createDurableChatWakeupRequest } from "../../../services/durable-chat-wakeup.js"; + +const support = await getEmbeddedPostgresTestSupport(); + +// A GitHub message arrives while the agent's run is active. The wake is saved +// behind that run. The run reads the message (it is in its wake input) and +// finishes. Before this fix the saved wake was still promoted, with its message +// ids removed. A reviewed chat run with no message cannot be attested, so it +// failed with `reviewed_chat_execution_binding_not_authorized` (context_mismatch) +// and blocked the task. +describe.skipIf(!support.supported)("release of a saved chat wake whose message was delivered", () => { + let database: Awaited>; + let db: ReturnType; + let companyId: string; + let agentId: string; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("wake-chat-delivered-"); + db = createDb(database.connectionString); + const [company] = await db.insert(companies).values({ name: "Chat wake fixture", issuePrefix: "CWD" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Reviewer", adapterType: "process" }).returning(); + companyId = company.id; + agentId = agent.id; + }, 60_000); + + afterAll(async () => { + await db?.$client.end(); + await database?.cleanup(); + }); + + async function seed(delivered: boolean) { + const [issue] = await db.insert(issues).values({ companyId, title: "Review a pull request", status: "in_review", + assigneeAgentId: agentId }).returning(); + const [message] = await db.insert(issueComments).values({ companyId, issueId: issue.id, authorUserId: "chat-user", + body: "GitHub review request.", createdAt: new Date(Date.now() - 60_000), updatedAt: new Date(Date.now() - 60_000) }).returning(); + // The finishing run started after the message and received it as wake input. + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", + invocationSource: "assignment", triggerDetail: "system", startedAt: new Date(Date.now() - 30_000), finishedAt: new Date(), + contextSnapshot: { issueId: issue.id, source: "chat:github", wakeReason: "External chat message received", + chatConversationId: randomUUID(), ...(delivered ? { wakeCommentIds: [message.id] } : { wakeCommentIds: [randomUUID()] }) }, + }).returning(); + await db.update(issues).set({ executionRunId: run.id, executionLockedAt: new Date() }).where(eq(issues.id, issue.id)); + const context = { issueId: issue.id, taskId: issue.id, source: "chat:github", wakeReason: "External chat message received", + wakeSource: "assignment", chatConversationId: randomUUID(), commentId: message.id, wakeCommentId: message.id, + wakeCommentIds: [message.id] }; + const wakeId = randomUUID(); + const { idempotencyKey } = createDurableChatWakeupRequest({ id: wakeId, companyId, agentId, issueId: issue.id, + commentId: message.id, requestedByActorType: "user", requestedByActorId: "chat-user", requestedAt: new Date(), + authorize: async () => {} }); + const [wake] = await db.insert(agentWakeupRequests).values({ id: wakeId, idempotencyKey, companyId, agentId, source: "assignment", + triggerDetail: "system", reason: "External chat message received", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: "chat-user", + payload: { issueId: issue.id, mutation: "chat_message_received", wakeCommentId: message.id, _paperclipWakeContext: context }, + }).returning(); + return { issue, run, wake }; + } + + function release(unexpected: () => never) { + return createReleaseIssueExecution({ + issueLock: createPostgresWakeQueueAdapter(db, { resolveResponsibleUserId: unexpected, getRoutineEnv: unexpected, + resolveSessionBeforeForWakeup: unexpected } as never), + recovery: { escalateStrandedAssignedIssue: unexpected, escalateStrandedRecoveryIssueInPlace: unexpected } as never, + }); + } + + it("consumes the saved chat wake instead of starting a run with no message", async () => { + const { run, wake } = await seed(true); + const unexpected = vi.fn((): never => { throw new Error("Unexpected promotion of a consumed chat wake"); }); + const result = await release(unexpected)({ companyId, runId: run.id, now: new Date() }); + expect(result.outcome.kind).toBe("released"); + expect(unexpected).not.toHaveBeenCalled(); + const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(saved).toMatchObject({ status: "cancelled", runId: null, error: "Queued messages were already delivered to a completed run" }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)).then(rows => + rows.filter(row => row.contextSnapshot?.issueId === saved.payload?.issueId && row.id !== run.id))).toEqual([]); + }); + + it("does not report a delivered chat message as not started", async () => { + const { wake } = await seed(true); + const action = { id: wake.id, companyId, kind: "inbound_wakeup", conversationId: randomUUID(), deliveryId: randomUUID(), + principalId: randomUUID(), createdAt: wake.requestedAt, status: "processed", + payload: { version: 1, agentId, issueId: wake.payload?.issueId, commentId: wake.payload?.wakeCommentId, + requestedByActorType: "user", requestedByActorId: "chat-user" } }; + const consumed = { ...wake, status: "cancelled", error: "Queued messages were already delivered to a completed run" }; + expect(resolveInboundWakeReceipt(action as never, consumed, consumed)).toBeNull(); + // Other declines keep their notice. + const refused = { ...wake, status: "cancelled", error: "Deferred wake suppressed by active subtree pause hold" }; + expect(resolveInboundWakeReceipt(action as never, refused, refused)).toMatchObject({ state: "not_started" }); + }); + + it("still promotes a saved chat wake whose message no run has received", async () => { + const { run, wake } = await seed(false); + // Promotion reaches the host; the refusal proves the wake was not consumed. + const unexpected = vi.fn((): never => { throw new Error("promotion reached the host"); }); + await expect(release(unexpected)({ companyId, runId: run.id, now: new Date() })).rejects.toThrow("promotion reached the host"); + expect(unexpected).toHaveBeenCalled(); + const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(saved.status).not.toBe("cancelled"); + }); +}); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index d59c9340a0a..93689a48dd1 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -19,7 +19,7 @@ import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm" import { readReleaseRun, type ReleaseRun } from "./release-run-read.js"; import { isGovernancePauseStop } from "../../../services/governance-pause-stop.js"; import type { Db } from "@paperclipai/db"; -import { deliveredQueuedCommentIds } from "../../../services/queued-comment-delivery.js"; +import { deliveredQueuedCommentIds, EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION, EXTERNAL_CHAT_MESSAGE_WAKE_REASON } from "../../../services/queued-comment-delivery.js"; import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared"; import { activityLog, @@ -177,7 +177,11 @@ function toDeferredWakeCandidate(row: typeof agentWakeupRequests.$inferSelect): !queuedReason || queuedReason === "issue_commented" || queuedReason === "issue_reopened_via_comment" || - queuedReason === "issue_comment_mentioned"; + queuedReason === "issue_comment_mentioned" || + // An inbound chat message wake carries only its saved messages. When a + // completed run already received them, nothing is left to start, and a + // reviewed chat run with no message cannot be attested. + (queuedReason === EXTERNAL_CHAT_MESSAGE_WAKE_REASON && payload.mutation === EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION); const preservesIndependentContinuation = hasInteractionContinuationWakeContext(deferredContextSeed) || deferredContextSeed.resumeIntent === true || diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 10e8c617f25..3a74a7aa2f9 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -1,5 +1,6 @@ import { readWakeChatConversationId } from "../../run-dispatch/index.js"; import { enrichPromotedWakeContext, wakeMustNotJoinActiveRun } from "../domain/context.js"; +import { QUEUED_MESSAGES_DELIVERED_REASON } from "../domain/policy.js"; import { decideQueuedCommentAction, decideReleaseRecovery, @@ -272,7 +273,7 @@ async function runReleaseDrain( reason: commentAction.selfAuthored ? "Deferred wake contained only comments authored by the finishing run" : liveness.deliveredCommentIds?.length - ? "Queued messages were already delivered to a completed run" + ? QUEUED_MESSAGES_DELIVERED_REASON : "Queued messages were discarded before promotion", now: input.now, }); diff --git a/server/src/modules/wake-queue/domain/policy.ts b/server/src/modules/wake-queue/domain/policy.ts index 11b904bdc61..8ef29468016 100644 --- a/server/src/modules/wake-queue/domain/policy.ts +++ b/server/src/modules/wake-queue/domain/policy.ts @@ -13,6 +13,9 @@ // This file only branches on that facts object; it never queries a // database, reads the clock, or reads the wake context payload directly. +/** The error a saved wake records when a completed run already received all of its messages. */ +export const QUEUED_MESSAGES_DELIVERED_REASON = "Queued messages were already delivered to a completed run"; + export type WakeAdmissionFacts = { /** True when the active execution run's agent and this wake's own agent share an execution-agent-name key. */ isSameExecutionAgent: boolean; diff --git a/server/src/modules/wake-queue/index.ts b/server/src/modules/wake-queue/index.ts index 15e86549d30..9ac0a76879b 100644 --- a/server/src/modules/wake-queue/index.ts +++ b/server/src/modules/wake-queue/index.ts @@ -29,6 +29,7 @@ export type { } from "./application/types.js"; export { WakeQueueApplicationError } from "./application/types.js"; export { hasRetainedExternalChatRetry } from "./domain/context.js"; +export { QUEUED_MESSAGES_DELIVERED_REASON } from "./domain/policy.js"; export type { IssueSnapshot, RunSnapshot, diff --git a/server/src/services/chat-channels/inbound/wakeups.ts b/server/src/services/chat-channels/inbound/wakeups.ts index d80fde3d6f9..feea97ba8f0 100644 --- a/server/src/services/chat-channels/inbound/wakeups.ts +++ b/server/src/services/chat-channels/inbound/wakeups.ts @@ -1,3 +1,4 @@ +import { EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION, EXTERNAL_CHAT_MESSAGE_WAKE_REASON } from "../../queued-comment-delivery.js"; 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"; @@ -367,8 +368,8 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) { await queueIssueAssignmentWakeup({ heartbeat: options.heartbeat, issue: context.issue, - reason: "External chat message received", - mutation: "chat_message_received", + reason: EXTERNAL_CHAT_MESSAGE_WAKE_REASON, + mutation: EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION, contextSource: `chat:${context.endpoint.provider}`, requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId, @@ -471,7 +472,7 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) { }); assertDurableChatWakeupReceipt(request, receipt); await queueIssueAssignmentWakeup({ heartbeat: options.heartbeat, issue: context.issue, - reason: "External chat message received", mutation: "chat_message_received", + reason: EXTERNAL_CHAT_MESSAGE_WAKE_REASON, mutation: EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION, contextSource: `chat:${context.endpoint.provider}`, taskKey: context.issue.identifier, requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId, wakeCommentId: request.commentId, chatConversationId: context.conversation.id, diff --git a/server/src/services/chat-inbound-wakeup-publications.ts b/server/src/services/chat-inbound-wakeup-publications.ts index 0f3915b2706..d037459beb5 100644 --- a/server/src/services/chat-inbound-wakeup-publications.ts +++ b/server/src/services/chat-inbound-wakeup-publications.ts @@ -3,6 +3,7 @@ import { assertDurableChatWakeupReceipt, createDurableChatWakeupRequest, } from "./durable-chat-wakeup.js"; +import { QUEUED_MESSAGES_DELIVERED_REASON } from "../modules/wake-queue/index.js"; type Action = typeof chatActions.$inferSelect; type Receipt = typeof agentWakeupRequests.$inferSelect; @@ -125,6 +126,10 @@ export function resolveInboundWakeReceipt( const declined = ["skipped", "cancelled", "failed"].includes(owner.status) || ["skipped", "cancelled", "failed"].includes(receipt.status); + // A completed run already received this message and answered it. Saying + // "not started" would be wrong; it keeps the same silence as a promotion. + if (declined && owner.status === "cancelled" && owner.error === QUEUED_MESSAGES_DELIVERED_REASON) + return null; if (declined) return owner.runId || receipt.runId ? null diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index a99654a7ba1..d504a101231 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { QUEUED_MESSAGES_DELIVERED_REASON } from "../../modules/wake-queue/index.js"; import { connectionIntentService } from "../connection-intents.js"; import { verifyBlockerStallWake } from "../recovery/blocker-stall-wakeups.js"; import { CONTEXT_COMPACTION_WAKE_REASON, ContextCompactionError, isContextCompactionRun, readContextCompactionRequest } from "../context-compaction.js"; @@ -1920,7 +1921,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const remaining = commentIds.filter(commentId => !delivered.has(commentId)); const now = new Date(); await db.update(agentWakeupRequests).set(remaining.length === 0 - ? { status: "cancelled", finishedAt: now, updatedAt: now, error: "Queued messages were already delivered to a completed run" } + ? { status: "cancelled", finishedAt: now, updatedAt: now, error: QUEUED_MESSAGES_DELIVERED_REASON } : { payload: withQueuedCommentIdsInWakePayload(parseObject(receipt.payload), remaining), updatedAt: now }) .where(and(eq(agentWakeupRequests.id, receipt.id), eq(agentWakeupRequests.status, "deferred_issue_execution"))); } diff --git a/server/src/services/queued-comment-delivery.ts b/server/src/services/queued-comment-delivery.ts index 6ef16eab4fd..9faf1b6a169 100644 --- a/server/src/services/queued-comment-delivery.ts +++ b/server/src/services/queued-comment-delivery.ts @@ -1,6 +1,11 @@ import { and, eq, inArray, sql } from "drizzle-orm"; import { heartbeatRunEvents, heartbeatRuns, issueComments, type Db } from "@paperclipai/db"; +/** The wake reason an inbound chat message records (see chat inbound wakeups). */ +export const EXTERNAL_CHAT_MESSAGE_WAKE_REASON = "External chat message received"; +/** The mutation an inbound chat message wake records. */ +export const EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION = "chat_message_received"; + /** * Per-comment consumption (work-intent outbox, invariant 4). A succeeded run * of the wake's agent that started after the comment's latest revision