Recover completed native task replies after workspace ownership handoff

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,

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,