From 8f53d44418d9b3bd067a658f64d3388d701bdb95 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 12 Sep 2026 21:58:36 -0500 Subject: [PATCH] fix(runner): adopt saved task messages during agent handoff Co-Authored-By: Paperclip --- doc/execution-semantics.md | 5 +++ .../explicit-native-continuation.test.ts | 33 +++++++++++++++++++ server/src/services/heartbeat.ts | 18 ++++++++-- 3 files changed, 54 insertions(+), 2 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index aa35b13517..0baec44cba 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1190,6 +1190,11 @@ semantic actions, and available result summary to the replacement agent. The replacement must inspect existing files and preserve completed content before editing. Source history is still scoped to the same company and task; prior results are untrusted evidence, not instructions or new authorization. +Saved task comments move into that successor's delivery receipt in the same +transaction that queues it. Their original authors remain intact. A former +assignee's ordinary comment wake must not start another execution or reopen a +completed task after the replacement finishes. Mentions, chat deliveries, and +dedicated interaction continuations retain their separate delivery contracts. A requested file is complete when the user can retrieve it. Native runners must register requested output files before reporting Done and link the resulting diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 7c235d6fc5..49b4708eee 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -42,6 +42,39 @@ const support = await getEmbeddedPostgresTestSupport(); actorType: "user", actorId: "board", reason: "issue_commented" }; } type Fixture = Awaited>; + it.each(["handoff", "foreign_task", "running_source", "different_owner", "mention", "interaction", "chat"])("adopts former-owner comments only during an authorized handoff (%s)", async kind => { + const f = await seed(), nextAgentId = randomUUID(), queueId = randomUUID(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.insert(agents).values({ id: nextAgentId, companyId: f.companyId, name: "Replacement", role: "engineer", adapterType: "paperclip_runner", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: nextAgentId, status: "running" }); + await db.update(issues).set({ status: "in_progress", assigneeAgentId: kind === "different_owner" ? f.agentId : nextAgentId }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: kind === "running_source" ? "running" : "cancelled", errorCode: "issue_reassigned", + nativeIssueId: kind === "foreign_task" ? null : f.issueId, + contextSnapshot: { issueId: kind === "foreign_task" ? randomUUID() : f.issueId } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const secondId = randomUUID(); + await db.insert(issueComments).values({ id: secondId, companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "second-user", body: "Preserve the existing draft." }); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: "board", idempotencyKey: kind === "chat" ? "chat-inbound:handoff-test" : null, + payload: { issueId: f.issueId, commentId: secondId, _paperclipWakeContext: { issueId: f.issueId, + wakeReason: kind === "mention" ? "issue_comment_mentioned" : "issue_commented", wakeCommentIds: [f.commentId, secondId], + ...(kind === "interaction" ? { interactionId: randomUUID(), wakeReason: "connection_intent.resolved" } : {}), + } }, + }); + await heartbeatService(db).wakeup(nextAgentId, { source: "assignment", triggerDetail: "system", reason: "issue_assigned", + requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, interruptedRunId: f.sourceRunId }, + contextSnapshot: { issueId: f.issueId, interruptedRunId: f.sourceRunId } }); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + if (kind === "handoff") { + expect(receipt.status).toBe("coalesced"); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)); + expect(successor).toMatchObject({ agentId: nextAgentId, status: "queued", contextSnapshot: { wakeCommentIds: [f.commentId, secondId] } }); + expect(receipt.requestedByActorId).toBe("board"); + const [comment] = await db.select().from(issueComments).where(eq(issueComments.id, secondId)); + expect(comment.authorUserId).toBe("second-user"); + } else expect(receipt.status).toBe("deferred_issue_execution"); + }); it.each(["ready", "unacknowledged", "pause", "recovery", "controller", "process_running", "identity_missing", "remote_pending", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"])("delivers a saved native message after run-only Stop exactly once (%s)", async gate => { const f = await seed(); if (gate !== "recovery") await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 3de2209b3c..741abbdbd2 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -27208,6 +27208,20 @@ export function heartbeatService( .returning() .then((rows) => rows[0]); + // A handoff changes the executor, not the owner of saved user input. + // Validate its exact stopped source while the issue row is locked; + // unrelated agents and dedicated continuations keep their own wakes. + const interruptedRunId = readNonEmptyString(enrichedContextSnapshot.interruptedRunId); + const handoffSource = source === "assignment" && reason === "issue_assigned" && + issue.assigneeAgentId === agentId && interruptedRunId && isUuidLike(interruptedRunId) + ? await tx.select({ agentId: heartbeatRuns.agentId }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, interruptedRunId), eq(heartbeatRuns.companyId, issue.companyId), + eq(heartbeatRuns.status, "cancelled"), eq(heartbeatRuns.errorCode, "issue_reassigned"), + ne(heartbeatRuns.agentId, agentId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`, + or(isNull(heartbeatRuns.nativeIssueId), eq(heartbeatRuns.nativeIssueId, issue.id)), + )).then(rows => rows[0] ?? null) + : null; const pendingComments = !isConversation(issue) && opts.allowRunCoalescing !== false && !(await getExecutionBlocker(tx as unknown as Db, issue.companyId, issue.id)) @@ -27217,7 +27231,7 @@ export function heartbeatService( .where( and( eq(agentWakeupRequests.companyId, issue.companyId), - eq(agentWakeupRequests.agentId, agentId), + inArray(agentWakeupRequests.agentId, handoffSource ? [agentId, handoffSource.agentId] : [agentId]), eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`, ), @@ -27250,7 +27264,7 @@ export function heartbeatService( ...queuedCommentIdsFromRunContext(enrichedContextSnapshot), ]), ]; - if (opts.queuedCommentRequestId) { + if (opts.queuedCommentRequestId || handoffSource) { adoptedCommentIds = await undeliveredLegacyUserCommentIds(tx as unknown as Db, agent.companyId, issueId, agentId, adoptedCommentIds); }