diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 13f907f61f0..175a0108647 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1996,10 +1996,54 @@ That is the execution contract Paperclip should present to operators. ### Cancellation during native startup -Core supports verified cleanup for explicitly recorded startup obligations. This -support does not create startup contexts, advertise transport support, or capture -a transport binding. Existing callers retain their legacy payloads and startup -budgets. This support alone makes no startup-latency claim. +A retained Codex or Claude session uses the same Core-owned startup context +for local execution and plugin execution. Core retains the selected native +execution input. Provider, model, instructions, skills and tool selection remain +in that input and its existing Runner launch requirements. Provider configuration defaults +changed after selection apply to later runs. A configuration edit does not hold +startup open or delay its commit. + +For a retained plugin task, this context starts inside the existing controller +claim after its authority locks and checks. Its 30-second budget includes claim +commit, retained cleanup, optional attachment staging and accounting setup before +session inspection. The claim records its exact context, task lease, owner and +deadline in the existing run profile update. Local retained sessions start the +context after the claim; their deadline remains the smaller of the 30-second warm +startup budget and the controller's remaining confirmed ownership budget. Fresh +launches, checkpoint recovery, Pi and other providers retain their existing +budgets. Inspection of a warm candidate can discover a stopped task and acquire +a replacement within this bound. Once Core rejects reuse, the later cold launch +uses its existing budget. Before an unsupported-plugin or incompatible warm +candidate falls back, Core clears the matching durable context with an ownership +check; a failed clear prevents fallback. This adds one database response only to +that remote fallback, and none to compatible warm or local startup. + +A compatible plugin receives an optional transient transport projection. It +contains an attempt ID, an absolute deadline, plugin-owned instance secrets and +the company transport configuration. Core reads this transport once for the +attempt. It does not put transport secrets in lease metadata, checkpoints or run +results. The existing warm-identity update records only a run ID, context ID and +deadline before the plugin receives a preparation request. The +plugin uses the existing Core-selected Runner requirements, verifies its actual +guest resources, and does not repeat Core's eligibility decision. Generic plugin +runtime reads remain fresh. Older plugins receive no transport projection. Their initial capability lookup +is still bounded by the Core startup context; after unsupported capability is +detected and the matching context is cleared, subsequent startup uses the +existing path and budgets. + +Cancellation aborts compatible plugin startup I/O and prevents later provider +dispatch. The initial capability lookup retains its legacy read path and can +finish after the caller stops waiting. Already accepted remote requests can still +commit. A caller timeout or deadline +does not prove that its worker or remote process stopped. Core retains its +startup responsibility until the actual worker reply or the deadline. A worker +that remains unsettled then requests audited cancellation and automatic cleanup +through the existing control queues. Normal Stop joins this work. Cleanup must +retire the exact physical task, verify that its process and preparation writes +stopped, and drain accepted host handlers before releasing resources. A process +stop alone is insufficient. The compatible backend fences later writes to a +retired physical task; retries allocate a new task ID. Missing proof or an +unreachable backend leaves visible cleanup pending and automatic retries. Restart recovery uses the claim breadcrumb only while the current run has no first session checkpoint and the prior controller and context deadlines have @@ -2013,7 +2057,6 @@ the duration of remote cleanup. Existing Stop receipts and process cleanup still destructive lifecycle or binding changes. Operator plugin restarts and external machine changes can still interrupt startup. - Cancellation records a preparation fence while holding the run row lock. Native runtime selection checks that fence, the running status, and the current startup controller lease in the same transaction that creates the native coordinator. diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index d94db9f0585..d0b66bf7fc1 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -185,7 +185,7 @@ export interface AdapterSandboxExecutionTarget extends AdapterExecutionTargetWor transport: "sandbox"; providerKey?: string | null; /** Host-bound plugin task RPC. Both sides validate the wire contract. */ - runnerTask?: (operation: unknown) => Promise; + runnerTask?: (operation: unknown, context?: { executionContext?: unknown }) => Promise; runnerTaskProjectIds?: readonly string[]; /** Private Core-issued identity for a bounded, tool-free setup probe. */ runnerTaskProbe?: { runId: string; sessionId: string }; diff --git a/server/src/__tests__/environment-task-runtime.test.ts b/server/src/__tests__/environment-task-runtime.test.ts index 96126fc6df8..8a175531594 100644 --- a/server/src/__tests__/environment-task-runtime.test.ts +++ b/server/src/__tests__/environment-task-runtime.test.ts @@ -71,24 +71,38 @@ describe("plugin-provided Runner execution", () => { if (preparationStopped) { expect(result).toEqual({ value: true, error: null }); expect(operations).toEqual(["stop", "status", "complete"]); } else { expect(result.error?.message).toContain("termination is unconfirmed"); expect(fixture.sets).toContainEqual(expect.objectContaining({ status: "pending_cleanup", cleanupStatus: "failed" })); } }); - it("keeps legacy payloads passive without capability or transport activation", async () => { + it("negotiates the transient context without reading transport for an older plugin", async () => { const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); const workers = worker(); workers.call.mockImplementation(async (...args: any[]) => { args[5]?.(); return { kind: "description", taskId: "attempt-1", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; }); try { - for (const executionContext of [undefined, context]) { - await executeEnvironmentTask(database().db, workers as never, { companyId: "company", leaseId, - operation: { kind: "describe" }, ...(executionContext ? { executionContext } : {}) }); - const sent = workers.call.mock.calls.at(-1) as unknown as any[]; - expect(sent[2]).not.toHaveProperty("executionContextSupported"); - expect(sent[2]).not.toHaveProperty("executionContext"); - if (!executionContext) expect(sent).toHaveLength(4); - } + await executeEnvironmentTask(database().db, workers as never, { companyId: "company", leaseId, operation: { kind: "describe" }, executionContext: context }); + const sent = workers.call.mock.calls[0] as unknown as any[]; + expect(sent[2]).toMatchObject({ executionContextSupported: 1 }); + expect(sent[2].executionContext).toBeUndefined(); expect(state.identity).not.toHaveBeenCalled(); expect(state.config).not.toHaveBeenCalled(); expect(context.pending).toBe(false); } finally { context.finish(); } }); + it("shares one transport lookup across the supported startup operations", async () => { + const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); + context.pluginSupported = true; + const workers = worker(); + workers.call.mockImplementation(async (...args: any[]) => { + args[5]?.(); + return { kind: "description", taskId: "attempt-1", executionContextVersion: 1, runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; + }); + try { + for (const kind of ["describe", "describe"] as const) await executeEnvironmentTask(database().db, workers as never, + { companyId: "company", leaseId, operation: { kind }, executionContext: context }); + expect(state.identity).toHaveBeenCalledOnce(); + expect(state.config).toHaveBeenCalledOnce(); + for (const call of workers.call.mock.calls as unknown as any[][]) expect(call[2].executionContext).toMatchObject({ + id: context.id, deadlineAtMs: context.deadlineAtMs, transport: { instanceSecrets: { launcher: "synthetic-transient-binding" } }, + }); + } finally { context.finish(); } + }); it.each(["json", "wrong-run"])("rejects a %s context before the worker", async mode => { const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: mode === "wrong-run" ? "other" : "run" } } as never, 1_000); const workers = worker(); @@ -98,6 +112,16 @@ describe("plugin-provided Runner execution", () => { expect(workers.call).not.toHaveBeenCalled(); } finally { context.finish(); } }); + it("removes transport material from worker error reporting", async () => { + const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); + context.pluginSupported = true; + const workers = worker(); + workers.call.mockImplementation(async (...args: any[]) => { args[5]?.(); throw new Error(JSON.stringify(args[2])); }); + try { + await expect(executeEnvironmentTask(database().db, workers as never, { companyId: "company", leaseId, operation: { kind: "describe" }, executionContext: context })) + .rejects.toThrow("Plugin-provided Runner execution operation unavailable; reconcile the same task before retrying"); + } finally { context.finish(); } + }); it("forwards a retained fresh directory intent on later launch calls and checks provider support", async () => { const value = row(); value.lease.metadata.runnerTaskProjectAuthValidation.projectIds = []; diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index 02d42e92072..e72af716e18 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -274,9 +274,10 @@ export async function resolveEnvironmentExecutionTarget(input: { concurrentSyncOperations: false, duplexCommandStream: false, runnerWebSocketIngress: true }, sandboxLeaseAcquisition:acquisition, ...(lease.metadata?.runnerTaskProbe ? { runnerTaskProbe: lease.metadata.runnerTaskProbe as { runId: string; sessionId: string } } : {}), - runnerTask: async operation => { + runnerTask: async (operation, context) => { const result = await runtime.task({ companyId: input.companyId, leaseId: lease.id, - operation: environmentTaskOperationSchema.parse(operation) }); + operation: environmentTaskOperationSchema.parse(operation), + ...(context?.executionContext ? { executionContext: context.executionContext } : {}) }); if (result.kind === "description" && result.taskId !== acquisition.providerLeaseId) { acquisition.outcome = "replacement"; acquisition.providerLeaseId = result.taskId; diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index 1ef71b839d2..a74aa9bf924 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -12,6 +12,7 @@ import { pluginRegistryService } from "./plugin-registry.js"; import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt } from "./remote-execution-termination.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import { NativeExecutionContext } from "./native-runtime/native-execution-context.js"; +import { pluginAgentRuntimeIdentity } from "./plugin-agent-runtime-state.js"; export async function pluginEnvironmentSupportsTasks(db: Db, workers: PluginWorkerManager, lease: { companyId: string; metadata: Record | null }) { const metadata = lease.metadata ?? {}; @@ -230,11 +231,13 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage } if (warmAttachment && operation.kind === "prepare") { const i = operation.requirements.identity; + const startup = context?.pluginSupported ? { contextId: context.id, deadlineAtMs: context.deadlineAtMs, runId: i.runId } : undefined; 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 }, + ...(startup ? { runnerTaskStartup: startup } : {}), })}::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}); @@ -335,6 +338,17 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage return snapshot.credential; }); } + const executionContext = context?.pluginSupported && ["describe", "prepare", "check_project_auth", "connection"].includes(operation.kind) ? await context.projection({ pluginId: plugin.id, + leaseId: lease.id, taskId, environmentId: environment!.id }, async () => { + try { + const [identity, configuration] = await Promise.all([ + pluginAgentRuntimeIdentity(db, plugin.id, input.companyId, run!.agentId), + pluginRegistryService(db).getConfig(plugin.id, input.companyId), + ]); + if (!identity) throw new Error("Missing transport binding"); + return { instanceSecrets: identity.instanceSecrets, configuration: configuration?.configJson ?? {} }; + } catch { throw new Error("Plugin task transport unavailable"); } + }) : undefined; const params = { driverKey: metadata.driverKey, companyId: input.companyId, environmentId: environment?.id ?? null, issueId: lease.issueId, config: driverConfig, @@ -342,10 +356,13 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage lease: { providerLeaseId: lease.providerLeaseId, metadata: lease.metadata ?? undefined }, ...(workspaceBinding ? { workspace: workspaceBinding } : {}), ...(freshWorkspace && scopesProjects ? { fresh: freshWorkspace } : {}), operation, + executionContextSupported: 1 as const, + ...(executionContext ? { executionContext } : {}), }; const timeout = operation.kind === "copilot_metadata" ? Math.max(1, Math.min(40_000, Number(probe?.expiresAtMs) - Date.now())) - : 15_000; + : executionContext ? Math.max(1, context!.deadlineAtMs - Date.now()) : 15_000; const result = await callWorker("environmentTask", params, timeout); + if (executionContext) context!.assertOpen(); const parsed = parseEnvironmentTaskResult(operation, taskId, result); if (operation.kind === "read_file") { const [current] = await db.select({ lease: environmentLeases, run: heartbeatRuns }).from(environmentLeases) 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 67194b6a9f3..35845b35f5d 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -86,6 +86,7 @@ import { import * as issueServiceModule from "../issues.js"; import * as runEventWriter from "../heartbeat-run-events.js"; import { NativePermissionDeclinedError } from "./native-permission-decline.js"; +import { NativeExecutionContext } from "./native-execution-context.js"; import { resolvePaperclipRunnerNativeProviderInput } from "./provider-profile.js"; import { createNativeHarnessBackupStamp, @@ -6669,6 +6670,86 @@ function cancellationDb(options?: { } describe("native startup cancellation fence", () => { + it.each(["deadline", "stop"])("retires a claimed warm owner when %s interrupts its reuse diagnostic", async mode => { + const first: NativeExecutionInputV1 = { ...execution, + binding: { ...execution.binding, runId: `reuse-log-first-${mode}`, executionWorkspaceId: `reuse-log-workspace-${mode}` }, + session: { ...execution.session, normalizedSessionId: `reuse-log-session-${mode}`, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } } }; + const second = { ...first, binding: { ...first.binding, runId: `reuse-log-second-${mode}` } }; + const third = { ...first, binding: { ...first.binding, runId: `reuse-log-third-${mode}` } }; + const close = vi.fn(async () => undefined), session = { close }; + const replacement = { close: vi.fn(async () => undefined) }; + const result = { result: { summary: "done" }, terminal: { runTerminalState: "succeeded" }, turnId: "first", + normalizedSessionId: first.session.normalizedSessionId, providerSessionId: "reuse-log-provider", driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + state.execute.mockReset().mockImplementationOnce(async options => { await options.onSession(session); return result; }) + .mockImplementationOnce(async options => { + expect(options.existingSession).toBeUndefined(); + await options.onSession(replacement); + return { ...result, turnId: "third" }; + }); + await executePaperclipNativeSession({ db: leaseDb(first), execution: first, runnerInstanceId: `reuse-log-${mode}` }); + const originalRenewal = nativeLeaseRenewal.startNativeSessionExecutionLeaseRenewal; + const renewal = vi.spyOn(nativeLeaseRenewal, "startNativeSessionExecutionLeaseRenewal").mockImplementationOnce(input => ({ + ...originalRenewal(input), remainingMs: () => mode === "deadline" ? 250 : 60_000, + })); + let entered!: () => void, release!: () => void; + const writing = new Promise(resolve => { entered = resolve; }); + const written = new Promise(resolve => { release = resolve; }); + const db = leaseDb(second); + try { + const startup = executePaperclipNativeSession({ db, execution: second, runnerInstanceId: `reuse-log-${mode}`, + onEvent: async event => { + if (event.eventType === "run.startup.reuse" && event.payload?.layer === "runner") { entered(); await written; } + } }); + const failure = startup.catch(error => error); + await writing; + const stopping = mode === "stop" ? cancelNativeSession(second.binding.runId, "Stop during reuse write") : undefined; + expect(await failure).toMatchObject({ message: mode === "deadline" ? "native_execution_startup_expired" : "native_cancellation_pending_recovery" }); + await stopping; + expect(close).toHaveBeenCalledOnce(); + release(); + await executePaperclipNativeSession({ db: leaseDb(third), execution: third, runnerInstanceId: `reuse-log-${mode}` }); + expect(state.execute).toHaveBeenCalledTimes(2); + } finally { release(); renewal.mockRestore(); await closeIdleWarmNativeSessionsForRestart(); } + }); + + it("uses the common startup context and durable Stop fence for a retained local session", async () => { + const first: NativeExecutionInputV1 = { ...execution, + binding: { ...execution.binding, runId: "local-context-first", executionWorkspaceId: "local-context-workspace" }, + session: { ...execution.session, normalizedSessionId: "local-context-session", lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } } }; + const second = { ...first, binding: { ...first.binding, runId: "local-context-second" } }; + const stopped: Record = {}; + const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); + const session = { cancel, close: vi.fn(async () => undefined) }; + const submit = vi.fn(); + state.execute.mockReset().mockImplementationOnce(async options => { + expect(options.startup).toBeUndefined(); + await options.onSession(session); + return { result: { summary: "done" }, terminal: { runTerminalState: "succeeded" }, turnId: "first", + normalizedSessionId: first.session.normalizedSessionId, providerSessionId: "retained-local", driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }).mockImplementationOnce(async options => { + expect(options.existingSession).toBe(session); + expect(options.startup).toMatchObject({ deadlineAtMs: expect.any(Number), signal: expect.any(AbortSignal) }); + expect(options.startup.execution).toBe(second); + expect(Number.isSafeInteger(options.startup.deadlineAtMs)).toBe(true); + expect(options.startup.deadlineAtMs - Date.now()).toBeLessThanOrEqual(1_000); + stopped.startupCancellation = { requestedAt: new Date().toISOString() }; + try { await options.onSession(session); submit(); } + finally { await options.onSession(null); } + throw new Error("unexpected provider dispatch"); + }); + await executePaperclipNativeSession({ db: leaseDb(first), execution: first, runnerInstanceId: "local-context" }); + const createRenewal = nativeLeaseRenewal.startNativeSessionExecutionLeaseRenewal; + const renewal = vi.spyOn(nativeLeaseRenewal, "startNativeSessionExecutionLeaseRenewal").mockImplementation(input => ({ + ...createRenewal(input), remainingMs: () => 1_000.75, + })); + onTestFinished(() => renewal.mockRestore()); + await expect(executePaperclipNativeSession({ db: leaseDb(second, {}, stopped), execution: second, runnerInstanceId: "local-context" })) + .rejects.toThrow("native_cancellation_pending_recovery"); + expect(cancel).toHaveBeenCalledOnce(); + expect(submit).not.toHaveBeenCalled(); + }); it.each(["current", "stale"] as const)("reattaches a %s remote recovery claim only to deliver an existing Stop", async mode => { const { currentNativeControllerIdentity } = await import("./native-restart-recovery.js"); const controller = await currentNativeControllerIdentity(); @@ -12069,8 +12150,8 @@ describe("runnerd provider runtime wiring", () => { }; } - it("admits a warm follow-up after resuming a suspended plugin session", async () => { - const first = { ...scopedNativeExecution("plugin-cold-then-warm"), + it.each(["reuse", "unsettled-describe", "claim-delay", "claim-expired", "claim-rollback", "unsupported", "fallback-clear-failed", "fallback-clear-expired", "fallback-clear-stopped"])("owns retained plugin startup through %s", async mode => { + const first = { ...scopedNativeExecution(`plugin-cold-then-warm-${mode}`), schema: "paperclip.native-execution-input.v5" as const, executionMode: "default" as const, provider: { kind: "codex" as const, model: null, approvalPolicy: "never" as const }, planningContext: null, runtimeContext: nativeRuntimeContextFixture(), @@ -12090,11 +12171,34 @@ describe("runnerd provider runtime wiring", () => { const home = join(isolatedStateDirectory, "plugin-account"); await mkdir(home); await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 }); - const task = vi.fn(async (operation: any) => operation.kind === "read_state" + let finishWorker = () => {}; + let finishFallbackClear = () => {}; + let fallbackStop: Promise | undefined; + let capturedContext: NativeExecutionContext | undefined; + const claimUpdates: Array<{ table: unknown; values: Record }> = []; + const savedContext = () => { + const profile = claimUpdates.find(update => update.table === heartbeatRuns && update.values.runnerProfileJson)?.values.runnerProfileJson; + if (!profile) return undefined; + const query = new PgDialect().sqlToQuery(profile as SQL); + const value = query.params.find(value => typeof value === "string" && value.includes('"nativeStartupContext"')); + return value ? JSON.parse(String(value)).nativeStartupContext : undefined; + }; + const task = vi.fn(async (operation: any, options?: { executionContext?: NativeExecutionContext }) => { + if (operation.kind === "describe" && options?.executionContext) { + capturedContext = options.executionContext; + if (mode.startsWith("claim-") || mode === "unsupported") expect(savedContext()).toMatchObject({ + contextId: capturedContext.id, deadlineAtMs: capturedContext.deadlineAtMs, leaseId: "followup-lease" }); + } + if (mode === "unsettled-describe" && operation.kind === "describe" && options?.executionContext) { + finishWorker = options.executionContext.trackWorker(); + throw new Error("worker caller timeout"); + } + return operation.kind === "read_state" ? { kind: "state", taskId: "replacement-task", state: { ...durableRunnerState(identity, "suspended"), outbox: [], pendingTerminalDelivery: null } } : { kind: "description", taskId: "replacement-task", launchPreparationVersion: 1, - projectAuthCheckVersion: 1, warmTaskVersion: 1, runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }); + ...(mode === "claim-delay" ? { executionContextVersion: 1 } : {}), projectAuthCheckVersion: 1, warmTaskVersion: 1, runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; + }); const resumed = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-resumed` } }; const followup = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-followup` } }; const session = { close: vi.fn(async () => undefined) }; @@ -12115,7 +12219,7 @@ describe("runnerd provider runtime wiring", () => { const options = { runnerInstanceId: identity.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, managedAiCredentialHome: home, managedAiCredentialIdentity: "revision", managedAiAccountIdentity: "account" }; const dbFor = (current: NativeExecutionInput, previous: NativeExecutionInput) => { - const db = leaseDb(current); + const db = leaseDb(current, {}, {}, current === followup ? claimUpdates : []); const select = db.select.bind(db); db.select = ((fields?: Record) => fields?.status === heartbeatRuns.status && fields?.runnerProfileJson === heartbeatRuns.runnerProfileJson ? priorNativeRunDb(previous).select(fields as never) @@ -12125,19 +12229,99 @@ describe("runnerd provider runtime wiring", () => { const target = { kind: "remote" as const, transport: "sandbox" as const, environmentId: "plugin-environment", remoteCwd: ".", leaseId: "replacement-lease", runnerTask: task, sandboxLeaseAcquisition: { outcome: "replacement" as const, providerLeaseId: "replacement-task" } }; + let budget: ReturnType | undefined; + const finish = NativeExecutionContext.prototype.finish; + const finished: NativeExecutionContext[] = []; + const finishSpy = vi.spyOn(NativeExecutionContext.prototype, "finish").mockImplementation(function (this: NativeExecutionContext) { finished.push(this); finish.call(this); }); try { await expect(executePaperclipNativeSession({ ...options, execution: resumed, db: dbFor(resumed, first), runnerExecutionTarget: target as never })).resolves.toMatchObject({ exitCode: 0 }); + if (mode === "unsettled-describe") { + const runtime = await import("../../vendor/paperclip-runner/index.js"); + budget = vi.spyOn(runtime, "coldAdmissionTimeoutMs").mockReturnValue(1_000); + await expect(executePaperclipNativeSession({ ...options, execution: followup, db: dbFor(followup, resumed), + runnerExecutionTarget: { ...target, leaseId: "followup-lease", sandboxLeaseAcquisition: { + outcome: "resumed", providerLeaseId: "replacement-task" } } as never })) + .rejects.toMatchObject({ message: "native_startup_cleanup_pending", requiresPreparationStop: false }); + expect(state.execute).toHaveBeenCalledOnce(); + expect(rebind).not.toHaveBeenCalled(); + return; + } const checkpointReads = task.mock.calls.filter(([operation]) => operation.kind === "read_state").length; - await expect(executePaperclipNativeSession({ ...options, execution: followup, db: dbFor(followup, resumed), - runnerExecutionTarget: { ...target, leaseId: "followup-lease", sandboxLeaseAcquisition: { - outcome: "resumed", providerLeaseId: "replacement-task" } } as never })).resolves.toMatchObject({ exitCode: 0 }); + const nextDb = dbFor(followup, resumed); + const transaction = nextDb.transaction.bind(nextDb); + if (mode === "claim-rollback") nextDb.transaction = (async (...args: Parameters) => { + const result = await transaction(...args); + if (savedContext()) throw new Error("claim rollback"); + return result; + }) as Db["transaction"]; + if (mode.startsWith("fallback-clear-")) { + const update = nextDb.update.bind(nextDb); + nextDb.update = ((table: unknown) => { + const builder = update(table as never); + return { ...builder, set: (values: Record) => { + if (table === heartbeatRuns && values.runnerProfileJson && + new PgDialect().sqlToQuery(values.runnerProfileJson as SQL).sql.includes("- 'nativeStartupContext'")) { + return { where: () => ({ returning: async () => { + if (mode === "fallback-clear-failed") return []; + if (mode === "fallback-clear-stopped") fallbackStop = cancelNativeSession(followup.binding.runId, "Stop during fallback clear"); + return new Promise(resolve => { finishFallbackClear = () => resolve([{ id: followup.binding.runId }]); }); + } }) }; + } + return builder.set(values as never); + } }; + }) as unknown as Db["update"]; + } + if (mode.startsWith("claim-") || mode === "unsupported" || mode.startsWith("fallback-clear-")) { + const runtime = await import("../../vendor/paperclip-runner/index.js"); + budget = vi.spyOn(runtime, "coldAdmissionTimeoutMs").mockReturnValue(mode === "claim-expired" ? 10 : mode === "fallback-clear-expired" ? 250 : 1_000); + } + const following = executePaperclipNativeSession({ ...options, execution: followup, db: nextDb, + onExecutionLeaseAcquired: async () => { + if (mode === "claim-delay" || mode === "claim-expired") { + expect(savedContext()).toMatchObject({ runId: followup.binding.runId, leaseId: "followup-lease" }); + await new Promise(resolve => setTimeout(resolve, 25)); + } + }, runnerExecutionTarget: { ...target, leaseId: "followup-lease", sandboxLeaseAcquisition: { + outcome: "resumed", providerLeaseId: "replacement-task" } } as never }); + if (mode === "claim-expired" || mode === "claim-rollback") { + await expect(following).rejects.toThrow(mode === "claim-expired" ? "native_execution_startup_expired" : "claim rollback"); + expect(capturedContext).toBeUndefined(); + expect(finished).toContainEqual(expect.objectContaining({ id: savedContext().contextId })); + expect(finished.find(context => context.id === savedContext().contextId)?.active).toBe(false); + expect(state.execute).toHaveBeenCalledOnce(); + return; + } + if (mode === "fallback-clear-expired" || mode === "fallback-clear-stopped") { + await expect(following).rejects.toThrow(mode === "fallback-clear-expired" ? "native_execution_startup_expired" : "native_cancellation_pending_recovery"); + expect(state.execute).toHaveBeenCalledOnce(); + expect(rebind).not.toHaveBeenCalled(); + return; + } + if (mode === "fallback-clear-failed") { + await expect(following).rejects.toThrow("Native startup fallback lost authority"); + expect(state.execute).toHaveBeenCalledOnce(); + expect(rebind).not.toHaveBeenCalled(); + return; + } + await expect(following).resolves.toMatchObject({ exitCode: 0 }); + const clears = claimUpdates.filter(update => update.table === heartbeatRuns && update.values.runnerProfileJson && + new PgDialect().sqlToQuery(update.values.runnerProfileJson as SQL).sql.includes("- 'nativeStartupContext'")); + expect(clears).toHaveLength(mode === "claim-delay" ? 0 : 1); + if (mode === "claim-delay") expect(state.execute.mock.calls[1]![0].startup).toMatchObject({ + id: savedContext().contextId, deadlineAtMs: savedContext().deadlineAtMs }); + if (mode === "unsupported") expect(capturedContext?.active).toBe(false); expect(state.execute.mock.calls[1]![0].existingSession).toBe(session); expect(rebind).toHaveBeenCalledOnce(); expect(session.close).not.toHaveBeenCalled(); expect(task.mock.calls.filter(([operation]) => operation.kind === "read_state")).toHaveLength(checkpointReads); expect(existsSync(receiptPath)).toBe(false); } finally { + finishWorker(); + finishFallbackClear(); + await fallbackStop; + finishSpy.mockRestore(); + budget?.mockRestore(); owner.mockRestore(); resume.mockRestore(); } @@ -12163,6 +12347,35 @@ describe("runnerd provider runtime wiring", () => { return state.createTransport.mock.calls[0]![0].stateDirectory!; } + it("rejects a late backend factory after its startup context has finished", async () => { + const current = scopedNativeExecution("late-context-factory"); + state.createBackend.mockClear(); + const db = leaseDb(current), select = db.select.bind(db); + let release!: (value: unknown[]) => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + db.select = ((fields?: Record) => { + if (fields?.contextSnapshot !== heartbeatRuns.contextSnapshot) return select(fields as never); + const query = { from: () => query, where: () => query, limit: () => { entered(); return held; } }; + return query; + }) as Db["select"]; + const context = new NativeExecutionContext(current, 1_000), task = vi.fn(); + let factory!: ReturnType; + const opening = context.wait(() => factory = createRunnerdBackend({ db, execution: current, runnerInstanceId: "late-context-runner", + executionContext: context, runnerIngressAuthorized: true, runnerExecutionTarget: { + kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: "lease", runnerTask: task, + } })); + await Promise.race([started, opening]); + const rejected = expect(opening).rejects.toThrow("operator Stop"); + context.abort(new Error("operator Stop")); + await rejected; + context.finish(); + release([{ contextSnapshot: {} }]); + await expect(factory).rejects.toThrow("operator Stop"); + expect(task).not.toHaveBeenCalled(); + expect(state.createBackend).not.toHaveBeenCalled(); + }); + async function writeRunnerAuthority(root: string, identity: Record, lifecycle: string) { await mkdir(join(root, "control-plane"), { recursive: true }); await mkdir(join(root, "runner"), { recursive: true }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 93992b4870d..12f079da14a 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -32,6 +32,9 @@ import { resolvePluginTaskResume } from "../environment-task-runtime.js"; import { environmentTaskResultSchema } from "@paperclipai/plugin-sdk"; import { preparePluginRunnerCredentials } from "./plugin-runner-credentials.js"; import { createPluginRunnerTaskOwner, verifyPluginRunnerTaskReattachment } from "./plugin-runner-task.js"; +import { NativeExecutionContext } from "./native-execution-context.js"; +import { NativeStartupCleanupPendingError, clearNativeStartupContext } from "./native-startup-cleanup.js"; +import { coldAdmissionTimeoutMs } from "../../vendor/paperclip-runner/index.js"; import { applyObservabilityEnvironment, type ProviderObservabilityLaunch } from "@paperclipai/adapter-utils/agent-observability"; import { museRunnerBroker } from "../muse-runner-broker.js"; import { nativeCaseHandoffResult } from "./native-case-handoff.js"; @@ -309,6 +312,7 @@ type NativeSessionStartup = { priorRestartDetach?: symbol; leaseOwner?: string; executionLeaseRenewal?: NativeExecutionLeaseRenewal; + executionContext?: NativeExecutionContext; settled: Promise; settle: () => void; }; @@ -7752,6 +7756,7 @@ async function performNativeCancellation( let settleActiveCancellation: (() => void) | undefined; if (startup) { startup.stopRequested = true; + startup.executionContext?.abort(new Error("native_session_cancelled")); startup.cancellationSettled = new Promise(resolve => { settleStartupCancellation = resolve; }); } try { @@ -8245,6 +8250,10 @@ export async function executePaperclipNativeSession(input: { await deferNativeControllerRecovery(input.db, { ...input.execution.binding, leaseOwner: input.leaseOwner, retryAt: error.retryAt }); } + if (!(error instanceof NativeRunnerOwnershipUnverifiedError) && !(error instanceof NativeStartupCleanupPendingError) && startup.stopRequested && startup.executionContext?.signal.reason instanceof Error && + startup.executionContext.signal.reason.message === "native_session_cancelled") { + throw new NativeCancellationPendingRecoveryError(); + } const remoteRunnerRecovery = readRemoteRunnerRecoveryError(error); if (remoteRunnerRecovery) { await input.onEvent?.({ @@ -8256,6 +8265,12 @@ export async function executePaperclipNativeSession(input: { } throw error; } finally { + let startupUnsettled = false; + if (startup.executionContext) { + if (startup.executionContext.active) startup.executionContext.abort(new Error("native_execution_startup_completed")); + startupUnsettled = !(await startup.executionContext.drain()); + startup.executionContext.finish(); + } if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) { executingNativeOwnerScopes.delete(ownerScope); } @@ -8301,6 +8316,9 @@ export async function executePaperclipNativeSession(input: { companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId, issueId: input.execution.binding.issueId }); + if (startupUnsettled) throw new NativeStartupCleanupPendingError(startup.executionContext!.id, + startup.executionContext!.deadlineAtMs, startup.leaseOwner!, startup.executionContext!.pluginSupported, + input.runnerExecutionTarget?.kind === "remote" ? input.runnerExecutionTarget.leaseId ?? undefined : undefined); } } @@ -8490,6 +8508,8 @@ async function executePaperclipNativeSessionWithinScope( let controllerGeneration = 0; const providerPackAdmissionEpoch = randomUUID(); let providerPackAdmissionSource: NativeProviderPackAdmissionSource | undefined; + let confirmedLeaseDeadline = 0; + let claimedStartupContext: NativeExecutionContext | undefined; try { attempt = await trace.measure( "native.coordinator.claim", @@ -8641,7 +8661,17 @@ async function executePaperclipNativeSessionWithinScope( : coordinator.controllerBootId === controller.bootId ? Math.max(1, coordinator.controllerGeneration) : coordinator.controllerGeneration + 1; + const target = input.runnerExecutionTarget; + const retained = warmNativeSessions.get(nativeSessionScopeKey(input.execution)); + if (!input.restartRecovery && target?.kind === "remote" && target.transport === "sandbox" && target.runnerTask && target.leaseId && + retained?.pluginTaskId && idleWarmNativeSessionOwnerRunId(input) !== undefined && + (input.execution.provider.kind === "codex" || (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "claude"))) { + claimedStartupContext = new NativeExecutionContext(input.execution, Math.min(executionLeaseTtlMs, + coldAdmissionTimeoutMs(input.execution.provider.kind, input.execution.provider.kind === "acpx" ? input.execution.provider.agent : undefined, false))); + nativeSessionStartups.get(input.execution.binding.runId)!.executionContext = claimedStartupContext; + } const leaseNow = new Date(); + confirmedLeaseDeadline = performance.now() + executionLeaseTtlMs; const claimed = await tx .update(nativeRunFinalizations) .set({ @@ -8686,7 +8716,12 @@ async function executePaperclipNativeSessionWithinScope( .set({ nativePhase: "observed", nativePhaseUpdatedAt: leaseNow, - runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ nativeProviderPackAdmissionState: nativeProviderPackAdmissionPending(providerPackAdmissionEpoch) })}::jsonb`, + runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ nativeProviderPackAdmissionState: nativeProviderPackAdmissionPending(providerPackAdmissionEpoch), + ...(claimedStartupContext ? { nativeStartupContext: { schema: "paperclip.native-startup-context.v1", + companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, runId: input.execution.binding.runId, + contextId: claimedStartupContext.id, deadlineAtMs: claimedStartupContext.deadlineAtMs, leaseId: target?.kind === "remote" ? target.leaseId : null, + leaseOwner, controllerGeneration: nextControllerGeneration, attempt: nextAttempt } } : {}), + })}::jsonb`, updatedAt: leaseNow, }) .where(eq(heartbeatRuns.id, coordinator.runId)); @@ -8702,11 +8737,17 @@ async function executePaperclipNativeSessionWithinScope( companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, leaseOwner, attempt: claimedAttempt, controller, remote: input.runnerExecutionTarget?.kind === "remote", - onLost: () => input.runnerExecutionTarget?.kind === "remote" - ? detachNativeSessionsForRestart([input.execution.binding.runId], leaseOwner) - : cancelNativeSession(input.execution.binding.runId, "native session execution lease lost"), + confirmedDeadline: confirmedLeaseDeadline, + onLost: () => { + startup.executionContext?.abort(new Error("native_session_lease_lost")); + return input.runnerExecutionTarget?.kind === "remote" + ? detachNativeSessionsForRestart([input.execution.binding.runId], leaseOwner) + : cancelNativeSession(input.execution.binding.runId, "native session execution lease lost"); + }, }); + claimedStartupContext?.assertOpen(); await input.onExecutionLeaseAcquired?.(); + claimedStartupContext?.assertOpen(); return claimedAttempt; }), { parentName: "task.prepare" }, @@ -8724,7 +8765,9 @@ async function executePaperclipNativeSessionWithinScope( nativeSessionOwnerScope(input.execution, input.runnerExecutionTarget?.environmentId ?? null), nativeSessionScopeKey(input.execution), nativeSessionKey(input.execution), ); + claimedStartupContext?.assertOpen(); if (prepareAfterClaim) input = await prepareAfterClaim(); + claimedStartupContext?.assertOpen(); await trace.end(taskPrepareScope); } catch (error) { await trace.end(taskPrepareScope, { outcome: "failed" }); @@ -9254,16 +9297,41 @@ async function executePaperclipNativeSessionWithinScope( let releaseGitHubRun: (() => void) | undefined; let persistedWarmSession: PersistedNativeSession | null | undefined = stagedRestart && !stagedRestart.hasTargetCheckpoint ? stagedRestart.checkpoint : undefined; + let executionContext: NativeExecutionContext | undefined; + let claimedStartupContextCleared = false; + const finishStartupContextForFallback = async () => { + if (claimedStartupContext && !claimedStartupContextCleared) { + const context = claimedStartupContext; + await context.wait(() => clearNativeStartupContext(input.db, { companyId: input.execution.binding.companyId, + runId: input.execution.binding.runId, contextId: context.id, leaseOwner, controllerGeneration })); + claimedStartupContextCleared = true; + } + (executionContext ?? claimedStartupContext)?.finish(); + executionContext = undefined; + startupForAdmission!.executionContext = undefined; + }; if (warmSessionId !== null && warmConfigDigest !== null && !input.restartRecovery) { const entry = warmNativeSessions.get(warmSessionId); if (entry) { + if (input.execution.provider.kind === "codex" || + (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "claude")) { + executionContext = claimedStartupContext ?? new NativeExecutionContext(input.execution, + Math.floor(Math.min(coldAdmissionTimeoutMs(input.execution.provider.kind, input.execution.provider.kind === "acpx" ? input.execution.provider.agent : undefined, false), + startupForAdmission!.executionLeaseRenewal!.remainingMs()))); + startupForAdmission!.executionContext = executionContext; + if (startupForAdmission!.stopRequested) executionContext.abort(new Error("native_session_cancelled")); + } if (entry.preparingRunId && entry.preparingRunId !== input.execution.binding.runId) throw new Error("native_session_supervisor_busy"); entry.preparingRunId = undefined; const taskTarget=input.runnerExecutionTarget; if (entry.pluginTaskId && taskTarget?.kind === "remote" && taskTarget.transport === "sandbox" && taskTarget.runnerTask) { - const description=environmentTaskResultSchema.parse(await taskTarget.runnerTask({kind:"describe"})); + const description=environmentTaskResultSchema.parse(await (executionContext + ? executionContext.wait(() => taskTarget.runnerTask!({kind:"describe"}, { executionContext })) + : taskTarget.runnerTask({kind:"describe"}))); if(description.kind !== "description") throw new Error("Plugin Runner description unavailable"); pluginTaskDescription=description; + if (executionContext && description.executionContextVersion === 1) executionContext.pluginSupported = true; + else if (executionContext) await finishStartupContextForFallback(); } // Old run-scoped environments still require process replacement. A // session-owned broker can change run authority without replacing it. @@ -9294,6 +9362,7 @@ async function executePaperclipNativeSessionWithinScope( (input.runnerEnvironment?.PAPERCLIP_RUNNER_NETWORK_ACCESS === "enabled") ) { + await finishStartupContextForFallback(); pluginTaskDescription=undefined; const preparedCopy = entry.instructionPreparationRunId === input.execution.binding.runId; if (entry.busy && !preparedCopy) throw new Error("native_session_supervisor_busy"); @@ -9315,7 +9384,8 @@ async function executePaperclipNativeSessionWithinScope( if (entry.instructionWorkingCopy?.runId && entry.instructionPreparationRunId !== input.execution.binding.runId) { throw new Error("native_instruction_materialization_not_claimed"); } - if (entry.instructionPreparationRunId !== undefined && await entry.instructionWorkingCopy!.hasChanges()) { + if (entry.instructionPreparationRunId !== undefined && await (executionContext + ? executionContext.wait(() => entry.instructionWorkingCopy!.hasChanges()) : entry.instructionWorkingCopy!.hasChanges())) { entry.closeOnReleaseReason = "warm instruction copy changed during preparation"; try { await closeWarmNativeSession(entry, entry.closeOnReleaseReason); @@ -9332,8 +9402,19 @@ async function executePaperclipNativeSessionWithinScope( if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); entry.idleTimer = null; existingWarmSession = entry.session; - await emitStartupReuse(input.onEvent, "runner", { state: "reused", reason: "verified_reuse" }); - await emitStartupReuse(input.onEvent, "provider_session", { state: "reused", reason: "verified_reuse" }); + const emitReuse = async () => { + await emitStartupReuse(input.onEvent, "runner", { state: "reused", reason: "verified_reuse" }); + executionContext?.assertOpen(); + await emitStartupReuse(input.onEvent, "provider_session", { state: "reused", reason: "verified_reuse" }); + }; + try { + if (executionContext) await executionContext.wait(emitReuse); else await emitReuse(); + } catch (error) { + if (lifecyclePolicy.mode === "warm") { + await releaseWarmNativeSession(warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, true).catch(() => undefined); + } + throw error; + } githubAccess = entry.githubAccess; } } else { @@ -9344,6 +9425,9 @@ async function executePaperclipNativeSessionWithinScope( ); } } + if ((claimedStartupContext && executionContext !== claimedStartupContext) || (executionContext && !existingWarmSession)) { + await finishStartupContextForFallback(); + } async function resolvePendingGovernedWait() { const handoff = await nativeCaseHandoffResult(input.db, { ...input.execution.binding, completionContract: input.execution.completionContract.contract }); @@ -9456,6 +9540,7 @@ async function executePaperclipNativeSessionWithinScope( resolveCredentials: (binding) => resolveGitHubOperationCredentials(input.db, binding), onLog: input.onLog, })); + executionContext?.assertOpen(); releaseGitHubRun = githubAccess.activate(input.execution.binding); input = { ...input, runnerEnvironment: { ...input.runnerEnvironment, ...githubAccess.env } }; } @@ -9466,17 +9551,19 @@ async function executePaperclipNativeSessionWithinScope( && (input.execution.provider.kind === "codex" || input.execution.provider.kind === "opencode") && input.execution.schema !== "paperclip.native-execution-input.v6" && input.execution.schema !== "paperclip.native-execution-input.v8"; - if (!overlapGitHubPreparation) await prepareRunnerEnvironment(); - const expectedCurrentWakeComments = await timeController("wake_binding", () => resolveCurrentWakeCommentsBinding( - input.db, - input.execution.binding, - )); + if (!overlapGitHubPreparation) { + if (executionContext) await executionContext.wait(prepareRunnerEnvironment); else await prepareRunnerEnvironment(); + } + const resolveWake = () => timeController("wake_binding", () => resolveCurrentWakeCommentsBinding(input.db, input.execution.binding)); + const expectedCurrentWakeComments = await (executionContext ? executionContext.wait(resolveWake) : resolveWake()); let remoteWarmRecoveryPending = remoteWarmRecovery !== undefined; let runnerdBackend = input.useRunnerd && input.backend === undefined - ? await timeController("backend_setup", () => createRunnerdBackend({ + ? await timeController("backend_setup", () => { + const create = () => createRunnerdBackend({ ...input, providerPackAdmissionSource, + executionContext, onAccountingEvent: producerUsage?.capture, controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? { bootId: controller.bootId, @@ -9499,7 +9586,9 @@ async function executePaperclipNativeSessionWithinScope( } : undefined, trace, toolTrace, - })) + }); + return executionContext ? executionContext.wait(create) : create(); + }) : null; if (remoteWarmRecovery) { await remoteWarmRecovery.revalidate(leaseOwner); @@ -9563,6 +9652,7 @@ async function executePaperclipNativeSessionWithinScope( }); }, input: runnerExecution, + startup: executionContext && (!usesPluginTask || executionContext.pluginSupported) ? executionContext : undefined, remoteCleanupScope: remoteCleanupLease ? remoteLeaseCleanupScope(remoteCleanupLease) : undefined, turnTimeoutMs: input.turnTimeoutMs, backend: @@ -9867,6 +9957,8 @@ async function executePaperclipNativeSessionWithinScope( await active.cancellationSettled; throw new NativeCancellationPendingRecoveryError(); } + executionContext?.assertOpen(); + executionContext?.finish(); } else { liveQuestions.close(); activeNativeSessions.delete(input.execution.binding.runId); @@ -9935,6 +10027,10 @@ async function executePaperclipNativeSessionWithinScope( clearSteeringDeliveries(input.execution.binding.runId); clearNativeRuntimeRequestResolutions(input.execution.binding.runId); } catch (error) { + executionContext?.abort(error instanceof Error ? error : new Error("native_execution_startup_failed")); + const startupUnsettled = executionContext ? !(await executionContext.drain()) : false; + if (startupUnsettled) throw new NativeStartupCleanupPendingError(executionContext!.id, executionContext!.deadlineAtMs, + leaseOwner, executionContext!.pluginSupported, input.runnerExecutionTarget?.kind === "remote" ? input.runnerExecutionTarget.leaseId ?? undefined : undefined); // Preserve typed failure semantics while removing credentials before any // persistence, reporting, logging, or rethrow at this boundary. if (error instanceof Error) { @@ -10001,7 +10097,7 @@ async function executePaperclipNativeSessionWithinScope( // finished; a missing result must not overwrite cancellation or trigger // another turn to perform completion bookkeeping. const stoppedBeforeFirstTurn = error instanceof Error && error.message === "native_session_cancelled"; - if (protocolIntegrityFailure === null && error instanceof Error && + if (!startupUnsettled && protocolIntegrityFailure === null && error instanceof Error && (stoppedBeforeFirstTurn || error instanceof NativePermissionDeclinedError || error.message === "native_finalization_missing: session returned no semantic result")) { const [stoppedRun] = await input.db.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.id, input.execution.binding.runId), @@ -12217,6 +12313,7 @@ export function remoteCheckpointIncompleteFailure( /** Production runnerd backend seam, exported so provider wiring can be regression tested. */ export async function createRunnerdBackend(input: { + executionContext?: NativeExecutionContext; casesEnabled?: boolean; db: Db; providerPackAdmissionSource?: NativeProviderPackAdmissionSource; @@ -12284,6 +12381,7 @@ export async function createRunnerdBackend(input: { } initializingSessionToolAuthorities.add(sessionScopeId); try { + input.executionContext?.signal.throwIfAborted(); let retainedTransition: VerifiedWarmTransitionBinding | undefined; // executePaperclipNativeSession holds the full session-scope claim and // verifies/migrates the durable root before it acquires the coordinator @@ -12307,11 +12405,21 @@ export async function createRunnerdBackend(input: { runnerExecutionTarget: input.runnerExecutionTarget, }); } - return await createRunnerdBackendWithinSessionClaim( + input.executionContext?.signal.throwIfAborted(); + const backend = await createRunnerdBackendWithinSessionClaim( input, sessionScopeId, retainedTransition, ); + input.executionContext?.signal.throwIfAborted(); + return backend; + } catch (error) { + const authority = sessionToolAuthorityEpochs.get(sessionScopeId); + if (input.executionContext?.signal.aborted && authority?.runId === input.execution.binding.runId) { + authority.revoke(); + sessionToolAuthorityEpochs.delete(sessionScopeId); + } + throw error; } finally { initializingSessionToolAuthorities.delete(sessionScopeId); releaseRetainedNativeSessionCleanup(input.db, { @@ -12439,7 +12547,12 @@ async function createRunnerdBackendWithinSessionClaim( let recoveryPending = retainedTransition !== undefined; const target = input.runnerExecutionTarget ?? { kind: "local" as const }; const remoteTarget = target.kind === "remote" ? target : null; - const pluginTask = remoteTarget?.transport === "sandbox" ? remoteTarget.runnerTask : undefined; + const rawPluginTask = remoteTarget?.transport === "sandbox" ? remoteTarget.runnerTask : undefined; + const pluginTask = rawPluginTask && input.executionContext ? (operation: unknown) => { + const startupOperation = ["describe", "prepare", "check_project_auth", "connection"].includes(String(record(operation).kind)); + if (startupOperation) input.executionContext?.signal.throwIfAborted(); + return rawPluginTask(operation, input.executionContext?.active && startupOperation ? { executionContext: input.executionContext } : undefined); + } : rawPluginTask; let runnerControlPlane: DurablePrpControlPlane | 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")) { @@ -12510,6 +12623,7 @@ async function createRunnerdBackendWithinSessionClaim( ? await input.db.select({ config: agents.adapterConfig }).from(agents).where(and( eq(agents.id, input.execution.binding.agentId), eq(agents.companyId, input.execution.binding.companyId), )).limit(1).then(rows => rows[0]) : undefined; + input.executionContext?.signal.throwIfAborted(); const authority = new PaperclipRunnerToolAuthority(input.db, { casesEnabled: input.casesEnabled === true, ...(nativeReview ? { nativeReview } : {}), @@ -12558,6 +12672,7 @@ async function createRunnerdBackendWithinSessionClaim( >; try { dynamicTools = await timeController("tool_authority", () => authorityEpoch.definitions()); + input.executionContext?.signal.throwIfAborted(); } catch (error) { authorityEpoch.revoke(); throw error; @@ -12646,6 +12761,7 @@ async function createRunnerdBackendWithinSessionClaim( credentials.accountIdentity, input.managedAiCredentialWriteback); } const ownerInput: Parameters[0] = { ...credentials, + executionContext: input.executionContext?.pluginSupported ? input.executionContext : undefined, // 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, @@ -12741,6 +12857,7 @@ async function createRunnerdBackendWithinSessionClaim( itemId: adopting ? prior!.itemId : `item-${input.execution.binding.runId}` }, }).transport, }); + input.executionContext?.signal.throwIfAborted(); sessionToolAuthorityEpochs.get(sessionScopeId)?.revoke(); sessionToolAuthorityEpochs.set(sessionScopeId, authorityEpoch); return { ...backend, descriptor: () => backend.descriptor(), openSession: session => backend.openSession(session), @@ -14460,6 +14577,7 @@ async function createRunnerdBackendWithinSessionClaim( }, }); }; + input.executionContext?.signal.throwIfAborted(); const priorAuthorityEpoch = sessionToolAuthorityEpochs.get(sessionScopeId); if (priorAuthorityEpoch && priorAuthorityEpoch !== authorityEpoch) { priorAuthorityEpoch.revoke(); 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 f592f4f421a..b1525d74cca 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest"; import type { RunnerProviderLaunchRequirements } from "@paperclipai/paperclip-runner"; import type { RunnerProcessLaunchSpec } from "../../vendor/paperclip-runner/index.js"; import { createPluginRunnerTaskOwner } from "./plugin-runner-task.js"; +import { NativeExecutionContext } from "./native-execution-context.js"; const ingress = vi.hoisted(() => ({ connect: vi.fn() })); vi.mock("../../realtime/runner-prp-outbound.js", () => ({ connectRunnerPrpIngress: ingress.connect })); @@ -53,6 +54,20 @@ function provider() { } describe("plugin native Runner ownership", () => { + it.each(["finish", "rebind"])("does not send startup Stop after the context %s", mode => { + const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); + const next = new NativeExecutionContext(context.execution, 1_000); + const remove = vi.spyOn(context.signal, "removeEventListener"); + const { task } = provider(); + const input = { task, projectIds: [], harness: "codex", executionContext: context, adoptIdentity: requirements.identity }; + const owner = createPluginRunnerTaskOwner(input); + try { + if (mode === "finish") context.finish(); + else { owner.rebind({ ...input, executionContext: next }); expect(remove).toHaveBeenCalledWith("abort", expect.any(Function)); } + context.abort(new Error("later lease loss")); + expect(task).not.toHaveBeenCalled(); + } finally { context.finish(); next.finish(); } + }); it("rechecks revoked grants for unchanged project IDs before adopting retained resources", async () => { const { task } = provider(); const perform = task.getMockImplementation()!; @@ -217,6 +232,22 @@ describe("plugin native Runner ownership", () => { await expect(prepared).rejects.toThrow("preparation cancelled"); expect(task.mock.calls.some(([op]) => op.kind === "submit")).toBe(false); }); + it("sends Stop before a late preparation can publish a launch", async () => { + const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); + const { task } = provider(), perform = task.getMockImplementation()!; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + task.mockImplementation(async operation => { if (operation.kind === "prepare") await gate; return perform(operation); }); + const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, executionContext: context }); + const preparing = prepare(owner); + await vi.waitFor(() => expect(task).toHaveBeenCalledWith(expect.objectContaining({ kind: "prepare" }))); + context.abort(new Error("operator Stop")); + expect(task).toHaveBeenCalledWith({ kind: "stop" }); + release(); + await expect(preparing).rejects.toThrow("preparation cancelled"); + expect(task.mock.calls.some(([operation]) => ["check_project_auth", "connection", "submit"].includes(operation.kind))).toBe(false); + context.finish(); + }); it.each(["unsupported", "denied", "wrong-configuration", "wrong-provider"])("withholds credentials when project validation is %s", async mode => { const { task } = provider(); const perform = task.getMockImplementation()!; task.mockImplementation(async op => op.kind === "check_project_auth" diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index c39b9f7c4fe..92e0de33507 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -6,6 +6,7 @@ import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-co import { RUNNER_PRP_PROTOCOL_MIN, RUNNER_PRP_PROTOCOL_MAX, type RunnerProcessHandle, type RunnerProcessLaunchSpec, type DurablePrpControlPlane, type RunnerExecutionDriver, type RunnerLaunchPreparer, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js"; import { connectRunnerPrpIngress } from "../../realtime/runner-prp-outbound.js"; +import type { NativeExecutionContext } from "./native-execution-context.js"; export async function verifyPluginRunnerTaskReattachment( task: (operation: unknown) => Promise, identity: unknown, @@ -27,6 +28,7 @@ export async function verifyPluginRunnerTaskReattachment( * No provider pid is treated as a local pid. Only executionStopped settles it. */ export function createPluginRunnerTaskOwner(input: { + executionContext?: NativeExecutionContext; task: (operation: unknown) => Promise; projectIds: readonly string[]; harness: string; @@ -76,6 +78,21 @@ export function createPluginRunnerTaskOwner(input: { } return result; }; + let removeStartupAbort = () => {}; + const bindStartupAbort = () => { + removeStartupAbort(); + const context = input.executionContext; + if (!context) return; + const abort = () => { + if (input.executionContext !== context || !context.active) return; + stopRequested = true; + void call({ kind: "stop" }).catch(rejectFailure); + }; + context.signal.addEventListener("abort", abort, { once: true }); + removeStartupAbort = () => context.signal.removeEventListener("abort", abort); + if (context.signal.aborted) abort(); + }; + bindStartupAbort(); 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; @@ -341,6 +358,7 @@ export function createPluginRunnerTaskOwner(input: { JSON.stringify(input.projectIds) !== JSON.stringify(next.projectIds)) throw new Error("Plugin warm task cannot be rebound"); input = next; released = false; + bindStartupAbort(); }, };