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); + }); +});