From 470cb34c0e42cb9d0b232d0bcf5d45cc7b0a81ca Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 24 Oct 2026 15:14:48 -0700 Subject: [PATCH] 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 --- doc/DATABASE.md | 7 + .../chat-monitor-continuation.test.ts | 11 +- .../src/services/conversation-continuation.ts | 9 +- .../native-runtime/native-run-finalizer.ts | 31 ++- .../presentation-ownership-read.test.ts | 214 ++++++++++++++++++ 5 files changed, 267 insertions(+), 5 deletions(-) create mode 100644 server/src/services/presentation-ownership-read.test.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 1ff88e3368d..13424c8fad6 100644 --- a/doc/DATABASE.md +++ b/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 diff --git a/server/src/services/chat-monitor-continuation.test.ts b/server/src/services/chat-monitor-continuation.test.ts index 01b094800e4..f8e277d587b 100644 --- a/server/src/services/chat-monitor-continuation.test.ts +++ b/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); diff --git a/server/src/services/conversation-continuation.ts b/server/src/services/conversation-continuation.ts index 36e8b7a4804..cd0c54a2dcd 100644 --- a/server/src/services/conversation-continuation.ts +++ b/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(), diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index ff5d3267a9d..8b2ae73c349 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/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; 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, accepted: typeof nativeRunResults.$inferSelect, ): Promise { 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 | 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( diff --git a/server/src/services/presentation-ownership-read.test.ts b/server/src/services/presentation-ownership-read.test.ts new file mode 100644 index 00000000000..08ee36f558e --- /dev/null +++ b/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>; + let db: ReturnType; + 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 = {}) { + 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>(executor: T): T { + return new Proxy(executor, { get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + 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) => { + 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>) { + 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>) { + 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 | 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; + 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 | 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).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 | null }); + expect((await compareRepair(f)).resultJson).toEqual(result); + }); +});