mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(chat): never batch messages of two chat conversations into one wake (#20697)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Chat channels let a linked user talk to an agent from GitHub, Slack and other platforms. Each external thread is a chat conversation. Many conversations can share one task. > - A reviewed chat run attests exactly one conversation. #17054 keeps each conversation in its own wake batch, but it compares only the conversation that a wake records. > - The work-intent outbox saves unanswered user messages as a plain comment wake. That wake records no conversation, even when its messages are chat messages. A wake with no recorded conversation matched every conversation, so a message from another conversation merged into it. > - The promoted run held messages of two conversations. The binding check correctly refused it (`reviewed_chat_execution_binding_not_authorized`, `binding_denied`), and the task needed Retry. > - This pull request compares the conversations of the queued messages themselves. Wakes of different conversations never merge. Each conversation gets its own run, in order. > - The benefit is that replies in several threads of one task each get an attested run, without a failed run or a Retry. ## Linked Issues or Issue Description Refs #17054 (the earlier per-conversation batching fix). Refs #20689 (a different shape of the same error). **What happened?** On staging, a GitHub pull-request review task had two runs fail with `setup_failed` / `reviewed_chat_execution_binding_not_authorized` (`reviewedChatBinding.denialReason: binding_denied`). Each failed run held the same two older saved messages plus one new message. The two runs recorded two different `chatConversationId` values. The work-intent outbox had saved the older messages as plain deferred wakes with no conversation. A new message from each conversation then merged into one of these saved wakes. **Steps to reproduce** 1. Bind two chat conversations to one `in_review` task. 2. Let the outbox save an unanswered message of conversation B as a deferred wake while a run is active. 3. As the linked user, send a message in conversation A while the run is still active. 4. See that the message of A merges into the saved wake of B. The promoted run fails the binding check. **Expected behavior** The saved message of B runs as its own wake. The message of A runs after it, in its own attested run. ## What Changed - New `server/src/modules/wake-queue/adapters/chat-conversation-batch.ts`: reads the inbound chat links (`chat_message_links`, direction `inbound`) of queued messages and finds messages of a conversation other than the expected one. - Deferred-wake lookup (`findExistingDeferredWake`): when the incoming wake has a conversation, a saved wake qualifies only if none of its queued messages is an inbound message of another conversation. This is one more predicate in the same query (no new round trip). It uses the existing `chat_message_links_company_comment_idx` index. - Coalescing into the active run (`coalesceIntoActiveExecutionRun`): returns no merge when the merged batch would hold messages of a conversation other than the one it records. The wake is then saved behind the run, which is the existing fallback. - Release drain (`findNextDeferredWake`), upgrade repair: a saved chat wake that older code already merged across two conversations is split. The other conversation's messages move to a plain saved wake (the outbox shape), ordered just before it. The original wake keeps its conversation and its own messages. The drain then reads the queue again. - `doc/execution-semantics.md`: describe the comparison and the upgrade repair. - New test `chat-wake-conversation-batch.test.ts` (embedded Postgres, real admission and release code, real `authorizeChatConversationForBoundRun`). ## Verification - Red on `origin/master` (test file only): `pnpm --dir server exec vitest run src/modules/wake-queue/adapters/chat-wake-conversation-batch.test.ts` gives 4 failed, 6 passed. The 4 failures are the bug: the new message merges into the other conversation's saved wake (`[[earlier, reply]]`), two conversations cross-merge, a message coalesces into a queued run of another conversation, and the old merged shape stays merged. The 6 passing cases pin unchanged batching. - Green with the fix: 10 of 10 pass. Each separated wake passes `authorizeChatConversationForBoundRun` for its own conversation. The old merged shape fails it with `paperclip_runner_chat_attachment_binding_denied` before the split. - Batching that stays the same (pinned, pass before and after): comment wakes with no conversation merge; messages of one conversation merge; an outbox-saved wake of the same conversation still absorbs the reply; a chat message and plain comments merge in both orders; same-conversation messages still coalesce into a queued run; two recorded conversations stay apart. - Related suites pass locally: `src/modules/wake-queue`, `src/modules/run-dispatch`, `chat-run-milestone-work`, `durable-chat-wakeup`, `heartbeat-comment-wake-batching`, `heartbeat-reviewed-chat-binding.integration`, `slack-conversation-lifecycle`, `queued-comment-consumption`, `issue-queued-comments-routes` (23 files, 1037 tests). `heartbeat-process-recovery`: see the comment below. - `tsc --noEmit -p server/tsconfig.json` passes. ## Risks - Low risk. No schema change and no migration. - Behavior change only for chat messages: a message no longer joins a saved wake or a run that holds another conversation's inbound messages. It gets its own wake behind them, and it runs after them. Before this change, that batch always failed the binding check. - Board comments have no inbound chat link, so they belong to no conversation. Wakes without a conversation, wakes of one conversation, and mixed board-comment and chat wakes batch exactly as before. - Ordering: the deferred queue still promotes one wake per release, oldest first. The second conversation waits for the issue execution lock. Single-assignee and atomic checkout are unchanged. - Upgrade: a saved wake that older code merged across conversations is split at promotion. No message is dropped or delivered twice: each message id moves to exactly one wake. A run that older code already promoted in the merged shape is not changed; it fails the binding check as before and keeps its existing Retry path. - No wait is added before a turn starts. Admission adds one predicate to an existing query. Coalescing adds one indexed read only when the merged batch has a chat conversation. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), in Claude Code with tool use (shell, file edits, GitHub CLI). ## 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 - [x] All Paperclip CI gates are green - [x] 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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude authored and GitHub committed
1 parent
d79b7f29f2
commit
4baff83225
4 files changed
+474
-1
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
}
|
||||
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<string[]> {
|
||||
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<string, unknown> },
|
||||
): Promise<boolean> {
|
||||
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;
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let f: Awaited<ReturnType<typeof seedBoundChatAuthorization>>;
|
||||
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<string> {
|
||||
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<string, unknown>, 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<string, Record<string, unknown>>)[CONTEXT_KEY]! }));
|
||||
}
|
||||
|
||||
async function attest(context: Record<string, unknown>) {
|
||||
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<string, Record<string, unknown>>)[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<string, Record<string, unknown>>)[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<string, unknown>;
|
||||
await db.update(agentWakeupRequests).set({ payload: { ...payload, mutation: "chat_message_received",
|
||||
[CONTEXT_KEY]: { ...(payload[CONTEXT_KEY] as Record<string, unknown>), 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<string, unknown>;
|
||||
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 });
|
||||
});
|
||||
});
|
||||
@@ -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 &&
|
||||
|
||||
Reference in new issue
Block a user