From 285073afe028ba6511fecc8c8d56bd5cd20b89a2 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 29 Sep 2026 21:38:37 -0500 Subject: [PATCH] fix(runner): settle ACPX cancellation after verified retirement Co-Authored-By: Paperclip --- .../runner-core/src/acpx_provider_backend.rs | 35 ++++++-- .../runner-core/src/acpx_provider_session.rs | 37 +++++++- .../runner-core/src/bin/fake-acpx-sidecar.rs | 50 +++++++++-- .../crates/runner-core/src/provider_events.rs | 8 +- .../tests/acpx_provider_resolutions.rs | 63 +++++++++++++ .../runner-core/tests/acpx_provider_turns.rs | 25 ++++++ .../backends/harness-driver-backend.test.ts | 52 +++++++++++ .../src/backends/harness-driver-backend.ts | 34 ++++++- .../src/cli/acpx-runtime-sidecar.test.ts | 32 ++++++- .../src/cli/acpx-runtime-sidecar.ts | 12 ++- .../acpx/codex-runtime-adapter.test.ts | 38 ++++++++ .../src/drivers/acpx/codex-runtime-adapter.ts | 15 ++-- .../src/drivers/acpx/runtime-host.test.ts | 6 +- .../src/drivers/acpx/runtime-host.ts | 18 ++-- .../drivers/acpx/turn-cancellation.test.ts | 79 ++++++++++++++++ .../src/drivers/acpx/turn-cancellation.ts | 89 +++++++++++++++++++ 16 files changed, 551 insertions(+), 42 deletions(-) create mode 100644 packages/paperclip-runner/src/drivers/acpx/turn-cancellation.test.ts create mode 100644 packages/paperclip-runner/src/drivers/acpx/turn-cancellation.ts diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index 3b07269d07d..9cf9a96e61d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -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(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs index 9958237a0db..f0c9c487fa6 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -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 { 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, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs index 58e94296539..c52cb86c339 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs @@ -232,6 +232,9 @@ fn run() -> Result<(), Box> { | "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> { | "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> { 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> { "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> { 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!({ diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs index e0eb95f611c..3d3f7cbc49e 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs @@ -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, }}), ) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs index 9028e297984..f14b0367ec3 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs @@ -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(); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs index a3b94bc8988..7e9e4393da1 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs @@ -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(); + } +} diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts index d550d594d64..470bd66d71b 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts @@ -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() { diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index fd280b5df81..3dd6a5b4650 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -511,10 +511,42 @@ class HarnessNativeSession implements NativeSession { let streamFailure: unknown = null; const observedPendingInputs = new Map>(); 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; diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts index 3179baf3ce6..18946a66f59 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts @@ -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(); - 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(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")')); diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index f6a3db693c0..37d5c83193f 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -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 { async function waitForPermission( activeTurnId: string, + agent: QualifiedAcpxAgent, request: AcpPermissionRequest, context: { signal: AbortSignal; responseDelivery?: Promise }, 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); }); } 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 cd28b25e574..d159d6fa50d 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 @@ -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()); 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 00b6209e29a..6cc5ccb847e 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -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 }, - 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 = 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 0c8ffa797ce..a5dc0584124 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts @@ -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", ); diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts index 559dc8344c8..5599838b641 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 { 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 | 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 { 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 { diff --git a/packages/paperclip-runner/src/drivers/acpx/turn-cancellation.test.ts b/packages/paperclip-runner/src/drivers/acpx/turn-cancellation.test.ts new file mode 100644 index 00000000000..52201e5bab6 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/turn-cancellation.test.ts @@ -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 = () => { + let resolve!: (value: T) => void, reject!: (reason: unknown) => void; + const promise = new Promise((r, j) => { resolve = r; reject = j; }); + return { promise, resolve, reject }; +}; +function fixture() { + const terminal = deferred(); + const stream = deferred>(); + const cleanup = deferred(); + 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" }); +}); diff --git a/packages/paperclip-runner/src/drivers/acpx/turn-cancellation.ts b/packages/paperclip-runner/src/drivers/acpx/turn-cancellation.ts new file mode 100644 index 00000000000..c5dc8f67c0f --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/turn-cancellation.ts @@ -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, + graceMs: number, +): AcpxRuntimeTurn { + let cancellation: Promise | undefined; + const retirement = new AbortController(); + let retire!: (value: AcpRuntimeTurnResult) => void; + let failRetirement!: (error: unknown) => void; + const retired = new Promise((resolve, reject) => { + retire = resolve; failRetirement = reject; + }); + void retired.catch(() => undefined); + const cancel = (input?: { reason?: string }): Promise => { + 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 | undefined; + const terminal = await Promise.race([ + raw.result, + new Promise(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); + } + })(), + }; +}