diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index 5188d9e8012..fb8099b141a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -387,6 +387,10 @@ pub trait CommandExecutor { Ok(()) } + fn provider_poll_is_live(&self) -> bool { + false + } + /// Removes the prefix returned by `poll_events` after every event in that /// prefix is durably committed to the PRP outbox. Implementations that retain /// provider events must not remove them before this acknowledgement. @@ -899,6 +903,7 @@ pub fn run_durable_runner_with_host_control( notifications.poll(&config, &mut state, &store); } let mut sent_source_seq = state.acked_source_seq; + let replayed_through_source_seq = state.highest_source_seq(); let mut ack_persistence = CumulativeAckPersistence::default(); let mut lifecycle_after_reply = CommandLifecycle::Continue; @@ -1055,6 +1060,8 @@ pub fn run_durable_runner_with_host_control( .min(reconnect_deadline.saturating_duration_since(Instant::now())); continue; } + let mut control_frame_consumed = false; + let mut quiet_through_source_seq = replayed_through_source_seq; loop { if !config.max_runtime.is_zero() && started.elapsed() >= config.max_runtime { break; @@ -1153,7 +1160,19 @@ pub fn run_durable_runner_with_host_control( } } } - let control_message = transport.receive_json(); + let control_message = if control_frame_consumed + && state.highest_source_seq() > quiet_through_source_seq + && executor.provider_poll_is_live() + { + transport.receive_queued_json() + } else { + transport.receive_json() + }; + control_frame_consumed = matches!(control_message, Ok(Some(_))); + let source_seq_before_poll = state.highest_source_seq(); + let polls_provider = matches!(control_message, Ok(None)) + && state.pending_provider_cleanup.is_none() + && state.acked_source_seq >= sent_source_seq; poll_executor_events_when_control_idle( &mut state, &store, @@ -1162,6 +1181,9 @@ pub fn run_durable_runner_with_host_control( sent_source_seq, &control_message, )?; + if polls_provider { + quiet_through_source_seq = source_seq_before_poll; + } let message = match control_message { Ok(Some(message)) => message, Ok(None) => continue, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/ack_release_poll.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/ack_release_poll.rs new file mode 100644 index 00000000000..25396a550ad --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/ack_release_poll.rs @@ -0,0 +1,322 @@ +#[test] +fn acknowledged_output_polls_the_provider_without_waiting_out_the_read_timeout() { + acknowledged_prefix_releases_provider_output(AckRelease::Idle); +} + +#[test] +fn control_queued_behind_an_ack_runs_before_the_next_provider_poll() { + acknowledged_prefix_releases_provider_output(AckRelease::QueuedCommand); +} + +#[test] +fn replayed_output_ack_keeps_the_read_timeout_before_the_provider_poll() { + acknowledged_prefix_releases_provider_output(AckRelease::Replayed); +} + +#[test] +fn welcome_command_output_polls_the_provider_without_waiting_out_the_read_timeout() { + acknowledged_prefix_releases_provider_output(AckRelease::WelcomeCommand); +} + +#[test] +fn restorable_provider_ack_keeps_the_read_timeout_before_the_provider_poll() { + acknowledged_prefix_releases_provider_output(AckRelease::NotLive); +} + +#[derive(Clone, Copy, Debug, PartialEq)] +enum AckRelease { + Idle, + QueuedCommand, + Replayed, + NotLive, + WelcomeCommand, +} + +fn acknowledged_prefix_releases_provider_output(mode: AckRelease) { + struct PollingExecutor { + live: bool, + opened: bool, + emitted: bool, + control_frame_read_at: Option, + poll_delay: Arc>>, + log: Arc>>, + } + + impl super::super::CommandExecutor for PollingExecutor { + fn execute( + &mut self, + command: &Command, + ) -> Result { + self.log.lock().unwrap().push(command.command_type.clone()); + if command.command_type != "session.open" { + return Ok(super::super::CommandExecution::result( + json!({"status": "completed"}), + )); + } + self.opened = true; + Ok(super::super::CommandExecution { + result: json!({"status": "completed"}), + events: vec![( + "provider.notice.recorded".to_owned(), + super::super::EventPriority::P1, + json!({"message": "accepted"}), + )], + }) + } + + fn poll_events(&mut self) -> Result, DurableRunnerError> { + if !self.opened || self.emitted { + return Ok(Vec::new()); + } + self.emitted = true; + *self.poll_delay.lock().unwrap() = + self.control_frame_read_at.map(|read_at| read_at.elapsed()); + self.log.lock().unwrap().push("poll".to_owned()); + Ok(vec![super::super::PolledEvent { + executor_event_id: "provider_event_1".to_owned(), + event_type: "provider.notice.recorded".to_owned(), + priority: super::super::EventPriority::P1, + payload: json!({"message": "started"}), + }]) + } + + fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> { + self.control_frame_read_at = Some(Instant::now()); + Ok(()) + } + + fn provider_poll_is_live(&self) -> bool { + self.live && self.opened + } + } + + let command = |seq: u64, command_type: &str| { + json!({ + "schema": "paperclip.prp.command.v1", + "commandId": format!("command_{seq}"), + "controllerSeq": seq, + "type": command_type, + "issuedAt": "2026-08-24T00:00:00.000Z", + "payload": {}, + }) + }; + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let mut config = config(port); + config.max_runtime = Duration::from_secs(5); + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-ack-poll-{}-{mode:?}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&directory); + config.state_dir = directory.clone(); + let state = test_state(&config); + let expires = current_unix_ms().unwrap() + 60_000; + let server_config = config.clone(); + let server_state = state.clone(); + let open_command = command(1, "session.open"); + let snapshot_command = command(2, "session.snapshot"); + let fast_poll = matches!(mode, AckRelease::Idle | AckRelease::WelcomeCommand); + let shutdown_seq = if fast_poll { 2 } else { 3 }; + let shutdown_command = command(shutdown_seq, "runner.shutdown"); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = accept(stream).unwrap(); + let mut secure = server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "bootstrap-secret", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: expires, + revocation_epoch: 0, + }, + ); + send_secure( + &mut socket, + &mut secure, + &server_config, + &welcome( + &server_state, + "connection_1", + Some("lease-secret"), + expires, + 0, + if mode == AckRelease::WelcomeCommand { + vec![open_command.clone()] + } else { + Vec::new() + }, + ), + ); + if mode != AckRelease::WelcomeCommand { + send_secure( + &mut socket, + &mut secure, + &server_config, + &control( + &server_state, + "connection_1", + "command", + open_command.clone(), + ), + ); + } + let open_result = receive_secure(&mut socket, &mut secure, &server_config); + assert_eq!(open_result["kind"], "command_result"); + let open_event = receive_secure(&mut socket, &mut secure, &server_config); + assert_eq!(open_event["kind"], "event"); + let mut connection = "connection_1"; + if mode == AckRelease::Replayed { + drop(socket); + let (stream, _) = listener.accept().unwrap(); + socket = accept(stream).unwrap(); + secure = server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "lease-secret", + kind: "lease", + lease_id: Some("lease_1"), + expires_at_unix_ms: expires, + revocation_epoch: 1, + }, + ); + connection = "connection_2"; + send_secure( + &mut socket, + &mut secure, + &server_config, + &welcome( + &server_state, + connection, + None, + expires, + 0, + vec![open_command], + ), + ); + assert_eq!( + receive_secure(&mut socket, &mut secure, &server_config), + open_result + ); + assert_eq!( + receive_secure(&mut socket, &mut secure, &server_config), + open_event + ); + } + let ack = control( + &server_state, + connection, + "ack", + json!({"ackedSourceSeq": 1}), + ); + let snapshot = control(&server_state, connection, "command", snapshot_command); + match mode { + AckRelease::Idle | AckRelease::WelcomeCommand => { + send_secure(&mut socket, &mut secure, &server_config, &ack) + } + AckRelease::QueuedCommand => { + for value in [ack, snapshot] { + let encrypted = secure + .encrypt(&serde_json::to_vec(&value).unwrap(), true) + .unwrap(); + socket + .write( + encode_plain_message(&encrypted, server_config.max_frame_bytes) + .unwrap(), + ) + .unwrap(); + } + socket.flush().unwrap(); + } + AckRelease::Replayed | AckRelease::NotLive => { + send_secure(&mut socket, &mut secure, &server_config, &ack); + thread::sleep(Duration::from_millis(50)); + send_secure(&mut socket, &mut secure, &server_config, &snapshot); + } + } + if !fast_poll { + let result = receive_secure(&mut socket, &mut secure, &server_config); + assert_eq!(result["kind"], "command_result"); + assert_eq!(result["payload"]["commandType"], "session.snapshot"); + } + let polled = receive_secure(&mut socket, &mut secure, &server_config); + assert_eq!(polled["kind"], "event"); + assert_eq!(polled["payload"]["payload"]["message"], "started"); + send_secure( + &mut socket, + &mut secure, + &server_config, + &control( + &server_state, + connection, + "ack", + json!({"ackedSourceSeq": 2}), + ), + ); + send_secure( + &mut socket, + &mut secure, + &server_config, + &control(&server_state, connection, "command", shutdown_command), + ); + let shutdown_result = receive_secure(&mut socket, &mut secure, &server_config); + assert_eq!(shutdown_result["kind"], "command_result"); + send_secure( + &mut socket, + &mut secure, + &server_config, + &control( + &server_state, + connection, + "command_result_ack", + json!({ + "commandId": format!("command_{shutdown_seq}"), + "commandType": "runner.shutdown", + "controllerSeq": shutdown_seq, + "status": "completed", + }), + ), + ); + }); + + let log = Arc::new(std::sync::Mutex::new(Vec::new())); + let poll_delay = Arc::new(std::sync::Mutex::new(None)); + super::super::run_durable_runner( + config, + BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(), + PollingExecutor { + live: mode != AckRelease::NotLive, + opened: false, + emitted: false, + control_frame_read_at: None, + poll_delay: poll_delay.clone(), + log: log.clone(), + }, + ) + .unwrap(); + server.join().unwrap(); + let expected: &[&str] = if fast_poll { + &["session.open", "poll", "runner.shutdown"] + } else { + &[ + "session.open", + "session.snapshot", + "poll", + "runner.shutdown", + ] + }; + assert_eq!(*log.lock().unwrap(), expected); + if fast_poll { + let delay = poll_delay.lock().unwrap().unwrap(); + assert!( + delay < RUNTIME_READ_TIMEOUT, + "the provider poll waited {delay:?} after its ACK was read" + ); + } + std::fs::remove_dir_all(directory).unwrap(); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs index ed60da24385..e64d60ddf9a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs @@ -28,6 +28,7 @@ const SECURE_FRAME_SCHEMA: &str = "paperclip.runner.secure-frame.v1"; const AUTH_TIMEOUT: Duration = Duration::from_secs(2); const WELCOME_TIMEOUT: Duration = CONNECT_TOTAL_TIMEOUT; const RUNTIME_READ_TIMEOUT: Duration = Duration::from_millis(250); +const QUEUED_CONTROL_READ_TIMEOUT: Duration = Duration::from_millis(1); const CONNECT_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(3); const CONNECT_TOTAL_TIMEOUT: Duration = Duration::from_secs(10); const MAX_CA_BUNDLE_BYTES: u64 = 4 * 1024 * 1024; @@ -1163,6 +1164,19 @@ impl AuthenticatedTransport { self.receive_json_until(None) } + pub(crate) fn receive_queued_json(&mut self) -> Result, DurableRunnerError> { + self.set_runtime_read_timeout(QUEUED_CONTROL_READ_TIMEOUT)?; + let message = self.receive_json_until(None); + self.set_runtime_read_timeout(RUNTIME_READ_TIMEOUT)?; + message + } + + fn set_runtime_read_timeout(&mut self, timeout: Duration) -> Result<(), DurableRunnerError> { + self.socket + .set_read_timeout(Some(timeout)) + .map_err(|error| DurableRunnerError::invalid(error.to_string())) + } + fn receive_json_until( &mut self, deadline: Option, @@ -1749,6 +1763,7 @@ fn hex_nibble(byte: u8) -> Result { mod tests { #[cfg(unix)] include!("tests/hosted_retention.rs"); + include!("tests/ack_release_poll.rs"); use std::net::{IpAddr, Ipv4Addr, TcpListener}; use std::path::PathBuf; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs index 5ce46394a8b..54fb7967c4f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs @@ -95,6 +95,13 @@ impl CommandExecutor for SelectedExecutor { } } + fn provider_poll_is_live(&self) -> bool { + match self { + Self::LocalFacade(executor) => executor.provider_poll_is_live(), + _ => false, + } + } + fn acknowledge_events(&mut self, count: usize) -> Result<(), DurableRunnerError> { match self { Self::LocalFacade(executor) => executor.acknowledge_events(count), @@ -454,6 +461,12 @@ impl CommandExecutor for NativeProviderCommandExecutor { .map_or(Ok(None), CommandExecutor::maintain_disconnected_provider) } + fn provider_poll_is_live(&self) -> bool { + self.selected + .as_ref() + .is_some_and(CommandExecutor::provider_poll_is_live) + } + fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> { // A provider that has not yet been selected/restored cannot have // in-process cleanup to advance. Never launch one merely for ACK debt. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index fd0ea148c9b..b94eac6ab45 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -5958,6 +5958,13 @@ impl CommandExecutor for CodexCommandExecutor { Ok(drained.next_drain_deadline) } + fn provider_poll_is_live(&self) -> bool { + self.provider.is_some() + && self.state.as_ref().is_some_and(|state| { + state.startup_attempt.is_none() && state.lifecycle != "reconciliation_required" + }) + } + fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> { if self.state.as_ref().is_some_and(|state| { state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some()