diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 1c259295ad5..7fe7b47478f 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2779,6 +2779,8 @@ not resource cleanup. `native-task-session-publication.ts` owns its accepted-res owner, task-epoch and reset checks; heartbeat supplies prepared session metadata. The goal-less regression in `native-session-resumption.test.ts` holds provider cleanup open and then exercises a normal follow-up and an explicit session reset. +Control-port tests also cover terminal snapshots before accepted results and +result-replay contention with the status writer's coordinator-first lock order. Run it against an isolated database with `PAPERCLIP_TEST_DATABASE_URL` when the host cannot start embedded Postgres: diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e012ea02b88..0db9d0df81c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2704,6 +2704,13 @@ so they cannot restore a cleared pointer or overwrite its successor. Unsupported legacy bindings keep the existing completion path. The receipt is excluded from public run projections and contains no session parameters or credentials. +A terminal snapshot can arrive before result acceptance. That snapshot publishes +neither a continuation pointer nor a handled publication receipt. The existing +post-acceptance checkpoint completes publication in its normal transaction. +The first production release containing recovery readers and redaction for both +private recovery fields is the disk-independent rollback floor. Earlier +development-reader probes do not establish a published rollback target. + For per-turn direct sandbox recovery, the executor reads the retained Runner's suspended lifecycle through the recorded sandbox lease and validates its complete identity and lack of pending delivery/cleanup. Only a lifecycle projection is diff --git a/server/src/services/native-runtime/native-task-session-publication.test.ts b/server/src/services/native-runtime/native-task-session-publication.test.ts index c5a68c91964..4ab4c73c059 100644 --- a/server/src/services/native-runtime/native-task-session-publication.test.ts +++ b/server/src/services/native-runtime/native-task-session-publication.test.ts @@ -4,7 +4,7 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { eq, sql } from "drizzle-orm"; import { drizzle } from "drizzle-orm/postgres-js"; import { agents, agentTaskSessions, companies, completionContracts, createDb, heartbeatRuns, issues, - nativeRunFinalizations, type Db } from "@paperclipai/db"; + nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { CONTROL_PLANE_CONFORMANCE_OPEN, CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../../vendor/paperclip-runner/testing.js"; import type { PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; @@ -13,7 +13,7 @@ import { startRecoveryPostgres } from "../../__tests__/helpers/recovery-postgres import { redactRegisteredSecretValues } from "../run-secret-redaction.js"; import { nativeSha256 } from "./canonical.js"; import { PaperclipControlPlanePort, type PaperclipControlPlaneBinding } from "./paperclip-control-plane-port.js"; -import { nativeTaskSessionPublicationHandled, NATIVE_TASK_SESSION_PUBLICATION_KEY, +import { nativeTaskSessionPublicationHandled, publishNativeTaskSessionInTransaction, NATIVE_TASK_SESSION_PUBLICATION_KEY, type NativeTaskSessionPublicationContext } from "./native-task-session-publication.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_PUBLICATION_TEST_DATABASE_URL"); @@ -40,7 +40,7 @@ describe("accepted native task session publication", () => { await temporary?.cleanup(); }); - async function fixture(source: "runner" | "control" = "runner") { + async function fixture(source: "runner" | "control" = "runner", accepted = true) { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); const sessionId = randomUUID(), contractId = randomUUID(), runner = randomUUID(), control = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Native pointer", issuePrefix: `P${companyId.slice(0, 7)}` }); @@ -69,7 +69,7 @@ describe("accepted native task session publication", () => { semanticResult: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }; const port = new PaperclipControlPlanePort(measured, binding, { taskSessionPublication: context, taskSessionPublicationOwner: owner }); await port.openRun({ ...CONTROL_PLANE_CONFORMANCE_OPEN, identity: snapshot.identity, sourceInstanceId: runner }); - await new PaperclipControlPlanePort(db, binding).completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, + if (accepted) await new PaperclipControlPlanePort(db, binding).completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId: "publication-turn" }); await db.insert(agentTaskSessions).values({ companyId, agentId, adapterType: context.adapterType, taskKey: issueId, goalSourceId: `${source === "runner" ? runner : control}:${runId}`, goalSourceCursor: 4 }); @@ -115,6 +115,66 @@ describe("accepted native task session publication", () => { expect(repeated.filter(query => /^update /i.test(query.trim()))).toHaveLength(1); }); + it("persists a pre-acceptance terminal observation and publishes only at the existing accepted checkpoint", async () => { + const f = await fixture("runner", false); + // Driver event/startup checkpoints can expose a semantic terminal observation + // before the runtime calls completeRun. That observation is not acceptance. + await f.port.checkpointSession(f.snapshot); + const observed = await f.read(); + expect(observed.run.runnerProfileJson?.sessionCheckpoint).toMatchObject({ semanticResult: CONTROL_PLANE_CONFORMANCE_RESULT }); + expect(observed.run.runnerProfileJson).not.toHaveProperty(NATIVE_TASK_SESSION_PUBLICATION_KEY); + expect(observed.session?.sessionParamsJson).toBeNull(); + expect(nativeTaskSessionPublicationHandled(observed.run.runnerProfileJson, { ...f.binding, ...f.context })).toBe(false); + await f.port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, + terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId: "publication-turn" }); + expect((await f.read()).session?.sessionParamsJson).toBeNull(); + await f.port.checkpointSession(f.snapshot); + const completed = await f.read(); + expect(completed.session?.lastRunId).toBe(f.binding.runId); + expect(nativeTaskSessionPublicationHandled(completed.run.runnerProfileJson, { ...f.binding, ...f.context })).toBe(true); + }); + + it("does not hold the issue parent lock while a result replay waits for the status coordinator", async () => { + const f = await fixture(); + let committed = 0; + const port = new PaperclipControlPlanePort(db, f.binding, { nativeRecovery: { owner: f.owner, + compatibility: { configDigest: nativeSha256("config"), processConfigDigest: null } }, + onResultCommitted: async () => { committed++; } }); + await port.openRun({ ...CONTROL_PLANE_CONFORMANCE_OPEN, identity: f.snapshot.identity, sourceInstanceId: f.binding.sourceInstanceId }); + let replay: Promise | undefined; + let statusLockError: unknown = null; + try { + await db.transaction(async tx => { + // This is commitNativeStatusDecision's actual lock prefix. Hold its + // coordinator first, then overlap an identical remote result replay. + await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.binding.runId)).for("update"); + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`); + replay = port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, + terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId: "publication-turn" }).then(() => null, error => error); + let waiting = false; + for (let attempt = 0; attempt < 100; attempt++) { + const [state] = await measured.execute(sql`select exists ( + select 1 from pg_stat_activity where ${backend.pid} = any(pg_blocking_pids(pid))) as waiting`); + if (state.waiting) { waiting = true; break; } + await new Promise(resolve => setTimeout(resolve, 10)); + } + expect(waiting).toBe(true); + // A bounded test-local timeout reports the inverse-order cycle without + // relying on which transaction PostgreSQL chooses as deadlock victim. + await tx.execute(sql`set local lock_timeout = '500ms'`); + await tx.select().from(issues).where(eq(issues.id, f.binding.issueId)).for("update"); + }); + } catch (error) { statusLockError = error; } + const replayError = await replay; + expect(statusLockError).toBeNull(); + expect(replayError).toBeNull(); + expect(committed).toBe(0); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.binding.runId))).toHaveLength(1); + await db.update(nativeRunFinalizations).set({ controllerGeneration: 2 }).where(eq(nativeRunFinalizations.runId, f.binding.runId)); + await expect(port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, + terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId: "publication-turn" })).rejects.toThrow("checkpoint_owner_lost"); + }); + it.each(["missing", "foreign_epoch", "newer_pointer", "newer_execution", "generation_reset", "stop"])("fences %s and records a handled suppression", async kind => { const f = await fixture(); if (kind === "missing") await db.delete(agentTaskSessions).where(eq(agentTaskSessions.taskKey, f.binding.issueId)); @@ -171,7 +231,11 @@ describe("accepted native task session publication", () => { .rejects.toMatchObject({ cause: { constraint_name: "native_run_finalizations_result_owner_fk" } }); expect((await foreign.read()).session?.sessionParamsJson).toBeNull(); await db.update(nativeRunFinalizations).set({ resultId: null }).where(eq(nativeRunFinalizations.runId, foreign.binding.runId)); - await expect(foreign.port.checkpointSession(foreign.snapshot)).rejects.toThrow("publication_result_missing"); + // The CP defers preacceptance snapshots, while the publication helper itself + // still requires accepted-result authority if called directly. + await expect(publishNativeTaskSessionInTransaction({} as never, { binding: foreign.binding, + context: foreign.context, snapshot: foreign.snapshot, run: (await foreign.read()).run, + issue: null, resultId: null })).rejects.toThrow("publication_result_missing"); }); it("adds no publication SQL for nonterminal or goal snapshots", async () => { diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index 81d764d2d4f..f7bac1622d3 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -300,7 +300,10 @@ export class PaperclipControlPlanePort implements ControlPlanePort { header ? { runId: this.#binding.runId, runnerInstanceId: header.runnerInstanceId, normalizedSessionId: header.normalizedSessionId } : null) } : {}), }; - if (publication && selected) { + // Driver/live snapshots may already describe a terminal result before + // completeRun accepts it. Persist that observation without a handled + // receipt; the existing post-acceptance checkpoint publishes the pointer. + if (publication && selected?.resultId) { const receipt = await publishNativeTaskSessionInTransaction(tx, { binding: this.#binding, context: publication, snapshot, run, issue: selected.publicationIssue ?? null, resultId: selected.resultId ?? null, recoveryReady: !recovery || header !== null }); @@ -485,14 +488,15 @@ export class PaperclipControlPlanePort implements ControlPlanePort { }); const insertedResult = await this.#db.transaction(async (tx) => { - // Result insertion checks the task foreign key. Acquire that parent lock - // before the run, matching task mutations that subsequently update a run. - // Otherwise concurrent chat/status writes can deadlock after generation. + // Status commit and snapshot publication lock coordinator → issue → run. + // A replay must follow that order too: holding the issue while waiting + // for the coordinator deadlocks a status commit's issue UPDATE lock. + const recovery = this.#nativeRecovery; + const owner = recovery ? await lockNativeRunnerCheckpointOwner(tx, recovery.owner) : null; + // Result insertion checks the task FK. Keep its parent lock before run. await tx.select({ id: issues.id }).from(issues) .where(and(eq(issues.id, this.#binding.issueId), eq(issues.companyId, this.#binding.companyId))) .for("key share"); - const recovery = this.#nativeRecovery; - const owner = recovery ? await lockNativeRunnerCheckpointOwner(tx, recovery.owner) : null; const run = await tx.select().from(heartbeatRuns) .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1) .then((rows) => rows[0] ?? null);