diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index ab44e35d6e1..8b36618d840 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -217,3 +217,25 @@ This native loop does not advertise hosted parking support: Launcher retention, Core orchestration, provider-specific qualification, and live VM sleep/wake verification are required before rollout. Providers without an eligibility implementation remain ineligible. + + +### Optional controller release and resume + +An explicit session `release()` returns a copy of A's confirmed six-field source +identity. It returns `false` when support is absent or work is still active. +Those refusals leave active work and its controller route in place. Release and +released-source adoption require a registration checkpoint hook that owns process +cleanup. Without it, release returns `false` and adoption fails before claim. + +For an eligible release, the controller first makes its existing route quiet, +then waits for the host acknowledgement before it closes that route. A failed +request can have taken effect. Its retry keeps the same source identity; a +resume retry also keeps the exact bootstrap ticket. The route stays quiet while +the result is uncertain. Resume's pinned ticket deadline bounds claim +reconciliation. A late reply cannot grant authentication or work permission. + +A `resumed` result permits authentication under A. Ordinary `run.attach` must +still establish B before B can start work. An `unavailable` result reports known +absence of the retained Runner; it does not prove that descendants have stopped +or release their cleanup ownership. This optional contract adds no automatic +release policy or capability advertisement. diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index 56caad7ecab..50e56be89cd 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -4,6 +4,7 @@ import type { HarnessSessionRecoveryOptions, PersistedHarnessSession, } from "../contracts/harness-driver.js"; +import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js"; import type { NativeSession, NativeSessionBackend, @@ -510,6 +511,11 @@ class HarnessNativeSession implements NativeSession { this.#explicitlyCancelled = false; } + async release(): Promise { + this.#assertProtocolIntegrity(); + return this.#session.release?.() ?? false; + } + async detachControllerForRestart(): Promise { if (this.#session.detachControllerForRestart === undefined) return; await this.#session.detachControllerForRestart(); diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index 94086d85ec8..e4d4387a54e 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -1,4 +1,5 @@ import type { NativeRunRuntimeGrant } from "./runtime-context.js"; +import type { DurableRecoveryIdentity } from "./durable-recovery.js"; import type { PrpEvent, PrpStructuredRunResult, @@ -583,6 +584,8 @@ export interface HarnessSession { usage?(): Promise | null>; transcript?(): Promise; snapshot(): Promise; + /** Return the confirmed released source, or false when active use was not released. */ + release?(): Promise; /** Relinquish controller authority without semantically closing the session. */ detachControllerForRestart?(): Promise; close(input: { reason: string; force?: boolean }): Promise; diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 333062e4703..77b8431eb89 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -1,4 +1,5 @@ import type { ProviderFailure } from "./provider-failure.generated.js"; +import type { DurableRecoveryIdentity } from "./durable-recovery.js"; import type { NativeRunRuntimeGrant } from "./runtime-context.js"; import type { NativeRunIdentity, @@ -201,6 +202,8 @@ export interface NativeSession { identity(): NativeRunIdentity; capabilities(): Promise; attachRun?(input: { identity: NativeRunIdentity; currentRunGrant?: NativeRunRuntimeGrant }): Promise; + /** Return the confirmed released source, or false when active use was not released. */ + release?(): Promise; /** Relinquish controller authority without suspending provider execution. */ detachControllerForRestart?(): Promise; events(input?: { afterCursor?: string | null }): AsyncIterable; diff --git a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts index abb1e85b2a5..79165267471 100644 --- a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts +++ b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts @@ -13,6 +13,14 @@ export interface RunnerExecutionDriver { codexSourceRecoveryVersion?: 1; /** The driver verifies a stopped, idle source for the selected ACPX provider before launch. */ acpxSourceRecoveryVersion?: 1; + /** An explicitly supported host release/resume pair; absence keeps ordinary warm ownership. + * Release and released-source adoption require the registration checkpoint cleanup hook. */ + releaseResume?: { + release(identity: DurableRecoveryIdentity): Promise; + resume(identity: DurableRecoveryIdentity, bootstrapTicket: string): Promise<"resumed" | "unavailable">; + /** Confirmed Core-owned completed-session proof, used only for cache adoption. */ + releasedSource?: DurableRecoveryIdentity; + }; /** Current authenticated lease/checkpoint-writer owner; may change during warm reuse. */ authorityRunId?: string; /** Own provider configuration and file preparation in the execution environment. */ diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index e06d5ca1263..32ae39d837f 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -1,4 +1,5 @@ import { configuredEnvironment } from "../../configured-environment.js"; +import type { DurableRecoveryIdentity } from "../../contracts/durable-recovery.js"; import type { NativeTurnControlCapabilities } from "../../contracts/types.js"; import type { NativeRunRuntimeGrant } from "../../contracts/runtime-context.js"; import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; @@ -64,6 +65,8 @@ export interface CodexAppServerTransport { * diagnostic context; transports must not forward it to the provider. */ close(reason?: string): Promise; + /** Return the confirmed released source, or false when active use was not released. */ + release?(): Promise; /** Relinquish controller authority while leaving durable runner work alive. */ detachControllerForRestart?(): Promise; processInfo?(): CodexTransportProcessInfo; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index eb60ed356b5..f4fdcd13d0c 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -1,4 +1,5 @@ import { readCodexThreadState, readCodexTurnMetadata, readCodexTurnItems } from "./codex-history.js"; +import type { DurableRecoveryIdentity } from "../../contracts/durable-recovery.js"; import { codexRunUsage, codexUsageBaselineForNextRun, observeCodexUsage } from "./codex-usage-baseline.js"; import { randomUUID } from "node:crypto"; import { isNativeRestartInterruption, NativeProviderTerminalFailure, NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js"; @@ -918,6 +919,11 @@ export class CodexHarnessSession await this.transport.close(input?.reason); } + async release(): Promise { + this.assertProtocolIntegrity(); + return this.transport.release?.() ?? false; + } + async detachControllerForRestart(): Promise { if (this.transport.detachControllerForRestart === undefined) return; await this.transport.detachControllerForRestart(); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-release-resume.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-release-resume.test.ts new file mode 100644 index 00000000000..fc5232c61d6 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-release-resume.test.ts @@ -0,0 +1,422 @@ +import { createHash, randomBytes } from "node:crypto"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { createServer } from "node:net"; +import { request as httpRequest } from "node:http"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import type { DurablePrpControlPlane, RunnerProcessHandle, PrpWireConnection, TransportCloseReason } from "../control-plane/durable-prp-control-plane.js"; +import { createCapabilityRunnerdCodexTransport, readRunnerdArtifactBinding } from "./runnerd-codex-transport.js"; +import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; +import { createPrivateHostControlFixture } from "../../test/support/private-host-control.js"; +import { CodexAppServerDriver } from "../drivers/codex/codex-app-server-driver.js"; +import { createCodexTaskEnvelope } from "../contracts/codex.js"; +import type { HarnessSession, PersistedHarnessSession } from "../contracts/harness-driver.js"; +import { QUALIFIED_ACPX_PROFILES } from "../drivers/acpx/qualified-profiles.js"; +import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js"; + +const digest = (value: string) => createHash("sha256").update(value).digest("hex"); +const read = async (path: string) => JSON.parse(await readFile(path, "utf8")); +const parkId = (source: Record) => `park_${digest(JSON.stringify(Object.fromEntries(Object.entries(source).sort(([a], [b]) => a.localeCompare(b)))))}`; + +async function nativeWire(port: number, path: string, closeGate?: Promise): Promise { + return new Promise((resolve, reject) => { + const request = httpRequest({ hostname: "127.0.0.1", port, path, headers: { + Connection: "Upgrade", Upgrade: "websocket", "Sec-WebSocket-Version": "13", "Sec-WebSocket-Key": randomBytes(16).toString("base64"), + } }); + request.on("error", () => reject(new Error("Fixture listener unavailable"))); + request.on("response", response => { response.resume(); reject(new Error("Fixture listener upgrade refused")); }); + request.setTimeout(1_000, () => request.destroy(new Error("Fixture listener timeout"))); + request.on("upgrade", (_response, socket, head) => { + request.setTimeout(0); + let buffer = Buffer.alloc(0), handler: ((value: unknown) => void) | undefined; + const pending: unknown[] = [], closed = new Set<(reason: TransportCloseReason) => void>(); + const send = (opcode: number, payload: Buffer) => { + const mask = randomBytes(4), bytes = Buffer.from(payload); + for (let index = 0; index < bytes.length; index++) bytes[index] ^= mask[index % 4]!; + const header = Buffer.alloc(payload.length < 126 ? 2 : payload.length < 65_536 ? 4 : 10); + header[0] = 0x80 | opcode; + if (header.length === 2) header[1] = 0x80 | payload.length; + else if (header.length === 4) { header[1] = 0xfe; header.writeUInt16BE(payload.length, 2); } + else { header[1] = 0xff; header.writeBigUInt64BE(BigInt(payload.length), 2); } + socket.write(Buffer.concat([header, mask, bytes])); + }; + const consume = (chunk: Buffer) => { + buffer = Buffer.concat([buffer, chunk]); + while (buffer.length >= 2) { + if ((buffer[0]! & 0x80) === 0 || (buffer[1]! & 0x80) !== 0) { socket.destroy(); return; } + let length = buffer[1]! & 0x7f, offset = 2; + if (length === 126) { if (buffer.length < 4) return; length = buffer.readUInt16BE(2); offset = 4; } + else if (length === 127) { + if (buffer.length < 10) return; + const size = buffer.readBigUInt64BE(2); + if (size > 4_194_304n) { socket.destroy(); return; } + length = Number(size); offset = 10; + } + if (buffer.length < offset + length) return; + const opcode = buffer[0]! & 0x0f, payload = buffer.subarray(offset, offset + length); + buffer = buffer.subarray(offset + length); + if (opcode === 8) { socket.destroy(); return; } + if (opcode === 9) { send(10, payload); continue; } + if (opcode !== 1) continue; + try { const value: unknown = JSON.parse(payload.toString("utf8")); if (handler) handler(value); else pending.push(value); } + catch { socket.destroy(); return; } + } + }; + socket.on("data", consume); + socket.on("close", () => { + const notify = () => { for (const close of closed) close({ message: "fixture route closed" }); }; + if (closeGate) void closeGate.then(notify); else notify(); + }); + socket.on("error", () => {}); + resolve({ sendJson: value => send(1, Buffer.from(JSON.stringify(value))), close: () => socket.destroy(), + onJson: receive => { handler = receive; for (const value of pending.splice(0)) receive(value); }, + onClose: close => { closed.add(close); } }); + consume(head); + }); + request.end(); + }); +} + +it.skipIf(process.platform === "win32").each(["cached", "lost claim reply", "cache adoption", "acpx cache adoption", "unavailable", "claim cancellation", "expired ticket", "late claim reply", "held past deadline", "release admission", "no cleanup callback", "adoption no cleanup callback", "resume no cleanup callback", "deferred turn after close", "late wire close", "empty goal snapshot"] as const)( + "releases and resumes the native Codex source before attachment (%s)", async mode => { + const root = await mkdtemp(join(tmpdir(), "runner-release-resume-")); + const acpx = mode === "acpx cache adoption"; + const sidecar = join(root, "sidecar"), sidecarScript = join(root, "sidecar-entry"); + if (acpx) { + const quote = (value: string) => `'${value.replaceAll("'", "'\"'\"'")}'`; + const fake = resolve(import.meta.dirname, "../../runner/target/debug/fake-acpx-sidecar"); + await writeFile(sidecarScript, "qualified fixture entry"); + await writeFile(sidecar, `#!/bin/sh\nexec ${quote(fake)} --mode source-reconcile --profile-digest ${quote(QUALIFIED_ACPX_PROFILES.claude.commandDigest)} --journal ${quote(join(root, "calls"))} --source-fault ${quote(join(root, "fault"))}\n`, { mode: 0o700 }); + } + const listener = createServer(); + await new Promise(resolve => listener.listen(0, "127.0.0.1", resolve)); + const address = listener.address(); + if (!address || typeof address === "string") throw new Error("Fixture listener unavailable"); + const port = address.port; + await new Promise(resolve => listener.close(() => resolve())); + const fixture = createPrivateHostControlFixture(); + const identity = { runnerInstanceId: "retained-runner", environmentLeaseId: "retained-lease", runId: "source-run", + normalizedSessionId: "retained-session", turnId: "source-turn", itemId: "source-item" }; + const binary = defaultCapabilityRunnerdBinary(); + const runnerFile = join(root, "runner/runner-state.json"); + const claimTicketDigests: string[] = [], routeRuns: string[] = []; + let deliverOldClose!: () => void; + const oldCloseGate = new Promise(resolve => { deliverOldClose = resolve; }); + const goalGate = join(root, "goal-tail-gate"); + let authority: DurablePrpControlPlane | undefined; + let claimantRunId = identity.runId; + let handle: RunnerProcessHandle | undefined; + let loseReply = mode === "lost claim reply"; + let closeCheckpoint = 0, claimAccepted = false, adopting = false; + let acceptClaim!: () => void; + const acceptedClaim = new Promise(resolve => { acceptClaim = resolve; }); + let releaseClaim!: () => void; + const claimReceipt = new Promise(resolve => { releaseClaim = resolve; }); + let expireClock: ReturnType | undefined; + let holdMeasuredTurn = false, enterMeasuredTurn!: () => void, finishMeasuredTurn!: () => void; + const measuredTurn = new Promise(resolve => { enterMeasuredTurn = resolve; }); + const heldMeasuredTurn = new Promise(resolve => { finishMeasuredTurn = resolve; }); + const sockets = new Set(); + const registration = async (core: DurablePrpControlPlane, source = core.store.state.identity) => { + if ((mode === "late claim reply" || mode === "held past deadline") && claimAccepted) throw new Error("Fixture source authentication started after expiry"); + authority = core; + let socket: PrpWireConnection | undefined; + return { + connection: { mode: "listen" as const, listenAddress: "0.0.0.0" as const, listenPort: port, listenPath: `/api/runner/v1/connect/${source.runId}` }, + activate: async () => { + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + let candidate: PrpWireConnection; + try { candidate = await nativeWire(port, `/api/runner/v1/connect/${source.runId}`, mode === "late wire close" && routeRuns.length === 0 ? oldCloseGate : undefined); } + catch { await new Promise(resolve => setTimeout(resolve, 10)); continue; } + sockets.add(candidate); + socket = candidate; + const attachment = core.attachWireConnection(candidate); + await vi.waitFor(() => expect(attachment.isAuthenticated()).toBe(true), { timeout: 5_000 }); + routeRuns.push(source.runId); + return; + } + throw new Error("Fixture source authentication timed out"); + }, + checkpoint: mode === "no cleanup callback" || mode === "deferred turn after close" || (mode === "adoption no cleanup callback" && adopting) || (mode === "resume no cleanup callback" && claimAccepted) ? undefined : async (settlement: "settled" | "unsettled") => { + expect(settlement).toBe("unsettled"); + closeCheckpoint += 1; + await fixture.close(); + }, + release: async () => { socket?.close(); }, + }; + }; + const pair = { + release: vi.fn(async (source: typeof identity) => { + const state = await read(runnerFile); + const result = await fixture.request({ op: "park", id: parkId(source), identity: source, + lastControllerCommandSeq: state.lastControllerCommandSeq, ackedSourceSeq: state.ackedSourceSeq }); + if (result.status !== "parked") throw new Error("Fixture park receipt unavailable"); + if (mode === "late wire close" && source.runId === identity.runId) expect(authority!.activeRunnerConnectionCount()).toBe(1); + }), + resume: vi.fn(async (source: typeof identity, ticket: string): Promise<"resumed" | "unavailable"> => { + claimTicketDigests.push(digest(ticket)); + if (mode === "unavailable") { + expect(handle?.isExitConfirmed?.()).toBe(true); + return "unavailable"; + } + const result = await fixture.request({ op: "claim", id: `claim_${claimantRunId}`, + parkId: parkId(source), bootstrapTicket: ticket }); + if (result.status !== "claimed") throw new Error("Fixture claim receipt unavailable"); + claimAccepted = true; acceptClaim(); + if (mode === "claim cancellation" || mode === "held past deadline") await claimReceipt; + if (mode === "expired ticket" || mode === "late claim reply") { + expireClock = vi.spyOn(Date, "now").mockReturnValue(Date.now() + 61_000); + if (mode === "expired ticket") throw new Error("Fixture lost claim reply past ticket lifetime"); + } + if (loseReply) { loseReply = false; throw new Error("Fixture lost claim reply"); } + return "resumed"; + }), + }; + const acpxArguments = acpx ? ["--acpx-launch-authority-digest", `sha256:${digest("fixture authority")}`, + "--acpx-sidecar-command", sidecar, "--acpx-sidecar-command-sha256", `sha256:${digest(await readFile(sidecar, "utf8"))}`, + "--acpx-sidecar-script", sidecarScript, "--acpx-sidecar-script-sha256", `sha256:${digest("qualified fixture entry")}`] : []; + const launch = vi.fn((spec: Parameters[0]) => { handle = fixture.launch({ ...spec, args: [...spec.args, ...acpxArguments] }); return handle; }); + const options = { runnerBinary: binary, runnerArtifact: readRunnerdArtifactBinding(binary), stateDirectory: root, + codexCommand: fakeCodex, codexArgs: fakeCodexArgs(root, ...(mode === "empty goal snapshot" ? ["--durable-turn-ids", "--late-accounting-tail-gate", goalGate, "--late-active-goal"] : [])), prpIdentity: identity, + get authorityRunId() { return claimantRunId; }, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, + runnerProcessLauncher: launch, controlPlaneRegistration: registration, releaseResume: pair, + ...((mode === "release admission" || mode === "deferred turn after close") ? { measureOperation: async (phase: string, work: () => Promise): Promise => { + if (phase === "runner.transport.turn_start" && holdMeasuredTurn) { enterMeasuredTurn(); await heldMeasuredTurn; } + return work(); + } } : {}), + ...(acpx ? { provider: "acpx" as const, acpxAgent: "claude" as const, acpxRuntimeDirectory: join(root, "acpx"), + prepareLaunch: async (input: RunnerProviderLaunchRequirements) => ({ configurationId: `fixture-${input.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(input), + provider: createRunnerProviderLaunch(input, { options: {}, root, cwd: root, codexHome: root, codexCommand: "", providerNodeCommand: sidecar, opencodeProxyPath: "", acpxSidecarPath: sidecarScript }) }) } : {}) }; + const intervals = new Set(), originalInterval = globalThis.setInterval, originalClear = globalThis.clearInterval; + const armInterval = vi.spyOn(globalThis, "setInterval").mockImplementation((...args: Parameters) => { + const timer = originalInterval(...args); + if (args[1] === 5 || args[1] === 250) intervals.add(timer); + return timer; + }); + const clearInterval = vi.spyOn(globalThis, "clearInterval").mockImplementation((...args: Parameters) => { + intervals.delete(args[0]); originalClear(...args); + }); + let bundle = createCapabilityRunnerdCodexTransport(options); + let recoveredSession: HarnessSession | undefined; + const complete = async () => { + if (recoveredSession) { + await recoveredSession.startTurn({ message: { role: "user", text: "Complete the fixture follow-up" } }); + for await (const event of recoveredSession.events()) if (event.eventType === "turn.completed") return; + throw new Error("Fixture follow-up terminal unavailable"); + } + const terminal = (async () => { for await (const event of bundle.transport.notifications()) if (event.method === "turn/completed") return; })(); + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Complete the fixture turn" }] }); + await terminal; + }; + try { + const opened = await bundle.transport.request("thread/start", { cwd: root, dynamicTools: [], ...(acpx ? { model: "explicit-source-model" } : {}) }); + const threadId = String((opened.thread as { id: string }).id); + await complete(); + if (mode === "empty goal snapshot") { + await writeFile(goalGate, "release late goal"); + await vi.waitFor(() => expect(authority!.store.state.committedEvents.some(event => event.eventType === "session.goal.updated" + && ((event.envelope.payload as { payload?: { goal?: { status?: string } } }).payload?.goal?.status === "active"))).toBe(true)); + await complete(); + await vi.waitFor(() => expect(authority!.store.state.committedEvents.filter(event => ["session.goal.updated", "session.goal.snapshot"].includes(event.eventType)).at(-1)?.envelope.payload).toMatchObject({ payload: { goal: null, workingNow: false } })); + const goalEvents = authority!.store.state.committedEvents.filter(event => ["session.goal.updated", "session.goal.snapshot"].includes(event.eventType)); + const latest = goalEvents.at(-1)!; + expect(latest.eventType).toBe("session.goal.snapshot"); + expect(latest.envelope.payload).toMatchObject({ payload: { goal: null, workingNow: false } }); + expect(goalEvents.slice(0, -1).some(event => (event.envelope.payload as { payload?: { goal?: { status?: string } } }).payload?.goal?.status === "active")).toBe(true); + } + if (mode === "deferred turn after close") { + holdMeasuredTurn = true; + const starting = bundle.transport.request("turn/start", { input: [{ type: "text", text: "Must not start after close" }] }); + void starting.catch(() => undefined); + await measuredTurn; + await bundle.transport.close(); + const commands = authority!.store.state.commands.length; + expect(handle!.isExitConfirmed?.()).toBe(true); + finishMeasuredTurn(); + await expect(starting).rejects.toThrow("closed"); + expect(authority!.store.state.commands).toHaveLength(commands); + return; + } + if (mode === "cached") { + const core = authority!, events = core.store.state.committedEvents, last = events.at(-1)!; + for (const [eventType, payload] of [["turn.started", {}], ["session.goal.updated", { goal: { status: "active" } }], + ["session.goal.updated", { goal: { status: "paused" }, workingNow: true }], + ["session.goal.cleared", { goal: null, workingNow: true }], ["session.goal.snapshot", { goal: null, workingNow: true }]] as const) { + events.push({ ...last, eventType, envelope: { ...last.envelope, payload: { payload } } }); + try { expect(await bundle.transport.release!()).toBe(false); } + finally { events.pop(); } + expect(pair.release).not.toHaveBeenCalled(); + expect(core.activeRunnerConnectionCount()).toBe(1); + } + const unsettled = vi.spyOn(core, "semanticToolResultsSettled").mockReturnValue(false); + try { expect(await bundle.transport.release!()).toBe(false); } + finally { unsettled.mockRestore(); } + expect(pair.release).not.toHaveBeenCalled(); + await complete(); + } + if (mode === "no cleanup callback") { + const released = await bundle.transport.release!(); + if (released === false) await complete(); + await expect(bundle.transport.close()).resolves.toBeUndefined(); + expect(released).toBe(false); + expect(pair.release).not.toHaveBeenCalled(); + expect(closeCheckpoint).toBe(0); + expect(handle!.isExitConfirmed?.()).toBe(true); + return; + } + if (mode === "release admission") { + holdMeasuredTurn = true; + const earlierStart = bundle.transport.request("turn/start", { input: [{ type: "text", text: "Deferred measured turn" }] }); + void earlierStart.catch(() => undefined); + await measuredTurn; + const core = authority!, originalFlush = core.flushCheckpoint.bind(core); + let entered!: () => void, finish!: () => void; + const flushing = new Promise(resolve => { entered = resolve; }); + const held = new Promise(resolve => { finish = resolve; }); + const flush = vi.spyOn(core, "flushCheckpoint").mockImplementationOnce(async () => { + entered(); await held; await originalFlush(); + }); + const releasing = bundle.transport.release!(); + try { + await flushing; + const commands = core.store.state.commands.length; + finishMeasuredTurn(); + await expect(earlierStart).rejects.toThrow("run_attachment_in_progress"); + for (const method of ["thread/read", "thread/resume", "thread/turns/list", "thread/goal/get"]) { + await expect(bundle.transport.request(method, { threadId })).rejects.toThrow("run_attachment_in_progress"); + } + await expect(bundle.transport.request("initialize", {})).resolves.toEqual({ user: {} }); + await expect(bundle.transport.request("turn/start", { input: [{ type: "text", text: "Must stay fenced" }] })) + .rejects.toThrow("run_attachment_in_progress"); + await expect(bundle.transport.request("thread/goal/set", { threadId, objective: "Must stay fenced" })) + .rejects.toThrow("run_attachment_in_progress"); + expect(core.store.state.commands).toHaveLength(commands); + } finally { finishMeasuredTurn(); finish(); flush.mockRestore(); await Promise.allSettled([earlierStart, releasing]); } + } + await vi.waitFor(async () => expect(await bundle.transport.release!()).toEqual(identity), { timeout: 5_000 }); + if (mode === "cached") { + const source = await bundle.transport.release!(); + expect(source).toEqual(identity); + if (source !== false) source.runId = "changed-return-value"; + expect(await bundle.transport.release!()).toEqual(identity); + expect(pair.release).toHaveBeenCalledOnce(); + } + if (mode === "late wire close") expect(authority!.activeRunnerConnectionCount()).toBe(0); + await vi.waitFor(() => expect(authority?.activeRunnerConnectionCount()).toBe(0)); + expect(intervals.size).toBe(0); + expect(launch).toHaveBeenCalledOnce(); + await expect(bundle.transport.request("turn/start", { input: [] })).rejects.toThrow("release_resume_pending"); + await expect(bundle.transport.request("thread/resume", { threadId })).rejects.toThrow("release_resume_pending"); + claimantRunId = "claimant-run"; + if (mode === "cache adoption" || mode === "acpx cache adoption" || mode === "adoption no cleanup callback") { + await bundle.transport.detachControllerForRestart!(); + adopting = true; + bundle = createCapabilityRunnerdCodexTransport({ ...options, get authorityRunId() { return claimantRunId; }, + releaseResume: { ...pair, releasedSource: identity }, + adoptExistingRunner: { pid: handle!.child.pid ?? null, processGroupId: handle!.processGroupId ?? null, + startedAt: handle!.startedAt ?? null, isAlive: () => handle!.isExitConfirmed?.() === false, + signal: () => { handle!.child.kill("SIGKILL"); return true; } }, + resumeProviderSession: { driverSessionId: threadId, providerSessionId: threadId }, + }); + const request = vi.spyOn(bundle.transport, "request"); + const driver = new CodexAppServerDriver({ conversationMode: "direct", + taskEnvelope: createCodexTaskEnvelope({ objective: "Complete the fixture turn", contractRevision: "fixture", criteria: [] }), + transportFactory: () => bundle.transport, workingDirectoryAuthority: "remote_runner", environment: { PAPERCLIP_WORKSPACE_CWD: root }, + ...(acpx ? { capabilities: { goals: false, threadLineage: false }, driverIdentity: { kind: "acpx_runtime", displayName: "Claude via ACPX", version: "fixture" } } : {}) }); + const original = opened.thread as { sessionId: string; providerIdentity?: PersistedHarnessSession["providerIdentity"] }; + const recovered = await driver.recoverSession({ driverKind: acpx ? "acpx_runtime" : "codex_app_server", + driverSessionId: threadId, providerSessionId: original.sessionId, providerIdentity: original.providerIdentity, + runId: identity.runId, normalizedSessionId: identity.normalizedSessionId }); + if (mode === "adoption no cleanup callback") { + expect(recovered.recovered).toBe(false); + expect(recovered.reason).toContain("released_cleanup_unavailable"); + expect(pair.resume).not.toHaveBeenCalled(); + await bundle.transport.close(); + expect(closeCheckpoint).toBe(0); + expect(handle!.isExitConfirmed?.()).toBe(false); + return; + } + expect(recovered.recovered, recovered.reason).toBe(true); + recoveredSession = recovered.session; + expect(request.mock.calls.map(([method]) => method)).toEqual(expect.arrayContaining(["thread/read", "thread/turns/list", "thread/resume"])); + expect((await recoveredSession!.snapshot()).driverSessionId).toBe(threadId); + await expect(bundle.transport.request("thread/resume", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch"); + await expect(bundle.transport.request("turn/start", { input: [] })).rejects.toThrow("release_resume_pending"); + await expect(bundle.transport.request("thread/goal/set", { threadId, objective: "Must stay fenced" })).rejects.toThrow("release_resume_pending"); + } + if (mode === "unavailable") await fixture.close(); + if (mode === "held past deadline") vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const attaching = Promise.resolve(recoveredSession + ? recoveredSession.attachRun!({ runId: claimantRunId }) + : bundle.transport.attachRun!({ runId: claimantRunId, turnId: "next-turn", itemId: "next-item" })); + void attaching.catch(() => undefined); + if (mode === "held past deadline") { + await acceptedClaim; + let failure: unknown; + const settled = attaching.catch(error => { failure = error; }); + await vi.advanceTimersByTimeAsync(61_000); + expect(failure).toBeInstanceOf(Error); + expect((failure as Error).message).toContain("reconciliation_expired"); + vi.useRealTimers(); + await settled; + await bundle.transport.close(); + releaseClaim(); + await Promise.resolve(); + expect(routeRuns).toEqual([identity.runId]); + } else if (mode === "claim cancellation") { + await vi.waitFor(() => expect(claimAccepted).toBe(true)); + const closing = bundle.transport.close(); + expect(closeCheckpoint).toBe(0); + releaseClaim(); + await expect(attaching).rejects.toThrow("closed"); + await closing; + } else if (mode === "unavailable" || mode === "expired ticket" || mode === "late claim reply" || mode === "resume no cleanup callback") { + await expect(attaching).rejects.toThrow(mode === "unavailable" ? "reuse_unavailable" : mode === "resume no cleanup callback" ? "released_cleanup_unavailable" : "reconciliation_expired"); + expireClock?.mockRestore(); expireClock = undefined; + await bundle.transport.close(); + } else await attaching; + expect(new Set(claimTicketDigests).size).toBe(1); + if (["unavailable", "claim cancellation", "expired ticket", "late claim reply", "held past deadline", "resume no cleanup callback"].includes(mode)) { + expect(pair.resume).toHaveBeenCalledOnce(); + expect(authority!.store.state.identity.runId).toBe(identity.runId); + expect(launch).toHaveBeenCalledOnce(); + expect(closeCheckpoint).toBe(1); + expect(handle!.isExitConfirmed?.()).toBe(true); + expect(intervals.size).toBe(0); + return; + } + expect(pair.resume).toHaveBeenCalledTimes(mode === "lost claim reply" ? 2 : 1); + if (mode === "cache adoption" || mode === "acpx cache adoption" || mode === "late wire close") { + expect(routeRuns.filter(runId => runId === identity.runId)).toHaveLength(2); + } + if (mode === "late wire close") { + deliverOldClose(); + await new Promise(resolve => setImmediate(resolve)); + expect(authority!.activeRunnerConnectionCount()).toBe(1); + } + expect(routeRuns).toContain(identity.runId); + expect((await read(runnerFile)).runId).toBe(claimantRunId); + expect(launch).toHaveBeenCalledOnce(); + await complete(); + const nextSource = structuredClone(authority!.store.state.identity); + await vi.waitFor(async () => expect(await bundle.transport.release!()).toEqual(nextSource), { timeout: 5_000 }); + expect(intervals.size).toBe(0); + await bundle.transport.close(); + expect(closeCheckpoint).toBe(1); + expect(handle!.isExitConfirmed?.()).toBe(true); + } finally { + if (mode === "held past deadline") vi.useRealTimers(); + expireClock?.mockRestore(); releaseClaim(); finishMeasuredTurn(); deliverOldClose(); + try { + for (const socket of sockets) socket.close(); + await fixture.close(); + await bundle.transport.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } finally { + for (const timer of intervals) originalClear(timer as ReturnType); + armInterval.mockRestore(); clearInterval.mockRestore(); + } + } + }, 30_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 89b1dd47967..9df1aba9e7a 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3869,6 +3869,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #detachPromise: Promise | null = null; #startupOperation: Promise | null = null; #attachmentOperation: Promise | null = null; + #releaseEpisode: { + source: DurableRecoveryIdentity; confirmed: boolean; claimed?: boolean; + claimantRunId?: string; ticket?: string; ticketFlushed?: boolean; deadline?: number; + } | null = null; #startupDrainTimedOut = false; #controllerDetachedForRestart = false; #failure: Error | null = null; @@ -4041,6 +4045,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }; } + #assertRunWorkAdmission(commandType?: string): void { + const held = this.#attachmentOperation !== null || (commandType !== undefined && this.#releaseEpisode !== null); + if (this.options.releaseResume && held && (commandType === undefined || !isRunnerRecoveryMaintenanceCommand(commandType))) { + throw new Error("native_runner_run_attachment_in_progress"); + } + } + #retainsSourceInputs(): boolean { const identity = this.#eventIdentity ?? this.options.prpIdentity; return this.#authorityRunId() !== undefined && identity !== undefined && identity !== null @@ -4057,6 +4068,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ): Promise> { if (this.#closed) throw new Error("PRP Codex transport is closed"); this.#throwIfFailed(); + if (!["initialize", "collaborationMode/list"].includes(method)) this.#assertRunWorkAdmission(); + if (this.#releaseEpisode && !(this.#releaseEpisode.claimed && ["initialize", "collaborationMode/list", "thread/read", "thread/resume", "thread/turns/list", "thread/items/list"].includes(method))) { + throw new Error("native_runner_release_resume_pending"); + } const expectedThread = this.#threadId || this.options.resumeProviderSession?.driverSessionId; if (expectedThread && params.threadId !== undefined && params.threadId !== expectedThread && ["thread/read", "thread/resume", "thread/turns/list", "thread/items/list", "thread/goal/get"].includes(method)) { @@ -4414,6 +4429,142 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); } + #quietReleasedRunner(): void { + if (this.#pump !== null) clearInterval(this.#pump); + this.#pump = null; + if (this.#adoptedRunnerMonitor !== null) clearInterval(this.#adoptedRunnerMonitor); + this.#adoptedRunnerMonitor = null; + } + + async #releaseSettledRunner(): Promise { + const pair = this.options.releaseResume; + const core = this.#core; + if (!pair || this.options.lifecyclePolicy?.mode !== "warm") return false; + this.#assertStartupOpen(); + this.#throwIfFailed(); + if (!core || !this.#startupComplete) throw new Error("native_runner_release_unavailable"); + if (!this.#controlPlaneCheckpoint) return false; + if (!this.#releaseEpisode) { + const knownBusy = () => { + const events = core.store.state.committedEvents; + const goalEvent = events.findLast(event => event.eventType === "session.goal.updated" || event.eventType === "session.goal.cleared" + || event.eventType === "session.goal.snapshot"); + const goalPayload = record(record(goalEvent?.envelope.payload).payload), goal = record(goalPayload.goal); + return this.#turnStartResponsePending || this.#bridgedRuntimeInputs.size > 0 + || !core.semanticToolResultsSettled() || providerTurnIsActiveFromCommittedEvents(events) + || goalPayload.workingNow === true + || (goalEvent?.eventType !== "session.goal.cleared" && (goal.status === "active" || goal.workingNow === true)); + }; + if (knownBusy() || !await this.#drainSettledProviderEventsBeforeSuspend() || knownBusy()) return false; + if (core.store.state.commands.some(command => command.status === "pending")) return false; + await core.flushCheckpoint(); + this.#assertStartupOpen(); + this.#releaseEpisode = { source: structuredClone(core.store.state.identity), confirmed: false }; + } + const episode = this.#releaseEpisode; + if (episode.ticket || !recoveryIdentityMatches(episode.source, core.store.state.identity)) { + throw new Error("native_runner_release_identity_conflict"); + } + this.#quietReleasedRunner(); + if (!episode.confirmed) { + await pair.release(episode.source); + this.#assertStartupOpen(); + core.disconnectActiveRunner(); + episode.confirmed = true; + } + return structuredClone(episode.source); + } + + release(): Promise { + if (this.#attachmentOperation) return Promise.reject(new Error("native_runner_run_attachment_in_progress")); + const operation = Promise.resolve().then(() => this.#releaseSettledRunner()); + const pending = operation.then(() => {}); + this.#attachmentOperation = pending; + void pending.catch(() => undefined); + return operation.finally(() => { if (this.#attachmentOperation === pending) this.#attachmentOperation = null; }); + } + + async #claimReleasedRunner(claimantRunId: string): Promise { + const episode = this.#releaseEpisode; + const pair = this.options.releaseResume; + const core = this.#core; + if (!episode || !episode.confirmed || !pair || !core || !recoveryIdentityMatches(episode.source, core.store.state.identity) + || !claimantRunId || claimantRunId === episode.source.runId + || (episode.claimantRunId && episode.claimantRunId !== claimantRunId)) { + throw new Error("native_runner_resume_identity_conflict"); + } + episode.claimantRunId = claimantRunId; + episode.deadline ??= Date.now() + RUNNER_BOOTSTRAP_TICKET_TTL_MS; + if (episode.claimed) return; + const expired = new Error("native_runner_resume_reconciliation_expired"); + const assertPending = () => { + this.#assertStartupOpen(); + if (Date.now() >= episode.deadline!) throw expired; + }; + assertPending(); + let deadlineTimer: ReturnType; + const deadline = new Promise((_resolve, reject) => { + deadlineTimer = setTimeout(() => reject(expired), episode.deadline! - Date.now()); + deadlineTimer.unref?.(); + }); + const closed = this.#closedSignal.then(() => { throw new Error("runnerd transport closed during provider startup"); }); + void closed.catch(() => undefined); + try { + while (!episode.claimed) { + assertPending(); + let outcome: "resumed" | "unavailable" | undefined; + try { + if (!episode.ticket) { + episode.ticket = core.issueBootstrapTicket(RUNNER_BOOTSTRAP_TICKET_TTL_MS); + } + if (!episode.ticketFlushed) { + await Promise.race([core.flushCheckpoint(), deadline, closed]); + assertPending(); + episode.ticketFlushed = true; + } + assertPending(); + outcome = await Promise.race([pair.resume(episode.source, episode.ticket), deadline, closed]); + } catch (error) { + // A generic RPC error cannot prove refusal. Keep this exact pending claim quiet. + assertPending(); + if (error === expired) throw error; + await Promise.race([new Promise(resolve => setTimeout(resolve, 250)), deadline, closed]); + continue; + } + assertPending(); + if (outcome === "unavailable") throw new Error("native_runner_reuse_unavailable"); + episode.claimed = true; + } + } finally { + clearTimeout(deadlineTimer!); + } + } + + async #resumeReleasedRunner(claimantRunId: string): Promise { + await this.#claimReleasedRunner(claimantRunId); + const core = this.#core!; + if (core.activeRunnerConnectionCount() !== 1) { + const registration = await this.options.controlPlaneRegistration?.(core, this.#releaseEpisode!.source); + try { + this.#assertStartupOpen(); + if (!registration?.checkpoint) throw new Error("native_runner_released_cleanup_unavailable"); + } catch (error) { + await Promise.resolve().then(() => registration?.release()).catch(() => undefined); + throw error; + } + this.#controlPlaneCheckpoint = registration.checkpoint; + this.#controlPlaneRelease = registration?.release ?? null; + await registration?.activate?.(); + await this.#awaitRegistrationReady(registration?.ready); + this.#assertStartupOpen(); + } + if (this.#pump === null) { + this.#pump = setInterval(() => this.#pumpEventsSafely(), 5); + this.#pump.unref?.(); + } + if (this.options.adoptExistingRunner && this.#adoptedRunnerMonitor === null) this.#watchAdoptedRunner(this.options.adoptExistingRunner); + } + attachRun(input: { runId: string; turnId: string; @@ -4460,6 +4611,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (retainingSource && this.#bridgedRuntimeInputs.size > 0) { throw new Error("native_runner_source_runtime_reconciliation_required"); } + if (this.#releaseEpisode) await this.#resumeReleasedRunner(input.runId); await this.#awaitWarmRunAttachmentReady(); if (retainingSource && !core.semanticToolResultsSettled()) { throw new Error("native_runner_source_semantic_reconciliation_required"); @@ -4624,6 +4776,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (!previousReleased) await previousRelease?.(); previousReleased = true; this.#controlPlaneRelease = registration?.release ?? null; + this.#releaseEpisode = null; if (retainingSource) { core.enableSemanticInputDispatch(desired); this.#recoverPendingSemanticInputsWhenReady(); @@ -5040,6 +5193,17 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } finally { clearTimeout(timer); } } this.#turnStartAdmission?.resolve(false); + if (this.#releaseEpisode) { + this.#quietReleasedRunner(); + try { + if (!this.#controlPlaneCheckpoint) throw new Error("native_runner_released_cleanup_unavailable"); + await this.#controlPlaneCheckpoint("unsettled"); + } finally { + await this.#retireControllerForRecovery(closeDeadline); + this.#controlPlaneCheckpoint = null; + } + return; + } const adoptedRunner = this.options.adoptExistingRunner; // `settled` is a durable-state assertion, not merely the absence of a // process handle. Registration can install a remote checkpoint callback @@ -6438,6 +6602,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await this.options.authorizeWarmTransitionRecovery?.("before_bootstrap"); } this.#assertStartupOpen(); + if (adoptedRunner && this.options.releaseResume?.releasedSource) { + const source = this.options.releaseResume.releasedSource; + if (!recoveryIdentityMatches(source, core.store.state.identity)) throw new Error("native_runner_released_source_mismatch"); + if (!this.#controlPlaneCheckpoint) throw new Error("native_runner_released_cleanup_unavailable"); + this.#releaseEpisode = { source: structuredClone(source), confirmed: true }; + await this.#claimReleasedRunner(this.#authorityRunId() ?? ""); + } const bootstrapTicket = adoptedRunner ? null : warmRecovery @@ -6813,6 +6984,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #startTurnWork( params: Record, ): Promise> { + this.#assertStartupOpen(); + this.#throwIfFailed(); + this.#assertRunWorkAdmission("turn.start"); const turnStartTimeoutMs = this.options.turnStartTimeoutMs ?? 30_000; if (!Number.isSafeInteger(turnStartTimeoutMs) || turnStartTimeoutMs <= 0) { throw new Error("turnStartTimeoutMs must be a positive safe integer"); @@ -7021,6 +7195,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { payload: Record, correlationId?: string, ): Promise { + this.#assertRunWorkAdmission(type); const core = this.#core; if (core === null) throw new Error("PRP provider thread is not started"); const commandId = correlationId @@ -7054,6 +7229,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { deadline?: number, abortOnClose = false, ): Promise> { + this.#assertRunWorkAdmission(type); const core = this.#core; if (core === null) throw new Error("PRP provider thread is not started"); const commandId = `command_lab_${randomUUID().replaceAll("-", "")}`; @@ -7794,11 +7970,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ): void { let checking = false; this.#adoptedRunnerMonitor = setInterval(() => { - if (checking || this.#closed) return; + if (checking || this.#closed || (this.#releaseEpisode && !this.#releaseEpisode.claimed)) return; checking = true; void Promise.resolve(adoptedRunner.isAlive()) .then((alive) => { - if (alive || this.#closed) return; + if (alive || this.#closed || (this.#releaseEpisode && !this.#releaseEpisode.claimed)) return; this.#evidence.runnerExited = true; this.#publish(); this.#failTransport( @@ -7808,7 +7984,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); }) .catch(() => { - if (!this.#closed) { + if (!this.#closed && !(this.#releaseEpisode && !this.#releaseEpisode.claimed)) { this.#failTransport( new Error( "native_adopted_runner_identity_unverifiable: runner liveness could not be revalidated", @@ -7868,7 +8044,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { const detail = result.stderr.trim() || result.stdout.trim(); if (detail) this.#diagnostic(detail.slice(-4_096)); this.#publish(); - if (this.#closed || this.#failure !== null || this.#handle !== handle) + if (this.#closed || this.#failure !== null || this.#handle !== handle || (this.#releaseEpisode && !this.#releaseEpisode.claimed)) return; // A per-turn runner exits after its terminal suffix is durably ACKed. // Drain that suffix into the provider-facing queue before classifying diff --git a/packages/paperclip-runner/test/support/private-host-control.py b/packages/paperclip-runner/test/support/private-host-control.py new file mode 100644 index 00000000000..b24a4b3d763 --- /dev/null +++ b/packages/paperclip-runner/test/support/private-host-control.py @@ -0,0 +1,127 @@ +"""Test-only unnamed fd4 handoff; stdout contains bounded control receipts only.""" +import fcntl +import json +import os +import signal +import socket +import subprocess +import sys +import threading + +LIMIT = 16 * 1024 + + +def process_identity(pid): + if sys.platform.startswith("linux"): + with open(f"/proc/{pid}/stat") as state: + fields = state.read(4096).rsplit(")", 1)[1].split() + return f"{int(fields[2])}:{int(fields[19])}" + if sys.platform == "darwin": + import ctypes + + class BsdInfo(ctypes.Structure): + _fields_ = [("prefix", ctypes.c_uint32 * 12), + ("comm", ctypes.c_char * 16), ("name", ctypes.c_char * 32), + ("suffix", ctypes.c_uint32 * 6), + ("seconds", ctypes.c_uint64), ("micros", ctypes.c_uint64)] + + info = BsdInfo() + lib = ctypes.CDLL("/usr/lib/libproc.dylib") + # PROC_PIDTBSDINFO includes PGID and microsecond process birth identity. + if lib.proc_pidinfo(pid, 3, 0, ctypes.byref(info), ctypes.sizeof(info)) != ctypes.sizeof(info) or info.prefix[3] != pid: + raise OSError("process identity unavailable") + return f"{info.suffix[1]}:{info.seconds}:{info.micros}" + raise OSError("unsupported test host") + + +def cleanup(): + pid, group = int(sys.argv[2]), int(sys.argv[3]) + expected = sys.argv[4] + if pid <= 1 or group <= 1 or not expected.startswith(f"{group}:") or process_identity(pid) != expected: + raise OSError("process ownership changed") + os.killpg(group, signal.SIGKILL) + + +def emit(value): + sys.stdout.write(json.dumps(value) + "\n") + sys.stdout.flush() + + +def main(): + left, right = socket.socketpair() + # Duplicate above fd4 before closing originals: either original may be fd4. + parent_fd = fcntl.fcntl(left.fileno(), fcntl.F_DUPFD_CLOEXEC, 5) + child_fd = fcntl.fcntl(right.fileno(), fcntl.F_DUPFD_CLOEXEC, 5) + left.close() + right.close() + parent = socket.socket(fileno=parent_fd) + os.dup2(child_fd, 4, inheritable=True) + os.close(child_fd) + try: + child = subprocess.Popen( + sys.argv[1:] + ["--host-control-fd", "4"], + pass_fds=(4,), stdin=subprocess.DEVNULL, stdout=sys.stderr, stderr=sys.stderr, + ) + finally: + os.close(4) + def close_channel(*_): + try: + parent.shutdown(socket.SHUT_RDWR) + except OSError: + pass + parent.close() + + signal.signal(signal.SIGTERM, close_channel) + output = threading.Lock() + try: + identity = process_identity(child.pid) + if not identity.startswith(f"{os.getpgrp()}:"): + raise OSError("unexpected native group") + except Exception: + close_channel() + try: + child.wait(timeout=3) + except subprocess.TimeoutExpired: + os.killpg(os.getpgrp(), signal.SIGKILL) + raise + emit({"fixture": "started", "pid": child.pid, "identity": identity}) + + def forward(): + try: + parent.settimeout(10) + reader = parent.makefile("rb") + while True: + frame = sys.stdin.buffer.readline(LIMIT + 1) + if not frame: + break + if len(frame) > LIMIT or not frame.endswith(b"\n"): + break + with output: + parent.sendall(frame) + reply = reader.readline(LIMIT + 1) + if not reply or len(reply) > LIMIT or not reply.endswith(b"\n"): + break + sys.stdout.buffer.write(reply) + sys.stdout.buffer.flush() + except (OSError, ValueError): + pass + finally: + close_channel() + + threading.Thread(target=forward, daemon=True).start() + # Never report completion based on the wrapper's own lifetime. + code = child.wait() + with output: + emit({"fixture": "exited", "code": code if code >= 0 else None, + "signal": signal.Signals(-code).name if code < 0 else None}) + + +if __name__ == "__main__": + try: + if len(sys.argv) == 5 and sys.argv[1] == "--cleanup": + cleanup() + else: + main() + except Exception: + # Do not reflect subprocess arguments, ticket frames, or provider output. + sys.exit(1) diff --git a/packages/paperclip-runner/test/support/private-host-control.ts b/packages/paperclip-runner/test/support/private-host-control.ts new file mode 100644 index 00000000000..b2e2894ad29 --- /dev/null +++ b/packages/paperclip-runner/test/support/private-host-control.ts @@ -0,0 +1,144 @@ +import { execFileSync, spawn, type ChildProcessByStdio } from "node:child_process"; +import { constants } from "node:os"; +import type { Readable, Writable } from "node:stream"; +import { fileURLToPath } from "node:url"; +import type { RunnerProcessHandle, RunnerProcessLaunchSpec, RunnerProcessResult } from "../../src/control-plane/durable-prp-control-plane.js"; + +const limit = 16 * 1024; +const wrapper = fileURLToPath(new URL("./private-host-control.py", import.meta.url)); +const failure = () => new Error("private_host_control_fixture_failed"); +type Reply = { id: string; status: string }; + +function bounded(promise: Promise, milliseconds: number): Promise { + let timer: NodeJS.Timeout; + return Promise.race([promise, new Promise((_, reject) => { + timer = setTimeout(() => reject(failure()), milliseconds); + })]).finally(() => clearTimeout(timer)); +} + +export function createPrivateHostControlFixture() { + let child: ChildProcessByStdio | undefined; + let nativePid: number | undefined; + let nativeIdentity: string | undefined; + let result: RunnerProcessResult | undefined; + let confirmed = false; + let closing = false; + let poisoned = false; + let buffer = Buffer.alloc(0); + let pending: { id: string; resolve(reply: Reply): void; reject(error: Error): void } | undefined; + let resolveReady!: () => void; + let rejectReady!: (error: Error) => void; + const ready = new Promise((resolve, reject) => { resolveReady = resolve; rejectReady = reject; }); + void ready.catch(() => {}); + let completion: Promise | undefined; + let serial: Promise = Promise.resolve(); + + const fail = () => { + poisoned = true; + rejectReady(failure()); + pending?.reject(failure()); + pending = undefined; + }; + const killGroup = () => { + if (child?.pid && child.exitCode === null && child.signalCode === null) { + try { process.kill(-child.pid, "SIGKILL"); } catch { /* already exited */ } + } else if (child?.pid && nativePid && nativeIdentity) { + // A dead wrapper does not release its native child's group. Recheck the + // original native generation before signaling; never bless a reused PID. + try { + execFileSync("python3", [wrapper, "--cleanup", String(nativePid), String(child.pid), nativeIdentity], { timeout: 2_000, stdio: "ignore" }); + } catch { /* absent or ownership unproven; completion still rejects */ } + } + }; + + return { + launch(spec: RunnerProcessLaunchSpec): RunnerProcessHandle { + if (child || closing || process.platform === "win32" || spec.nativeRunner?.connection?.mode !== "listen" || spec.args.includes("--host-control-fd")) throw failure(); + child = spawn("python3", [wrapper, spec.command, ...spec.args], { + cwd: spec.cwd, env: spec.environment, detached: true, stdio: ["pipe", "pipe", "ignore"], + }); + // OS-level discard also prevents descendants retaining a diagnostic pipe. + child.stdin.on("error", fail); + child.stdout.on("error", fail); + child.stdout.on("data", (chunk: Buffer) => { + buffer = Buffer.concat([buffer, chunk]); + let newline: number; + while ((newline = buffer.indexOf(10)) >= 0) { + if (newline + 1 > limit) { fail(); return; } + const line = buffer.subarray(0, newline); + buffer = buffer.subarray(newline + 1); + try { + const value: unknown = JSON.parse(line.toString("utf8")); + if (!value || typeof value !== "object" || Array.isArray(value)) throw failure(); + const record = value as Record; + if (record.fixture === "started" && Object.keys(record).length === 3 && nativePid === undefined && Number.isSafeInteger(record.pid) && Number(record.pid) > 0 && + typeof record.identity === "string" && /^\d+:\d+(?::\d+)?$/.test(record.identity) && record.identity.startsWith(`${child!.pid}:`)) { + nativePid = Number(record.pid); nativeIdentity = record.identity; resolveReady(); + } else if (record.fixture === "exited" && Object.keys(record).length === 3 && nativePid !== undefined && result === undefined && + (record.code === null || Number.isSafeInteger(record.code) && Number(record.code) >= 0) && + (record.signal === null || typeof record.signal === "string" && Object.hasOwn(constants.signals, record.signal)) && + (record.code === null) !== (record.signal === null)) { + result = { code: record.code as number | null, signal: record.signal as NodeJS.Signals | null, stdout: "", stderr: "" }; + } else if (pending && Object.keys(record).length === 2 && record.id === pending.id && + typeof record.status === "string" && ["parked", "claimed", "retired", "conflict"].includes(record.status)) { + pending.resolve({ id: pending.id, status: record.status }); pending = undefined; + } else throw failure(); + } catch { fail(); return; } + } + if (buffer.length > limit) fail(); + }); + completion = new Promise((resolve, reject) => { + child!.once("error", () => { fail(); reject(failure()); }); + child!.once("close", () => { + fail(); + if (result && !buffer.length) { confirmed = true; resolve(result); } + else reject(failure()); + }); + }); + void completion.catch(() => {}); + return { + child: { + get pid() { return nativePid; }, + get exitCode() { return confirmed ? result?.code ?? null : null; }, + get signalCode() { return confirmed ? result?.signal ?? null : null; }, + kill(signal = "SIGTERM") { + if (signal === "SIGKILL" || signal === 9) killGroup(); + else { + closing = true; + child!.stdin.end(); + if (nativePid !== undefined) child!.kill("SIGTERM"); + } + return !confirmed; + }, + }, + completion, processGroupId: child.pid, startedAt: new Date().toISOString(), + isExitConfirmed: () => confirmed, restart: null, + }; + }, + request(frame: Record): Promise { + const operation = serial.then(async () => { + if (!child || closing || poisoned || typeof frame.id !== "string" || !frame.id) throw failure(); + const id = frame.id; + let wire: string; + try { wire = `${JSON.stringify(frame)}\n`; } catch { throw failure(); } + if (Buffer.byteLength(wire) > limit) throw failure(); + await bounded(ready, 10_000); + if (closing || poisoned) throw failure(); + return bounded(new Promise((resolve, reject) => { + pending = { id, resolve, reject }; + child!.stdin.write(wire, (error) => { if (error) fail(); }); + }), 12_000).catch(() => { fail(); throw failure(); }); + }); + serial = operation.catch(() => {}); + return operation; + }, + async close(): Promise { + closing = true; + if (!child || !completion) return; + child.stdin.end(); + if (nativePid !== undefined && !confirmed) child.kill("SIGTERM"); + try { await bounded(completion, 15_000); } + catch { killGroup(); throw failure(); } + }, + }; +} diff --git a/server/src/realtime/runner-prp-outbound.test.ts b/server/src/realtime/runner-prp-outbound.test.ts index 86a702775f1..5c8523cb4dc 100644 --- a/server/src/realtime/runner-prp-outbound.test.ts +++ b/server/src/realtime/runner-prp-outbound.test.ts @@ -366,6 +366,67 @@ describe("runner provider-ingress WebSocket wire", () => { } }); + it.each(["readiness", "authenticated observer"])("does not start liveness when quiesce wins %s", async mode => { + vi.useFakeTimers(); + const create = sockets.create, socket = new FakeSocket(); + sockets.create = () => openingSocket(socket); + const endpoint = { kind: "authenticated_websocket" as const, + websocketUrl: "wss://fixture.example.test/connect", secretHeaders: [], generation: "fixture", + refresh: vi.fn(async () => endpoint), close: vi.fn(async () => {}) }; + const handle = connectRunnerPrpIngress({ authority: { + attachWireConnection: () => ({ isAuthenticated: () => { + if (mode === "readiness") handle.quiesce(); + return true; + } }), + } as unknown as DurablePrpControlPlane, endpoint, + onStateChange: state => { if (state === "authenticated" && mode === "authenticated observer") handle.quiesce(); } }); + try { + await vi.advanceTimersByTimeAsync(100_000); + expect(socket.ping).not.toHaveBeenCalled(); + expect(socket.terminate).not.toHaveBeenCalled(); + expect(socket.close).not.toHaveBeenCalled(); + expect(endpoint.refresh).not.toHaveBeenCalled(); + } finally { await handle.close(); sockets.create = create; vi.useRealTimers(); } + }); + + it("quiets liveness and reconnect while preserving the final receive/ACK path", async () => { + vi.useFakeTimers(); + const create = sockets.create; + const socket = new FakeSocket(); + let opens = 0; + sockets.create = () => { opens += 1; return openingSocket(socket); }; + const endpoint = { kind: "authenticated_websocket" as const, + websocketUrl: "wss://fixture.example.test/connect", secretHeaders: [], generation: "fixture", + refresh: vi.fn(async () => endpoint), close: vi.fn(async () => {}) }; + const received = vi.fn(); + const attach = vi.fn((wire: WsJsonWireConnection) => { + wire.onJson(value => { received(value); wire.sendJson({ kind: "ack" }); }); + return { isAuthenticated: () => true }; + }); + const handle = connectRunnerPrpIngress({ authority: { attachWireConnection: attach } as unknown as DurablePrpControlPlane, + endpoint, livenessIntervalMs: 10, livenessTimeoutMs: 30 }); + const failure = vi.fn(); + void handle.failure.catch(failure); + try { + await vi.advanceTimersByTimeAsync(0); + await handle.ready; + handle.quiesce(); + socket.ping.mockClear(); + socket.emit("message", Buffer.from(JSON.stringify({ kind: "last_event" })), false); + expect(received).toHaveBeenCalledWith({ kind: "last_event" }); + expect(socket.send).toHaveBeenCalledWith(JSON.stringify({ kind: "ack" })); + await vi.advanceTimersByTimeAsync(100_000); + expect(socket.ping).not.toHaveBeenCalled(); + expect(socket.terminate).not.toHaveBeenCalled(); + expect(socket.close).not.toHaveBeenCalled(); + socket.close(); + await vi.advanceTimersByTimeAsync(100_000); + expect(opens).toBe(1); + expect(endpoint.refresh).not.toHaveBeenCalled(); + expect(failure).not.toHaveBeenCalled(); + } finally { await handle.close(); sockets.create = create; vi.useRealTimers(); } + }); + it("does not report intentional close as an active-run ingress failure", async () => { const endpoint = { kind: "authenticated_websocket" as const, diff --git a/server/src/realtime/runner-prp-outbound.ts b/server/src/realtime/runner-prp-outbound.ts index f563d2a3d67..393bdbbce61 100644 --- a/server/src/realtime/runner-prp-outbound.ts +++ b/server/src/realtime/runner-prp-outbound.ts @@ -305,6 +305,8 @@ export interface RunnerPrpOutboundHandle { readonly ready: Promise; /** Rejects only when active-run recovery has exhausted its fixed budget. */ readonly failure: Promise; + /** Stop liveness and reconnect work, preserving an established receive/ACK path. */ + quiesce(): void; close(): Promise; } @@ -337,6 +339,7 @@ export function connectRunnerPrpIngress(input: { let readyResolve!: () => void; let readyReject!: (error: unknown) => void; let readySettled = false; + let stopLiveness = () => {}; const ready = new Promise((resolve, reject) => { readyResolve = resolve; readyReject = reject; @@ -386,22 +389,26 @@ export function connectRunnerPrpIngress(input: { "Runner ingress did not complete PRP authentication before the deadline.", ); } + if (abort.signal.aborted) return; if (!readySettled) { readySettled = true; readyResolve(); } input.onStateChange?.("authenticated"); + if (abort.signal.aborted) return; attempt = 0; leaseFailureCode = null; - const stopLiveness = opened.wire.monitorLiveness({ + stopLiveness = opened.wire.monitorLiveness({ intervalMs: input.livenessIntervalMs ?? DEFAULT_LIVENESS_INTERVAL_MS, timeoutMs: input.livenessTimeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS, }); + if (abort.signal.aborted) { stopLiveness(); return; } try { await socketClosed(opened.wire); } finally { stopLiveness(); } + if (abort.signal.aborted) return; recoveryDeadline = Date.now() + recoveryGraceMs; const leaseFailure = input.authority.connectionLeaseFailure ?? null; if (leaseFailure) { @@ -480,6 +487,11 @@ export function connectRunnerPrpIngress(input: { return { ready, failure, + quiesce() { + stopLiveness(); + // Established wires removed their handshake abort listener. Keep their ACK path. + abort.abort(); + }, async close() { abort.abort(); await Promise.all([...ownedWires].map(closeOwnedWire)); 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 0e866b013a8..34b7de9bbe0 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -538,6 +538,62 @@ describe("plugin Runner release/resume facade", () => { ? { kind: "released", taskId: "task" } : { kind: "resume", taskId: "task", outcome: "resumed" }); + it.each(["codex", "claude"])("quiesces %s before release and closes only after its ACK", async harness => { + const { task } = provider(); + const perform = task.getMockImplementation()!; + const order: string[] = []; + let acknowledge!: (value: unknown) => void; + task.mockImplementation(op => { + if (op.kind !== "release") return perform(op); + order.push("release"); + return new Promise(resolve => { acknowledge = resolve; }); + }); + const quiesce = vi.fn(() => { order.push("quiesce"); }); + const close = vi.fn(async () => { order.push("close"); }); + ingress.connect.mockReset().mockReturnValue({ ready: Promise.resolve(), quiesce, close }); + const owner = createPluginRunnerTaskOwner({ ...input(harness), task }); + const registration = await owner.registration({} as never, "run"); + registration.activate(); await registration.ready(); + try { + const releasing = owner.driver.releaseResume!.release(requirements.identity); + expect(order).toEqual(["quiesce", "release"]); + expect(close).not.toHaveBeenCalled(); + acknowledge({ kind: "released", taskId: "task" }); + await releasing; + expect(order).toEqual(["quiesce", "release", "close"]); + expect(task.mock.calls.at(-1)![0]).toEqual({ kind: "release", identity: requirements.identity }); + expect(ingress.connect).toHaveBeenCalledOnce(); + } finally { await registration.release("detach"); } + }); + + it.each(["codex", "claude"])("keeps an uncertain %s release quiet for an exact retry", async harness => { + const { task } = provider(); + const perform = task.getMockImplementation()!; + const lost = new Error("fixture release reply lost"); + let attempts = 0; + task.mockImplementation(async op => { + if (op.kind !== "release") return perform(op); + if (attempts++ === 0) throw lost; + return { kind: "released", taskId: "task" }; + }); + const quiesce = vi.fn(), close = vi.fn(async () => {}); + ingress.connect.mockReset().mockReturnValue({ ready: Promise.resolve(), quiesce, close }); + const owner = createPluginRunnerTaskOwner({ ...input(harness), task }); + const registration = await owner.registration({} as never, "run"); + registration.activate(); await registration.ready(); + try { + await expect(owner.driver.releaseResume!.release(requirements.identity)).rejects.toBe(lost); + expect(quiesce).toHaveBeenCalledOnce(); + expect(close).not.toHaveBeenCalled(); + await owner.driver.releaseResume!.release(requirements.identity); + expect(quiesce).toHaveBeenCalledTimes(2); + expect(close).toHaveBeenCalledOnce(); + expect(task.mock.calls.filter(([op]) => op.kind === "release").map(([op]) => op)) + .toEqual([{ kind: "release", identity: requirements.identity }, { kind: "release", identity: requirements.identity }]); + expect(ingress.connect).toHaveBeenCalledOnce(); + } finally { await registration.release("detach"); } + }); + it.each(["codex", "claude"])("forwards both %s operations through the current checked owner", async harness => { const firstTask = lifecycleTask(); const currentTask = lifecycleTask(); @@ -741,3 +797,44 @@ it.each(["codex", "claude"])("refuses warm %s account, harness, project, or cred await expect(prepare(owner)).rejects.toThrow("credential broker v1"); expect(task.mock.calls.map(([op])=>op.kind)).toEqual(["describe"]); }); + +it.each((["codex", "claude"] as const).flatMap(harness => ["prepared", "unprepared", "foreign target"].map(mode => [harness, mode] as const)))( + "authenticates adopted %s source before activating only its exact %s successor", async (harness, mode) => { + const { task } = provider(), perform = task.getMockImplementation()!; + const source = launchRequirements(harness), target = { ...source, identity: { ...source.identity, runId: "successor", turnId: "next-turn", itemId: "next-item" } }; + let nativeIdentity = source.identity; + task.mockImplementation(async operation => operation.kind === "read_state" + ? { kind: "state", taskId: "task", state: { schema: "paperclip.runner.durable.state.v1", lifecycle: "ready", ...nativeIdentity } } + : perform(operation)); + const connect = vi.fn(() => ({ ready: Promise.resolve(), close: async () => {} })); + ingress.connect.mockImplementation(connect); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness, authorityRunId: target.identity.runId, + projectAuthCheckVersion: 1, accountIdentity: "account", adoptIdentity: source.identity }); + const initial = await owner.driver.controlPlaneRegistration!({} as never, source.identity); + await initial.activate!(); await initial.ready!(); + expect(task.mock.calls.some(([operation]) => operation.kind === "read_state" && operation.identity.runId === source.identity.runId)).toBe(true); + const requested = mode === "foreign target" ? { ...target.identity, itemId: "unprepared-item" } : target.identity; + const next = await owner.driver.controlPlaneRegistration!({} as never, requested); + if (mode !== "unprepared") await prepare(owner, target); + nativeIdentity = target.identity; + try { + await next.activate!(); + if (mode === "prepared") { await expect(next.ready!()).resolves.toBeUndefined(); expect(connect).toHaveBeenCalledTimes(2); } + else { await expect(next.ready!()).rejects.toThrow("adoption identity mismatch"); expect(connect).toHaveBeenCalledTimes(1); } + } finally { await next.release!("detach"); await initial.release!(); } + }, +); + +it.each(["codex", "claude"] as const)("rejects a foreign %s source tuple before opening adopted ingress", async harness => { + const { task } = provider(); + const connect = vi.fn(() => ({ ready: Promise.resolve(), close: async () => {} })); + ingress.connect.mockImplementation(connect); + const source = launchRequirements(harness).identity; + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness, adoptIdentity: source }); + const registration = await owner.driver.controlPlaneRegistration!({} as never, { ...source, itemId: "foreign-item" }); + try { + await registration.activate!(); + await expect(registration.ready!()).rejects.toThrow("adoption identity mismatch"); + expect(task).not.toHaveBeenCalled(); expect(connect).not.toHaveBeenCalled(); + } finally { await registration.release!("detach"); } +}); diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index bf99a5ed958..cd020ceb0a1 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -48,6 +48,9 @@ export function createPluginRunnerTaskOwner(input: { pollMs?: number; warmTaskVersion?: 1; releaseResumeVersion?: 1; + releasedSource?: DurableRecoveryIdentity; + onResumed?: (identity: DurableRecoveryIdentity) => Promise; + onRetired?: (identity: DurableRecoveryIdentity) => Promise; githubBroker?: { url: string; token: string }; adoptIdentity?: DurableRecoveryIdentity; }) { @@ -268,7 +271,8 @@ export function createPluginRunnerTaskOwner(input: { }); return { child, completion, restart: null }; }; - const registration = async (authority: DurablePrpControlPlane, runId: string, signal?: AbortSignal) => { + const registration = async (authority: DurablePrpControlPlane, runId: string, signal?: AbortSignal, expectedIdentity?: DurableRecoveryIdentity) => { + const desired = expectedIdentity ? structuredClone(expectedIdentity) : undefined; const generation = ++registrationGeneration; let registrationReleased = false; let outbound: ReturnType | null = null; @@ -284,7 +288,15 @@ export function createPluginRunnerTaskOwner(input: { activate: () => { activation = (async () => { if (input.adoptIdentity) { - await verifyPluginRunnerTaskReattachment(input.task, input.adoptIdentity); + if (!desired || runId === input.adoptIdentity.runId) { + if (desired && Object.entries(input.adoptIdentity).some(([key, value]) => desired[key as keyof DurableRecoveryIdentity] !== value)) { + throw new Error("Plugin Runner adoption identity mismatch"); + } + await verifyPluginRunnerTaskReattachment(input.task, input.adoptIdentity); + } else if (!preparedIdentity || Object.entries(preparedIdentity).some(([key, value]) => desired[key as keyof DurableRecoveryIdentity] !== value) + || ["runnerInstanceId", "environmentLeaseId", "normalizedSessionId"].some(key => desired[key as keyof DurableRecoveryIdentity] !== input.adoptIdentity![key as keyof DurableRecoveryIdentity])) { + throw new Error("Plugin Runner adoption identity mismatch"); + } } else await running; if (registrationReleased || signal?.aborted) throw new Error("Plugin Runner startup cancelled"); const ingress = await endpoint(); @@ -318,6 +330,8 @@ export function createPluginRunnerTaskOwner(input: { if (submitted) { await call({ kind: "complete" }); completed = true; + const identity = preparedIdentity ?? launchIdentity ?? input.priorIdentity; + if (identity) await input.onRetired?.(identity); } }, release: async (reason?: "detach") => { @@ -330,6 +344,25 @@ export function createPluginRunnerTaskOwner(input: { }; }; const driver: RunnerExecutionDriver = { + get releaseResume() { + if (input.releaseResumeVersion !== 1 || input.warmTaskVersion !== 1 || !input.accountIdentity) return undefined; + return { + releasedSource: input.releasedSource, + release: async (identity: DurableRecoveryIdentity) => { + requireReleaseResume(); + const outbound = activeOutbound; + outbound?.quiesce(); + await release(identity); + await outbound?.close(); + if (activeOutbound === outbound) activeOutbound = null; + }, + resume: async (identity: DurableRecoveryIdentity, ticket: string) => { + const outcome = await resume(identity, ticket); + if (outcome === "resumed") await input.onResumed?.(identity); + return outcome; + }, + }; + }, get authorityRunId() { return input.authorityRunId ?? input.executionContext?.execution.binding.runId; }, get codexSourceRecoveryVersion() { return input.codexSourceRecoveryVersion; }, prepareLaunch, authorizeRunnerAdoption, readRunnerState, runnerProcessLauncher: launcher, @@ -341,7 +374,7 @@ export function createPluginRunnerTaskOwner(input: { if (result.kind !== "state") throw new Error("Plugin Runner archived state unavailable"); return result.state; }, - controlPlaneRegistration: (authority, identity, signal) => registration(authority, identity?.runId ?? "", signal), + controlPlaneRegistration: (authority, identity, signal) => registration(authority, identity?.runId ?? "", signal, identity), ...(input.adoptIdentity ? { adoptExistingRunner: { // Provider ownership is attested by Launcher, never by a controller-local PID. pid: null, processGroupId: null, startedAt: null,