From d79b7f29f2695651004b3755c5fc44d6970f46bd Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:10:28 -0600 Subject: [PATCH] fix(heartbeat): bind queued card-answer interrupts to the delivered answer (#20698) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A person can queue direction while a task's run is active, then use Interrupt to stop the run and start a successor that receives the direction > - The queued direction can be an answer to a card, for example an accepted confirmation of a saved plan > - `explicitOperatorRunIdentity` verifies the click from the coalesced queue receipt before the successor run can start > - That check required comment ids in the receipt, but a card answer carries its interaction and no comment ids > - So every claim of the successor failed with 403, and the direction never reached a run > - This pull request binds a card-answer receipt to the answer that the successor run delivers, and to the user who gave it > - The benefit is that Interrupt works for queued card answers, and the operator identity stays verified ## Linked Issues or Issue Description Refs #14043, #14510 **What happened** A person accepted a `request_confirmation` card on a task while its run was active, and then clicked Interrupt. The active run stopped ("Interrupted to send queued messages"). The successor run was then cancelled with "Cancelled because the queued run cannot be claimed: Queued-message interrupt authority is unavailable". The coalesced receipt had the interaction id and no `queuedCommentIds`. The successor run's context had the same `interactionId` and the receipt id in `queuedCommentSourceWakeIds`. The chat eval `steering.interrupt-with-direction` recorded this. **Expected behavior** The successor run starts, receives the accepted direction once, and runs with the clicking user as its responsible user. A receipt that the run does not deliver, or an answer that another user gave, still fails closed. **Steps to reproduce** 1. Start a run on a task. 2. While the run is active, accept a confirmation card on that task. The answer goes into the deferred queue. 3. Click Interrupt for the queued messages. 4. Before this change, the successor run is cancelled with `queued_run_claim_rejected`. ## What Changed - `server/src/services/run-identity.ts`: the shared receipt binding (consumed `coalesced` receipt for the same company, agent and run; same task; `user` request from the marker's actor) runs first, as before. Then: - A receipt with comment ids keeps the existing rule: all ids must be delivered to the run. - A receipt without comment ids must be a queued card answer (`queuedInteractionId`). The run context must deliver the same `interactionId`. The interaction must be on the same company and task, with the receipt's status, and `resolvedByUserId` must be the clicking user. Then the run gets the operator identity (`queued_comment_interrupt`). Otherwise the claim still fails with 403. - `server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts`: a successor of a card-answer interrupt is claimed, runs, and has the clicking user as responsible user. Two negative cases: the run delivers a different answer, and another user resolved the answer. Both stay rejected. Relation to #14510: that open PR fixes the same claim failure (#14043) by returning `null` for card-answer receipts, so the run starts without the operator identity. Its Risks section notes that keeping the operator identity needs a check that the actor resolved the interaction. This PR adds that check instead, so the successor keeps the identity that enqueue time records. If maintainers prefer #14510's approach, this PR's regression test still applies to it. ## Verification - `cd server && npx vitest run src/__tests__/heartbeat-queued-run-claim-isolation.test.ts`: the new positive test fails before the change (successor cancelled) and passes after it. The negative tests and the existing tests pass. - Suites that cover interrupt identity, each run separately: `run-identity` (37), `heartbeat-responsible-user-invariant` (18), `issue-queued-comments-routes` (129), `deferred-execution-receipt-work` (50), `task-interruption-query` (31), `explicit-native-continuation` (436), `run-cancellation` (8), `workspace-recovery-admission` (147), `wake-queue/adapters/postgres` (113): all pass. In `heartbeat-process-recovery`, the interrupt-named tests pass except `dispatches interrupted CLI input and cleans up early fixture exits`, which fails before any interrupt (the CLI child does not start: `controller_lease_lost`). It fails the same way on unchanged master on this loaded host (3 of 3 attempts failed at least one case), so it is not caused by this change. CI covers it. - `server` `tsc --noEmit`: no errors in the changed files. - Live chat eval `chat-features.runner-opencode.local.steering.interrupt-with-direction` at this branch (run 42938970466): passed. The old run stopped with "Interrupted to send queued messages". Exactly one successor run started, delivered the accepted card answer, ran with the clicking user as responsible user, and saved the approved single page and its delivery document. Before this change, the same case failed with the successor cancelled as `queued_run_claim_rejected`. ## Risks - Low risk. The change adds a verified path only for receipts that previously always failed. Comment receipts keep their checks. - If a different user answered the card than the user who clicked Interrupt, the claim still fails. The person who answered can click Interrupt. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use and code execution. ## 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 Opus 5.5 --- ...artbeat-queued-run-claim-isolation.test.ts | 57 +++++++++++++++++++ server/src/services/run-identity.ts | 28 +++++++-- 2 files changed, 80 insertions(+), 5 deletions(-) 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" }; }