diff --git a/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts b/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts index ab100717603..b5fc8e27060 100644 --- a/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts +++ b/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts @@ -11,6 +11,7 @@ import { heartbeatRuns, externalAgentHolds, issues, + issueThreadInteractions, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, @@ -294,6 +295,62 @@ describeEmbeddedPostgres("heartbeat queued-run claim isolation", () => { expect(mockAdapterExecute).toHaveBeenCalledOnce(); }); + async function insertCardAnswerInterruptRun(fault?: "other-answer" | "other-resolver") { + const { companyId, agentId } = await insertAgent(); + const userId = `card-owner-${randomUUID()}`; + await db.insert(authUsers).values({ id: userId, name: "Card owner", email: `${userId}@paperclip.test`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, + status: "active", membershipRole: "owner" }); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Website", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: userId }); + const [interaction] = await db.insert(issueThreadInteractions).values({ companyId, issueId, + kind: "request_confirmation", status: "accepted", payload: { version: 1, prompt: "Use the simpler layout?" }, + result: { version: 1, outcome: "accepted" }, createdByUserId: userId, + resolvedByUserId: fault === "other-resolver" ? "another-user" : userId, resolvedAt: new Date() }).returning(); + const queueId = randomUUID(); + const { runId, wakeupRequestId } = await insertQueuedRun(companyId, agentId, { + reason: "issue_commented", + idempotencyKey: `queued-comment-interrupt:${queueId}`, + requestedByActorType: "user", + requestedByActorId: userId, + payload: { issueId, interactionId: interaction!.id, explicitUserContinuation: { queuedCommentInterruptId: queueId } }, + }, new Date(Date.now() - 60_000)); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId, agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "coalesced", runId, + requestedByActorType: "user", requestedByActorId: userId, + payload: { issueId, mutation: "interaction", interactionId: interaction!.id, interactionKind: "request_confirmation", + interactionStatus: "accepted", queuedCommentInterrupt: { actorId: userId, requestedAt: new Date().toISOString() } } }); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, taskId: issueId, + interactionId: fault === "other-answer" ? randomUUID() : interaction!.id, interactionKind: "request_confirmation", + interactionStatus: "accepted", queuedCommentSourceWakeIds: [queueId] } }).where(sql`${heartbeatRuns.id} = ${runId}`); + return { runId, wakeupRequestId, userId }; + } + + it("claims the successor of an interrupt that delivers a queued card answer", async () => { + const { runId, userId } = await insertCardAnswerInterruptRun(); + + await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined(); + await heartbeat.drainActiveRunExecutions(); + + expect(await runStatus(runId)).toMatchObject({ status: "succeeded", errorCode: null }); + const [run] = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns) + .where(sql`${heartbeatRuns.id} = ${runId}`); + expect(run).toMatchObject({ responsibleUserId: userId }); + }); + + it.each(["other-answer", "other-resolver"] as const)( + "rejects a card-answer interrupt whose %s does not match the receipt", async (fault) => { + const { runId } = await insertCardAnswerInterruptRun(fault); + + await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined(); + + expect(await runStatus(runId)).toMatchObject({ status: "cancelled", errorCode: "queued_run_claim_rejected" }); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + }, + ); + it("keeps processing other agents' queued runs after one agent's run is rejected", async () => { const first = await insertAgent(); const second = await insertAgent(); diff --git a/server/src/services/run-identity.ts b/server/src/services/run-identity.ts index 9b8d845c098..e9ebc3219ed 100644 --- a/server/src/services/run-identity.ts +++ b/server/src/services/run-identity.ts @@ -15,6 +15,7 @@ import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { boundRunIssueId, boundRunIssueIdSql, runTaskContextColumn } from "./run-task-binding.js"; import { isUuidLike } from "@paperclipai/shared"; import { queuedCommentIdsFromRunContext, queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; +import { queuedInteractionId } from "./queued-interaction-response.js"; /** Resolve an explicit click from persisted receipts, never caller context or message authors. */ export async function explicitOperatorRunIdentity( @@ -46,14 +47,31 @@ export async function explicitOperatorRunIdentity( )); const marker = receipt?.payload?.queuedCommentInterrupt; const actorId = marker && typeof marker === "object" && "actorId" in marker ? marker.actorId : null; - const ids = queuedCommentIdsFromWakePayload(receipt?.payload); - const deliveredIds = queuedCommentIdsFromRunContext(run.contextSnapshot); - if (typeof actorId !== "string" || !actorId || !ids.length || - receipt?.payload?.issueId !== run.contextSnapshot?.issueId || - !ids.every(id => deliveredIds.includes(id)) || + if (typeof actorId !== "string" || !actorId || !receipt || + receipt.payload?.issueId !== run.contextSnapshot?.issueId || request.requestedByActorType !== "user" || request.requestedByActorId !== actorId) { throw forbidden("Queued-message interrupt authority is unavailable"); } + const ids = queuedCommentIdsFromWakePayload(receipt.payload); + if (!ids.length) { + const interactionId = queuedInteractionId(receipt.payload); + const [interaction] = interactionId && run.contextSnapshot?.interactionId === interactionId + ? await executor.select({ resolvedByUserId: issueThreadInteractions.resolvedByUserId }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, interactionId), eq(issueThreadInteractions.companyId, run.companyId), + eq(issueThreadInteractions.issueId, String(run.contextSnapshot.issueId)), + eq(issueThreadInteractions.status, String(receipt.payload?.interactionStatus)), + )) + : []; + if (!interaction || interaction.resolvedByUserId !== actorId) { + throw forbidden("Queued-message interrupt authority is unavailable"); + } + return { actorId, cause: "queued_comment_interrupt" }; + } + const deliveredIds = queuedCommentIdsFromRunContext(run.contextSnapshot); + if (!ids.every(id => deliveredIds.includes(id))) { + throw forbidden("Queued-message interrupt authority is unavailable"); + } return { actorId, cause: "queued_comment_interrupt" }; }