mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
Release and resume retained plugin Runner sessions in Core (#19611)
This commit is contained in:
Nicky Leach authored and GitHub committed
1 parent
9efee7ee16
commit
b21884bdcf
7 files changed
+965
-80
No files matched your search
@@ -117,20 +117,47 @@ warm or source recovery capabilities do not imply support. Older strict Core
|
||||
schemas reject this description field, so coordinate advertisement with a
|
||||
compatible Core version.
|
||||
|
||||
Core's lease-bound task owner exposes capability-gated `release` and `resume`
|
||||
methods for both Codex and ACPX. They require `releaseResumeVersion: 1`, warm-task
|
||||
support and a selected account, and validate the operation result and exact task
|
||||
ID with the SDK parser. These methods are a forwarding facade; Core does not yet
|
||||
call them automatically or change its warm-session timeout policy.
|
||||
Core uses this pair only for explicitly supported warm tasks with a selected
|
||||
account. It releases after the turn's required events, tool results, journal
|
||||
writes and cleanup have settled. It keeps the exact session busy until the host
|
||||
confirms release and Core records that source. Existing tasks without the
|
||||
capability retain their ordinary warm lifecycle.
|
||||
|
||||
The caller must serialize settled release against new work and disposal, quiet
|
||||
its owned connection's liveness checks and reconnect loop before release, and
|
||||
retain the same pending ticket for ambiguous resume retries. Ordinary owner
|
||||
rebind must first check the new run's account, credentials and project scope.
|
||||
After resume, authenticate the source with the fresh ticket before warm readiness
|
||||
and ordinary `run.attach`. An RPC error does not prove that release or resume had
|
||||
no effect; it must not trigger reconnect, a replacement ticket or turn replay.
|
||||
The facade does not supply this orchestration or enable provider retention.
|
||||
Before release, Core stops connection pings and reconnect work while keeping the
|
||||
existing receive/ACK path open. A confirmed release closes the route and stops
|
||||
its idle polling. An ambiguous release stays with the original run for bounded
|
||||
reconciliation. It cannot pass to a new run's task scope. Cancellation or an
|
||||
expired reconciliation budget uses ordinary host stop and physical cleanup;
|
||||
it does not change an already completed business result.
|
||||
|
||||
Resume first checks the new run's account, credential revision and project
|
||||
scope. It uses one fresh transient ticket, preserving that exact ticket through
|
||||
ambiguous replies. Core authenticates the source before warm readiness, provider
|
||||
preparation and ordinary `run.attach`; no new turn can pass that handover.
|
||||
Generic RPC errors do not prove refusal or permit a new ticket, reconnect or
|
||||
turn replay. A known `unavailable` result still requires ordinary physical
|
||||
cleanup. Transparent cold fallback remains a separate qualification requirement.
|
||||
|
||||
A normal Core cache timeout detaches only Core's quiet local owner. The host
|
||||
keeps its retained resources under its own retention policy. Core's existing
|
||||
completed-session record stores only the confirmed source, physical task and a
|
||||
digest of the selected account, credential revision and workspace scope. The
|
||||
transport returns the exact source after the host's release receipt; Core does
|
||||
not infer it from a local checkpoint marker, which may still name an older run.
|
||||
An invalid source receipt or failed proof write keeps cleanup under the same
|
||||
owner until physical retirement finishes. Later
|
||||
cache adoption validates that Core-authored proof before claiming or
|
||||
reconnecting. A stale or missing proof never licenses a claim. A changed
|
||||
selection remains blocked until guarded retirement/recovery; it does not imply
|
||||
that a cold launch is safe. Confirmed claim or physical cleanup clears only the
|
||||
matching proof. Raw tickets are never stored in this record.
|
||||
|
||||
These Core paths do not advertise the capability or opt ordinary tasks into a
|
||||
private host channel. Existing durable lease and plugin-worker idle checks still
|
||||
apply; retaining a released Runner does not by itself make a tenant safe to sleep.
|
||||
Provider recovery, bounded host retention and deployment qualification are still
|
||||
required before activation. Recovery from a Core crash
|
||||
between claim and source authentication is not qualified by this integration.
|
||||
|
||||
## Error codes
|
||||
|
||||
|
||||
@@ -324,6 +324,7 @@ const ingressRetrySuitePath = "server/src/services/native-runtime/native-ingress
|
||||
const usageDeliverySuitePath = "server/src/services/native-runtime/native-usage-delivery.integration.test.ts";
|
||||
const warmCheckpointOwnershipSuitePath = "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts";
|
||||
const pluginWarmOwnerSuitePath = "server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts";
|
||||
const pluginReleaseResumeSuitePath = "server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts";
|
||||
const dotRunnerSuitePath = "server/src/__tests__/dot-runner.test.ts";
|
||||
const restartRecoverySuitePath =
|
||||
"server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts";
|
||||
@@ -373,7 +374,8 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes
|
||||
assert.ok(!files.includes(usageDeliverySuitePath));
|
||||
assert.ok(!files.includes(warmCheckpointOwnershipSuitePath));
|
||||
assert.ok(!files.includes(pluginWarmOwnerSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.ok(!files.includes(pluginReleaseResumeSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, pluginReleaseResumeSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.equal(new Set(files).size, files.length);
|
||||
const defaultRun = dryRunJson([], prEnv);
|
||||
assert.ok(defaultRun.generalServerSuiteCount === full.generalServerSuiteCount);
|
||||
@@ -395,6 +397,7 @@ for (const [caller, envOverrides] of [["Release", { GITHUB_WORKFLOW: "Release" }
|
||||
assert.ok(files.includes(restartRecoverySuitePath));
|
||||
assert.ok(files.includes(warmCheckpointOwnershipSuitePath));
|
||||
assert.ok(files.includes(pluginWarmOwnerSuitePath));
|
||||
assert.ok(files.includes(pluginReleaseResumeSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.equal(new Set(files).size, files.length);
|
||||
});
|
||||
@@ -429,6 +432,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => {
|
||||
usageDeliverySuitePath,
|
||||
warmCheckpointOwnershipSuitePath,
|
||||
pluginWarmOwnerSuitePath,
|
||||
pluginReleaseResumeSuitePath,
|
||||
]);
|
||||
});
|
||||
|
||||
|
||||
@@ -106,6 +106,7 @@ const nativeRunnerSuites = [
|
||||
"server/src/services/native-runtime/native-usage-delivery.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts",
|
||||
"server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts",
|
||||
"server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts",
|
||||
];
|
||||
// In the PR workflow (pr.yml, the caller of pr-trusted.yml — reusable
|
||||
// workflows inherit the caller's GITHUB_WORKFLOW), the last Verify Paperclip
|
||||
|
||||
@@ -6,6 +6,7 @@ import * as harnessStorage from "./native-harness-backup-storage.js";
|
||||
import * as stagedAttachmentRecovery from "./native-staged-attachment-recovery.js";
|
||||
import * as controllerRecovery from "./native-controller-recovery.js";
|
||||
import { isSettledRemoteWarmControlState } from "./remote-warm-session-recovery.js";
|
||||
import * as remoteWarmRecoveryRuntime from "./remote-warm-session-recovery.js";
|
||||
import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
|
||||
import * as environmentTaskRuntime from "../environment-task-runtime.js";
|
||||
import * as pluginTaskRuntime from "./plugin-runner-task.js";
|
||||
@@ -171,7 +172,7 @@ type BackendFactoryOptions = {
|
||||
type RunnerTransportOptions = {
|
||||
authorityRunId?: string;
|
||||
onAccountingEvent?: (event: PrpEvent) => Promise<void>;
|
||||
onRecoveryAccountingEvent?: unknown;
|
||||
onRecoveryAccountingEvent?: (event: PrpEvent, identity: import("../../vendor/paperclip-runner/index.js").DurableRecoveryIdentity) => Promise<void>;
|
||||
environment?: NodeJS.ProcessEnv;
|
||||
acpxRuntimeDirectory?: string;
|
||||
adoptExistingRunner?: { pid: number; isAlive: () => Promise<boolean> | boolean };
|
||||
@@ -8381,6 +8382,13 @@ describe("native terminal-turn accounting", () => {
|
||||
expect(firstCapture).toHaveBeenCalledTimes(2);
|
||||
expect(secondCapture).toHaveBeenCalledTimes(1);
|
||||
await expect(retainedHook({ ...firstReport, runId: "unrelated-run" })).rejects.toThrow("native_usage_run_binding_mismatch");
|
||||
const recoveryHook = state.createTransport.mock.calls.at(-1)![0].onRecoveryAccountingEvent!;
|
||||
const source = { runnerInstanceId: "warm-producer", environmentLeaseId: "warm-usage-workspace",
|
||||
runId: first.binding.runId, normalizedSessionId: first.session.normalizedSessionId!, turnId: "source-turn", itemId: "source-item" };
|
||||
await recoveryHook(firstReport, source);
|
||||
expect(firstCapture).toHaveBeenCalledTimes(3);
|
||||
expect(secondCapture).toHaveBeenCalledTimes(1);
|
||||
await expect(recoveryHook({ ...firstReport, runId: "foreign-source" }, source)).rejects.toThrow("native_recovery_usage_source_unproven");
|
||||
} finally {
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
@@ -15857,6 +15865,318 @@ describe("runnerd provider runtime wiring", () => {
|
||||
} finally {owner.mockRestore();await rm(home,{recursive:true,force:true});}
|
||||
});
|
||||
|
||||
it.each((["codex", "claude"] as const).flatMap(provider => [[false, false, false, false], [true, false, false, false], [false, true, false, false], [false, false, true, false], ...(provider === "codex" ? [[false, false, false, true]] : [])].map(([unknown, shutdown, journalFailure, deadline]) => [provider, unknown, shutdown, journalFailure, deadline] as const)))(
|
||||
"releases settled %s only under its current owner and detaches its cache locally (unknown=%s, shutdown=%s, journalFailure=%s, deadline=%s)", async (provider, unknown, shutdown, journalFailure, deadline) => {
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
expect(readNativeSessionIdleWork()).toBe("none");
|
||||
const previousHome = process.env.PAPERCLIP_HOME;
|
||||
process.env.PAPERCLIP_HOME = isolatedStateDirectory;
|
||||
const current = scopedNativeExecution(`released-${provider}-${unknown}`);
|
||||
current.session.lifecyclePolicy = { mode: "warm", idleTimeoutMs: unknown ? 750 : 250 };
|
||||
if (provider === "claude") current.provider = { kind: "acpx", agent: "claude", model: "fixture", permissionMode: "deny-all" } as never;
|
||||
const source = { runnerInstanceId: "released-runner", environmentLeaseId: "released-lease", runId: current.binding.runId,
|
||||
normalizedSessionId: current.session.normalizedSessionId!, turnId: "settled-turn", itemId: "settled-item" };
|
||||
const root = await initializeScopedRunnerRoot(current, source.runnerInstanceId);
|
||||
await writeRunnerAuthority(root, source, "ready");
|
||||
const home = join(isolatedStateDirectory, "selected-account");
|
||||
await mkdir(home, { recursive: true });
|
||||
await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 });
|
||||
const task = vi.fn(async (_operation: unknown) => ({ kind: "description", taskId: "retained-task",
|
||||
launchPreparationVersion: 1, projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1,
|
||||
runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }));
|
||||
const target = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: ".", environmentId: "retained-environment",
|
||||
leaseId: source.environmentLeaseId, runnerTask: task, sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId: "retained-task" } };
|
||||
const owners = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner");
|
||||
let allowRelease!: () => void;
|
||||
const receipt = new Promise<void>(resolve => { allowRelease = resolve; });
|
||||
const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
|
||||
let attempts = 0;
|
||||
const release = vi.fn(async () => {
|
||||
expect(owners.mock.calls.at(-1)![0].authorityRunId).toBe(current.binding.runId);
|
||||
expect(updates.some(update => update.values.leaseOwner === null && update.values.recoveryState === null)).toBe(true);
|
||||
if (unknown && attempts++ === 0) throw new Error("fixture lost release reply");
|
||||
await receipt;
|
||||
return structuredClone(source);
|
||||
});
|
||||
let allowPhysicalStop!: () => void;
|
||||
const physicalStop = new Promise<void>(resolve => { allowPhysicalStop = resolve; });
|
||||
const close = vi.fn(async () => { if (shutdown || deadline || journalFailure) await physicalStop; }), detachControllerForRestart = vi.fn(async () => {});
|
||||
const session = { release, close, detachControllerForRestart };
|
||||
const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
|
||||
summary: "Completed", completionClaim: { contractRevision: current.completionContract.contract.revision,
|
||||
objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], artifacts: [], attentionRequests: [] });
|
||||
if (!validated.ok) throw new Error("Invalid completion fixture");
|
||||
state.execute.mockReset().mockImplementationOnce(async options => {
|
||||
await options.onCheckpoint?.({ backendKind: "runner", driverKind: current.session.driverKind, sessionId: "driver-session",
|
||||
identity: { ...current.binding, sessionId: source.normalizedSessionId }, providerSessionId: "provider-session",
|
||||
activeTurnId: null, pendingRuntimeRequests: [] });
|
||||
await options.onSession?.(session);
|
||||
return { result: validated.result, terminal: { runTerminalState: "succeeded" }, turnId: source.turnId,
|
||||
normalizedSessionId: source.normalizedSessionId, providerSessionId: "provider-session", driverKind: "test", driverVersion: "1",
|
||||
nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
|
||||
});
|
||||
const execute = executePaperclipNativeSession({ db: leaseDb(current, {}, {}, updates), execution: current,
|
||||
runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, runnerExecutionTarget: target,
|
||||
managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account",
|
||||
runnerEnvironment: provider === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {} });
|
||||
void execute.catch(() => undefined);
|
||||
try {
|
||||
await vi.waitFor(() => expect(release, inspect(updates.map(update => update.values.failureDetail))).toHaveBeenCalledTimes(unknown ? 2 : 1));
|
||||
expect(readNativeSessionIdleWork()).toBe("present");
|
||||
const next = { ...current, binding: { ...current.binding, runId: `${current.binding.runId}-next` } };
|
||||
const nextTask = vi.fn();
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next,
|
||||
runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerExecutionTarget: { ...target, runnerTask: nextTask } })).rejects.toThrow("native_session_supervisor_busy");
|
||||
expect(nextTask).not.toHaveBeenCalled();
|
||||
if (provider === "codex" && !unknown && !shutdown && !journalFailure && !deadline) {
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next,
|
||||
runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerExecutionTarget: { ...target, environmentId: "competing-environment", runnerTask: nextTask,
|
||||
sandboxLeaseAcquisition: { ...target.sandboxLeaseAcquisition, providerLeaseId: "competing-task" } } })).rejects.toThrow("native_session_supervisor_busy");
|
||||
expect(nextTask).not.toHaveBeenCalled();
|
||||
expect(release).toHaveBeenCalledTimes(1);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(readNativeSessionIdleWork()).toBe("present");
|
||||
}
|
||||
if (shutdown || deadline) {
|
||||
if (shutdown) expect(await closeIdleWarmNativeSessionsForRestart()).toMatchObject({ busy: 1 });
|
||||
await vi.waitFor(() => expect(close).toHaveBeenCalledOnce());
|
||||
expect(readNativeSessionIdleWork()).toBe("present");
|
||||
expect(detachControllerForRestart).not.toHaveBeenCalled();
|
||||
// A late host receipt cannot publish a released owner during retirement.
|
||||
allowRelease();
|
||||
await receipt;
|
||||
await Promise.resolve();
|
||||
expect(readNativeSessionIdleWork()).toBe("present");
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next,
|
||||
runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerExecutionTarget: { ...target, runnerTask: nextTask } })).rejects.toThrow("native_session_supervisor_busy");
|
||||
expect(nextTask).not.toHaveBeenCalled();
|
||||
allowPhysicalStop();
|
||||
await expect(execute).resolves.toMatchObject({ sessionId: source.normalizedSessionId });
|
||||
expect(readNativeSessionIdleWork()).toBe("none");
|
||||
const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions");
|
||||
expect(JSON.parse(await readFile(join(directory, (await readdir(directory))[0]!), "utf8")).releasedRunner).toBeUndefined();
|
||||
return;
|
||||
}
|
||||
if (journalFailure) {
|
||||
const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions");
|
||||
const path = join(directory, (await readdir(directory))[0]!);
|
||||
await rename(path, `${path}.saved`);
|
||||
await mkdir(path);
|
||||
let finished = false;
|
||||
void execute.finally(() => { finished = true; }).catch(() => undefined);
|
||||
allowRelease();
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
expect(finished).toBe(false);
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next,
|
||||
runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account",
|
||||
runnerExecutionTarget: { ...target, remoteCwd: "different-workspace", runnerTask: nextTask } })).rejects.toThrow("native_session_supervisor_busy");
|
||||
expect(nextTask).not.toHaveBeenCalled();
|
||||
await vi.waitFor(() => expect(close).toHaveBeenCalledOnce());
|
||||
expect(readNativeSessionIdleWork()).toBe("present");
|
||||
expect(finished).toBe(false);
|
||||
await rm(path, { recursive: true });
|
||||
await rename(`${path}.saved`, path);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toBeUndefined();
|
||||
allowPhysicalStop();
|
||||
await expect(execute).resolves.toMatchObject({ sessionId: source.normalizedSessionId });
|
||||
expect(readNativeSessionIdleWork()).toBe("none");
|
||||
expect(detachControllerForRestart).not.toHaveBeenCalled();
|
||||
return;
|
||||
}
|
||||
allowRelease();
|
||||
await expect(execute).resolves.toMatchObject({ sessionId: source.normalizedSessionId });
|
||||
expect(readNativeSessionIdleWork()).toBe("none");
|
||||
const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions");
|
||||
const files = await readdir(directory);
|
||||
const path = join(directory, files[0]!);
|
||||
const envelope = JSON.parse(await readFile(path, "utf8"));
|
||||
expect(envelope.releasedRunner).toMatchObject({ source, taskId: "retained-task", scopeDigest: expect.any(String) });
|
||||
expect(JSON.stringify(envelope)).not.toContain("synthetic-fixture");
|
||||
if (!unknown) {
|
||||
const calls = owners.mock.calls.length;
|
||||
for (const changed of [{ remoteCwd: "different-workspace" }, { runnerTaskProjectIds: ["different-project"] }]) {
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next,
|
||||
runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account",
|
||||
runnerExecutionTarget: { ...target, ...changed, runnerTask: nextTask },
|
||||
runnerEnvironment: provider === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {} })).rejects.toThrow("released_owner_scope_mismatch");
|
||||
}
|
||||
expect(owners).toHaveBeenCalledTimes(calls);
|
||||
expect(nextTask).not.toHaveBeenCalled();
|
||||
expect(state.execute).toHaveBeenCalledTimes(1);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toEqual(envelope.releasedRunner);
|
||||
}
|
||||
await vi.waitFor(() => expect(detachControllerForRestart).toHaveBeenCalledOnce());
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(task.mock.calls.map(([operation]) => (operation as { kind: string }).kind)).toEqual(["describe"]);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toEqual(envelope.releasedRunner);
|
||||
if (provider === "codex" && !unknown) {
|
||||
const inspected = vi.spyOn(remoteWarmRecoveryRuntime, "inspectRemoteWarmSessionRecovery").mockImplementation(async request => {
|
||||
request.loadCheckpoint(current);
|
||||
throw new Error("fixture completed-source proof checked");
|
||||
});
|
||||
try {
|
||||
const calls = owners.mock.calls.length;
|
||||
const base = { db: leaseDb(next), execution: next, runnerInstanceId: source.runnerInstanceId,
|
||||
useRunnerd: true, runnerIngressAuthorized: true, runnerExecutionTarget: target,
|
||||
managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account" };
|
||||
await expect(executePaperclipNativeSession(base)).rejects.toThrow("fixture completed-source proof checked");
|
||||
for (const changed of [
|
||||
{ managedAiAccountIdentity: "different-account" }, { managedAiCredentialIdentity: "different-revision" },
|
||||
{ runnerExecutionTarget: { ...target, runnerTaskProjectIds: ["different-project"] } },
|
||||
{ runnerExecutionTarget: { ...target, remoteCwd: "different-workspace" } },
|
||||
{ runnerExecutionTarget: { ...target, sandboxLeaseAcquisition: { ...target.sandboxLeaseAcquisition, providerLeaseId: "different-task" } } },
|
||||
]) await expect(executePaperclipNativeSession({ ...base, ...changed })).rejects.toThrow("released_owner_scope_mismatch");
|
||||
expect(owners).toHaveBeenCalledTimes(calls);
|
||||
expect(state.execute).toHaveBeenCalledTimes(1);
|
||||
} finally { inspected.mockRestore(); }
|
||||
}
|
||||
await owners.mock.calls.at(-1)![0].onResumed!(source);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toBeUndefined();
|
||||
const replacement = { ...envelope, snapshot: { ...envelope.snapshot, identity: { ...envelope.snapshot.identity, runId: "new-owner" } },
|
||||
releasedRunner: { ...envelope.releasedRunner, source: { ...source, runId: "new-owner" } } };
|
||||
await writeFile(path, JSON.stringify(replacement));
|
||||
await expect(owners.mock.calls.at(-1)![0].onResumed!(source)).rejects.toThrow("released_source_mismatch");
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toEqual(replacement.releasedRunner);
|
||||
} finally {
|
||||
allowRelease(); allowPhysicalStop(); await execute.catch(() => undefined); await closeIdleWarmNativeSessionsForRestart(); owners.mockRestore();
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome;
|
||||
}
|
||||
});
|
||||
|
||||
it.each((["codex", "claude"] as const).flatMap(provider => (["current", "stale", ...(provider === "codex" ? ["foreign runner", "foreign lease"] : [])] as const)
|
||||
.map(receipt => [provider, receipt] as const)))(
|
||||
"records the confirmed %s source after external checkpoint registration and attachment (receipt=%s)", async (provider, receipt) => {
|
||||
const invalidReceipt = receipt !== "current";
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
const previousHome = process.env.PAPERCLIP_HOME;
|
||||
process.env.PAPERCLIP_HOME = isolatedStateDirectory;
|
||||
const { permissionPolicy: _permissionPolicy, ...claudeProfile } = QUALIFIED_ACPX_PROFILES.claude;
|
||||
let first: NativeExecutionInput = scopedNativeExecution(`released-external-${provider}-${receipt.replaceAll(" ", "-")}`);
|
||||
first.session.lifecyclePolicy = { mode: "warm", idleTimeoutMs: 5_000 };
|
||||
if (provider === "claude") first = { ...first, schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null,
|
||||
runtimeContext: nativeRuntimeContextFixture(), session: { ...first.session, driverKind: "acpx_runtime" },
|
||||
provider: { kind: "acpx", agent: "claude", model: "claude-sonnet-4-6", permissionMode: "approve-all", profile: claudeProfile } };
|
||||
parseNativeExecutionInput(first);
|
||||
const next = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-next` } };
|
||||
const source = { runnerInstanceId: "external-runner", environmentLeaseId: "external-lease", runId: first.binding.runId,
|
||||
normalizedSessionId: first.session.normalizedSessionId!, turnId: `turn-${first.binding.runId}`, itemId: `item-${first.binding.runId}` };
|
||||
const successor = { ...source, runId: next.binding.runId, turnId: "attached-turn", itemId: "attached-item" };
|
||||
const root = await initializeScopedRunnerRoot(first, source.runnerInstanceId);
|
||||
await writeRunnerAuthority(root, source, "ready");
|
||||
const home = join(isolatedStateDirectory, "external-selected-account");
|
||||
await mkdir(home, { recursive: true });
|
||||
await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 });
|
||||
const task = vi.fn(async () => ({ kind: "description", taskId: "external-task", launchPreparationVersion: 1,
|
||||
projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1,
|
||||
runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }));
|
||||
const target = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: ".", environmentId: "external-environment",
|
||||
leaseId: source.environmentLeaseId, runnerTask: task, sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId: "external-task" } };
|
||||
const createOwner = pluginTaskRuntime.createPluginRunnerTaskOwner;
|
||||
const owners = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockImplementation(input => createOwner({ ...input, adoptIdentity: source }));
|
||||
const saved = new Map<string, import("../../vendor/paperclip-runner/index.js").DurablePrpCheckpoint>();
|
||||
const restore = vi.spyOn(runnerCheckpoint, "restoreNativeRunnerCheckpoint").mockImplementation(async (_db, binding, directory) => {
|
||||
const checkpoint = saved.get(binding.runId);
|
||||
if (!checkpoint) throw new Error("Missing fixture controller checkpoint");
|
||||
checkpoint.restoreSessionDirectory(directory);
|
||||
return checkpoint;
|
||||
});
|
||||
let authority: DurablePrpControlPlane | undefined;
|
||||
let register: import("../../vendor/paperclip-runner/index.js").RunnerExecutionDriver["controlPlaneRegistration"];
|
||||
let releasePhysical!: () => void;
|
||||
const physical = new Promise<void>(resolve => { releasePhysical = resolve; });
|
||||
const close = vi.fn(async () => { if (invalidReceipt) await physical; });
|
||||
const release = vi.fn(async () => {
|
||||
await authority!.flushCheckpoint();
|
||||
const identity = authority!.store.state.identity;
|
||||
if (identity.runId !== successor.runId || receipt === "current") return structuredClone(identity);
|
||||
if (receipt === "stale") return structuredClone(source);
|
||||
return { ...identity, ...(receipt === "foreign runner" ? { runnerInstanceId: "successor-candidate-runner" }
|
||||
: { environmentLeaseId: "successor-controller-lease" }) };
|
||||
});
|
||||
const session = { close, release, detachControllerForRestart: vi.fn(async () => {}) };
|
||||
const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Completed",
|
||||
completionClaim: { contractRevision: first.completionContract.contract.revision, objectiveSatisfied: true, criteria: [], remainingWork: [] },
|
||||
evidence: [], verification: [], artifacts: [], attentionRequests: [] });
|
||||
if (!validated.ok) throw new Error("Invalid completion fixture");
|
||||
state.execute.mockReset().mockImplementation(async options => {
|
||||
const current = options.input.binding.runId === first.binding.runId ? first : next;
|
||||
if (!authority) {
|
||||
state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
|
||||
const transport = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & {
|
||||
controlPlaneRegistration: NonNullable<import("../../vendor/paperclip-runner/index.js").RunnerExecutionDriver["controlPlaneRegistration"]>;
|
||||
};
|
||||
await rm(join(root, "control-plane"), { recursive: true });
|
||||
authority = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), identity: source,
|
||||
expectedRunnerVersion: "fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
const persist = authority.setCheckpointPersistence.bind(authority);
|
||||
// Only the backing store is modeled; registration and the fail-closed marker are real.
|
||||
vi.spyOn(authority, "setCheckpointPersistence").mockImplementation(() => persist(async checkpoint => { saved.set(checkpoint.runId, checkpoint); }));
|
||||
register = transport.controlPlaneRegistration;
|
||||
await register(authority, source);
|
||||
expect(JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).schema)
|
||||
.toBe("paperclip.runner.external-control-plane-state.v1");
|
||||
} else {
|
||||
expect(options.existingSession).toBe(session);
|
||||
await owners.mock.calls[0]![0].onResumed!(source);
|
||||
authority.rotateRunIdentity(successor);
|
||||
await register!(authority, successor);
|
||||
await writeFile(join(root, "runner/runner-state.json"), JSON.stringify(durableRunnerState(successor, "ready")));
|
||||
expect(JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).identity.runId).toBe(source.runId);
|
||||
}
|
||||
await options.onCheckpoint?.({ backendKind: "runner", driverKind: current.session.driverKind, sessionId: "driver-session",
|
||||
identity: { ...current.binding, sessionId: source.normalizedSessionId }, providerSessionId: "provider-session", activeTurnId: null, pendingRuntimeRequests: [] });
|
||||
await options.onSession?.(session);
|
||||
return { result: validated.result, terminal: { runTerminalState: "succeeded" }, turnId: "provider-turn",
|
||||
normalizedSessionId: source.normalizedSessionId, providerSessionId: "provider-session", driverKind: "test", driverVersion: "1",
|
||||
nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
|
||||
});
|
||||
const run = (execution: NativeExecutionInput) => {
|
||||
const db = leaseDb(execution, { leaseOwner: "external-controller", controllerGeneration: 1 });
|
||||
if (execution.binding.runId === next.binding.runId) {
|
||||
const select = db.select.bind(db);
|
||||
db.select = ((fields?: Record<string, unknown>) => fields?.status === heartbeatRuns.status && fields?.runnerProfileJson === heartbeatRuns.runnerProfileJson
|
||||
? priorNativeRunDb(first).select(fields as never) : select(fields as never)) as Db["select"];
|
||||
}
|
||||
return executePaperclipNativeSession({ db, leaseOwner: "external-controller", execution,
|
||||
runnerInstanceId: execution === next ? "successor-candidate-runner" : source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerExecutionTarget: execution === next ? { ...target, leaseId: "successor-controller-lease" } : target,
|
||||
managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision",
|
||||
managedAiAccountIdentity: "selected-account", runnerEnvironment: provider === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {} });
|
||||
};
|
||||
let following: ReturnType<typeof run> | undefined;
|
||||
try {
|
||||
await expect(run(first)).resolves.toMatchObject({ sessionId: source.normalizedSessionId });
|
||||
expect(release).toHaveBeenCalledOnce();
|
||||
const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions");
|
||||
const path = join(directory, (await readdir(directory))[0]!);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner.source).toEqual(source);
|
||||
following = run(next);
|
||||
void following.catch(() => undefined);
|
||||
if (invalidReceipt) {
|
||||
await vi.waitFor(() => expect(close).toHaveBeenCalledOnce());
|
||||
expect(readNativeSessionIdleWork()).toBe("present");
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toBeUndefined();
|
||||
releasePhysical();
|
||||
await expect(following).resolves.toMatchObject({ sessionId: source.normalizedSessionId });
|
||||
expect(readNativeSessionIdleWork()).toBe("none");
|
||||
} else {
|
||||
await expect(following).resolves.toMatchObject({ sessionId: source.normalizedSessionId });
|
||||
expect(release).toHaveBeenCalledTimes(2);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).releasedRunner.source).toEqual(successor);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
}
|
||||
} finally {
|
||||
releasePhysical(); await following?.catch(() => undefined); await closeIdleWarmNativeSessionsForRestart();
|
||||
await authority?.stop(); owners.mockRestore(); restore.mockRestore();
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome;
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["codex", "opencode"] as const)("keeps assigned MCP credentials on the control plane for %s", async (providerKind) => {
|
||||
const selectedExecution: NativeExecutionInput = providerKind === "opencode"
|
||||
? { ...execution, provider: { kind: "opencode", model: "openrouter/deepseek/deepseek-v4.1-flash-1017", permissionMode: "allow" } }
|
||||
|
||||
@@ -45,7 +45,7 @@ import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
||||
import { createRunUsageRecorder } from "../usage-receipts.js";
|
||||
import { registerPluginTaskCredentialHandoff, type PluginTaskCredentialWritebackBinding } from "../plugin-task-credential-writeback.js";
|
||||
import { resolvePluginTaskResume } from "../environment-task-runtime.js";
|
||||
import { environmentTaskResultSchema } from "@paperclipai/plugin-sdk";
|
||||
import { environmentTaskOperationSchema, environmentTaskResultSchema } from "@paperclipai/plugin-sdk";
|
||||
import { preparePluginRunnerCredentials } from "./plugin-runner-credentials.js";
|
||||
import { createPluginRunnerTaskOwner, verifyPluginRunnerTaskReattachment } from "./plugin-runner-task.js";
|
||||
import { NativeExecutionContext } from "./native-execution-context.js";
|
||||
@@ -669,6 +669,8 @@ type WarmNativeSession = {
|
||||
companyId: string;
|
||||
environmentId: string | null;
|
||||
busy: boolean;
|
||||
released?: boolean;
|
||||
finishRelease?: () => Promise<void>;
|
||||
closeOnReleaseReason?: string;
|
||||
idleTimer: ReturnType<typeof setTimeout> | null;
|
||||
lastActivityAt: string;
|
||||
@@ -716,7 +718,9 @@ const pluginTaskSessionOwners = new WeakMap<NativeSession, ReturnType<typeof cre
|
||||
const closingWarmNativeSessions = new Map<string, Promise<void>>();
|
||||
|
||||
export function readNativeSessionIdleWork(): IdleLocalWork {
|
||||
return warmNativeSessions.size !== 0 || closingWarmNativeSessions.size !== 0 ? "present" : "none";
|
||||
return closingWarmNativeSessions.size !== 0 || [...warmNativeSessions.values()].some(entry =>
|
||||
entry.busy || entry.preparingRunId || entry.instructionPreparationRunId || entry.closeOnReleaseReason || !entry.released)
|
||||
? "present" : "none";
|
||||
}
|
||||
registerIdleLocalWorkProbe(readNativeSessionIdleWork);
|
||||
|
||||
@@ -876,7 +880,8 @@ async function closeIdleWarmNativeSessions(input: {
|
||||
if (input.environmentId !== undefined && entry.environmentId !== input.environmentId) {
|
||||
continue;
|
||||
}
|
||||
if (entry.busy || entry.preparingRunId || executingRunnerdSessionScopes.has(sessionId)) {
|
||||
if (entry.busy || entry.preparingRunId || executingRunnerdSessionScopes.has(sessionId)
|
||||
|| (input.canRetire && (entry.instructionPreparationRunId || closingWarmNativeSessions.has(sessionId)))) {
|
||||
// A busy turn can complete while another idle session is checkpointing.
|
||||
// Fence that entry now so its eventual release cannot leave a new idle
|
||||
// owner behind after the shutdown sweep has already passed it.
|
||||
@@ -884,6 +889,8 @@ async function closeIdleWarmNativeSessions(input: {
|
||||
busy += 1;
|
||||
continue;
|
||||
}
|
||||
// The host owns retention after release; a quiet local handle needs no retirement for sleep.
|
||||
if (input.canRetire && entry.released && !entry.closeOnReleaseReason) continue;
|
||||
if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
|
||||
// Keep a fenced owner until both retirement and stopped collection succeed.
|
||||
// Failure must not make a live directory eligible for a replacement owner.
|
||||
@@ -1927,6 +1934,10 @@ class SessionAuthorityEpoch {
|
||||
if (!isNativeUsageAccountingEvent(event)) return;
|
||||
this.#assertCurrent();
|
||||
if (event.runId === this.runId) throw new Error("native_recovery_usage_source_unproven");
|
||||
if (event.runId === this.#drainingRun?.runId) {
|
||||
await this.acceptAccountingEvent(event);
|
||||
return;
|
||||
}
|
||||
await capture();
|
||||
this.#assertCurrent();
|
||||
}
|
||||
@@ -7324,27 +7335,92 @@ function legacyNativeSessionCheckpointPath(
|
||||
);
|
||||
}
|
||||
|
||||
type ReleasedWarmRunner = { source: DurableRecoveryIdentity; taskId: string; scopeDigest: string };
|
||||
|
||||
function releasedWarmScope(input: Parameters<typeof executePaperclipNativeSession>[0]): string {
|
||||
const target = input.runnerExecutionTarget;
|
||||
return nativeSha256({ companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
|
||||
environmentId: target?.kind === "remote" ? target.environmentId ?? null : null, workspace: nativeSessionWorkspaceScope(input.execution),
|
||||
cwd: input.execution.workspace.cwd, remoteCwd: target?.kind === "remote" ? target.remoteCwd : null,
|
||||
projectIds: target?.kind === "remote" && target.transport === "sandbox" ? target.runnerTaskProjectIds ?? [] : [],
|
||||
accountIdentity: input.managedAiAccountIdentity ?? null, credentialIdentity: input.managedAiCredentialIdentity ?? null });
|
||||
}
|
||||
|
||||
function requireReleasedWarmScope(input: Parameters<typeof executePaperclipNativeSession>[0],
|
||||
proof: ReleasedWarmRunner | undefined, identity: unknown): DurableRecoveryIdentity {
|
||||
const target = input.runnerExecutionTarget;
|
||||
if (!proof || target?.kind !== "remote" || target.transport !== "sandbox" || !target.runnerTask
|
||||
|| proof.taskId !== target.sandboxLeaseAcquisition?.providerLeaseId
|
||||
|| proof.scopeDigest !== releasedWarmScope(input) || canonicalJson(proof.source) !== canonicalJson(identity)) {
|
||||
throw new Error("native_session_released_owner_scope_mismatch");
|
||||
}
|
||||
return proof.source;
|
||||
}
|
||||
|
||||
function readWarmNativeEnvelope(execution: NativeExecutionInput): Record<string, unknown> | null {
|
||||
const path = nativeSessionCheckpointPath(execution);
|
||||
if (!existsSync(path)) return null;
|
||||
const envelope = record(JSON.parse(readBoundedNativeFile(path, NATIVE_WARM_CHECKPOINT_MAX_BYTES,
|
||||
"native_session_supervisor_checkpoint_too_large").toString("utf8")));
|
||||
if (envelope.schema !== "paperclip.native-session-supervisor.v1") throw new Error("native_session_supervisor_checkpoint_mismatch");
|
||||
return envelope;
|
||||
}
|
||||
|
||||
function releasedWarmRunner(envelope: Record<string, unknown> | null): ReleasedWarmRunner | undefined {
|
||||
if (!envelope || envelope.releasedRunner === undefined) return undefined;
|
||||
const proof = record(envelope.releasedRunner), source = record(proof.source);
|
||||
const snapshot = record(envelope.snapshot), identity = record(snapshot.identity);
|
||||
for (const key of ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"]) {
|
||||
if (typeof source[key] !== "string" || !source[key]) throw new Error("native_session_released_source_mismatch");
|
||||
}
|
||||
if (typeof proof.taskId !== "string" || !proof.taskId || typeof proof.scopeDigest !== "string"
|
||||
|| identity.runId !== source.runId || identity.sessionId !== source.normalizedSessionId) {
|
||||
throw new Error("native_session_released_source_mismatch");
|
||||
}
|
||||
return proof as ReleasedWarmRunner;
|
||||
}
|
||||
|
||||
function writeWarmNativeEnvelope(execution: NativeExecutionInput, envelope: Record<string, unknown>): void {
|
||||
const path = nativeSessionCheckpointPath(execution);
|
||||
const temporary = `${path}.${process.pid}.${randomUUID()}.tmp`;
|
||||
writeFileSync(temporary, JSON.stringify(envelope), { encoding: "utf8", mode: 0o600 });
|
||||
renameSync(temporary, path);
|
||||
chmodSync(path, 0o600);
|
||||
}
|
||||
|
||||
function updateReleasedWarmRunner(execution: NativeExecutionInput, source: DurableRecoveryIdentity,
|
||||
taskId: string, proof: ReleasedWarmRunner | null): void {
|
||||
const envelope = readWarmNativeEnvelope(execution);
|
||||
const previous = releasedWarmRunner(envelope);
|
||||
if (!envelope) {
|
||||
if (proof) throw new Error("native_session_supervisor_checkpoint_missing");
|
||||
return;
|
||||
}
|
||||
if (!proof && !previous) return;
|
||||
const identity = record(record(envelope.snapshot).identity);
|
||||
if (identity.companyId !== execution.binding.companyId || identity.agentId !== execution.binding.agentId
|
||||
|| identity.runId !== source.runId || identity.sessionId !== source.normalizedSessionId
|
||||
|| (previous && (previous.taskId !== taskId || canonicalJson(previous.source) !== canonicalJson(source)))) {
|
||||
throw new Error("native_session_released_source_mismatch");
|
||||
}
|
||||
if (proof) envelope.releasedRunner = proof;
|
||||
else delete envelope.releasedRunner;
|
||||
writeWarmNativeEnvelope(execution, envelope);
|
||||
}
|
||||
|
||||
function persistWarmNativeCheckpoint(
|
||||
execution: NativeExecutionInput,
|
||||
configDigest: string,
|
||||
snapshot: PersistedNativeSession,
|
||||
processConfigDigest?: string,
|
||||
): void {
|
||||
const path = nativeSessionCheckpointPath(execution);
|
||||
const temporary = `${path}.${process.pid}.${randomUUID()}.tmp`;
|
||||
writeFileSync(
|
||||
temporary,
|
||||
JSON.stringify({
|
||||
schema: "paperclip.native-session-supervisor.v1",
|
||||
configDigest,
|
||||
processConfigDigest,
|
||||
updatedAt: new Date().toISOString(),
|
||||
snapshot,
|
||||
}),
|
||||
{ encoding: "utf8", mode: 0o600 },
|
||||
);
|
||||
renameSync(temporary, path);
|
||||
chmodSync(path, 0o600);
|
||||
const previous = releasedWarmRunner(readWarmNativeEnvelope(execution));
|
||||
writeWarmNativeEnvelope(execution, {
|
||||
schema: "paperclip.native-session-supervisor.v1", configDigest, processConfigDigest,
|
||||
updatedAt: new Date().toISOString(), snapshot,
|
||||
...(previous && previous.source.runId === snapshot.identity.runId
|
||||
&& previous.source.normalizedSessionId === snapshot.identity.sessionId ? { releasedRunner: previous } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
function needsCoreGitHubAccess(input: Pick<Parameters<typeof executePaperclipNativeSession>[0], "managedGitHub" | "runnerExecutionTarget">): boolean {
|
||||
@@ -7497,10 +7573,17 @@ function armWarmNativeSessionIdleTimer(
|
||||
return;
|
||||
}
|
||||
current.busy = true;
|
||||
current.closeOnReleaseReason = "warm native session idle timeout";
|
||||
const closing = closeWarmNativeSession(current, current.closeOnReleaseReason)
|
||||
const released = current.released === true;
|
||||
if (!released) current.closeOnReleaseReason = "warm native session idle timeout";
|
||||
const closing = (released ? (async () => {
|
||||
if (!current.session.detachControllerForRestart) throw new Error("native_session_released_detach_unavailable");
|
||||
await current.session.detachControllerForRestart();
|
||||
await current.githubAccess?.stop();
|
||||
})() : closeWarmNativeSession(current, current.closeOnReleaseReason!))
|
||||
.then(() => { forgetWarmNativeSession(sessionId, current); })
|
||||
.catch(() => {})
|
||||
.catch(() => {
|
||||
if (released) current.closeOnReleaseReason = "released Runner local detach is unconfirmed";
|
||||
})
|
||||
.finally(() => {
|
||||
current.busy = false;
|
||||
if (closingWarmNativeSessions.get(sessionId) === closing) closingWarmNativeSessions.delete(sessionId);
|
||||
@@ -9098,29 +9181,8 @@ export async function executePaperclipNativeSession(input: {
|
||||
startupUnsettled = !(await startup.executionContext.drain());
|
||||
startup.executionContext.finish();
|
||||
}
|
||||
if (ownsPhysicalIngress && physicalIngressKey && executingNativeIngressAllocations.get(physicalIngressKey) === ownerToken) {
|
||||
executingNativeIngressAllocations.delete(physicalIngressKey);
|
||||
}
|
||||
if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) {
|
||||
executingNativeOwnerScopes.delete(ownerScope);
|
||||
}
|
||||
startup.resolve(null);
|
||||
startup.settle();
|
||||
if (nativeSessionStartups.get(runId) === startup) {
|
||||
nativeSessionStartups.delete(runId);
|
||||
setNativeExecutionActive(runId, false);
|
||||
}
|
||||
if (
|
||||
ownsSessionScope &&
|
||||
sessionScopeId !== null &&
|
||||
executingRunnerdSessionScopes.get(sessionScopeId) ===
|
||||
input.execution.binding.runId
|
||||
) {
|
||||
executingRunnerdSessionScopes.delete(sessionScopeId);
|
||||
}
|
||||
// Preparation can fail before the execution loop owns its normal cleanup.
|
||||
// Remove the settled startup before yielding, so detachment cannot reacquire
|
||||
// its old resolved session while renewal cleanup drains.
|
||||
// Join preparation cleanup before release; keep cancellation reachable
|
||||
// until the exact retained owner has finished the handover.
|
||||
await startup.executionLeaseRenewal?.stop().catch(() => undefined);
|
||||
if (startup.unstartedClaim && !nativeRunsDetachingForRestart.has(runId)) {
|
||||
await releaseUnstartedNativeExecutionClaim({ db: input.db, runId, companyId: input.execution.binding.companyId,
|
||||
@@ -9128,9 +9190,11 @@ export async function executePaperclipNativeSession(input: {
|
||||
logger.warn({ err: error, runId }, "Native preparation claim release failed; the control deadline will settle it");
|
||||
});
|
||||
}
|
||||
let stagedCleanupFailed = false;
|
||||
try {
|
||||
await cleanupStagedAttachments();
|
||||
} catch (cleanupError) {
|
||||
stagedCleanupFailed = true;
|
||||
// Cleanup is a confidentiality incident, but it occurs after the native
|
||||
// provider may already have completed the turn. Reclassifying that turn
|
||||
// as failed could replay provider side effects. Emit a private runtime
|
||||
@@ -9152,6 +9216,38 @@ export async function executePaperclipNativeSession(input: {
|
||||
companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
|
||||
}, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId,
|
||||
issueId: input.execution.binding.issueId });
|
||||
const pendingRelease = ownsSessionScope && sessionScopeId
|
||||
&& executingRunnerdSessionScopes.get(sessionScopeId) === runId
|
||||
? warmNativeSessions.get(sessionScopeId) : undefined;
|
||||
try {
|
||||
if (pendingRelease?.runId === runId && pendingRelease.finishRelease) {
|
||||
if (executionFailure !== undefined || startupUnsettled || stagedCleanupFailed) {
|
||||
pendingRelease.closeOnReleaseReason = "warm native session final cleanup failed";
|
||||
}
|
||||
await pendingRelease.finishRelease();
|
||||
}
|
||||
} finally {
|
||||
if (ownsPhysicalIngress && physicalIngressKey && executingNativeIngressAllocations.get(physicalIngressKey) === ownerToken) {
|
||||
executingNativeIngressAllocations.delete(physicalIngressKey);
|
||||
}
|
||||
if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) {
|
||||
executingNativeOwnerScopes.delete(ownerScope);
|
||||
}
|
||||
if (
|
||||
ownsSessionScope &&
|
||||
sessionScopeId !== null &&
|
||||
executingRunnerdSessionScopes.get(sessionScopeId) ===
|
||||
input.execution.binding.runId
|
||||
) {
|
||||
executingRunnerdSessionScopes.delete(sessionScopeId);
|
||||
}
|
||||
startup.resolve(null);
|
||||
startup.settle();
|
||||
if (nativeSessionStartups.get(runId) === startup) {
|
||||
nativeSessionStartups.delete(runId);
|
||||
setNativeExecutionActive(runId, false);
|
||||
}
|
||||
}
|
||||
if (startupUnsettled) throw new NativeStartupCleanupPendingError(startup.executionContext!.id,
|
||||
startup.executionContext!.deadlineAtMs, startup.leaseOwner!, startup.executionContext!.pluginSupported,
|
||||
input.runnerExecutionTarget?.kind === "remote" ? input.runnerExecutionTarget.leaseId ?? undefined : undefined);
|
||||
@@ -9270,6 +9366,9 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
}
|
||||
let retainedTransition: VerifiedWarmTransitionBinding | undefined;
|
||||
let remoteWarmRecovery: RemoteWarmSessionRecovery | undefined;
|
||||
const cachedReleasedOwner = warmNativeSessions.get(nativeSessionScopeKey(input.execution));
|
||||
if (cachedReleasedOwner?.released) requireReleasedWarmScope(input, releasedWarmRunner(readWarmNativeEnvelope(input.execution)),
|
||||
readRunnerdDurableIdentity(scopedRunnerdStateRoot(input.execution)));
|
||||
if (input.useRunnerd && !input.restartRecovery && !input.backend &&
|
||||
input.runnerExecutionTarget?.kind === "remote" &&
|
||||
input.execution.session.lifecyclePolicy.mode === "warm" &&
|
||||
@@ -9295,6 +9394,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
}
|
||||
if (durableIdentityMatchesSession(identity, input.execution) && identity.runId !== input.execution.binding.runId &&
|
||||
runnerdAuthorityLifecycleWithVerifiedBackup({ root, identity, execution: input.execution, allowVerifiedBackup: true }) !== "suspended") {
|
||||
let releasedSource: DurableRecoveryIdentity | undefined;
|
||||
remoteWarmRecovery = await inspectRemoteWarmSessionRecovery({
|
||||
db: input.db, execution: input.execution, target: input.runnerExecutionTarget, identity,
|
||||
readControlState: () => record(JSON.parse(readBoundedNativeFile(
|
||||
@@ -9302,7 +9402,13 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
"runner_durable_identity_too_large").toString("utf8"))),
|
||||
matchesExecution: previous => nativeSessionScopeKey(previous) === nativeSessionScopeKey(input.execution) &&
|
||||
nativeSessionConfigDigest(previous, "remote") === nativeSessionConfigDigest(input.execution, "remote"),
|
||||
loadCheckpoint: previous => loadWarmNativeCheckpoint(previous, nativeSessionConfigDigest(previous, "remote"), "remote"),
|
||||
loadCheckpoint: previous => {
|
||||
const proof = releasedWarmRunner(readWarmNativeEnvelope(previous));
|
||||
if (proof) {
|
||||
releasedSource = requireReleasedWarmScope(input, proof, identity);
|
||||
}
|
||||
return loadWarmNativeCheckpoint(previous, nativeSessionConfigDigest(previous, "remote"), "remote");
|
||||
},
|
||||
canReuseProcess: previous => {
|
||||
const expected = warmNativeProcessConfigDigest(input);
|
||||
if (!expected) return false;
|
||||
@@ -9311,6 +9417,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
return envelope.processConfigDigest === expected;
|
||||
},
|
||||
});
|
||||
remoteWarmRecovery.releasedSource = releasedSource;
|
||||
}
|
||||
}
|
||||
if (input.useRunnerd) {
|
||||
@@ -9517,7 +9624,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
const target = input.runnerExecutionTarget;
|
||||
const retained = warmNativeSessions.get(nativeSessionScopeKey(input.execution));
|
||||
if (!input.restartRecovery && target?.kind === "remote" && target.transport === "sandbox" && target.runnerTask && target.leaseId &&
|
||||
retained?.pluginTaskId && idleWarmNativeSessionOwnerRunId(input) !== undefined &&
|
||||
((retained?.pluginTaskId && idleWarmNativeSessionOwnerRunId(input) !== undefined) || remoteWarmRecovery?.releasedSource) &&
|
||||
(input.execution.provider.kind === "codex" || (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "claude"))) {
|
||||
claimedStartupContext = new NativeExecutionContext(input.execution, Math.min(executionLeaseTtlMs,
|
||||
coldAdmissionTimeoutMs(input.execution.provider.kind, input.execution.provider.kind === "acpx" ? input.execution.provider.agent : undefined, false)));
|
||||
@@ -10216,7 +10323,9 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
};
|
||||
if (warmSessionId !== null && warmConfigDigest !== null && !input.restartRecovery) {
|
||||
const entry = warmNativeSessions.get(warmSessionId);
|
||||
if (entry) {
|
||||
if (entry?.released) requireReleasedWarmScope(input, releasedWarmRunner(readWarmNativeEnvelope(input.execution)),
|
||||
readRunnerdDurableIdentity(scopedRunnerdStateRoot(input.execution)));
|
||||
if (entry || remoteWarmRecovery?.releasedSource) {
|
||||
if (input.execution.provider.kind === "codex" ||
|
||||
(input.execution.provider.kind === "acpx" && input.execution.provider.agent === "claude")) {
|
||||
executionContext = claimedStartupContext ?? new NativeExecutionContext(input.execution,
|
||||
@@ -10225,10 +10334,10 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
startupForAdmission!.executionContext = executionContext;
|
||||
if (startupForAdmission!.stopRequested) executionContext.abort(new Error("native_session_cancelled"));
|
||||
}
|
||||
if (entry.preparingRunId && entry.preparingRunId !== input.execution.binding.runId) throw new Error("native_session_supervisor_busy");
|
||||
entry.preparingRunId = undefined;
|
||||
if (entry?.preparingRunId && entry.preparingRunId !== input.execution.binding.runId) throw new Error("native_session_supervisor_busy");
|
||||
if (entry) entry.preparingRunId = undefined;
|
||||
const taskTarget=input.runnerExecutionTarget;
|
||||
if (entry.pluginTaskId && taskTarget?.kind === "remote" && taskTarget.transport === "sandbox" && taskTarget.runnerTask) {
|
||||
if ((entry?.pluginTaskId || remoteWarmRecovery?.releasedSource) && taskTarget?.kind === "remote" && taskTarget.transport === "sandbox" && taskTarget.runnerTask) {
|
||||
const description=environmentTaskResultSchema.parse(await (executionContext
|
||||
? executionContext.wait(() => taskTarget.runnerTask!({kind:"describe"}, { executionContext }))
|
||||
: taskTarget.runnerTask({kind:"describe"})));
|
||||
@@ -10237,6 +10346,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
if (executionContext && description.executionContextVersion === 1) executionContext.pluginSupported = true;
|
||||
else if (executionContext) await finishStartupContextForFallback();
|
||||
}
|
||||
}
|
||||
if (entry) {
|
||||
// Old run-scoped environments still require process replacement. A
|
||||
// session-owned broker can change run authority without replacing it.
|
||||
const hasBrokerCapability = Boolean(
|
||||
@@ -10301,6 +10412,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
entry.instructionWorkingCopy = input.instructionWorkingCopy?.checkpointWarm ? undefined : input.instructionWorkingCopy;
|
||||
entry.busy = true;
|
||||
entry.ownerToken = warmSessionOwnerToken;
|
||||
entry.released = false;
|
||||
entry.environmentId =
|
||||
input.runnerExecutionTarget?.environmentId ?? null;
|
||||
if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
|
||||
@@ -10329,7 +10441,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
);
|
||||
}
|
||||
}
|
||||
if ((claimedStartupContext && executionContext !== claimedStartupContext) || (executionContext && !existingWarmSession)) {
|
||||
if ((claimedStartupContext && executionContext !== claimedStartupContext) || (executionContext && !existingWarmSession && !remoteWarmRecovery?.releasedSource)) {
|
||||
await finishStartupContextForFallback();
|
||||
}
|
||||
async function resolvePendingGovernedWait() {
|
||||
@@ -10497,9 +10609,10 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
})
|
||||
: null;
|
||||
if (remoteWarmRecovery) {
|
||||
await remoteWarmRecovery.revalidate(leaseOwner);
|
||||
if (executionContext) await executionContext.wait(() => remoteWarmRecovery.revalidate(leaseOwner));
|
||||
else await remoteWarmRecovery.revalidate(leaseOwner);
|
||||
const recovered = await runnerdBackend?.recoverSession?.(remoteWarmRecovery.checkpoint, {
|
||||
signal: new AbortController().signal,
|
||||
signal: executionContext?.signal ?? new AbortController().signal,
|
||||
});
|
||||
if (!recovered?.recovered || !recovered.session) throw new Error("runner_state_identity_mismatch: remote_warm_session_unproven");
|
||||
if (remoteWarmRecovery.reuseProcess) {
|
||||
@@ -10509,6 +10622,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
persistedWarmSession = null;
|
||||
} else {
|
||||
await recovered.session.close({ reason: "recovered warm session requires current process configuration" });
|
||||
if (executionContext) await finishStartupContextForFallback();
|
||||
// Closing an externally persisted controller leaves an identity marker,
|
||||
// not its attachment template. The next backend must rotate the saved
|
||||
// PRP authority from the committed suspension checkpoint.
|
||||
@@ -10710,8 +10824,10 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
rmSync(resolve(scopedRunnerdStateRoot(input.execution), "plugin-runner-state.json"), { force: true });
|
||||
}
|
||||
if (session && session === existingWarmSession) remoteWarmRecoveryPending = false;
|
||||
if (session && input.managedAiCredentialHome) {
|
||||
managedCredentialSession = runnerdBackend?.bindManagedSession(session) ?? session;
|
||||
if (session && (input.managedAiCredentialHome || (input.runnerExecutionTarget?.kind === "remote"
|
||||
&& input.runnerExecutionTarget.transport === "sandbox" && input.runnerExecutionTarget.runnerTask))) {
|
||||
const bound = runnerdBackend?.bindManagedSession(session, input.managedAiCredentialHome ? undefined : { releaseResumeOnly: true }) ?? session;
|
||||
if (input.managedAiCredentialHome) managedCredentialSession = bound;
|
||||
}
|
||||
releaseRegisteredGoalController();
|
||||
liveQuestions.close();
|
||||
@@ -11662,11 +11778,101 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
if (collectInstructions && ownedSession) {
|
||||
ownedSession.closeOnReleaseReason = "registered instruction edits require stopped-provider collection";
|
||||
}
|
||||
await timeController("warm_release", () => releaseWarmNativeSession(
|
||||
warmSessionId,
|
||||
warmSessionOwnerToken,
|
||||
lifecyclePolicy.idleTimeoutMs,
|
||||
!retainSession,
|
||||
if (retainSession && ownedSession?.ownerToken === warmSessionOwnerToken && !ownedSession.closeOnReleaseReason
|
||||
&& pluginTaskSessionOwners.get(ownedSession.session)?.driver.releaseResume) {
|
||||
const scope = warmSessionId, entry = ownedSession, token = warmSessionOwnerToken;
|
||||
entry.finishRelease = async () => {
|
||||
const current = () => warmNativeSessions.get(scope) === entry && entry.ownerToken === token;
|
||||
if (!current()) return;
|
||||
try {
|
||||
if (nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) entry.closeOnReleaseReason = "warm native session cancelled";
|
||||
if (entry.closeOnReleaseReason) {
|
||||
await closeWarmNativeSession(entry, entry.closeOnReleaseReason);
|
||||
forgetWarmNativeSession(scope, entry);
|
||||
return;
|
||||
}
|
||||
const deadline = Date.now() + Math.min(lifecyclePolicy.idleTimeoutMs,
|
||||
coldAdmissionTimeoutMs(input.execution.provider.kind,
|
||||
input.execution.provider.kind === "acpx" ? input.execution.provider.agent : undefined, false));
|
||||
let released: DurableRecoveryIdentity | false | undefined;
|
||||
for (;;) {
|
||||
let gateTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
const gate = new Promise<never>((_resolve, reject) => {
|
||||
const check = () => {
|
||||
if (!current() || Date.now() >= deadline || entry.closeOnReleaseReason
|
||||
|| nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) {
|
||||
reject(new Error("native_session_release_reconciliation_stopped"));
|
||||
} else gateTimer = setTimeout(check, Math.min(25, deadline - Date.now()));
|
||||
};
|
||||
check();
|
||||
});
|
||||
// Observe late replies, but only this reservation may publish their result.
|
||||
released = await Promise.race([Promise.resolve().then(() => {
|
||||
if (!current() || Date.now() >= deadline || entry.closeOnReleaseReason
|
||||
|| nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) {
|
||||
throw new Error("native_session_release_reconciliation_stopped");
|
||||
}
|
||||
return entry.session.release?.();
|
||||
}), gate]);
|
||||
break;
|
||||
}
|
||||
catch {
|
||||
if (!current()) return;
|
||||
if (Date.now() >= deadline || entry.closeOnReleaseReason
|
||||
|| nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) {
|
||||
entry.closeOnReleaseReason ??= "warm Runner release reconciliation expired";
|
||||
await closeWarmNativeSession(entry, entry.closeOnReleaseReason);
|
||||
forgetWarmNativeSession(scope, entry);
|
||||
return;
|
||||
}
|
||||
await new Promise(resolve => setTimeout(resolve, 250));
|
||||
}
|
||||
finally { clearTimeout(gateTimer); }
|
||||
}
|
||||
if (!current()) return;
|
||||
if (nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) entry.closeOnReleaseReason = "warm native session cancelled";
|
||||
if (entry.closeOnReleaseReason) {
|
||||
await closeWarmNativeSession(entry, entry.closeOnReleaseReason);
|
||||
forgetWarmNativeSession(scope, entry);
|
||||
return;
|
||||
}
|
||||
if (released !== false && released !== undefined) {
|
||||
try {
|
||||
const operation = environmentTaskOperationSchema.parse({ kind: "release", identity: released });
|
||||
const target = input.runnerExecutionTarget;
|
||||
if (operation.kind !== "release" || operation.identity.runId !== input.execution.binding.runId
|
||||
|| operation.identity.normalizedSessionId !== nativeSessionKey(input.execution)
|
||||
|| operation.identity.runnerInstanceId !== effectiveRunnerInstanceId
|
||||
|| target?.kind !== "remote" || operation.identity.environmentLeaseId !== (durableRunnerBinding?.environmentLeaseId ?? target.leaseId)
|
||||
|| target.transport !== "sandbox" || !entry.pluginTaskId
|
||||
|| entry.pluginTaskId !== target.sandboxLeaseAcquisition?.providerLeaseId) {
|
||||
throw new Error("native_session_released_source_mismatch");
|
||||
}
|
||||
const source = operation.identity;
|
||||
updateReleasedWarmRunner(input.execution, source, entry.pluginTaskId,
|
||||
{ source, taskId: entry.pluginTaskId, scopeDigest: releasedWarmScope(input) });
|
||||
} catch {
|
||||
entry.closeOnReleaseReason = "released Runner source proof could not be verified or saved";
|
||||
await closeWarmNativeSession(entry, entry.closeOnReleaseReason);
|
||||
forgetWarmNativeSession(scope, entry);
|
||||
return;
|
||||
}
|
||||
entry.released = true;
|
||||
}
|
||||
} catch {
|
||||
// The business turn already settled. An unknown release stays owned and active.
|
||||
logger.warn({ runId: input.execution.binding.runId }, "Retained Runner release remains unconfirmed");
|
||||
} finally {
|
||||
if (current()) {
|
||||
delete entry.finishRelease;
|
||||
entry.busy = false;
|
||||
armWarmNativeSessionIdleTimer(scope, entry, lifecyclePolicy.idleTimeoutMs);
|
||||
}
|
||||
}
|
||||
};
|
||||
} else await timeController("warm_release", () => releaseWarmNativeSession(
|
||||
warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, !retainSession,
|
||||
));
|
||||
if (collectInstructions && !collectedByOwner) await timeController("instruction_collection", () => instructionCopy!.collectStopped());
|
||||
}
|
||||
@@ -13741,7 +13947,7 @@ async function prepareRemoteExternalRunner({ input, target, runner, root, identi
|
||||
}
|
||||
|
||||
type ManagedRunnerdBackend = NativeSessionBackend & {
|
||||
bindManagedSession(session: NativeSession): NativeSession;
|
||||
bindManagedSession(session: NativeSession, options?: { releaseResumeOnly: true }): NativeSession;
|
||||
bindDiagnosticSession?(session: NativeSession): void;
|
||||
sessionCheckpointPersisted?(snapshot: PersistedNativeSession): void;
|
||||
providerAdmissionPersisted?(): Promise<void>;
|
||||
@@ -13941,8 +14147,11 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
} : undefined,
|
||||
onRecoveryAccountingEvent: input.onAccountingEvent ? async (event: PrpEvent, source: Parameters<NonNullable<typeof recoveryUsage>>[1]) => {
|
||||
const current = sessionToolAuthorityEpochs.get(sessionScopeId);
|
||||
if (!current || !recoveryUsage) throw new Error("native_recovery_usage_source_unproven");
|
||||
await current.acceptRecoveryAccountingEvent(event, () => recoveryUsage(event, source));
|
||||
if (!current) throw new Error("native_recovery_usage_source_unproven");
|
||||
await current.acceptRecoveryAccountingEvent(event, () => {
|
||||
if (!recoveryUsage) throw new Error("native_recovery_usage_source_unproven");
|
||||
return recoveryUsage(event, source);
|
||||
});
|
||||
} : undefined,
|
||||
};
|
||||
const effectiveRunnerInstanceId =
|
||||
@@ -14003,6 +14212,10 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
onCredentials: input.execution.session.lifecyclePolicy.mode === "warm" ? undefined : credentials.onCredentials,
|
||||
warmTaskVersion: description.warmTaskVersion,
|
||||
releaseResumeVersion: description.releaseResumeVersion,
|
||||
authorityRunId: description.releaseResumeVersion === 1 ? input.execution.binding.runId : undefined,
|
||||
releasedSource: input.remoteWarmRecovery?.releasedSource,
|
||||
onResumed: async source => updateReleasedWarmRunner(input.execution, source, description.taskId, null),
|
||||
onRetired: async source => updateReleasedWarmRunner(input.execution, source, description.taskId, null),
|
||||
adoptIdentity: adoptionIdentity ?? (adopting && prior ? {
|
||||
runnerInstanceId: prior.runnerInstanceId, environmentLeaseId: prior.environmentLeaseId,
|
||||
runId: prior.runId, normalizedSessionId: prior.normalizedSessionId,
|
||||
@@ -14031,6 +14244,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
});
|
||||
}, onDispatched: input.onTaskDispatched,
|
||||
};
|
||||
if (input.remoteWarmRecovery?.releasedSource && description.releaseResumeVersion !== 1) throw new Error("Plugin Runner release/resume unavailable");
|
||||
if (input.execution.session.lifecyclePolicy.mode === "warm" && (!credentials.accountIdentity || description.warmTaskVersion !== 1)) throw new Error("Plugin warm task capability or account unavailable");
|
||||
const owner = retainedOwner ?? createPluginRunnerTaskOwner(ownerInput);
|
||||
if (retainedOwner) retainedOwner.rebind(ownerInput);
|
||||
@@ -14107,7 +14321,10 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
sessionToolAuthorityEpochs.set(sessionScopeId, authorityEpoch);
|
||||
return { ...backend, descriptor: () => backend.descriptor(), openSession: session => backend.openSession(session),
|
||||
recoverSession: (snapshot, options) => backend.recoverSession!(snapshot, options),
|
||||
bindManagedSession: session => { pluginTaskSessionOwners.set(session, owner); return session; } };
|
||||
bindManagedSession: (session, options) => {
|
||||
if (!options?.releaseResumeOnly || description.releaseResumeVersion === 1) pluginTaskSessionOwners.set(session, owner);
|
||||
return session;
|
||||
} };
|
||||
}
|
||||
if (input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8") {
|
||||
if (target.kind === "remote" && input.dotWorkspaceRoot) throw new Error("dot_remote_workspace_access_unavailable");
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdir, readFile, readdir, writeFile } from "node:fs/promises";
|
||||
import https, { createServer } from "node:https";
|
||||
import { createServer as createTcpServer } from "node:net";
|
||||
import { join, resolve } from "node:path";
|
||||
import { getCACertificates, setDefaultCACertificates } from "node:tls";
|
||||
import { eq } from "drizzle-orm";
|
||||
import WebSocket, { WebSocketServer, type RawData } from "ws";
|
||||
import { expect, it, vi } from "vitest";
|
||||
import { completionContracts, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db";
|
||||
import { environmentTaskOperationSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk";
|
||||
import { readRunnerdArtifactBinding, defaultCapabilityRunnerdBinary,
|
||||
parseNativeExecutionInput, parseNativeRuntimeContext, QUALIFIED_ACPX_PROFILES, type DurableRecoveryIdentity, type RunnerProcessHandle, type RunnerProcessLaunchSpec } from "../../vendor/paperclip-runner/index.js";
|
||||
import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, canonicalNativeRuntimeContextDigest } from "@paperclipai/paperclip-runner";
|
||||
import { materializeAsset } from "./runtime-context.js";
|
||||
import { spawnRunner } from "../../vendor/paperclip-runner/testing.js";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture, realProcessIt, fakeCodexAppServer, waitForCondition } from "../../__tests__/helpers/native-restart.js";
|
||||
import { executePaperclipNativeSession, cancelNativeSession, closeIdleWarmNativeSessionsForRestart, readNativeSessionIdleWork } from "./native-session-executor.js";
|
||||
import { resolvePaperclipInstanceRoot } from "../../home-paths.js";
|
||||
import * as remoteWarmRecovery from "./remote-warm-session-recovery.js";
|
||||
import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js";
|
||||
import { buildNativeCompletionContract } from "./completion-contracts.js";
|
||||
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
||||
|
||||
const { createPrivateHostControlFixture } = await import(new URL("../../../../packages/paperclip-runner/test/support/private-host-control.ts", import.meta.url).href) as {
|
||||
createPrivateHostControlFixture(): {
|
||||
launch(spec: RunnerProcessLaunchSpec): RunnerProcessHandle;
|
||||
request(frame: Record<string, unknown>): Promise<{ id: string; status: string }>;
|
||||
close(): Promise<void>;
|
||||
};
|
||||
};
|
||||
|
||||
const fakeAcpxSidecar = resolve(import.meta.dirname, "../../../../packages/paperclip-runner/runner/target/debug/fake-acpx-sidecar");
|
||||
const sha = (value: string) => createHash("sha256").update(value).digest("hex");
|
||||
const parkId = (source: DurableRecoveryIdentity) => `park_${sha(JSON.stringify(Object.fromEntries(Object.entries(source).sort(([a], [b]) => a.localeCompare(b)))))}`;
|
||||
|
||||
// The only replaced service is the external environment-task/Launcher edge.
|
||||
// Runtime, backend, owner, PRP ingress, checkpoint persistence and native Runner are real.
|
||||
describeEmbeddedPostgres("plugin Runner release/resume through the Core execution entry", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
const nativeIt = process.platform === "win32" ? it.skip : realProcessIt;
|
||||
nativeIt.for([...(["codex", "claude"] as const).flatMap(harness => (["cached", "adopted"] as const).map(mode => [harness, mode] as const)), ["codex", "adopted cancellation"] as const, ["codex", "adopted legacy cancellation"] as const])("completes %s runs through a retained %s process", { timeout: 60_000 }, async ([harness, mode], context) => {
|
||||
if (harness === "claude" && !existsSync(fakeAcpxSidecar)) context.skip();
|
||||
const adopted = mode.startsWith("adopted"), cancellation = mode.includes("cancellation"), qualifiedCancellation = mode === "adopted cancellation";
|
||||
const first = await environment.seedRun(`RELEASE-ENTRY-${harness}-${mode}`);
|
||||
const { db, runId, issueId, native } = first;
|
||||
const { companyId, agentId, runtimeRoot } = environment.current;
|
||||
const root = join(runtimeRoot, runId), workspace = join(root, "workspace"), home = join(root, "credential-home");
|
||||
await mkdir(workspace, { recursive: true }); await mkdir(home, { recursive: true });
|
||||
await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 });
|
||||
const workspaceId = randomUUID(), environmentId = randomUUID(), leaseA = randomUUID(), taskId = randomUUID();
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId,
|
||||
mode: "isolated", strategyType: "directory", name: "Release fixture", cwd: workspace });
|
||||
await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "plugin" });
|
||||
await db.insert(environmentLeases).values({ id: leaseA, companyId, environmentId, heartbeatRunId: runId, issueId,
|
||||
executionWorkspaceId: workspaceId, provider: "plugin:fixture:tasks", providerLeaseId: taskId, status: "active", leasePolicy: "reuse_by_environment", metadata: { remoteCwd: workspace } });
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId));
|
||||
const runtimeContext = nativeRuntimeContextFixture();
|
||||
runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444, content: Buffer.from("Complete only the controlled task.\n") }]);
|
||||
runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext);
|
||||
const { permissionPolicy: _permissionPolicy, ...claudeProfile } = QUALIFIED_ACPX_PROFILES.claude;
|
||||
const execution = parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null,
|
||||
runtimeContext, binding: { companyId, agentId, issueId, runId, executionWorkspaceId: workspaceId },
|
||||
provider: harness === "codex" ? { kind: "codex", model: "fixture-model", approvalPolicy: "never" }
|
||||
: { kind: "acpx", agent: "claude", model: "claude-sonnet-4-6", permissionMode: "approve-all", profile: claudeProfile },
|
||||
task: { identifier: "RELEASE-ENTRY", title: issue!.title, description: null, prompt: "Complete the controlled task.", workMode: "standard" },
|
||||
workspace: { cwd: workspace, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: native.normalizedSessionId, driverKind: harness === "codex" ? "codex_app_server" : "acpx_runtime", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "warm", idleTimeoutMs: adopted ? 5_000 : 60_000 } },
|
||||
completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, interactionResponses: [], credentialBindings: [] });
|
||||
const resultFile = join(root, "result.json"), calls = join(root, "provider-calls");
|
||||
await writeFile(resultFile, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Controlled task complete.",
|
||||
completionClaim: { contractRevision: execution.completionContract.contract.revision, objectiveSatisfied: true,
|
||||
criteria: execution.completionContract.contract.criteria.map(c => ({ criterionId: c.id, status: "satisfied", evidenceRefs: [] })), remainingWork: [] },
|
||||
evidence: [], verification: [], artifacts: [], attentionRequests: [] }));
|
||||
const sidecar = join(root, "sidecar"), sidecarScript = join(root, "sidecar-entry");
|
||||
if (harness === "claude") {
|
||||
const quote = (value: string) => `'${value.replaceAll("'", "'\"'\"'")}'`;
|
||||
await writeFile(sidecarScript, "qualified fixture entry");
|
||||
await writeFile(sidecar, `#!/bin/sh\nexec ${quote(fakeAcpxSidecar)} --mode source-reconcile --profile-digest ${quote(claudeProfile.commandDigest)} --journal ${quote(calls)} --source-fault ${quote(join(root, "fault"))} --wait-completion-feedback\n`, { mode: 0o700 });
|
||||
}
|
||||
const key = join(root, "tls.key"), certificate = join(root, "tls.crt");
|
||||
execFileSync("openssl", ["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key, "-out", certificate,
|
||||
"-days", "1", "-subj", "/CN=localhost", "-addext", "subjectAltName=IP:127.0.0.1,DNS:localhost"], { stdio: "ignore" });
|
||||
const oldCAs = getCACertificates("default"), oldAgent = https.globalAgent;
|
||||
const certificates = [...oldCAs, await readFile(certificate, "utf8")];
|
||||
const probe = createTcpServer(); await new Promise<void>(done => probe.listen(0, "127.0.0.1", done));
|
||||
const address = probe.address(); if (!address || typeof address === "string") throw new Error("Fixture listener unavailable");
|
||||
const runnerPort = address.port; await new Promise<void>(done => probe.close(() => done()));
|
||||
const tls = createServer({ key: await readFile(key), cert: await readFile(certificate) });
|
||||
const sockets = new Set<WebSocket>(), websocket = new WebSocketServer({ noServer: true });
|
||||
let listenPath = "";
|
||||
const wireEvents: string[] = [];
|
||||
tls.on("upgrade", (request, socket, head) => {
|
||||
if (request.headers["x-fixture-token"] !== "synthetic-ingress") { socket.destroy(); return; }
|
||||
websocket.handleUpgrade(request, socket, head, front => {
|
||||
const back = new WebSocket(`ws://127.0.0.1:${runnerPort}${listenPath}`, { perMessageDeflate: false });
|
||||
const pending: Array<{ data: RawData; binary: boolean }> = []; sockets.add(front); sockets.add(back);
|
||||
front.on("message", (data, binary) => {
|
||||
if (!binary) { const frame = JSON.parse(data.toString()); wireEvents.push(`core:${frame.kind ?? frame.type}:${frame.payload?.type ?? frame.payload?.eventType ?? frame.error?.code ?? ""}`); }
|
||||
if (back.readyState === WebSocket.OPEN) back.send(data, { binary }); else pending.push({ data, binary });
|
||||
});
|
||||
back.on("open", () => { for (const message of pending.splice(0)) back.send(message.data, { binary: message.binary }); });
|
||||
back.on("message", (data, binary) => {
|
||||
if (!binary) { const frame = JSON.parse(data.toString()); wireEvents.push(`runner:${frame.kind ?? frame.type}:${frame.payload?.type ?? frame.payload?.eventType ?? frame.error?.code ?? ""}`); }
|
||||
if (front.readyState === WebSocket.OPEN) front.send(data, { binary });
|
||||
});
|
||||
front.on("close", () => { sockets.delete(front); back.terminate(); }); back.on("close", () => { sockets.delete(back); front.terminate(); });
|
||||
front.on("error", () => back.terminate()); back.on("error", () => front.terminate());
|
||||
});
|
||||
});
|
||||
await new Promise<void>(done => tls.listen(0, "127.0.0.1", done));
|
||||
const tlsAddress = tls.address(); if (!tlsAddress || typeof tlsAddress === "string") throw new Error("Fixture TLS unavailable");
|
||||
const host = createPrivateHostControlFixture(), runnerRoot = join(root, "native"), binary = defaultCapabilityRunnerdBinary();
|
||||
const artifact = readRunnerdArtifactBinding(binary), preparations = new Map<string, { operation: Extract<PluginEnvironmentTaskOperation, { kind: "prepare" }>; launch: { configurationId: string } }>();
|
||||
const releases: DurableRecoveryIdentity[] = [], claims: DurableRecoveryIdentity[] = [];
|
||||
let handle: RunnerProcessHandle | undefined, submissions = 0, stops = 0, completes = 0;
|
||||
let releaseClaimReply!: () => void, observeHeldClaim!: () => void;
|
||||
const claimReply = new Promise<void>(done => { releaseClaimReply = done; });
|
||||
const heldClaim = new Promise<void>(done => { observeHeldClaim = done; });
|
||||
let pendingFollowup: Promise<unknown> | undefined;
|
||||
const providerStarts = async () => (await readFile(calls, "utf8")).trim().split(/\r?\n/)
|
||||
.filter(line => harness === "codex" ? line === "process-start" : JSON.parse(line).processStart).length;
|
||||
const state = async () => JSON.parse(await readFile(join(runnerRoot, "runner-state.json"), "utf8"));
|
||||
const sameSource = (a: DurableRecoveryIdentity, b: DurableRecoveryIdentity) => Object.entries(a).every(([key, value]) => b[key as keyof DurableRecoveryIdentity] === value);
|
||||
const task = async (raw: unknown, authorityRunId: string): Promise<unknown> => {
|
||||
const operation = environmentTaskOperationSchema.parse(raw);
|
||||
if (operation.kind === "describe") return { kind: "description", taskId, launchPreparationVersion: 1,
|
||||
projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact, ...(qualifiedCancellation ? { executionContextVersion: 1 } : {}) };
|
||||
if (operation.kind === "prepare") {
|
||||
expect(operation.requirements.identity.runId).toBe(authorityRunId);
|
||||
expect(operation.accountIdentity).toBe("fixture-account");
|
||||
if (operation.priorIdentity) expect(sameSource(operation.priorIdentity, await state())).toBe(true);
|
||||
expect(operation.requirements.settings.managedProfile).toBeUndefined();
|
||||
expect(operation.requirements.settings.agentCoreProfile).toBeUndefined();
|
||||
expect(operation.requirements.settings.acpxAgent).toBe(harness === "claude" ? "claude" : undefined);
|
||||
const requirements = { ...operation.requirements, runtimeContext: operation.requirements.runtimeContext ? parseNativeRuntimeContext(operation.requirements.runtimeContext) : null, settings: { ...operation.requirements.settings, acpxAgent: harness === "claude" ? "claude" as const : undefined, managedProfile: undefined, agentCoreProfile: undefined } };
|
||||
const launch = { configurationId: `configuration-${operation.requirements.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(requirements),
|
||||
provider: createRunnerProviderLaunch(requirements, { options: { codexArgs: ["--state-file", join(root, "provider.json"), "--terminal-result-file", resultFile,
|
||||
"--call-log", calls, "--record-process-start"] }, root, cwd: workspace, codexHome: home, codexCommand: fakeCodexAppServer,
|
||||
providerNodeCommand: harness === "claude" ? sidecar : process.execPath, opencodeProxyPath: "", acpxSidecarPath: harness === "claude" ? sidecarScript : "" }) };
|
||||
preparations.set(launch.configurationId, { operation, launch }); return { kind: "prepared", taskId, launch };
|
||||
}
|
||||
if (operation.kind === "check_project_auth") return { kind: "project_auth", taskId, configurationId: operation.configurationId, provider: operation.provider, allowed: true };
|
||||
if (operation.kind === "submit") {
|
||||
if (!operation.preparedLaunchId || !operation.launch) throw new Error("Fixture structured launch unavailable");
|
||||
const prepared = preparations.get(operation.preparedLaunchId)!; submissions++;
|
||||
expect(operation.runner).toMatchObject({ runnerId: prepared.operation.requirements.identity.runnerInstanceId,
|
||||
leaseId: prepared.operation.requirements.identity.environmentLeaseId, runId: authorityRunId,
|
||||
sessionId: prepared.operation.requirements.identity.normalizedSessionId, turnId: prepared.operation.requirements.identity.turnId,
|
||||
itemId: prepared.operation.requirements.identity.itemId });
|
||||
listenPath = `/api/runner/v1/connect/${operation.runner.runId}`;
|
||||
handle = spawnRunner({ ...operation.launch, stateDirectory: runnerRoot, identity: prepared.operation.requirements.identity, ticket: operation.bootstrapTicket,
|
||||
runnerBinaryPath: binary, maxOutboxBytes: operation.launch.maxOutboxBytes, p0ReserveBytes: operation.launch.p0ReserveBytes,
|
||||
connection: { mode: "listen", listenAddress: "0.0.0.0", listenPort: runnerPort, listenPath }, processLauncher: host.launch,
|
||||
environment: { PATH: process.env.PATH },
|
||||
...(harness === "claude" ? { acpxLaunchProfile: { authorityDigest: `sha256:${sha("fixture authority")}`, command: sidecar,
|
||||
commandSha256: `sha256:${sha(await readFile(sidecar, "utf8"))}`, sidecarScript,
|
||||
sidecarScriptSha256: `sha256:${sha(await readFile(sidecarScript, "utf8"))}` } } : {}) });
|
||||
return { kind: "accepted", taskId };
|
||||
}
|
||||
if (operation.kind === "connection") return { kind: "connection", taskId, endpoint: { kind: "authenticated_websocket",
|
||||
websocketUrl: `wss://127.0.0.1:${tlsAddress.port}/runner`, secretHeaders: [{ name: "X-Fixture-Token", value: "synthetic-ingress" }], generation: "fixture" } };
|
||||
if (operation.kind === "status") return { kind: "status", taskId, phase: handle?.isExitConfirmed?.() ? "cancelled" : "running", executionStopped: handle?.isExitConfirmed?.() ?? false };
|
||||
if (operation.kind === "read_state") {
|
||||
const live = await state();
|
||||
const previous = [...preparations.values()].some(({ operation: prepared }) => prepared.priorIdentity && sameSource(prepared.priorIdentity, operation.identity));
|
||||
expect(sameSource(operation.identity, live) || previous).toBe(true);
|
||||
return { kind: "state", taskId, state: live };
|
||||
}
|
||||
if (operation.kind === "release") {
|
||||
const live = await state(); const reply = await host.request({ op: "park", id: parkId(operation.identity), identity: operation.identity,
|
||||
lastControllerCommandSeq: live.lastControllerCommandSeq, ackedSourceSeq: live.ackedSourceSeq });
|
||||
if (reply.status !== "parked") throw new Error("Fixture native release refused");
|
||||
releases.push(operation.identity); return { kind: "released", taskId };
|
||||
}
|
||||
if (operation.kind === "resume") {
|
||||
const retainedProviderCount = await providerStarts();
|
||||
const reply = await host.request({ op: "claim", id: `claim_${sha(operation.bootstrapTicket)}`, parkId: parkId(operation.identity), bootstrapTicket: operation.bootstrapTicket });
|
||||
if (reply.status !== "claimed") throw new Error("Fixture native claim refused");
|
||||
expect(await providerStarts()).toBe(retainedProviderCount);
|
||||
claims.push(operation.identity);
|
||||
if (cancellation) { observeHeldClaim(); await claimReply; }
|
||||
return { kind: "resume", taskId, outcome: "resumed" };
|
||||
}
|
||||
if (operation.kind === "stop") { stops++; await host.close(); }
|
||||
if (operation.kind === "complete") { expect(handle?.isExitConfirmed?.()).toBe(true); completes++; }
|
||||
return { kind: "accepted", taskId };
|
||||
};
|
||||
const supervisor = async () => {
|
||||
const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions");
|
||||
for (const file of await readdir(directory)) {
|
||||
const envelope = JSON.parse(await readFile(join(directory, file), "utf8"));
|
||||
if (envelope.snapshot?.identity.sessionId === native.normalizedSessionId) return envelope;
|
||||
}
|
||||
throw new Error("Fixture supervisor record unavailable");
|
||||
};
|
||||
const proof = async () => (await supervisor()).releasedRunner;
|
||||
const execute = async (current: typeof execution, leaseId: string) => {
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: current }, runnerInstanceId: native.runnerInstanceId, completionContractId: contract!.id, completionContractSha256: contract!.canonicalSha256, nativeSessionId: current.session.normalizedSessionId })
|
||||
.where(eq(heartbeatRuns.id, current.binding.runId));
|
||||
return executePaperclipNativeSession({ db, execution: current, runnerInstanceId: native.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerEnvironment: harness === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {},
|
||||
managedAiCredentialHome: home, managedAiCredentialIdentity: "fixture-revision", managedAiAccountIdentity: "fixture-account", turnTimeoutMs: 15_000,
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: workspace, environmentId, leaseId, providerKey: "plugin:fixture:tasks", runnerTask: operation => task(operation, current.binding.runId),
|
||||
sandboxLeaseAcquisition: { outcome: leaseId === leaseA ? "created" : "resumed", providerLeaseId: taskId } } });
|
||||
};
|
||||
const adoption = vi.spyOn(remoteWarmRecovery, "inspectRemoteWarmSessionRecovery");
|
||||
try {
|
||||
setDefaultCACertificates(certificates); https.globalAgent = new https.Agent({ ca: certificates });
|
||||
const a = await execute(execution, leaseA); expect(a).toMatchObject({ exitCode: 0, timedOut: false });
|
||||
await waitForCondition("automatic release A", () => releases.length === 1);
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId)))[0]).toMatchObject({ schemaStatus: "accepted", resultJson: { result: { reportedWorkDisposition: "done" } } });
|
||||
expect(await proof()).toMatchObject({ source: releases[0], taskId });
|
||||
const providerSessionId = (await supervisor()).snapshot.providerSessionId;
|
||||
expect(providerSessionId).toEqual(expect.any(String));
|
||||
const pid = handle!.child.pid;
|
||||
const { heartbeatService } = await import("../heartbeat.js");
|
||||
expect(await heartbeatService(db).prepareIdleWarmNativeSessionsForSleep({ canRetire: () => true }))
|
||||
.toEqual({ closed: 0, busy: 0, failed: 0 });
|
||||
expect(stops).toBe(0); expect(handle!.isExitConfirmed?.()).toBe(false);
|
||||
if (adopted) {
|
||||
await new Promise(done => setTimeout(done, 5_200));
|
||||
expect(stops).toBe(0); expect(handle!.isExitConfirmed?.()).toBe(false);
|
||||
expect(await proof()).toMatchObject({ source: releases[0], taskId });
|
||||
}
|
||||
expect(readNativeSessionIdleWork()).toBe("none");
|
||||
await reconcileNativeFinalizations(db, [runId]);
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]).toMatchObject({ phase: "committed", leaseOwner: null });
|
||||
const admitNext = async () => {
|
||||
const runB = randomUUID(), leaseB = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runB, companyId, agentId, nativeIssueId: issueId, status: "running", runtimeMode: "native" });
|
||||
await db.insert(nativeRunFinalizations).values({ runId: runB, companyId, issueId, phase: "observed" });
|
||||
await db.update(issues).set({ executionRunId: runB, status: "in_progress" }).where(eq(issues.id, issueId));
|
||||
await db.insert(environmentLeases).values({ id: leaseB, companyId, environmentId, heartbeatRunId: runB, issueId, executionWorkspaceId: workspaceId,
|
||||
provider: "plugin:fixture:tasks", providerLeaseId: taskId, status: "active", leasePolicy: "reuse_by_environment", metadata: { remoteCwd: workspace } });
|
||||
const next = { ...execution, binding: { ...execution.binding, runId: runB } };
|
||||
pendingFollowup = execute(next, leaseB);
|
||||
if (cancellation) {
|
||||
void pendingFollowup.catch(() => undefined);
|
||||
await heldClaim;
|
||||
const competitor = { ...next, binding: { ...next.binding, runId: randomUUID() } };
|
||||
await expect(execute(competitor, leaseB)).rejects.toThrow("native_session_supervisor_busy");
|
||||
const stopping = cancelNativeSession(runB, "Stop retained adoption", { db });
|
||||
void stopping.catch(() => undefined);
|
||||
await waitForCondition("durable cancellation intent", async () => Boolean((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runB)))[0]?.resultJson?.nativeCancellation));
|
||||
if (qualifiedCancellation) await waitForCondition("Core-triggered native stop", () => stops > 0 && handle!.isExitConfirmed!());
|
||||
else await expect(execute(competitor, leaseB)).rejects.toThrow("native_session_supervisor_busy");
|
||||
expect(claims).toEqual([releases[0]]);
|
||||
expect(releases).toHaveLength(1);
|
||||
releaseClaimReply();
|
||||
expect(await pendingFollowup.catch(error => error)).toBeInstanceOf(Error);
|
||||
await stopping.catch(() => undefined);
|
||||
expect(completes).toBeGreaterThan(0);
|
||||
expect(handle!.isExitConfirmed!()).toBe(true);
|
||||
expect((await readFile(calls, "utf8")).split(/\r?\n/).filter(method => method === "turn/start")).toHaveLength(1);
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runB)))).toHaveLength(0);
|
||||
expect(releases).toHaveLength(1); expect((await proof())?.source.runId).not.toBe(runB);
|
||||
return runB;
|
||||
}
|
||||
expect(await pendingFollowup).toMatchObject({ exitCode: 0, timedOut: false });
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runB)))[0]).toMatchObject({ schemaStatus: "accepted", resultJson: { result: { reportedWorkDisposition: "done" } } });
|
||||
await reconcileNativeFinalizations(db, [runB]);
|
||||
return runB;
|
||||
};
|
||||
const runB = await admitNext();
|
||||
if (cancellation) return;
|
||||
await waitForCondition("automatic release B", () => releases.length === 2);
|
||||
expect(await proof()).toMatchObject({ source: releases[1], taskId });
|
||||
expect((await supervisor()).snapshot.providerSessionId).toBe(providerSessionId);
|
||||
expect(adoption).toHaveBeenCalledTimes(adopted ? 1 : 0);
|
||||
if (adopted) expect(adoption.mock.calls[0]![0]).toMatchObject({ identity: releases[0], execution: { binding: { runId: runB } } });
|
||||
if (adopted) {
|
||||
const runC = await admitNext();
|
||||
await waitForCondition("automatic release C", () => releases.length === 3);
|
||||
expect(releases[2]!.runId).toBe(runC); expect(await proof()).toMatchObject({ source: releases[2], taskId });
|
||||
expect((await supervisor()).snapshot.providerSessionId).toBe(providerSessionId);
|
||||
expect(adoption).toHaveBeenCalledTimes(1);
|
||||
}
|
||||
expect(claims).toEqual(releases.slice(0, -1)); expect(releases[1]!.runId).toBe(runB);
|
||||
expect(releases[1]!.environmentLeaseId).toBe(leaseA); expect(handle!.child.pid).toBe(pid); expect(submissions).toBe(1);
|
||||
const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/);
|
||||
if (harness === "codex") {
|
||||
expect(methods.filter(method => method === "process-start")).toHaveLength(1);
|
||||
expect(methods.filter(method => method === "turn/start")).toHaveLength(adopted ? 3 : 2);
|
||||
} else {
|
||||
const journal = methods.map(line => JSON.parse(line));
|
||||
expect(journal.filter(entry => entry.processStart)).toHaveLength(adopted ? 3 : 2);
|
||||
expect(journal.filter(entry => entry.command === "turn.start")).toHaveLength(adopted ? 3 : 2);
|
||||
}
|
||||
} catch (error) {
|
||||
const live = await state().catch(() => ({}));
|
||||
console.info("entry-fixture", { releases: releases.length, claims: claims.length, lifecycle: live.lifecycle,
|
||||
lastCommand: live.lastControllerCommandSeq, ackedSource: live.ackedSourceSeq, outbox: live.outbox?.map((e: { envelope?: { payload?: { eventType?: string } } }) => e.envelope?.payload?.eventType), wire: wireEvents.slice(-70) });
|
||||
throw error;
|
||||
} finally {
|
||||
releaseClaimReply();
|
||||
await pendingFollowup?.catch(() => undefined);
|
||||
try {
|
||||
const cleanup = await Promise.allSettled([closeIdleWarmNativeSessionsForRestart(), host.close()]);
|
||||
for (const socket of sockets) socket.terminate(); websocket.close();
|
||||
await new Promise<void>(done => tls.close(() => done()));
|
||||
const failures = cleanup.filter((result): result is PromiseRejectedResult => result.status === "rejected");
|
||||
if (failures.length) throw new AggregateError(failures.map(result => result.reason), "Fixture cleanup failed");
|
||||
} finally {
|
||||
adoption.mockRestore();
|
||||
https.globalAgent.destroy(); https.globalAgent = oldAgent; setDefaultCACertificates(oldCAs);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -1,7 +1,7 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js";
|
||||
import { currentNativeControllerIdentity, evaluateNativeControllerTakeover } from "./native-restart-recovery.js";
|
||||
import { NativeControllerRecoveryWaitingError, readNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
|
||||
@@ -10,6 +10,7 @@ export type RemoteWarmSessionRecovery = {
|
||||
checkpoint: PersistedNativeSession;
|
||||
previousExecution: NativeExecutionInput;
|
||||
reuseProcess: boolean;
|
||||
releasedSource?: DurableRecoveryIdentity;
|
||||
remote: { providerLeaseId: string; remoteCwd: string };
|
||||
revalidate: (leaseOwner: string) => Promise<void>;
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user