From 51db5098994400463c7669f70bffe4b2691dc86c Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 28 Sep 2026 12:04:42 -0500 Subject: [PATCH] Expire durable ACP interactions when their provider turn is lost Retain a bounded private pending-request ledger across controller and runner reconnects. Emit non-replayable expirations before terminal facts, preserve complete questions, and never restore approvals or mutations after provider loss. Co-Authored-By: Paperclip --- .../runner-core/src/acpx_provider_backend.rs | 260 +++++++++++++++++- .../runner-core/src/acpx_provider_state.rs | 38 ++- .../crates/runner-core/src/durable/state.rs | 10 +- .../crates/runner-core/src/provider_events.rs | 56 +++- .../runner-core/tests/acpx_provider_state.rs | 69 +++++ .../src/reducer/session-reducer.test.ts | 16 ++ 6 files changed, 434 insertions(+), 15 deletions(-) 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 65d56263d7f..602255ad0e4 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 @@ -1,4 +1,4 @@ -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::fs::{self, DirBuilder, File}; use std::io::{Read, Write}; use std::net::TcpListener; @@ -427,6 +427,8 @@ struct AcpxDurableState { goal_source_revision: Option, #[serde(default)] pending_events: VecDeque, + #[serde(default)] + pending_runtime_requests: BTreeMap, #[serde(default = "initial_event_sequence")] next_event_sequence: u64, } @@ -452,6 +454,7 @@ impl AcpxDurableState { goal_revision: 0, goal_source_revision: None, pending_events: VecDeque::new(), + pending_runtime_requests: BTreeMap::new(), next_event_sequence: initial_event_sequence(), } } @@ -472,6 +475,7 @@ impl AcpxDurableState { "ACPX durable launch profile digest does not match runner startup", )); } + validate_pending_runtime_requests(&self.pending_runtime_requests)?; let mut ids = HashSet::new(); if self.schema != ACPX_PROVIDER_STATE_SCHEMA || self.launch_profile_digest.len() != 71 @@ -491,6 +495,7 @@ impl AcpxDurableState { .attempted_turn_controls .iter() .any(|id| !is_stable_id(id, 160)) + || (!self.pending_runtime_requests.is_empty() && self.active_turn_id.is_none()) || self.pending_events.len() > MAX_PENDING_EVENTS || self.pending_events.iter().any(|event| { event_sequence(&event.executor_event_id) @@ -528,12 +533,74 @@ impl AcpxDurableState { Ok(()) } + fn expire_runtime_requests( + &mut self, + reason: &str, + cancelled: bool, + ) -> Result<(), DurableRunnerError> { + for request in self + .pending_runtime_requests + .values() + .cloned() + .collect::>() + { + let input = request.get("type").and_then(Value::as_str) == Some("input"); + let mut payload = json!({ + "provider":"acpx", "requestId":request["requestId"], + "requestKind":request["requestKind"], "requestType":request["type"], + "turnId":request["turnId"], "itemId":request["itemId"], + "reason":reason, "replayAllowed":false, "adapter":"acpx-runtime-sidecar", + }); + if input { + payload["request"] = request; + } + self.push(NormalizedProviderEvent { + event_type: if cancelled { + "runtime_request.cancelled" + } else { + "runtime_request.expired" + } + .to_owned(), + priority: EventPriority::P0, + payload, + })?; + } + Ok(()) + } + fn push(&mut self, event: NormalizedProviderEvent) -> Result<(), DurableRunnerError> { if self.pending_events.len() >= MAX_PENDING_EVENTS { return Err(DurableRunnerError::invalid( "ACPX provider event backlog exceeds its durable limit", )); } + if event.event_type == "runtime_request.created" { + let request = event.payload.get("request").cloned().ok_or_else(|| { + DurableRunnerError::invalid("ACPX runtime request omitted its payload") + })?; + let request_id = request + .get("requestId") + .and_then(Value::as_str) + .ok_or_else(|| { + DurableRunnerError::invalid("ACPX runtime request omitted its identity") + })? + .to_owned(); + let mut next = self.pending_runtime_requests.clone(); + if next.insert(request_id, request).is_some() { + return Err(DurableRunnerError::invalid( + "ACPX runtime request reused a pending durable identity", + )); + } + validate_pending_runtime_requests(&next)?; + self.pending_runtime_requests = next; + } else if matches!( + event.event_type.as_str(), + "runtime_request.resolved" | "runtime_request.expired" | "runtime_request.cancelled" + ) { + if let Some(id) = event.payload.get("requestId").and_then(Value::as_str) { + self.pending_runtime_requests.remove(id); + } + } let sequence = self.next_event_sequence; self.next_event_sequence = sequence .checked_add(1) @@ -548,6 +615,53 @@ impl AcpxDurableState { } } +fn validate_pending_runtime_requests( + requests: &BTreeMap, +) -> Result<(), DurableRunnerError> { + if requests.len() > 1024 + || serde_json::to_vec(requests).map_or(true, |bytes| bytes.len() > 8 * 1024 * 1024) + { + return Err(DurableRunnerError::invalid( + "ACPX durable runtime request ledger exceeds its bound", + )); + } + for (id, request) in requests { + let input = request.get("type").and_then(Value::as_str) == Some("input"); + if !is_stable_id(id, 160) + || request.get("requestId").and_then(Value::as_str) != Some(id) + || request.get("schema").and_then(Value::as_str) != Some("paperclip.runtime_request.v2") + || request.get("status").and_then(Value::as_str) != Some("pending") + || request.get("requestKind").and_then(Value::as_str) + != Some(if input { + "runtime" + } else { + "permission_approval" + }) + || (!input && request.get("type").and_then(Value::as_str) != Some("permission")) + || ["turnId", "itemId"].iter().any(|field| { + request + .get(field) + .and_then(Value::as_str) + .is_none_or(|value| !is_stable_id(value, DURABLE_STABLE_ID_CHARS)) + }) + || (input + && request.pointer("/input/schema").and_then(Value::as_str) + != Some("paperclip.question_set.v1")) + || (!input && request.get("input").is_some()) + || serde_json::to_vec(request).map_or(true, |bytes| bytes.len() > 256 * 1024) + { + return Err(DurableRunnerError::invalid( + "ACPX durable runtime request ledger has an invalid identity or type", + )); + } + if input { + crate::acpx_event_payload::validate_question_set(&request["input"]) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + } + } + Ok(()) +} + pub struct AcpxCommandExecutor { state_dir: PathBuf, context: AcpxEventProjectionContext, @@ -675,6 +789,7 @@ impl AcpxCommandExecutor { .state .as_mut() .expect("ACPX state remains available during recovery"); + state.expire_runtime_requests("provider_process_lost", false)?; state.lifecycle = "closed".to_owned(); state.active_turn_id = None; state.provider_exit_unconfirmed = true; @@ -1285,6 +1400,7 @@ impl AcpxCommandExecutor { .state .as_mut() .expect("ACPX state remains available after provider termination"); + state.expire_runtime_requests("explicit_cancellation", true)?; state.active_turn_id = None; self.context.provider_turn_id = None; // Persist a non-attachable, recoverable boundary before the fallible @@ -1358,6 +1474,10 @@ impl AcpxCommandExecutor { .map_err(|error| { DurableRunnerError::invalid(format!("ACPX runtime response failed: {error}")) })?; + if let Some(state) = self.state.as_mut() { + state.pending_runtime_requests.remove(request_id); + } + self.save_state()?; Ok(CommandExecution { result: json!({"status": "delivered", "requestId": request_id}), events: vec![( @@ -1421,6 +1541,7 @@ impl AcpxCommandExecutor { .state .as_mut() .ok_or_else(|| DurableRunnerError::invalid("ACPX provider is not prepared"))?; + state.expire_runtime_requests("explicit_cancellation", true)?; state.lifecycle = "closed".to_owned(); state.active_turn_id = None; self.context.provider_turn_id = None; @@ -1480,7 +1601,7 @@ impl AcpxCommandExecutor { if self .state .as_ref() - .is_some_and(|state| !state.pending_events.is_empty()) + .is_some_and(|state| !state.pending_events.is_empty() || state.lifecycle == "closed") || self.session.is_none() { return Ok(()); @@ -1490,10 +1611,33 @@ impl AcpxCommandExecutor { .session .as_mut() .expect("ACPX session remains available while polling") - .poll_event(Duration::from_millis(1)) - .map_err(|error| { - DurableRunnerError::invalid(format!("ACPX provider failed: {error}")) - })?; + .poll_event(Duration::from_millis(1)); + let events = match events { + Ok(events) => events, + Err(error) => { + let state = self + .state + .as_mut() + .expect("ACPX failed provider has durable state"); + state.expire_runtime_requests("provider_process_lost", false)?; + let failed_turn = state.active_turn_id.take(); + state.lifecycle = "closed".to_owned(); + state.provider_exit_unconfirmed = true; + if let Some(turn_id) = failed_turn { + state.push(NormalizedProviderEvent { event_type: "turn.failed".to_owned(), priority: EventPriority::P0, + payload: json!({"provider":"acpx","providerTurnId":turn_id,"status":"failed","providerTerminalObserved":false, + "code":"acpx_provider_transport_failed","message":crate::durable::redact_text(&error.to_string())}) })?; + state.push(NormalizedProviderEvent { event_type: "run.terminal".to_owned(), priority: EventPriority::P0, + payload: json!({"schema":"paperclip.prp.terminal.v1","status":"failed","turnTerminalState":"failed","runTerminalState":"failed","reportedWorkDisposition":"unknown","provider":"acpx"}) })?; + } + self.context.provider_turn_id = None; + // A failed transport cannot attest escaped provider cleanup. + // Drop the closed process owner; later cleanup must prove the lifetime fence. + self.session = None; + self.save_state()?; + return Ok(()); + } + }; let Some(events) = events else { break }; let mut provider_turn_settled = false; for event in events { @@ -1976,6 +2120,87 @@ mod tests { }) } + fn pending_input_request(id: &str) -> Value { + json!({"schema":"paperclip.runtime_request.v2","requestId":id,"requestKind":"runtime","type":"input","status":"pending", + "turnId":"turn-1","itemId":"item-1","prompt":"Choose","input":{"schema":"paperclip.question_set.v1","questions":[{"id":"q","prompt":"Choose","answerMode":"text","required":true}]}, + "origin":{"adapter":"acpx-runtime-sidecar","provider":"cursor","method":"cursor/ask_question"}}) + } + + #[test] + fn durable_runtime_ledger_retains_only_unsettled_requests_and_rejects_forged_types() { + let operations = Vec::new(); + let tool_set = AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.into(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(&operations).unwrap(), + operations, + }; + let mut state = AcpxDurableState::new( + serde_json::from_value(descriptor("codex")).unwrap(), + tool_set, + "test".into(), + ); + state + .push(NormalizedProviderEvent { + event_type: "runtime_request.created".into(), + priority: EventPriority::P0, + payload: json!({"request":pending_input_request("input-1")}), + }) + .unwrap(); + assert!(state + .push(NormalizedProviderEvent { + event_type: "runtime_request.created".into(), + priority: EventPriority::P0, + payload: json!({"request":pending_input_request("input-1")}) + }) + .is_err()); + let mut forged = pending_input_request("forged"); + forged["requestKind"] = json!("permission_approval"); + assert!(state + .push(NormalizedProviderEvent { + event_type: "runtime_request.created".into(), + priority: EventPriority::P0, + payload: json!({"request":forged}) + }) + .is_err()); + state.pending_events.clear(); + let mut restored: AcpxDurableState = + serde_json::from_value(serde_json::to_value(&state).unwrap()).unwrap(); + restored + .expire_runtime_requests("provider_process_lost", false) + .unwrap(); + assert_eq!(restored.pending_events.len(), 1); + assert_eq!( + restored.pending_events[0].event_type, + "runtime_request.expired" + ); + assert_eq!( + restored.pending_events[0].payload["request"], + pending_input_request("input-1") + ); + assert_eq!(restored.pending_events[0].payload["replayAllowed"], false); + restored + .expire_runtime_requests("provider_process_lost", false) + .unwrap(); + assert_eq!(restored.pending_events.len(), 1); + state + .push(NormalizedProviderEvent { + event_type: "runtime_request.resolved".into(), + priority: EventPriority::P0, + payload: json!({"requestId":"input-1"}), + }) + .unwrap(); + assert!(state.pending_runtime_requests.is_empty()); + let mut old = serde_json::to_value(state).unwrap(); + old.as_object_mut() + .unwrap() + .remove("pendingRuntimeRequests"); + assert!(serde_json::from_value::(old) + .unwrap() + .pending_runtime_requests + .is_empty()); + } + #[test] fn retained_events_exposes_terminal_suffix_without_restoring_provider() { let directory = temporary_directory("retained-terminal-suffix"); @@ -2382,6 +2607,15 @@ mod tests { state.lifecycle = "turn_active".to_owned(); state.identity = Some(identity); state.active_turn_id = Some("turn-1".to_owned()); + state + .push(NormalizedProviderEvent { + event_type: "runtime_request.created".into(), + priority: EventPriority::P0, + payload: json!({"request":pending_input_request("restart-input")}), + }) + .unwrap(); + state.pending_events.clear(); // The controller already acknowledged creation. + let config = test_config(&directory, Some(launch_profile)); let mut original = AcpxCommandExecutor::with_runner_config(&directory, &config); original.state = Some(state); @@ -2443,11 +2677,15 @@ mod tests { assert_eq!(snapshot.result["status"], "closed"); assert!(!marker.exists()); let events = recovered.poll_events().unwrap(); - assert_eq!(events[0].event_type, "turn.failed"); - assert_eq!(events[0].payload["providerShutdownFailed"], true); - assert_eq!(events[1].event_type, "run.terminal"); - assert_eq!(events[1].payload["schema"], "paperclip.prp.terminal.v1"); - assert_eq!(events[1].payload["turnTerminalState"], "failed"); + assert_eq!(events[0].event_type, "runtime_request.expired"); + assert_eq!(events[0].payload["requestId"], "restart-input"); + assert_eq!(events[0].payload["replayAllowed"], false); + assert_eq!(events[1].event_type, "turn.failed"); + assert_eq!(events[1].payload["providerShutdownFailed"], true); + assert_eq!(events[2].event_type, "run.terminal"); + assert_eq!(events[2].payload["schema"], "paperclip.prp.terminal.v1"); + assert_eq!(events[2].payload["turnTerminalState"], "failed"); + assert_eq!(recovered.poll_events().unwrap(), events); let cleanup_error = recovered .shutdown() .expect_err("cleanup must not succeed while the original lifetime remains active"); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs index 01d4c2294c6..876cf9d00f1 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs @@ -61,6 +61,12 @@ pub enum AcpxProviderStateEvent { question_set: Value, origin: Option, }, + RuntimeRequestEnded { + request_id: String, + question_set: Option, + origin: Option, + status: AcpxTurnStatus, + }, SemanticResult(AcpxSemanticResult), AssistantMessage { turn_id: String, @@ -84,6 +90,7 @@ struct PendingInput { runtime_request_id: String, value_bytes: usize, question_set: Value, + origin: Option, } /// Reduces validated sidecar events into bounded provider state. @@ -273,7 +280,8 @@ impl AcpxProviderState { question_set, origin, } => { - let value_bytes = value_bytes(&question_set)?; + let value_bytes = value_bytes(&question_set)? + + origin.as_ref().map(value_bytes).transpose()?.unwrap_or(0); self.admit_runtime_request(&request_id, value_bytes)?; let runtime_request_id = project_acpx_runtime_request_id(&request_id) .expect("a decoded ACPX input request has a bounded identity"); @@ -295,6 +303,7 @@ impl AcpxProviderState { runtime_request_id, value_bytes, question_set: question_set.clone(), + origin: origin.clone(), }, ) .is_some() @@ -355,8 +364,8 @@ impl AcpxProviderState { .expect("a decoded terminal event has a turn binding") .to_owned(); self.scope.clear_turn(&turn_id)?; + let mut events = self.end_pending_runtime_requests(status); self.clear_pending_requests(); - let mut events = Vec::new(); if status == AcpxTurnStatus::Completed && !self.assistant_text.is_empty() { events.push(AcpxProviderStateEvent::AssistantMessage { turn_id: turn_id.clone(), @@ -612,6 +621,31 @@ impl AcpxProviderState { Ok(()) } + pub(crate) fn end_pending_runtime_requests( + &mut self, + status: AcpxTurnStatus, + ) -> Vec { + let mut events = Vec::new(); + for (request_id, _) in std::mem::take(&mut self.pending_permissions) { + events.push(AcpxProviderStateEvent::RuntimeRequestEnded { + request_id, + question_set: None, + origin: None, + status, + }); + } + for (request_id, pending) in std::mem::take(&mut self.pending_inputs) { + events.push(AcpxProviderStateEvent::RuntimeRequestEnded { + request_id, + question_set: Some(pending.question_set), + origin: pending.origin, + status, + }); + } + self.pending_runtime_request_bytes = 0; + events + } + fn clear_pending_requests(&mut self) { self.pending_tools.clear(); self.pending_tool_input_bytes = 0; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs index 4c41ae5c689..9f853788900 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs @@ -1638,7 +1638,7 @@ fn preserve_bounded_display_content( } return Ok(preserved); } - if event_type == "runtime_request.created" { + if matches!(event_type, "runtime_request.created" | "runtime_request.expired" | "runtime_request.cancelled") { if let Some(input) = original.pointer("/request/input") { if input.get("schema").and_then(Value::as_str) == Some("paperclip.question_set.v1") { validate_question_set(input) @@ -3258,6 +3258,14 @@ mod tests { assert_eq!(stored["output"], text); assert_eq!(stored["outputTruncated"], false); + for event_type in ["runtime_request.expired", "runtime_request.cancelled"] { + state.enqueue_event(&config, event_type, EventPriority::P0, json!({ + "requestId":"request-1", "replayAllowed":false, + "request":{"schema":"paperclip.runtime_request.v2","requestId":"request-1","type":"input","status":"pending","input":input} + })).unwrap(); + assert_eq!(state.outbox.last().unwrap().envelope.pointer("/payload/payload/request/input/description"), Some(&json!(text))); + } + let mut oversized = input; oversized["description"] = json!("漢".repeat(70_000)); assert!(state 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 9fb42a14a6f..5e7429fbbf1 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 @@ -254,7 +254,7 @@ pub fn project_acpx_state_event( .and_then(Value::as_str) }) .map(|value| bounded_text(value, MAX_TEXT_CHARS)) - .unwrap_or_else(|| "Codex needs your input".to_owned()); + .unwrap_or_else(|| "Provider needs your input".to_owned()); let origin = project_runtime_request_origin(origin.as_ref())?; one( "runtime_request.created", @@ -275,6 +275,60 @@ pub fn project_acpx_state_event( }), ) } + AcpxProviderStateEvent::RuntimeRequestEnded { + request_id, + question_set, + origin, + status, + } => { + let cancelled = matches!( + status, + AcpxTurnStatus::Cancelled | AcpxTurnStatus::Interrupted + ); + let reason = match status { + AcpxTurnStatus::Completed => "turn_completed", + AcpxTurnStatus::Failed => "provider_process_lost", + _ => "explicit_cancellation", + }; + let (request_id, request) = if let Some(question_set) = question_set { + let created = project_acpx_state_event( + context, + &AcpxProviderStateEvent::InputRequest { + request_id: request_id.clone(), + question_set: question_set.clone(), + origin: origin.clone(), + }, + )?; + let request = created[0].payload["request"].clone(); + (request["requestId"].clone(), Some(request)) + } else { + validate_projection_identity( + request_id, + "permission request", + SHORT_STABLE_ID_CHARS, + )?; + (json!(request_id), None) + }; + let mut payload = json!({ + "provider":"acpx", "requestId":request_id, + "requestKind": if request.is_some() {"runtime"} else {"permission_approval"}, + "requestType": if request.is_some() {"input"} else {"permission"}, + "turnId":context.turn_id, "itemId":context.item_id, + "reason":reason, "replayAllowed":false, "adapter":"acpx-runtime-sidecar", + }); + if let Some(request) = request { + payload["request"] = request; + } + one( + if cancelled { + "runtime_request.cancelled" + } else { + "runtime_request.expired" + }, + EventPriority::P0, + payload, + ) + } AcpxProviderStateEvent::SemanticResult(result) => { validate_semantic_projection_identity(&result.call_id, &result.operation_id)?; if is_reserved_terminal_operation(&result.operation_id) { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs index 4c3361de93e..072a812d563 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs @@ -608,3 +608,72 @@ fn mutation_prose_survives_sidecar_decode_pending_state_and_semantic_projection( assert!(!input.to_string().contains("actual-credential")); } } + +#[test] +fn terminal_requests_expire_with_their_projected_identity_before_terminal_and_never_replay() { + for status in ["failed", "cancelled", "interrupted", "completed"] { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + let long_id = format!("input-{}", "x".repeat(200)); + let input = state.accept_event(&event(1, GeneratedAcpxSidecarEventType::RuntimeInputRequested, Some("turn-1"), + json!({"requestId":long_id,"questionSet":question_set(),"origin":{"adapter":"acpx-runtime-sidecar","provider":"cursor","method":"cursor/create_plan"}}))).unwrap(); + state.accept_event(&event(2, GeneratedAcpxSidecarEventType::RuntimePermissionRequested, Some("turn-1"), + json!({"requestId":"permission-1","title":"Run validation","choices":[{"key":"cancel","label":"Cancel"}]}))).unwrap(); + let context = AcpxEventProjectionContext { + run_id: "run-1".into(), + normalized_session_id: "session-1".into(), + turn_id: "turn-1".into(), + provider_turn_id: Some("turn-1".into()), + item_id: "item-1".into(), + }; + let created = project_acpx_state_event(&context, &input[0]).unwrap(); + let projected_id = created[0].payload["request"]["requestId"].clone(); + let terminal = state + .accept_event(&event( + 3, + GeneratedAcpxSidecarEventType::RuntimeTurnTerminal, + Some("turn-1"), + json!({"status":status}), + )) + .unwrap(); + let projected: Vec<_> = terminal + .iter() + .flat_map(|event| project_acpx_state_event(&context, event).unwrap()) + .collect(); + assert_eq!(projected.len(), 3); + assert_eq!(projected[0].payload["requestId"], "permission-1"); + assert_eq!(projected[1].payload["requestId"], projected_id); + assert_eq!(projected[1].payload["request"]["input"], question_set()); + assert_eq!( + projected[1].payload["request"]["origin"]["method"], + "cursor/create_plan" + ); + for ended in &projected[..2] { + assert_eq!(ended.payload["replayAllowed"], false); + assert_eq!( + ended.event_type, + if matches!(status, "cancelled" | "interrupted") { + "runtime_request.cancelled" + } else { + "runtime_request.expired" + } + ); + } + assert!(projected[2].event_type.starts_with("turn.")); + assert!(state + .pending_question_set(projected_id.as_str().unwrap()) + .is_none()); + assert!(state.complete_permission("permission-1").is_err()); + assert!(state + .complete_input(projected_id.as_str().unwrap()) + .is_err()); + assert!(state + .accept_event(&event( + 4, + GeneratedAcpxSidecarEventType::RuntimeTurnTerminal, + Some("turn-1"), + json!({"status":status}) + )) + .is_err()); + } +} diff --git a/packages/paperclip-runner/src/reducer/session-reducer.test.ts b/packages/paperclip-runner/src/reducer/session-reducer.test.ts index 8abc6267a7b..e85359f14c7 100644 --- a/packages/paperclip-runner/src/reducer/session-reducer.test.ts +++ b/packages/paperclip-runner/src/reducer/session-reducer.test.ts @@ -113,4 +113,20 @@ describe("deterministic PRP session reducer", () => { "Resolved permission: Allow the fake command?", ]); }); + + it.each(["expired", "cancelled"] as const)("replays authoritative ACP request %s before terminal without leaving an actionable prompt", async status => { + const fixture = await loadFixture("interrupted-run"); + const base = fixture.events.find(event => event.eventType === "runtime_request.created")!; + const created: PrpEvent = { ...base, sourceEventId: "acpx-created", sourceSeq: 1 }; + const ended: PrpEvent = { ...base, sourceEventId: "acpx-ended", sourceSeq: 2, + eventType: `runtime_request.${status}`, + payload: { requestId: "request_interrupted_permission", requestKind: "permission_approval", requestType: "permission", + reason: status === "expired" ? "provider_process_lost" : "explicit_cancellation", replayAllowed: false } }; + const terminal: PrpEvent = { ...base, sourceEventId: "acpx-terminal", sourceSeq: 3, eventType: "run.terminal", + payload: { schema: "paperclip.prp.terminal.v1", status: "failed", turnTerminalState: "failed", runTerminalState: "failed", reportedWorkDisposition: "unknown" } }; + const events = [created, ended, terminal]; + const result = reduceSessionEvents(createSessionSnapshot(fixture), events); + expect(result.requests).toEqual([expect.objectContaining({ requestId: "request_interrupted_permission", status })]); + expect(reduceSessionEvents(result, events).requests).toEqual(result.requests); + }); });