Support Codex warm plugin task reuse and recovery (#17390)

This commit is contained in:
Nicky Leach authored and GitHub committed 2026-10-23 19:14:14 -07:00
1 parent 9fb28d06ff
commit aa0037a137
26 files changed
+845 -83

No files matched your search

+3 -1
View File
@@ -139,7 +139,9 @@ the environment configuration cannot redirect cleanup to a new provider.
Before preparation or submission, Core checks the active lease, run, issue, and every project
in the company. It checks the submitting run ID separately from a retained PRP
identity. Session reuse requires the same agent, environment, workspace, and
provider. It also requires a retained source task and proof of termination.
provider. Checkpoint resumption also requires a retained source task and proof of termination.
Opt-in Codex warm reuse instead hands off the same retained process through an
atomic lease transition and an exact, authorized `priorIdentity` preparation.
An internal setup probe can use a private, time-limited lease without a run row.
Core checks the saved probe identity, rejects project mounts, and limits the
+40 -1
View File
@@ -23,7 +23,7 @@ bounded `read_file` operation for attachments and deliverables.
Core supports per-turn workspace sessions with Codex, OpenCode, and ACPX.
OpenCode and ACPX require the existing qualified provider profiles.
Workspace-free providers and warm process retention are not supported here.
Workspace-free providers are not supported here. Codex warm retention is opt-in through `warmTaskVersion: 1`; ACPX and OpenCode remain per-turn.
The plugin owns project workspace preparation. Core does not prepare or reuse
a Core Git worktree for a task-only environment.
For plugin-backed project storage, Core uses the workspace provider contract.
@@ -161,3 +161,42 @@ Recovery receives the original run and agent context.
The existing agent-independent `inspect` and `remove` operations continue to
protect storage maintenance. Providers that do not declare finalization version 1
retain the inspection-based completion path. No stored binding migration is required.
## Codex warm task retention
A driver declares `warmTaskVersion: 1`, implements `onEnvironmentResumeLease`,
and returns `warmTaskVersion: 1` from `describe` only when its execution host
supports authorized run attachment. Select the agent's warm lifecycle or set
`runnerLifecycleMode: "warm"` and `runnerIdleTimeoutMs` on its plugin environment.
Core requires an explicitly selected account. Setup probes remain ephemeral.
A Launcher task owns one Runner process, independently of Core heartbeat runs.
Successful warm run completion retains the task and its mounts. The reuse
fingerprint includes the authorized project set; changing it retires the previous
task before checkpoint recovery. Adoption and each warm handoff require a current,
configuration-bound project-auth check against the exact retained mount scope. Core atomically
hands its retained provider identity to one new run lease, revalidates current
project authentication, prepares the new run's resources, and uses PRP
`run.attach`. Preparation includes the exact `priorIdentity` and stable
`accountIdentity`. The provider journals that transition idempotently before
allowing the new authority. The original ingress address remains unchanged.
Runner accepts `connection: { mode: "external" }` on warm attachment to preserve
the process owner's listener and TLS settings while rotating the run identity.
Idle retirement suspends and stops Runner, saves its exact checkpoint, writes
credentials through the durable lease handoff, then completes the provider task.
Temporary credential homes are not needed after the originating run ends.
Termination receipts cover all run leases that owned the retired physical task.
A stopped task gets a new provider task ID for checkpoint resumption. Inspection
precedes in-memory session selection so a replacement cannot inherit the stopped
session. Core saves the fresh provider expiry and forwards and enforces any
caller-imposed deadline. A missing
or unreachable computer never authorizes a replacement. Controller recovery can
adopt a verified surviving idle process through its authenticated PRP connection;
ambiguous transitions remain fenced by the existing durable recovery checks.
Changes to account, credential generation, workspace, runtime or process policy
retire the old owner before recovery. A new authority epoch archives the stopped
checkpoint under its stable archive key and preserves provider session files.
Warm status monitoring relies on PRP after startup and does not poll an idle
Launcher. General ACPX warm support is deliberately excluded.
@@ -152,7 +152,9 @@ pub(crate) fn next_authority_config(
.ok_or_else(|| {
DurableRunnerError::invalid("run.attach authority connection is required")
})?;
let external = connection.get("mode").and_then(Value::as_str) == Some("external");
let connect_url = match connection.get("mode").and_then(Value::as_str) {
Some("external") => current.connect_url.clone(),
Some("connect") => connection
.get("connectUrl")
.and_then(Value::as_str)
@@ -184,11 +186,13 @@ pub(crate) fn next_authority_config(
};
let mut next = current.clone();
next.connect_url = connect_url;
next.ca_bundle_path = connection
.get("caBundlePath")
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.map(Into::into);
if !external {
next.ca_bundle_path = connection
.get("caBundlePath")
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.map(Into::into);
}
next.runner_instance_id = read_identity("runnerInstanceId")?;
next.environment_lease_id = read_identity("environmentLeaseId")?;
next.run_id = read_identity("runId")?;
@@ -225,6 +229,14 @@ fn apply_authority_rotation(
"warm authority rotation requires native v2 session state acknowledgement",
));
}
let preserve_endpoint = state.warm_transition.as_ref().is_some_and(|transition| {
transition
.receipt
.connection
.get("mode")
.and_then(Value::as_str)
== Some("external")
});
let reconnect_count = state.reconnect_count.saturating_add(1);
let mut diagnostics = state.diagnostics.clone();
let mut rotated = DurableState::new(&next);
@@ -239,7 +251,12 @@ fn apply_authority_rotation(
store.save(&rotated)?;
*state = rotated;
*config = next;
endpoint.rotate(&config.connect_url, &config.run_id)
if preserve_endpoint {
// The process owner keeps its original ingress while PRP rotates run authority.
Ok(())
} else {
endpoint.rotate(&config.connect_url, &config.run_id)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -3506,6 +3523,80 @@ mod tests {
}
}
#[test]
fn warm_external_attachment_preserves_the_original_ingress() {
let directory =
std::env::temp_dir().join(format!("paperclip-warm-external-{}", std::process::id()));
let _ = fs::remove_dir_all(&directory);
let listener = std::net::TcpListener::bind("0.0.0.0:0").unwrap();
let port = listener.local_addr().unwrap().port();
let path = "/api/runner/v1/connect/run_1".to_owned();
let mut current = config(directory.clone());
current.connect_url = format!("listen://0.0.0.0:{port}{path}");
current.ca_bundle_path = Some(PathBuf::from("/fixture/trusted-ca.pem"));
current.runner_digest = format!("sha256:{}", "a".repeat(64));
let original_url = current.connect_url.clone();
let mut endpoint = RunnerTransportEndpoint::Listen {
listener,
path: path.clone(),
};
let store = DurableStateStore::new(&directory).unwrap();
let (mut state, _) = store.load_or_create(&current).unwrap();
for run in 2..=3 {
let mut attach = command("run.attach");
attach.payload = json!({"paperclipNextAuthority":{
"identity":{"runnerInstanceId":current.runner_instance_id,"environmentLeaseId":current.environment_lease_id,
"runId":format!("run_{run}"),"normalizedSessionId":current.normalized_session_id,
"turnId":format!("turn_{run}"),"itemId":format!("item_{run}")},
"connection":{"mode":"external"}
}});
let next = next_authority_config(&attach, &current).unwrap().unwrap();
assert_eq!(next.connect_url, original_url);
assert_eq!(next.ca_bundle_path, current.ca_bundle_path);
let result = StoredCommandResult {
command_id: attach.command_id.clone(),
controller_seq: attach.controller_seq,
command_type: "run.attach".to_owned(),
status: "completed".to_owned(),
result: json!({"status":"resumed"}),
};
let receipt = WarmRunTransition::new(
&current,
&next,
&attach,
&result,
0,
"lease".to_owned(),
u64::MAX,
0,
)
.unwrap();
state.schema = TRANSITION_STATE_SCHEMA.to_owned();
state.warm_transition = Some(PendingWarmRunTransition {
receipt,
phase: "prepared".to_owned(),
command: attach,
result,
});
apply_authority_rotation(&mut state, &store, &mut current, &mut endpoint, next)
.unwrap();
assert_eq!(current.run_id, format!("run_{run}"));
match &endpoint {
RunnerTransportEndpoint::Listen {
listener,
path: retained_path,
} => {
assert_eq!(listener.local_addr().unwrap().port(), port);
assert_eq!(retained_path, &path);
}
_ => panic!("external attachment must preserve the listener"),
}
state.warm_transition = None;
state.schema = "paperclip.runner.durable.state.v1".to_owned();
}
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn warm_run_attachment_reuses_the_provider_ingress_listener() {
let directory = std::env::temp_dir().join(format!(
@@ -55,6 +55,7 @@ export const environmentTaskOperationSchema = z.discriminatedUnion("kind", [
z.object({ kind: z.literal("copilot_metadata"), token: credentialValue, model: z.string().min(1).max(1024).optional() }).strict(),
z.object({ kind: z.literal("prepare"), requirements: environmentRunnerLaunchRequirementsSchema,
recovery: preparedLaunch.optional(), resumeFrom: providerId.optional(),
priorIdentity: durableIdentity.optional(), accountIdentity: z.string().min(1).max(1024).optional(),
projectIds: z.array(z.string().uuid()).max(64).refine(ids => new Set(ids).size === ids.length, "Duplicate project IDs"),
}).strict(),
z.object({
@@ -83,7 +84,10 @@ export const environmentTaskOperationSchema = z.discriminatedUnion("kind", [
maxOutboxBytes: z.number().int().positive(), p0ReserveBytes: z.number().int().positive(),
maxRuntimeMs: z.number().int().nonnegative().optional(), maxLifetimeMs: z.number().int().nonnegative().optional(),
reconnectGraceMs: z.number().int().nonnegative().optional(),
lifecyclePolicy: z.object({ mode: z.literal("per_turn"), idleTimeoutMs: z.null() }).strict(),
lifecyclePolicy: z.discriminatedUnion("mode", [
z.object({ mode: z.literal("per_turn"), idleTimeoutMs: z.null() }).strict(),
z.object({ mode: z.literal("warm"), idleTimeoutMs: z.number().int().positive().max(86_400_000) }).strict(),
]),
}).strict().optional(),
}).strict(),
z.object({ kind: z.literal("check_project_auth"), configurationId: providerId.max(1024),
@@ -120,6 +124,7 @@ export const environmentTaskResultSchema = z.discriminatedUnion("kind", [
credential: environmentTaskCredentialSchema.nullable() }).strict(),
z.object({ kind: z.literal("description"), taskId: providerId,
runner: z.object({ version: z.string().min(1), digest: z.string().regex(/^sha256:[a-f0-9]{64}$/) }).strict(),
warmTaskVersion: z.literal(1).optional(),
launchPreparationVersion: z.literal(1).optional(),
freshWorkspaceVersion: z.literal(1).optional(),
projectAuthCheckVersion: z.literal(1).optional(),
+2
View File
@@ -669,6 +669,8 @@ export interface PluginEnvironmentAcquireLeaseParams extends PluginEnvironmentDr
}
export interface PluginEnvironmentResumeLeaseParams extends PluginEnvironmentDriverBaseParams {
/** New run authority for a retained task; omitted by legacy environment consumers. */
runId?: string;
providerLeaseId: string;
leaseMetadata?: Record<string, unknown>;
}
@@ -0,0 +1,13 @@
import {describe,expect,it} from "vitest";
import {environmentTaskOperationSchema} from "../src/environment-tasks.js";
const submit={kind:"submit",projectIds:[],bootstrapTicket:"synthetic",runner:{harness:"codex",protocolMin:1,protocolMax:3,runnerId:"runner",leaseId:"lease",runId:"run",sessionId:"session",turnId:"turn",itemId:"item"},
launch:{runnerVersion:"fixture",runnerDigest:`sha256:${"a".repeat(64)}`,maxOutboxBytes:262144,p0ReserveBytes:65536,lifecyclePolicy:{mode:"warm",idleTimeoutMs:300000}}};
describe("task lifecycle policy",()=>{
it("accepts bounded warm policy and the existing per-turn policy",()=>{
expect(environmentTaskOperationSchema.parse(submit)).toEqual(submit);
expect(environmentTaskOperationSchema.safeParse({...submit,launch:{...submit.launch,lifecyclePolicy:{mode:"per_turn",idleTimeoutMs:null}}}).success).toBe(true);
});
it.each([0,-1,86400001,null,1.5])("rejects invalid warm idle timeout %s",idleTimeoutMs=>{
expect(environmentTaskOperationSchema.safeParse({...submit,launch:{...submit.launch,lifecyclePolicy:{mode:"warm",idleTimeoutMs}}}).success).toBe(false);
});
});
+2
View File
@@ -66,6 +66,8 @@ export type SandboxEnvironmentConfig =
| PluginSandboxEnvironmentConfig;
export interface PluginEnvironmentConfig {
runnerLifecycleMode?: "inherit" | "per_turn" | "warm";
runnerIdleTimeoutMs?: number;
pluginKey: string;
driverKey: string;
driverConfig: Record<string, unknown>;
+2
View File
@@ -194,6 +194,8 @@ export interface PluginEnvironmentDriverDeclaration {
/** Provides Runner execution, resource preparation, and PRP connections via environmentTask. */
supportsTasks?: boolean;
freshWorkspaceVersion?: 1;
/** Codex task retention and authorized run attachment, version 1. */
warmTaskVersion?: 1;
/** Stable driver key, unique within the plugin. Namespaced by plugin ID at runtime. */
driverKey: string;
/**
+1
View File
@@ -179,6 +179,7 @@ export type SandboxProviderCapabilitiesInput = z.infer<typeof sandboxProviderCap
export const pluginEnvironmentDriverDeclarationSchema = z.object({
supportsTasks: z.boolean().optional(),
freshWorkspaceVersion: z.literal(1).optional(),
warmTaskVersion: z.literal(1).optional(),
driverKey: z.string().min(1).regex(
/^[a-z0-9][a-z0-9._-]*$/,
"Environment driver key must start with a lowercase alphanumeric and contain only lowercase letters, digits, dots, hyphens, or underscores",
@@ -776,6 +776,7 @@ describe("native runner lifecycle changes before lease acquisition", () => {
const input = {
companyId: "company-1", selectedEnvironmentId: "env-1", localEnvironmentId: "local",
adapterType: "paperclip_runner", issueId: "existing-task", heartbeatRunId: "run-1", agentId: "agent-1",
runnerTaskProjectIds:["anchor","authorized-reference"],
persistedExecutionWorkspace: { id: "existing-workspace", mode: "shared_workspace" as const },
executionWorkspaceSettings: null,
};
@@ -786,6 +787,7 @@ describe("native runner lifecycle changes before lease acquisition", () => {
expect(warm.environment.config).toEqual({ provider: "daytona", reuseLease: true });
expect(acquireRunLease.mock.calls[1][0].persistedExecutionWorkspace).toEqual(input.persistedExecutionWorkspace);
expect(acquireRunLease.mock.calls[1][0].configuredEnvironment).toBe(environment);
expect(acquireRunLease.mock.calls[1][0].runnerTaskProjectIds).toEqual(input.runnerTaskProjectIds);
// Run-scoped adaptation must not change the shared environment or other agents.
expect(environment.config).toEqual({ provider: "daytona", reuseLease: false });
const coldAgain = await orchestrator.acquireForRun({ ...input, heartbeatRunId: "run-3", adapterConfig: { lifecycleMode: "per_turn" } });
@@ -1,5 +1,6 @@
import { JsonRpcCallError } from "@paperclipai/plugin-sdk";
import { collectRunFailureDiagnostics } from "../services/run-failure-diagnostics.js";
import { createSandboxOrphanCleanupSpool } from "../services/sandbox-orphan-cleanup-spool.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
import { createHash, randomUUID } from "node:crypto";
import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
@@ -2369,6 +2370,34 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
}
});
it("ingests a replacement orphan appended after startup without duplicate concurrent cleanup", async () => {
const {companyId,environment,runId}=await seedEnvironment({driver:"sandbox",name:"Late plugin orphan",config:{provider:"fake",image:"ubuntu:24.04",reuseLease:false}});
let release!:()=>void,entered!:()=>void;
const pending=new Promise<void>(resolve=>{release=resolve;}),started=new Promise<void>(resolve=>{entered=resolve;});
const realEnvironmentService=environmentsModule.environmentService;
const factory=vi.spyOn(environmentsModule,"environmentService").mockImplementation(database=>{
const real=realEnvironmentService(database);
return {...real,insertPendingCleanupLease:async input=>{entered();await pending;return real.insertPendingCleanupLease(input);}};
});
try {
const spool=createSandboxOrphanCleanupSpool(process.env.SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR!);
const active=environmentRuntimeService(db,{orphanCleanupSpool:spool});
expect(await active.flushDeferredOrphanCleanups()).toEqual({recovered:0,pending:0});
const record={companyId,environmentId:environment.id,executionWorkspaceId:null,issueId:null,heartbeatRunId:runId,
provider:"plugin:fixture:tasks",providerLeaseId:"late-task",metadata:{driver:"plugin",runnerTaskDispatched:false}};
expect(await spool.append(record)).toBe(true);
const first=active.flushDeferredOrphanCleanups();await started;
const second=active.flushDeferredOrphanCleanups();
release();
expect(await first).toEqual({recovered:1,pending:0});
expect(await second).toEqual({recovered:1,pending:0});
expect(await spool.load()).toEqual([]);
expect(await active.flushDeferredOrphanCleanups()).toEqual({recovered:0,pending:0});
const rows=await db.select().from(environmentLeases).where(eq(environmentLeases.environmentId,environment.id));
expect(rows).toEqual([expect.objectContaining({providerLeaseId:"late-task",status:"pending_cleanup"})]);
} finally {release();factory.mockRestore();}
});
it("persists the orphan to the durable spool so a restart recovers and lands the durable row", async () => {
const { companyId, environment, runId } = await seedEnvironment({
driver: "sandbox",
@@ -163,7 +163,7 @@ databaseSuite("Core plugin Runner task lifecycle", () => {
warnings: [], created: false, branchCreatedByRuntime: false },
effectiveExecutionWorkspaceMode: null, persistedExecutionWorkspace: null });
const target = realization.executionTarget;
expect(target).toMatchObject({ kind: "remote", transport: "sandbox", remoteCwd: "/local/project", leaseId: acquired.lease.id });
expect(target).toMatchObject({ kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: acquired.lease.id });
if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runnerTask) throw new Error("Missing plugin target");
expect(target.runner).toBeUndefined();
await expect(target.runnerTask({ kind: "describe" }))
@@ -0,0 +1,171 @@
import { randomUUID } from "node:crypto";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { eq } from "drizzle-orm";
import { agents, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, plugins, projects } from "@paperclipai/db";
import { environmentRuntimeService } from "../services/environment-runtime.js";
import * as orphanCleanup from "../services/sandbox-orphan-cleanup-spool.js";
import { resolvePluginTaskResume } from "../services/environment-task-runtime.js";
import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
const support=await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("retained Codex plugin task leases",()=>{
let db:ReturnType<typeof createDb>,stop:()=>Promise<void>;
beforeAll(async()=>{const started=await startEmbeddedPostgresTestDatabase("plugin-warm-tasks");db=createDb(started.connectionString);stop=started.stop;});
afterAll(async()=>{await stop?.();});
async function fixture(initialProjects:string[]=[],initialExpiry?:string,deadline?:Date){
const companyId=randomUUID(),agentId=randomUUID(),issueId=randomUUID(),environmentId=randomUUID(),pluginId=randomUUID();
let runId=randomUUID(),allocation=0,stopped=false,outage=false,releaseFails=false;
const calls:string[]=[];
let projectIds=[...initialProjects],providerExpiry=initialExpiry;
const taskExpiries=new Map<string,string|undefined>(),acquireRequests:any[]=[];
let resumeGate:Promise<void>|undefined,releaseResume:(()=>void)|undefined,resumes=0;
await db.insert(companies).values({id:companyId,name:"Warm fixture",issuePrefix:`W${companyId.slice(0,7)}`});
for(const id of initialProjects) await db.insert(projects).values({id,companyId,name:"Authorized project"});
await db.insert(agents).values({id:agentId,companyId,name:"Codex",adapterType:"paperclip_runner"});
await db.insert(issues).values({id:issueId,companyId,title:"Warm task",status:"in_progress",assigneeAgentId:agentId});
await db.insert(heartbeatRuns).values({id:runId,companyId,agentId,nativeIssueId:issueId,invocationSource:"manual",status:"running"});
const [environment]=await db.insert(environments).values({id:environmentId,name:environmentId,driver:"plugin",status:"active",
config:{pluginKey:pluginId,driverKey:"tasks",driverConfig:{},runnerLifecycleMode:"warm",runnerIdleTimeoutMs:300000}}).returning();
await db.insert(plugins).values({id:pluginId,pluginKey:pluginId,packageName:"fixture-warm",version:"1.0.0",status:"ready",manifestJson:{
capabilities:["environment.drivers.register"],environmentDrivers:[{driverKey:"tasks",supportsTasks:true,warmTaskVersion:1}],
} as never});
const workers={isRunning:()=>true,getWorker:()=>({supportedMethods:["environmentTask","environmentAcquireLease","environmentResumeLease","environmentReleaseLease"]}),
call:async(_id:string,method:string,p:any)=>{
calls.push(method==="environmentTask" ? p.operation.kind : method);
if(method==="environmentAcquireLease") {
const id=`task-${++allocation}`;taskExpiries.set(id,providerExpiry);acquireRequests.push(p);
return {providerLeaseId:id,...(providerExpiry ? {expiresAt:providerExpiry}:{}),metadata:{runId:p.runId}};
}
if(method==="environmentResumeLease") {if(outage) throw Error("unreachable");
if(resumeGate){if(++resumes===2)releaseResume!();await resumeGate;}return {providerLeaseId:stopped?null:p.providerLeaseId,expiresAt:taskExpiries.get(p.providerLeaseId),metadata:{runId:p.runId}};}
if(method==="environmentReleaseLease") {if(releaseFails) throw Error("release unavailable");return {providerLeaseId:p.providerLeaseId,state:"stopped"};}
if(p.operation.kind==="status") return {kind:"status",taskId:p.taskId,phase:stopped?"completed":"running",executionStopped:stopped,sessionStateAvailable:stopped};
if(p.operation.kind==="read_state") return {kind:"state",taskId:p.taskId,state:{...p.operation.identity,lifecycle:"ready"}};
if(p.operation.kind==="stop") stopped=true;
if(p.operation.kind==="prepare") return {kind:"prepared",taskId:p.taskId,launch:{configurationId:"configuration",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}}};
if(p.operation.kind==="check_project_auth") return {kind:"project_auth",taskId:p.taskId,configurationId:p.operation.configurationId,provider:"openai",allowed:true};
if(p.operation.kind==="describe") return {kind:"description",taskId:p.taskId,runner:{version:"fixture",digest:`sha256:${"a".repeat(64)}`},warmTaskVersion:1};
return {kind:"accepted",taskId:p.taskId};
}} as unknown as PluginWorkerManager;
const runtime=()=>environmentRuntimeService(db,{pluginWorkerManager:workers});
const acquire=()=>runtime().acquireRunLease({companyId,environment:environment as never,issueId,agentId,heartbeatRunId:runId,persistedExecutionWorkspace:null,adapterType:"paperclip_runner",runnerTaskProjectIds:projectIds,requestedExpiresAt:deadline});
const first=await acquire();
const identity={runnerInstanceId:"runner",environmentLeaseId:first.lease.id,runId,normalizedSessionId:"session",turnId:"turn",itemId:"item"};
await runtime().task({companyId,leaseId:first.lease.id,operation:{kind:"check_project_auth",configurationId:"configuration",provider:"openai",projectIds:initialProjects}});
await runtime().task({companyId,leaseId:first.lease.id,operation:{kind:"submit",preparedLaunchId:"configuration",projectIds:initialProjects,bootstrapTicket:"synthetic",accountIdentity:"account",
runner:{harness:"codex",protocolMin:1,protocolMax:3,runnerId:identity.runnerInstanceId,leaseId:identity.environmentLeaseId,runId,sessionId:identity.normalizedSessionId,turnId:identity.turnId,itemId:identity.itemId},
launch:{runnerVersion:"fixture",runnerDigest:`sha256:${"a".repeat(64)}`,maxOutboxBytes:262144,p0ReserveBytes:65536,lifecyclePolicy:{mode:"warm",idleTimeoutMs:300000}}}});
await runtime().releaseRunLeases(runId,"released",undefined,"keep_running");
await db.update(heartbeatRuns).set({status:"succeeded"}).where(eq(heartbeatRuns.id,runId));
return {companyId,first,identity,calls,runtime,acquire,acquireRequests,
setExpiry(next?:string){providerExpiry=next;},
async nextRun(){runId=randomUUID();await db.insert(heartbeatRuns).values({id:runId,companyId,agentId,nativeIssueId:issueId,invocationSource:"manual",status:"running"});return runId;},
setProjects(next:string[]){projectIds=next;},stopped(){stopped=true;},failRelease(){releaseFails=true;},outage(){outage=true;},
raceResume(){resumeGate=new Promise<void>(resolve=>{releaseResume=resolve;});}};
}
it("hands a retained process to a new run after controller reconstruction",async()=>{
const f=await fixture();
expect(f.first.lease.leasePolicy).toBe("reuse_by_environment");
expect(f.calls).not.toContain("stop");expect(f.calls).not.toContain("environmentReleaseLease");
const runId=await f.nextRun(),next=await f.acquire();
expect(next.lease.providerLeaseId).toBe(f.first.lease.providerLeaseId);
expect(next.lease.id).not.toBe(f.first.lease.id);
const identity={...f.identity,runId,turnId:"next-turn",itemId:"next-item"};
const operation={kind:"prepare" as const,projectIds:[],priorIdentity:f.identity,accountIdentity:"account",requirements:{
schema:"paperclip.runner.launch-requirements.v1" as const,identity,provider:"codex" as const,model:null,instructions:"Current run",runtimeContext:null,settings:{},approvalPolicy:"never" as const,readOnly:false,preparedConversation:false,includeCollaborationModeInstructions:false}};
await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{kind:"read_state",identity:f.identity}})).resolves.toMatchObject({kind:"state"});
await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation})).resolves.toMatchObject({kind:"prepared"});
await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation})).resolves.toMatchObject({kind:"prepared"});
await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{...operation,priorIdentity:{...f.identity,itemId:"foreign"}}})).rejects.toThrow("prior identity mismatch");
await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{kind:"read_state",identity:f.identity}})).resolves.toMatchObject({kind:"state"});
await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{kind:"read_state",identity:{...f.identity,itemId:"foreign"}}})).rejects.toThrow("state identity mismatch");
expect(f.calls.filter(call=>call==="submit")).toHaveLength(1);
const [old]=await db.select().from(environmentLeases).where(eq(environmentLeases.id,f.first.lease.id));
expect(old.status).toBe("expired");
});
it("resumes stopped checkpoints through a new physical task instead of reusing a tombstone",async()=>{
const f=await fixture();f.stopped();await f.nextRun();
const next=await f.acquire();
expect(next.lease.providerLeaseId).not.toBe(f.first.lease.providerLeaseId);
expect(f.calls).toContain("complete");
await expect(resolvePluginTaskResume(db,{companyId:f.companyId,leaseId:next.lease.id,priorIdentity:f.identity})).resolves.toBe(f.first.lease.providerLeaseId);
});
it("does not allocate a replacement when the retained computer cannot be inspected",async()=>{
const f=await fixture();f.outage();await f.nextRun();
await expect(f.acquire()).rejects.toThrow("unreachable");
expect(f.calls.filter(call=>call==="environmentAcquireLease")).toHaveLength(1);
});
it("atomically admits only one successor to a retained task",async()=>{
const f=await fixture();await f.nextRun();f.raceResume();
const results=await Promise.allSettled([f.acquire(),f.acquire()]);
expect(results.filter(result=>result.status==="fulfilled")).toHaveLength(1);
expect(results.filter(result=>result.status==="rejected")).toHaveLength(1);
expect(f.calls.filter(call=>call==="environmentAcquireLease")).toHaveLength(1);
});
it("replaces a task that stops after lease handoff before preparing the next run",async()=>{
const f=await fixture();await f.nextRun();const next=await f.acquire();f.stopped();
const result=await f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{kind:"describe"}});
expect(result.taskId).not.toBe(f.first.lease.providerLeaseId);
await expect(resolvePluginTaskResume(db,{companyId:f.companyId,leaseId:next.lease.id,priorIdentity:f.identity})).resolves.toBe(f.first.lease.providerLeaseId);
});
it.each(["ownership-lost","database-error","cleanup-error","spool"] as const)("compensates an unowned replacement task: %s",async mode=>{
const f=await fixture();await f.nextRun();const next=await f.acquire();f.stopped();
if(mode==="cleanup-error" || mode==="spool") f.failRelease();
const append=vi.fn(async()=>true);
const spool=mode==="spool" ? vi.spyOn(orphanCleanup,"createSandboxOrphanCleanupSpool").mockReturnValue({append,load:async()=>[],remove:async()=>{}}) : undefined;
const transaction=mode==="spool" ? vi.spyOn(db,"transaction").mockRejectedValue(Error("database unavailable")) : undefined;
const original=db.update.bind(db);
const update=vi.spyOn(db,"update").mockImplementation(((table:any)=>{
const query=original(table),set=query.set.bind(query);
query.set=((values:any)=>values.providerLeaseId === "task-2"
? {where:()=>({returning:async()=>{if(mode!=="ownership-lost") throw Error("replacement write failed");return [];}})}
: set(values)) as never;
return query;
}) as never);
try {
await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{kind:"describe"}}))
.rejects.toThrow("reconcile the same task");
} finally {update.mockRestore();transaction?.mockRestore();spool?.mockRestore();}
expect(f.calls).toContain("environmentReleaseLease");
if(mode==="spool") expect(append).toHaveBeenCalledWith(expect.objectContaining({providerLeaseId:"task-2",metadata:expect.objectContaining({driver:"plugin",runnerTaskDispatched:false})}));
const rows=await db.select().from(environmentLeases).where(eq(environmentLeases.companyId,f.companyId));
expect(rows.filter(row=>row.providerLeaseId === "task-2")).toEqual(mode==="cleanup-error"
? [expect.objectContaining({status:"pending_cleanup",failureReason:"plugin_task_replacement_unowned",metadata:expect.objectContaining({runnerTaskDispatched:false})})] : []);
});
it.each(["added","removed"] as const)("retires retained mounts when an authorized project is %s after controller restart",async mode=>{
const projectId=randomUUID(),f=await fixture(mode==="removed" ? [projectId] : []);
if(mode==="added") await db.insert(projects).values({id:projectId,companyId:f.companyId,name:"Added project"});
f.setProjects(mode==="added" ? [projectId] : []);await f.nextRun();
const next=await f.acquire();
expect(next.lease.providerLeaseId).not.toBe(f.first.lease.providerLeaseId);
expect(next.lease.metadata?.runnerTaskProjectIds).toEqual(mode==="added" ? [projectId] : []);
expect(f.calls).toContain("stop");expect(f.calls).toContain("complete");
expect(f.calls).not.toContain("environmentResumeLease");
await expect(resolvePluginTaskResume(db,{companyId:f.companyId,leaseId:next.lease.id,priorIdentity:f.identity})).resolves.toBe(f.first.lease.providerLeaseId);
});
it.each(["later","none","bounded","violating","missing-bounded","invalid","expired"] as const)("replaces task expiry without losing caller bounds: %s",async mode=>{
const bounded=["bounded","violating","missing-bounded"].includes(mode);
const deadline=bounded ? new Date("2099-03-01T00:00:00.000Z") : undefined;
const f=await fixture([],"2099-01-01T00:00:00.000Z",deadline);
await f.nextRun();const next=await f.acquire();f.stopped();
const expiry=mode === "invalid" ? "invalid-expiry" : mode === "expired" ? "2000-01-01T00:00:00.000Z" : mode === "none" || mode === "missing-bounded" ? undefined
: mode === "violating" ? "2099-04-01T00:00:00.000Z" : "2099-02-01T00:00:00.000Z";
f.setExpiry(expiry);
const replacing=f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{kind:"describe"}});
if(["violating","missing-bounded","invalid","expired"].includes(mode)) {
await expect(replacing).rejects.toThrow("reconcile the same task");
expect(f.calls).toContain("environmentReleaseLease");
} else {
const description=await replacing;
const [saved]=await db.select().from(environmentLeases).where(eq(environmentLeases.id,next.lease.id));
expect(saved.providerLeaseId).toBe(description.taskId);
expect(saved.expiresAt?.toISOString() ?? null).toBe(expiry ?? null);
}
expect(f.acquireRequests.at(-1)?.requestedExpiresAt).toBe(deadline?.toISOString());
});
});
@@ -111,6 +111,8 @@ const pluginSandboxEnvironmentConfigSchema = z.object({
}).catchall(z.unknown());
const pluginEnvironmentConfigSchema = z.object({
runnerLifecycleMode: z.enum(["inherit", "per_turn", "warm"]).optional(),
runnerIdleTimeoutMs: z.number().int().positive().max(86_400_000).optional(),
pluginKey: z.string().min(1),
driverKey: z.string().min(1).regex(
/^[a-z0-9][a-z0-9._-]*$/,
@@ -263,16 +263,28 @@ export async function resolveEnvironmentExecutionTarget(input: {
throw new Error("Plugin environment does not provide native Runner task execution");
}
// The Runner resolves its workspace locally; this is only a relative scope marker.
return {
kind: "remote", transport: "sandbox", providerKey: "plugin",
const acquisition:SandboxLeaseAcquisition = {outcome:lease.metadata?.runnerTaskReused === true ? "resumed" : "created",providerLeaseId:lease.providerLeaseId!};
const target: Extract<AdapterExecutionTarget,{kind:"remote";transport:"sandbox"}> = {
kind: "remote", transport: "sandbox", providerKey: lease.provider ?? "plugin",
environmentId: input.environment.id, leaseId: lease.id, remoteCwd: ".",
runnerLifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
sandboxLeaseAcquisition: { outcome: "created", providerLeaseId: lease.providerLeaseId! },
runnerLifecyclePolicy: lease.metadata?.runnerTaskLifecyclePolicy as { mode: "warm"; idleTimeoutMs: number } | undefined ?? { mode: "per_turn", idleTimeoutMs: null },
reusableLeaseConfigured: lease.leasePolicy === "reuse_by_environment",
effectiveCapabilities: { reusableLeases: lease.leasePolicy === "reuse_by_environment", nativeSyncIn: false, nativeSyncOut: false,
persistentProcessSessions: false, independentControlCommands: false, incrementalSessionOutput: false,
concurrentSyncOperations: false, duplexCommandStream: false, runnerWebSocketIngress: true },
sandboxLeaseAcquisition:acquisition,
...(lease.metadata?.runnerTaskProbe ? { runnerTaskProbe: lease.metadata.runnerTaskProbe as { runId: string; sessionId: string } } : {}),
runnerTask: operation => runtime.task({ companyId: input.companyId, leaseId: lease.id,
operation: environmentTaskOperationSchema.parse(operation) }),
runnerTask: async operation => {
const result = await runtime.task({ companyId: input.companyId, leaseId: lease.id,
operation: environmentTaskOperationSchema.parse(operation) });
if (result.kind === "description" && result.taskId !== acquisition.providerLeaseId) {
acquisition.outcome = "replacement";
acquisition.providerLeaseId = result.taskId;
}
return result;
},
};
return target;
}
if (input.environment.driver === "sandbox") {
@@ -214,6 +214,7 @@ export function environmentRunOrchestrator(
heartbeatRunId: string;
persistedExecutionWorkspace: Pick<ExecutionWorkspace, "id" | "mode"> | null;
executionWorkspaceSettings: IssueExecutionWorkspaceSettings | null;
runnerTaskProjectIds?: readonly string[];
adapterType: string | null;
}): Promise<EnvironmentRuntimeLeaseRecord> {
try {
@@ -277,6 +278,7 @@ export function environmentRunOrchestrator(
agentId: string;
persistedExecutionWorkspace: Pick<ExecutionWorkspace, "id" | "mode"> | null;
executionWorkspaceSettings: IssueExecutionWorkspaceSettings | null;
runnerTaskProjectIds?: readonly string[];
/** Existing live runner recovery must use its original active lease, including ephemeral leases. */
reattachRemoteLease?: { leaseId: string; providerLeaseId: string; remoteCwd: string };
/** A new controller must stop the previous plugin task before allocating a replacement. */
@@ -317,6 +319,7 @@ export function environmentRunOrchestrator(
heartbeatRunId: input.heartbeatRunId,
persistedExecutionWorkspace: input.persistedExecutionWorkspace,
executionWorkspaceSettings: input.executionWorkspaceSettings,
runnerTaskProjectIds: input.runnerTaskProjectIds,
adapterType: input.adapterType ?? null,
};
let leaseRecord: EnvironmentRuntimeLeaseRecord;
+50 -27
View File
@@ -457,6 +457,7 @@ export interface EnvironmentDriverAcquireInput {
executionWorkspaceId: string | null;
executionWorkspaceMode: ExecutionWorkspace["mode"] | null;
executionWorkspaceSettings: IssueExecutionWorkspaceSettings | null;
runnerTaskProjectIds?: readonly string[];
/**
* The harness/adapter type for this run (the agent's adapter). Drivers that
* materialize a per-run sandbox use it to select the runtime image so a single
@@ -1388,34 +1389,13 @@ function createSandboxEnvironmentDriver(
return true;
};
// Load the durable spool into the in-process buffer exactly once. The first
// flush after a restart runs this load, so the records a previous process
// could not land re-enter the flush path. A later flush sees the resolved
// promise and never reloads, so a record this process already buffered never
// doubles. The load runs before the flush splices the buffer, so a
// restart-recovered record flushes on the same tick.
let spoolLoadPromise: Promise<void> | null = null;
const ensureSpoolLoaded = (): Promise<void> => {
if (!spoolLoadPromise) {
spoolLoadPromise = (async () => {
const persisted = await orphanCleanupSpool.load();
for (const record of persisted) {
enqueueDeferredOrphanCleanup(record);
}
})();
}
return spoolLoadPromise;
};
const flushDeferredOrphanCleanups = async (): Promise<{ recovered: number; pending: number }> => {
// Load the durable spool into the buffer first, so a restart-recovered record
// flushes on this tick. The load runs once, then a resolved promise returns.
await ensureSpoolLoaded();
const flushDeferredOrphanCleanupsOnce = async (): Promise<{ recovered: number; pending: number }> => {
// Other task paths can append after startup; each pass must ingest their records.
for (const record of await orphanCleanupSpool.load()) enqueueDeferredOrphanCleanup(record);
if (deferredOrphanCleanups.length === 0) {
return { recovered: 0, pending: 0 };
}
// Take the whole batch synchronously before the first await, so a second
// concurrent flush sees an empty buffer and never re-inserts the same record.
// New acquisitions can enqueue while this batch is being persisted.
const batch = deferredOrphanCleanups.splice(0, deferredOrphanCleanups.length);
let recovered = 0;
for (const record of batch) {
@@ -1463,6 +1443,11 @@ function createSandboxEnvironmentDriver(
}
return { recovered, pending: deferredOrphanCleanups.length };
};
let orphanFlush: Promise<{ recovered: number; pending: number }> | null = null;
const flushDeferredOrphanCleanups = (): Promise<{ recovered: number; pending: number }> => {
orphanFlush ??= flushDeferredOrphanCleanupsOnce().finally(() => { orphanFlush = null; });
return orphanFlush;
};
// Build the safe diagnostic fields for an orphan record. The provider
// identifiers are the only fields any diagnostic log emits, and they carry no
@@ -3670,7 +3655,38 @@ function createPluginEnvironmentDriver(
driver.defaultAcquireTimeoutMs,
);
const timeoutOverride: [number?] = rpcTimeoutMs === undefined ? [] : [rpcTimeoutMs];
const providerLease = await workerManager.call(plugin.id, "environmentAcquireLease", {
const warm = input.adapterType === "paperclip_runner" && parsed.config.runnerLifecycleMode === "warm" &&
driver.warmTaskVersion === 1 && !!input.issueId && !!input.agentId && !!input.heartbeatRunId &&
workerManager.getWorker(plugin.id)?.supportedMethods.includes("environmentResumeLease");
const lifecyclePolicy = warm ? { mode: "warm", idleTimeoutMs: parsed.config.runnerIdleTimeoutMs ?? 300_000 } : { mode: "per_turn", idleTimeoutMs: null };
const fingerprint = createHash("sha256").update(JSON.stringify([parsed.config, input.agentId, input.runnerTaskProjectIds ?? []])).digest("hex");
const candidates = warm ? await db.select().from(environmentLeases).where(and(
eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, input.environment.id),
eq(environmentLeases.status, "retained"), eq(environmentLeases.issueId, input.issueId!),
sql`${environmentLeases.executionWorkspaceId} IS NOT DISTINCT FROM ${input.executionWorkspaceId ?? null}::uuid`,
sql`${environmentLeases.metadata}->>'agentId' = ${input.agentId!}`,
sql`${environmentLeases.metadata}->>'pluginId' = ${plugin.id}`,
sql`${environmentLeases.metadata}->>'driverKey' = ${parsed.config.driverKey}`,
)) : [];
if (candidates.length > 1) throw new Error("Ambiguous retained plugin task ownership");
const previous = candidates[0];
let resumed: PluginEnvironmentLease | undefined;
if (previous?.providerLeaseId && previous.metadata?.runnerTaskFingerprint !== fingerprint) {
// A changed mount scope must retire the old process before checkpoint recovery.
await this.releaseRunLease({environment:input.environment,lease:toEnvironmentLeaseSnapshot(previous),status:"released"});
} else if (previous?.providerLeaseId) {
// Resume only inspects the exact computer; an unavailable reply must not create a replacement.
resumed = await workerManager.call(plugin.id, "environmentResumeLease", {
driverKey: parsed.config.driverKey, companyId: input.companyId, environmentId: input.environment.id,
issueId: input.issueId, config: parsed.config.driverConfig, runId: input.heartbeatRunId!,
providerLeaseId: previous.providerLeaseId, leaseMetadata: previous.metadata ?? undefined,
});
if (!resumed?.providerLeaseId) {
await this.releaseRunLease({ environment: input.environment, lease: toEnvironmentLeaseSnapshot(previous), status: "released" });
resumed = undefined;
}
}
const providerLease = resumed ?? await workerManager.call(plugin.id, "environmentAcquireLease", {
driverKey: parsed.config.driverKey,
companyId: input.companyId,
environmentId: input.environment.id,
@@ -3695,11 +3711,16 @@ function createPluginEnvironmentDriver(
executionWorkspaceId: input.executionWorkspaceId,
issueId: input.issueId,
heartbeatRunId: input.heartbeatRunId,
leasePolicy: "ephemeral",
leasePolicy: warm ? "reuse_by_environment" : "ephemeral",
...(resumed && previous ? { replacesReusableLeaseId: previous.id } : {}),
provider: `plugin:${parsed.config.pluginKey}:${parsed.config.driverKey}`,
providerLeaseId: providerLease.providerLeaseId,
expiresAt: providerAttestedLeaseExpiry(input.requestedExpiresAt, parseExpiresAt(providerLease.expiresAt)),
metadata: {
...(resumed && previous ? { ...previous.metadata, runnerTaskReused: true, runnerTaskPreviousLeaseId: previous.id } : {}),
runnerTaskLifecyclePolicy: lifecyclePolicy, runnerTaskFingerprint: fingerprint,
runnerTaskRequestedExpiresAt:requestedExpiresAtParam(input.requestedExpiresAt) ?? null,
...(warm ? {runnerTaskProjectIds:[...input.runnerTaskProjectIds ?? []]} : {}),remoteCwd:".",
...(input.agentId ? { agentId: input.agentId } : {}),
providerMetadata: providerLease.metadata ?? {},
environmentId: input.environment.id,
@@ -4105,6 +4126,7 @@ export function environmentRuntimeService(
heartbeatRunId: string | null;
persistedExecutionWorkspace: Pick<ExecutionWorkspace, "id" | "mode"> | null;
executionWorkspaceSettings?: IssueExecutionWorkspaceSettings | null;
runnerTaskProjectIds?: readonly string[];
/** The agent's adapter type for this run (mixed-harness environments). */
adapterType?: string | null;
resolveSnapshotIdentity?: boolean;
@@ -4145,6 +4167,7 @@ export function environmentRuntimeService(
executionWorkspaceId: leaseContext.executionWorkspaceId,
executionWorkspaceMode: leaseContext.executionWorkspaceMode,
executionWorkspaceSettings: input.executionWorkspaceSettings ?? null,
runnerTaskProjectIds: input.runnerTaskProjectIds,
adapterType: input.adapterType ?? null,
resolveSnapshotIdentity: input.resolveSnapshotIdentity,
applyCustomImageTemplate: input.applyCustomImageTemplate ?? false,
+103 -7
View File
@@ -1,3 +1,6 @@
import { environmentService } from "./environments.js";
import { createSandboxOrphanCleanupSpool, type DeferredOrphanCleanupRecord } from "./sandbox-orphan-cleanup-spool.js";
import { logger } from "../middleware/logger.js";
import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js";
import { readWorkspaceProviderBinding } from "@paperclipai/shared";
import { settlePluginTaskCredentialHandoff } from "./plugin-task-credential-writeback.js";
@@ -5,7 +8,7 @@ import { and, eq, inArray, or, sql } from "drizzle-orm";
import { environmentLeases, executionWorkspaces, environments, heartbeatRuns, issues, projects, type Db } from "@paperclipai/db";
import { freshWorkspaceSchema, environmentTaskOperationSchema, parseEnvironmentTaskResult, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk";
import { pluginRegistryService } from "./plugin-registry.js";
import { hasRemoteTerminationReceipt, remoteExecutionHasStopped } from "./remote-execution-termination.js";
import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt } from "./remote-execution-termination.js";
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
export async function pluginEnvironmentSupportsTasks(db: Db, workers: PluginWorkerManager, lease: { companyId: string; metadata: Record<string, unknown> | null }) {
@@ -43,7 +46,7 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
const activeProbe = lease.heartbeatRunId === null && lease.issueId === null && probe
&& typeof probe.runId === "string" && typeof probe.sessionId === "string"
&& typeof probe.expiresAtMs === "number" && probe.expiresAtMs > Date.now();
if ((operation.kind === "read_file" || operation.kind === "describe" || operation.kind === "prepare" || operation.kind === "submit" || operation.kind === "connection" || operation.kind === "read_state" || operation.kind === "archive_state" || operation.kind === "check_project_auth" || operation.kind === "copilot_metadata") &&
if ((operation.kind === "read_file" || operation.kind === "describe" || operation.kind === "prepare" || operation.kind === "submit" || operation.kind === "connection" || (operation.kind === "read_state" && metadata.runnerTaskExecutionStopped !== true) || operation.kind === "archive_state" || operation.kind === "check_project_auth" || operation.kind === "copilot_metadata") &&
(metadata.runnerTaskRetiring === true || !environment || (!activeProbe && run?.status !== "running") || lease.status !== "active" || (lease.expiresAt && lease.expiresAt.getTime() <= Date.now()))) throw new Error("Plugin-provided Runner execution lease is not active");
if (operation.kind === "copilot_metadata" && !activeProbe) throw new Error("Copilot metadata requires a private setup lease");
const checkpointIdentity = operation.kind === "read_state" || operation.kind === "read_file" ? operation.identity
@@ -58,12 +61,20 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
const scopesProjects = launches || operation.kind === "check_project_auth" || operation.kind === "read_file";
const resumeFrom = launches || checkpointIdentity ? (operation as { resumeFrom?: string }).resumeFrom : undefined;
const savedIdentity = metadata.runnerTaskIdentity as Record<string, unknown> | undefined;
const ownCheckpoint = checkpointIdentity && !resumeFrom && savedIdentity
&& savedIdentity.runnerId === checkpointIdentity.runnerInstanceId && savedIdentity.leaseId === checkpointIdentity.environmentLeaseId
&& savedIdentity.runId === checkpointIdentity.runId && savedIdentity.sessionId === checkpointIdentity.normalizedSessionId
&& savedIdentity.turnId === checkpointIdentity.turnId && savedIdentity.itemId === checkpointIdentity.itemId;
const prior = operation.kind === "prepare" ? operation.priorIdentity : undefined;
const priorSaved = metadata.runnerTaskPriorIdentity as Record<string, unknown> | undefined;
const matches = (identity: typeof prior, saved: Record<string, unknown> | undefined) => !!identity && !!saved &&
identity.runnerInstanceId === saved.runnerId && identity.environmentLeaseId === saved.leaseId &&
identity.normalizedSessionId === saved.sessionId && identity.runId === saved.runId &&
identity.turnId === saved.turnId && identity.itemId === saved.itemId;
const ownCheckpoint = checkpointIdentity && !resumeFrom &&
(matches(checkpointIdentity,savedIdentity) || operation.kind !== "read_file" && matches(checkpointIdentity,priorSaved));
if (checkpointIdentity && !resumeFrom && !ownCheckpoint) throw new Error("Plugin Runner state identity mismatch");
if (runner && ((!checkpointIdentity || !resumeFrom) && runner.runId !== (activeProbe ? probe.runId : run?.id) || (!resumeFrom && runner.leaseId !== lease.id && !ownCheckpoint)
const warmAttachment = !!prior && metadata.runnerTaskReused === true &&
(matches(prior, savedIdentity) || matches(prior, priorSaved) && operation.kind === "prepare" && matches(operation.requirements.identity, savedIdentity)) && runner?.runnerId === prior.runnerInstanceId &&
runner?.leaseId === prior.environmentLeaseId && runner?.sessionId === prior.normalizedSessionId;
if (prior && !warmAttachment) throw new Error("Plugin warm task prior identity mismatch");
if (runner && ((!checkpointIdentity || !resumeFrom) && !(ownCheckpoint && operation.kind !== "read_file") && runner.runId !== (activeProbe ? probe.runId : run?.id) || (!resumeFrom && runner.leaseId !== lease.id && !ownCheckpoint && !warmAttachment)
|| (activeProbe && (runner.sessionId !== probe.sessionId || (launches && operation.projectIds.length > 0))))) {
throw new Error("Plugin-provided Runner execution identity mismatch");
}
@@ -71,6 +82,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
const sources = await db.select({ runId: environmentLeases.heartbeatRunId, metadata: environmentLeases.metadata }).from(environmentLeases).where(and(
eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.provider, lease.provider!),
eq(environmentLeases.providerLeaseId, resumeFrom),
sql`${environmentLeases.metadata}->>'runnerTaskSessionRetained' = 'true'`,
checkpointIdentity ? eq(environmentLeases.heartbeatRunId, checkpointIdentity.runId) : undefined,
)).limit(2);
const priorRunId = sources.length === 1 ? sources[0]!.runId : null;
const sourceIdentity = sources.length === 1 ? sources[0]!.metadata?.runnerTaskIdentity as Record<string, unknown> | undefined : undefined;
@@ -115,6 +128,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
issueId: lease.issueId, ...freshWorkspace, existing: freshWorkspace });
}
const projectIds = scopesProjects ? operation.projectIds : [];
if (scopesProjects && Array.isArray(metadata.runnerTaskProjectIds) &&
JSON.stringify(projectIds) !== JSON.stringify(metadata.runnerTaskProjectIds)) throw new Error("Plugin Runner project scope changed");
if (activeProbe && projectIds.length > 0) throw new Error("Plugin Runner probe cannot use project mounts");
if (projectIds.length > 0) {
const authorizedProjects = await db.select({ id: projects.id }).from(projects)
@@ -166,7 +181,75 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
)).returning({ id: environmentLeases.id });
if (marked.length !== 1) throw new Error("Plugin-provided Runner execution lease is not active");
}
if (warmAttachment && operation.kind === "prepare") {
const i = operation.requirements.identity;
const marked = await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({
runnerTaskPriorIdentity: { runnerId: prior!.runnerInstanceId, leaseId: prior!.environmentLeaseId, runId: prior!.runId,
sessionId: prior!.normalizedSessionId, turnId: prior!.turnId, itemId: prior!.itemId },
runnerTaskIdentity: { runnerId: i.runnerInstanceId, leaseId: i.environmentLeaseId, runId: i.runId,
sessionId: i.normalizedSessionId, turnId: i.turnId, itemId: i.itemId },
})}::jsonb` }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.status, "active"),
sql`coalesce(${environmentLeases.metadata}->>'runnerTaskRetiring', 'false') != 'true'`,
sql`${environmentLeases.metadata}->'runnerTaskIdentity' = ${JSON.stringify(savedIdentity)}::jsonb`)).returning({id:environmentLeases.id});
if (marked.length !== 1) throw new Error("Plugin warm task attachment lost authority");
}
try {
if (operation.kind === "describe" && metadata.runnerTaskReused === true) {
const state = parseEnvironmentTaskResult({kind:"status"}, taskId, await workers.call(plugin.id, "environmentTask", {
driverKey: metadata.driverKey, companyId: input.companyId, environmentId: environment?.id ?? null,
issueId: lease.issueId, config: driverConfig, taskId, runId: run?.id ?? null, agentId: run?.agentId ?? null,
projectIds, lease: {providerLeaseId: taskId, metadata}, operation:{kind:"status"},
}));
if (state.kind === "status" && state.executionStopped === true) {
await executeEnvironmentTask(db, workers, {...input, operation:{kind:"status"}});
await executeEnvironmentTask(db, workers, {...input, operation:{kind:"complete"}});
// Preserve the predecessor's stop proof for checkpoint authorization.
if (typeof metadata.runnerTaskPreviousLeaseId === "string") await db.update(environmentLeases).set({
metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({runnerTaskExecutionStopped:true,
runnerTaskSessionRetained:state.sessionStateAvailable === true})}::jsonb`,
}).where(and(eq(environmentLeases.id, metadata.runnerTaskPreviousLeaseId), eq(environmentLeases.companyId, input.companyId)));
const fresh = await workers.call(plugin.id, "environmentAcquireLease", {driverKey:metadata.driverKey,
companyId:input.companyId,environmentId:environment!.id,issueId:lease.issueId,config:driverConfig,
runId:run!.id,agentId:run!.agentId,adapterType:"paperclip_runner",
...(typeof metadata.runnerTaskRequestedExpiresAt === "string" ? {requestedExpiresAt:metadata.runnerTaskRequestedExpiresAt} : {})});
if (!fresh.providerLeaseId || fresh.providerLeaseId === taskId) throw new Error("Plugin replacement task identity unavailable");
const {runnerTaskIdentity:_identity,runnerTaskPriorIdentity:_prior,runnerTaskCredentialHandoff:_handoff,
runnerTaskResumeFrom:_resume,runnerTaskPreviousLeaseId:_previous,...rest}=metadata;
const replacementMetadata={...rest,providerMetadata:fresh.metadata ?? {},runnerTaskReused:false,runnerTaskDispatched:false,
runnerTaskExecutionStopped:false,runnerTaskSessionRetained:false};
try {
const expiresAt=fresh.expiresAt ? new Date(fresh.expiresAt) : null;
const deadline=typeof metadata.runnerTaskRequestedExpiresAt === "string" ? Date.parse(metadata.runnerTaskRequestedExpiresAt) : null;
if ((expiresAt && (!Number.isFinite(expiresAt.getTime()) || expiresAt.getTime() <= Date.now())) ||
(deadline !== null && (!Number.isFinite(deadline) || !expiresAt || expiresAt.getTime() > deadline)))
throw new Error("Plugin replacement task expiry is not bounded");
const replacement = await db.update(environmentLeases).set({providerLeaseId:fresh.providerLeaseId,expiresAt,
metadata:replacementMetadata,
}).where(and(eq(environmentLeases.id,lease.id),eq(environmentLeases.providerLeaseId,taskId),
eq(environmentLeases.status,"active"))).returning({id:environmentLeases.id});
if (replacement.length !== 1) throw new Error("Plugin task replacement ownership changed");
} catch (error) {
try {
await workers.call(plugin.id,"environmentReleaseLease",{driverKey:metadata.driverKey,companyId:input.companyId,
environmentId:environment!.id,issueId:lease.issueId,config:driverConfig,providerLeaseId:fresh.providerLeaseId,
leaseMetadata:replacementMetadata});
} catch {
const orphan:DeferredOrphanCleanupRecord={companyId:input.companyId,environmentId:environment!.id,
executionWorkspaceId:lease.executionWorkspaceId,issueId:lease.issueId,heartbeatRunId:lease.heartbeatRunId,
provider:lease.provider!,providerLeaseId:fresh.providerLeaseId,metadata:replacementMetadata};
try {
await environmentService(db).insertPendingCleanupLease({...orphan,failureReason:"plugin_task_replacement_unowned"});
} catch {
// The existing orphan spool recovers cleanup even when the database is unavailable.
const persisted=await createSandboxOrphanCleanupSpool().append(orphan);
logger.error({providerLeaseId:fresh.providerLeaseId,persisted},"Plugin replacement cleanup requires recovery");
}
}
throw error;
}
return executeEnvironmentTask(db, workers, input);
}
}
if (operation.kind === "complete") {
await settlePluginTaskCredentialHandoff(db, { companyId: input.companyId, leaseId: lease.id }, metadata, async accountIdentity => {
const snapshot = await executeEnvironmentTask(db, workers, { ...input, operation: { kind: "credentials", accountIdentity } });
@@ -221,6 +304,19 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
if (operation.kind === "complete") {
if (freshWorkspace && metadata.runnerTaskExecutionStopped === true)
await releaseProviderRecoveryTransfers(input.companyId, freshWorkspace);
// A physical task can back several run leases. Confirm retirement for every prior owner.
if (lease.leasePolicy === "reuse_by_environment" && metadata.runnerTaskExecutionStopped === true) {
const owners=await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId,input.companyId),
eq(environmentLeases.provider,lease.provider!),eq(environmentLeases.providerLeaseId,taskId)));
for(const owner of owners) {
const receipt=remoteTerminationReceipt(owner,{providerLeaseId:taskId,state:"stopped"});
if(!receipt) continue;
await db.update(environmentLeases).set({
...(owner.status === "retained" ? {status:"released" as const,releasedAt:new Date(),cleanupStatus:"success" as const}:{}),
metadata:sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({remoteExecutionTermination:receipt,runnerTaskExecutionStopped:true})}::jsonb`,
}).where(eq(environmentLeases.id,owner.id));
}
}
await db.update(environmentLeases).set({
metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskDispatched":false}'::jsonb`,
}).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId),
@@ -607,6 +607,7 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary",
await db.insert(projectWorkspaces).values({ companyId: issue.companyId, projectId: readable.id, name: "Provider-only storage", cwd: unavailableCwd });
await db.update(issues).set({ description: [readable, privateProject].map(project => `[project](${buildProjectMentionHref(project.id)})`).join(" ") }).where(eq(issues.id, issue.id));
const stopped = new Error("Fixture reached plugin realization");
const acquire=vi.spyOn(envOrchestrator,"acquireForRun");
const realize = vi.spyOn(envOrchestrator, "realizeForRun").mockRejectedValue(stopped);
try {
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(stopped);
@@ -614,8 +615,11 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary",
runnerTaskProjectIds: enabled ? [issue.projectId, readable.id] : undefined,
executionWorkspace: expect.objectContaining({ additionalWorkspaces: [] }),
}));
expect(acquire).toHaveBeenCalledWith(expect.objectContaining({runnerTaskProjectIds:
enabled ? [issue.projectId,readable.id] : [issue.projectId]}));
await expect(access(unavailableCwd)).rejects.toMatchObject({ code: "ENOENT" });
} finally {
acquire.mockRestore();
vi.stubEnv("PAPERCLIP_MULTI_PROJECT_WORKSPACE_SYNC", "true");
vi.stubEnv("PAPERCLIP_MULTI_PROJECT_WORKSPACE_SYNC_REMOTE", "true");
}
@@ -1106,6 +1106,9 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace
adapterType: agent.adapterType,
adapterConfig: parseObject(agent.adapterConfig),
admittedLifecycleMode: persistedNativeExecutionInput?.session.lifecyclePolicy.mode,
runnerTaskProjectIds:[...new Set([executionWorkspace.projectId,...resolvedWorkspace.runnerTaskProjectIds ?? [],
...(executionWorkspace.additionalWorkspaces ?? []).map(workspace=>workspace.projectId)]
.filter((id):id is string=>Boolean(id)))],
recoverPluginTask: runOptions.nativeRestartRecovery?.kind === "resume_dead_runner" || runOptions.nativeRestartRecovery?.kind === "bootstrap_incomplete",
issueId: issueId ?? null,
heartbeatRunId: run.id,
@@ -10378,14 +10378,14 @@ describe("runnerd provider runtime wiring", () => {
await rm(isolatedStateDirectory, { recursive: true, force: true });
});
it.each(["unchanged", "changed", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment"] as const)("rotates an owned warm remote runner before checking changed configuration: %s", async scenario => {
it.each(["unchanged", "changed", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment", "plugin-replaced"] as const)("rotates an owned warm remote runner before checking changed configuration: %s", async scenario => {
const first = { ...scopedNativeExecution(`warm-admission-${scenario}`),
schema: "paperclip.native-execution-input.v5" as const,
provider: { kind: "codex" as const, model: null, approvalPolicy: "never" as const },
executionMode: "default" as const, planningContext: null, runtimeContext: nativeRuntimeContextFixture() };
first.session.lifecyclePolicy = { mode: "warm", idleTimeoutMs: 60_000 };
const next = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-next` },
session: { ...first.session, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: ["unchanged", "runtime-context", "stale-owner", "missing-owner"].includes(scenario) ? 60_000 : 90_000 } } };
session: { ...first.session, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: ["unchanged", "runtime-context", "stale-owner", "missing-owner", "plugin-replaced"].includes(scenario) ? 60_000 : 90_000 } } };
if (scenario === "runtime-context") {
next.runtimeContext = { ...first.runtimeContext,
mcp: { ...first.runtimeContext.mcp, digest: "1".repeat(64) } };
@@ -10396,6 +10396,24 @@ describe("runnerd provider runtime wiring", () => {
remoteCwd: "/home/daytona/workspace", runner: { execute: vi.fn() } };
const identity = { runId: first.binding.runId, normalizedSessionId: first.session.normalizedSessionId,
runnerInstanceId: `runner-${scenario}`, environmentLeaseId: first.binding.executionWorkspaceId };
let replacementPending=false;
const acquisition={outcome:"resumed" as "resumed"|"replacement",providerLeaseId:"task-first"};
const task=vi.fn(async()=>{
if(replacementPending) {acquisition.outcome="replacement";acquisition.providerLeaseId="task-next";replacementPending=false;}
return {kind:"description",taskId:acquisition.providerLeaseId,launchPreparationVersion:1,projectAuthCheckVersion:1,warmTaskVersion:1,
runner:{version:"fixture",digest:`sha256:${"a".repeat(64)}`}};
});
const pluginTarget=scenario === "plugin-replaced" ? {...target,remoteCwd:".",leaseId:identity.environmentLeaseId,
runnerTask:task,sandboxLeaseAcquisition:acquisition} : target;
const home=join(isolatedStateDirectory,"warm-plugin-home");
if(scenario === "plugin-replaced") {
await mkdir(home,{recursive:true});await writeFile(join(home,"auth.json"),JSON.stringify({OPENAI_API_KEY:"synthetic-fixture"}),{mode:0o600});
}
const pluginOptions=scenario === "plugin-replaced" ? {runnerIngressAuthorized:true,managedAiCredentialHome:home,
managedAiCredentialIdentity:"revision",managedAiAccountIdentity:"account"} : {};
const resume=scenario === "plugin-replaced" ? vi.spyOn(environmentTaskRuntime,"resolvePluginTaskResume").mockResolvedValue("task-first") : undefined;
const owners=scenario === "plugin-replaced" ? vi.spyOn(pluginTaskRuntime,"createPluginRunnerTaskOwner") : undefined;
const root = await initializeScopedRunnerRoot(first, identity.runnerInstanceId);
await writeRunnerAuthority(root, identity, "ready");
const order: string[] = [];
@@ -10428,7 +10446,7 @@ describe("runnerd provider runtime wiring", () => {
: ordinarySelect(fields as never)) as Db["select"];
try {
await executePaperclipNativeSession({ db: leaseDb(first), execution: first,
runnerInstanceId: identity.runnerInstanceId, useRunnerd: true, runnerExecutionTarget: target as never });
runnerInstanceId: identity.runnerInstanceId, useRunnerd: true, runnerExecutionTarget: pluginTarget as never,...pluginOptions });
if (scenario === "stale-owner") await writeRunnerAuthority(root, { ...identity, runId: older.binding.runId }, "ready");
if (scenario === "missing-owner") {
await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "fixture owner loss" });
@@ -10436,9 +10454,11 @@ describe("runnerd provider runtime wiring", () => {
order.length = 0;
firstSession.close.mockClear();
}
if(scenario === "plugin-replaced") replacementPending=true;
const continuation = executePaperclipNativeSession({ db: nextDb, execution: next,
runnerInstanceId: `runner-next-${scenario}`, useRunnerd: true,
runnerExecutionTarget: (scenario === "foreign-environment" ? { ...target, environmentId: `${environmentId}-other` } : target) as never });
runnerExecutionTarget: (scenario === "foreign-environment" ? { ...target, environmentId: `${environmentId}-other` }
: scenario === "plugin-replaced" ? {...pluginTarget,leaseId:"next-core-lease"} : target) as never,...pluginOptions });
if (["stale-owner", "active-owner", "missing-owner", "foreign-environment"].includes(scenario)) {
await expect(continuation).rejects.toThrow(scenario === "active-owner"
? "prior_owner_active" : "prior_owner_terminal_state_indeterminate");
@@ -10451,6 +10471,11 @@ describe("runnerd provider runtime wiring", () => {
expect(order).toEqual(["retire"]);
} else {
await expect(continuation).resolves.toMatchObject({ exitCode: 0 });
if(scenario === "plugin-replaced") {
expect(resume).toHaveBeenCalledWith(nextDb,expect.objectContaining({leaseId:"next-core-lease"}));
expect(owners!.mock.calls.at(-1)![0]).toMatchObject({resumeFrom:"task-first"});
expect(owners!.mock.calls.at(-1)![0].adoptIdentity).toBeUndefined();
}
expect(order).toEqual(scenario === "unchanged" ? ["execute-next"] : ["retire", "execute-next"]);
if (scenario === "unchanged") {
await writeRunnerAuthority(root, { ...identity, runId: next.binding.runId }, "ready");
@@ -10475,6 +10500,7 @@ describe("runnerd provider runtime wiring", () => {
}
} finally {
failRetirement = false;
resume?.mockRestore();owners?.mockRestore();
await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "test cleanup" });
}
});
@@ -11114,7 +11140,7 @@ describe("runnerd provider runtime wiring", () => {
const driver = { prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration };
state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []);
const readWorkspaceFile = vi.fn(async () => Buffer.from("proof"));
const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile });
const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, rebind:vi.fn() });
const task = vi.fn(async (operation: unknown) => {
environmentTaskOperationSchema.parse(operation);
return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
@@ -11170,6 +11196,58 @@ describe("runnerd provider runtime wiring", () => {
} finally { owner.mockRestore(); }
});
it.each(["codex", "fresh", "missing-capability", "missing-account", "acpx"] as const)("negotiates task-only warm execution: %s", async mode => {
const current = scopedNativeExecution(`plugin-warm-${mode}`);
current.session.lifecyclePolicy = {mode:"warm",idleTimeoutMs:300000};
if(mode==="acpx") current.provider={kind:"acpx",agent:"claude",model:"fixture",permissionMode:"deny-all"} as never;
const home=await mkdtemp(join(tmpdir(),"plugin-warm-account-"));
await writeFile(join(home,"auth.json"),JSON.stringify({OPENAI_API_KEY:"synthetic-fixture"}),{mode:0o600});
const owner=vi.spyOn(pluginTaskRuntime,"createPluginRunnerTaskOwner");
const task=vi.fn(async(operation:any)=> operation.kind === "stop" || operation.kind === "complete" ? {kind:"accepted",taskId:"warm-task"}
: operation.kind === "status" ? {kind:"status",taskId:"warm-task",phase:"completed",executionStopped:true}
: {kind:"description",taskId:"warm-task",launchPreparationVersion:1,projectAuthCheckVersion:1,
...(mode!=="missing-capability" ? {warmTaskVersion:1}:{}),runner:{version:"fixture",digest:`sha256:${"a".repeat(64)}`}});
state.createBackend.mockClear();state.createTransport.mockClear();
try {
const backend=createRunnerdBackend({db:leaseDb(current),execution:current,runnerInstanceId:"runner",runnerIngressAuthorized:true,
managedAiCredentialHome:home,managedAiCredentialIdentity:mode==="missing-account" ? undefined : "revision",managedAiAccountIdentity:"account",
runnerExecutionTarget:{kind:"remote",transport:"sandbox",leaseId:"lease",remoteCwd:".",runnerTask:task,
...(mode==="fresh" ? {sandboxLeaseAcquisition:{outcome:"resumed",providerLeaseId:"old-task"}} : {})}});
if(mode!=="codex" && mode!=="fresh") {await expect(backend).rejects.toThrow(mode==="acpx" ? "Codex workspace" : "capability or account");return;}
await backend;
if(mode==="fresh") expect(task.mock.calls.map(([operation])=>operation.kind)).toEqual(["stop","status","complete","describe"]);
state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
expect(state.createTransport.mock.calls.at(-1)![0]).toMatchObject({lifecyclePolicy:{mode:"warm",idleTimeoutMs:300000}});
expect(owner.mock.calls.at(-1)![0]).toMatchObject({warmTaskVersion:1,accountIdentity:"account",onCredentials:undefined});
} finally {owner.mockRestore();await rm(home,{recursive:true,force:true});}
});
it.each([true,false])("reconnects task-only warm recovery under the saved identity before reuse or retirement: %s", async reuseProcess => {
const current=scopedNativeExecution(`plugin-restart-${reuseProcess}`);
current.session.lifecyclePolicy={mode:"warm",idleTimeoutMs:300000};
const identity={runnerInstanceId:"old-runner",environmentLeaseId:"old-lease",runId:"old-run",
normalizedSessionId:current.session.normalizedSessionId!,turnId:"old-turn",itemId:"old-item"};
const home=await mkdtemp(join(tmpdir(),"plugin-restart-account-"));
await writeFile(join(home,"auth.json"),JSON.stringify({OPENAI_API_KEY:"synthetic-fixture"}),{mode:0o600});
const owner=vi.spyOn(pluginTaskRuntime,"createPluginRunnerTaskOwner");
const task=vi.fn(async()=>({kind:"description",taskId:"saved-task",launchPreparationVersion:1,projectAuthCheckVersion:1,
warmTaskVersion:1,runner:{version:"fixture",digest:`sha256:${"a".repeat(64)}`}}));
state.createBackend.mockClear();state.createTransport.mockClear();
try {
await createRunnerdBackend({db:leaseDb(current),execution:current,runnerInstanceId:"new-runner",runnerIngressAuthorized:true,
managedAiCredentialHome:home,managedAiCredentialIdentity:reuseProcess ? "old-revision" : "new-revision",managedAiAccountIdentity:"account",
managedAiCredentialWriteback:{} as never,
remoteWarmRecovery:{identity,reuseProcess,previousExecution:current,checkpoint:{} as never,
remote:{providerLeaseId:"saved-task",remoteCwd:"."},revalidate:async()=>{}},
runnerExecutionTarget:{kind:"remote",transport:"sandbox",leaseId:"new-lease",remoteCwd:".",runnerTask:task,
sandboxLeaseAcquisition:{outcome:"resumed",providerLeaseId:"saved-task"}}});
state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
expect(state.createTransport.mock.calls.at(-1)![0].prpIdentity).toEqual(identity);
expect(owner.mock.calls.at(-1)![0].adoptIdentity).toEqual(identity);
expect(task.mock.calls).toEqual([[{kind:"describe"}]]);
} finally {owner.mockRestore();await rm(home,{recursive:true,force:true});}
});
it("keeps assigned MCP credentials on the control plane for remote Codex", async () => {
const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test");
@@ -472,6 +472,7 @@ export type NativeInstructionWorkingCopy = {
};
type WarmNativeSession = {
pluginTaskId?: string;
observabilityCompatibilityKey?: string;
runId: string;
instructionWorkingCopy?: NativeInstructionWorkingCopy;
@@ -517,6 +518,7 @@ async function closeWarmNativeSession(entry: WarmNativeSession, reason: string,
}
const warmNativeSessions = new Map<string, WarmNativeSession>();
const pluginTaskSessionOwners = new WeakMap<NativeSession, ReturnType<typeof createPluginRunnerTaskOwner>>();
// Closing a remote owner saves its checkpoint asynchronously. A new turn must
// not inspect or quarantine that owner's state until the save has finished.
const closingWarmNativeSessions = new Map<string, Promise<void>>();
@@ -4891,7 +4893,7 @@ async function migrateRunnerdStateRootForExecution(input: {
const taskTarget = input.runnerExecutionTarget;
if (input.restartRecovery?.kind === "resume_dead_runner" && durableIdentityMatchesExecution(identity, input.execution)
&& taskTarget?.kind === "remote" && taskTarget.transport === "sandbox" && taskTarget.runnerTask && taskTarget.leaseId) {
const resumeFrom = await resolvePluginTaskResume(input.db, { companyId: input.execution.binding.companyId,
const resumeFrom = taskTarget.sandboxLeaseAcquisition?.outcome === "resumed" ? undefined : await resolvePluginTaskResume(input.db, { companyId: input.execution.binding.companyId,
leaseId: taskTarget.leaseId, priorIdentity: identity });
const result = environmentTaskResultSchema.parse(await taskTarget.runnerTask({ kind: "read_state", identity, resumeFrom }));
if (result.kind !== "state") throw new Error("Plugin Runner checkpoint unavailable");
@@ -7914,6 +7916,18 @@ async function executePaperclipNativeSessionWithinScope(
!idleWarmNativeSessionOwnerRunId(input)) {
const root = scopedRunnerdStateRoot(input.execution);
const identity = readRunnerdDurableIdentity(root);
const taskTarget = input.runnerExecutionTarget;
if (taskTarget.transport === "sandbox" && taskTarget.runnerTask && taskTarget.leaseId &&
durableIdentityMatchesSession(identity,input.execution) && identity.runId !== input.execution.binding.runId) {
// Inspection can replace a stopped retained task before deciding between adoption and checkpoint recovery.
await taskTarget.runnerTask({kind:"describe"});
if (taskTarget.sandboxLeaseAcquisition?.outcome !== "resumed") {
const resumeFrom=await resolvePluginTaskResume(input.db,{companyId:input.execution.binding.companyId,leaseId:taskTarget.leaseId,priorIdentity:identity});
const checkpoint=environmentTaskResultSchema.parse(await taskTarget.runnerTask({kind:"read_state",identity,resumeFrom}));
if(checkpoint.kind!=="state") throw new Error("Plugin Runner checkpoint unavailable");
savePluginRunnerCheckpoint(root,checkpoint.state,identity);
}
}
if (durableIdentityMatchesSession(identity, input.execution) && identity.runId !== input.execution.binding.runId &&
runnerdAuthorityLifecycleWithVerifiedBackup({ root, identity, execution: input.execution, allowVerifiedBackup: true }) !== "suspended") {
remoteWarmRecovery = await inspectRemoteWarmSessionRecovery({
@@ -8631,6 +8645,7 @@ async function executePaperclipNativeSessionWithinScope(
`native-warm-session:${input.execution.binding.runId}`,
);
let existingWarmSession: NativeSession | undefined;
let pluginTaskDescription: Extract<ReturnType<typeof environmentTaskResultSchema.parse>,{kind:"description"}> | undefined;
let managedCredentialSession: NativeSession | undefined;
let githubAccess: NativeGitHubAccess | undefined;
let releaseGitHubRun: (() => void) | undefined;
@@ -8640,6 +8655,12 @@ async function executePaperclipNativeSessionWithinScope(
if (entry) {
if (entry.preparingRunId && entry.preparingRunId !== input.execution.binding.runId) throw new Error("native_session_supervisor_busy");
entry.preparingRunId = undefined;
const taskTarget=input.runnerExecutionTarget;
if (entry.pluginTaskId && taskTarget?.kind === "remote" && taskTarget.transport === "sandbox" && taskTarget.runnerTask) {
const description=environmentTaskResultSchema.parse(await taskTarget.runnerTask({kind:"describe"}));
if(description.kind !== "description") throw new Error("Plugin Runner description unavailable");
pluginTaskDescription=description;
}
// Old run-scoped environments still require process replacement. A
// session-owned broker can change run authority without replacing it.
const hasBrokerCapability = Boolean(
@@ -8650,6 +8671,8 @@ async function executePaperclipNativeSessionWithinScope(
(hasBrokerCapability &&
entry.credentialRunId !== input.execution.binding.runId);
if (
(entry.pluginTaskId !== undefined && (input.runnerExecutionTarget?.kind !== "remote" || input.runnerExecutionTarget.transport !== "sandbox" ||
entry.pluginTaskId !== input.runnerExecutionTarget.sandboxLeaseAcquisition?.providerLeaseId)) ||
input.refreshTools === true ||
entry.closeOnReleaseReason !== undefined ||
entry.configDigest !== warmConfigDigest ||
@@ -8667,6 +8690,7 @@ async function executePaperclipNativeSessionWithinScope(
(input.runnerEnvironment?.PAPERCLIP_RUNNER_NETWORK_ACCESS ===
"enabled")
) {
pluginTaskDescription=undefined;
const preparedCopy = entry.instructionPreparationRunId === input.execution.binding.runId;
if (entry.busy && !preparedCopy) throw new Error("native_session_supervisor_busy");
if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
@@ -8860,6 +8884,8 @@ async function executePaperclipNativeSessionWithinScope(
// projects remoteCwd into the provider execution boundary.
execution: input.execution,
runnerInstanceId: effectiveRunnerInstanceId,
pluginTaskDescription,
retainedPluginTaskOwner: existingWarmSession ? pluginTaskSessionOwners.get(existingWarmSession) : undefined,
durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId,
remoteWarmRecovery: remoteWarmRecovery ? {
...remoteWarmRecovery,
@@ -9094,6 +9120,8 @@ async function executePaperclipNativeSessionWithinScope(
existing.runId = input.execution.binding.runId;
} else
warmNativeSessions.set(warmSessionId, {
pluginTaskId: usesPluginTask && input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "sandbox"
? input.runnerExecutionTarget.sandboxLeaseAcquisition?.providerLeaseId : undefined,
runId: input.execution.binding.runId,
instructionWorkingCopy: input.instructionWorkingCopy?.checkpointWarm ? undefined : input.instructionWorkingCopy,
instructionRoot: input.instructionWorkingCopy?.root,
@@ -11461,6 +11489,8 @@ export async function createRunnerdBackend(input: {
restartRecovery?: NativeRestartRecoveryClaim;
durableEnvironmentLeaseId?: string;
remoteWarmRecovery?: Omit<RemoteWarmSessionRecovery, "revalidate"> & { revalidate: () => Promise<void> };
retainedPluginTaskOwner?: ReturnType<typeof createPluginRunnerTaskOwner>;
pluginTaskDescription?: Extract<ReturnType<typeof environmentTaskResultSchema.parse>,{kind:"description"}>;
onSpawn?: (meta: {
pid: number;
processGroupId: number | null;
@@ -11658,8 +11688,8 @@ async function createRunnerdBackendWithinSessionClaim(
const remoteTarget = target.kind === "remote" ? target : null;
const pluginTask = remoteTarget?.transport === "sandbox" ? remoteTarget.runnerTask : undefined;
if (pluginTask && input.runnerIngressAuthorized !== true) throw new Error("runner_ingress_unavailable");
if (pluginTask && (input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8" || input.execution.session.lifecyclePolicy.mode !== "per_turn")) {
throw new Error("Plugin Runner tasks currently require a per-turn workspace session");
if (pluginTask && (input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8" || (input.execution.session.lifecyclePolicy.mode === "warm" && input.execution.provider.kind !== "codex"))) {
throw new Error("Plugin warm Runner tasks require a Codex workspace session");
}
const remoteCommandRunner = remoteTarget
? remoteTarget.transport === "ssh"
@@ -11791,10 +11821,26 @@ async function createRunnerdBackendWithinSessionClaim(
if (pluginTask && remoteTarget?.transport === "sandbox" && input.execution.schema !== "paperclip.native-execution-input.v6" && input.execution.schema !== "paperclip.native-execution-input.v8") {
const provider = input.execution.provider;
if (provider.kind !== "codex" && provider.kind !== "opencode" && provider.kind !== "acpx") throw new Error("Plugin Runner tasks require a workspace harness");
const description = environmentTaskResultSchema.parse(await pluginTask({ kind: "describe" }));
if (remoteTarget.sandboxLeaseAcquisition?.outcome === "resumed" && !durableBinding && !input.retainedPluginTaskOwner && !input.remoteWarmRecovery) {
// A new logical session (including fresh retries) cannot inherit the old process authority.
await pluginTask({kind:"stop"});
const stopped=environmentTaskResultSchema.parse(await pluginTask({kind:"status"}));
if(stopped.kind!=="status" || stopped.executionStopped!==true) throw new Error("Plugin Runner termination is unconfirmed; reconcile the saved task");
await pluginTask({kind:"complete"});
}
const description = input.pluginTaskDescription ?? environmentTaskResultSchema.parse(await pluginTask({ kind: "describe" }));
if (description.kind !== "description" || description.launchPreparationVersion !== 1 || description.projectAuthCheckVersion !== 1) throw new Error("Plugin Runner launch preparation v1 unavailable");
const prior = durableBinding ? durableIdentity as RunnerdDurableIdentity : undefined;
const resumeFrom = prior && prior.environmentLeaseId !== remoteTarget.leaseId
const retainedOwner = input.retainedPluginTaskOwner;
const adopting = input.remoteWarmRecovery !== undefined;
const adoptionIdentity = input.remoteWarmRecovery ? {
runnerInstanceId: input.remoteWarmRecovery.identity.runnerInstanceId,
environmentLeaseId: input.remoteWarmRecovery.identity.environmentLeaseId,
runId: input.remoteWarmRecovery.identity.runId,
normalizedSessionId: input.remoteWarmRecovery.identity.normalizedSessionId,
turnId: input.remoteWarmRecovery.identity.turnId, itemId: input.remoteWarmRecovery.identity.itemId,
} : undefined;
const resumeFrom = !retainedOwner && !adopting && prior && prior.environmentLeaseId !== remoteTarget.leaseId
? await resolvePluginTaskResume(input.db, { companyId: input.execution.binding.companyId,
leaseId: remoteTarget.leaseId!, priorIdentity: prior }) : undefined;
const credentials = await preparePluginRunnerCredentials({
@@ -11803,12 +11849,16 @@ async function createRunnerdBackendWithinSessionClaim(
home: input.managedAiCredentialHome, environment: input.runnerEnvironment ?? {},
});
if (credentials.needsWriteback && !input.managedAiCredentialWriteback) throw new Error("Plugin Runner credential writeback authority unavailable");
if (input.managedAiCredentialWriteback && credentials.accountIdentity) {
if (!adopting && input.managedAiCredentialWriteback && credentials.accountIdentity) {
if (!remoteTarget.leaseId) throw new Error("Plugin Runner credential lease unavailable");
await registerPluginTaskCredentialHandoff(input.db, { companyId: input.execution.binding.companyId, leaseId: remoteTarget.leaseId },
credentials.accountIdentity, input.managedAiCredentialWriteback);
}
const owner = createPluginRunnerTaskOwner({ ...credentials, projectAuthCheckVersion: description.projectAuthCheckVersion, task: pluginTask, projectIds: remoteTarget.runnerTaskProjectIds ?? [],
const ownerInput: Parameters<typeof createPluginRunnerTaskOwner>[0] = { ...credentials,
// Warm retirement can outlive the run's temporary credential home; complete uses the durable handoff.
onCredentials: input.execution.session.lifecyclePolicy.mode === "warm" ? undefined : credentials.onCredentials,
warmTaskVersion: description.warmTaskVersion,
adoptIdentity: adoptionIdentity, projectAuthCheckVersion: description.projectAuthCheckVersion, task: pluginTask, projectIds: remoteTarget.runnerTaskProjectIds ?? [],
harness: provider.kind === "acpx" ? provider.agent : provider.kind, resumeFrom, priorIdentity: prior ? {
runnerInstanceId: prior.runnerInstanceId, environmentLeaseId: prior.environmentLeaseId,
runId: prior.runId, normalizedSessionId: prior.normalizedSessionId,
@@ -11826,7 +11876,10 @@ async function createRunnerdBackendWithinSessionClaim(
})}::jsonb` })
.where(and(eq(environmentLeases.id, remoteTarget.leaseId!), eq(environmentLeases.companyId, input.execution.binding.companyId)));
}, onDispatched: input.onTaskDispatched,
});
};
if (input.execution.session.lifecyclePolicy.mode === "warm" && (!credentials.accountIdentity || description.warmTaskVersion !== 1)) throw new Error("Plugin Codex warm task capability or account unavailable");
const owner = retainedOwner ?? createPluginRunnerTaskOwner(ownerInput);
if (retainedOwner) retainedOwner.rebind(ownerInput);
if (description.workspaceFileReadVersion === 1) {
authority.binding.readRemoteApiFile = owner.readWorkspaceFile;
authority.binding.readRemoteTaskFile = file => owner.readWorkspaceFile(file.contentRef, file.byteSize);
@@ -11851,13 +11904,13 @@ async function createRunnerdBackendWithinSessionClaim(
claudeEffort: "claudeEffort" in provider ? provider.claudeEffort : undefined,
acpxPermissionModePinned: true } : {}),
...(provider.kind === "opencode" ? { opencodePermissionMode: provider.permissionMode } : {}),
runnerReconnectGraceMs: 120_000, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
runnerReconnectGraceMs: 120_000, lifecyclePolicy: input.execution.session.lifecyclePolicy,
runtimeContext: "runtimeContext" in input.execution ? input.execution.runtimeContext : null,
externallySandboxed: true, baseInstructions: recovery?.baseInstructions,
resumeDynamicTools: dynamicTools, resumeCompletionContract: { revision: input.execution.completionContract.contract.revision,
criterionIds: input.execution.completionContract.contract.criteria.map(criterion => criterion.id) },
resumeActiveTurnId: recovery?.persistedSession?.activeTurnId ?? null, resumeProviderSession: recovery?.persistedSession,
prpIdentity: { runnerInstanceId: prior?.runnerInstanceId ?? effectiveRunnerInstanceId,
prpIdentity: adoptionIdentity ?? { runnerInstanceId: prior?.runnerInstanceId ?? effectiveRunnerInstanceId,
environmentLeaseId: prior?.environmentLeaseId ?? effectiveEnvironmentLeaseId,
runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution),
turnId: `turn-${input.execution.binding.runId}`, itemId: `item-${input.execution.binding.runId}` },
@@ -11866,7 +11919,8 @@ async function createRunnerdBackendWithinSessionClaim(
sessionToolAuthorityEpochs.get(sessionScopeId)?.revoke();
sessionToolAuthorityEpochs.set(sessionScopeId, authorityEpoch);
return { ...backend, descriptor: () => backend.descriptor(), openSession: session => backend.openSession(session),
recoverSession: (snapshot, options) => backend.recoverSession!(snapshot, options), bindManagedSession: session => session };
recoverSession: (snapshot, options) => backend.recoverSession!(snapshot, options),
bindManagedSession: session => { 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");
@@ -322,3 +322,78 @@ describe("plugin task workspace file transfer", () => {
} finally {await registration.release();await registration.checkpoint("unsettled");await handle.completion;}
});
});
it("reuses a warm Codex process across run authorities without idle status polling", async () => {
const {task} = provider();
ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}});
const retained = vi.fn(async()=>{});
const input = {task,projectIds:[],harness:"codex",projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const,
accountIdentity:"account",credentialIdentity:"revision",pollMs:1,onSessionRetained:retained};
const owner = createPluginRunnerTaskOwner(input);
await prepare(owner);
const registration = await owner.registration({} as never,"run");
const spec = launchSpec(); spec.nativeRunner!.lifecyclePolicy = {mode:"warm",idleTimeoutMs:300000};
const handle = owner.launcher(spec); registration.activate(); await registration.ready();
const observed = task.mock.calls.length;
await new Promise(resolve=>setTimeout(resolve,10));
expect(task.mock.calls).toHaveLength(observed);
const originalTask=task.getMockImplementation()!;
const nextTask=vi.fn(async op=>{
const result=await originalTask(op) as Record<string,unknown>;
return op.kind === "prepare" ? {...result,projectAuthCheckAtSubmit:true} : result;
});
owner.rebind({...input,task:nextTask});
const nextRegistration=await owner.registration({} as never,"next-run");
const nextIdentity={...requirements.identity,runId:"next-run",turnId:"next-turn",itemId:"next-item"};
await owner.driver.prepareLaunch!({...requirements,identity:nextIdentity});
expect(nextTask.mock.calls[0][0]).toMatchObject({kind:"prepare",priorIdentity:requirements.identity,accountIdentity:"account"});
expect(nextTask.mock.calls.map(([op])=>op.kind)).toEqual(["prepare","check_project_auth"]);
await registration.release(); nextRegistration.activate(); await nextRegistration.ready();
expect(nextTask.mock.calls.some(([op])=>op.kind==="submit" || op.kind==="stop" || op.kind==="complete")).toBe(false);
await nextRegistration.release(); await nextRegistration.checkpoint("settled"); await handle.completion;
expect(retained).toHaveBeenCalledWith(expect.objectContaining(nextIdentity));
expect(nextTask.mock.calls.filter(([op])=>op.kind==="stop")).toHaveLength(1);
});
it("adopts a verified retained Codex task without submitting a replacement", async () => {
const {task}=provider();
ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}});
const owner=createPluginRunnerTaskOwner({task,projectIds:[],harness:"codex",projectAuthCheckVersion:1,
warmTaskVersion:1,accountIdentity:"account",adoptIdentity:requirements.identity,pollMs:1});
const recovered = {configurationId:"saved",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}};
await expect(owner.driver.prepareLaunch!(requirements,recovered)).resolves.toEqual(recovered);
expect(task.mock.calls.map(([op])=>op.kind)).toEqual(["check_project_auth"]);
expect(task.mock.calls[0][0]).toMatchObject({configurationId:"saved",projectIds:[]});
expect(await owner.driver.adoptExistingRunner!.isAlive()).toBe(true);
const registration=await owner.registration({} as never,"run");
registration.activate(); await registration.ready();
await registration.release(); await registration.checkpoint("settled");
expect(task.mock.calls.some(([op])=>op.kind==="submit")).toBe(false);
expect(task.mock.calls.some(([op])=>op.kind==="complete")).toBe(true);
});
it("denies retained process adoption when current project authentication is revoked",async()=>{
const {task}=provider();
task.mockImplementation(async op=>op.kind === "check_project_auth"
? {kind:"project_auth",taskId:"task",configurationId:op.configurationId,provider:op.provider,allowed:false} as never
: {kind:"accepted",taskId:"task"} as never);
const owner=createPluginRunnerTaskOwner({task,projectIds:["project"],harness:"codex",projectAuthCheckVersion:1,
warmTaskVersion:1,accountIdentity:"account",adoptIdentity:requirements.identity});
await expect(owner.driver.prepareLaunch!(requirements,{configurationId:"saved",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}}))
.rejects.toThrow("project authentication conflicts");
expect(task.mock.calls.map(([op])=>op.kind)).toEqual(["check_project_auth"]);
});
it("refuses warm account or credential generation changes", async()=>{
const {task}=provider();
const input={task,projectIds:[],harness:"codex",projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const,
accountIdentity:"account",credentialIdentity:"revision",pollMs:1};
const owner=createPluginRunnerTaskOwner(input);
await prepare(owner);
const spec=launchSpec();spec.nativeRunner!.lifecyclePolicy={mode:"warm",idleTimeoutMs:300000};
const process=owner.launcher(spec);
expect(()=>owner.rebind({...input,accountIdentity:"other"})).toThrow("cannot be rebound");
expect(()=>owner.rebind({...input,credentialIdentity:"rotated"})).toThrow("cannot be rebound");
const registration=await owner.registration({} as never,"run");
await registration.checkpoint("unsettled"); await process.completion;
});
@@ -24,17 +24,20 @@ export function createPluginRunnerTaskOwner(input: {
onSessionRetained?: (state: Record<string, unknown>) => Promise<void>;
onDispatched?: () => Promise<void> | void;
pollMs?: number;
warmTaskVersion?: 1;
adoptIdentity?: DurableRecoveryIdentity;
}) {
const pollMs = input.pollMs ?? 500;
let registrationGeneration = 0;
let activeOutbound: ReturnType<typeof connectRunnerPrpIngress> | null = null;
let preparedLaunchId: string | undefined;
let preparedIdentity: DurableRecoveryIdentity | undefined;
let launched = false;
let preparedIdentity: DurableRecoveryIdentity | undefined = input.adoptIdentity;
let launched = !!input.adoptIdentity;
let stopRequested = false;
let released = false;
let submitted = false;
let submitted = !!input.adoptIdentity;
let detached = false;
let adoptionVerified = false;
let markReady!: () => void;
let failReady!: (error: unknown) => void;
let confirmStopped: (code: number) => void = () => {};
@@ -44,6 +47,7 @@ export function createPluginRunnerTaskOwner(input: {
void failure.catch(() => undefined);
const running = new Promise<void>((resolve, reject) => { markReady = resolve; failReady = reject; });
void running.catch(() => undefined);
if (input.adoptIdentity) markReady();
const call = async (operation: PluginEnvironmentTaskOperation) => {
const phase = ({ prepare: "execution.plugin_prepare", check_project_auth: "execution.plugin_project_auth",
submit: "execution.plugin_submit", connection: "execution.plugin_connection" } as const)[operation.kind as "prepare" | "check_project_auth" | "submit" | "connection"];
@@ -54,23 +58,32 @@ export function createPluginRunnerTaskOwner(input: {
}
return result;
};
const checkProjectAuth = async (configurationId:string) => {
if (input.projectAuthCheckVersion !== 1) throw new Error("Plugin Runner managed project authentication validation unavailable");
const provider = input.harness === "codex" ? "openai" : input.harness === "claude" ? "anthropic" : input.harness;
const validation = await call({kind:"check_project_auth",configurationId,provider,projectIds:[...input.projectIds]});
if (released || stopRequested) throw new Error("Plugin Runner preparation cancelled");
if (validation.kind !== "project_auth" || !validation.allowed || validation.configurationId !== configurationId
|| validation.provider !== provider) throw new Error("Plugin Runner project authentication conflicts with the selected managed account");
};
const prepareLaunch: RunnerLaunchPreparer = async (requirements, recovery) => {
if (released || stopRequested) throw new Error("Plugin Runner preparation cancelled");
// Reconnecting the previous authority must not prepare it through the new run's lease.
if (recovery && input.adoptIdentity && Object.entries(input.adoptIdentity).every(([key,value]) => requirements.identity[key as keyof DurableRecoveryIdentity] === value)) {
await checkProjectAuth(recovery.configurationId);
preparedLaunchId = recovery.configurationId;
return recovery;
}
const result = await call(environmentTaskOperationSchema.parse({ kind: "prepare", requirements,
projectIds: [...input.projectIds], ...(recovery ? { recovery } : {}),
...(input.resumeFrom ? { resumeFrom: input.resumeFrom } : {}),
...(launched && preparedIdentity && requirements.identity.runId !== preparedIdentity.runId
? { priorIdentity: preparedIdentity, accountIdentity: input.accountIdentity } : {}),
}));
if (released || stopRequested) throw new Error("Plugin Runner preparation cancelled");
if (result.kind !== "prepared") throw new Error("Plugin Runner preparation unavailable");
if (input.projectAuthCheckVersion !== 1) throw new Error("Plugin Runner managed project authentication validation unavailable");
if (!result.projectAuthCheckAtSubmit) {
const provider = input.harness === "codex" ? "openai" : input.harness === "claude" ? "anthropic" : input.harness;
const validation = await call({ kind: "check_project_auth", configurationId: result.launch.configurationId,
provider, projectIds: [...input.projectIds] });
if (released || stopRequested) throw new Error("Plugin Runner preparation cancelled");
if (validation.kind !== "project_auth" || !validation.allowed || validation.configurationId !== result.launch.configurationId
|| validation.provider !== provider) throw new Error("Plugin Runner project authentication conflicts with the selected managed account");
}
if (launched || !result.projectAuthCheckAtSubmit) await checkProjectAuth(result.launch.configurationId);
preparedLaunchId = result.launch.configurationId;
preparedIdentity = requirements.identity;
return result.launch;
@@ -115,7 +128,7 @@ export function createPluginRunnerTaskOwner(input: {
if (launched || released) throw new Error("Plugin Runner replacement requires a new task lease");
const native = spec.nativeRunner;
if (!preparedLaunchId) throw new Error("Plugin Runner launch requires preparation");
if (!native || native.lifecyclePolicy?.mode !== "per_turn"
if (!native || (native.lifecyclePolicy?.mode !== "per_turn" && !(native.lifecyclePolicy?.mode === "warm" && input.harness === "codex" && input.accountIdentity && input.warmTaskVersion === 1))
|| !preparedIdentity || Object.entries(preparedIdentity).some(([key, value]) => native.identity[key as keyof DurableRecoveryIdentity] !== value)) {
throw new Error("Plugin Runner requires a structured per-turn native launch");
}
@@ -138,7 +151,11 @@ export function createPluginRunnerTaskOwner(input: {
});
launched = true;
const child: RunnerProcessHandle["child"] = { exitCode: null, signalCode: null,
kill: () => { stopRequested = true; return true; } };
kill: () => {
stopRequested = true;
if (native.lifecyclePolicy?.mode === "warm" && submitted) void stopAndConfirm().catch(rejectFailure);
return true;
} };
const completion = new Promise<Awaited<RunnerProcessHandle["completion"]>>(resolve => {
confirmStopped = code => {
if (child.exitCode !== null) return;
@@ -172,7 +189,11 @@ export function createPluginRunnerTaskOwner(input: {
if (detached) return;
if (stopRequested && !stopSent) { await call({ kind: "stop" }); stopSent = true; }
const state = await readStatus();
if (state.phase === "running" && !stopRequested) markReady();
if (state.phase === "running" && !stopRequested) {
markReady();
// PRP owns live failure detection; warm idle tasks must not keep polling Launcher.
if (native.lifecyclePolicy?.mode === "warm") return;
}
if (state.executionStopped === true) return;
await delay(pollMs);
}
@@ -251,6 +272,16 @@ export function createPluginRunnerTaskOwner(input: {
return result.state;
},
controlPlaneRegistration: (authority, identity, signal) => registration(authority, identity?.runId ?? "", signal),
...(input.adoptIdentity ? { adoptExistingRunner: {
// Provider ownership is attested by Launcher, never by a controller-local PID.
pid: 0, processGroupId: null, startedAt: new Date().toISOString(),
isAlive: async () => {
if (!adoptionVerified) { if ((await readStatus()).executionStopped === true) return false; adoptionVerified = true; }
// Authenticated PRP detects disconnects; this cache never asserts termination.
return stoppedState === undefined;
},
signal: async () => { stopRequested = true; await stopAndConfirm(); return true; },
} } : {}),
};
const readWorkspaceFile = async (path: string, maxBytes: number): Promise<Buffer> => {
if (!submitted || !preparedIdentity || released || stopRequested) throw new Error("Plugin Runner file access unavailable");
@@ -262,5 +293,14 @@ export function createPluginRunnerTaskOwner(input: {
createHash("sha256").update(bytes).digest("hex") !== result.sha256) throw new Error("Plugin Runner file integrity check failed");
return bytes;
};
return { launcher, registration, driver, readWorkspaceFile };
return { launcher, registration, driver, readWorkspaceFile,
rebind(next: typeof input) {
if (!launched || stopRequested || stoppedState || detached || next.harness !== "codex" ||
input.accountIdentity !== next.accountIdentity || input.credentialIdentity !== next.credentialIdentity ||
JSON.stringify(input.projectIds) !== JSON.stringify(next.projectIds)) throw new Error("Plugin warm task cannot be rebound");
input = next;
released = false;
},
};
}
@@ -0,0 +1,8 @@
import {expect,it} from "vitest";
import {resolveRunnerEnvironmentForRun} from "./runner-environment-lifecycle.js";
it("opts Codex plugin tasks into warm retention without opting ACPX in",()=>{
const environment={driver:"plugin" as const,config:{pluginKey:"fixture",driverKey:"tasks",driverConfig:{}}};
expect(resolveRunnerEnvironmentForRun(environment,"paperclip_runner",{provider:"codex",lifecycleMode:"warm",idleTimeoutMs:1234})).toMatchObject({config:{runnerLifecycleMode:"warm",runnerIdleTimeoutMs:1234}});
expect(resolveRunnerEnvironmentForRun(environment,"paperclip_runner",{provider:"acpx",lifecycleMode:"warm"})).toMatchObject({config:{runnerLifecycleMode:"per_turn"}});
expect(environment.config).not.toHaveProperty("runnerLifecycleMode");
});
@@ -1,3 +1,4 @@
import { resolvePaperclipRunnerIdleTimeoutMs } from "@paperclipai/adapter-utils";
import type { Environment } from "@paperclipai/shared";
/** Resolve runner retention before acquiring a sandbox, not after it is ephemeral.
@@ -10,11 +11,15 @@ export function resolveRunnerEnvironmentForRun<T extends Pick<Environment, "driv
adapterConfig: Record<string, unknown> = {},
admittedLifecycleMode?: "warm" | "per_turn",
): T {
if (adapterType !== "paperclip_runner" || environment.driver !== "sandbox") return environment;
if (adapterType !== "paperclip_runner" || !["sandbox", "plugin"].includes(environment.driver)) return environment;
const config = environment.config ?? {};
const lifecycleMode = admittedLifecycleMode ?? (config.runnerLifecycleMode === "warm" || config.runnerLifecycleMode === "per_turn"
? config.runnerLifecycleMode
: adapterConfig.lifecycleMode);
if (environment.driver === "plugin") return { ...environment, config: { ...config,
runnerLifecycleMode: lifecycleMode === "warm" && (adapterConfig.provider ?? "codex") === "codex" ? "warm" : "per_turn",
runnerIdleTimeoutMs: resolvePaperclipRunnerIdleTimeoutMs(config.runnerIdleTimeoutMs ?? adapterConfig.idleTimeoutMs),
} };
if (lifecycleMode !== "warm") return environment;
return {
...environment,