diff --git a/packages/paperclip-runner/runner/CODEX_PROVIDER.md b/packages/paperclip-runner/runner/CODEX_PROVIDER.md index 7a100aa7f32..38c36126c29 100644 --- a/packages/paperclip-runner/runner/CODEX_PROVIDER.md +++ b/packages/paperclip-runner/runner/CODEX_PROVIDER.md @@ -29,7 +29,10 @@ execution and finalization path. The provider descriptor, Codex thread ID, account session ID, active Codex turn ID, and unacknowledged normalized-event prefix are written to a private, -bounded, atomically replaced sidecar in the runner state directory. On restart, +bounded, atomically replaced sidecar in the runner state directory. runnerd +writes the sidecar once for the provider frames read in one poll, not once per +frame. The write always comes before an event, a command result, or a reply to +the provider that depends on those frames leaves runnerd. On restart, runnerd resumes that exact thread and reconciles the active turn from `thread/read`. An unexpected provider exit retains the last active turn until that reconciliation proves whether it is still running, so cancellation and a diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index 033e4fb8dfe..e67c96c0f2f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -869,6 +869,11 @@ pub struct CodexProvider { quarantined: bool, trace: Option, last_trace_frame_id: Option, + held_courtesy_responses: Option>, + #[cfg(test)] + pub(crate) sent_frame_observer: Option>, + #[cfg(test)] + pub(crate) interpretation_observer: Option>, opencode_launch_profile: Option, cursor_cloud_launch_profile: Option, provider_termination_receipt: Option, @@ -1300,6 +1305,11 @@ impl CodexProvider { quarantined: false, trace: ProviderTraceSink::from_environment(), last_trace_frame_id: None, + held_courtesy_responses: None, + #[cfg(test)] + sent_frame_observer: None, + #[cfg(test)] + interpretation_observer: None, opencode_launch_profile: opencode_launch_profile.cloned(), cursor_cloud_launch_profile: cursor_cloud_launch_profile.cloned(), provider_termination_receipt: None, @@ -1467,6 +1477,11 @@ impl CodexProvider { self.last_trace_frame_id.take() } + #[cfg(test)] + pub(crate) fn enable_trace_for_test(&mut self, trace_path: PathBuf) { + self.trace = Some(ProviderTraceSink::at_path(trace_path, 64 * 1024 * 1024)); + } + pub(crate) fn record_provider_trace_interpretation( &mut self, frame_id: u64, @@ -1475,6 +1490,10 @@ impl CodexProvider { emitted_event_ids: Vec, reason: &str, ) { + #[cfg(test)] + if let Some(observer) = self.interpretation_observer.as_mut() { + observer(&emitted_event_ids); + } if let Some(trace) = self.trace.as_mut() { trace.interpretation( frame_id, @@ -3831,10 +3850,36 @@ impl CodexProvider { { return Ok(()); } + if let Some(held) = self.held_courtesy_responses.as_mut() { + held.push(value.clone()); + return Ok(()); + } self.send_frame(value) } + pub(crate) fn hold_courtesy_responses(&mut self) { + self.held_courtesy_responses.get_or_insert_with(Vec::new); + } + + pub(crate) fn release_courtesy_responses(&mut self, keep_holding: bool) { + let held = if keep_holding { + self.held_courtesy_responses + .as_mut() + .map(std::mem::take) + .unwrap_or_default() + } else { + self.held_courtesy_responses.take().unwrap_or_default() + }; + for value in held { + let _ = self.send_frame(&value); + } + } + fn send_frame(&mut self, value: &Value) -> Result<(), LocalRunnerError> { + #[cfg(test)] + if let Some(observer) = self.sent_frame_observer.as_mut() { + observer(value); + } if let Some(trace) = self.trace.as_mut() { let raw = serde_json::to_vec(value).unwrap_or_default(); if let Some(frame_id) = trace.frame("client_to_provider", &raw) { 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 af7e43def90..fd0ea148c9b 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 @@ -79,6 +79,18 @@ fn invalid_tool_call_result( } } +fn serialize_provider_state(state: &CodexProviderState) -> Result, DurableRunnerError> { + let bytes = serde_json::to_vec_pretty(state).map_err(|error| { + DurableRunnerError::invalid(format!("failed to serialize Codex provider state: {error}")) + })?; + if bytes.len() as u64 > MAX_PROVIDER_STATE_BYTES { + return Err(DurableRunnerError::invalid( + "Codex provider state exceeds the 16 MiB limit", + )); + } + Ok(bytes) +} + fn receipt_limit_deadline_after(timeout_ms: u64) -> Result { current_unix_ms()?.checked_add(timeout_ms).ok_or_else(|| { DurableRunnerError::invalid("Codex receipt-limit interruption deadline overflowed") @@ -1973,6 +1985,13 @@ impl CodexProviderState { } } +struct NormalizationTrace { + frame_id: u64, + rule_id: String, + mapped: bool, + emitted_event_ids: Vec, +} + pub struct CodexCommandExecutor { state_dir: PathBuf, state: Option, @@ -1985,6 +2004,10 @@ pub struct CodexCommandExecutor { startup_command: Option, startup_evidence_error: Option, recovery_pending_effects: Option, + unsaved_frame_state: std::cell::Cell, + unsaved_trace_interpretations: Vec, + #[cfg(test)] + persist_count: std::cell::Cell, } impl CodexCommandExecutor { @@ -2001,6 +2024,10 @@ impl CodexCommandExecutor { startup_command: None, startup_evidence_error: None, recovery_pending_effects: None, + unsaved_frame_state: std::cell::Cell::new(false), + unsaved_trace_interpretations: Vec::new(), + #[cfg(test)] + persist_count: std::cell::Cell::new(0), } } @@ -2960,11 +2987,69 @@ impl CodexCommandExecutor { .state .as_ref() .ok_or_else(|| DurableRunnerError::invalid("Codex provider state is unavailable"))?; - self.persist_state(state) + self.persist_state(state)?; + self.unsaved_frame_state.set(false); + Ok(()) + } + + fn save_unsaved_frame_state(&self) -> Result<(), DurableRunnerError> { + if !self.unsaved_frame_state.get() { + return Ok(()); + } + let state = self + .state + .as_ref() + .ok_or_else(|| DurableRunnerError::invalid("Codex provider state is unavailable"))?; + let bytes = serialize_provider_state(state)?; + if fs::read(self.state_path()).is_ok_and(|saved| saved == bytes) { + self.unsaved_frame_state.set(false); + return Ok(()); + } + state.validate()?; + self.write_state_bytes(&bytes)?; + self.unsaved_frame_state.set(false); + Ok(()) + } + + fn record_saved_trace_interpretations(&mut self) { + let traces = std::mem::take(&mut self.unsaved_trace_interpretations); + let Some(provider) = self.provider.as_mut() else { + return; + }; + for trace in traces { + provider.record_provider_trace_interpretation( + trace.frame_id, + &trace.rule_id, + if trace.mapped { "mapped" } else { "ignored" }, + trace.emitted_event_ids, + if trace.mapped { + "Provider notification normalized into durable PRP events" + } else { + "Provider notification did not produce a durable PRP event" + }, + ); + } + } + + fn save_before_provider_writes( + &mut self, + still_polling: bool, + ) -> Result<(), DurableRunnerError> { + self.save_unsaved_frame_state()?; + self.record_saved_trace_interpretations(); + if let Some(provider) = self.provider.as_mut() { + provider.release_courtesy_responses(still_polling); + } + Ok(()) } fn persist_state(&self, state: &CodexProviderState) -> Result<(), DurableRunnerError> { state.validate()?; + let bytes = serialize_provider_state(state)?; + self.write_state_bytes(&bytes) + } + + fn write_state_bytes(&self, bytes: &[u8]) -> Result<(), DurableRunnerError> { fs::create_dir_all(&self.state_dir).map_err(|error| { DurableRunnerError::invalid(format!( "failed to create provider state directory: {error}" @@ -2980,19 +3065,9 @@ impl CodexCommandExecutor { )?; verify_private_directory(&self.state_dir)?; let path = self.state_path(); - let bytes = serde_json::to_vec_pretty(state).map_err(|error| { - DurableRunnerError::invalid(format!( - "failed to serialize Codex provider state: {error}" - )) - })?; - if bytes.len() as u64 > MAX_PROVIDER_STATE_BYTES { - return Err(DurableRunnerError::invalid( - "Codex provider state exceeds the 16 MiB limit", - )); - } let (temporary, mut file) = create_private_temporary_file(&path)?; let result = (|| -> std::io::Result<()> { - file.write_all(&bytes)?; + file.write_all(bytes)?; file.sync_all()?; drop(file); fs::rename(&temporary, &path)?; @@ -3010,6 +3085,8 @@ impl CodexCommandExecutor { fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).map_err(|error| { DurableRunnerError::invalid(format!("failed to protect provider state: {error}")) })?; + #[cfg(test)] + self.persist_count.set(self.persist_count.get() + 1); Ok(()) } @@ -4515,6 +4592,7 @@ impl CodexCommandExecutor { call_id: String, operation_id: String, ) -> Result<(), DurableRunnerError> { + self.save_before_provider_writes(true)?; let deadline_unix_ms = receipt_limit_deadline_after(RECEIPT_LIMIT_INTERRUPT_TERMINAL_DEADLINE_MS)?; let state = self @@ -4562,7 +4640,12 @@ impl CodexCommandExecutor { .expect("Codex state remains available before receipt-limit interruption") .record_receipt_limit_interrupt_attempt()?; self.save_state()?; - match self.interrupt_turn("semantic_tool_turn_receipt_limit") { + self.save_before_provider_writes(false)?; + let interrupted = self.interrupt_turn("semantic_tool_turn_receipt_limit"); + if let Some(provider) = self.provider.as_mut() { + provider.hold_courtesy_responses(); + } + match interrupted { Ok(_) => { let accepted_deadline_unix_ms = receipt_limit_deadline_after(RECEIPT_LIMIT_ACCEPTED_TERMINAL_DEADLINE_MS)?; @@ -4782,6 +4865,7 @@ impl CodexCommandExecutor { .admit_tool_call(&call_id, &operation_id, &input); match admission { Ok(ToolCallAdmission::CompletedReplay(result)) => { + self.save_before_provider_writes(true)?; self.provider .as_mut() .expect("provider remains present while handling its tool call") @@ -5113,6 +5197,25 @@ impl CodexCommandExecutor { if self.provider.is_none() { return self.settle_receipt_limit_interrupt_if_deadline_elapsed(); } + if let Some(provider) = self.provider.as_mut() { + provider.hold_courtesy_responses(); + } + let polled = self.poll_provider_frames(allow_control, receipt_limit_terminal_poll); + let saved = self.save_before_provider_writes(false); + polled?; + saved?; + // Check the durable deadline only after a complete provider poll. A + // terminal that arrived after interruption acceptance but before this + // observation remains authoritative even when several fast controller + // polls have already exhausted the interruption-attempt budget. + self.settle_receipt_limit_interrupt_if_deadline_elapsed() + } + + fn poll_provider_frames( + &mut self, + allow_control: bool, + receipt_limit_terminal_poll: bool, + ) -> Result<(), DurableRunnerError> { for _ in 0..MAX_EVENTS_PER_POLL { let provider = self .provider @@ -5175,6 +5278,7 @@ impl CodexCommandExecutor { ))?; // Commit the authoritative failure before best-effort provider cleanup. self.save_state()?; + self.record_saved_trace_interpretations(); if let Some(mut provider) = self.provider.take() { if let Some(frame_id) = trace_frame_id { provider.record_provider_trace_interpretation( @@ -5225,7 +5329,7 @@ impl CodexCommandExecutor { "method": method.chars().take(128).collect::(), "receivedThreadId": child, "expectedThreadId": root_thread, "receivedTurnId": child_turn, "expectedTurnId": root_turn }), }])?; - self.save_state()?; + self.unsaved_frame_state.set(true); if let (Some(frame_id), Some(provider)) = (trace_frame_id, self.provider.as_mut()) { @@ -5245,6 +5349,9 @@ impl CodexCommandExecutor { }; let normalized_terminal_type = normalized_codex_terminal_event_type(&method, ¶ms); + if normalized_terminal_type.is_some() { + self.save_before_provider_writes(true)?; + } let cursor_cloud_delivery = if self .state .as_ref() @@ -5486,12 +5593,16 @@ impl CodexCommandExecutor { .collect::>() }) .unwrap_or_default(); - self.save_state()?; - if self - .state - .as_ref() - .is_some_and(|state| state.lifecycle == "reconciliation_required") + let reconciliation_required = state.lifecycle == "reconciliation_required"; + if terminal_event_type.is_some() + || cursor_cloud_delivery.is_some() + || reconciliation_required { + self.save_state()?; + } else { + self.unsaved_frame_state.set(true); + } + if reconciliation_required { break; } if let (Some(delivery), Some(provider)) = @@ -5499,24 +5610,16 @@ impl CodexCommandExecutor { { let _ = provider.acknowledge_cursor_cloud_terminal(delivery); } - if let (Some(frame_id), Some(provider)) = - (trace_frame_id, self.provider.as_mut()) - { - provider.record_provider_trace_interpretation( + if let Some(frame_id) = trace_frame_id { + self.unsaved_trace_interpretations.push(NormalizationTrace { frame_id, - &format!("codex.normalize.{}", method.replace('/', ".")), - if normalized_event_count > 0 { - "mapped" - } else { - "ignored" - }, - trace_emitted_event_ids, - if normalized_event_count > 0 { - "Provider notification normalized into durable PRP events" - } else { - "Provider notification did not produce a durable PRP event" - }, - ); + rule_id: format!("codex.normalize.{}", method.replace('/', ".")), + mapped: normalized_event_count > 0, + emitted_event_ids: trace_emitted_event_ids, + }); + } + if !self.unsaved_frame_state.get() { + self.record_saved_trace_interpretations(); } } CodexProviderEvent::McpPermissionRequest { request_id, params } => { @@ -5582,6 +5685,8 @@ impl CodexCommandExecutor { process_generation, completed_turn_process_generation, } => { + self.save_unsaved_frame_state()?; + self.record_saved_trace_interpretations(); self.provider = None; if !success { let state = self @@ -5624,16 +5729,13 @@ impl CodexCommandExecutor { } } } - // Check the durable deadline only after a complete provider poll. A - // terminal that arrived after interruption acceptance but before this - // observation remains authoritative even when several fast controller - // polls have already exhausted the interruption-attempt budget. - self.settle_receipt_limit_interrupt_if_deadline_elapsed() + Ok(()) } } impl CommandExecutor for CodexCommandExecutor { fn host_park_ready(&mut self) -> Result { + self.save_before_provider_writes(false)?; let Some(state) = self.state.as_ref() else { return Ok(false); }; @@ -5701,6 +5803,7 @@ impl CommandExecutor for CodexCommandExecutor { command_type: command.command_type.clone(), }); let outcome = (|| { + self.save_before_provider_writes(false)?; self.restore()?; if self .state @@ -5805,6 +5908,7 @@ impl CommandExecutor for CodexCommandExecutor { if let Some(error) = &self.startup_evidence_error { return Err(error.clone()); } + self.save_before_provider_writes(false)?; Ok(self .state .as_ref() @@ -9312,6 +9416,10 @@ done #[path = "tests/provider_api_validation.rs"] mod api_validation_tests; +#[cfg(test)] +#[path = "tests/provider_frame_save_batching.rs"] +mod frame_save_batching_tests; + #[cfg(test)] #[path = "tests/provider_assigned_tool_validation.rs"] mod assigned_tool_validation_tests; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/tests/provider_frame_save_batching.rs b/packages/paperclip-runner/runner/crates/runner-core/src/tests/provider_frame_save_batching.rs new file mode 100644 index 00000000000..a41aa1047ca --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/tests/provider_frame_save_batching.rs @@ -0,0 +1,579 @@ +use super::*; + +fn frame_test_state() -> CodexProviderState { + let mut state = CodexProviderState::new( + CodexProviderConfig { + provider: "opencode".to_owned(), + driver: "opencode_server".to_owned(), + provider_version: "1.18.34".to_owned(), + command: PathBuf::from("node"), + args: Vec::new(), + cwd: std::env::current_dir() + .unwrap() + .to_string_lossy() + .into_owned(), + model: Some("openrouter/model".to_owned()), + provider_session_id: None, + instructions: String::new(), + approval_policy: "never".to_owned(), + externally_sandboxed: false, + read_only: false, + include_skill_instructions: None, + conversation_mode: None, + }, + Some(CompletionContractBinding { + revision: "revision-1".to_owned(), + criterion_ids: vec!["criterion-1".to_owned()], + }), + ProviderToolBridge::default(), + ); + state.thread_id = Some("thread-1".to_owned()); + state.active_provider_turn_id = Some("turn-1".to_owned()); + state.lifecycle = "turn_active".to_owned(); + state +} + +fn streamed_task_context_operation() -> crate::provider_bridge::AuthorizedTool { + crate::provider_bridge::AuthorizedTool { + operation_id: "get_task_context".to_owned(), + version: 1, + description: "Read the active task context.".to_owned(), + input_schema: json!({"type": "object"}), + response_schema: json!({"type": "object"}), + } +} + +#[cfg(unix)] +fn streamed_turn_before_tool_call( + frames: usize, + retained_results: usize, + tool: &str, + wait_for_stream: bool, +) -> (PathBuf, CodexCommandExecutor) { + let tail = format!( + r#"{{"id":"tool-1","method":"item/tool/call","params":{{"threadId":"thread-1","turnId":"turn-1","callId":"call-streamed","tool":"{tool}","arguments":{{}}}}}}"# + ); + streamed_turn(frames, retained_results, &tail, wait_for_stream, |_| {}) +} + +#[cfg(unix)] +fn streamed_turn( + frames: usize, + retained_results: usize, + tail: &str, + wait_for_stream: bool, + prepare_provider: impl FnOnce(&mut CodexProvider), +) -> (PathBuf, CodexCommandExecutor) { + streamed_turn_then( + frames, + retained_results, + tail, + ":", + wait_for_stream, + prepare_provider, + ) +} + +#[cfg(unix)] +fn streamed_turn_then( + frames: usize, + retained_results: usize, + tail: &str, + after: &str, + wait_for_stream: bool, + prepare_provider: impl FnOnce(&mut CodexProvider), +) -> (PathBuf, CodexCommandExecutor) { + let directory = std::env::temp_dir().join(format!( + "paperclip-streamed-frames-{}", + uuid::Uuid::new_v4() + )); + fs::create_dir_all(&directory).unwrap(); + let streamed = directory.join("streamed"); + let script = r#" +while IFS= read -r line; do + case "$line" in + *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{}}' ;; + *'"method":"thread/start"'*) printf '%s\n' '{"id":2,"result":{"thread":{"id":"thread-1"}}}' ;; + *'"method":"turn/start"'*) + printf '%s\n' '{"id":3,"result":{"turn":{"id":"turn-1"}}}' + printf '%s\n' '{"method":"turn/started","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"inProgress"}}}' + i=0 + while [ "$i" -lt FRAMES ]; do + printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","itemId":"message-1","delta":"x"}}' + printf '%s\n' '{"method":"item/updated","params":{"threadId":"thread-1","turnId":"turn-1","item":{"id":"reasoning-1","type":"reasoning","summary":[],"content":[]}}}' + i=$((i+1)) + done + printf '%s\n' 'TAIL' + touch 'STREAMED' + AFTER ;; + esac +done +"# + .replace("FRAMES", &frames.to_string()) + .replace("TAIL", tail) + .replace("AFTER", after) + .replace("STREAMED", &streamed.display().to_string()); + let operation = streamed_task_context_operation(); + let mut bridge = ProviderToolBridge::default(); + bridge + .prepare(AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation)) + .unwrap(), + operations: vec![operation.clone()], + }) + .unwrap(); + for index in 0..retained_results { + let call_id = format!("retained-{index}"); + bridge + .begin_call(call_id.clone(), "get_task_context".to_owned(), json!({})) + .unwrap(); + bridge + .apply_result(ToolResult { + call_id, + operation_id: "get_task_context".to_owned(), + result: json!({"body": "x".repeat(10_000)}), + is_error: false, + }) + .unwrap(); + } + let mut state = frame_test_state(); + state.config.provider = "codex".to_owned(); + state.config.driver = "codex_app_server".to_owned(); + state.config.command = PathBuf::from("/bin/sh"); + state.config.args = vec!["-c".to_owned(), script]; + state.config.model = None; + state.tool_bridge = bridge; + let mut provider = CodexProvider::start_with_tools(&state.config, [operation], None).unwrap(); + prepare_provider(&mut provider); + provider.start_turn("Test", &state.config.cwd).unwrap(); + state.provider_process_generation = provider.process_generation(); + let mut executor = CodexCommandExecutor::new(&directory); + executor.event_identity = Some(ProviderEventIdentity { + runner_instance_id: "runner-1".to_owned(), + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + }); + executor.state = Some(state); + executor.provider = Some(provider); + executor.restore_checked = true; + executor.save_state().unwrap(); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + while wait_for_stream && !streamed.exists() { + assert!(std::time::Instant::now() < deadline); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + (directory, executor) +} + +fn poll_until_semantic_tool_call(executor: &mut CodexCommandExecutor) -> Vec { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + let mut observed = Vec::new(); + loop { + assert!(std::time::Instant::now() < deadline); + let events = executor.poll_events().unwrap(); + let count = events.len(); + let delivered = events + .iter() + .any(|event| event.event_type == "semantic_tool.input"); + observed.extend(events); + if delivered { + return observed; + } + executor.acknowledge_events(count).unwrap(); + } +} + +#[cfg(unix)] +#[test] +fn streamed_frames_do_not_multiply_durable_saves_before_a_tool_call() { + let mut saves = Vec::new(); + for frames in [8, 48] { + let (directory, mut executor) = + streamed_turn_before_tool_call(frames, 4, "get_task_context", true); + let before = executor.persist_count.get(); + let observed = poll_until_semantic_tool_call(&mut executor); + saves.push(executor.persist_count.get() - before); + assert_eq!( + observed + .iter() + .filter(|event| event.event_type == "item.delta") + .count(), + frames + ); + executor.provider.as_mut().unwrap().shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } + eprintln!("durable saves before the tool call: {saves:?}"); + assert_eq!( + saves[0], saves[1], + "saves grew with streamed frames: {saves:?}" + ); +} + +fn saved_provider_state(executor: &CodexCommandExecutor) -> CodexProviderState { + serde_json::from_slice(&fs::read(executor.state_path()).unwrap()).unwrap() +} + +#[cfg(unix)] +#[test] +fn batched_frame_state_is_saved_before_events_leave_and_survives_a_restart() { + let (directory, mut executor) = + streamed_turn_before_tool_call(300, 4, "get_task_context", false); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + assert!(std::time::Instant::now() < deadline); + let events = executor.poll_events().unwrap(); + assert!(!executor.unsaved_frame_state.get()); + let saved = saved_provider_state(&executor); + assert_eq!( + serde_json::to_value(&saved).unwrap(), + serde_json::to_value(executor.state.as_ref().unwrap()).unwrap() + ); + assert_eq!(saved.pending_events, events); + if events + .iter() + .any(|event| event.event_type == "semantic_tool.input") + { + break; + } + let count = events.len(); + executor.acknowledge_events(count).unwrap(); + } + let live = executor.state.clone().unwrap(); + executor.provider.as_mut().unwrap().shutdown().unwrap(); + drop(executor); + + let mut resumed = CodexCommandExecutor::new(&directory); + resumed.load_state_without_provider().unwrap(); + let resumed_state = resumed.state.as_ref().unwrap(); + assert_eq!( + serde_json::to_value(resumed_state).unwrap(), + serde_json::to_value(&live).unwrap() + ); + assert!(resumed_state + .tool_bridge + .pending_calls() + .any(|call| call.call_id == "call-streamed")); + assert_eq!( + resumed.retained_events().unwrap(), + live.pending_events.iter().cloned().collect::>() + ); + fs::remove_dir_all(directory).unwrap(); +} + +#[cfg(unix)] +#[test] +fn batched_frame_state_is_saved_when_a_later_frame_fails() { + let (directory, mut executor) = + streamed_turn_before_tool_call(20, 0, "unauthorized_tool", true); + assert!(executor.poll_events().is_err()); + assert!(!executor.unsaved_frame_state.get()); + let saved = saved_provider_state(&executor); + assert_eq!( + serde_json::to_value(&saved).unwrap(), + serde_json::to_value(executor.state.as_ref().unwrap()).unwrap() + ); + assert_eq!( + saved + .pending_events + .iter() + .filter(|event| event.event_type == "item.delta") + .count(), + 20 + ); + let _ = executor.provider.as_mut().unwrap().shutdown(); + fs::remove_dir_all(directory).unwrap(); +} + +fn previous_release_state() -> CodexProviderState { + let mut state = frame_test_state(); + state.config.provider = "codex".to_owned(); + state.config.driver = "codex_app_server".to_owned(); + state.config.model = None; + state +} + +#[test] +fn previous_release_state_is_rewritten_by_the_first_batched_save() { + let directory = std::env::temp_dir().join(format!( + "paperclip-batched-save-upgrade-{}", + uuid::Uuid::new_v4() + )); + fs::create_dir_all(&directory).unwrap(); + #[cfg(unix)] + fs::set_permissions(&directory, fs::Permissions::from_mode(0o700)).unwrap(); + let mut previous = serde_json::to_value(previous_release_state()).unwrap(); + let object = previous.as_object_mut().unwrap(); + assert!(object.remove("activeTerminalToolRejected").is_some()); + assert!(object.remove("descendantThreadIds").is_some()); + let previous_bytes = serde_json::to_vec(&previous).unwrap(); + let path = directory.join(CODEX_PROVIDER_STATE_FILE); + fs::write(&path, &previous_bytes).unwrap(); + #[cfg(unix)] + fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap(); + + let mut resumed = CodexCommandExecutor::new(&directory); + resumed.load_state_without_provider().unwrap(); + assert_eq!(resumed.state.as_ref().unwrap(), &previous_release_state()); + resumed.unsaved_frame_state.set(true); + resumed.save_unsaved_frame_state().unwrap(); + assert_eq!(resumed.persist_count.get(), 1); + assert_ne!(fs::read(&path).unwrap(), previous_bytes); + assert_eq!( + serde_json::to_value(saved_provider_state(&resumed)).unwrap(), + serde_json::to_value(resumed.state.as_ref().unwrap()).unwrap() + ); + + resumed.unsaved_frame_state.set(true); + resumed.save_unsaved_frame_state().unwrap(); + assert_eq!(resumed.persist_count.get(), 1); + + resumed.state.as_mut().unwrap().last_agent_message = Some("streamed".to_owned()); + resumed.unsaved_frame_state.set(true); + resumed.save_unsaved_frame_state().unwrap(); + assert_eq!(resumed.persist_count.get(), 2); + assert_eq!( + saved_provider_state(&resumed).last_agent_message.as_deref(), + Some("streamed") + ); + fs::remove_dir_all(directory).unwrap(); +} + +#[cfg(unix)] +#[test] +#[ignore = "timing measurement; run with --ignored --nocapture"] +fn measure_streamed_frame_cost_before_a_tool_call() { + for retained in [0, 20, 80] { + for frames in [50, 200] { + let (directory, mut executor) = + streamed_turn_before_tool_call(frames, retained, "get_task_context", false); + let state_bytes = fs::metadata(executor.state_path()).unwrap().len(); + let before = executor.persist_count.get(); + let started = std::time::Instant::now(); + poll_until_semantic_tool_call(&mut executor); + let elapsed = started.elapsed(); + eprintln!( + "retained={retained} stateBytes={state_bytes} frames={} saves={} toolCallLatencyMs={:.1}", + frames * 2, + executor.persist_count.get() - before, + elapsed.as_secs_f64() * 1000.0 + ); + executor.provider.as_mut().unwrap().shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } + } +} + +#[cfg(unix)] +fn event_identity_for_tests() -> ProviderEventIdentity { + ProviderEventIdentity { + runner_instance_id: "runner-1".to_owned(), + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + } +} + +#[cfg(unix)] +fn read_saved_state(path: &std::path::Path) -> CodexProviderState { + serde_json::from_slice(&fs::read(path).unwrap()).unwrap() +} + +#[cfg(unix)] +#[test] +fn courtesy_rejections_reach_the_provider_only_after_earlier_frames_are_saved() { + let directory_slot = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let saved_at_rejection = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let observer_directory = std::sync::Arc::clone(&directory_slot); + let observer_saved = std::sync::Arc::clone(&saved_at_rejection); + let tail = r#"{"id":"courtesy-1","method":"paperclip/test/unsupported","params":{"threadId":"thread-1","turnId":"turn-1"}}"#; + let (directory, mut executor) = streamed_turn(20, 0, tail, true, move |provider| { + provider.sent_frame_observer = Some(Box::new(move |frame: &Value| { + if frame["id"] != "courtesy-1" { + return; + } + let directory = observer_directory.lock().unwrap().clone().unwrap(); + *observer_saved.lock().unwrap() = + Some(read_saved_state(&directory.join(CODEX_PROVIDER_STATE_FILE))); + })); + }); + *directory_slot.lock().unwrap() = Some(directory.clone()); + let events = executor.poll_events().unwrap(); + assert!(events.iter().any(|event| event.event_type == "item.delta")); + + let resumed = saved_at_rejection + .lock() + .unwrap() + .take() + .expect("runnerd sent the courtesy rejection"); + assert_eq!( + resumed + .pending_events + .iter() + .filter(|event| event.event_type == "item.delta") + .count(), + 20 + ); + assert_eq!( + serde_json::to_value(&resumed).unwrap(), + serde_json::to_value(executor.state.as_ref().unwrap()).unwrap() + ); + executor.provider.as_mut().unwrap().shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + +#[cfg(unix)] +#[test] +fn trace_interpretations_name_only_saved_events() { + let directory_slot = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let checked = std::sync::Arc::new(std::sync::Mutex::new((0usize, Vec::::new()))); + let observer_directory = std::sync::Arc::clone(&directory_slot); + let observer_checked = std::sync::Arc::clone(&checked); + let trace_path = std::env::temp_dir().join(format!( + "paperclip-frame-trace-{}.jsonl", + uuid::Uuid::new_v4() + )); + let provider_trace_path = trace_path.clone(); + let (directory, mut executor) = streamed_turn_before_tool_call_with(40, move |provider| { + provider.enable_trace_for_test(provider_trace_path); + provider.interpretation_observer = Some(Box::new(move |ids: &[String]| { + if ids.is_empty() { + return; + } + let directory = observer_directory.lock().unwrap().clone().unwrap(); + let saved = read_saved_state(&directory.join(CODEX_PROVIDER_STATE_FILE)); + let identity = event_identity_for_tests(); + let saved_ids = (0..saved.next_provider_event_seq) + .map(|sequence| identity.source_event_id(&provider_event_id(sequence))) + .collect::>(); + let mut checked = observer_checked.lock().unwrap(); + checked.0 += ids.len(); + checked + .1 + .extend(ids.iter().filter(|id| !saved_ids.contains(*id)).cloned()); + })); + }); + *directory_slot.lock().unwrap() = Some(directory.clone()); + poll_until_semantic_tool_call(&mut executor); + let (count, unsaved) = checked.lock().unwrap().clone(); + assert!(count >= 40, "trace named {count} events"); + assert!( + unsaved.is_empty(), + "trace named unsaved events: {unsaved:?}" + ); + executor.provider.as_mut().unwrap().shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + let _ = fs::remove_file(trace_path); +} + +#[cfg(unix)] +#[test] +fn a_crash_inside_an_unsaved_batch_leaves_saved_state_and_provider_output_unchanged() { + let sent = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); + let observer_sent = std::sync::Arc::clone(&sent); + let tail = r#"{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","itemId":"message-1","delta":"last"}}"#; + let (directory, mut executor) = streamed_turn(60, 0, tail, true, move |provider| { + provider.sent_frame_observer = Some(Box::new(move |frame: &Value| { + observer_sent.lock().unwrap().push(frame.clone()); + })); + }); + let path = executor.state_path(); + let saved_before = fs::read(&path).unwrap(); + let sent_before = sent.lock().unwrap().len(); + executor + .provider + .as_mut() + .unwrap() + .hold_courtesy_responses(); + executor.poll_provider_frames(true, false).unwrap(); + assert!(executor.unsaved_frame_state.get()); + let live_next = executor.state.as_ref().unwrap().next_provider_event_seq; + let saved_next = read_saved_state(&path).next_provider_event_seq; + assert!(live_next >= saved_next + 121); + assert_eq!(fs::read(&path).unwrap(), saved_before); + assert_eq!(sent.lock().unwrap().len(), sent_before); + executor.provider.as_mut().unwrap().shutdown().unwrap(); + drop(executor); + + let mut resumed = CodexCommandExecutor::new(&directory); + resumed.load_state_without_provider().unwrap(); + let resumed_state = resumed.state.as_ref().unwrap(); + assert_eq!(resumed_state.next_provider_event_seq, saved_next); + assert!(resumed_state.pending_events.is_empty()); + fs::remove_dir_all(directory).unwrap(); +} + +#[cfg(unix)] +fn streamed_turn_before_tool_call_with( + frames: usize, + prepare_provider: impl FnOnce(&mut CodexProvider), +) -> (PathBuf, CodexCommandExecutor) { + let tail = r#"{"id":"tool-1","method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"call-streamed","tool":"get_task_context","arguments":{}}}"#; + streamed_turn(frames, 0, tail, true, prepare_provider) +} + +#[cfg(unix)] +#[test] +fn a_held_cancellation_that_cannot_reach_the_provider_keeps_the_completed_turn() { + let tail = r#"{"id":"tool-1","method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"call-streamed","tool":"get_task_context","arguments":{}}}"#; + let after = r#"exec 0<&-; printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'; sleep 2; exit 0"#; + let (directory, mut executor) = streamed_turn_then(10, 0, tail, after, true, |_| {}); + std::thread::sleep(std::time::Duration::from_millis(200)); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + let mut observed = Vec::new(); + while !observed + .iter() + .any(|event: &PolledEvent| event.event_type == "turn.completed") + { + assert!(std::time::Instant::now() < deadline); + let events = executor.poll_events().unwrap(); + let count = events.len(); + observed.extend(events); + executor.acknowledge_events(count).unwrap(); + } + assert!(observed + .iter() + .any(|event| event.event_type == "semantic_tool.input")); + let saved = read_saved_state(&executor.state_path()); + assert!(saved.completed_turn_authoritative); + if let Some(provider) = executor.provider.as_mut() { + let _ = provider.shutdown(); + } + fs::remove_dir_all(directory).unwrap(); +} + +#[cfg(unix)] +#[test] +fn trace_interpretations_survive_a_provider_exit_in_the_same_poll() { + let named = std::sync::Arc::new(std::sync::Mutex::new(0usize)); + let observer_named = std::sync::Arc::clone(&named); + let tail = r#"{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","itemId":"message-1","delta":"last"}}"#; + let trace_path = std::env::temp_dir().join(format!( + "paperclip-frame-trace-exit-{}.jsonl", + uuid::Uuid::new_v4() + )); + let provider_trace_path = trace_path.clone(); + let (directory, mut executor) = + streamed_turn_then(20, 0, tail, "exit 0", true, move |provider| { + provider.enable_trace_for_test(provider_trace_path); + provider.interpretation_observer = Some(Box::new(move |ids: &[String]| { + *observer_named.lock().unwrap() += ids.len(); + })); + }); + std::thread::sleep(std::time::Duration::from_millis(200)); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + while executor.provider.is_some() { + assert!(std::time::Instant::now() < deadline); + let events = executor.poll_events().unwrap(); + let count = events.len(); + executor.acknowledge_events(count).unwrap(); + } + assert!(*named.lock().unwrap() >= 41); + fs::remove_dir_all(directory).unwrap(); + let _ = fs::remove_file(trace_path); +}