mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 06:42:06 +01:00
feat(native): record verified provider-pack session admission (#18469)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Remote native sessions can use an installed pack or a retained earlier pack. > - The current app revision does not identify the pack that a session admitted. > - Core already verifies the selected pack and binds its launch artifacts. > - Normal run records lack a closed receipt for that admitted selection. > - This pull request records that receipt after session admission and attachment. > - Operators can compare the actual Core admission identity with their expected pack. ## Linked Issues or Issue Description Refs #17749 for protected provider-pack selection and #18343 for retained controller recovery. The related OpenCode producer repair is #18456. Those changes do not publish the closed run receipt added here. A search found no existing `nativeProviderPackAdmission` PR. **What happened?** Core can admit a compatible installed provider pack with a source revision that differs from the current app. A retained session can also keep its original pack. The normal run record does not show the admitted pack digest, proxy digest, and source revision. An expected launch path or current app revision cannot fill that evidence gap. **Expected behavior** The run has a closed, server-owned receipt after actual session admission and attachment. The first receipt remains immutable history. A separate current-proof state verifies only the actual current admission epoch. Retained recovery needs positive original provenance, and missing legacy evidence stays unknown. **Steps to reproduce** 1. Select and verify an installed pack whose digest differs from the current app pack. 2. Open the native session successfully. 3. Inspect the normal run profile for a bound admission receipt. **Paperclip version or commit** The original selected base was `8bcc5a01edd5d87540401521e6f30d3df43e1848`. The branch now includes exact actual prerequisite parent `3ee4c5b6138da82d0b1f51aa010540561bf43c22`; the PR diff against it is the five receipt, executor, test and documentation files. **Deployment mode** Core remote native execution with provider-pack selection and retained session attachment. ## What Changed - Retain the verified selected pack identity only after a backend returns the matching admitted session. - Publish `runnerProfileJson.nativeProviderPackAdmission` from the session callback after attachment succeeds. - Mark the receipt with `stage: session_admitted` and `authority: core_verified_selection`. - Store bound company, agent, run, session, runner, logical `runnerEnvironmentLeaseId`, pack digest, OpenCode proxy digest, and source revision fields. Retained attachments identify `inheritedFromRunId`. The logical Runner scope is distinct from the physical environment lease. - Lock the current coordinator claim before the profile write. Check owner, attempt, generation, controller boot, lease expiry, phase, and absence of an accepted result. Merge JSON fields without replacing other profile data. - Restore retained metadata only from the exact prior closed Core receipt, positively verified prior state and matching retained binding. Capture same-run prior provenance from the locked claim row before invalidation; cross-run recovery reads only the exact source. Legacy absence remains unknown while recovery continues. - Preserve the first receipt as history. Atomically mark a new opaque admission epoch pending in the existing mandatory claim transaction. Publish current verification only for the matching claim and epoch after actual attachment; changed or unsupported admission becomes unknown without blocking authorized execution. - Preserve synchronous session publication and quarantine ordering before the best-effort receipt write. Fresh replacement selects its newly verified pack; a captured retained-process flag cannot relabel the replacement with an older pack. Preserve the existing private pack authority even when a local dirty revision is ineligible for the closed receipt. - Add focused admission, recovery, identity, PostgreSQL ownership, lock, and JSON-preservation tests. Document the receipt contract. ## Verification - RED: the actual managed-backend path verified and admitted an installed pack but had no closed metadata. Failed open published no evidence. Additional controls fail when retained recovery invents missing history or dirty-revision eligibility gates the existing private authority. - RED calibration: removing mandatory pending invalidation fails the actual pre-admission control; reusing the retained flag for fresh replacement falsely restores the old pack. Removing the epoch fence fails the real PostgreSQL stale-callback case; reintroducing a receipt-specific execution error fails both authorized Runner/logical-lease replacement controls. - GREEN: 49 focused Core tests, all 33 admission identity/transaction tests (20 pure and 13 PostgreSQL), and all 23 existing private-authority tests passed on exact integrated head `165a9096cf3e0d34d2db1e6b52960ad4ac0ad132`. No all-skipped run is counted. - Final test-only head `11ac34baf00552a5d322b5f5f1a8e491859cacec` adds the positive executor-callback SQL-writer wiring control, exact session/identity/claim/epoch assertions, profile preservation, and null-session zero-write check. Removing only the production writer call makes it fail on a missing receipt. All70 focused identity/Core tests pass with this closure; production code, the13 PostgreSQL tests and the23 private-authority tests are byte-identical to the qualified165 source (106 distinct passing controls across the combined proof). - The PostgreSQL tests used an owned loopback-only cached Postgres 17 container with the supported `PAPERCLIP_TEST_POSTGRES_ADMIN_URL` helper and aliases to current DB/shared source. The container was removed afterward. They cover immutable history, JSON preservation, stale claims and epochs, expired leases, terminal runs, changed/unsupported replacements, and a failed write retaining pending proof. - Historical pushed `f7f12fe` CI executed all 24 earlier admission tests, including nine PostgreSQL cases, in server5. Its warm-quarantine failure exposed the new hook-ordering issue fixed here. The same run also encountered a six-character UUID fixture-prefix collision and the independently repaired cached issue-link browser selector; those historical failures are not claimed as green. - Node 24.22.2 and Vitest 5.0.3. The credential-free normal QA runtime import preflight passes with networking disabled. Map contracts, token gates and `git diff --check` pass. - Local TypeScript validation still sees stale borrowed Runner/shared declaration mismatches outside the changed ranges. No diagnostics remain in the new helper or changed test/hook ranges; clean current-head CI is required. - Repository-wide typecheck, full tests and build were not completed locally. A full local executor attempt reached this host's unsupported native PostgreSQL listener and was interrupted; its result is not counted. Canonical run `41214512239`, attempt 2, is successful on exact final head `11ac34baf00552a5d322b5f5f1a8e491859cacec`. All 59 checks are terminal: 55 success, 3 skipped and 1 neutral; none failed or pending. The actual Linux suites execute all 33 admission tests (including 13 PostgreSQL cases) and all 1,022 executor tests successfully. The [original five failures and bounded unchanged-source controls](https://github.com/paperclipai/paperclip/pull/18469#issuecomment-6545593767) remain preserved. One failed-only request was made; GitHub expanded the reusable workflow call and dependents. No second request was made. Fresh exact-head Greptile review is 5/5, all review threads are resolved, and the PR is mergeable. - These local results claim no live provider admission, rollout or recovery. ## Risks - The authority is Core's verified selection plus successful session admission. It is not an independent provider statement or continuous runtime measurement. - Receipt fields cannot grant execution or replace the existing provider, process, lease, workspace, or recovery checks. - Legacy missing receipts and unavailable database reads remain unknown. The current release cannot label an older admitted pack. - The closed receipt requires a clean, 40-character hexadecimal source revision. Verified local `-dirty` packs still execute and retain their existing private recovery authority; their closed admission proof stays `unknown`, so it cannot certify an immutable source revision. - The first receipt is historical. Consumers must require current state `verified`; pending or unknown cannot certify the current pack. Changed Runner scope or pack evidence does not add execution policy. - A receipt save failure leaves the atomically invalidated state pending. The existing artifact authority, cancellation, workspace and process checks remain responsible for execution safety. - The receipt adds no manifest, environment, credential, or filesystem path to the run record. ## Model Used - OpenAI GPT-6 through Codex, with reasoning, code editing, and tool use. The session does not expose an exact model build ID or context-window capacity. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes: #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub references) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
51ea624344
commit
b7f1c3769c
5 files changed
+643
-13
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
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<void>(resolve => { locked = resolve; });
|
||||
const hold = new Promise<void>(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);
|
||||
});
|
||||
});
|
||||
@@ -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<Selection, "admittedRunId" | "environmentLeaseId"> & {
|
||||
schema: "paperclip.native-provider-pack-admission.v1";
|
||||
stage: "session_admitted";
|
||||
authority: "core_verified_selection";
|
||||
runId: string;
|
||||
runnerEnvironmentLeaseId: string;
|
||||
inheritedFromRunId?: string;
|
||||
};
|
||||
|
||||
const admitted = new WeakMap<NativeSession, Readonly<Selection>>();
|
||||
const digest = /^sha256:[a-f0-9]{64}$/;
|
||||
type NativeRunIdentity = ReturnType<NativeSession["identity"]>;
|
||||
|
||||
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<string, unknown>;
|
||||
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<boolean> {
|
||||
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<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeProviderPackAdmission'`,
|
||||
state: sql<unknown>`${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<string, unknown>;
|
||||
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<boolean> {
|
||||
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<string, unknown> | 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<string, unknown>), 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;
|
||||
});
|
||||
}
|
||||
@@ -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<typeof import("node:fs")>();
|
||||
@@ -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<typeof admittedNativeProviderPack> = 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<typeof createRunnerdBackend>[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<string, unknown>) => 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<string, unknown> = { preserved: { value: "unchanged" } };
|
||||
const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
|
||||
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<unknown>) => target.transaction(tx => operation(wrap(tx as unknown as Db)) as never);
|
||||
if (key === "select") return (fields?: Record<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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<string, unknown>).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<string, unknown> }> = [];
|
||||
let observed: Record<string, unknown> | 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<unknown>) => 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<string, unknown>) => {
|
||||
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()
|
||||
|
||||
@@ -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<void>;
|
||||
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<NativeSessionBackend["openSession"]>[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;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user