fix(runner): settle ACPX cancellation after verified retirement

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-19 10:36:26 +02:00 · 2026-09-29 21:40:17 -05:00
1 parent 951b06d08a
commit 285073afe0
@@ -207,7 +207,7 @@ impl AcpxProviderDescriptor {
                "0.0.33",
                Some("@earendil-works/pi-coding-agent"),
                Some("0.84.2"),
-                "sha256:843d30e419914529755c9827d9151a306da1b50b643be7eab1abe797641a37ce",
+                "sha256:edf058835ee84de3869c4a8e8bdb71a934ffdb9f34daa37ae6faeb92371d1cdf",
            ),
            "cursor" => (
                self.model.as_str(),
@@ -215,7 +215,7 @@ impl AcpxProviderDescriptor {
                "2026.09.26-dd393fe",
                None,
                None,
-                "sha256:aa8c0b2b84786982bcd06b7634bf95be6f2bc42bb8def48a8751dc50cf739b6b",
+                "sha256:377dcea64a727ce799cc112458d4b40ba4bc6574cd6c6f7233b6efd5917a6c4b",
            ),
            "copilot" => (
                self.model.as_str(),
@@ -223,7 +223,7 @@ impl AcpxProviderDescriptor {
                "1.0.88",
                None,
                None,
-                "sha256:ece77e40876631a69a828b91813722001d71b6fc81be47ecd4b3dced84ff9473",
+                "sha256:0fe49c2f8a2b144344d0de745fed1e4568df305de3a26c3a121dec21c8d4b751",
            ),
            "grok" => (
                "grok-4.7",
@@ -1906,7 +1906,16 @@ impl AcpxCommandExecutor {
                }
                if let Some(event_type) = terminal {
                    state.active_turn_id = None;
-                    state.lifecycle = "session_open".to_owned();
+                    state.lifecycle = if self
+                        .session
+                        .as_ref()
+                        .is_some_and(|session| session.runtime_retired())
+                    {
+                        "closed"
+                    } else {
+                        "session_open"
+                    }
+                    .to_owned();
                    provider_turn_settled = true;
                    // An ACP goal has session lifetime, not prompt lifetime.
                    // Out-of-prompt goal updates remain observable after quiescence.
@@ -2794,7 +2803,7 @@ mod tests {
                "cursor",
                "cursor-agent",
                "2026.09.26-dd393fe",
-                "sha256:aa8c0b2b84786982bcd06b7634bf95be6f2bc42bb8def48a8751dc50cf739b6b",
+                "sha256:377dcea64a727ce799cc112458d4b40ba4bc6574cd6c6f7233b6efd5917a6c4b",
                None,
                None,
                "explicit-model",
@@ -2803,7 +2812,7 @@ mod tests {
                "copilot",
                "@github/copilot",
                "1.0.88",
-                "sha256:ece77e40876631a69a828b91813722001d71b6fc81be47ecd4b3dced84ff9473",
+                "sha256:0fe49c2f8a2b144344d0de745fed1e4568df305de3a26c3a121dec21c8d4b751",
                None,
                None,
                "explicit-model",
@@ -2812,7 +2821,7 @@ mod tests {
                "pi",
                "pi-acp",
                "0.0.33",
-                "sha256:843d30e419914529755c9827d9151a306da1b50b643be7eab1abe797641a37ce",
+                "sha256:edf058835ee84de3869c4a8e8bdb71a934ffdb9f34daa37ae6faeb92371d1cdf",
                Some("@earendil-works/pi-coding-agent"),
                Some("0.84.2"),
                "openrouter/deepseek/deepseek-v4-flash-0731",
@@ -2834,6 +2843,18 @@ mod tests {
            }
            let valid: AcpxProviderDescriptor = serde_json::from_value(value.clone()).unwrap();
            valid.validate(&context()).unwrap();
+            let mut previous_contract = value.clone();
+            previous_contract["commandDigest"] = json!(match agent {
+                "cursor" =>
+                    "sha256:aa8c0b2b84786982bcd06b7634bf95be6f2bc42bb8def48a8751dc50cf739b6b",
+                "copilot" =>
+                    "sha256:ece77e40876631a69a828b91813722001d71b6fc81be47ecd4b3dced84ff9473",
+                "pi" => "sha256:843d30e419914529755c9827d9151a306da1b50b643be7eab1abe797641a37ce",
+                _ => unreachable!(),
+            });
+            let previous_contract: AcpxProviderDescriptor =
+                serde_json::from_value(previous_contract).unwrap();
+            assert!(previous_contract.validate(&context()).is_err());
            if agent == "cursor" {
                assert_eq!(valid.public_descriptor(None)["cursorMode"], json!("agent"));
                let mut previous_v4 = value.clone();
@@ -10,7 +10,8 @@ use crate::acpx_provider_state::{
 };
 use crate::acpx_sidecar_transport::{AcpxSidecarTransport, AcpxSidecarTransportConfig};
 use crate::generated_acpx_sidecar_contract::{
-    GeneratedAcpxSidecarCommand, GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
+    GeneratedAcpxSidecarCommand, GeneratedAcpxSidecarEventType,
+    GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
 };
 use crate::local_runner::LocalRunnerError;
 use crate::provider_bridge::{
@@ -278,6 +279,7 @@ pub struct AcpxProviderSession {
    working_directory: PathBuf,
    closed: bool,
    transport_terminated: bool,
+    runtime_retired: bool,
 }

 impl AcpxProviderSession {
@@ -312,9 +314,14 @@ impl AcpxProviderSession {
            working_directory: config.working_directory.clone(),
            closed: false,
            transport_terminated: false,
+            runtime_retired: false,
        })
    }

+    pub fn runtime_retired(&self) -> bool {
+        self.runtime_retired
+    }
+
    pub fn process_id(&self) -> u32 {
        self.transport.process_id()
    }
@@ -360,6 +367,11 @@ impl AcpxProviderSession {
        working_directory: &Path,
    ) -> Result<Value, LocalRunnerError> {
        self.ensure_open()?;
+        if self.runtime_retired {
+            return Err(LocalRunnerError::invalid(
+                "ACPX provider runtime was retired by cancellation",
+            ));
+        }
        validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?;
        validate_turn_message(message)?;
        if working_directory != self.working_directory {
@@ -526,6 +538,9 @@ impl AcpxProviderSession {
                "ACPX sidecar did not confirm turn cancellation",
            )));
        }
+        // Polling still owns the terminal frame queued before this response.
+        // Only future prompt admission is revoked by the confirmed close.
+        self.runtime_retired = response.get("sessionClosed").and_then(Value::as_bool) == Some(true);
        Ok(response)
    }

@@ -538,9 +553,27 @@ impl AcpxProviderSession {
            Ok(event) => event,
            Err(error) => return Err(self.fail_closed(error)),
        };
-        let Some(event) = event else {
+        let Some(mut event) = event else {
            return Ok(None);
        };
+        if event.event_type == GeneratedAcpxSidecarEventType::RuntimePermissionRequested {
+            // Authority comes from this admitted connection's profile. A sidecar
+            // claim cannot relabel another provider; old frames may omit origin.
+            let origin = json!({"adapter":"acpx-runtime-sidecar", "provider":self.config.agent,
+                "method":"session/request_permission"});
+            if event
+                .payload
+                .get("origin")
+                .is_some_and(|claimed| claimed != &origin)
+            {
+                return Err(self.fail_closed(LocalRunnerError::invalid(
+                    "ACPX permission origin conflicts with the admitted provider profile",
+                )));
+            }
+            if let Some(payload) = event.payload.as_object_mut() {
+                payload.insert("origin".to_owned(), origin);
+            }
+        }
        let mut next_state = self.state.clone();
        let events = match next_state.accept_event(&event) {
            Ok(events) => events,
@@ -232,6 +232,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
            | "turns-mismatched-reserved-result-terminal"
            | "turns-unauthorized-tool"
            | "turns-permission"
+            | "permissions-forged-origin"
+            | "turns-retired"
+            | "turns-retired-terminal-first"
            | "permissions-interactive"
            | "permissions-wrong-ack"
            | "resolutions"
@@ -242,6 +245,21 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
            | "suspend-wrong-ack"
            | "suspend-wrong-identity"
            | "suspend-missing-identity" => {
+                if command == "turn.cancel" && mode == "turns-retired-terminal-first" {
+                    let turn_id = request
+                        .pointer("/params/turnId")
+                        .and_then(Value::as_str)
+                        .unwrap();
+                    write_turn_event(
+                        &mut stdout,
+                        next_sequence,
+                        "runtime.turn_terminal",
+                        "run-1",
+                        turn_id,
+                        json!({"status":"interrupted"}),
+                    )?;
+                    next_sequence += 1;
+                }
                write_json(
                    &mut stdout,
                    &bootstrap_success(id, command, &request, mode, profile_digest),
@@ -398,7 +416,10 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                if command == "turn.start"
                    && matches!(
                        mode,
-                        "turns-permission" | "permissions-interactive" | "permissions-wrong-ack"
+                        "turns-permission"
+                            | "permissions-interactive"
+                            | "permissions-wrong-ack"
+                            | "permissions-forged-origin"
                    )
                {
                    write_turn_event(
@@ -407,12 +428,20 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                        "runtime.permission_requested",
                        "run-1",
                        turn_id,
-                        json!({
-                            "requestId":"permission-1",
-                            "kind":"execute",
-                            "title":"Run a command?",
-                            "choices":[{"key":"accept","label":"Allow once"},{"key":"cancel","label":"Cancel"}],
-                        }),
+                        if mode == "permissions-forged-origin" {
+                            json!({
+                                "requestId":"permission-1", "kind":"execute", "title":"Run a command?",
+                                "choices":[{"key":"cancel","label":"Cancel"}],
+                                "origin":{"adapter":"acpx-runtime-sidecar","provider":"cursor","method":"session/request_permission"},
+                            })
+                        } else {
+                            json!({
+                                "requestId":"permission-1",
+                                "kind":"execute",
+                                "title":"Run a command?",
+                                "choices":[{"key":"accept","label":"Allow once"},{"key":"cancel","label":"Cancel"}],
+                            })
+                        },
                    )?;
                    next_sequence += 1;
                }
@@ -654,7 +683,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                        next_sequence += 1;
                    }
                }
-                if command == "turn.cancel" && mode == "turns" {
+                if command == "turn.cancel" && matches!(mode, "turns" | "turns-retired") {
                    write_turn_event(
                        &mut stdout,
                        next_sequence,
@@ -731,6 +760,7 @@ fn bootstrap_success(
                    "requestedModel": model,
                    "effectiveModel": if mode == "bootstrap-wrong-model" { "wrong-model" } else { model },
                    "permissionMode": params.get("permissionMode"),
+                    "cursorMode": params.get("cursorMode"),
                    "providerLifetimeFenceCandidates": [60001, 60002, 60003],
                },
                "status": {},
@@ -749,7 +779,9 @@ fn bootstrap_success(
            "accepted": true, "turnId": params.get("turnId"), "controlId": params.get("controlId"),
            "mode": if mode == "controls-wrong-ack" { json!("wrong") } else { params["mode"].clone() },
        }),
-        "turn.cancel" => json!({"cancelled":mode != "turns-wrong-cancel"}),
+        "turn.cancel" => {
+            json!({"cancelled":mode != "turns-wrong-cancel", "sessionClosed":matches!(mode, "turns-retired" | "turns-retired-terminal-first")})
+        }
        "session.suspend" => json!({
            "suspended":mode != "suspend-wrong-ack",
            "identity": if mode == "suspend-missing-identity" { Value::Null } else { json!({
@@ -224,6 +224,12 @@ pub fn project_acpx_state_event(
                    ));
                }
            }
+            let origin = match details.get("origin") {
+                Some(origin) => project_runtime_request_origin(Some(origin))?,
+                None => {
+                    json!({"adapter":"acpx-runtime-sidecar", "provider":"acpx", "method":"session/request_permission"})
+                }
+            };
            one(
                "runtime_request.created",
                EventPriority::P0,
@@ -232,7 +238,7 @@ pub fn project_acpx_state_event(
                    "requestId":request_id, "turnId":context.turn_id, "itemId":context.item_id,
                    "type":"permission", "status":"pending", "prompt":title, "choices":choices,
                    "details":details,
-                    "origin":{"adapter":"acpx-runtime-sidecar","provider":"acpx","method":"session/request_permission"},
+                    "origin":origin,
                }}),
            )
        }
@@ -301,3 +301,66 @@ fn interactive_permission_requires_an_offered_action_and_acknowledgement() {
        session.shutdown("verified").unwrap();
    }
 }
+
+#[test]
+fn permission_origin_is_bound_to_the_admitted_connection_and_survives_projection() {
+    for agent in ["claude", "copilot", "cursor", "pi"] {
+        let mut cfg = config("permissions-interactive");
+        cfg.agent = agent.to_owned();
+        cfg.model = if agent == "claude" {
+            "claude-sonnet-5"
+        } else if agent == "pi" {
+            "openrouter/deepseek/deepseek-v4-flash-0731"
+        } else {
+            "explicit-test-model"
+        }
+        .to_owned();
+        if agent == "cursor" {
+            cfg.cursor_mode = Some(paperclip_runner_core::acpx_provider_session::CursorMode::Agent);
+        }
+        cfg.provider_policy = Some(
+            paperclip_runner_core::acpx_provider_session::AcpxProviderRuntimePolicy {
+                read_only: false,
+            },
+        );
+        let mut session = AcpxProviderSession::start(&cfg).unwrap();
+        session
+            .start_turn("turn-1", "Request permission", &std::env::temp_dir())
+            .unwrap();
+        let events = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
+        let expected = json!({"adapter":"acpx-runtime-sidecar","provider":agent,"method":"session/request_permission"});
+        let projected = project_acpx_state_event(
+            &AcpxEventProjectionContext {
+                run_id: "run-1".to_owned(),
+                normalized_session_id: "session-1".to_owned(),
+                turn_id: "turn-1".to_owned(),
+                provider_turn_id: None,
+                item_id: "permission-1".to_owned(),
+            },
+            &events[0],
+        )
+        .unwrap();
+        assert_eq!(projected[0].payload["request"]["origin"], expected);
+        assert_eq!(projected[0].payload["request"]["turnId"], "turn-1");
+        assert_eq!(projected[0].payload["request"]["requestId"], "permission-1");
+        session.shutdown("provenance checked").unwrap();
+    }
+}
+
+#[test]
+fn rejects_a_permission_origin_claim_from_another_provider() {
+    let mut cfg = config("permissions-forged-origin");
+    cfg.agent = "claude".to_owned();
+    cfg.model = "claude-sonnet-5".to_owned();
+    let mut session = AcpxProviderSession::start(&cfg).unwrap();
+    session
+        .start_turn("turn-1", "Request permission", &std::env::temp_dir())
+        .unwrap();
+    assert!(session
+        .poll_event(Duration::from_secs(1))
+        .unwrap_err()
+        .to_string()
+        .contains("origin conflicts"));
+    assert!(session.state().pending_permission("permission-1").is_none());
+    session.shutdown("rejected provenance").unwrap();
+}
@@ -779,3 +779,28 @@ fn lazy_warm_handshake_updates_live_turn_control_discovery() {
        .unwrap();
    session.shutdown("verified live handshake").unwrap();
 }
+
+#[test]
+fn forced_cancellation_retirement_preserves_terminal_polling_but_rejects_live_reuse() {
+    for mode in ["turns-retired", "turns-retired-terminal-first"] {
+        let mut session = AcpxProviderSession::start(&config(mode)).unwrap();
+        session
+            .start_turn("turn-1", "Please help", &std::env::temp_dir())
+            .unwrap();
+        session
+            .interrupt_turn("turn-1", "Stop stuck provider")
+            .unwrap();
+        assert!(session.runtime_retired());
+        let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
+        assert!(
+            matches!(terminal.last().unwrap(), AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1")
+        );
+        assert!(session.state().active_turn_id().is_none());
+        assert!(session
+            .start_turn("turn-2", "Must not reuse", &std::env::temp_dir())
+            .unwrap_err()
+            .to_string()
+            .contains("runtime was retired"));
+        session.shutdown("test complete").unwrap();
+    }
+}
@@ -988,6 +988,58 @@ describe("HarnessDriverBackend", () => {
    expect(start).not.toHaveBeenCalled();
  });

+  it("preserves a failed provider terminal after Stop instead of waiting for timeout", async () => {
+    class FailedAfterStop extends FakeHarnessSession {
+      override async *events() {
+        yield prpEvent(1, "item.delta", { text: "before stop" });
+        yield prpEvent(2, "item.delta", { text: "suppressed after stop" });
+        yield prpEvent(3, "runtime_request.cancelled", { requestId: "permission-1", turnId: "turn-1" });
+        yield prpEvent(4, "turn.failed", { status: "failed" });
+      }
+      async interrupt() {}
+    }
+    const backend = new HarnessDriverBackend({ ...driver, async openSession() { return new FailedAfterStop(); } });
+    const session = await backend.openSession({
+      identity: { runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1" },
+      workingDirectory: "/workspace",
+    });
+    const iterator = session.events()[Symbol.asyncIterator]();
+    await iterator.next();
+    await session.cancel({ reason: "operator stop", signal: new AbortController().signal }).cleanup;
+    await expect(iterator.next()).resolves.toMatchObject({ value: { eventType: "runtime_request.cancelled", sourceSeq: 3 } });
+    await expect(iterator.next()).resolves.toMatchObject({ value: { eventType: "turn.failed", sourceSeq: 4 } });
+    await expect(iterator.next()).resolves.toMatchObject({ done: true });
+  });
+
+  it("settles Stop after provider completion but before its queued terminal is consumed", async () => {
+    class CompletedBeforeStop extends FakeHarnessSession {
+      override async *events() {
+        yield prpEvent(1, "item.delta", { text: "before stop" });
+        yield prpEvent(2, "turn.completed", { status: "completed" });
+        yield prpEvent(3, "run.terminal", {
+          schema: "paperclip.prp.terminal.v1", turnTerminalState: "completed",
+          runTerminalState: "succeeded", reportedWorkDisposition: "done",
+        });
+      }
+      async interrupt() { throw new Error("no active provider turn"); }
+    }
+    const backend = new HarnessDriverBackend({ ...driver, async openSession() { return new CompletedBeforeStop(); } });
+    const session = await backend.openSession({
+      identity: { runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1" },
+      workingDirectory: "/workspace",
+    });
+    const iterator = session.events()[Symbol.asyncIterator]();
+    await iterator.next();
+    await expect(session.cancel({ reason: "operator stop", signal: new AbortController().signal }).cleanup).rejects.toThrow("no active provider turn");
+    await expect(iterator.next()).resolves.toMatchObject({ value: { eventType: "turn.cancelled", sourceSeq: 2,
+      payload: { status: "cancelled", providerTerminalState: "completed", reason: "cancelled_after_provider_completed" } } });
+    await expect(iterator.next()).resolves.toMatchObject({ value: { eventType: "run.terminal", sourceSeq: 3,
+      payload: { turnTerminalState: "cancelled", runTerminalState: "cancelled", reportedWorkDisposition: "yielded" } } });
+    await expect(iterator.next()).resolves.toMatchObject({ done: true });
+    await expect(session.result()).resolves.toBeNull();
+    await expect(session.snapshot()).resolves.toMatchObject({ terminal: { runTerminalState: "cancelled" } });
+  });
+
  it("does not synthesize a fallback after explicit run cancellation", async () => {
    class CancelledProviderSession extends FakeHarnessSession {
      override async *events() {
@@ -511,10 +511,42 @@ class HarnessNativeSession implements NativeSession {
    let streamFailure: unknown = null;
    const observedPendingInputs = new Map<string, Record<string, unknown>>();
    try {
-      for await (const event of this.#session.events()) {
+      for await (const providerEvent of this.#session.events()) {
+        // Stop may win publication after the provider result already settled.
+        // Preserve that terminal proof, with the operator's cancelled disposition.
+        let event = providerEvent;
+        if (this.#explicitlyCancelled && providerEvent.eventType === "turn.completed") {
+          event = {
+            ...providerEvent,
+            eventType: "turn.cancelled",
+            payload: {
+              ...providerEvent.payload,
+              status: "cancelled",
+              providerTerminalState: "completed",
+              reason: "cancelled_after_provider_completed",
+            },
+          };
+        } else if (this.#explicitlyCancelled && providerEvent.eventType === "run.terminal"
+          && providerEvent.payload.runTerminalState === "succeeded") {
+          event = {
+            ...providerEvent,
+            payload: {
+              ...providerEvent.payload,
+              turnTerminalState: "cancelled",
+              runTerminalState: "cancelled",
+              reportedWorkDisposition: "yielded",
+            },
+          };
+        }
        const isCancellationEvent =
          event.eventType === "turn.cancelled" ||
          event.eventType === "turn.interrupted" ||
+          // Failure is an authoritative terminal fact, not accepted output.
+          // Dropping it after Stop leaves the controller waiting until timeout.
+          event.eventType === "turn.failed" ||
+          event.eventType === "runtime_request.cancelled" ||
+          event.eventType === "runtime_request.expired" ||
+          (event.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(event.payload.runTerminalState))) ||
          (event.eventType === "item.completed" &&
            event.payload.kind === "interrupt_acknowledgement");
        if (this.#explicitlyCancelled && !isCancellationEvent) continue;
@@ -94,23 +94,47 @@ describe("qualified ACPX runtime sidecar", () => {
    const end = source.indexOf("\nasync function waitForInput", start);
    expect(start).toBeGreaterThan(0);
    const permissions = new Map<string, unknown>();
-    const emitted: Array<{ choices: Array<{ key: string }> }> = [];
+    const emitted: Array<{ choices: Array<{ key: string }>; origin: unknown }> = [];
    const wait = new Function("permissions", "openParams", "normalizeAcpxPermission", "emit",
      `let turnId = "turn-1", requestSequence = 0; const MAX_PENDING_INPUTS = 512;
       const stableRequestId = () => "request-1"; const requireAcpxResponseDelivery = c => c.responseDelivery;
-       return async function(activeTurnId, request, context, toolEvidence) { ${source.slice(start, end)}`)(
-      permissions, { agent }, normalizeAcpxPermission, (_event: string, payload: { choices: Array<{ key: string }> }) => emitted.push(payload),
+       return async function(activeTurnId, agent, request, context, toolEvidence) { ${source.slice(start, end)}`)(
+      permissions, { agent }, normalizeAcpxPermission, (_event: string, payload: { choices: Array<{ key: string }>; origin: unknown }) => emitted.push(payload),
    );
    const abort = new AbortController();
-    const pending = wait("turn-1", { sessionId: "session", inferredKind: "edit", raw: {
+    const pending = wait("turn-1", agent, { sessionId: "session", inferredKind: "edit", raw: {
      sessionId: "session", toolCall: { toolCallId: "call", title: "Edit file" },
      options: ["allow_once", "allow_always", "reject_once"].map(kind => ({ kind, optionId: kind, name: kind })),
    } }, { signal: abort.signal, responseDelivery: Promise.resolve() });
+    expect(emitted[0]!.origin).toEqual({ adapter: "acpx-runtime-sidecar", provider: agent, method: "session/request_permission" });
    expect(emitted[0]!.choices.some(choice => choice.key === "accept_for_session")).toBe(agent === "pi" || agent === "copilot");
    abort.abort();
    await expect(pending).resolves.toEqual({ outcome: "cancel" });
    expect(permissions.size).toBe(0);
  });
+  it("returns the sidecar retirement flag only after cancellation settlement and rejects stale turns", async () => {
+    const source = readFileSync(new URL("./acpx-runtime-sidecar.ts", import.meta.url), "utf8");
+    const start = source.indexOf('    const expected = boundedIdentity', source.indexOf('if (request.command === "turn.cancel")'));
+    const end = source.indexOf('\n  if (request.command === "permission.resolve")', start);
+    let settle!: () => void;
+    const cleanup = new Promise<void>(resolve => { settle = resolve; });
+    let closed = false;
+    const host = { interruptActiveTurn: vi.fn(async () => { await cleanup; closed = true; }), isClosed: () => closed };
+    const cancel = new Function("requireHost", `
+      const turnId = "active-turn";
+      const boundedIdentity = x => x, boundedOptionalText = (x, fallback) => x ?? fallback;
+      return async function(request) { ${source.slice(start, end)}
+    `)(() => host);
+    await expect(cancel({ params: { turnId: "stale" } })).rejects.toThrow("stale");
+    expect(host.interruptActiveTurn).not.toHaveBeenCalled();
+    const pending = cancel({ params: { turnId: "active-turn", reason: "Stop" } });
+    let acknowledged = false; void pending.then(() => { acknowledged = true; });
+    await Promise.resolve(); expect(acknowledged).toBe(false);
+    settle(); await expect(pending).resolves.toEqual({ cancelled: true, sessionClosed: true });
+    host.interruptActiveTurn.mockRejectedValueOnce(new Error("provider cleanup incomplete"));
+    await expect(cancel({ params: { turnId: "active-turn" } })).rejects.toThrow("cleanup incomplete");
+  });
+
  it("emits permission delivery evidence only after the actual response write settles", async () => {
    const source = readFileSync(new URL("./acpx-runtime-sidecar.ts", import.meta.url), "utf8");
    const start = source.indexOf('    const requestId = boundedIdentity', source.indexOf('if (request.command === "permission.resolve")'));
@@ -359,6 +359,7 @@ async function dispatch(
    if (!runId) throw new Error("attach a run before starting an ACPX turn");
    if (turnId) throw new Error("ACPX sidecar already has an active turn");
    const currentTurnId = boundedIdentity(request.params.turnId, "turnId");
+    const activeAgent = openParams!.agent;
    turnId = currentTurnId;
    turnControls.begin(currentTurnId);
    let runtimeTurn: AcpxRuntimeTurn;
@@ -391,7 +392,7 @@ async function dispatch(
        onElicitation: (providerRequest, context) =>
          waitForInput(currentTurnId, providerRequest, context),
        onPermissionRequest: (providerRequest, context) =>
-          waitForPermission(currentTurnId, providerRequest, context, toolEvidence),
+          waitForPermission(currentTurnId, activeAgent, providerRequest, context, toolEvidence),
      });
    } catch (error) {
      turnId = null;
@@ -421,14 +422,15 @@ async function dispatch(
  if (request.command === "turn.cancel") {
    const expected = boundedIdentity(request.params.turnId, "turnId");
    if (expected !== turnId) throw new Error("cannot cancel a stale ACPX turn");
-    await requireHost().interruptActiveTurn(
+    const activeHost = requireHost();
+    await activeHost.interruptActiveTurn(
      boundedOptionalText(
        request.params.reason,
        "Paperclip cancellation",
        4_000,
      ),
    );
-    return { cancelled: true };
+    return { cancelled: true, sessionClosed: activeHost.isClosed() };
  }
  if (request.command === "permission.resolve") {
    const requestId = boundedIdentity(request.params.requestId, "requestId");
@@ -758,6 +760,7 @@ async function waitForTool(call: RunnerToolCall): Promise<unknown> {

 async function waitForPermission(
  activeTurnId: string,
+  agent: QualifiedAcpxAgent,
  request: AcpPermissionRequest,
  context: { signal: AbortSignal; responseDelivery?: Promise<void> },
  toolEvidence?: CopilotToolEvidence | CursorToolEvidence,
@@ -766,7 +769,7 @@ async function waitForPermission(
  if (turnId !== activeTurnId || signal.aborted || permissions.size >= MAX_PENDING_INPUTS) {
    return { outcome: "cancel" };
  }
-  const normalized = normalizeAcpxPermission(request, ["pi", "copilot"].includes(openParams?.agent ?? "") ? { allowAlwaysScope: "session" } : {});
+  const normalized = normalizeAcpxPermission(request, ["pi", "copilot"].includes(agent) ? { allowAlwaysScope: "session" } : {});
  const responseDelivery = requireAcpxResponseDelivery(context);
  const requestId = stableRequestId(activeTurnId, ++requestSequence, normalized.toolCallId);
  return await new Promise((settle) => {
@@ -785,6 +788,7 @@ async function waitForPermission(
    emit("runtime.permission_requested", {
      requestId, kind: normalized.kind, title: normalized.title,
      toolCallId: normalized.toolCallId, choices: normalized.choices,
+      origin: { adapter: "acpx-runtime-sidecar", provider: agent, method: "session/request_permission" },
    }, activeTurnId);
  });
 }
@@ -71,6 +71,44 @@ describe("Codex ACPX runtime adapter", () => {
    await expect(created.onPermissionRequest!({ sessionId: "forged", raw: {}, inferredKind: "edit" } as never, { signal })).resolves.toEqual({ outcome: "reject_once" });
    pending.settle(); await turn.result; await port.close({ reason: "session checks complete" });
  });
+  it("cancels an outstanding native permission callback on the exact active turn", async () => {
+    const pending = pendingExtensionTurn("turn-1");
+    const runtime = fakeRuntime(); vi.mocked(runtime.startTurn).mockReturnValue(pending.turn);
+    let created!: AcpRuntimeOptions;
+    const options = openOptions(fakeCommand());
+    const port = await openCodexAcpxRuntime(options, {
+      createRegistry: () => registry(), createStore: () => store(), createRuntime: value => { created = value; return runtime; },
+    });
+    let callbackSignal!: AbortSignal;
+    const handler = vi.fn((_request, context) => new Promise<{ outcome: "cancel" }>(resolve => {
+      callbackSignal = context.signal;
+      context.signal.addEventListener("abort", () => resolve({ outcome: "cancel" }), { once: true });
+    }));
+    const turn = port.startTurn({ text: "Write", requestId: "turn-1", onPermissionRequest: handler });
+    const permission = created.onPermissionRequest!({ sessionId: "backend-1", raw: { sessionId: "backend-1" }, inferredKind: "edit" } as never,
+      { signal: new AbortController().signal, responseDelivery: Promise.resolve() });
+    expect(callbackSignal.aborted).toBe(false);
+    const oldMode = options.permissionMode;
+    options.permissionMode = "approve-all";
+    const expiredContext = new AbortController(); expiredContext.abort();
+    await expect(created.onPermissionRequest!({ sessionId: "backend-1", raw: {}, inferredKind: "edit" } as never,
+      { signal: expiredContext.signal })).resolves.toEqual({ outcome: "cancel" });
+    options.permissionMode = oldMode;
+    await turn.cancel();
+    expect(callbackSignal.aborted).toBe(true);
+    await expect(permission).resolves.toEqual({ outcome: "cancel" });
+    await expect(created.onPermissionRequest!({ sessionId: "backend-1", raw: {}, inferredKind: "edit" } as never,
+      { signal: new AbortController().signal })).resolves.toEqual({ outcome: "cancel" });
+    expect(handler).toHaveBeenCalledOnce();
+    pending.settle(); await turn.result;
+    // The active-turn pointer has been cleared. Even full-auto policy cannot
+    // authorize a late callback from the retired turn.
+    options.permissionMode = "approve-all";
+    await expect(created.onPermissionRequest!({ sessionId: "backend-1", raw: {}, inferredKind: "edit" } as never,
+      { signal: new AbortController().signal })).resolves.toEqual({ outcome: "cancel" });
+    await port.close({ reason: "test complete" });
+  });
+
  it("fails Cursor adapter admission when the provider never acknowledges instructions", async () => {
    const runtime = fakeRuntime();
    const options = openOptions(fakeCommand());
@@ -268,8 +268,9 @@ export async function openQualifiedAcpxRuntime(
  );
  const permissionBoundary: {
    active: AbortController | null;
+    hasAdmittedTurn: boolean;
    handler?: AcpRuntimeOptions["onPermissionRequest"];
-  } = { active: null };
+  } = { active: null, hasAdmittedTurn: false };
  const extensionProfile = ACPX_CAPABILITY_PROFILES[options.profile.agent];
  const extensionRequests = new Set(extensionProfile.extensionRequests);
  const extensionNotifications = new Set(extensionProfile.extensionNotifications);
@@ -374,6 +375,8 @@ export async function openQualifiedAcpxRuntime(
        || (rawSessionId !== undefined && rawSessionId !== request.sessionId)) {
        return { outcome: "reject_once" };
      }
+      if ((permissionBoundary.hasAdmittedTurn && !extensionBoundary.active)
+        || extensionBoundary.active?.signal.aborted || context.signal.aborted) return { outcome: "cancel" };
      const disposition = decideAcpxPermission(
        options.profile.agent,
        options.permissionMode,
@@ -387,15 +390,16 @@ export async function openQualifiedAcpxRuntime(
      );
      if (disposition === "delegate") {
        const active = permissionBoundary.active;
+        const turn = extensionBoundary.active;
        const handler = permissionBoundary.handler;
-        if (active && handler && !active.signal.aborted && !context.signal.aborted) {
+        if (active && turn && handler && !active.signal.aborted && !turn.signal.aborted && !context.signal.aborted) {
          // Capture this turn's callback before awaiting. A session-lifetime
          // callback must never acquire the next turn's approval authority.
          const decision = await handler(request, {
-            signal: AbortSignal.any([active.signal, context.signal]),
+            signal: AbortSignal.any([active.signal, turn.signal, context.signal]),
            responseDelivery: context.responseDelivery,
          });
-          if (permissionBoundary.active !== active || active.signal.aborted || context.signal.aborted) {
+          if (permissionBoundary.active !== active || extensionBoundary.active !== turn || active.signal.aborted || turn.signal.aborted || context.signal.aborted) {
            return { outcome: "cancel" };
          }
          return decision ?? { outcome: "cancel" };
@@ -1013,7 +1017,7 @@ function runtimePort(
  runtimeCloseTimeoutMs: number,
  goalState: AcpxRuntimeGoalState,
  commandLaunches: { count: number; refreshConsumedCommand?: () => Promise<void> },
-  permissionBoundary: { active: AbortController | null; handler?: AcpRuntimeOptions["onPermissionRequest"] },
+  permissionBoundary: { active: AbortController | null; hasAdmittedTurn: boolean; handler?: AcpRuntimeOptions["onPermissionRequest"] },
  extensionBoundary: AcpxRuntimeExtensionBoundary,
  assertPromptPolicy?: (text: string) => void,
 ): AcpxRuntimePort {
@@ -1388,6 +1392,7 @@ function runtimePort(
        controller.abort(new Error("ACPX extension turn expired"));
        if (extensionBoundary.active === extensionTurn) extensionBoundary.active = null;
      };
+      permissionBoundary.hasAdmittedTurn = true;
      permissionBoundary.active = approval;
      permissionBoundary.handler = input.onPermissionRequest;
      const finishOwnershipAdmission =
@@ -1372,7 +1372,7 @@ describe("ACPX runtime host", () => {
        onExtensionRequest,
        onExtensionNotification,
      }),
-    ).toBe(turn);
+    ).toMatchObject({ requestId: turn.requestId });
    expect(startTurn).toHaveBeenCalledWith({
      text: "Complete the task.",
      requestId: "turn-1",
@@ -1388,8 +1388,8 @@ describe("ACPX runtime host", () => {
    expect(turn.cancel).toHaveBeenCalledWith({ reason: "user interrupt" });

    await host.close({ reason: "shutdown" });
-    expect(turn.cancel).toHaveBeenCalledWith({ reason: "shutdown" });
-    expect(runtime.close).toHaveBeenCalledOnce();
+    expect(turn.cancel).toHaveBeenCalledTimes(2);
+    expect(runtime.close).toHaveBeenCalledExactlyOnceWith({ reason: "user interrupt" });
    expect(() => host.startTurn({ text: "Late", requestId: "turn-3" })).toThrow(
      "is closing",
    );
@@ -1,3 +1,4 @@
+import { withAcpxTurnCancellation } from "./turn-cancellation.js";
 import { resolveCursorSessionMode, type CursorSessionMode } from "./cursor-mode.js";
 import { dirname, join } from "node:path";
 import { bindAcpxAgentFiles } from "./agent-files-binding.js";
@@ -286,6 +287,7 @@ export class AcpxRuntimeHost {
  readonly #claudeSkillNames: readonly string[];
  readonly #assertAgentFilesHeld: (() => void) | undefined;
  #activeTurn: AcpxRuntimeTurn | null = null;
+  #activeTurnInterruption: AcpxRuntimeTurn["cancel"] | null = null;
  #closingStarted = false;
  #closePromise: Promise<void> | null = null;
  #closed = false;
@@ -802,26 +804,30 @@ export class AcpxRuntimeHost {
      ...(input.onExtensionNotification ? { onExtensionNotification: input.onExtensionNotification } : {}),
    });
    this.#activeTurn = turn;
-    void turn.result
+    const managed = withAcpxTurnCancellation(turn, reason => this.close({ reason }), ACPX_TURN_CANCELLATION_SHUTDOWN_BOUND_MS);
+    this.#activeTurnInterruption = managed.cancel;
+    void managed.result
      .finally(() => {
        // Once shutdown owns this turn, retain its cancellation handle until
        // runtime cleanup succeeds. The result may settle while cleanup is
        // failing, and a later close must still be able to retry cancellation.
        if (this.#activeTurn === turn && !this.#closingStarted) {
          this.#activeTurn = null;
+          this.#activeTurnInterruption = null;
        }
      })
      .catch(() => undefined);
-    return turn;
+    return managed;
+  }
+
+  isClosed(): boolean {
+    return this.#closed;
  }

  async interruptActiveTurn(reason: string): Promise<void> {
    const turn = this.#activeTurn;
    if (!turn) throw new Error("ACPX runtime host has no active turn");
-    const cancellationError = await boundedCancellation(
-      turn.cancel({ reason: boundedReason(reason) }),
-    );
-    if (cancellationError) throw cancellationError;
+    await this.#activeTurnInterruption!({ reason: boundedReason(reason) });
  }

  async close(input: { reason: string }): Promise<void> {
@@ -0,0 +1,79 @@
+import type { AcpRuntimeTurnResult } from "acpx/runtime";
+import { afterEach, expect, it, vi } from "vitest";
+import { withAcpxTurnCancellation } from "./turn-cancellation.js";
+import type { AcpxRuntimeTurn } from "./runtime-host.js";
+
+const deferred = <T>() => {
+  let resolve!: (value: T) => void, reject!: (reason: unknown) => void;
+  const promise = new Promise<T>((r, j) => { resolve = r; reject = j; });
+  return { promise, resolve, reject };
+};
+function fixture() {
+  const terminal = deferred<AcpRuntimeTurnResult>();
+  const stream = deferred<IteratorResult<{ type: "text_delta"; text: string }>>();
+  const cleanup = deferred<void>();
+  const cancel = vi.fn(async () => undefined), close = vi.fn(() => cleanup.promise);
+  const raw: AcpxRuntimeTurn = {
+    requestId: "turn", promptStarted: Promise.resolve(), result: terminal.promise,
+    cancel, closeStream: vi.fn(async () => undefined),
+    events: { [Symbol.asyncIterator]: () => ({ next: () => stream.promise }) },
+  };
+  return { terminal, stream, cleanup, cancel, close, turn: withAcpxTurnCancellation(raw, close, 20) };
+}
+afterEach(() => vi.useRealTimers());
+
+it("does not treat acknowledgement or stdout EOF during cleanup as terminal proof", async () => {
+  vi.useFakeTimers(); const f = fixture();
+  const next = f.turn.events[Symbol.asyncIterator]().next();
+  const stopped = f.turn.cancel({ reason: "operator stop" });
+  let settled = false; void f.turn.result.finally(() => { settled = true; });
+  await vi.advanceTimersByTimeAsync(19);
+  expect(f.cancel).toHaveBeenCalledExactlyOnceWith({ reason: "operator stop" });
+  expect(f.close).not.toHaveBeenCalled(); expect(settled).toBe(false);
+  await vi.advanceTimersByTimeAsync(1);
+  expect(f.close).toHaveBeenCalledExactlyOnceWith("operator stop");
+  f.stream.reject(new Error("stdout closed during cleanup"));
+  f.terminal.resolve({ status: "failed", error: { message: "EOF", retryable: false } });
+  await Promise.resolve(); expect(settled).toBe(false);
+  f.cleanup.resolve();
+  await stopped;
+  await expect(f.turn.result).resolves.toEqual({ status: "cancelled", stopReason: "cancelled_after_runtime_close" });
+  await expect(next).resolves.toMatchObject({ done: true });
+});
+
+it("never publishes cancellation when owned provider cleanup fails", async () => {
+  vi.useFakeTimers(); const f = fixture();
+  const result = expect(f.turn.result).rejects.toThrow("provider still alive");
+  const next = expect(f.turn.events[Symbol.asyncIterator]().next()).rejects.toThrow("provider still alive");
+  const stopped = expect(f.turn.cancel()).rejects.toThrow("provider still alive");
+  await vi.advanceTimersByTimeAsync(20); f.cleanup.reject(new Error("provider still alive"));
+  await Promise.all([result, next, stopped]);
+});
+
+it.each(["completed", "cancelled", "failed"] as const)("settles a racing provider %s terminal without waiting for the run timeout", async status => {
+  const f = fixture();
+  const stopped = f.turn.cancel();
+  const terminal: AcpRuntimeTurnResult = status === "failed" ? { status, error: { message: "provider failure", retryable: false } } : { status };
+  f.terminal.resolve(terminal);
+  await stopped;
+  await expect(f.turn.result).resolves.toEqual(status === "completed" ? { status: "cancelled", stopReason: "cancelled_after_provider_completed" } : terminal);
+  expect(f.close).not.toHaveBeenCalled();
+});
+
+it("keeps a provider transport failure distinct from a cancellation timeout", async () => {
+  const f = fixture();
+  const result = expect(f.turn.result).rejects.toThrow("provider stdout closed");
+  const stopped = expect(f.turn.cancel()).rejects.toThrow("provider stdout closed");
+  f.terminal.reject(new Error("provider stdout closed"));
+  await Promise.all([result, stopped]); expect(f.close).not.toHaveBeenCalled();
+});
+
+it("drains ordinary output even when its normal result is already settled", async () => {
+  const f = fixture();
+  f.terminal.resolve({ status: "completed" });
+  const iterator = f.turn.events[Symbol.asyncIterator]();
+  f.stream.resolve({ value: { type: "text_delta", text: "final bytes" }, done: false });
+  await expect(iterator.next()).resolves.toEqual({ value: { type: "text_delta", text: "final bytes" }, done: false });
+  await iterator.return?.();
+  await expect(f.turn.result).resolves.toEqual({ status: "completed" });
+});
@@ -0,0 +1,89 @@
+import type { AcpRuntimeTurnResult } from "acpx/runtime";
+import type { AcpxRuntimeTurn } from "./runtime-host.js";
+
+/** An ACP cancel acknowledgement is not terminal evidence. Await a provider
+ * terminal or owned runtime retirement before ending the caller's stream.
+ */
+export function withAcpxTurnCancellation(
+  raw: AcpxRuntimeTurn,
+  closeRuntime: (reason: string) => Promise<void>,
+  graceMs: number,
+): AcpxRuntimeTurn {
+  let cancellation: Promise<AcpRuntimeTurnResult> | undefined;
+  const retirement = new AbortController();
+  let retire!: (value: AcpRuntimeTurnResult) => void;
+  let failRetirement!: (error: unknown) => void;
+  const retired = new Promise<AcpRuntimeTurnResult>((resolve, reject) => {
+    retire = resolve; failRetirement = reject;
+  });
+  void retired.catch(() => undefined);
+  const cancel = (input?: { reason?: string }): Promise<void> => {
+    if (!cancellation) {
+      const reason = input?.reason ?? "Paperclip cancellation";
+      // Install the gate before invoking provider code, including synchronous
+      // callback cancellation and a terminal racing the acknowledgement.
+      cancellation = Promise.resolve().then(async () => {
+        void Promise.resolve().then(() => raw.cancel({ reason })).catch(() => undefined);
+        let timer: ReturnType<typeof setTimeout> | undefined;
+        const terminal = await Promise.race([
+          raw.result,
+          new Promise<null>(resolve => { timer = setTimeout(() => resolve(null), graceMs); }),
+        ]).finally(() => clearTimeout(timer));
+        if (terminal !== null) {
+          return terminal.status === "completed"
+            ? { status: "cancelled" as const, stopReason: "cancelled_after_provider_completed" }
+            : terminal;
+        }
+        // closeRuntime owns process identities and lifetime leases. A rejected
+        // close must never turn an unproven process exit into cancellation.
+        await closeRuntime(reason);
+        return { status: "cancelled" as const, stopReason: "cancelled_after_runtime_close" };
+      });
+      void cancellation.then(
+        value => { retire(value); retirement.abort(); },
+        error => { failRetirement(error); retirement.abort(); },
+      );
+    }
+    return cancellation.then(() => undefined);
+  };
+  const result = Promise.race([
+    raw.result.then(value => cancellation ?? value, error => cancellation ?? Promise.reject(error)),
+    retired,
+  ]);
+  void result.catch(() => undefined);
+  return {
+    requestId: raw.requestId,
+    promptStarted: raw.promptStarted,
+    result,
+    cancel,
+    closeStream: input => raw.closeStream(input),
+    events: (async function* () {
+      const iterator = raw.events[Symbol.asyncIterator]();
+      try {
+        while (true) {
+          if (retirement.signal.aborted) { await retired; return; }
+          let onRetirement!: () => void;
+          const interrupted = new Promise<{ kind: "retired" }>((resolve, reject) => {
+            onRetirement = () => { void retired.then(() => resolve({ kind: "retired" }), reject); };
+            retirement.signal.addEventListener("abort", onRetirement, { once: true });
+          });
+          const next = await Promise.race([
+            iterator.next().then(value => ({ kind: "event" as const, value })),
+            interrupted,
+          ]).finally(() => retirement.signal.removeEventListener("abort", onRetirement)).catch(async error => {
+            if (!cancellation) throw error;
+            // EOF during owned close is not itself proof that cleanup worked.
+            await cancellation;
+            return { kind: "retired" as const };
+          });
+          if (next.kind === "retired" || next.value.done) return;
+          yield next.value.value;
+        }
+      } finally {
+        // Some provider iterators await a stuck RPC even after retirement.
+        // Do not make their optional return() another unbounded shutdown wait.
+        void iterator.return?.().catch(() => undefined);
+      }
+    })(),
+  };
+}