diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs index 1c82e7ec66b..ade615f7ffc 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs @@ -828,6 +828,20 @@ fn parse_admission_diagnostic(line: &str) -> Option<(&'static str, u64)> { .split_once(": elapsedMs=")?; let stage = match stage { "binding" => "binding", + "turn_prepare" => "turn_prepare", + "turn_spawn" => "turn_spawn", + "turn_spawned" => "turn_spawned", + "turn_initialize" => "turn_initialize", + "turn_initialized" => "turn_initialized", + "turn_session_load" => "turn_session_load", + "turn_session_loaded" => "turn_session_loaded", + "turn_session_resume" => "turn_session_resume", + "turn_session_resumed" => "turn_session_resumed", + "turn_session_new" => "turn_session_new", + "turn_session_created" => "turn_session_created", + "turn_prompt_submitted" => "turn_prompt_submitted", + "turn_ownership" => "turn_ownership", + "turn_ownership_verified" => "turn_ownership_verified", "installation" => "installation", "sandbox" => "sandbox", "lifetime" => "lifetime", @@ -1112,6 +1126,38 @@ mod tests { .is_err()); } + #[cfg(unix)] + #[test] + fn turn_start_timeout_distinguishes_lazy_initialize_load_and_ownership() { + for stage in ["turn_initialize", "turn_session_load", "turn_ownership"] { + let script = format!( + "printf '%s\\n' '[paperclip-acpx-sidecar] admission_{stage}: elapsedMs=42' '[paperclip-acpx-sidecar] admission_turn_prompt_submitted: elapsedMs=43 /private/token-canary' >&2; sleep 2" + ); + let config = AcpxSidecarTransportConfig { + command: PathBuf::from("/bin/sh"), + args: vec!["-c".into(), script], + verified_launch: None, + request_timeout: Duration::from_millis(200), + shutdown_grace: Duration::from_millis(10), + }; + let mut transport = AcpxSidecarTransport::start(&config).unwrap(); + let error = transport + .request(GeneratedAcpxSidecarCommand::TurnStart, json!({})) + .err() + .unwrap() + .to_string(); + assert!(error.contains("request timed out at turn.start"), "{error}"); + assert!( + error.contains(&format!("admissionStage={stage} admissionElapsedMs=42")), + "{error}" + ); + assert!( + !error.contains("private") && !error.contains("canary"), + "{error}" + ); + } + } + #[cfg(unix)] #[test] fn admission_rejection_retains_sequenced_stage_and_preserves_buffered_frames() { diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts index a5457816bff..588401a613a 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts @@ -1768,6 +1768,59 @@ describe("Codex ACPX runtime adapter", () => { await owner.command.close(); }); + it("reports deferred turn initialization, session load and ownership without acknowledging early", async () => { + const runtime = fakeRuntime(); + const child = fakeChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + let created!: AcpRuntimeOptions; + let submit!: () => void; + let verifyOwner!: () => void; + const promptStarted = new Promise(resolve => { submit = resolve; }); + const owned = new Promise(resolve => { verifyOwner = resolve; }); + const phases: string[] = []; + let observedOwnership!: () => void; + const ownershipEntered = new Promise(resolve => { observedOwnership = resolve; }); + const options = openOptions(command); + options.onTurnStage = stage => { + phases.push(stage); + if (stage === "turn_ownership") observedOwnership(); + }; + vi.mocked(runtime.startTurn).mockImplementation(() => { + created.spawnAgent!({ command: "private-command-canary", args: [], options: {} }); + created.onAgentSpawn!({ pid: 123, startedAt: "private-time-canary" }); + created.onAcpMessage!("outbound", { id: "private-id-canary", method: "initialize", params: { secret: "private-payload-canary" } }); + return { requestId: "resumed", promptStarted, events: { async *[Symbol.asyncIterator]() {} }, + result: new Promise(() => undefined), cancel: vi.fn(), closeStream: vi.fn() }; + }); + const port = await openCodexAcpxRuntime(options, { + createRegistry: () => registry(), createStore: () => store(), + awaitProviderOwnership: () => owned, awaitProviderExit: providerOwnershipEstablished, + createRuntime: value => { created = value; return runtime; }, + }); + const turn = port.startTurn({ text: "Resume", requestId: "resumed" }); + let acknowledged = false; + void turn.promptStarted.then(() => { acknowledged = true; }); + expect(phases.at(-1)).toBe("turn_initialize"); + await Promise.resolve(); + expect(acknowledged).toBe(false); + created.onAcpMessage!("inbound", { id: "private-id-canary", result: {} }); + created.onAcpMessage!("outbound", { id: 2, method: "session/load", params: { sessionId: "private-session-canary" } }); + expect(phases.at(-1)).toBe("turn_session_load"); + await Promise.resolve(); + expect(acknowledged).toBe(false); + created.onAcpMessage!("inbound", { id: 2, result: {} }); + submit(); + await ownershipEntered; + expect(phases.at(-1)).toBe("turn_ownership"); + expect(acknowledged).toBe(false); + verifyOwner(); + await expect(turn.promptStarted).resolves.toBeUndefined(); + expect(phases).toEqual(["turn_prepare", "turn_spawn", "turn_spawned", "turn_initialize", "turn_initialized", + "turn_session_load", "turn_session_loaded", "turn_prompt_submitted", "turn_ownership", "turn_ownership_verified"]); + expect(JSON.stringify(phases)).not.toContain("canary"); + }); + it("admits a verified provider that starts with the first recovered turn", async () => { const runtime = fakeRuntime(); const child = fakeChild(); diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts index 9255b0bc94e..a7ab68ce7dd 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -1,3 +1,4 @@ +import { createTurnAdmissionDiagnostics } from "./turn-admission-diagnostics.js"; import { createPiThinkingAdmission, resolvePiThinkingLevel } from "./pi-thinking.js"; import type { ChildProcess } from "node:child_process"; @@ -190,6 +191,7 @@ export async function openQualifiedAcpxRuntime( const createRegistry = dependencies.createRegistry ?? createAgentRegistry; const createStore = dependencies.createStore ?? createRuntimeStore; const createRuntime = dependencies.createRuntime ?? createAcpRuntime; + const turnDiagnostics = createTurnAdmissionDiagnostics(options.onTurnStage); const runtimeCloseTimeoutMs = dependencies.runtimeCloseTimeoutMs ?? DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS; const retainedCleanupOwners = new WeakSet>(); @@ -426,6 +428,8 @@ export async function openQualifiedAcpxRuntime( : null; }, onSessionNotification: acceptGoalNotification, + onAcpMessage: turnDiagnostics.onAcpMessage, + onAgentSpawn: () => turnDiagnostics.report("turn_spawned"), spawnEnvironment: () => ({ ...definedEnvironment(options.launchEnvironment), ...(options.profile.agent === "claude" @@ -447,6 +451,7 @@ export async function openQualifiedAcpxRuntime( options.signal?.throwIfAborted(); options.assertWorkspaceHeld?.(); commandLaunches.count += 1; + turnDiagnostics.report("turn_spawn"); return children.add( options.command.spawn(input.args, input.options, { credentialFenceFds, @@ -605,6 +610,7 @@ export async function openQualifiedAcpxRuntime( commandLaunches, permissionBoundary, extensionBoundary, + turnDiagnostics, options.profile.agent === "copilot" ? assertCopilotPromptPolicy : undefined, ); } catch (error) { @@ -1035,6 +1041,7 @@ function runtimePort( commandLaunches: { count: number; refreshConsumedCommand?: () => Promise }, permissionBoundary: { active: AbortController | null; hasAdmittedTurn: boolean; handler?: AcpRuntimeOptions["onPermissionRequest"] }, extensionBoundary: AcpxRuntimeExtensionBoundary, + turnDiagnostics: ReturnType, assertPromptPolicy?: (text: string) => void, ): AcpxRuntimePort { extensionBoundary.sessionIds = new Set([identity.backendSessionId]); @@ -1395,6 +1402,7 @@ function runtimePort( startTurn(input) { assertPromptPolicy?.(input.text); if (extensionBoundary.active) throw new Error("ACPX runtime already has an active turn"); + const turnProgress = turnDiagnostics.begin(); const approval = new AbortController(); const controller = new AbortController(); const extensionTurn: AcpxRuntimeExtensionTurn = { @@ -1405,6 +1413,7 @@ function runtimePort( }; extensionBoundary.active = extensionTurn; const releaseExtensionTurn = (): void => { + turnProgress.end(); controller.abort(new Error("ACPX extension turn expired")); if (extensionBoundary.active === extensionTurn) extensionBoundary.active = null; }; @@ -1433,7 +1442,11 @@ function runtimePort( void finishOwnershipAdmission().catch(() => undefined); throw error; } - const guarded = turnWithVerifiedLifetimeOwnership(turn, finishOwnershipAdmission); + const guarded = turnWithVerifiedLifetimeOwnership(turn, async () => { + turnProgress.report("turn_ownership"); + await finishOwnershipAdmission(); + turnProgress.report("turn_ownership_verified"); + }, () => turnProgress.report("turn_prompt_submitted")); extensionTurn.promptStarted = guarded.promptStarted; const result = guarded.result.then( (value) => { @@ -1484,6 +1497,7 @@ function runtimePort( function turnWithVerifiedLifetimeOwnership( turn: AcpxRuntimeTurn, finishOwnershipAdmission: () => Promise, + onPromptSubmitted: () => void, ): AcpxRuntimeTurn { // A persisted ACPX session can be loaded without starting an agent process. // Keep the narrowly scoped turn admission open until either the provider has @@ -1491,7 +1505,7 @@ function turnWithVerifiedLifetimeOwnership( // stable-empty seal in SpawnedChildSet then rejects every later spawn. const reachedAdmissionBoundary = Promise.race([ turn.promptStarted.then( - () => undefined, + () => onPromptSubmitted(), () => undefined, ), turn.result.then( diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts index da484936613..9af98b14c11 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts @@ -207,7 +207,7 @@ describe("ACPX runtime host", () => { const gate = deferred(); const stages: Array<{ stage: string; elapsedMs: number }> = []; const enteredRuntime = deferred(); - const openRuntime = vi.fn(() => { + const openRuntime = vi.fn((_options: AcpxRuntimePortOpenOptions) => { enteredRuntime.resolve(); return gate.promise; }); @@ -239,6 +239,10 @@ describe("ACPX runtime host", () => { expect(elapsedMs).toBeGreaterThanOrEqual(stages[index - 1]?.elapsedMs ?? 0); expect(elapsedMs).toBeLessThanOrEqual(3_600_000); } + expect(() => openRuntime.mock.calls[0]![0].onTurnStage?.("turn_initialize")).not.toThrow(); + await Promise.resolve(); + expect(stages.at(-1)?.stage).toBe("turn_initialize"); + expect(Number.isInteger(stages.at(-1)?.elapsedMs)).toBe(true); expect(host.identity().acpxRecordId).toBe("record-1"); await host.close({ reason: "diagnostic observer test complete" }); expect(port.close).toHaveBeenCalledOnce(); diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts index f2b1ab14156..64db6a4d836 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts @@ -1,3 +1,4 @@ +import type { AcpxTurnAdmissionStage } from "./turn-admission-diagnostics.js"; import { piProviderConfiguration } from "./pi-provider-config.js"; import { withAcpxTurnCancellation } from "./turn-cancellation.js"; import { resolveAcpxProviderMode } from "./provider-mode.js"; @@ -149,6 +150,8 @@ export interface AcpxRuntimePort { } export interface AcpxRuntimePortOpenOptions { + /** Closed active-turn progress, measured from the host admission origin. */ + onTurnStage?: (stage: AcpxTurnAdmissionStage) => void; /** A durable identity was admitted; restored native mode must already match. */ restoringSession?: boolean; /** Ephemeral provider extension metadata; mandatory ACP caps remain host-owned. */ @@ -199,7 +202,7 @@ export interface AcpxRetainedCleanupFailure { error: unknown; } -export type AcpxAdmissionStage = "binding" | "installation" | "sandbox" | "lifetime" +export type AcpxAdmissionStage = AcpxTurnAdmissionStage | "binding" | "installation" | "sandbox" | "lifetime" | "agent_files" | "skills" | "command" | "tool_bridge" | "handshake" | "verification" | "ready" | "cleanup"; export interface AcpxRuntimeHostDependencies { @@ -590,6 +593,7 @@ export class AcpxRuntimeHost { agentFiles?.assertHeld(); await assertAcpxProfileWorkspace(options.agent, binding.workspacePath); return dependencies.openRuntime({ + onTurnStage: reportStage, ...(options.clientCapabilities === undefined ? {} : { clientCapabilities: structuredClone(options.clientCapabilities), }), diff --git a/packages/paperclip-runner/src/drivers/acpx/turn-admission-diagnostics.test.ts b/packages/paperclip-runner/src/drivers/acpx/turn-admission-diagnostics.test.ts new file mode 100644 index 00000000000..7cde70399f1 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/turn-admission-diagnostics.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from "vitest"; +import { createTurnAdmissionDiagnostics, type AcpxTurnAdmissionStage } from "./turn-admission-diagnostics.js"; + +describe("closed turn admission diagnostics", () => { + it.each([ + ["initialize", "turn_initialize", "turn_initialized"], + ["session/load", "turn_session_load", "turn_session_loaded"], + ["session/resume", "turn_session_resume", "turn_session_resumed"], + ["session/new", "turn_session_new", "turn_session_created"], + ] as const)("retains pending %s without exposing wire data", (method, sent, completed) => { + const stages: AcpxTurnAdmissionStage[] = []; + const diagnostics = createTurnAdmissionDiagnostics(stage => stages.push(stage)); + diagnostics.onAcpMessage("outbound", { id: 1, method }); + expect(stages).toEqual([]); + diagnostics.begin(); + diagnostics.onAcpMessage("outbound", { id: "private-id-canary", method, params: { path: "/secret-canary", token: "token-canary" } }); + expect(stages.at(-1)).toBe(sent); + diagnostics.onAcpMessage("inbound", { id: "unrelated", result: {} }); + expect(stages.at(-1)).toBe(sent); + diagnostics.onAcpMessage("inbound", { id: "private-id-canary", result: { content: "response-canary" } }); + expect(stages.at(-1)).toBe(completed); + expect(JSON.stringify(stages)).not.toMatch(/canary/); + }); + + it("ignores rejected, unknown, stale and oversized requests", () => { + const stages: AcpxTurnAdmissionStage[] = []; + const diagnostics = createTurnAdmissionDiagnostics(stage => stages.push(stage)); + const old = diagnostics.begin(); + diagnostics.onAcpMessage("outbound", { id: 1, method: "session/load" }); + diagnostics.onAcpMessage("inbound", { id: 1, error: { message: "private-canary" } }); + expect(stages.at(-1)).toBe("turn_session_load"); + diagnostics.onAcpMessage("outbound", { id: 2, method: "initialize" }); + diagnostics.begin(); + const count = stages.length; + old.report("turn_ownership_verified"); + old.end(); + diagnostics.onAcpMessage("inbound", { id: 2, result: {} }); + diagnostics.onAcpMessage("outbound", { id: 3, method: "constructor" }); + diagnostics.onAcpMessage("outbound", { id: "x".repeat(129), method: "initialize" }); + expect(stages).toHaveLength(count); + diagnostics.report("turn_spawn"); + expect(stages.at(-1)).toBe("turn_spawn"); + }); + + it.each([false, true])("isolates throwing/rejecting observers (async=%s)", async (async) => { + const diagnostics = createTurnAdmissionDiagnostics(() => { + if (async) return Promise.reject(new Error("observer-canary")); + throw new Error("observer-canary"); + }); + expect(() => { + const turn = diagnostics.begin(); + diagnostics.onAcpMessage("outbound", { id: 1, method: "initialize" }); + diagnostics.onAcpMessage("inbound", { id: 1, result: {} }); + turn.report("turn_prompt_submitted"); + turn.end(); + }).not.toThrow(); + await Promise.resolve(); + }); +}); diff --git a/packages/paperclip-runner/src/drivers/acpx/turn-admission-diagnostics.ts b/packages/paperclip-runner/src/drivers/acpx/turn-admission-diagnostics.ts new file mode 100644 index 00000000000..337113ccb05 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/turn-admission-diagnostics.ts @@ -0,0 +1,70 @@ +/** Closed progress only. Wire IDs are correlated in memory and never reported. */ +export type AcpxTurnAdmissionStage = + | "turn_prepare" | "turn_spawn" | "turn_spawned" + | "turn_initialize" | "turn_initialized" + | "turn_session_load" | "turn_session_loaded" + | "turn_session_resume" | "turn_session_resumed" + | "turn_session_new" | "turn_session_created" + | "turn_prompt_submitted" | "turn_ownership" | "turn_ownership_verified"; + +type WireStage = readonly [AcpxTurnAdmissionStage, AcpxTurnAdmissionStage]; +const WIRE_STAGES: Readonly> = { + initialize: ["turn_initialize", "turn_initialized"], + "session/load": ["turn_session_load", "turn_session_loaded"], + "session/resume": ["turn_session_resume", "turn_session_resumed"], + "session/new": ["turn_session_new", "turn_session_created"], +}; + +export function createTurnAdmissionDiagnostics( + observe?: (stage: AcpxTurnAdmissionStage) => void, +) { + let active: symbol | null = null; + const pending = new Map(); + const report = (stage: AcpxTurnAdmissionStage): void => { + if (!active) return; + try { + void Promise.resolve(observe?.(stage)).catch(() => undefined); + } catch { /* Progress observers cannot change provider authority. */ } + }; + return { + begin() { + const token = Symbol(); + active = token; + pending.clear(); + report("turn_prepare"); + return { + report(stage: AcpxTurnAdmissionStage) { + if (active === token) report(stage); + }, + end() { + if (active !== token) return; + active = null; + pending.clear(); + }, + }; + }, + report, + onAcpMessage(direction: "inbound" | "outbound", value: unknown) { + if (!active || !value || typeof value !== "object" || Array.isArray(value)) return; + const message = value as Record; + const id = message.id; + if (!(typeof id === "string" && id.length <= 128) + && !(typeof id === "number" && Number.isSafeInteger(id))) return; + if (direction === "outbound") { + const method = message.method; + const stages = typeof method === "string" && Object.hasOwn(WIRE_STAGES, method) + ? WIRE_STAGES[method] : undefined; + if (!stages || pending.size >= 16) return; + pending.set(id, stages); + report(stages[0]); + } else { + const stages = pending.get(id); + if (!stages || Object.hasOwn(message, "method")) return; + pending.delete(id); + // This marks a successful wire response, not later auth/replay/checkpoint work. + // A rejected response is not evidence that initialize/load completed. + if (Object.hasOwn(message, "result") && !Object.hasOwn(message, "error")) report(stages[1]); + } + }, + }; +}