From aa0037a13782e8218ea0f5e1ae3ed39d68d82c40 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 23 Oct 2026 19:14:14 -0700 Subject: [PATCH] Support Codex warm plugin task reuse and recovery (#17390) --- doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md | 4 +- doc/plugins/PLUGIN_RUNNER_EXECUTION.md | 41 ++++- .../crates/runner-core/src/durable/runner.rs | 103 ++++++++++- packages/plugins/sdk/src/environment-tasks.ts | 7 +- packages/plugins/sdk/src/protocol.ts | 2 + .../sdk/tests/environment-tasks.test.ts | 13 ++ packages/shared/src/types/environment.ts | 2 + packages/shared/src/types/plugin.ts | 2 + packages/shared/src/validators/plugin.ts | 1 + .../environment-run-orchestrator.test.ts | 2 + .../src/__tests__/environment-runtime.test.ts | 29 +++ .../plugin-runner-task.integration.test.ts | 2 +- .../plugin-warm-tasks.integration.test.ts | 171 ++++++++++++++++++ server/src/services/environment-config.ts | 2 + .../services/environment-execution-target.ts | 26 ++- .../services/environment-run-orchestrator.ts | 3 + server/src/services/environment-runtime.ts | 77 +++++--- .../src/services/environment-task-runtime.ts | 110 ++++++++++- .../heartbeat/workspace-preparation.test.ts | 4 + .../heartbeat/workspace-preparation.ts | 3 + .../native-session-executor.test.ts | 88 ++++++++- .../native-runtime/native-session-executor.ts | 76 ++++++-- .../native-runtime/plugin-runner-task.test.ts | 75 ++++++++ .../native-runtime/plugin-runner-task.ts | 70 +++++-- .../runner-environment-lifecycle.test.ts | 8 + .../services/runner-environment-lifecycle.ts | 7 +- 26 files changed, 845 insertions(+), 83 deletions(-) create mode 100644 packages/plugins/sdk/tests/environment-tasks.test.ts create mode 100644 server/src/__tests__/plugin-warm-tasks.integration.test.ts create mode 100644 server/src/services/runner-environment-lifecycle.test.ts diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index c2ccb0461c8..55e3d57b15c 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -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 diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index 0be326e6821..e632a50a1c3 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -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. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index f59ca939ec1..ce05d2acce3 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -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(¤t).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, ¤t).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( + ¤t, + &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!( diff --git a/packages/plugins/sdk/src/environment-tasks.ts b/packages/plugins/sdk/src/environment-tasks.ts index 59b05730b48..57f0dea51fa 100644 --- a/packages/plugins/sdk/src/environment-tasks.ts +++ b/packages/plugins/sdk/src/environment-tasks.ts @@ -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(), diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index c3c721b6dff..6fa7d47e944 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -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; } diff --git a/packages/plugins/sdk/tests/environment-tasks.test.ts b/packages/plugins/sdk/tests/environment-tasks.test.ts new file mode 100644 index 00000000000..9db7cbfe2d5 --- /dev/null +++ b/packages/plugins/sdk/tests/environment-tasks.test.ts @@ -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); + }); +}); diff --git a/packages/shared/src/types/environment.ts b/packages/shared/src/types/environment.ts index 54251055a3c..9f929fcb71b 100644 --- a/packages/shared/src/types/environment.ts +++ b/packages/shared/src/types/environment.ts @@ -66,6 +66,8 @@ export type SandboxEnvironmentConfig = | PluginSandboxEnvironmentConfig; export interface PluginEnvironmentConfig { + runnerLifecycleMode?: "inherit" | "per_turn" | "warm"; + runnerIdleTimeoutMs?: number; pluginKey: string; driverKey: string; driverConfig: Record; diff --git a/packages/shared/src/types/plugin.ts b/packages/shared/src/types/plugin.ts index acfebb4395a..e4b437f77ea 100644 --- a/packages/shared/src/types/plugin.ts +++ b/packages/shared/src/types/plugin.ts @@ -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; /** diff --git a/packages/shared/src/validators/plugin.ts b/packages/shared/src/validators/plugin.ts index 16b8092f0c0..4663958df23 100644 --- a/packages/shared/src/validators/plugin.ts +++ b/packages/shared/src/validators/plugin.ts @@ -179,6 +179,7 @@ export type SandboxProviderCapabilitiesInput = z.infer { 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" } }); diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 3149a3b5ba1..8eb816fd340 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -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(resolve=>{release=resolve;}),started=new Promise(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", diff --git a/server/src/__tests__/plugin-runner-task.integration.test.ts b/server/src/__tests__/plugin-runner-task.integration.test.ts index 45f347726b8..ab9e91759db 100644 --- a/server/src/__tests__/plugin-runner-task.integration.test.ts +++ b/server/src/__tests__/plugin-runner-task.integration.test.ts @@ -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" })) diff --git a/server/src/__tests__/plugin-warm-tasks.integration.test.ts b/server/src/__tests__/plugin-warm-tasks.integration.test.ts new file mode 100644 index 00000000000..aee224def81 --- /dev/null +++ b/server/src/__tests__/plugin-warm-tasks.integration.test.ts @@ -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,stop:()=>Promise; + 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(),acquireRequests:any[]=[]; + let resumeGate:Promise|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(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()); + }); + +}); diff --git a/server/src/services/environment-config.ts b/server/src/services/environment-config.ts index 002cea6a477..b9c041a35d1 100644 --- a/server/src/services/environment-config.ts +++ b/server/src/services/environment-config.ts @@ -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._-]*$/, diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index d497e32805c..02d42e92072 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -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 = { + 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") { diff --git a/server/src/services/environment-run-orchestrator.ts b/server/src/services/environment-run-orchestrator.ts index 287b4a9fd74..caa024c79a4 100644 --- a/server/src/services/environment-run-orchestrator.ts +++ b/server/src/services/environment-run-orchestrator.ts @@ -214,6 +214,7 @@ export function environmentRunOrchestrator( heartbeatRunId: string; persistedExecutionWorkspace: Pick | null; executionWorkspaceSettings: IssueExecutionWorkspaceSettings | null; + runnerTaskProjectIds?: readonly string[]; adapterType: string | null; }): Promise { try { @@ -277,6 +278,7 @@ export function environmentRunOrchestrator( agentId: string; persistedExecutionWorkspace: Pick | 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; diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index da016c92491..71cca45a3e1 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -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 | null = null; - const ensureSpoolLoaded = (): Promise => { - 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 | 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, diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index 937822cfb95..ffdbf703123 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -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 | 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 | 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 | undefined; + const matches = (identity: typeof prior, saved: Record | 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 | 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), diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index 0b40ec91564..ccbcd9066c5 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -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"); } diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index c083c51f263..a23665a96b6 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -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, diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 2dc81a2d161..e881b964bb6 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -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"); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 65fa5c31ede..c0446b8385d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -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(); +const pluginTaskSessionOwners = new WeakMap>(); // 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>(); @@ -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,{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 & { revalidate: () => Promise }; + retainedPluginTaskOwner?: ReturnType; + pluginTaskDescription?: Extract,{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[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"); diff --git a/server/src/services/native-runtime/plugin-runner-task.test.ts b/server/src/services/native-runtime/plugin-runner-task.test.ts index 4ee51bc072e..27f7f4f5364 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -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; + 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; +}); diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index 19aa8f8aea1..8cbb01a9f4c 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -24,17 +24,20 @@ export function createPluginRunnerTaskOwner(input: { onSessionRetained?: (state: Record) => Promise; onDispatched?: () => Promise | void; pollMs?: number; + warmTaskVersion?: 1; + adoptIdentity?: DurableRecoveryIdentity; }) { const pollMs = input.pollMs ?? 500; let registrationGeneration = 0; let activeOutbound: ReturnType | 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((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>(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 => { 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; + }, + }; + } diff --git a/server/src/services/runner-environment-lifecycle.test.ts b/server/src/services/runner-environment-lifecycle.test.ts new file mode 100644 index 00000000000..d6aaeb0d9bf --- /dev/null +++ b/server/src/services/runner-environment-lifecycle.test.ts @@ -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"); +}); diff --git a/server/src/services/runner-environment-lifecycle.ts b/server/src/services/runner-environment-lifecycle.ts index fcb898bae20..b2908196eb7 100644 --- a/server/src/services/runner-environment-lifecycle.ts +++ b/server/src/services/runner-environment-lifecycle.ts @@ -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 = {}, 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,