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