diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 17ee6c4f445..970d2c28670 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2240,6 +2240,27 @@ replacement launches use current release authority. A missing historical receipt retains the current release gate; guest files cannot establish old authority. A missing, changed, or incompatible package must be replaced and verified before launch. +The server records the closed `runnerProfileJson.nativeProviderPackAdmission` +receipt after actual session admission and attachment. Its `session_admitted` +stage and `core_verified_selection` authority describe Core's verified selection, +not an independent provider statement. It includes only the bound run, session, +runner, lease and pack identities. Retained attachments keep the original pack +and identify `inheritedFromRunId`. The first receipt is immutable history. +`nativeProviderPackAdmissionState` separately describes current admission proof: +`pending`, `verified` or `unknown`, with an opaque admission epoch. The mandatory +coordinator claim atomically sets a new pending epoch before session admission. +Only the matching owner and epoch can publish post-attachment proof. Failed +receipt writes leave pending evidence; changed or unsupported admission leaves +unknown evidence and does not prevent authorized execution. A missing legacy +receipt remains unknown; the current release cannot supply that missing history. +Retained recovery needs positively verified prior provenance, captured before +same-run invalidation or read from the exact source run. A fresh replacement +uses its actual newly verified selection. Recording the receipt does not grant +execution or change the existing recovery gates. Admission proof does not certify +provider liveness, physical termination or workspace export. +The `runnerEnvironmentLeaseId` names the logical Runner and provider-pack +authority scope. It can differ from a physical `environment_leases.id`. + Warm attachment requires two consecutive authenticated readiness snapshots. Blocked readiness probes back off within the reconnect deadline so they do not fill the durable command journal while waiting. The fast ready path keeps its diff --git a/server/src/services/native-runtime/native-provider-pack-admission.test.ts b/server/src/services/native-runtime/native-provider-pack-admission.test.ts new file mode 100644 index 00000000000..b5380d173ac --- /dev/null +++ b/server/src/services/native-runtime/native-provider-pack-admission.test.ts @@ -0,0 +1,229 @@ +import { createHash, randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { eq } from "drizzle-orm"; +import { agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import type { NativeSession } from "../../vendor/paperclip-runner/index.js"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { canonicalProviderPackJson, type ProviderPackIdentity } from "./remote-provider-pack-verification.js"; +import { admittedNativeProviderPack, nativeProviderPackAdmissionPending, recordNativeProviderPackAdmission, + rememberAdmittedNativeProviderPack, rememberRecoveredNativeProviderPack } from "./native-provider-pack-admission.js"; + +const hash = (value: string) => `sha256:${createHash("sha256").update(value).digest("hex")}`; +function manifest(revision = "a".repeat(40)): ProviderPackIdentity { + const artifact = { path: "/private/provider-path", sha256: hash(revision) }; + const payload = { runnerSourceRevision: revision, pins: { opencode: "1.18.34" }, + target: { platform: "linux", architecture: "x64" }, + artifacts: { nodeCommand: artifact, opencodeCommand: artifact, opencodeProxy: artifact } }; + return { schema: "paperclip-runner/remote-provider-pack/v1", payload, digest: hash(canonicalProviderPackJson(payload)) }; +} +const initial = { companyId: "company", agentId: "agent", issueId: "issue", runId: "source-run", sessionId: "session" }; +function sessionFixture(identity = initial) { + let current = { ...identity }; + const session = { identity: () => ({ ...current }), attachRun: async ({ identity: next }: { identity: typeof initial }) => { current = { ...next }; } } as NativeSession; + return { session, attach: (next: typeof initial) => session.attachRun!({ identity: next }) }; +} +const binding = { runnerInstanceId: "runner", environmentLeaseId: "lease" }; +function remember(session: NativeSession, identity = initial, selected = manifest()) { + return rememberAdmittedNativeProviderPack(session, { identity, ...binding, manifest: selected }); +} +const verified = { ...nativeProviderPackAdmissionPending(randomUUID()), state: "verified" }; +function receiptDb(receipt: unknown, state: unknown = verified): Db { + return { select: () => ({ from: () => ({ where: () => ({ limit: async () => [{ receipt, state }] }) }) }) } as unknown as Db; +} + +describe("provider pack admission identity", () => { + it("does not publish selected configuration before successful session admission", async () => { + const { session } = sessionFixture(); + expect(admittedNativeProviderPack(session, initial)).toBeNull(); + }); + it("keeps a closed original pack identity through an actual target attachment", async () => { + const { session, attach } = sessionFixture(); + const original = manifest(); + expect(remember(session, initial, original)).toBe(true); + const before = admittedNativeProviderPack(session, initial)!; + const target = { ...initial, runId: "target-run", issueId: "target-issue" }; + expect(admittedNativeProviderPack(session, target)).toBeNull(); + await attach(target); + const next = admittedNativeProviderPack(session, target)!; + expect(next).toEqual({ ...before, runId: target.runId, inheritedFromRunId: initial.runId }); + expect(JSON.stringify(next)).not.toContain("private/provider-path"); + expect(Object.keys(next).sort()).toEqual(["schema", "stage", "authority", "companyId", "agentId", "runId", "sessionId", + "runnerInstanceId", "runnerEnvironmentLeaseId", "providerPackDigest", "opencodeProxySha256", "sourceRevision", "inheritedFromRunId"].sort()); + expect(remember(session, target, original)).toBe(true); + expect(admittedNativeProviderPack(session, target)).toEqual(next); + expect(remember(session, target, manifest("b".repeat(40)))).toBe(true); + expect(admittedNativeProviderPack(session, target)).toMatchObject({ sourceRevision: "b".repeat(40), runId: target.runId }); + }); + it.each(["companyId", "agentId", "issueId", "runId", "sessionId"] as const)("rejects a returned session with a different %s", key => { + const { session } = sessionFixture(); + expect(remember(session, { ...initial, [key]: "foreign" })).toBe(false); + expect(admittedNativeProviderPack(session, initial)).toBeNull(); + }); + it("rejects stale manifest digests and unavailable source identity", () => { + const { session } = sessionFixture(); + const stale = manifest(); stale.payload.artifacts.opencodeProxy.sha256 = hash("changed"); + expect(remember(session, initial, stale)).toBe(false); + expect(remember(session, initial, manifest("unknown"))).toBe(false); + expect(admittedNativeProviderPack(session, initial)).toBeNull(); + }); + it("does not reuse an earlier in-memory proof for an unsupported new admission", async () => { + const { session } = sessionFixture(); remember(session); + expect(remember(session, initial, manifest("unknown"))).toBe(false); + expect(admittedNativeProviderPack(session, initial)).toBeNull(); + remember(session); + await expect(rememberRecoveredNativeProviderPack(receiptDb(undefined), session, + { identity: initial, sourceRunId: initial.runId, ...binding })).resolves.toBe(false); + expect(admittedNativeProviderPack(session, initial)).toBeNull(); + }); + it("restores only the exact prior closed Core receipt and leaves legacy absence unknown", async () => { + const source = sessionFixture(); remember(source.session); + const saved = admittedNativeProviderPack(source.session, initial)!; + const recovered = sessionFixture(); + await expect(rememberRecoveredNativeProviderPack(receiptDb(undefined), recovered.session, + { identity: initial, sourceRunId: initial.runId, ...binding })).resolves.toBe(false); + expect(admittedNativeProviderPack(recovered.session, initial)).toBeNull(); + await expect(rememberRecoveredNativeProviderPack(receiptDb(saved), recovered.session, + { identity: initial, sourceRunId: initial.runId, ...binding })).resolves.toBe(true); + const target = { ...initial, runId: "target-run" }; + expect(admittedNativeProviderPack(recovered.session, target)).toBeNull(); + await recovered.attach(target); + expect(admittedNativeProviderPack(recovered.session, target)).toEqual({ ...saved, runId: target.runId, inheritedFromRunId: initial.runId }); + }); + it.each(["companyId", "agentId", "runId", "sessionId", "runnerInstanceId", "runnerEnvironmentLeaseId"] as const)("rejects a prior receipt with a different %s", async key => { + const source = sessionFixture(); remember(source.session); + const saved = { ...admittedNativeProviderPack(source.session, initial)!, [key]: "foreign" }; + const recovered = sessionFixture(); + await expect(rememberRecoveredNativeProviderPack(receiptDb(saved), recovered.session, + { identity: initial, sourceRunId: initial.runId, ...binding })).resolves.toBe(false); + expect(admittedNativeProviderPack(recovered.session, initial)).toBeNull(); + }); + it.each([undefined, { ...verified, state: "pending" }, { ...verified, state: "unknown" }])("does not revive a historical receipt without verified current provenance: %j", async state => { + const source = sessionFixture(); remember(source.session); + const recovered = sessionFixture(); + const absentState = state === undefined ? null : state; + await expect(rememberRecoveredNativeProviderPack(receiptDb(admittedNativeProviderPack(source.session, initial), absentState), recovered.session, + { identity: initial, sourceRunId: initial.runId, ...binding })).resolves.toBe(false); + expect(admittedNativeProviderPack(recovered.session, initial)).toBeNull(); + }); + it("uses only the exact same-run locked pre-invalidation source snapshot", async () => { + const source = sessionFixture(); remember(source.session); + const saved = admittedNativeProviderPack(source.session, initial)!; + const recovered = sessionFixture(); + const db = receiptDb(saved, nativeProviderPackAdmissionPending(randomUUID())); + await expect(rememberRecoveredNativeProviderPack(db, recovered.session, + { identity: initial, sourceRunId: initial.runId, ...binding, source: { runId: "foreign-run", receipt: saved, state: verified } })).resolves.toBe(false); + await expect(rememberRecoveredNativeProviderPack(db, recovered.session, + { identity: initial, sourceRunId: initial.runId, ...binding, source: { runId: initial.runId, receipt: saved, state: verified } })).resolves.toBe(true); + }); +}); + +describe("provider pack admission PostgreSQL fencing", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("pack-admission-"); db = createDb(database.connectionString); }, 90_000); + afterAll(async () => { await database?.cleanup(); }); + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Pack admission", issuePrefix: randomUUID().slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Pack agent" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Pack task", assigneeAgentId: agentId }); + const identity = { companyId, agentId, issueId, runId, sessionId: `session-${runId}` }; + const admissionEpoch = randomUUID(); + const profile = { preserved: { value: "unchanged" }, nativeExecutionInput: { binding: identity }, + nativeProviderPackAdmissionState: nativeProviderPackAdmissionPending(admissionEpoch) }; + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, runtimeMode: "native", status: "running", runnerProfileJson: profile }); + const claim = { leaseOwner: randomUUID(), attempt: 1, controllerGeneration: 2, controllerBootId: randomUUID() }; + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed", ...claim, + leaseExpiresAt: new Date(Date.now() + 60_000) }); + const { session } = sessionFixture(identity); remember(session, identity); + return { identity, session, claim, profile, admissionEpoch }; + } + const read = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]!; + it("merges the exact closed receipt without changing other profile data and is idempotent", async () => { + const f = await fixture(); + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(true); + const saved = await read(f.identity.runId); + expect(saved.runnerProfileJson).toEqual({ ...f.profile, nativeProviderPackAdmission: admittedNativeProviderPack(f.session, f.identity), + nativeProviderPackAdmissionState: { ...f.profile.nativeProviderPackAdmissionState, state: "verified" } }); + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(true); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual(saved.runnerProfileJson); + }); + it.each(["leaseOwner", "attempt", "controllerGeneration", "controllerBootId"] as const)("rejects stale %s without publishing a receipt", async key => { + const f = await fixture(); + const changed = typeof f.claim[key] === "number" ? 99 : randomUUID(); + await expect(recordNativeProviderPackAdmission(db, { ...f, claim: { ...f.claim, [key]: changed } as typeof f.claim })).resolves.toBe(false); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual(f.profile); + }); + it("rejects an expired claim and a terminal run", async () => { + const f = await fixture(); + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.identity.runId)); + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(false); + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.identity.runId)); + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, f.identity.runId)); + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(false); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual(f.profile); + }); + it("preserves the original immutable receipt when another pack is proposed", async () => { + const f = await fixture(); await recordNativeProviderPackAdmission(db, f); + const before = (await read(f.identity.runId)).runnerProfileJson!; + const other = sessionFixture(f.identity); remember(other.session, f.identity, manifest("b".repeat(40))); + await expect(recordNativeProviderPackAdmission(db, { ...f, session: other.session })).resolves.toBe(false); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual({ ...before, + nativeProviderPackAdmissionState: { ...f.profile.nativeProviderPackAdmissionState, state: "unknown" } }); + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(false); + }); + it("preserves the original inheritance when the same run recovers the same verified pack", async () => { + const f = await fixture(), sourceRunId = randomUUID(); + const receipt = admittedNativeProviderPack(f.session, f.identity)!; + await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId: f.identity.companyId, agentId: f.identity.agentId, + nativeIssueId: f.identity.issueId, runtimeMode: "native", status: "succeeded", + runnerProfileJson: { nativeProviderPackAdmission: { ...receipt, runId: sourceRunId } } }); + const original = { ...f.profile, nativeProviderPackAdmission: { ...receipt, inheritedFromRunId: sourceRunId } }; + await db.update(heartbeatRuns).set({ runnerProfileJson: original }).where(eq(heartbeatRuns.id, f.identity.runId)); + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(true); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual({ ...original, + nativeProviderPackAdmissionState: { ...f.profile.nativeProviderPackAdmissionState, state: "verified" } }); + }); + it("leaves a newer pending epoch untouched by an older same-claim callback", async () => { + const f = await fixture(); await recordNativeProviderPackAdmission(db, f); + const previous = (await read(f.identity.runId)).runnerProfileJson!; + const next = { ...previous, nativeProviderPackAdmissionState: nativeProviderPackAdmissionPending(randomUUID()) }; + await db.update(heartbeatRuns).set({ runnerProfileJson: next }).where(eq(heartbeatRuns.id, f.identity.runId)); + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(false); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual(next); + }); + it.each(["runner", "lease", "unsupported"])("keeps the first receipt historical after an authorized %s replacement", async change => { + const f = await fixture(); await recordNativeProviderPackAdmission(db, f); + const previous = (await read(f.identity.runId)).runnerProfileJson!; + const admissionEpoch = randomUUID(); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...previous, + nativeProviderPackAdmissionState: nativeProviderPackAdmissionPending(admissionEpoch) } }).where(eq(heartbeatRuns.id, f.identity.runId)); + const replacement = sessionFixture(f.identity); + if (change !== "unsupported") rememberAdmittedNativeProviderPack(replacement.session, { identity: f.identity, manifest: manifest(), + ...binding, ...(change === "runner" ? { runnerInstanceId: "replacement" } : { environmentLeaseId: "replacement" }) }); + await expect(recordNativeProviderPackAdmission(db, { ...f, admissionEpoch, session: replacement.session })).resolves.toBe(false); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual({ ...previous, + nativeProviderPackAdmissionState: { ...nativeProviderPackAdmissionPending(admissionEpoch), state: "unknown" } }); + }); + it("cannot publish while another transaction holds the coordinator claim", async () => { + const f = await fixture(); + await recordNativeProviderPackAdmission(db, f); + const previous = (await read(f.identity.runId)).runnerProfileJson!; + f.admissionEpoch = randomUUID(); + const pending = { ...previous, nativeProviderPackAdmissionState: nativeProviderPackAdmissionPending(f.admissionEpoch) }; + await db.update(heartbeatRuns).set({ runnerProfileJson: pending }).where(eq(heartbeatRuns.id, f.identity.runId)); + let release!: () => void, locked!: () => void; + const acquired = new Promise(resolve => { locked = resolve; }); + const hold = new Promise(resolve => { release = resolve; }); + const owner = db.transaction(async tx => { + await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.identity.runId)).for("update"); + locked(); await hold; + }); + try { + await acquired; + await expect(recordNativeProviderPackAdmission(db, f)).rejects.toThrow(); + expect((await read(f.identity.runId)).runnerProfileJson).toEqual(pending); + } finally { release(); await owner; } + await expect(recordNativeProviderPackAdmission(db, f)).resolves.toBe(true); + }); +}); diff --git a/server/src/services/native-runtime/native-provider-pack-admission.ts b/server/src/services/native-runtime/native-provider-pack-admission.ts new file mode 100644 index 00000000000..b50a57bbfef --- /dev/null +++ b/server/src/services/native-runtime/native-provider-pack-admission.ts @@ -0,0 +1,167 @@ +import { and, eq, gt, isNull, sql } from "drizzle-orm"; +import { createHash } from "node:crypto"; +import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import type { NativeSession } from "../../vendor/paperclip-runner/index.js"; +import { canonicalProviderPackJson, type ProviderPackIdentity } from "./remote-provider-pack-verification.js"; + +type Selection = { + admittedRunId: string; + companyId: string; + agentId: string; + sessionId: string; + runnerInstanceId: string; + environmentLeaseId: string; + providerPackDigest: string; + opencodeProxySha256: string; + sourceRevision: string; +}; + +export type NativeProviderPackAdmission = Omit & { + schema: "paperclip.native-provider-pack-admission.v1"; + stage: "session_admitted"; + authority: "core_verified_selection"; + runId: string; + runnerEnvironmentLeaseId: string; + inheritedFromRunId?: string; +}; + +const admitted = new WeakMap>(); +const digest = /^sha256:[a-f0-9]{64}$/; +type NativeRunIdentity = ReturnType; + +export type NativeProviderPackAdmissionSource = { runId: string; receipt: unknown; state: unknown }; + +export function nativeProviderPackAdmissionPending(admissionEpoch: string) { + return { schema: "paperclip.native-provider-pack-admission-state.v1" as const, state: "pending" as const, admissionEpoch }; +} + +function verifiedState(value: unknown): boolean { + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const state = value as Record; + return Object.keys(state).sort().join(",") === "admissionEpoch,schema,state" + && state.schema === "paperclip.native-provider-pack-admission-state.v1" && state.state === "verified" + && typeof state.admissionEpoch === "string" && /^[a-f0-9-]{36}$/.test(state.admissionEpoch); +} + +function rememberSelection(session: NativeSession, selected: Selection): boolean { + const prior = admitted.get(session); + if (prior && canonicalProviderPackJson(prior) === canonicalProviderPackJson({ ...selected, admittedRunId: prior.admittedRunId })) return true; + admitted.set(session, Object.freeze(selected)); + return true; +} + +export function rememberAdmittedNativeProviderPack(session: NativeSession, input: { + identity: NativeRunIdentity; + runnerInstanceId: string; + environmentLeaseId: string; + manifest: ProviderPackIdentity; +}): boolean { + const unavailable = () => { admitted.delete(session); return false; }; + if (typeof session.identity !== "function") return unavailable(); + const actual = session.identity(); + if (canonicalProviderPackJson(actual) !== canonicalProviderPackJson(input.identity)) return unavailable(); + const proxy = input.manifest.payload.artifacts.opencodeProxy?.sha256; + const revision = input.manifest.payload.runnerSourceRevision; + if (!digest.test(input.manifest.digest) + || input.manifest.digest !== `sha256:${createHash("sha256").update(canonicalProviderPackJson(input.manifest.payload)).digest("hex")}` + || !digest.test(proxy ?? "") || !/^[a-f0-9]{40}$/.test(revision) + || !input.runnerInstanceId || !input.environmentLeaseId) return unavailable(); + const selected: Selection = { + admittedRunId: actual.runId, + companyId: actual.companyId, agentId: actual.agentId, sessionId: actual.sessionId, + runnerInstanceId: input.runnerInstanceId, environmentLeaseId: input.environmentLeaseId, + providerPackDigest: input.manifest.digest, opencodeProxySha256: proxy!, sourceRevision: revision, + }; + return rememberSelection(session, selected); +} + +export async function rememberRecoveredNativeProviderPack(db: Db, session: NativeSession, input: { + identity: NativeRunIdentity; + sourceRunId: string; + runnerInstanceId: string; + environmentLeaseId: string; + source?: NativeProviderPackAdmissionSource; +}): Promise { + admitted.delete(session); + if (typeof session.identity !== "function") return false; + if (canonicalProviderPackJson(session.identity()) !== canonicalProviderPackJson(input.identity)) return false; + const source = input.source?.runId === input.sourceRunId ? input.source : (await db.select({ + receipt: sql`${heartbeatRuns.runnerProfileJson}->'nativeProviderPackAdmission'`, + state: sql`${heartbeatRuns.runnerProfileJson}->'nativeProviderPackAdmissionState'`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.sourceRunId), eq(heartbeatRuns.companyId, input.identity.companyId), + eq(heartbeatRuns.agentId, input.identity.agentId), eq(heartbeatRuns.runtimeMode, "native"))).limit(1))[0]; + if (!verifiedState(source?.state)) return false; + const value = source?.receipt; + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const receipt = value as Record; + const keys = new Set(["schema", "stage", "authority", "companyId", "agentId", "runId", "sessionId", "runnerInstanceId", + "runnerEnvironmentLeaseId", "providerPackDigest", "opencodeProxySha256", "sourceRevision", "inheritedFromRunId"]); + if (Object.keys(receipt).some(key => !keys.has(key)) || receipt.schema !== "paperclip.native-provider-pack-admission.v1" + || receipt.stage !== "session_admitted" || receipt.authority !== "core_verified_selection" + || receipt.runId !== input.sourceRunId || receipt.companyId !== input.identity.companyId || receipt.agentId !== input.identity.agentId + || receipt.sessionId !== input.identity.sessionId || receipt.runnerInstanceId !== input.runnerInstanceId + || receipt.runnerEnvironmentLeaseId !== input.environmentLeaseId || !digest.test(String(receipt.providerPackDigest)) + || !digest.test(String(receipt.opencodeProxySha256)) || !/^[a-f0-9]{40}$/.test(String(receipt.sourceRevision)) + || (receipt.inheritedFromRunId !== undefined && (typeof receipt.inheritedFromRunId !== "string" || !receipt.inheritedFromRunId))) return false; + return rememberSelection(session, { admittedRunId: String(receipt.inheritedFromRunId ?? receipt.runId), + companyId: input.identity.companyId, agentId: input.identity.agentId, sessionId: input.identity.sessionId, + runnerInstanceId: input.runnerInstanceId, environmentLeaseId: input.environmentLeaseId, + providerPackDigest: String(receipt.providerPackDigest), opencodeProxySha256: String(receipt.opencodeProxySha256), + sourceRevision: String(receipt.sourceRevision) }); +} + +export function admittedNativeProviderPack(session: NativeSession, identity: NativeRunIdentity): NativeProviderPackAdmission | null { + const selected = admitted.get(session); + if (!selected) return null; + const actual = session.identity(); + if (canonicalProviderPackJson(actual) !== canonicalProviderPackJson(identity) + || actual.companyId !== selected.companyId || actual.agentId !== selected.agentId || actual.sessionId !== selected.sessionId) return null; + const { admittedRunId, environmentLeaseId, ...identityFields } = selected; + return { schema: "paperclip.native-provider-pack-admission.v1", stage: "session_admitted", + authority: "core_verified_selection", ...identityFields, runId: actual.runId, runnerEnvironmentLeaseId: environmentLeaseId, + ...(admittedRunId === actual.runId ? {} : { inheritedFromRunId: admittedRunId }) }; +} + +export async function recordNativeProviderPackAdmission(db: Db, input: { + session: NativeSession; + identity: NativeRunIdentity; + admissionEpoch: string; + claim: { leaseOwner: string; attempt: number; controllerGeneration: number; controllerBootId: string }; +}): Promise { + const receipt = admittedNativeProviderPack(input.session, input.identity); + return db.transaction(async tx => { + const [owner] = await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, input.identity.runId), eq(nativeRunFinalizations.companyId, input.identity.companyId), + eq(nativeRunFinalizations.issueId, input.identity.issueId), eq(nativeRunFinalizations.leaseOwner, input.claim.leaseOwner), + eq(nativeRunFinalizations.attempt, input.claim.attempt), eq(nativeRunFinalizations.controllerGeneration, input.claim.controllerGeneration), + eq(nativeRunFinalizations.controllerBootId, input.claim.controllerBootId), gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), + eq(nativeRunFinalizations.phase, "observed"), isNull(nativeRunFinalizations.resultId), + )).for("update", { noWait: true }).limit(1); + if (!owner) return false; + const [run] = await tx.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.identity.runId), eq(heartbeatRuns.companyId, input.identity.companyId), + eq(heartbeatRuns.agentId, input.identity.agentId), eq(heartbeatRuns.nativeIssueId, input.identity.issueId), + eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "running"), + )).for("update", { noWait: true }).limit(1); + if (!run) return false; + const current = run.profile?.nativeProviderPackAdmissionState as Record | undefined; + if (current?.schema !== "paperclip.native-provider-pack-admission-state.v1" || current.admissionEpoch !== input.admissionEpoch + || !["pending", "verified", "unknown"].includes(String(current.state))) return false; + if (current.state === "unknown") return false; + const previous = run.profile?.nativeProviderPackAdmission; + const verified = receipt != null && (previous == null ? current.state === "pending" + : canonicalProviderPackJson({ ...(previous as Record), inheritedFromRunId: null }) + === canonicalProviderPackJson({ ...receipt, inheritedFromRunId: null })); + if (verified && current.state === "verified") return true; + const patch = { ...(verified && previous == null ? { nativeProviderPackAdmission: receipt } : {}), + nativeProviderPackAdmissionState: { ...nativeProviderPackAdmissionPending(input.admissionEpoch), state: verified ? "verified" : "unknown" } }; + const saved = await tx.update(heartbeatRuns).set({ + runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify(patch)}::jsonb`, + updatedAt: new Date(), + }).where(and(eq(heartbeatRuns.id, input.identity.runId), eq(heartbeatRuns.companyId, input.identity.companyId), + eq(heartbeatRuns.agentId, input.identity.agentId), eq(heartbeatRuns.nativeIssueId, input.identity.issueId), + eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "running"), + )).returning({ id: heartbeatRuns.id }); + return verified && saved.length === 1; + }); +} diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 59d7a94e408..f8684817dd9 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -100,6 +100,9 @@ import { NativeRunnerOwnershipUnverifiedError } from "./native-runner-ownership. import type { AdapterRuntimeEvent } from "../../adapters/index.js"; import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; import { acknowledgedNativeStopExecutionHasStopped, isAcknowledgedNativeStop } from "../acknowledged-native-stop.js"; +import { admittedNativeProviderPack, nativeProviderPackAdmissionPending, rememberAdmittedNativeProviderPack } from "./native-provider-pack-admission.js"; +import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js"; +import * as providerPackAdmission from "./native-provider-pack-admission.js"; vi.mock("node:fs", async (importOriginal) => { const actual = await importOriginal(); @@ -1700,7 +1703,7 @@ describe("remote provider pack manifest", () => { return JSON.stringify(value); }; - it.each(["opencode", "opencode-diagnostics", "opencode-diagnostics-stale", "cursor", "claude-x64", "claude-arm64"] as const)("uses the bundled Linux image authority for remote %s without controller overrides", async provider => { + it.each(["opencode", "opencode-dirty", "opencode-diagnostics", "opencode-diagnostics-stale", "cursor", "claude-x64", "claude-arm64"] as const)("uses the bundled Linux image authority for remote %s without controller overrides", async provider => { const architecture = provider === "claude-arm64" ? "arm64" : "x64"; const claude = provider.startsWith("claude-"); const opencode = provider.startsWith("opencode"); @@ -1746,7 +1749,7 @@ describe("remote provider pack manifest", () => { const syncIn = vi.fn(async (_operations: { files: { kind: string }[] }[]) => undefined); const installed = structuredClone(manifest); if (opencode) { - installed.payload.runnerSourceRevision = "c".repeat(40); + installed.payload.runnerSourceRevision = "c".repeat(40) + (provider === "opencode-dirty" ? "-dirty" : ""); installed.payload.artifacts.nodeCommand.sha256 = "sha256:" + "d".repeat(64); if (provider !== "opencode-diagnostics") installed.payload.artifacts.opencodeProxy.sha256 = "sha256:" + "e".repeat(64); installed.digest = "sha256:" + createHash("sha256").update(canonical(installed.payload)).digest("hex"); @@ -1761,6 +1764,7 @@ describe("remote provider pack manifest", () => { }) : "", })); try { + let coldAdmission: ReturnType = null; const prepare = createRunnerdBackend({ db: leaseDb(remoteExecution), execution: remoteExecution, runnerInstanceId: "bundled-image-runner", runnerIngressAuthorized: true, ...(opencode ? { runnerEnvironment: { ...configuredEnvironmentProjection(diagnostics ? { PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS: "1" } : {}), PAPERCLIP_NATIVE_MCP_NAME: "assigned-fixture", PAPERCLIP_NATIVE_MCP_URL: "http://127.0.0.1/mcp/gateways/assigned-fixture", PAPERCLIP_NATIVE_MCP_TOKEN: "private-fixture" } } : {}), @@ -1772,7 +1776,7 @@ describe("remote provider pack manifest", () => { expect(state.createBackend).not.toHaveBeenCalled(); return; } - await prepare; + const managedBackend = await prepare; state.createTransport.mockClear(); state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); const transport = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & { @@ -1798,6 +1802,28 @@ describe("remote provider pack manifest", () => { expect(transport).toMatchObject({ providerPackAuthorityDigest: installed.digest, providerNodeCommand: "/opt/paperclip-runner/provider-pack/node_modules/node/bin/node", providerNodeCommandSha256: installed.payload.artifacts.nodeCommand.sha256 }); + if (provider === "opencode" || provider === "opencode-dirty") { + const identity = { ...remoteExecution.binding, sessionId: remoteExecution.session.normalizedSessionId! }; + const session = { identity: () => identity } as never; + const backend = state.createBackend.mock.results.at(-1)!.value; + const openSession = vi.fn().mockRejectedValueOnce(new Error("provider open failed")).mockResolvedValue(session); + Object.assign(backend, { openSession }); + expect(admittedNativeProviderPack(session, identity)).toBeNull(); + await expect(managedBackend.openSession({ identity })).rejects.toThrow("provider open failed"); + expect(admittedNativeProviderPack(session, identity)).toBeNull(); + const opened = await managedBackend.openSession({ identity }); + if (provider === "opencode-dirty") { + expect(admittedNativeProviderPack(opened, identity)).toBeNull(); + expect(existsSync(join(transport.stateDirectory!, "provider-pack-authority.json"))).toBe(true); + } else expect(admittedNativeProviderPack(opened, identity)).toMatchObject({ + schema: "paperclip.native-provider-pack-admission.v1", stage: "session_admitted", authority: "core_verified_selection", + companyId: identity.companyId, agentId: identity.agentId, runId: identity.runId, sessionId: identity.sessionId, + runnerInstanceId: transport.prpIdentity!.runnerInstanceId, runnerEnvironmentLeaseId: transport.prpIdentity!.environmentLeaseId, + providerPackDigest: installed.digest, opencodeProxySha256: installed.payload.artifacts.opencodeProxy.sha256, + sourceRevision: installed.payload.runnerSourceRevision, + }); + coldAdmission = admittedNativeProviderPack(opened, identity); + } expect(execute.mock.calls.filter(([call]) => call.args?.[0] === "--build-metadata")).toHaveLength(1); expect(execute.mock.calls.some(([call]) => call.args?.[0] === "-e" || call.args?.[0] === "--version")).toBe(false); expect(syncIn.mock.calls.every(([operations]) => operations.every(operation => operation.files.every(file => file.kind === "file")))).toBe(true); @@ -1831,7 +1857,7 @@ describe("remote provider pack manifest", () => { : script.includes("cat --") ? `nonce\n123\n2026-09-19T10:00:00.000Z\n${identity.runnerInstanceId}\nlinux:abcd-1234:100\n` : ""; return { stdout, stderr: "", exitCode: 0, timedOut: false }; }); - await createRunnerdBackend({ db: leaseDb(remoteExecution), execution: remoteExecution, + const recoveryInput: Parameters[0] = { db: leaseDb(remoteExecution), execution: remoteExecution, runnerInstanceId: identity.runnerInstanceId, runnerIngressAuthorized: true, ...(opencode ? { runnerEnvironment: configuredEnvironmentProjection({ PAPERCLIP_OPENCODE_PERFORMANCE_DIAGNOSTICS: "1" }) } : {}), runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", @@ -1840,10 +1866,58 @@ describe("remote provider pack manifest", () => { restartRecovery: { kind: "reattach_remote_runner", runId: remoteExecution.binding.runId, leaseOwner: "controller", controllerGeneration: 2, providerAttempt: 1, restartKind: "graceful", recoveryRequestId: null, remote: { providerLeaseId: "sandbox", remoteCwd: "/workspace" } }, - }); + }; + const recoveredBackend = await createRunnerdBackend(recoveryInput); state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); const recovered = state.createTransport.mock.calls.at(-1)![0]; expect(recovered).toMatchObject({ providerPackAuthorityDigest: admitted.digest, runnerArtifact: artifact, adoptExistingRunner: { pid: 123 } }); + if (provider === "opencode") { + const nativeIdentity = { ...remoteExecution.binding, sessionId: remoteExecution.session.normalizedSessionId! }; + const restored = { identity: () => nativeIdentity } as never; + const checkpoint = { backendKind: "runner", identity: nativeIdentity, sessionId: nativeIdentity.sessionId } as never; + Object.assign(state.createBackend.mock.results.at(-1)!.value, { + recoverSession: async () => ({ recovered: true, session: restored }), + }); + const result = await recoveredBackend.recoverSession!(checkpoint, { signal: new AbortController().signal }); + expect(result.recovered).toBe(true); + expect(admittedNativeProviderPack(result.session!, nativeIdentity)).toBeNull(); + const legacyBackend = await createRunnerdBackend({ ...recoveryInput, + controllerPersistence: { bootId: "replacement-controller", runnerArtifact: artifact } }); + const legacy = { identity: () => nativeIdentity } as never; + Object.assign(state.createBackend.mock.results.at(-1)!.value, { + recoverSession: async () => ({ recovered: true, session: legacy }), + }); + const legacyResult = await legacyBackend.recoverSession!(checkpoint, { signal: new AbortController().signal }); + expect(legacyResult.recovered).toBe(true); + expect(admittedNativeProviderPack(legacyResult.session!, nativeIdentity)).toBeNull(); + const withPrior = new Proxy(recoveryInput.db, { + get(target, key, receiver) { + if (key !== "select") return Reflect.get(target, key, receiver); + return (fields?: Record) => fields && "receipt" in fields + ? { from: () => ({ where: () => ({ limit: async () => [{ receipt: coldAdmission, + state: { ...nativeProviderPackAdmissionPending(randomUUID()), state: "verified" } }] }) }) } + : target.select(fields as never); + }, + }); + for (const pinned of [false, true]) { + const knownBackend = await createRunnerdBackend({ ...recoveryInput, db: withPrior, + ...(pinned ? { controllerPersistence: { bootId: "replacement-controller", runnerArtifact: artifact }, runnerRemoteBinaryPath: state.bundledRunnerBinary! } : {}) }); + const known = { identity: () => nativeIdentity } as never; + Object.assign(state.createBackend.mock.results.at(-1)!.value, { + recoverSession: async () => ({ recovered: true, session: known }), + }); + const knownResult = await knownBackend.recoverSession!(checkpoint, { signal: new AbortController().signal }); + expect(knownResult.recovered).toBe(true); + expect(admittedNativeProviderPack(knownResult.session!, nativeIdentity)).toEqual(coldAdmission); + if (pinned) { + const replacement = { identity: () => nativeIdentity } as never; + Object.assign(state.createBackend.mock.results.at(-1)!.value, { openSession: async () => replacement }); + const replaced = await knownBackend.openReplacementSession!({ identity: nativeIdentity }, checkpoint); + expect(replaced).toBe(replacement); + expect(admittedNativeProviderPack(replaced, nativeIdentity)).toBeNull(); + } + } + } } } finally { state.bundledPackManifestPath = null; state.bundledRunnerBinary = null; @@ -8060,6 +8134,105 @@ describe("native session cancellation", () => { }); describe("native session execution lease fencing", () => { + it("persists current admission from the executor callback and never writes for null publication", async () => { + const identity = { runId: execution.binding.runId, companyId: execution.binding.companyId, agentId: execution.binding.agentId, + issueId: execution.binding.issueId, sessionId: execution.session.normalizedSessionId! }; + const session = { identity: () => identity } as never; + const artifact = { path: "/fixture/provider", sha256: "sha256:" + "a".repeat(64) }; + const payload = { runnerSourceRevision: "b".repeat(40), target: { platform: "linux", architecture: "x64" }, + pins: { opencode: "1.18.34" }, artifacts: { nodeCommand: artifact, opencodeCommand: artifact, opencodeProxy: artifact } }; + expect(rememberAdmittedNativeProviderPack(session, { identity, runnerInstanceId: "actual-runner", environmentLeaseId: "runner-scope", + manifest: { schema: "paperclip-runner/remote-provider-pack/v1", payload, + digest: "sha256:" + createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex") } })).toBe(true); + const profile: Record = { preserved: { value: "unchanged" } }; + const updates: Array<{ table: unknown; values: Record }> = []; + const recordAdmission = vi.spyOn(providerPackAdmission, "recordNativeProviderPackAdmission"); + let profileWrites = 0; + const wrap = (source: Db): Db => new Proxy(source, { get(target, key, receiver) { + if (key === "transaction") return (operation: (tx: Db) => Promise) => target.transaction(tx => operation(wrap(tx as unknown as Db)) as never); + if (key === "select") return (fields?: Record) => { + if (!fields || !("profile" in fields)) return target.select(fields as never); + const query = { where: () => query, for: () => query, limit: async () => [{ profile }] }; + return { from: () => query }; + }; + if (key === "update") return (table: unknown) => ({ set: (values: Record) => { + const update = target.update(table as never).set(values); + return { where: (condition: unknown) => { + if (table === heartbeatRuns && values.runnerProfileJson) { + const query = new PgDialect().sqlToQuery(values.runnerProfileJson as SQL); + const patch = query.params.map(value => typeof value === "string" ? (() => { try { return JSON.parse(value); } catch { return null; } })() : null) + .find(value => value?.nativeProviderPackAdmissionState); + if (patch) { Object.assign(profile, patch); profileWrites++; } + } + return update.where(condition as never); + } }; + } }); + return Reflect.get(target, key, receiver); + } }); + const db = wrap(leaseDb(execution, {}, {}, updates, profile)); + let nullWrites = -1; + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession(session); + const before = profileWrites; + await options.onSession(null); + nullWrites = profileWrites - before; + return { result: { summary: "cancelled" }, terminal: { runTerminalState: "cancelled" }, turnId: "turn", + normalizedSessionId: identity.sessionId, providerSessionId: null, driverKind: "test", driverVersion: "1", + nativeEventCount: 0, highestContiguousSourceSeq: 0, usage: null }; + }); + await executePaperclipNativeSession({ db, execution, runnerInstanceId: "actual-runner" }); + expect(profile.preserved).toEqual({ value: "unchanged" }); + expect(profile.nativeProviderPackAdmission).toEqual(admittedNativeProviderPack(session, identity)); + expect(profile.nativeProviderPackAdmissionState).toMatchObject({ state: "verified", admissionEpoch: expect.any(String) }); + expect(profileWrites).toBe(2); + expect(nullWrites).toBe(0); + const claim = updates.find(update => update.table === nativeRunFinalizations && update.values.phase === "observed" && update.values.leaseOwner)!.values; + expect(recordAdmission).toHaveBeenCalledExactlyOnceWith(db, { session, identity, + admissionEpoch: (profile.nativeProviderPackAdmissionState as Record).admissionEpoch, + claim: { leaseOwner: claim.leaseOwner, attempt: claim.attempt, controllerGeneration: claim.controllerGeneration, controllerBootId: claim.controllerBootId } }); + }); + it("atomically invalidates previous admission proof before provider admission", async () => { + const updates: Array<{ table: unknown; values: Record }> = []; + let observed: Record | null = null; + const profile = { nativeProviderPackAdmission: { historical: "preserved" }, + nativeProviderPackAdmissionState: { ...nativeProviderPackAdmissionPending(randomUUID()), state: "verified" } }; + state.execute.mockReset().mockImplementationOnce(async () => { + const invalidation = updates.find(update => update.table === heartbeatRuns && update.values.nativePhase === "observed")!; + if (invalidation.values.runnerProfileJson) { + const query = new PgDialect().sqlToQuery(invalidation.values.runnerProfileJson as SQL); + observed = query.params.map(value => typeof value === "string" ? (() => { try { return JSON.parse(value); } catch { return null; } })() : null) + .find(value => value?.nativeProviderPackAdmissionState); + } + return { result: { summary: "cancelled" }, terminal: { runTerminalState: "cancelled" }, turnId: "turn", + normalizedSessionId: "session", providerSessionId: null, driverKind: "test", driverVersion: "1", + nativeEventCount: 0, highestContiguousSourceSeq: 0, usage: null }; + }); + await executePaperclipNativeSession({ db: leaseDb(execution, {}, {}, updates, profile), execution, runnerInstanceId: "runner" }); + expect(state.execute).toHaveBeenCalledOnce(); + expect(observed).toEqual({ nativeProviderPackAdmissionState: { schema: "paperclip.native-provider-pack-admission-state.v1", + state: "pending", admissionEpoch: expect.any(String) } }); + expect(JSON.stringify(observed)).not.toContain(profile.nativeProviderPackAdmissionState.admissionEpoch); + expect(profile.nativeProviderPackAdmissionState.state).toBe("verified"); + }); + it("does not admit a session if mandatory claim invalidation fails", async () => { + const source = leaseDb(execution); + const failed = new Error("claim profile write failed"); + const db = new Proxy(source, { get(target, key, receiver) { + if (key !== "transaction") return Reflect.get(target, key, receiver); + return (operation: (tx: Db) => Promise) => target.transaction(tx => operation(new Proxy(tx, { get(transaction, name, innerReceiver) { + if (name !== "update") return Reflect.get(transaction, name, innerReceiver); + return (table: unknown) => ({ set: (values: Record) => { + if (table === heartbeatRuns && values.nativePhase === "observed" && values.runnerProfileJson) throw failed; + return transaction.update(table as never).set(values); + } }); + } }) as unknown as Db) as never); + } }); + state.execute.mockClear(); state.createBackend.mockClear(); state.createTransport.mockClear(); + await expect(executePaperclipNativeSession({ db, execution, runnerInstanceId: "runner" })).rejects.toBe(failed); + expect(state.execute).not.toHaveBeenCalled(); + expect(state.createBackend).not.toHaveBeenCalled(); + expect(state.createTransport).not.toHaveBeenCalled(); + }); it("renews only when the exact fenced owner remains current", async () => { const returning = vi .fn() diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 9cec1870142..c78016c1ad6 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -16,6 +16,7 @@ import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export- import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { providerPackWorkspaceId, recordAdmittedRemoteProviderPack, resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js"; +import { nativeProviderPackAdmissionPending, rememberAdmittedNativeProviderPack, rememberRecoveredNativeProviderPack, recordNativeProviderPackAdmission, type NativeProviderPackAdmissionSource } from "./native-provider-pack-admission.js"; import { type ProviderPackInstallation, verifyRemoteProviderPack as verifyRemoteProviderPackIdentity } from "./remote-provider-pack-verification.js"; import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { NATIVE_SESSION_EXECUTION_LEASE_TTL_MS, startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js"; @@ -8437,6 +8438,8 @@ async function executePaperclipNativeSessionWithinScope( ? REMOTE_CONTROLLER_LEASE_MS : NATIVE_SESSION_EXECUTION_LEASE_TTL_MS; let attempt: number; let controllerGeneration = 0; + const providerPackAdmissionEpoch = randomUUID(); + let providerPackAdmissionSource: NativeProviderPackAdmissionSource | undefined; try { attempt = await trace.measure( "native.coordinator.claim", @@ -8478,6 +8481,7 @@ async function executePaperclipNativeSessionWithinScope( resultJson: heartbeatRuns.resultJson, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, + runnerProfileJson: heartbeatRuns.runnerProfileJson, }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.execution.binding.runId)) @@ -8617,12 +8621,16 @@ async function executePaperclipNativeSessionWithinScope( .returning({ runId: nativeRunFinalizations.runId }) .then((rows) => rows[0] ?? null); if (!claimed) throw new Error("native_session_lease_lost"); + providerPackAdmissionSource = { runId: coordinator.runId, + receipt: boundRun.runnerProfileJson?.nativeProviderPackAdmission, + state: boundRun.runnerProfileJson?.nativeProviderPackAdmissionState }; controllerGeneration = nextControllerGeneration; await tx .update(heartbeatRuns) .set({ nativePhase: "observed", nativePhaseUpdatedAt: leaseNow, + runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ nativeProviderPackAdmissionState: nativeProviderPackAdmissionPending(providerPackAdmissionEpoch) })}::jsonb`, updatedAt: leaseNow, }) .where(eq(heartbeatRuns.id, coordinator.runId)); @@ -9412,6 +9420,7 @@ async function executePaperclipNativeSessionWithinScope( input.useRunnerd && input.backend === undefined ? await timeController("backend_setup", () => createRunnerdBackend({ ...input, + providerPackAdmissionSource, onAccountingEvent: producerUsage?.capture, controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? { bootId: controller.bootId, @@ -9459,7 +9468,7 @@ async function executePaperclipNativeSessionWithinScope( persistWarmNativeCheckpoint(remoteWarmRecovery.previousExecution, nativeSessionConfigDigest(remoteWarmRecovery.previousExecution, "remote"), await recovered.session.snapshot()); persistedWarmSession = loadWarmNativeCheckpoint(input.execution, warmConfigDigest!, "remote"); - runnerdBackend = await createRunnerdBackend({ ...input, onAccountingEvent: producerUsage?.capture, runnerInstanceId: effectiveRunnerInstanceId, + runnerdBackend = await createRunnerdBackend({ ...input, providerPackAdmissionSource, onAccountingEvent: producerUsage?.capture, runnerInstanceId: effectiveRunnerInstanceId, controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? { bootId: controller.bootId } : undefined, durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, trace, toolTrace }); } @@ -9784,6 +9793,18 @@ async function executePaperclipNativeSessionWithinScope( if (session.detachControllerForRestart) await detachActiveNativeSessionForRestart(active); throw new NativeControllerDetachedForRestartError(); } + try { + await recordNativeProviderPackAdmission(input.db, { + session, + admissionEpoch: providerPackAdmissionEpoch, + identity: { runId: input.execution.binding.runId, companyId: input.execution.binding.companyId, + agentId: input.execution.binding.agentId, issueId: input.execution.binding.issueId, + sessionId: nativeSessionKey(input.execution) }, + claim: { leaseOwner, attempt, controllerGeneration, controllerBootId: controller.bootId }, + }); + } catch { + try { await input.onLog?.("stderr", "[paperclip-runner] native provider-pack admission receipt could not be saved\n"); } catch {} + } if (session.resolveRuntimeRequest) await liveQuestions.attach(); if (active.cancelRequested) { await active.cancellationSettled; @@ -12136,6 +12157,7 @@ export function remoteCheckpointIncompleteFailure( export async function createRunnerdBackend(input: { casesEnabled?: boolean; db: Db; + providerPackAdmissionSource?: NativeProviderPackAdmissionSource; controllerPersistence?: { bootId: string; runnerArtifact?: { version: string; digest: string } }; onAccountingEvent?: (event: PrpEvent) => Promise; execution: NativeExecutionInput; @@ -14382,7 +14404,22 @@ async function createRunnerdBackendWithinSessionClaim( priorAuthorityEpoch.revoke(); } sessionToolAuthorityEpochs.set(sessionScopeId, authorityEpoch); - const recordAdmittedProviderPack = async () => { + const recordAdmittedProviderPack = async (session: NativeSession, identity: Parameters[0]["identity"], retained: boolean) => { + if (retained && remoteTarget && requiresRemoteProviderPack) { + const sourceRunId = input.remoteWarmRecovery?.identity.runId ?? durableIdentity?.runId; + if (typeof sourceRunId === "string" && sourceRunId) { + try { + await rememberRecoveredNativeProviderPack(input.db, session, { identity, sourceRunId, + runnerInstanceId: effectiveRunnerInstanceId, environmentLeaseId: effectiveEnvironmentLeaseId, + source: input.providerPackAdmissionSource }); + } catch { + try { await input.onLog?.("stderr", "[paperclip-runner] prior provider-pack admission receipt is unavailable\n"); } catch {} + } + } + } else if (providerPackVerified && providerPackAuthority && expectedProviderPackManifest) { + rememberAdmittedNativeProviderPack(session, { identity, manifest: expectedProviderPackManifest, + runnerInstanceId: effectiveRunnerInstanceId, environmentLeaseId: effectiveEnvironmentLeaseId }); + } if (providerPackVerified && providerPackAuthority) { await recordAdmittedRemoteProviderPack(providerPackAuthority, message => input.onLog?.("stderr", message)); } @@ -14401,13 +14438,15 @@ async function createRunnerdBackendWithinSessionClaim( descriptor: () => backend.descriptor(), openSession: async (sessionInput) => { const session = await backend.openSession(sessionInput); - await recordAdmittedProviderPack(); - return wrapManagedSession(session); + const managed = wrapManagedSession(session); + await recordAdmittedProviderPack(managed, sessionInput.identity, Boolean(adoptedProcess)); + return managed; }, recoverSession: async (snapshot, options) => { const result = backend.recoverSession ? await backend.recoverSession(snapshot, options) : { recovered: false, reason: "driver does not support recovery" }; - if (result.session) await recordAdmittedProviderPack(); - return result.session ? { ...result, session: wrapManagedSession(result.session) } : result; + const session = result.session ? wrapManagedSession(result.session) : undefined; + if (result.recovered && session) await recordAdmittedProviderPack(session, snapshot.identity, Boolean(adoptedProcess)); + return session ? { ...result, session } : result; }, openReplacementSession: async (sessionInput) => { await measureNativeRunnerSpan( @@ -14417,8 +14456,9 @@ async function createRunnerdBackendWithinSessionClaim( { parentName: "native.session.execute" }, ); const session = await backend.openSession(sessionInput); - await recordAdmittedProviderPack(); - return wrapManagedSession(session); + const managed = wrapManagedSession(session); + await recordAdmittedProviderPack(managed, sessionInput.identity, false); + return managed; }, } satisfies ManagedRunnerdBackend; }