Diagnose deferred ACPX turn admission with closed phases

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() {

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<void>(resolve => { submit = resolve; });
+    const owned = new Promise<void>(resolve => { verifyOwner = resolve; });
+    const phases: string[] = [];
+    let observedOwnership!: () => void;
+    const ownershipEntered = new Promise<void>(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<never>(() => 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();

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<Promise<void>>();
@@ -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<void> },
   permissionBoundary: { active: AbortController | null; hasAdmittedTurn: boolean; handler?: AcpRuntimeOptions["onPermissionRequest"] },
   extensionBoundary: AcpxRuntimeExtensionBoundary,
+  turnDiagnostics: ReturnType<typeof createTurnAdmissionDiagnostics>,
   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<void>,
+  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(

packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts

@@ -207,7 +207,7 @@ describe("ACPX runtime host", () => {
     const gate = deferred<AcpxRuntimePort>();
     const stages: Array<{ stage: string; elapsedMs: number }> = [];
     const enteredRuntime = deferred<void>();
-    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();

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),
             }),

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();
+  });
+});

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<Record<string, WireStage>> = {
+  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<string | number, WireStage>();
+  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<string, unknown>;
+      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]);
+      }
+    },
+  };
+}