From c786e9c181a0a3e358b05d14a7e6d91e3382be0b Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 8 Oct 2026 20:03:23 -0500 Subject: [PATCH] Recover completed native task replies after workspace ownership handoff --- .../native-runtime/native-run-finalizer.ts | 77 +++++++++++++++++-- .../paperclip-control-plane-port.test.ts | 35 ++++++++- 2 files changed, 102 insertions(+), 10 deletions(-) diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index d986baf2c70..9d8c3f085d7 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -11,7 +11,7 @@ import { conversationNativeDecision, isConversation } from "../agent-conversatio import { issueTreeControlService } from "../issue-tree-control.js"; import { randomUUID } from "node:crypto"; import { preserveNativeWorkspaceExportLease } from "./native-workspace-export-resume.js"; -import { and, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { approvals, @@ -20,6 +20,7 @@ import { heartbeatRuns, heartbeatRunEvents, issueApprovals, + issueComments, issueRecoveryActions, issueThreadInteractions, issues, @@ -62,7 +63,11 @@ import { import { logger } from "../../middleware/logger.js"; import { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON, + findHeartbeatRunCompletionComment, + isExternalChatPresentationContext, + readCompletedAssistantMessageCandidate, resolveHeartbeatRunResponse, + selectHeartbeatRunFinalAgentMessage, } from "../heartbeat-run-summary.js"; import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js"; import { @@ -940,13 +945,6 @@ export async function repairCommittedNativeChatResponse( !accepted || !decision || result.schema !== "paperclip.run_result.v1" || - result.reportedWorkDisposition !== "yielded" || - record(result.continuation).kind !== "response_wake" || - !Array.isArray(result.attentionRequests) || - result.attentionRequests.length > 0 || - terminal.runTerminalState !== "succeeded" || - terminal.turnTerminalState !== "completed" || - terminal.reportedWorkDisposition !== "yielded" || !(await acceptedResponseDigestMatches(tx, run, accepted)) ) return false; @@ -954,6 +952,69 @@ export async function repairCommittedNativeChatResponse( // presentation contract, including selected attachments. Do not bypass it. if (record(decision.decisionJson).externalChatReviewPresentation) return false; + // Copyback can be owned by reconciliation before the live heartbeat + // reaches its presentation step. Recover a completed internal task's + // exact final reply from the same accepted turn, without rerunning work + // or granting external-chat publication authority. + if ( + !isConversation(issue) && + run.status === "succeeded" && + record(run.contextSnapshot).skipIssueComment !== true && + !isExternalChatPresentationContext(run.contextSnapshot) && + result.reportedWorkDisposition === "done" && + terminal.runTerminalState === "succeeded" && + terminal.turnTerminalState === "completed" && + terminal.reportedWorkDisposition === "done" && + (await resolveChatRunPresentationAuthorizationReason(tx, input)) === null + ) { + const rows = await tx.select({ seq: heartbeatRunEvents.seq, payload: heartbeatRunEvents.payload }) + .from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, input.companyId), + eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, "item.completed"), + sql`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId' = ${accepted.turnId}`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`, + )).orderBy(desc(heartbeatRunEvents.seq)).limit(200); + const finalAgentMessage = selectHeartbeatRunFinalAgentMessage({ + candidates: rows.flatMap((row) => { + const candidate = readCompletedAssistantMessageCandidate({ seq: row.seq, prpEvent: record(row.payload).prpEvent }); + return candidate ? [candidate] : []; + }), + }); + if (!finalAgentMessage) return false; + const comments = await tx.select({ id: issueComments.id, body: issueComments.body }) + .from(issueComments).where(and( + eq(issueComments.companyId, input.companyId), + eq(issueComments.issueId, input.issueId), + eq(issueComments.createdByRunId, run.id), + )).orderBy(desc(issueComments.createdAt), desc(issueComments.id)); + const resolved = resolveHeartbeatRunResponse({ + resultJson: { ...record(run.resultJson), nativeResult: result }, + existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson), + finalAgentMessage, + }); + if (!resolved.text || resolved.decision.commentAction !== "create") return false; + const comment = await issueService(db).addComment(input.issueId, resolved.text, + { agentId: run.agentId, runId: run.id }, { completionReply: true }, tx); + await tx.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ + presentationDecision: { ...resolved.decision, commentId: comment.id, + reasonCodes: [...resolved.decision.reasonCodes, "committed_task_response_recovered"] }, + })}::jsonb`, updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, run.id)); + agentId = run.agentId; + return true; + } + if ( + result.reportedWorkDisposition !== "yielded" || + record(result.continuation).kind !== "response_wake" || + !Array.isArray(result.attentionRequests) || + result.attentionRequests.length > 0 || + terminal.runTerminalState !== "succeeded" || + terminal.turnTerminalState !== "completed" || + terminal.reportedWorkDisposition !== "yielded" + ) return false; await authorizeCommittedChatResponse(db, tx, { ...input, agentId: run.agentId, diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 129ffdcf23a..9acb7e6a091 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -14,6 +14,7 @@ import { createDb, heartbeatRunEvents, heartbeatRuns, + issueComments, issueRecoveryActions, issueRelations, issueThreadInteractions, @@ -42,7 +43,7 @@ import { import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; -import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-run-finalizer.js"; import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; import { claimNativeReviewExecutionLock, getNativeReviewAssignment } from "./native-review-participant.js"; @@ -974,7 +975,37 @@ describe("PaperclipControlPlanePort conformance", () => { expect(response.decision.chosenSource).toBe("final_agent_message"); expect(stored.findIndex((row) => row.eventType === "run.result.accepted")) .toBeGreaterThan(stored.findIndex((row) => row.eventType === "item.completed")); - await finalizeNativeRun({ db, runId: taskRunId, workspaceFinalizeStatus: "succeeded" }); + // A workspace recovery owner can commit before the live heartbeat reaches + // presentation. The file-preparation comment must not hide the real reply. + const [prepared] = await db.insert(issueComments).values({ + companyId: identity.companyId, issueId: taskIssueId, + authorAgentId: identity.agentId, authorType: "agent", createdByRunId: taskRunId, + body: "Prepared Continuity file for this response.", + }).returning(); + const receipt = { semanticToolReceipts: { file: { + operationId: "register_deliverable", result: { + commandId: "deliverable-prepared:attachment-1", disposition: "applied", + attachmentId: "attachment-1", entityRefs: ["attachment-1", prepared!.id], + }, + } } }; + await db.update(heartbeatRuns).set({ + contextSnapshot: { skipIssueComment: true }, + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(receipt)}::jsonb`, + }).where(eq(heartbeatRuns.id, taskRunId)); + await finalizeNativeRun({ db, runId: taskRunId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const repair = () => repairCommittedNativeChatResponse(db, { + companyId: identity.companyId, issueId: taskIssueId, runId: taskRunId, + }); + expect(await repair()).toBe(false); + await db.update(heartbeatRuns).set({ contextSnapshot: { externalChatContinuation: true } }) + .where(eq(heartbeatRuns.id, taskRunId)); + expect(await repair()).toBe(false); + await db.update(heartbeatRuns).set({ contextSnapshot: {} }).where(eq(heartbeatRuns.id, taskRunId)); + expect(await repair()).toBe(true); + expect(await repair()).toBe(false); + expect(await db.select({ body: issueComments.body }).from(issueComments) + .where(eq(issueComments.createdByRunId, taskRunId)).orderBy(asc(issueComments.createdAt))) + .toEqual([{ body: prepared!.body }, { body: finalText }]); await expect(port.completeRun({ result: taskResult, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL,