diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6c2c318e36b..d58586a675b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -328,6 +328,15 @@ Messages from the same conversation still coalesce into one run. Messages from different conversations queue separately and run one after the other. Wakes without a conversation keep their normal coalescing. +The comparison uses the conversations of the messages themselves, not only the +conversation a wake records. A wake that the work-intent outbox saved for +unanswered chat messages records no conversation, so a message from another +conversation does not join it, and is not folded into a run promoted from it. +Board comments have no conversation and still join any batch. A saved chat wake +that older code already merged across two conversations is split when it is +promoted: the other conversation's messages run first as saved messages, then +the recorded conversation runs with its own messages. + If reviewed chat attestation exhausts its short lock or delivery wait before provider dispatch, the run records `reviewed_chat_execution_binding_not_ready`. The existing durable retry scheduler permits at most two further attempts. Each diff --git a/server/src/modules/wake-queue/adapters/chat-conversation-batch.ts b/server/src/modules/wake-queue/adapters/chat-conversation-batch.ts new file mode 100644 index 00000000000..d0109cf4f19 --- /dev/null +++ b/server/src/modules/wake-queue/adapters/chat-conversation-batch.ts @@ -0,0 +1,113 @@ +import { and, eq, inArray, ne, sql, type SQL } from "drizzle-orm"; +import { agentWakeupRequests, chatMessageLinks, type Db } from "@paperclipai/db"; +import { isUuidLike } from "@paperclipai/shared"; +import { withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js"; + +// A reviewed chat run attests exactly one conversation. A wake records its +// conversation in `chatConversationId`, but a wake can also hold messages of +// a conversation it does not record: the work-intent outbox saves unanswered +// chat messages as a plain comment wake. These checks compare the +// conversations of the messages themselves, so a batch never holds messages +// of two conversations. Comments with no inbound chat link (board comments) +// belong to no conversation and batch as before. + +const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; +const CHAT_CONVERSATION_ID_KEY = "chatConversationId"; + +function record(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record : {}; +} +const UUID_PATTERN = "^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"; + +/** + * SQL predicate on a deferred `agent_wakeup_requests` row: none of its queued + * messages is an inbound chat message of a conversation other than + * `conversationId`. + */ +export function deferredWakeHoldsNoOtherConversation(companyId: string, conversationId: string): SQL { + const ids = sql`${agentWakeupRequests.payload} -> ${DEFERRED_WAKE_CONTEXT_KEY}::text -> 'wakeCommentIds'`; + return sql`not exists ( + select 1 from ${chatMessageLinks} + where ${chatMessageLinks.companyId} = ${companyId} + and ${chatMessageLinks.direction} = 'inbound' + and ${chatMessageLinks.conversationId}::text <> ${conversationId} + and ${chatMessageLinks.commentId} = any(array( + select queued.id::uuid + from jsonb_array_elements_text(case when jsonb_typeof(${ids}) = 'array' then ${ids} else '[]'::jsonb end) as queued(id) + where queued.id ~ ${UUID_PATTERN} + )) + )`; +} + +/** The ids among `commentIds` that are inbound chat messages of a conversation other than `conversationId`. */ +export async function otherConversationCommentIds( + tx: Db, + input: { companyId: string; conversationId: string; commentIds: string[] }, +): Promise { + const queryable = input.commentIds.filter((id) => isUuidLike(id)); + if (queryable.length === 0) return []; + const rows = await tx + .selectDistinct({ commentId: chatMessageLinks.commentId }) + .from(chatMessageLinks) + .where(and( + eq(chatMessageLinks.companyId, input.companyId), + eq(chatMessageLinks.direction, "inbound"), + inArray(chatMessageLinks.commentId, queryable), + ne(sql`${chatMessageLinks.conversationId}::text`, input.conversationId), + )); + const other = new Set(rows.map((row) => row.commentId)); + return input.commentIds.filter((id) => other.has(id)); +} + +/** + * Upgrade repair for a deferred chat wake that older code merged across two + * conversations. The messages of other conversations move to their own plain + * saved wake (the shape the work-intent outbox writes), ordered just before + * this wake so they keep their place. This wake keeps its id, its recorded + * conversation and the rest of its messages. When no message of its own + * conversation is left, the wake keeps its id and becomes a plain saved wake. + * No message is dropped. + * Returns true when the wake changed and must be read again. + */ +export async function splitOtherConversationMessages( + tx: Db, + row: typeof agentWakeupRequests.$inferSelect, + input: { companyId: string; issueId: string; conversationId: string; queuedCommentIds: string[]; payload: Record }, +): Promise { + const moved = await otherConversationCommentIds(tx, { + companyId: input.companyId, conversationId: input.conversationId, commentIds: input.queuedCommentIds, + }); + if (moved.length === 0) return false; + const kept = input.queuedCommentIds.filter((id) => !moved.includes(id)); + const now = new Date(); + const deferred = and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, row.id), + eq(agentWakeupRequests.status, "deferred_issue_execution")); + if (kept.length === 0) { + // Every message belongs to another conversation. Keep the wake and its id + // (an interrupt names the wake by id) and drop only its chat binding, so + // it runs as plain saved messages. + const { [CHAT_CONVERSATION_ID_KEY]: _dropped, ...context } = record(input.payload[DEFERRED_WAKE_CONTEXT_KEY]); + await tx.update(agentWakeupRequests).set({ updatedAt: now, + payload: withQueuedCommentIdsInWakePayload({ ...input.payload, mutation: "comment", + [DEFERRED_WAKE_CONTEXT_KEY]: { ...context, source: "issue.comment", wakeReason: "issue_commented" } }, moved) }) + .where(deferred); + return true; + } + const updated = await tx.update(agentWakeupRequests) + .set({ payload: withQueuedCommentIdsInWakePayload(input.payload, kept), updatedAt: now }) + .where(deferred) + .returning({ id: agentWakeupRequests.id }); + // Another writer already moved the wake on; its messages stay where they are. + if (updated.length === 0) return true; + const latest = moved.at(-1)!; + await tx.insert(agentWakeupRequests).values({ + companyId: input.companyId, agentId: row.agentId, source: "automation", triggerDetail: "system", + reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: row.requestedByActorType, requestedByActorId: row.requestedByActorId, + requestedAt: new Date(row.requestedAt.getTime() - 1), updatedAt: now, + payload: withQueuedCommentIdsInWakePayload({ issueId: input.issueId, mutation: "comment", commentId: latest, + [DEFERRED_WAKE_CONTEXT_KEY]: { issueId: input.issueId, taskId: input.issueId, wakeReason: "issue_commented", + source: "issue.comment", wakeCommentIds: moved, wakeCommentId: latest } }, moved), + }); + return true; +} diff --git a/server/src/modules/wake-queue/adapters/chat-wake-conversation-batch.test.ts b/server/src/modules/wake-queue/adapters/chat-wake-conversation-batch.test.ts new file mode 100644 index 00000000000..760708c4549 --- /dev/null +++ b/server/src/modules/wake-queue/adapters/chat-wake-conversation-batch.test.ts @@ -0,0 +1,335 @@ +import { randomUUID } from "node:crypto"; +import { and, asc, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { agentWakeupRequests, chatConversations, chatDeliveries, chatMessageLinks, createDb, heartbeatRuns, + issueComments, issues, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { seedBoundChatAuthorization } from "../../../__tests__/helpers/bound-chat-authorization.js"; +import { authorizeChatConversationForBoundRun } from "../../../services/native-runtime/chat-attachment-reuse.js"; +import { mergeCoalescedContextSnapshot } from "../../../services/heartbeat/run-preparation.js"; +import { saveUndeliveredUserComments } from "../../../services/work-intent-outbox.js"; +import { queuedCommentIdsFromWakePayload } from "../../../services/issue-queued-comment-queue.js"; +import { createAdmitWakeBehindIssueExecution, createReleaseIssueExecution } from "../application/use-cases.js"; +import type { WakeAdmissionActiveExecutionRun } from "../application/ports.js"; +import { createAdmissionTransactionScope, createPostgresWakeQueueAdapter, createWakeAdmissionReader, + createWakeAdmissionWriter } from "./postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const CONTEXT_KEY = "_paperclipWakeContext"; + +// PAP-4244 on staging. The work-intent outbox saved two unanswered GitHub +// messages as a plain deferred wake: it records no chat conversation. A new +// message from another conversation then merged into that saved wake, because +// a wake with no recorded conversation matched every conversation. The +// promoted run held messages of two conversations, and the reviewed chat +// binding refused it: `reviewed_chat_execution_binding_not_authorized` +// (`binding_denied`). Each conversation must get its own run, in order. +describe.skipIf(!support.supported)("chat wake batches stay within one conversation", () => { + let database: Awaited>; + let db: ReturnType; + let f: Awaited>; + let otherConversationId: string; + let deferOnly = true; + + const admit = createAdmitWakeBehindIssueExecution({ + reader: createWakeAdmissionReader(), + writer: createWakeAdmissionWriter(), + helpers: { + filterZombieCoalesceTarget: (target) => target, + mergeCoalescedContextSnapshot, + shouldDeferFollowupWakeForSameIssue: () => deferOnly, + shouldQueueFollowupForRunningIssueWake: () => false, + }, + }); + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("wake-chat-conversation-batch-"); + db = createDb(database.connectionString); + }, 60_000); + + afterAll(async () => { + await db?.$client.end(); + await database?.cleanup(); + }); + + beforeEach(async () => { + deferOnly = true; + // The fixture uses fixed identifiers; move earlier cases' rows aside. + await db.execute(sql`update companies set issue_prefix = 'X' || upper(left(replace(id::text, '-', ''), 6))`); + await db.execute(sql`update issues set identifier = id::text`); + f = await seedBoundChatAuthorization(db); + otherConversationId = randomUUID(); + await db.insert(chatConversations).values({ id: otherConversationId, companyId: f.companyId, endpointId: f.endpointId, + resourceId: f.resourceId, issueId: f.issueId, externalConversationId: "channel-1", externalThreadId: "thread-2", + externalLabel: "Second review thread", state: "active" }); + // The fixture's own message is not part of these cases. + await db.delete(chatMessageLinks).where(eq(chatMessageLinks.commentId, f.commentId)); + await db.delete(issueComments).where(eq(issueComments.id, f.commentId)); + await db.insert(heartbeatRuns).values({ id: f.runId, companyId: f.companyId, agentId: f.agentId, status: "running", + invocationSource: "assignment", triggerDetail: "system", startedAt: new Date(), + contextSnapshot: { issueId: f.issueId, taskId: f.issueId } }); + await db.update(issues).set({ executionRunId: f.runId, executionLockedAt: new Date() }).where(eq(issues.id, f.issueId)); + }); + + let messageClock = Date.now() - 600_000; + async function message(conversationId: string | null): Promise { + messageClock += 1_000; + const at = new Date(messageClock); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: f.userId, body: `Message ${messageClock}`, createdAt: at, updatedAt: at }).returning(); + if (conversationId) { + const deliveryId = randomUUID(); + await db.insert(chatDeliveries).values({ id: deliveryId, companyId: f.companyId, endpointId: f.endpointId, + conversationId, principalId: f.principalId, providerEventId: deliveryId, deduplicationKey: deliveryId, + eventKind: "message", normalizedEvent: {}, state: "processed", attempts: 1, processedAt: at }); + await db.insert(chatMessageLinks).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId, + deliveryId, commentId: comment!.id, providerMessageId: deliveryId, direction: "inbound" }); + } + return comment!.id; + } + + function chatContext(conversationId: string, commentId: string) { + return { issueId: f.issueId, taskId: f.issueId, source: "chat:slack", wakeReason: "External chat message received", + wakeSource: "assignment", chatConversationId: conversationId, commentId, wakeCommentId: commentId, + wakeCommentIds: [commentId] }; + } + + function commentContext(commentId: string) { + return { issueId: f.issueId, taskId: f.issueId, source: "issue.comment", wakeReason: "issue_commented", + commentId, wakeCommentId: commentId, wakeCommentIds: [commentId] }; + } + + async function admitWake(contextSnapshot: Record, activeExecutionRun?: WakeAdmissionActiveExecutionRun) { + const commentId = contextSnapshot.wakeCommentId as string; + const chat = contextSnapshot.source === "chat:slack"; + return db.transaction(async (tx) => admit(createAdmissionTransactionScope(f.companyId, tx as unknown as Db), { + companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, agentNameKey: "chat runner", + issueExecutionAgentNameKey: null, + activeExecutionRun: activeExecutionRun ?? { id: f.runId, agentId: f.agentId, status: "running", + contextSnapshot: { issueId: f.issueId, taskId: f.issueId }, wakeupRequestId: null }, + allowRunCoalescing: true, liveRunExecutions: new Set([f.runId]), wakeCommentId: commentId, + forceFreshSession: false, contextSnapshot, source: "assignment", triggerDetail: "system", + reason: chat ? "External chat message received" : "issue_commented", + payload: { issueId: f.issueId, mutation: chat ? "chat_message_received" : "comment", wakeCommentId: commentId }, + requestedByActorType: "user", requestedByActorId: f.userId, idempotencyKey: null, + })); + } + + async function deferredWakes() { + const rows = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"))).orderBy(asc(agentWakeupRequests.requestedAt)); + return rows.map(row => ({ id: row.id, commentIds: queuedCommentIdsFromWakePayload(row.payload), + context: (row.payload as Record>)[CONTEXT_KEY]! })); + } + + async function attest(context: Record) { + return authorizeChatConversationForBoundRun(db, { companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: f.runId }, { ...context, paperclipHarnessCheckedOut: true }, "read"); + } + + async function saveUndelivered(commentIds: string[]) { + const saved = await saveUndeliveredUserComments(db, { companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + commentIds, allowedStatuses: ["in_review"] }); + expect(saved).toBeTruthy(); + return saved!; + } + + it("keeps a message out of a saved wake that holds another conversation's messages", async () => { + const earlier = await message(otherConversationId); + const saved = await saveUndelivered([earlier]); + const reply = await message(f.conversationId); + + expect(await admitWake(chatContext(f.conversationId, reply))).toEqual({ kind: "deferred" }); + + const wakes = await deferredWakes(); + // The saved messages run first, then the new conversation, each on its own. + expect(wakes.map(wake => wake.commentIds)).toEqual([[earlier], [reply]]); + expect(wakes[0]!.id).toBe(saved); + expect(wakes[0]!.context.chatConversationId).toBeUndefined(); + expect(wakes[0]!.context.source).toBe("issue.comment"); + await expect(attest(wakes[1]!.context)).resolves.toMatchObject({ conversationId: f.conversationId }); + }); + + it("keeps two conversations apart when both arrive behind saved wakes", async () => { + const first = await message(f.conversationId); + const second = await message(otherConversationId); + await saveUndelivered([first]); + await saveUndelivered([second]); + const replyToFirst = await message(f.conversationId); + const replyToSecond = await message(otherConversationId); + + await admitWake(chatContext(otherConversationId, replyToSecond)); + await admitWake(chatContext(f.conversationId, replyToFirst)); + + const wakes = await deferredWakes(); + expect(wakes.map(wake => wake.commentIds)).toEqual([[first, replyToFirst], [second, replyToSecond]]); + expect(wakes.map(wake => wake.context.chatConversationId)).toEqual([f.conversationId, otherConversationId]); + await expect(attest(wakes[0]!.context)).resolves.toMatchObject({ conversationId: f.conversationId }); + await expect(attest(wakes[1]!.context)).resolves.toMatchObject({ conversationId: otherConversationId }); + }); + + describe("batching that stays the same", () => { + it("merges comment wakes that have no conversation", async () => { + const one = await message(null); + const two = await message(null); + await admitWake(commentContext(one)); + await admitWake(commentContext(two)); + expect((await deferredWakes()).map(wake => wake.commentIds)).toEqual([[one, two]]); + }); + + it("merges messages of one conversation", async () => { + const one = await message(f.conversationId); + const two = await message(f.conversationId); + await admitWake(chatContext(f.conversationId, one)); + await admitWake(chatContext(f.conversationId, two)); + const wakes = await deferredWakes(); + expect(wakes.map(wake => wake.commentIds)).toEqual([[one, two]]); + await expect(attest(wakes[0]!.context)).resolves.toMatchObject({ conversationId: f.conversationId }); + }); + + it("merges a saved wake of the same conversation's messages", async () => { + const earlier = await message(f.conversationId); + await saveUndelivered([earlier]); + const reply = await message(f.conversationId); + await admitWake(chatContext(f.conversationId, reply)); + const wakes = await deferredWakes(); + expect(wakes.map(wake => wake.commentIds)).toEqual([[earlier, reply]]); + await expect(attest(wakes[0]!.context)).resolves.toMatchObject({ conversationId: f.conversationId }); + }); + + it("merges a chat message into a wake of plain comments, and a plain comment into a chat wake", async () => { + const plain = await message(null); + const chat = await message(f.conversationId); + const laterPlain = await message(null); + await admitWake(commentContext(plain)); + await admitWake(chatContext(f.conversationId, chat)); + await admitWake(commentContext(laterPlain)); + const wakes = await deferredWakes(); + expect(wakes.map(wake => wake.commentIds)).toEqual([[plain, chat, laterPlain]]); + expect(wakes[0]!.context.chatConversationId).toBe(f.conversationId); + }); + + it("keeps messages of two recorded conversations apart", async () => { + const one = await message(f.conversationId); + const two = await message(otherConversationId); + await admitWake(chatContext(f.conversationId, one)); + await admitWake(chatContext(otherConversationId, two)); + expect((await deferredWakes()).map(wake => wake.commentIds)).toEqual([[one], [two]]); + }); + }); + + it("does not fold a message into a queued run that holds another conversation's messages", async () => { + deferOnly = false; + const earlier = await message(otherConversationId); + const savedId = await saveUndelivered([earlier]); + // The saved wake was promoted: a queued run that the finished run left behind. + const queuedRunId = randomUUID(); + const [savedWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, savedId)); + const runContext = (savedWake!.payload as Record>)[CONTEXT_KEY]!; + await db.insert(heartbeatRuns).values({ id: queuedRunId, companyId: f.companyId, agentId: f.agentId, status: "queued", + invocationSource: "automation", triggerDetail: "system", wakeupRequestId: savedId, contextSnapshot: runContext }); + await db.update(agentWakeupRequests).set({ status: "queued", runId: queuedRunId }).where(eq(agentWakeupRequests.id, savedId)); + await db.update(issues).set({ executionRunId: queuedRunId }).where(eq(issues.id, f.issueId)); + const reply = await message(f.conversationId); + + const result = await admitWake(chatContext(f.conversationId, reply), { id: queuedRunId, agentId: f.agentId, + status: "queued", contextSnapshot: runContext, wakeupRequestId: savedId }); + + expect(result).toEqual({ kind: "deferred" }); + const [queuedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queuedRunId)); + expect(queuedRun!.contextSnapshot).toMatchObject({ wakeCommentIds: [earlier], source: "issue.comment" }); + expect((await deferredWakes()).map(wake => wake.commentIds)).toEqual([[reply]]); + }); + + it("still folds a same-conversation message into a queued run", async () => { + deferOnly = false; + const earlier = await message(f.conversationId); + const savedId = await saveUndelivered([earlier]); + const queuedRunId = randomUUID(); + const [savedWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, savedId)); + const runContext = (savedWake!.payload as Record>)[CONTEXT_KEY]!; + await db.insert(heartbeatRuns).values({ id: queuedRunId, companyId: f.companyId, agentId: f.agentId, status: "queued", + invocationSource: "automation", triggerDetail: "system", wakeupRequestId: savedId, contextSnapshot: runContext }); + await db.update(agentWakeupRequests).set({ status: "queued", runId: queuedRunId }).where(eq(agentWakeupRequests.id, savedId)); + await db.update(issues).set({ executionRunId: queuedRunId }).where(eq(issues.id, f.issueId)); + const reply = await message(f.conversationId); + + const result = await admitWake(chatContext(f.conversationId, reply), { id: queuedRunId, agentId: f.agentId, + status: "queued", contextSnapshot: runContext, wakeupRequestId: savedId }); + + expect(result.kind).toBe("coalesced"); + const [queuedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queuedRunId)); + expect(queuedRun!.contextSnapshot).toMatchObject({ wakeCommentIds: [earlier, reply], chatConversationId: f.conversationId }); + }); + + function releaseIssue() { + return createReleaseIssueExecution({ + issueLock: createPostgresWakeQueueAdapter(db, { + resolveResponsibleUserId: async () => f.userId, + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), + resolveSessionBeforeForWakeup: async () => null, + } as never), + recovery: { escalateStrandedAssignedIssue: async () => {}, escalateStrandedRecoveryIssueInPlace: async () => {} } as never, + }); + } + + // An older merged wake can be left with only the other conversation's + // messages (its own message was discarded). An interrupt names the wake by + // id, so the wake keeps its id and runs as plain saved messages. + it("keeps an interrupted wake's id when all of its messages belong to another conversation", async () => { + const earlier = await message(otherConversationId); + const savedId = await saveUndelivered([earlier]); + const [savedWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, savedId)); + const payload = savedWake!.payload as Record; + await db.update(agentWakeupRequests).set({ payload: { ...payload, mutation: "chat_message_received", + [CONTEXT_KEY]: { ...(payload[CONTEXT_KEY] as Record), source: "chat:slack", + wakeReason: "External chat message received", chatConversationId: f.conversationId } } }) + .where(eq(agentWakeupRequests.id, savedId)); + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), + resultJson: { queuedCommentInterruptQueueId: savedId, executionCancellation: { state: "acknowledged" }, + conversationContinuation: "continue_conversation_v1" } }) + .where(eq(heartbeatRuns.id, f.runId)); + + const result = await releaseIssue()({ companyId: f.companyId, runId: f.runId, now: new Date() }); + + expect(result.outcome.kind).toBe("promoted"); + const promoted = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued"))); + expect(promoted).toHaveLength(1); + expect(promoted[0]!.wakeupRequestId).toBe(savedId); + expect(promoted[0]!.contextSnapshot).toMatchObject({ wakeCommentIds: [earlier], source: "issue.comment" }); + expect(promoted[0]!.contextSnapshot?.chatConversationId).toBeUndefined(); + expect(await deferredWakes()).toEqual([]); + }); + + // A wake that older code already merged across two conversations. Release + // splits it: the other conversation's messages run first as plain saved + // messages, then the recorded conversation runs with its own messages. + it("splits a saved wake that older code merged across two conversations", async () => { + const earlier = await message(otherConversationId); + const savedId = await saveUndelivered([earlier]); + const reply = await message(f.conversationId); + const [savedWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, savedId)); + const payload = savedWake!.payload as Record; + const merged = mergeCoalescedContextSnapshot(payload[CONTEXT_KEY], chatContext(f.conversationId, reply)); + await db.update(agentWakeupRequests).set({ payload: { ...payload, mutation: "chat_message_received", + commentId: reply, wakeCommentId: reply, [CONTEXT_KEY]: merged } }).where(eq(agentWakeupRequests.id, savedId)); + expect(merged).toMatchObject({ wakeCommentIds: [earlier, reply], chatConversationId: f.conversationId }); + // This is the shape that failed on staging. + await expect(attest(merged)).rejects.toThrow("paperclip_runner_chat_attachment_binding_denied"); + + const release = releaseIssue(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + const first = await release({ companyId: f.companyId, runId: f.runId, now: new Date() }); + expect(first.outcome.kind).toBe("promoted"); + + const promoted = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued"))); + expect(promoted).toHaveLength(1); + expect(promoted[0]!.contextSnapshot).toMatchObject({ wakeCommentIds: [earlier], source: "issue.comment" }); + expect(promoted[0]!.contextSnapshot?.chatConversationId).toBeUndefined(); + const waiting = await deferredWakes(); + expect(waiting).toEqual([expect.objectContaining({ id: savedId, commentIds: [reply] })]); + await expect(attest(waiting[0]!.context)).resolves.toMatchObject({ conversationId: f.conversationId }); + }); +}); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 93689a48dd1..07c2bb718c2 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -55,7 +55,8 @@ import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInWakePayload, } from "../../../services/issue-queued-comment-queue.js"; -import { CHAT_CONVERSATION_ID_KEY, extractWakeCommentIds } from "../../run-dispatch/index.js"; +import { CHAT_CONVERSATION_ID_KEY, extractWakeCommentIds, readWakeChatConversationId } from "../../run-dispatch/index.js"; +import { deferredWakeHoldsNoOtherConversation, otherConversationCommentIds, splitOtherConversationMessages } from "./chat-conversation-batch.js"; import { hasInteractionContinuationWakeContext } from "../domain/context.js"; import { decidePreDrain, type PreDrainFacts } from "../domain/policy.js"; import { @@ -259,6 +260,12 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .then((rows) => rows[0] ?? null); if (!row) return null; const candidate = toDeferredWakeCandidate(row); + // Upgrade repair: older code could merge two conversations into one + // saved chat wake. Split it and read the queue again. + const conversationId = readWakeChatConversationId(candidate.deferredContextSeed); + if (conversationId && candidate.queuedCommentIds.length > 0 && await splitOtherConversationMessages(tx, row, { + companyId, issueId, conversationId, queuedCommentIds: candidate.queuedCommentIds, payload: candidate.payload, + })) continue; try { const authorizedFailedChatRetry = await authorizeFailedChatRunRetryWake( db, @@ -960,6 +967,9 @@ export function createWakeAdmissionReader(): WakeAdmissionReader { ...(chatConversationId ? [ sql`coalesce(${agentWakeupRequests.payload} -> ${DEFERRED_WAKE_CONTEXT_KEY}::text ->> ${CHAT_CONVERSATION_ID_KEY}::text, ${chatConversationId}) = ${chatConversationId}`, + // A wake with no recorded conversation can still hold + // another conversation's messages (an outbox-saved wake). + deferredWakeHoldsNoOtherConversation(companyId, chatConversationId), ] : []), ...(durableActor @@ -1016,6 +1026,12 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { )).for("update").limit(1); if (!current || current.status !== input.activeExecutionRunStatus || current.wakeupRequestId !== candidate.wakeupRequestId) return null; const commentIds = queuedCommentIdsFromRunContext(input.mergedContextSnapshot); + // A reviewed chat run attests one conversation. Never fold a message + // into a run that already holds another conversation's messages; the + // wake is saved behind the run instead. + const mergedConversationId = readWakeChatConversationId(input.mergedContextSnapshot); + if (mergedConversationId && commentIds.length > 0 && (await otherConversationCommentIds(tx, { + companyId: input.companyId, conversationId: mergedConversationId, commentIds })).length > 0) return null; const synchronizeBatch = current.status === "queued" && commentIds.length > 0 && primary?.status === "queued" && queuedCommentIdsFromWakePayload(primary.payload).length > 0; if (current.status === "queued" && commentIds.length && current.wakeupRequestId &&