mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
Continue a released Runner after confirmed graceful eviction (#19760)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Core can release a completed Runner while its host retains the session. > - A host can later retire that Runner under its retention policy. > - Graceful retirement saves a suspended session before physical shutdown. > - A cached Core owner still expects the old physical task and rejects a new allocation. > - This pull request retires that dead local controller after verified stop. > - The benefit is ordinary session continuation for Codex and qualified Claude. ## Linked Issues or Issue Description Refs #19611, #19683, #19196, #19733 and #19157. After a host gracefully evicts a released task, the next run can receive a new allocation. A cached controller rejects it with `native_session_released_owner_scope_mismatch`. The detached-controller path can already recover the suspended session. ## What Changed - Require the exact source, old task and current account/workspace scope before releasing a cached controller. - Use the existing physical-stop receipt checks. Detach only the local controller, then use the same suspended-state admission as the detached path. - Preserve the current run's instruction reservation. Retain a failed detach for cleanup. - Add actual Core, PostgreSQL and native Runner tests for Codex and qualified Claude through ACPX. Route them once through the native CI lane. ## Verification - The original real-entry Codex case failed at the cached-task scope check after native retirement, confirmed process exit and a recorded stop receipt. - The current head's native CI lane passed all 16 eviction cases for Codex and qualified Claude. These check cached and detached owners, current instruction preparation, reused-allocation inspection, missing stop/state, foreign identity and non-suspended state. They verify the same provider session, two ordinary turns, one accepted source result and no replacement submit on refusal. - Local checks passed the composition matrix, 12 targeted executor controls, 43 CI lane-selection checks and server TypeScript. All eight refusal cases now assert the expected error and whether the checkpoint request or host read was reached. This addresses the first standard review's test finding. - Commands: `pnpm exec vitest run src/services/native-runtime/plugin-runner-eviction.integration.test.ts`, focused `native-session-executor.test.ts` controls, `node ../node_modules/typescript/bin/tsc --noEmit -p tsconfig.json`, and `node --test scripts/__tests__/run-vitest-stable-shard.test.mjs`. - [Full CI](https://github.com/paperclipai/paperclip/actions/runs/42349579107) passed on head `9a25709ef43365b1f1849413b935785838eca19b`. [Standard review](https://github.com/paperclipai/paperclip/pull/19760#issuecomment-6703629255) is 5/5 on this head, with no unresolved findings. Whitespace and added-line secret/PII checks passed. ## Risks - Controlled provider processes and an authenticated local host fixture model the Launcher edge. This does not qualify live vendor transcript storage, mounts or a deployment. - This covers graceful eviction before allocation or before stopped-task inspection during `describe`. Loss after that inspection or an ambiguous claim remains held for reconciliation. No unavailable reply proves physical stop. - Missing or invalid suspended state still refuses continuation. The deterministic refusal rule from #19733 remains unchanged. Abrupt death without a suspended checkpoint stays with the separate #19157 work. - No new capability, mode, schema, dependency, deployment or activation is added. The existing draft #19157 and dependent branches remain unchanged. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning, through Codex with repository tools and code execution. The runtime does not expose a context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <[email protected]>
This commit is contained in:
Nicky LeachandPaperclip authored and GitHub committed
1 parent
e95dcadbd2
commit
beb3e55072
5 files changed
+403
-3
No files matched your search
@@ -152,6 +152,16 @@ 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.
|
||||
|
||||
When graceful host eviction has saved a suspended session and Core has recorded
|
||||
physical stop, a later allocation can use ordinary `resumeFrom` recovery. Core
|
||||
checks the old task, source identity and current scope before detaching a cached
|
||||
released controller. Cached and detached paths then use the same retained-state
|
||||
checks. Missing state, a foreign identity or an unsettled checkpoint refuses
|
||||
continuation; Core does not replace the provider session on that refusal. This
|
||||
covers eviction before allocation and stopped-task inspection during `describe`.
|
||||
A later loss during claim still needs reconciliation. Abrupt process death without
|
||||
a suspended checkpoint remains a separate recovery path.
|
||||
|
||||
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.
|
||||
|
||||
@@ -325,6 +325,7 @@ const usageDeliverySuitePath = "server/src/services/native-runtime/native-usage-
|
||||
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 pluginEvictionSuitePath = "server/src/services/native-runtime/plugin-runner-eviction.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";
|
||||
@@ -375,7 +376,8 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes
|
||||
assert.ok(!files.includes(warmCheckpointOwnershipSuitePath));
|
||||
assert.ok(!files.includes(pluginWarmOwnerSuitePath));
|
||||
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.ok(!files.includes(pluginEvictionSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, pluginReleaseResumeSuitePath, pluginEvictionSuitePath, "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);
|
||||
@@ -398,6 +400,7 @@ for (const [caller, envOverrides] of [["Release", { GITHUB_WORKFLOW: "Release" }
|
||||
assert.ok(files.includes(warmCheckpointOwnershipSuitePath));
|
||||
assert.ok(files.includes(pluginWarmOwnerSuitePath));
|
||||
assert.ok(files.includes(pluginReleaseResumeSuitePath));
|
||||
assert.ok(files.includes(pluginEvictionSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.equal(new Set(files).size, files.length);
|
||||
});
|
||||
@@ -433,6 +436,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => {
|
||||
warmCheckpointOwnershipSuitePath,
|
||||
pluginWarmOwnerSuitePath,
|
||||
pluginReleaseResumeSuitePath,
|
||||
pluginEvictionSuitePath,
|
||||
]);
|
||||
});
|
||||
|
||||
|
||||
@@ -107,6 +107,7 @@ const nativeRunnerSuites = [
|
||||
"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",
|
||||
"server/src/services/native-runtime/plugin-runner-eviction.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
|
||||
|
||||
@@ -9510,8 +9510,39 @@ 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 (cachedReleasedOwner?.released) {
|
||||
const proof = releasedWarmRunner(readWarmNativeEnvelope(input.execution));
|
||||
const identity = readRunnerdDurableIdentity(scopedRunnerdStateRoot(input.execution));
|
||||
const target = input.runnerExecutionTarget;
|
||||
if (proof && target?.kind === "remote" && target.transport === "sandbox" && target.runnerTask && target.leaseId
|
||||
&& target.sandboxLeaseAcquisition?.outcome === "created" && proof.taskId !== target.sandboxLeaseAcquisition.providerLeaseId) {
|
||||
const ownInstructionPreparation = cachedReleasedOwner.instructionPreparationRunId === input.execution.binding.runId
|
||||
&& input.instructionWorkingCopy?.runId === input.execution.binding.runId;
|
||||
if ((cachedReleasedOwner.busy && !ownInstructionPreparation)
|
||||
|| cachedReleasedOwner.preparingRunId && cachedReleasedOwner.preparingRunId !== input.execution.binding.runId
|
||||
|| cachedReleasedOwner.instructionPreparationRunId && !ownInstructionPreparation
|
||||
|| cachedReleasedOwner.closeOnReleaseReason) throw new Error("native_session_supervisor_busy");
|
||||
if (proof.taskId !== cachedReleasedOwner.pluginTaskId || proof.scopeDigest !== releasedWarmScope(input)
|
||||
|| canonicalJson(proof.source) !== canonicalJson(identity)) throw new Error("native_session_released_owner_scope_mismatch");
|
||||
const wasBusy = cachedReleasedOwner.busy;
|
||||
cachedReleasedOwner.busy = true;
|
||||
try {
|
||||
// Only physical stop releases the cached controller; ordinary checkpoint admission still proves continuation.
|
||||
const resumeFrom = await resolvePluginTaskResume(input.db, { companyId: input.execution.binding.companyId,
|
||||
leaseId: target.leaseId, priorIdentity: proof.source, inspectCheckpoint: true });
|
||||
if (resumeFrom !== proof.taskId) throw new Error("native_session_released_owner_scope_mismatch");
|
||||
if (!cachedReleasedOwner.session.detachControllerForRestart) throw new Error("native_session_released_detach_unavailable");
|
||||
try {
|
||||
await cachedReleasedOwner.session.detachControllerForRestart();
|
||||
await cachedReleasedOwner.githubAccess?.stop();
|
||||
} catch (error) {
|
||||
cachedReleasedOwner.closeOnReleaseReason = "released Runner local detach is unconfirmed";
|
||||
throw error;
|
||||
}
|
||||
forgetWarmNativeSession(nativeSessionScopeKey(input.execution), cachedReleasedOwner);
|
||||
} finally { cachedReleasedOwner.busy = wasBusy; }
|
||||
} else requireReleasedWarmScope(input, proof, identity);
|
||||
}
|
||||
if (input.useRunnerd && !input.restartRecovery && !input.backend &&
|
||||
input.runnerExecutionTarget?.kind === "remote" &&
|
||||
input.execution.session.lifecyclePolicy.mode === "warm" &&
|
||||
|
||||
@@ -0,0 +1,354 @@
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdir, cp, readFile, readdir, rename, 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 } from "vitest";
|
||||
import { completionContracts, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, plugins } 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, closeIdleWarmNativeSessionsForRestart, readNativeSessionIdleWork, claimWarmNativeInstructionCopy } from "./native-session-executor.js";
|
||||
import { resolvePaperclipInstanceRoot } from "../../home-paths.js";
|
||||
import { environmentRuntimeService } from "../environment-runtime.js";
|
||||
import { environmentService } from "../environments.js";
|
||||
import { resolveEnvironmentExecutionTarget } from "../environment-execution-target.js";
|
||||
import { executeEnvironmentTask } from "../environment-task-runtime.js";
|
||||
import type { PluginWorkerManager } from "../plugin-worker-manager.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 graceful eviction 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", "detached", "reused lease", "prepared copy", "missing stop", "missing state", "foreign state", "working state"] as const).map(mode => [harness, mode] as const))])("continues %s after graceful eviction with a %s Core owner", { timeout: 60_000 }, async ([harness, mode], context) => {
|
||||
if (harness === "claude" && !existsSync(fakeAcpxSidecar)) context.skip();
|
||||
const detached = mode === "detached" || mode === "missing state" || mode === "foreign state";
|
||||
const denied = mode.startsWith("missing") || mode.endsWith("state");
|
||||
const instructionCopy = (id: string) => ({ runId: id, root: workspace, preparationKey: "unchanged-fixture-instructions", hasChanges: async () => false, collectStopped: async () => {} });
|
||||
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(), taskA = randomUUID(), taskB = randomUUID(), pluginId = randomUUID();
|
||||
const providerKey = `plugin:${pluginId}:tasks`;
|
||||
const metadata = { driver: "plugin", pluginId, driverKey: "tasks", runnerTaskLease: true };
|
||||
await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "eviction-fixture", version: "1.0.0", status: "ready", manifestJson: { capabilities: ["environment.drivers.register"], environmentDrivers: [{ driverKey: "tasks", supportsTasks: true }] } as never });
|
||||
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", config: { pluginKey: pluginId, driverKey: "tasks" } });
|
||||
await db.insert(environmentLeases).values({ id: leaseA, companyId, environmentId, heartbeatRunId: runId, issueId,
|
||||
executionWorkspaceId: workspaceId, provider: providerKey, providerLeaseId: taskA, status: "active", leasePolicy: "reuse_by_environment", metadata: { ...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: detached ? 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 hosts = new Map<string, ReturnType<typeof createPrivateHostControlFixture>>([[taskA, createPrivateHostControlFixture()], [taskB, createPrivateHostControlFixture()]]);
|
||||
const binary = defaultCapabilityRunnerdBinary();
|
||||
const handles = new Map<string, RunnerProcessHandle>();
|
||||
const runnerRoot = (taskId: string) => join(root, taskId);
|
||||
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, cleaning = false, sourceReadRequests = 0, sourceReadResponses = 0;
|
||||
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 (taskId: string = taskA) => JSON.parse(await readFile(join(runnerRoot(taskId), "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, taskId: string): Promise<unknown> => {
|
||||
const host = hosts.get(taskId)!;
|
||||
const operation = environmentTaskOperationSchema.parse(raw);
|
||||
if (operation.kind === "describe") return { kind: "description", taskId, launchPreparationVersion: 1,
|
||||
projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact };
|
||||
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(taskId))).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", "--durable-turn-ids"] }, 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++;
|
||||
if (operation.resumeFrom) {
|
||||
expect(operation.resumeFrom).toBe(taskA);
|
||||
expect(handles.get(taskA)?.isExitConfirmed?.()).toBe(true);
|
||||
await cp(runnerRoot(taskA), runnerRoot(taskId), { recursive: true });
|
||||
}
|
||||
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(taskId), 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"))}` } } : {}) });
|
||||
handles.set(taskId, handle);
|
||||
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: handles.get(taskId)?.isExitConfirmed?.() ? "cancelled" : "running", executionStopped: (cleaning || mode !== "missing stop") && (handles.get(taskId)?.isExitConfirmed?.() ?? false), sessionStateAvailable: (cleaning || mode !== "missing state" && mode !== "missing stop") && handles.get(taskId)?.isExitConfirmed?.() === true && (await state(taskId)).lifecycle === "suspended" };
|
||||
if (operation.kind === "read_state") {
|
||||
if (operation.resumeFrom === taskA) sourceReadResponses++;
|
||||
const live = await state(operation.resumeFrom ?? taskId);
|
||||
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: operation.resumeFrom && mode === "foreign state" ? { ...live, turnId: "foreign-turn" }
|
||||
: operation.resumeFrom && mode === "working state" ? { ...live, lifecycle: "ready" } : live };
|
||||
}
|
||||
if (operation.kind === "archive_state") {
|
||||
expect(operation.resumeFrom).toBe(taskA);
|
||||
expect(handles.get(taskA)?.isExitConfirmed?.()).toBe(true);
|
||||
const path = join(runnerRoot(taskA), "runner-state.json"), archived = join(runnerRoot(taskA), `archive-${operation.archiveKey}.json`);
|
||||
try { await rename(path, archived); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }
|
||||
const saved = JSON.parse(await readFile(archived, "utf8"));
|
||||
expect(sameSource(operation.priorIdentity, saved)).toBe(true);
|
||||
return { kind: "state", taskId, state: saved };
|
||||
}
|
||||
if (operation.kind === "release") {
|
||||
const live = await state(taskId); 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);
|
||||
return { kind: "resume", taskId, outcome: "resumed" };
|
||||
}
|
||||
if (operation.kind === "stop") { await host.close(); }
|
||||
if (operation.kind === "complete") { expect(handles.get(taskId)?.isExitConfirmed?.()).toBe(true); }
|
||||
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 workers = { getWorker: () => ({ supportedMethods: ["environmentTask"] }), call: async (_id: string, method: string, params: any, _timeout?: number, _unused?: unknown, settled?: () => void) => {
|
||||
if (method === "environmentAcquireLease") return { providerLeaseId: taskB, metadata: { creationOperationId: params.creationOperationId } };
|
||||
try { return await task(params.operation, params.runId, params.taskId); } finally { settled?.(); }
|
||||
} } as unknown as PluginWorkerManager;
|
||||
const dispatch = (leaseId: string, operation: PluginEnvironmentTaskOperation, options?: { executionContext?: unknown; expectedTaskId?: string }) => {
|
||||
if (operation.kind === "read_state" && operation.resumeFrom === taskA) sourceReadRequests++;
|
||||
return executeEnvironmentTask(db, workers, { companyId, leaseId, operation, ...options });
|
||||
};
|
||||
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));
|
||||
const target = mode === "reused lease" ? await resolveEnvironmentExecutionTarget({ db, companyId, adapterType: "paperclip_runner",
|
||||
environment: { id: environmentId, driver: "plugin", config: { pluginKey: pluginId, driverKey: "tasks" } }, leaseId,
|
||||
lease: await environmentService(db).getLeaseById(leaseId), leaseMetadata: metadata,
|
||||
environmentRuntime: environmentRuntimeService(db, { pluginWorkerManager: workers }) }) : {
|
||||
kind: "remote" as const, transport: "sandbox" as const, remoteCwd: workspace, environmentId, leaseId, providerKey,
|
||||
runnerTask: (operation: unknown, options?: { executionContext?: unknown; expectedTaskId?: string }) => dispatch(leaseId, operation as PluginEnvironmentTaskOperation, options),
|
||||
sandboxLeaseAcquisition: { outcome: "created" as const, providerLeaseId: leaseId === leaseA ? taskA : taskB } };
|
||||
if (!target) throw new Error("Fixture execution target unavailable");
|
||||
return executePaperclipNativeSession({ db, execution: current, runnerInstanceId: native.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerEnvironment: harness === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {},
|
||||
...(mode === "prepared copy" ? { instructionWorkingCopy: instructionCopy(current.binding.runId) } : {}),
|
||||
managedAiCredentialHome: home, managedAiCredentialIdentity: "fixture-revision", managedAiAccountIdentity: "fixture-account", turnTimeoutMs: 15_000,
|
||||
runnerExecutionTarget: target });
|
||||
};
|
||||
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: taskA });
|
||||
const providerSessionId = (await supervisor()).snapshot.providerSessionId;
|
||||
expect(providerSessionId).toEqual(expect.any(String));
|
||||
if (detached) await new Promise(done => setTimeout(done, 5_200));
|
||||
expect(readNativeSessionIdleWork()).toBe("none");
|
||||
await reconcileNativeFinalizations(db, [runId]);
|
||||
const evict = async () => {
|
||||
const retired = await hosts.get(taskA)!.request({ op: "retire", id: "host-eviction", parkId: parkId(releases[0]!) });
|
||||
expect(retired.status).toBe("retired");
|
||||
await handles.get(taskA)!.completion;
|
||||
expect((await state()).lifecycle).toBe("suspended");
|
||||
// The existing task runtime records physical stop and state availability.
|
||||
expect(await dispatch(leaseA, { kind: "status" })).toMatchObject({ executionStopped: mode !== "missing stop", sessionStateAvailable: mode !== "missing state" && mode !== "missing stop" });
|
||||
if (mode !== "missing stop") {
|
||||
await dispatch(leaseA, { kind: "complete" });
|
||||
await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success" }).where(eq(environmentLeases.id, leaseA));
|
||||
}
|
||||
};
|
||||
if (mode !== "reused lease") await evict();
|
||||
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));
|
||||
const [oldLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseA));
|
||||
await db.insert(environmentLeases).values({ id: leaseB, companyId, environmentId, heartbeatRunId: runB, issueId, executionWorkspaceId: workspaceId,
|
||||
provider: providerKey, providerLeaseId: mode === "reused lease" ? taskA : taskB, status: "active", leasePolicy: "reuse_by_environment",
|
||||
metadata: mode === "reused lease" ? { ...oldLease!.metadata, runnerTaskReused: true, runnerTaskPreviousLeaseId: leaseA,
|
||||
runnerTaskExecutionStopped: false } : { ...metadata, remoteCwd: workspace } });
|
||||
if (mode === "reused lease") {
|
||||
expect(handles.get(taskA)!.isExitConfirmed?.()).toBe(false);
|
||||
await evict(); // B already holds A when the host evicts it, before normal describe/claim.
|
||||
}
|
||||
const next = { ...execution, binding: { ...execution.binding, runId: runB } };
|
||||
const releasePreparation = mode === "prepared copy" ? await claimWarmNativeInstructionCopy({ priorExecution: execution,
|
||||
companyId, agentId, executionWorkspaceId: workspaceId, workspace: execution.workspace, environmentId, runId: runB,
|
||||
preparationKey: "unchanged-fixture-instructions", adopt: async () => instructionCopy(runB) }) : null;
|
||||
if (mode === "prepared copy") expect(releasePreparation).toBeTypeOf("function");
|
||||
pendingFollowup = execute(next, leaseB);
|
||||
if (denied) {
|
||||
const refusal = mode === "working state" ? "checkpoint_not_settled"
|
||||
: mode === "foreign state" ? "Plugin Runner checkpoint identity mismatch"
|
||||
: "Plugin Runner retained session is not authorized";
|
||||
await expect(pendingFollowup).rejects.toThrow(refusal);
|
||||
if (mode === "missing stop") expect(sourceReadRequests).toBe(0);
|
||||
else expect(sourceReadRequests).toBeGreaterThan(0);
|
||||
if (mode === "foreign state" || mode === "working state") expect(sourceReadResponses).toBeGreaterThan(0);
|
||||
else expect(sourceReadResponses).toBe(0);
|
||||
expect(submissions).toBe(1); expect(claims).toEqual([]);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runB))).toHaveLength(0);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toHaveLength(1);
|
||||
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]);
|
||||
await releasePreparation?.();
|
||||
return runB;
|
||||
};
|
||||
const runB = await admitNext();
|
||||
if (denied) return;
|
||||
await waitForCondition("automatic release B", () => releases.length === 2);
|
||||
expect(await proof()).toMatchObject({ source: releases[1], taskId: taskB });
|
||||
expect((await supervisor()).snapshot.providerSessionId).toBe(providerSessionId);
|
||||
const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/);
|
||||
expect(submissions).toBe(2);
|
||||
expect(releases[1]!.runId).toBe(runB);
|
||||
expect(claims).toEqual([]);
|
||||
if (harness === "codex") {
|
||||
expect(methods.filter(method => method === "process-start")).toHaveLength(2);
|
||||
expect(methods.filter(method => method === "turn/start")).toHaveLength(2);
|
||||
} else {
|
||||
const journal = methods.map(line => JSON.parse(line));
|
||||
expect(journal.filter(entry => entry.command === "turn.start")).toHaveLength(2);
|
||||
}
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId)))).toHaveLength(1);
|
||||
} 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 {
|
||||
cleaning = true;
|
||||
await pendingFollowup?.catch(() => undefined);
|
||||
try {
|
||||
const cleanup = await Promise.allSettled([closeIdleWarmNativeSessionsForRestart(), ...[...hosts.values()].map(host => 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 {
|
||||
https.globalAgent.destroy(); https.globalAgent = oldAgent; setDefaultCACertificates(oldCAs);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user