diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 0875b177449..01e7a7031c4 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -499,3 +499,9 @@ for a native run. It keeps the same task-binding and native-finalization skip reasons, including the original nonblank binding values. Native input and history stay stored. Real legacy recovery still reads its complete context, profile, excerpts, and required failure evidence. + +The native Stop continuation recheck reads current run identity, the full Stop +acknowledgement receipt, and process identity. It does not load saved execution +input, runner profiles, excerpts, or unrelated result fields. Lease cleanup, +remote termination receipts, local process checks, task ownership, and admission +checks keep their existing behavior. Stored runs retain all of their data. diff --git a/server/src/services/acknowledged-native-stop.ts b/server/src/services/acknowledged-native-stop.ts index bb6b7fa74c3..9d07f7b61a7 100644 --- a/server/src/services/acknowledged-native-stop.ts +++ b/server/src/services/acknowledged-native-stop.ts @@ -37,7 +37,8 @@ export function isAcknowledgedNativeStop(run: Parameters) { if (!isAcknowledgedNativeStop(run)) return false; const leases = await db.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), diff --git a/server/src/services/heartbeat/native-stop-evidence-read.test.ts b/server/src/services/heartbeat/native-stop-evidence-read.test.ts new file mode 100644 index 00000000000..d53a5d4c417 --- /dev/null +++ b/server/src/services/heartbeat/native-stop-evidence-read.test.ts @@ -0,0 +1,93 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; +import { eq, sql } from "drizzle-orm"; +import { describe, expect, it, vi } from "vitest"; +import { isAcknowledgedNativeStop } from "../acknowledged-native-stop.js"; +import { callbacks, runControlPostgresSupport, setupRunControlDatabase } from "./__tests__/run-control-fixtures.js"; +import { readNativeStopExecutionEvidence } from "./native-stop-evidence-read.js"; +import { createHeartbeatRunControl } from "./run-control.js"; + +vi.mock("../local-service-supervisor.js", async importOriginal => ({ + ...await importOriginal(), + terminateLocalService: vi.fn(async () => undefined), +})); + +describe.skipIf(!runControlPostgresSupport.supported)("native Stop evidence reads", () => { + let db: Db; + const { fixture } = setupRunControlDatabase("native-stop-evidence-", value => { db = value; }); + + it("reads current receipts below the byte budget without changing saved data", async () => { + const { company, agent } = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Stop evidence fixture" }).returning(); + const id = randomUUID(), bulk = randomBytes(128 * 1024).toString("base64"); + const receipt = { schema: "paperclip.native-cancellation.v1", runId: id, companyId: company.id, issueId: issue.id, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID(), futureProof: { version: 2 } }; + const result = { nativeCancellation: receipt, cancelledByActorType: "user", cancelledByUserId: randomUUID(), + stdout: bulk, stderr: bulk, futureExecutionEvidence: { savedInput: bulk } }; + const [before] = await db.insert(heartbeatRuns).values({ id, companyId: company.id, agentId: agent.id, + nativeIssueId: issue.id, runtimeMode: "native", status: "cancelled", processPid: 123, processGroupId: 456, + contextSnapshot: { savedInput: bulk }, runnerProfileJson: { savedInput: bulk }, stdoutExcerpt: bulk, + stderrExcerpt: bulk, resultJson: result }).returning(); + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + let evidence: Awaited>; + try { evidence = await readNativeStopExecutionEvidence(db, company.id, id); } + finally { capture.flush(); } + expect(evidence!).toEqual({ id, companyId: company.id, nativeIssueId: issue.id, status: "cancelled", processPid: 123, + processGroupId: 456, resultJson: { nativeCancellation: receipt, cancelledByActorType: "user", cancelledByUserId: result.cancelledByUserId } }); + expect(isAcknowledgedNativeStop(evidence!)).toBe(true); + expect(events.at(-1)).toMatchObject({ completedQueries: 1, groups: 1 }); + expect(Number(events.at(-1)?.responseValueBytes)).toBeLessThan(2048); + expect(JSON.stringify(events)).not.toContain(bulk.slice(0, 64)); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]).toEqual(before); + await db.update(heartbeatRuns).set({ resultJson: { ...result, nativeCancellation: { ...receipt, dispatchState: "pending" } } }) + .where(eq(heartbeatRuns.id, id)); + expect(isAcknowledgedNativeStop((await readNativeStopExecutionEvidence(db, company.id, id))!)).toBe(false); + expect(await readNativeStopExecutionEvidence(db, randomUUID(), id)).toBeNull(); + expect(await readNativeStopExecutionEvidence(db, company.id, randomUUID())).toBeNull(); + }); + + it.each([null, [], false, 42, "not a receipt", { nativeCancellation: null }, { nativeCancellation: { dispatched: "true" } }])( + "keeps malformed receipts fail-closed: %j", async result => { + const { company, agent } = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + runtimeMode: "native", status: "cancelled" }).returning(); + await db.execute(sql`update ${heartbeatRuns} set result_json = ${JSON.stringify(result)}::jsonb where id = ${run.id}`); + expect(isAcknowledgedNativeStop((await readNativeStopExecutionEvidence(db, company.id, run.id))!)).toBe(false); + }, + ); + + it.each(["string", "array"])("does not transfer a large malformed %s result", async kind => { + const { company, agent } = await fixture(); + const bulk = randomBytes(128 * 1024).toString("base64"); + const saved = kind === "array" ? [bulk] : bulk; + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + runtimeMode: "native", status: "cancelled" }).returning(); + await db.execute(sql`update ${heartbeatRuns} set result_json = ${JSON.stringify(saved)}::jsonb where id = ${run.id}`); + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + try { + const evidence = await readNativeStopExecutionEvidence(db, company.id, run.id); + expect(evidence?.resultJson).toBeNull(); + expect(isAcknowledgedNativeStop(evidence!)).toBe(false); + } finally { capture.flush(); } + expect(events.at(-1)).toMatchObject({ completedQueries: 1, groups: 1 }); + expect(Number(events.at(-1)?.responseValueBytes)).toBeLessThan(1024); + expect((await db.select({ result: heartbeatRuns.resultJson }).from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].result).toEqual(saved); + }); + + it("uses the scoped receipt reader for the native continuation recheck", async () => { + const { company, agent } = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Native continuation fixture" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + nativeIssueId: issue.id, runtimeMode: "native", status: "succeeded" }).returning(); + const deps = callbacks(db); + const control = createHeartbeatRunControl(db, deps); + await control.resumeRemoteStopComments(run); + expect(deps.getRun).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/services/heartbeat/native-stop-evidence-read.ts b/server/src/services/heartbeat/native-stop-evidence-read.ts new file mode 100644 index 00000000000..6ee776bf03f --- /dev/null +++ b/server/src/services/heartbeat/native-stop-evidence-read.ts @@ -0,0 +1,20 @@ +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { and, eq, sql } from "drizzle-orm"; + +export async function readNativeStopExecutionEvidence(db: Db, companyId: string, runId: string) { + const [run] = await db.select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + nativeIssueId: heartbeatRuns.nativeIssueId, + status: heartbeatRuns.status, + processPid: heartbeatRuns.processPid, + processGroupId: heartbeatRuns.processGroupId, + resultJson: sql`case + when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object' then ( + select coalesce(jsonb_object_agg(key, value), '{}'::jsonb) + from jsonb_each(${heartbeatRuns.resultJson}) + where key = any(array['nativeCancellation', 'cancelledByActorType', 'cancelledByUserId']) + ) else null end`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).limit(1); + return run ?? null; +} diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index d7086c52508..b4ba5f0ddea 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,4 +1,5 @@ import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; +import { readNativeStopExecutionEvidence } from "./native-stop-evidence-read.js"; import { readTaskRetryIntent, reconcileTaskChatRetryReceipts } from "../task-retry.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, readNativeLocalProcessStop, recordedLocalExecutionStopped } from "../native-local-process-stop.js"; import { isHeartbeatRunTerminalStatus } from "./run-lifecycle.js"; @@ -522,7 +523,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont !(await recordedLocalExecutionStopped(db, run))) return; const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null); if (!issueId) return; - const currentRun = run.runtimeMode === "native" ? await getRun(run.id, { includeExecutionEvidence: true }) : null; + const currentRun = run.runtimeMode === "native" ? await readNativeStopExecutionEvidence(db, run.companyId, run.id) : null; const [coordinator] = currentRun ? await db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id),