Read reply repair and ownership evidence without execution payloads (#17501)

Narrow native reply repair and conversation ownership reads to their consumed evidence while preserving full stored history, result receipts, locks, current authority checks, and publication authorization.

Verify successful repair from equivalent unpublished states, exact final replies and decisions, no duplicate repair, preserved history, and 213 recovery/presentation tests. Build/typecheck and all CI pass. Greptile 5/5 Medium.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/DATABASE.md

@@ -68,6 +68,13 @@ Runner profiles retain the adapter type and complete native execution input and
 workspace-sync receipts. Full result evidence stays available to the existing
 retry and workspace checks. Stored context, profiles, and output are unchanged.
 
+Conversation ownership checks select process identity and full recovery result
+evidence without execution prompts or profiles. Committed native reply repair
+selects its persisted result and identity bindings plus typed presentation flags.
+It keeps its existing row locks, accepted-result digest checks, publication
+authorization, and selected-answer protections. Full execution history stays in
+storage.
+
 Run-context lookups use expression indexes for both `context_snapshot.issueId`
 and `context_snapshot.paperclipIssue.id`. Task creation provenance has its own
 company-scoped expression index that follows the query's native issue ID and

server/src/services/chat-monitor-continuation.test.ts

@@ -1,4 +1,4 @@
-import { randomUUID } from "node:crypto";
+import { randomBytes, randomUUID } from "node:crypto";
 import { and, eq } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import {
@@ -237,6 +237,12 @@ if (!support.supported) console.warn(`Monitor publication database unavailable:
 
   it("recovers the accepted continuation explanation after interrupted finalization without another provider result", async () => {
     const f = await fixture();
+    const executionPayload = randomBytes(128 * 1024).toString("base64");
+    await db.update(heartbeatRuns).set({
+      contextSnapshot: { ...f.source.contextSnapshot, largePrompt: executionPayload,
+        paperclipWake: { prompt: executionPayload } },
+      runnerProfileJson: { instructions: executionPayload }, stdoutExcerpt: executionPayload,
+    }).where(eq(heartbeatRuns.id, f.source.id));
     const [contract] = await db.insert(completionContracts).values({ companyId: f.company.id, issueId: f.issue.id,
       revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard",
       completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
@@ -270,6 +276,9 @@ if (!support.supported) console.warn(`Monitor publication database unavailable:
     expect(coordinator).toMatchObject({ phase: "committed" });
     const [settledRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.source.id));
     expect(settledRun).toMatchObject({ status: "succeeded", resultJson: { finalizationPhase: "committed" } });
+    expect(settledRun.contextSnapshot?.largePrompt).toBe(executionPayload);
+    expect(settledRun.runnerProfileJson?.instructions).toBe(executionPayload);
+    expect(settledRun.stdoutExcerpt).toBe(executionPayload);
     const priorComments = await db.select({ body: issueComments.body }).from(issueComments).where(eq(issueComments.createdByRunId, f.source.id));
     expect(priorComments).toEqual([{ body: summary }]);
     expect(await repair()).toBe(false);

server/src/services/conversation-continuation.ts

@@ -111,7 +111,14 @@ export async function getConversationOwnershipBlocker(db: Db, companyId: string,
       and (${environmentLeases.releasedAt} is null
         or ${environmentLeases.status} = 'pending_cleanup'
         or ${environmentLeases.cleanupStatus} = 'failed'))`;
-  const candidates = await db.select({ run: heartbeatRuns, activeLease }).from(heartbeatRuns)
+  const candidates = await db.select({ run: {
+    id: heartbeatRuns.id,
+    agentId: heartbeatRuns.agentId,
+    processPid: heartbeatRuns.processPid,
+    processGroupId: heartbeatRuns.processGroupId,
+    processStartedAt: heartbeatRuns.processStartedAt,
+    resultJson: heartbeatRuns.resultJson,
+  }, activeLease }).from(heartbeatRuns)
     .where(and(
       eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.runtimeMode, "legacy"),
       conversationRunPredicate(),

server/src/services/native-runtime/native-run-finalizer.ts

@@ -596,7 +596,7 @@ function recoveredExecutionFailureMetadata() {
 
 async function projectCommittedRun(input: {
   db: Db;
-  run: typeof heartbeatRuns.$inferSelect;
+  run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "status" | "finishedAt" | "runtimeMode" | "nativePhase">;
   coordinator: typeof nativeRunFinalizations.$inferSelect;
 }) {
   if (!input.coordinator.resultId) return;
@@ -764,7 +764,9 @@ async function materializeCommittedReviewResponse(db: Db, runId: string) {
 
 async function acceptedResponseDigestMatches(
   db: Db,
-  run: typeof heartbeatRuns.$inferSelect,
+  run: Pick<typeof heartbeatRuns.$inferSelect,
+    "id" | "companyId" | "agentId" | "nativeIssueId" | "nativeSessionId" |
+    "completionContractId" | "completionContractSha256" | "runnerInstanceId">,
   accepted: typeof nativeRunResults.$inferSelect,
 ): Promise<boolean> {
   const envelope = record(accepted.resultJson);
@@ -842,7 +844,30 @@ export async function repairCommittedNativeChatResponse(
         .for("update");
       if (!issue) return false;
       const [run] = await tx
-        .select()
+        .select({
+          id: heartbeatRuns.id,
+          companyId: heartbeatRuns.companyId,
+          agentId: heartbeatRuns.agentId,
+          nativeIssueId: heartbeatRuns.nativeIssueId,
+          nativeSessionId: heartbeatRuns.nativeSessionId,
+          completionContractId: heartbeatRuns.completionContractId,
+          completionContractSha256: heartbeatRuns.completionContractSha256,
+          runnerInstanceId: heartbeatRuns.runnerInstanceId,
+          runtimeMode: heartbeatRuns.runtimeMode,
+          status: heartbeatRuns.status,
+          errorCode: heartbeatRuns.errorCode,
+          nativePhase: heartbeatRuns.nativePhase,
+          finishedAt: heartbeatRuns.finishedAt,
+          resultJson: heartbeatRuns.resultJson,
+          contextSnapshot: sql<Record<string, unknown> | null>`case when ${heartbeatRuns.contextSnapshot} is null then null
+            else jsonb_build_object(
+              'skipIssueComment', ${heartbeatRuns.contextSnapshot}->'skipIssueComment',
+              'source', ${heartbeatRuns.contextSnapshot}->'source',
+              'externalChatContinuation', ${heartbeatRuns.contextSnapshot}->'externalChatContinuation',
+              'paperclipWake', jsonb_build_object('externalInteractionContinuation',
+                ${heartbeatRuns.contextSnapshot} #> '{paperclipWake,externalInteractionContinuation}')
+            ) end`,
+        })
         .from(heartbeatRuns)
         .where(
           and(

server/src/services/presentation-ownership-read.test.ts

@@ -0,0 +1,214 @@
+import { randomBytes, randomUUID } from "node:crypto";
+import { agents, companies, completionContracts, createDb, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
+import { eq, type SQL } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
+import { NativeRunCoordinatorStore } from "./native-runtime/native-run-coordinator-store.js";
+import { getConversationOwnershipBlocker } from "./conversation-continuation.js";
+import { isExternalChatPresentationContext } from "./heartbeat-run-summary.js";
+import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read projections", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  beforeAll(async () => {
+    database = await startEmbeddedPostgresTestDatabase("presentation-ownership-read-");
+    db = createDb(database.connectionString);
+  }, 60_000);
+  afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
+
+  async function fixture(patch: Partial<typeof heartbeatRuns.$inferInsert> = {}) {
+    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Read fixture", issuePrefix: companyId.slice(0, 8) });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Read agent", adapterType: "codex_local" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Read task", status: "blocked", assigneeAgentId: agentId });
+    const payload = randomBytes(128 * 1024).toString("base64");
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: issueId,
+      status: "failed", runtimeMode: "legacy", finishedAt: new Date("2020-01-01T00:00:00Z"),
+      contextSnapshot: { issueId, source: "chat:fixture", prompt: payload,
+        paperclipWake: { externalInteractionContinuation: true, prompt: payload } },
+      runnerProfileJson: { adapterDispatch: { adapterType: "codex_local", instructions: payload }, nativeExecutionInput: { prompt: payload } },
+      resultJson: { unknownFutureRecoveryReceipt: { preserved: true } }, stdoutExcerpt: payload, stderrExcerpt: payload,
+      ...patch,
+    }).returning();
+    return { companyId, agentId, issueId, run };
+  }
+
+  function captureReads(fullRows = false) {
+    const rows: unknown[][] = [];
+    function recordSelects<T extends Pick<Db, "select">>(executor: T): T {
+      return new Proxy(executor, { get(target, key) {
+        if (key !== "select") return Reflect.get(target, key);
+        return (...fields: Parameters<Db["select"]>) => {
+          const builder = target.select(...fields), from = builder.from.bind(builder);
+          builder.from = ((...args: Parameters<typeof from>) => {
+            const selection = fields[0];
+            const nested = selection && "run" in selection && "activeLease" in selection;
+            const repairSource = selection && "nativeSessionId" in selection && "completionContractId" in selection && "finishedAt" in selection;
+            const sourceRead = args[0] === heartbeatRuns && (nested || repairSource);
+            const query = fullRows && sourceRead
+              ? (nested ? target.select({ run: heartbeatRuns, activeLease: selection.activeLease as SQL }) : target.select()).from(heartbeatRuns)
+              : from(...args);
+            if (sourceRead) {
+              const execute = query.execute.bind(query);
+              query.execute = async (...executeArgs: Parameters<typeof execute>) => {
+                const result = await execute(...executeArgs); rows.push(result); return result;
+              };
+            }
+            return query;
+          }) as typeof builder.from;
+          return builder;
+        };
+      }});
+    }
+    const root = recordSelects(db);
+    const transaction: Db["transaction"] = (callback, config) => db.transaction(tx => callback(recordSelects(tx)), config);
+    const recorded = new Proxy(root, { get(target, key) { return key === "transaction" ? transaction : Reflect.get(target, key); } });
+    return { recorded, rows };
+  }
+
+  async function compareOwnership(f: Awaited<ReturnType<typeof fixture>>) {
+    const baseline = captureReads(true), projected = captureReads();
+    const expected = await getConversationOwnershipBlocker(baseline.recorded, f.companyId, f.issueId);
+    const actual = await getConversationOwnershipBlocker(projected.recorded, f.companyId, f.issueId);
+    expect(actual).toEqual(expected);
+    expect(projected.rows.flat()).toHaveLength(1);
+    expect(Buffer.byteLength(JSON.stringify(projected.rows))).toBeLessThan(4096);
+    expect(Buffer.byteLength(JSON.stringify(baseline.rows))).toBeGreaterThan(128 * 1024);
+    return actual;
+  }
+
+  it("keeps current process authority without reading execution payloads", async () => {
+    const f = await fixture({ processPid: process.pid });
+    expect(await compareOwnership(f)).toMatchObject({ runId: f.run.id, cause: "execution_owner_active" });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]).toEqual(f.run);
+    expect(await getConversationOwnershipBlocker(db, randomUUID(), f.issueId)).toBeNull();
+    await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.run.id));
+    expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toBeNull();
+  });
+
+  it("retains an active lease until the current database row releases it", async () => {
+    const f = await fixture();
+    const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.run.id,
+      issueId: f.issueId, provider: "daytona", providerLeaseId: "fixture-lease", status: "active" }).returning();
+    expect(await compareOwnership(f)).toMatchObject({ runId: f.run.id, cause: "execution_owner_active" });
+    expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0]).toEqual(lease);
+    await db.update(environmentLeases).set({ status: "released", releasedAt: new Date("2020-01-01T00:00:00Z") }).where(eq(environmentLeases.id, lease.id));
+    expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toBeNull();
+  });
+
+  it("keeps unconfirmed remote cleanup and all recovery evidence", async () => {
+    const f = await fixture();
+    const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.run.id,
+      issueId: f.issueId, provider: "daytona", providerLeaseId: "retained-fixture", status: "released",
+      releasedAt: new Date("2020-01-01T00:00:00Z"), cleanupStatus: "success" }).returning();
+    const result = { ...f.run.resultJson, workspaceRestoreFailure: "restore_unsafe_archive",
+      workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [lease.id] } };
+    await db.update(heartbeatRuns).set({ resultJson: result }).where(eq(heartbeatRuns.id, f.run.id));
+    expect(await compareOwnership(f)).toMatchObject({ runId: f.run.id, cause: "execution_owner_active" });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].resultJson).toEqual(result);
+  });
+
+  async function compareRepair(f: Awaited<ReturnType<typeof fixture>>) {
+    const baseline = captureReads(true), projected = captureReads();
+    const input = { companyId: f.companyId, issueId: f.issueId, runId: f.run.id };
+    expect(await repairCommittedNativeChatResponse(projected.recorded, input))
+      .toBe(await repairCommittedNativeChatResponse(baseline.recorded, input));
+    expect(projected.rows.flat()).toHaveLength(1);
+    expect(Buffer.byteLength(JSON.stringify(projected.rows))).toBeLessThan(4096);
+    expect(Buffer.byteLength(JSON.stringify(baseline.rows))).toBeGreaterThan(128 * 1024);
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]).toEqual(f.run);
+    return projected.rows.flat()[0] as { contextSnapshot: Record<string, unknown> | null; resultJson: unknown };
+  }
+
+  it("preserves selected answers without downloading prompts, profiles, or excerpts", async () => {
+    const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: {
+      presentationDecision: { commentId: randomUUID() }, unknownFutureReceipt: { preserved: true } } });
+    const row = await compareRepair(f);
+    expect(row.resultJson).toEqual(f.run.resultJson);
+    expect(await repairCommittedNativeChatResponse(db, { companyId: randomUUID(), issueId: f.issueId, runId: f.run.id })).toBe(false);
+    const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task" }).returning();
+    expect(await repairCommittedNativeChatResponse(db, { companyId: f.companyId, issueId: other.id, runId: f.run.id })).toBe(false);
+  });
+
+  async function unpublishedAcceptedReply() {
+    const payload = randomBytes(128 * 1024).toString("base64");
+    const f = await fixture({ runtimeMode: "native", status: "running", finishedAt: null,
+      contextSnapshot: { source: "internal", prompt: payload, paperclipWake: { prompt: payload } } });
+    const [contract] = await db.insert(completionContracts).values({ companyId: f.companyId, issueId: f.issueId,
+      revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard",
+      completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
+      contractJson: { revision: "test", objective: "Finish the fixture task", criteria: [] },
+      canonicalSha256: "read-fixture-contract", createdByActorType: "system", createdByActorId: "test" }).returning();
+    const sessionId = randomUUID(), runnerId = randomUUID();
+    await db.update(heartbeatRuns).set({ nativeSessionId: sessionId, runnerInstanceId: runnerId,
+      completionContractId: contract.id, completionContractSha256: contract.canonicalSha256 }).where(eq(heartbeatRuns.id, f.run.id));
+    await db.update(issues).set({ status: "in_progress", executionRunId: f.run.id }).where(eq(issues.id, f.issueId));
+    const port = new NativeRunCoordinatorStore(db, { companyId: f.companyId, issueId: f.issueId,
+      runId: f.run.id, agentId: f.agentId, normalizedSessionId: sessionId, completionContractId: contract.id,
+      completionContractSha256: contract.canonicalSha256, runnerSourceInstanceId: runnerId,
+      completionContractRevision: "test", completionContractCriterionIds: [] });
+    const reply = "The accepted provider reply is preserved exactly.";
+    await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.run.id, agentId: f.agentId,
+      seq: 1_000_000, eventType: "item.completed", payload: { prpEvent: { turnId: "reply-turn", sourceEventId: "fixture-final-answer",
+        payload: { kind: "agentMessage", channel: "final", text: reply } } } });
+    await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
+      summary: "The accepted summary differs from its final answer.", completionClaim: { contractRevision: "test",
+        objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: [], artifacts: [] },
+      terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" },
+      turnId: "reply-turn" });
+    await finalizeNativeRun({ db, runId: f.run.id, workspaceFinalizeStatus: "succeeded", projectRunStatus: false });
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)))[0]).toMatchObject({ phase: "committed" });
+    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:00:00Z") }).where(eq(heartbeatRuns.id, f.run.id));
+    expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id))).toEqual([]);
+    const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
+    expect(before.resultJson?.presentationDecision).toBeUndefined();
+    return { ...f, before, reply };
+  }
+
+  it("publishes the same accepted reply and decision from equivalent full-row and projected starting states", async () => {
+    const outcomes: unknown[] = [];
+    for (const fullRows of [true, false]) {
+      const f = await unpublishedAcceptedReply(), captured = captureReads(fullRows);
+      const input = { companyId: f.companyId, issueId: f.issueId, runId: f.run.id };
+      expect(await repairCommittedNativeChatResponse(captured.recorded, input)).toBe(true);
+      const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id));
+      expect(comments).toHaveLength(1); expect(comments[0].body).toBe(f.reply);
+      const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
+      const decision = after.resultJson!.presentationDecision as Record<string, unknown>;
+      expect(decision.commentId).toBe(comments[0].id);
+      const { commentId: _, ...stableDecision } = decision;
+      outcomes.push({ body: comments[0].body, decision: stableDecision });
+      expect(after.resultJson).toMatchObject(f.before.resultJson!);
+      expect(after.contextSnapshot).toEqual(f.before.contextSnapshot);
+      expect(after.runnerProfileJson).toEqual(f.before.runnerProfileJson);
+      expect(after.stdoutExcerpt).toBe(f.before.stdoutExcerpt);
+      expect(after.stderrExcerpt).toBe(f.before.stderrExcerpt);
+      expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.run.id))).toHaveLength(1);
+      expect(await repairCommittedNativeChatResponse(db, input)).toBe(false);
+      expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id))).toHaveLength(1);
+      expect(captured.rows.flat()).toHaveLength(1);
+      if (fullRows) expect(Buffer.byteLength(JSON.stringify(captured.rows))).toBeGreaterThan(128 * 1024);
+      else expect(Buffer.byteLength(JSON.stringify(captured.rows))).toBeLessThan(4096);
+    }
+    expect(outcomes[1]).toEqual(outcomes[0]);
+  });
+
+  it.each([null, [], "scalar", 42, { source: 42, skipIssueComment: "true", paperclipWake: [] },
+    { source: "issue.comment.created", externalChatContinuation: true },
+    { source: "chat:fixture", skipIssueComment: true, paperclipWake: { externalInteractionContinuation: "true" } },
+    { source: "internal", paperclipWake: { externalInteractionContinuation: true } },
+  ].map(context => ({ context })))("preserves typed publication flags for $context", async ({ context }) => {
+    const f = await fixture({ runtimeMode: "native", status: "succeeded", contextSnapshot: context as Record<string, unknown> | null });
+    const row = await compareRepair(f);
+    expect(isExternalChatPresentationContext(row.contextSnapshot)).toBe(isExternalChatPresentationContext(context));
+    const original = context && typeof context === "object" && !Array.isArray(context) ? context : {};
+    expect(row.contextSnapshot?.skipIssueComment === true).toBe((original as Record<string, unknown>).skipIssueComment === true);
+  });
+
+  it.each([null, ["stdout", { receipt: true }], "scalar", 42].map(result => ({ result })))("preserves non-object result evidence $result", async ({ result }) => {
+    const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: result as Record<string, unknown> | null });
+    expect((await compareRepair(f)).resultJson).toEqual(result);
+  });
+});