From cf6db7b523cd6ce6cbace5d6926eb13f91d39efa Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 30 Aug 2026 07:56:17 -0500 Subject: [PATCH] feat(runner): bridge Codex dynamic tools (#12382) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The native runner keeps provider execution separate from Paperclip tool authority. > - The Rust authorization bridge can now validate a run-scoped semantic tool catalog. > - Codex still needs a bounded protocol adapter for that validated catalog. > - The adapter must advertise the same catalog after provider recovery. > - The adapter must reject unadvertised calls and mismatched results. > - This pull request adds that Codex-only transport boundary. > - The benefit is a fail-closed provider bridge that remains inactive until durable wiring supplies tools. ## Linked Issues or Issue Description **Subsystem affected** `packages/paperclip-runner` and its Rust Codex app-server provider. **Problem or motivation** The native runner has a validated semantic tool catalog, but the Rust Codex provider cannot project that catalog or correlate Codex tool calls with Paperclip results. A resumed Codex thread also needs the same run-scoped catalog. **Proposed solution** Add an explicit Codex start path that accepts validated tools. Send those tools through `dynamicTools` for both thread start and thread resume. Validate thread, turn, call, operation, result identity, and size before data crosses the provider boundary. **Roadmap alignment** This work supports the shipped governed MCP Tool Gateway and self-healing run milestones. It does not add a user-facing adapter or enable the experimental runner. ## What Changed - Add a Codex dynamic-tool projection for explicit authorized tool sets. - Advertise the same tool set on new and resumed provider threads. - Correlate bounded Codex tool calls and Paperclip semantic results. - Reject calls outside the active thread or turn, unadvertised tools, reused request identities, mismatched results, and oversized values. - Keep the current durable backend on the zero-tool path until the next wiring change. - Extend the fake Codex app server and integration tests for success, recovery, denial, and correlation behavior. ## Verification - `cargo test --manifest-path packages/paperclip-runner/runner/Cargo.toml --workspace` - `cargo clippy --manifest-path packages/paperclip-runner/runner/Cargo.toml --workspace --all-targets -- -D warnings -A clippy::manual_is_multiple_of -A clippy::filter_map_bool_then` - `pnpm -r typecheck` - `pnpm build` - The PR changes 4 files relative to `runner-rust-semantic-tool-bridge`. ## Risks Low activation risk. The existing `CodexProvider::start` path still supplies an empty tool set. The durable backend treats a tool event as an error until a later PR attaches the durable authorization bridge. The new transport rejects unknown tools and invalid provider bindings. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI Codex with GPT-5 and repository tool use. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --- .../src/bin/fake-codex-app-server.rs | 309 +++- .../crates/runner-core/src/codex_provider.rs | 765 ++++++++- .../runner-core/src/process_supervisor.rs | 8 +- .../runner-core/src/provider_backend.rs | 306 +++- .../runner-core/tests/codex_provider.rs | 1368 +++++++++++++++++ 5 files changed, 2675 insertions(+), 81 deletions(-) diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index 2284ea25a2..1095e28495 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -2,6 +2,8 @@ use std::fs::{self, OpenOptions}; use std::io::{self, BufRead, Write}; use std::path::{Path, PathBuf}; use std::process::ExitCode; +use std::thread; +use std::time::Duration; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; @@ -47,6 +49,19 @@ fn log_call(path: Option<&Path>, method: &str) -> io::Result<()> { writeln!(file, "{method}") } +fn has_task_context_tool(message: &Value) -> bool { + message + .pointer("/params/dynamicTools") + .and_then(Value::as_array) + .is_some_and(|tools| { + tools.iter().any(|tool| { + tool.get("name").and_then(Value::as_str) == Some("get_task_context") + && tool.get("description").and_then(Value::as_str) == Some("Read task context.") + && tool.pointer("/inputSchema/type").and_then(Value::as_str) == Some("object") + }) + }) +} + fn finish_turn(state_path: &Path, state: &mut FakeState, status: &str) -> io::Result<()> { let turn_id = state .active_turn_id @@ -79,18 +94,110 @@ fn finish_turn(state_path: &Path, state: &mut FakeState, status: &str) -> io::Re save_state(state_path, state) } +fn emit_ambiguous_turn_evidence( + state_path: &Path, + state: &mut FakeState, + emit_turn_started: bool, + conflicting_identity: bool, +) -> io::Result<()> { + let turn_id = state + .active_turn_id + .clone() + .unwrap_or_else(|| "provider-turn-2".to_owned()); + if emit_turn_started { + send(json!({ + "method": "turn/started", + "params": {"turn": {"id": turn_id}} + }))?; + } + if conflicting_identity { + send(json!({ + "method": "turn/completed", + "params": {"turn": {"id": "provider-turn-conflict", "status": "completed"}} + })) + } else { + finish_turn(state_path, state, "completed") + } +} + fn run() -> Result<(), Box> { let args = std::env::args().skip(1).collect::>(); let state_path = PathBuf::from(argument(&args, "--state-file").ok_or("--state-file is required")?); let call_log = argument(&args, "--call-log").map(PathBuf::from); let emit_question = args.iter().any(|value| value == "--emit-question"); + let emit_tool_call = args.iter().any(|value| value == "--emit-tool-call"); + let replay_completed_tool_call = args + .iter() + .any(|value| value == "--replay-completed-tool-call"); + let complete_after_tool_call = args + .iter() + .any(|value| value == "--complete-after-tool-call"); + let exit_after_tool_call_completion = args + .iter() + .any(|value| value == "--exit-after-tool-call-completion"); + let require_dynamic_tool = args.iter().any(|value| value == "--require-dynamic-tool"); let hold_turn = args.iter().any(|value| value == "--hold-turn"); let exit_after_turn_start = args.iter().any(|value| value == "--exit-after-turn-start"); + let exit_after_turn_completion = args + .iter() + .any(|value| value == "--exit-after-turn-completion"); + let emit_post_completion_warning = args + .iter() + .any(|value| value == "--emit-post-completion-warning"); + let fail_after_turn_completion = args + .iter() + .any(|value| value == "--fail-after-turn-completion"); + let fail_after_second_turn_start = args + .iter() + .any(|value| value == "--fail-after-second-turn-start"); + let reject_second_turn_start = args + .iter() + .any(|value| value == "--reject-second-turn-start"); + let emit_turn_before_rejected_second_start = args + .iter() + .any(|value| value == "--emit-turn-before-rejected-second-start"); + let malformed_error_second_turn_start = args + .iter() + .any(|value| value == "--malformed-error-second-turn-start"); + let missing_id_second_turn_start = args + .iter() + .any(|value| value == "--missing-id-second-turn-start"); + let fail_after_accepting_second_turn_before_response = args + .iter() + .any(|value| value == "--fail-after-accepting-second-turn-before-response"); + let exit_after_accepting_second_turn_before_response = args + .iter() + .any(|value| value == "--exit-after-accepting-second-turn-before-response"); + let complete_ambiguous_second_turn = args + .iter() + .any(|value| value == "--complete-ambiguous-second-turn"); + let retain_ambiguous_second_turn_active = args + .iter() + .any(|value| value == "--retain-ambiguous-second-turn-active"); + let hold_ambiguous_second_turn_after_item = args + .iter() + .any(|value| value == "--hold-ambiguous-second-turn-after-item"); + let complete_ambiguous_second_turn_before_response = args + .iter() + .any(|value| value == "--complete-ambiguous-second-turn-before-response"); + let conflicting_ambiguous_second_turn = args + .iter() + .any(|value| value == "--conflicting-ambiguous-second-turn"); + let omit_ambiguous_turn_started = args + .iter() + .any(|value| value == "--omit-ambiguous-turn-started"); + let fail_after_thread_read = args.iter().any(|value| value == "--fail-after-thread-read"); + let exit_after_thread_read = args.iter().any(|value| value == "--exit-after-thread-read"); + let fail_after_turn_completion_delay_ms = + argument(&args, "--fail-after-turn-completion-delay-ms") + .map(|value| value.parse::()) + .transpose()?; let pre_response_notification = args .iter() .any(|value| value == "--notification-before-response"); let mut state = load_state(&state_path); + let mut turn_start_count = 0_u64; for line in io::stdin().lock().lines() { let message: Value = serde_json::from_str(&line?)?; @@ -99,6 +206,42 @@ fn run() -> Result<(), Box> { finish_turn(&state_path, &mut state, "completed")?; continue; } + if message.get("method").is_none() && message.get("id") == Some(&json!("tool-request-1")) { + if message.pointer("/result/success") == Some(&json!(false)) { + log_call(call_log.as_deref(), "tool-response:failure")?; + if state.active_turn_id.is_some() { + finish_turn(&state_path, &mut state, "failed")?; + } + continue; + } + if message.pointer("/result/success") != Some(&json!(true)) { + return Err("semantic tool response omitted success".into()); + } + let text = message + .pointer("/result/contentItems/0/text") + .and_then(Value::as_str) + .ok_or("semantic tool response omitted content text")?; + let result: Value = serde_json::from_str(text)?; + if result != json!({"ok": true, "task": {"id": "task-1"}}) { + return Err("semantic tool response changed the operation result".into()); + } + if replay_completed_tool_call { + send(json!({ + "id": "tool-request-replay", + "method": "item/tool/call", + "params": { + "threadId": state.thread_id, + "turnId": state.active_turn_id, + "callId": "semantic-call-1", + "tool": "get_task_context", + "arguments": {} + } + }))?; + continue; + } + finish_turn(&state_path, &mut state, "completed")?; + continue; + } let Some(method) = message.get("method").and_then(Value::as_str) else { continue; }; @@ -111,6 +254,9 @@ fn run() -> Result<(), Box> { }))?, "initialized" => {} "thread/start" => { + if require_dynamic_tool && !has_task_context_tool(&message) { + return Err("thread/start omitted the authorized dynamic tool".into()); + } state.thread_id = "codex-thread-1".to_owned(); state.active_turn_id = None; save_state(&state_path, &state)?; @@ -125,10 +271,15 @@ fn run() -> Result<(), Box> { "result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}} }))?; } - "thread/resume" => send(json!({ - "id": id, - "result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}} - }))?, + "thread/resume" => { + if require_dynamic_tool && !has_task_context_tool(&message) { + return Err("thread/resume omitted the authorized dynamic tool".into()); + } + send(json!({ + "id": id, + "result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}} + }))?; + } "thread/read" => { let turns = state .active_turn_id @@ -139,10 +290,113 @@ fn run() -> Result<(), Box> { "id": id, "result": {"thread": {"id": state.thread_id, "turns": turns}} }))?; + if fail_after_thread_read { + return Err("configured failure after thread read".into()); + } else if exit_after_thread_read { + return Ok(()); + } } "turn/start" => { - state.active_turn_id = Some("provider-turn-1".to_owned()); + turn_start_count += 1; + if reject_second_turn_start && turn_start_count == 2 { + send(json!({ + "method": "warning", + "params": {"message": "buffered before replacement rejection"} + }))?; + if emit_turn_before_rejected_second_start { + send(json!({ + "method": "turn/started", + "params": {"turn": {"id": "provider-turn-contradiction"}} + }))?; + } + send(json!({ + "id": id, + "error": {"code": -32000, "message": "replacement turn rejected"} + }))?; + return Err("configured failure after second turn rejection".into()); + } + let emits_ambiguous_turn_evidence = turn_start_count == 2 + && (complete_ambiguous_second_turn + || complete_ambiguous_second_turn_before_response + || conflicting_ambiguous_second_turn); + let provider_turn_id = if emits_ambiguous_turn_evidence + || (turn_start_count == 2 + && (retain_ambiguous_second_turn_active + || hold_ambiguous_second_turn_after_item)) + { + "provider-turn-2" + } else { + "provider-turn-1" + }; + state.active_turn_id = Some(provider_turn_id.to_owned()); save_state(&state_path, &state)?; + if complete_ambiguous_second_turn_before_response && turn_start_count == 2 { + emit_ambiguous_turn_evidence( + &state_path, + &mut state, + !omit_ambiguous_turn_started, + false, + )?; + } + if fail_after_accepting_second_turn_before_response && turn_start_count == 2 { + if emits_ambiguous_turn_evidence + && !complete_ambiguous_second_turn_before_response + { + emit_ambiguous_turn_evidence( + &state_path, + &mut state, + !omit_ambiguous_turn_started, + conflicting_ambiguous_second_turn, + )?; + } + return Err("configured failure after accepting second turn".into()); + } + if exit_after_accepting_second_turn_before_response && turn_start_count == 2 { + return Ok(()); + } + if malformed_error_second_turn_start && turn_start_count == 2 { + send(json!({"id": id, "error": {}}))?; + if emits_ambiguous_turn_evidence + && !complete_ambiguous_second_turn_before_response + { + emit_ambiguous_turn_evidence( + &state_path, + &mut state, + !omit_ambiguous_turn_started, + conflicting_ambiguous_second_turn, + )?; + } + return Err("configured failure after malformed turn error".into()); + } + if missing_id_second_turn_start && turn_start_count == 2 { + send(json!({ + "id": id, + "result": {"turn": {"status": "inProgress"}} + }))?; + if hold_ambiguous_second_turn_after_item { + send(json!({ + "method": "item/completed", + "params": {"item": { + "id": "replacement-message-before-terminal", + "type": "agentMessage", + "status": "completed", + "text": "Replacement output before terminal authority." + }} + }))?; + continue; + } + if emits_ambiguous_turn_evidence + && !complete_ambiguous_second_turn_before_response + { + emit_ambiguous_turn_evidence( + &state_path, + &mut state, + !omit_ambiguous_turn_started, + conflicting_ambiguous_second_turn, + )?; + } + return Err("configured failure after missing turn identity".into()); + } send(json!({ "id": id, "result": {"turn": {"id": "provider-turn-1", "status": "inProgress"}} @@ -151,8 +405,28 @@ fn run() -> Result<(), Box> { "method": "turn/started", "params": {"turn": {"id": "provider-turn-1"}} }))?; - if exit_after_turn_start { + if fail_after_second_turn_start && turn_start_count == 2 { + return Err("configured failure after second turn start".into()); + } else if exit_after_turn_start { return Ok(()); + } else if emit_tool_call { + send(json!({ + "id": "tool-request-1", + "method": "item/tool/call", + "params": { + "threadId": state.thread_id, + "turnId": "provider-turn-1", + "callId": "semantic-call-1", + "tool": "get_task_context", + "arguments": {} + } + }))?; + if complete_after_tool_call { + finish_turn(&state_path, &mut state, "completed")?; + if exit_after_tool_call_completion { + return Ok(()); + } + } } else if emit_question { send(json!({ "id": "runtime-request-1", @@ -176,6 +450,29 @@ fn run() -> Result<(), Box> { }))?; } else if !hold_turn { finish_turn(&state_path, &mut state, "completed")?; + if emit_post_completion_warning { + send(json!({ + "method": "warning", + "params": {"message": "provider remained live after terminal"} + }))?; + } + if fail_after_turn_completion { + if let Some(delay_ms) = fail_after_turn_completion_delay_ms { + thread::sleep(Duration::from_millis(delay_ms)); + // Make the post-terminal liveness observation + // deterministic even when parallel tests delay the + // controller's next poll until after this process + // exits. + send(json!({ + "method": "warning", + "params": {"message": "provider remained live after terminal"} + }))?; + } + return Err("configured failure after turn completion".into()); + } + if exit_after_turn_completion { + return Ok(()); + } } } "turn/steer" => send(json!({"id": id, "result": {"accepted": true}}))?, 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 74fede89fb..322236013d 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 @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, VecDeque}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -8,10 +8,14 @@ use serde_json::{json, Value}; use crate::durable::redact_text; use crate::local_runner::LocalRunnerError; use crate::process_supervisor::SupervisedProcess; +use crate::provider_bridge::{AuthorizedTool, ToolResult}; pub const CODEX_APP_SERVER_MAX_FRAME_BYTES: usize = 4 * 1024 * 1024; const MAX_BUFFERED_MESSAGES: usize = 1_024; const MAX_INSTRUCTIONS_BYTES: usize = 1024 * 1024; +const MAX_PENDING_TOOL_REQUESTS: usize = 4_096; +const MAX_PENDING_TOOL_REQUEST_BYTES: usize = 16 * 1024 * 1024; +const MAX_COMPLETED_TOOL_CALL_IDS: usize = 4_096; type QuestionOptionLabels = BTreeMap>; type QuestionSetMapping = (String, Value, QuestionOptionLabels); @@ -101,6 +105,11 @@ impl CodexProviderConfig { #[derive(Clone, Debug, PartialEq)] pub enum CodexProviderEvent { + ToolCall { + call_id: String, + operation_id: String, + input: Value, + }, Notification { method: String, params: Value, @@ -112,9 +121,41 @@ pub enum CodexProviderEvent { Exited { exit_code: Option, success: bool, + completed_turn_authoritative: bool, + completed_turn_observed_by_process: bool, + completion_reconciles_exit: bool, + process_generation: u64, + completed_turn_process_generation: Option, }, } +#[derive(Clone, Debug, PartialEq, Eq)] +struct CompletedTurnAuthority { + process_generation: u64, + provider_turn_id: String, +} + +enum ProviderRequestError { + Rejected(LocalRunnerError), + Ambiguous(LocalRunnerError), +} + +impl ProviderRequestError { + fn into_inner(self) -> LocalRunnerError { + match self { + Self::Rejected(error) | Self::Ambiguous(error) => error, + } + } +} + +#[derive(Clone, Debug, PartialEq)] +struct PendingToolRequest { + rpc_id: Value, + operation_id: String, + input: Value, + retained_bytes: usize, +} + #[derive(Clone, Debug, PartialEq)] struct PendingRuntimeRequest { rpc_id: Value, @@ -124,23 +165,79 @@ struct PendingRuntimeRequest { option_labels: QuestionOptionLabels, } +struct BufferedProviderMessage { + value: Value, +} + +enum AmbiguousTurnMessage { + Ready, + Deferred, + ReconciledWithStart, + ReconciledNeedsStart { provider_turn_id: String }, +} + pub struct CodexProvider { process: SupervisedProcess, next_request_id: u64, thread_id: String, provider_session_id: Option, active_provider_turn_id: Option, - pending_messages: VecDeque, + pending_messages: VecDeque, + deferred_ambiguous_messages: VecDeque, + authorized_tool_ids: BTreeSet, + pending_tool_requests: BTreeMap, + completed_tool_call_ids: BTreeSet, + pending_tool_request_bytes: usize, pending_runtime_requests: BTreeMap, expected_shutdown: bool, + process_generation: u64, + completed_turn_authority: Option, + completion_reconciliation_pending: bool, + ambiguous_turn_start_pending: bool, } impl CodexProvider { pub fn start( config: &CodexProviderConfig, resume_thread_id: Option<&str>, + ) -> Result { + Self::start_with_tools_for_generation(config, std::iter::empty(), resume_thread_id, 1) + } + + pub fn start_with_tools( + config: &CodexProviderConfig, + authorized_tools: impl IntoIterator, + resume_thread_id: Option<&str>, + ) -> Result { + Self::start_with_tools_for_generation(config, authorized_tools, resume_thread_id, 1) + } + + pub(crate) fn start_for_generation( + config: &CodexProviderConfig, + resume_thread_id: Option<&str>, + process_generation: u64, + ) -> Result { + Self::start_with_tools_for_generation( + config, + std::iter::empty(), + resume_thread_id, + process_generation, + ) + } + + fn start_with_tools_for_generation( + config: &CodexProviderConfig, + authorized_tools: impl IntoIterator, + resume_thread_id: Option<&str>, + process_generation: u64, ) -> Result { config.validate()?; + if process_generation == 0 { + return Err(LocalRunnerError::invalid( + "Codex process generation must be positive", + )); + } + let (dynamic_tools, authorized_tool_ids) = codex_dynamic_tools(authorized_tools)?; let mut provider = Self { process: SupervisedProcess::spawn( &config.command, @@ -153,8 +250,17 @@ impl CodexProvider { provider_session_id: None, active_provider_turn_id: None, pending_messages: VecDeque::new(), + deferred_ambiguous_messages: VecDeque::new(), + authorized_tool_ids, + pending_tool_requests: BTreeMap::new(), + completed_tool_call_ids: BTreeSet::new(), + pending_tool_request_bytes: 0, pending_runtime_requests: BTreeMap::new(), expected_shutdown: false, + process_generation, + completed_turn_authority: None, + completion_reconciliation_pending: false, + ambiguous_turn_start_pending: false, }; let initialized = provider.request( "initialize", @@ -179,6 +285,7 @@ impl CodexProvider { "permissions": "paperclip-runner-workspace-only", "runtimeWorkspaceRoots": [config.cwd], "baseInstructions": config.instructions, + "dynamicTools": dynamic_tools, }); let params_object = params .as_object_mut() @@ -187,9 +294,6 @@ impl CodexProvider { params_object.insert("threadId".to_owned(), json!(thread_id)); "thread/resume" } else { - // This PR does not grant any semantic tools. A later catalog and - // authorization layer can project a run-scoped inventory here. - params_object.insert("dynamicTools".to_owned(), json!([])); params_object.insert("experimentalRawEvents".to_owned(), json!(false)); "thread/start" }; @@ -215,7 +319,9 @@ impl CodexProvider { if resume_thread_id.is_some() { let snapshot = provider.read_thread()?; - provider.active_provider_turn_id = latest_active_turn_id(&snapshot); + provider.active_provider_turn_id = latest_active_turn_id(&snapshot) + .map(|turn_id| bounded_identifier(Some(&turn_id), "Codex turn id")) + .transpose()?; } Ok(provider) } @@ -236,18 +342,68 @@ impl CodexProvider { self.active_provider_turn_id.as_deref() } + pub(crate) fn ambiguous_turn_start_pending(&self) -> bool { + self.ambiguous_turn_start_pending + } + + pub(crate) fn restore_completed_turn_authority( + &mut self, + authoritative: bool, + process_generation: Option, + provider_turn_id: Option<&str>, + ) { + self.completed_turn_authority = authoritative.then(|| CompletedTurnAuthority { + // Legacy state did not record the generation. Generation zero is + // deliberately older than every supervised process generation. + process_generation: process_generation.unwrap_or(0), + provider_turn_id: provider_turn_id + .unwrap_or("durable-completed-turn") + .to_owned(), + }); + // Resuming a completed durable thread and reading its provider state + // is recovery, not new turn work. Keep the prior terminal authoritative + // until start_turn explicitly revokes it. + self.expected_shutdown = authoritative; + // Completion authority is durable across provider generations. Output, + // probes, and process restarts are session-liveness observations; none + // of them supersedes a completed result. Only accepting a replacement + // turn identity revokes this authority. + self.completion_reconciliation_pending = false; + } + + pub(crate) fn completed_turn_authority(&self) -> Option<(u64, &str)> { + self.completed_turn_authority.as_ref().map(|authority| { + ( + authority.process_generation, + authority.provider_turn_id.as_str(), + ) + }) + } + pub fn start_turn(&mut self, message: &str, cwd: &str) -> Result { if self.active_provider_turn_id.is_some() { return Err(LocalRunnerError::invalid( "Codex already has an active provider turn", )); } + if self.ambiguous_turn_start_pending { + return Err(LocalRunnerError::invalid( + "Codex has an unresolved ambiguous provider turn start", + )); + } if message.is_empty() || message.len() > MAX_INSTRUCTIONS_BYTES { return Err(LocalRunnerError::invalid( "Codex turn text is empty or exceeds the 1 MiB limit", )); } - let result = self.request( + // Preserve the prior durable result until a replacement turn identity + // is accepted. A rejected, ambiguous, or transport-failed attempt does + // not prove that replacement work superseded the completed turn. + let prior_reconciliation_pending = self.completion_reconciliation_pending; + self.completion_reconciliation_pending = false; + let prior_buffered_message_count = self.pending_messages.len(); + self.ambiguous_turn_start_pending = true; + let result = match self.request_classified( "turn/start", json!({ "threadId": self.thread_id, @@ -256,18 +412,92 @@ impl CodexProvider { "runtimeWorkspaceRoots": [cwd], "input": [{"type": "text", "text": message, "text_elements": []}], }), + ) { + Ok(result) => result, + Err(ProviderRequestError::Rejected(error)) => { + // A definite rejection proves no replacement work began. + // Only diagnostics without provider-work identity belong to + // that rejected request. Contradictory turn/item evidence or a + // server request leaves the start ambiguous until a validated + // replacement identity is observed. + let definite_rejection = self + .pending_messages + .iter() + .skip(prior_buffered_message_count) + .all(|buffered| is_unbound_rejected_turn_diagnostic(&buffered.value)); + if definite_rejection { + self.ambiguous_turn_start_pending = false; + self.completion_reconciliation_pending = prior_reconciliation_pending; + } + return Err(error); + } + Err(ProviderRequestError::Ambiguous(error)) => return Err(error), + }; + let provider_turn_id = bounded_identifier( + result + .pointer("/turn/id") + .or_else(|| result.get("turnId")) + .and_then(Value::as_str), + "Codex turn id", )?; - let provider_turn_id = result - .pointer("/turn/id") - .or_else(|| result.get("turnId")) - .and_then(Value::as_str) - .filter(|value| !value.is_empty()) - .ok_or_else(|| LocalRunnerError::invalid("Codex turn/start omitted turn.id"))? - .to_owned(); - self.active_provider_turn_id = Some(provider_turn_id); + // Only a validated provider turn identity proves that replacement + // work exists and supersedes the prior completed result. + self.accept_replacement_turn(provider_turn_id); Ok(result) } + fn classify_ambiguous_turn_message( + &mut self, + message: &Value, + ) -> Result { + if !self.ambiguous_turn_start_pending { + return Ok(AmbiguousTurnMessage::Ready); + } + + let Some(method) = message.get("method").and_then(Value::as_str) else { + return Ok(AmbiguousTurnMessage::Deferred); + }; + let params = message.get("params").cloned().unwrap_or(Value::Null); + let provider_turn_id = notification_turn_id(¶ms); + let identity_required = matches!(method, "turn/started" | "turn/completed"); + if provider_turn_id.is_none() && !identity_required { + return Ok(AmbiguousTurnMessage::Deferred); + } + validate_notification_binding(&self.thread_id, None, ¶ms)?; + let provider_turn_id = + bounded_identifier(provider_turn_id, "Codex turn id").map_err(|_| { + LocalRunnerError::invalid(format!( + "Codex {method} cannot resolve an ambiguous turn start without a valid turn id" + )) + })?; + + if self + .completed_turn_authority + .as_ref() + .is_some_and(|authority| authority.provider_turn_id == provider_turn_id) + { + return Err(LocalRunnerError::invalid(format!( + "Codex {method} notification reused the previously completed turn id while resolving an ambiguous turn start" + ))); + } + + self.accept_replacement_turn(provider_turn_id.clone()); + if method == "turn/started" && message.get("id").is_none() { + Ok(AmbiguousTurnMessage::ReconciledWithStart) + } else { + Ok(AmbiguousTurnMessage::ReconciledNeedsStart { provider_turn_id }) + } + } + + fn accept_replacement_turn(&mut self, provider_turn_id: String) { + self.ambiguous_turn_start_pending = false; + self.expected_shutdown = false; + self.completed_turn_authority = None; + self.completion_reconciliation_pending = false; + self.completed_tool_call_ids.clear(); + self.active_provider_turn_id = Some(provider_turn_id); + } + pub fn steer_turn(&mut self, message: &str) -> Result { let turn_id = self .active_provider_turn_id @@ -293,7 +523,7 @@ impl CodexProvider { .active_provider_turn_id .clone() .ok_or_else(|| LocalRunnerError::invalid("Codex has no active provider turn"))?; - self.cancel_pending_runtime_requests()?; + self.cancel_pending_requests()?; self.request( "turn/interrupt", json!({"threadId": self.thread_id, "turnId": turn_id}), @@ -301,6 +531,11 @@ impl CodexProvider { } pub fn read_thread(&mut self) -> Result { + // Probing provider state does not supersede an authoritative terminal. + // A later replacement turn must still establish its own identity. + // It does prove the provider remained live after that terminal, so a + // subsequent nonzero exit is a separate idle-session failure. + self.completion_reconciliation_pending = false; self.request( "thread/read", json!({"threadId": self.thread_id, "includeTurns": true}), @@ -327,14 +562,48 @@ impl CodexProvider { } pub fn poll(&mut self) -> Result, LocalRunnerError> { - let message = if let Some(message) = self.pending_messages.pop_front() { - message + let buffered = self.pending_messages.pop_front(); + let message = if let Some(buffered) = buffered { + buffered.value } else { let Some(line) = self.process.receive_stdout_line(Duration::from_millis(1))? else { - return if let Some(exit) = self.process.try_wait()? { + let exit = self.process.try_wait()?; + return if let Some(exit) = exit { + let completed_turn_authoritative = self.completed_turn_authority.is_some() + && self.active_provider_turn_id.is_none(); + let completed_turn_observed_by_process = self + .completed_turn_authority + .as_ref() + .is_some_and(|authority| { + authority.process_generation == self.process_generation + }); + // A durable terminal remains the run outcome, but it only + // reconciles the process generation that produced it. A + // later recovered provider can fail independently while + // leaving the already-recorded turn result intact. + let completion_reconciles_exit = completed_turn_authoritative + && completed_turn_observed_by_process + && self.completion_reconciliation_pending; Ok(Some(CodexProviderEvent::Exited { exit_code: exit.exit_code, - success: exit.success && self.expected_shutdown, + // A clean idle exit after a terminal is healthy. A + // nonzero exit still makes the provider unavailable, + // but the durable terminal reconciles it instead of + // allowing the session to fail retroactively. Fresh + // turn work explicitly revokes the prior authority. + // An unresolved start may already have created fresh + // work, so even a clean exit must fail that session. + success: exit.success + && !self.ambiguous_turn_start_pending + && (self.expected_shutdown || completed_turn_authoritative), + completed_turn_authoritative, + completed_turn_observed_by_process, + completion_reconciles_exit, + process_generation: self.process_generation, + completed_turn_process_generation: self + .completed_turn_authority + .as_ref() + .map(|authority| authority.process_generation), })) } else { Ok(None) @@ -343,10 +612,157 @@ impl CodexProvider { parse_provider_message(&line)? }; + if self.completed_turn_authority.is_some() + && self.active_provider_turn_id.is_none() + && message.get("method").and_then(Value::as_str) != Some("turn/completed") + { + // Output after the terminal proves the provider entered an idle + // liveness phase. Keep the result authoritative, but do not let it + // hide a later process failure. + self.completion_reconciliation_pending = false; + } + + match self.classify_ambiguous_turn_message(&message)? { + AmbiguousTurnMessage::Ready => {} + AmbiguousTurnMessage::Deferred => { + if self.deferred_ambiguous_messages.len() >= MAX_BUFFERED_MESSAGES { + return Err(LocalRunnerError::invalid( + "Codex emitted too many messages before resolving an ambiguous turn start", + )); + } + self.deferred_ambiguous_messages + .push_back(BufferedProviderMessage { value: message }); + return Ok(None); + } + AmbiguousTurnMessage::ReconciledWithStart => { + let mut replay = std::mem::take(&mut self.deferred_ambiguous_messages); + replay.append(&mut self.pending_messages); + self.pending_messages = replay; + } + AmbiguousTurnMessage::ReconciledNeedsStart { provider_turn_id } => { + let mut replay = std::mem::take(&mut self.deferred_ambiguous_messages); + replay.push_back(BufferedProviderMessage { value: message }); + replay.append(&mut self.pending_messages); + self.pending_messages = replay; + return Ok(Some(CodexProviderEvent::Notification { + method: "turn/started".to_owned(), + params: json!({ + "threadId": self.thread_id, + "turn": {"id": provider_turn_id, "status": "inProgress"}, + "reconciled": true, + }), + })); + } + } + if let (Some(rpc_id), Some(method)) = ( message.get("id").cloned(), message.get("method").and_then(Value::as_str), ) { + if method == "item/tool/call" { + let params = message.get("params").cloned().unwrap_or(Value::Null); + if params.get("threadId").and_then(Value::as_str) != Some(self.thread_id.as_str()) { + return Err(LocalRunnerError::invalid( + "Codex tool call named another thread", + )); + } + let active_turn_id = self.active_provider_turn_id.as_deref().ok_or_else(|| { + LocalRunnerError::invalid("Codex tool call arrived outside an active turn") + })?; + if params.get("turnId").and_then(Value::as_str) != Some(active_turn_id) { + return Err(LocalRunnerError::invalid( + "Codex tool call named another turn", + )); + } + let call_id = bounded_identifier( + params.get("callId").and_then(Value::as_str), + "Codex tool callId", + )?; + let operation_id = bounded_identifier( + params.get("tool").and_then(Value::as_str), + "Codex tool name", + )?; + if !self.authorized_tool_ids.contains(&operation_id) { + self.process.send(&json!({ + "id": rpc_id, + "result": codex_tool_failure("Paperclip did not authorize this tool for the run"), + }))?; + return Err(LocalRunnerError::invalid(format!( + "Codex requested unauthorized tool {}", + bounded_method(&operation_id) + ))); + } + let input = params.get("arguments").cloned().unwrap_or(Value::Null); + let input_bytes = serde_json::to_vec(&input) + .map_err(|error| { + LocalRunnerError::invalid(format!( + "Codex tool arguments are not serializable: {error}" + )) + })? + .len(); + let rpc_id_bytes = serde_json::to_vec(&rpc_id) + .map_err(|error| { + LocalRunnerError::invalid(format!( + "Codex JSON-RPC id is not serializable: {error}" + )) + })? + .len(); + let retained_bytes = pending_tool_request_size([ + input_bytes, + rpc_id_bytes, + call_id.len(), + operation_id.len(), + ])?; + let pending = PendingToolRequest { + rpc_id: rpc_id.clone(), + operation_id: operation_id.clone(), + input: input.clone(), + retained_bytes, + }; + if self.completed_tool_call_ids.contains(&call_id) { + return Err(LocalRunnerError::invalid( + "Codex reused a completed tool call id", + )); + } + if let Some(existing) = self.pending_tool_requests.get(&call_id) { + if existing != &pending { + return Err(LocalRunnerError::invalid( + "Codex reused a tool call id with different input", + )); + } + return Ok(None); + } + if self + .pending_tool_requests + .values() + .any(|existing| existing.rpc_id == rpc_id) + { + return Err(LocalRunnerError::invalid( + "Codex reused a pending JSON-RPC id for another tool call", + )); + } + if self.pending_tool_requests.len() >= MAX_PENDING_TOOL_REQUESTS { + return Err(LocalRunnerError::invalid( + "Codex emitted too many pending tool calls", + )); + } + if self.completed_tool_call_ids.len() >= MAX_COMPLETED_TOOL_CALL_IDS { + return Err(LocalRunnerError::invalid( + "Codex emitted too many completed tool calls in one turn", + )); + } + let retained_request_bytes = retain_pending_tool_request_bytes( + self.pending_tool_request_bytes, + retained_bytes, + )?; + self.pending_tool_requests.insert(call_id.clone(), pending); + self.pending_tool_request_bytes = retained_request_bytes; + return Ok(Some(CodexProviderEvent::ToolCall { + call_id, + operation_id, + input, + })); + } if method == "item/tool/requestUserInput" { let params = message.get("params").cloned().unwrap_or(Value::Null); if params.get("threadId").and_then(Value::as_str) != Some(self.thread_id.as_str()) { @@ -407,7 +823,24 @@ impl CodexProvider { ¶ms, )?; if method == "turn/completed" { + let provider_turn_id = self.active_provider_turn_id.clone().ok_or_else(|| { + LocalRunnerError::invalid( + "Codex completion arrived outside an active provider turn", + ) + })?; self.active_provider_turn_id = None; + self.expected_shutdown = true; + self.completed_turn_authority = Some(CompletedTurnAuthority { + process_generation: self.process_generation, + provider_turn_id, + }); + self.completion_reconciliation_pending = true; + // The provider terminal is authoritative once received. Clear + // local request ownership and attempt courtesy responses, but + // a provider that already closed stdin must not turn the + // completed turn back into a transport failure. + let _ = self.cancel_pending_requests(); + self.completed_tool_call_ids.clear(); } return Ok(Some(CodexProviderEvent::Notification { method: method.to_owned(), @@ -417,60 +850,248 @@ impl CodexProvider { Ok(None) } - pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> { - self.expected_shutdown = true; - self.cancel_pending_runtime_requests()?; - self.process.terminate_group().map(|_| ()) - } - - fn cancel_pending_runtime_requests(&mut self) -> Result<(), LocalRunnerError> { - let pending = std::mem::take(&mut self.pending_runtime_requests); - for request in pending.into_values() { - self.process.send(&json!({ - "id": request.rpc_id, - "result": {"answers": {}}, - }))?; + pub fn deliver_tool_result(&mut self, result: &ToolResult) -> Result<(), LocalRunnerError> { + let pending = self + .pending_tool_requests + .get(&result.call_id) + .cloned() + .ok_or_else(|| { + LocalRunnerError::invalid("Codex tool result has no pending JSON-RPC request") + })?; + if pending.operation_id != result.operation_id { + return Err(LocalRunnerError::invalid( + "Codex tool result operation does not match its call", + )); + } + let result_bytes = serde_json::to_vec(&result.result).map_err(|error| { + LocalRunnerError::invalid(format!("Codex tool result is not serializable: {error}")) + })?; + if result_bytes.len() > 1024 * 1024 { + return Err(LocalRunnerError::invalid( + "Codex tool result exceeds the 1 MiB limit", + )); + } + let text = String::from_utf8(result_bytes) + .expect("serde_json always serializes JSON values as valid UTF-8"); + self.process.send(&json!({ + "id": pending.rpc_id, + "result": { + "success": !result.is_error, + "contentItems": [{"type": "inputText", "text": text}], + }, + }))?; + if let Some(completed) = self.pending_tool_requests.remove(&result.call_id) { + self.pending_tool_request_bytes = self + .pending_tool_request_bytes + .saturating_sub(completed.retained_bytes); + self.completed_tool_call_ids.insert(result.call_id.clone()); } Ok(()) } + pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> { + self.expected_shutdown = true; + self.cancel_pending_requests()?; + self.process.terminate_group().map(|_| ()) + } + + fn cancel_pending_requests(&mut self) -> Result<(), LocalRunnerError> { + let pending_runtime = std::mem::take(&mut self.pending_runtime_requests); + let pending = std::mem::take(&mut self.pending_tool_requests); + self.pending_tool_request_bytes = 0; + let mut first_error = None; + for request in pending_runtime.into_values() { + if let Err(error) = self.process.send(&json!({ + "id": request.rpc_id, + "result": {"answers": {}}, + })) { + first_error.get_or_insert(error); + } + } + for request in pending.into_values() { + if let Err(error) = self.process.send(&json!({ + "id": request.rpc_id, + "result": codex_tool_failure("Paperclip stopped the active provider turn"), + })) { + first_error.get_or_insert(error); + } + } + match first_error { + Some(error) => Err(error), + None => Ok(()), + } + } + fn request(&mut self, method: &str, params: Value) -> Result { + self.request_classified(method, params) + .map_err(ProviderRequestError::into_inner) + } + + fn request_classified( + &mut self, + method: &str, + params: Value, + ) -> Result { let request_id = self.next_request_id; - self.next_request_id = self - .next_request_id - .checked_add(1) - .ok_or_else(|| LocalRunnerError::invalid("Codex request id exhausted"))?; + self.next_request_id = self.next_request_id.checked_add(1).ok_or_else(|| { + ProviderRequestError::Rejected(LocalRunnerError::invalid("Codex request id exhausted")) + })?; self.process - .send(&json!({"id": request_id, "method": method, "params": params}))?; + .send(&json!({"id": request_id, "method": method, "params": params})) + .map_err(ProviderRequestError::Ambiguous)?; loop { let line = self .process - .receive_stdout_line(Duration::from_secs(30))? + .receive_stdout_line(Duration::from_secs(30)) + .map_err(ProviderRequestError::Ambiguous)? .ok_or_else(|| { - LocalRunnerError::invalid(format!("Codex {method} response timed out")) + ProviderRequestError::Ambiguous(LocalRunnerError::invalid(format!( + "Codex {method} response timed out" + ))) })?; - let message = parse_provider_message(&line)?; + let message = parse_provider_message(&line).map_err(ProviderRequestError::Ambiguous)?; if message.get("id").and_then(Value::as_u64) == Some(request_id) && message.get("method").is_none() { if let Some(error) = message.get("error") { - return Err(LocalRunnerError::invalid(format!( - "Codex {method} failed: {}", - redact_text(&error.to_string()) + let well_formed_rejection = error.get("code").and_then(Value::as_i64).is_some() + && error.get("message").and_then(Value::as_str).is_some(); + if !well_formed_rejection || message.get("result").is_some() { + return Err(ProviderRequestError::Ambiguous(LocalRunnerError::invalid( + format!("Codex {method} returned an invalid error response"), + ))); + } + return Err(ProviderRequestError::Rejected(LocalRunnerError::invalid( + format!("Codex {method} failed: {}", redact_text(&error.to_string())), ))); } return Ok(message.get("result").cloned().unwrap_or(Value::Null)); } if self.pending_messages.len() >= MAX_BUFFERED_MESSAGES { - return Err(LocalRunnerError::invalid( + return Err(ProviderRequestError::Ambiguous(LocalRunnerError::invalid( "Codex emitted too many messages before a request response", - )); + ))); } - self.pending_messages.push_back(message); + self.pending_messages + .push_back(BufferedProviderMessage { value: message }); } } } +fn pending_tool_request_size( + parts: impl IntoIterator, +) -> Result { + parts.into_iter().try_fold(0usize, |total, part| { + total + .checked_add(part) + .ok_or_else(|| LocalRunnerError::invalid("Codex pending tool request size overflowed")) + }) +} + +fn retain_pending_tool_request_bytes( + current: usize, + incoming: usize, +) -> Result { + current + .checked_add(incoming) + .filter(|total| *total <= MAX_PENDING_TOOL_REQUEST_BYTES) + .ok_or_else(|| { + LocalRunnerError::invalid( + "Codex pending tool requests exceed the 16 MiB aggregate limit", + ) + }) +} + +fn codex_dynamic_tools( + authorized_tools: impl IntoIterator, +) -> Result<(Vec, BTreeSet), LocalRunnerError> { + let mut dynamic_tools = Vec::new(); + let mut operation_ids = BTreeSet::new(); + for tool in authorized_tools { + if dynamic_tools.len() >= 256 { + return Err(LocalRunnerError::invalid( + "Codex authorized tool set exceeds the operation limit", + )); + } + let operation_id = bounded_identifier(Some(&tool.operation_id), "Codex tool name")?; + let mut characters = operation_id.chars(); + let valid_first = characters + .next() + .is_some_and(|character| character.is_ascii_alphanumeric()); + let valid_rest = characters.all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '_' | '-' | '.' | ':') + }); + if !valid_first || !valid_rest { + return Err(LocalRunnerError::invalid( + "Codex tool name is not a valid operation id", + )); + } + if tool.version != 1 + || tool.description.trim().is_empty() + || tool.description.len() > 16 * 1024 + || tool.description.contains('\0') + || !tool.input_schema.is_object() + || !tool.response_schema.is_object() + { + return Err(LocalRunnerError::invalid(format!( + "Codex tool {} has an incomplete provider contract", + bounded_method(&operation_id) + ))); + } + let input_schema_bytes = serde_json::to_vec(&tool.input_schema).map_err(|error| { + LocalRunnerError::invalid(format!("Codex tool input schema is invalid: {error}")) + })?; + if input_schema_bytes.len() > 1024 * 1024 { + return Err(LocalRunnerError::invalid( + "Codex tool input schema exceeds the 1 MiB limit", + )); + } + jsonschema::validator_for(&tool.input_schema).map_err(|_| { + LocalRunnerError::invalid(format!( + "Codex tool {} has an invalid input JSON Schema", + bounded_method(&operation_id) + )) + })?; + if !operation_ids.insert(operation_id.clone()) { + return Err(LocalRunnerError::invalid( + "Codex authorized tool names must be unique", + )); + } + dynamic_tools.push(json!({ + "name": operation_id, + "description": tool.description, + "inputSchema": tool.input_schema, + })); + } + if serde_json::to_vec(&dynamic_tools) + .map_err(|error| { + LocalRunnerError::invalid(format!("Codex dynamic tool set is invalid: {error}")) + })? + .len() + > 4 * 1024 * 1024 + { + return Err(LocalRunnerError::invalid( + "Codex dynamic tool set exceeds the 4 MiB limit", + )); + } + Ok((dynamic_tools, operation_ids)) +} + +fn bounded_identifier(value: Option<&str>, label: &str) -> Result { + let value = value.ok_or_else(|| LocalRunnerError::invalid(format!("{label} is required")))?; + if value.is_empty() || value.len() > 160 || value.chars().any(char::is_control) { + return Err(LocalRunnerError::invalid(format!("{label} is invalid"))); + } + Ok(value.to_owned()) +} + +fn codex_tool_failure(message: &str) -> Value { + json!({ + "success": false, + "contentItems": [{"type": "inputText", "text": message}], + }) +} + fn parse_provider_message(line: &str) -> Result { let value: Value = serde_json::from_str(line).map_err(|error| { LocalRunnerError::invalid(format!("Codex emitted invalid JSON-RPC: {error}")) @@ -483,6 +1104,28 @@ fn parse_provider_message(line: &str) -> Result { Ok(value) } +fn is_unbound_rejected_turn_diagnostic(message: &Value) -> bool { + message.get("id").is_none() + && message.get("method").and_then(Value::as_str) == Some("warning") + && message + .get("params") + .is_none_or(|params| !contains_provider_work_binding(params)) +} + +fn contains_provider_work_binding(value: &Value) -> bool { + match value { + Value::Array(values) => values.iter().any(contains_provider_work_binding), + Value::Object(fields) => fields.iter().any(|(key, child)| { + (matches!(key.as_str(), "threadId" | "turnId" | "itemId" | "requestId") + && !child.is_null()) + || (matches!(key.as_str(), "thread" | "turn" | "item" | "request") + && child.get("id").is_some_and(|id| !id.is_null())) + || contains_provider_work_binding(child) + }), + _ => false, + } +} + fn validate_notification_binding( thread_id: &str, active_turn_id: Option<&str>, @@ -498,11 +1141,7 @@ fn validate_notification_binding( "Codex notification named another thread", )); } - let notification_turn_id = params - .get("turnId") - .or_else(|| params.pointer("/turn/id")) - .and_then(Value::as_str) - .filter(|value| !value.is_empty()); + let notification_turn_id = notification_turn_id(params); if let Some(active_turn_id) = active_turn_id { if notification_turn_id.is_some_and(|value| value != active_turn_id) { return Err(LocalRunnerError::invalid( @@ -513,6 +1152,14 @@ fn validate_notification_binding( Ok(()) } +fn notification_turn_id(params: &Value) -> Option<&str> { + params + .get("turnId") + .or_else(|| params.pointer("/turn/id")) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) +} + fn latest_active_turn_id(snapshot: &Value) -> Option { snapshot .pointer("/thread/turns") @@ -855,4 +1502,18 @@ mod tests { ) .is_ok()); } + + #[test] + fn bounds_all_retained_pending_tool_request_data_in_aggregate() { + let request_bytes = pending_tool_request_size([1, 2, 3, 4]).unwrap(); + assert_eq!(request_bytes, 10); + assert_eq!( + retain_pending_tool_request_bytes(MAX_PENDING_TOOL_REQUEST_BYTES - 10, request_bytes) + .unwrap(), + MAX_PENDING_TOOL_REQUEST_BYTES + ); + assert!(retain_pending_tool_request_bytes(MAX_PENDING_TOOL_REQUEST_BYTES, 1).is_err()); + assert!(retain_pending_tool_request_bytes(usize::MAX, 1).is_err()); + assert!(pending_tool_request_size([usize::MAX, 1]).is_err()); + } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs b/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs index 768e7dba25..7d2d058f00 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs @@ -308,9 +308,11 @@ impl SupervisedProcess { } Ok(ProcessOutput::StdoutClosed) => return Ok(None), Err(RecvTimeoutError::Timeout) => return Ok(None), - Err(RecvTimeoutError::Disconnected) => { - return Err(LocalRunnerError::invalid("process output channel closed")); - } + // The reader threads end when the child closes its output. + // Let the caller reconcile that closure with the authoritative + // process exit status instead of turning the channel teardown + // into a transport failure. + Err(RecvTimeoutError::Disconnected) => return Ok(None), } } } 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 d814f4abe0..860362da03 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 @@ -164,6 +164,15 @@ struct CodexProviderState { #[serde(default)] active_provider_turn_id: Option, #[serde(default)] + ambiguous_turn_start_pending: bool, + #[serde(default)] + completed_turn_authoritative: bool, + #[serde(default)] + provider_process_generation: u64, + #[serde(default)] + completed_turn_process_generation: Option, + #[serde(default)] + completed_provider_turn_id: Option, last_agent_message: Option, #[serde(default)] pending_events: VecDeque, @@ -185,6 +194,11 @@ impl CodexProviderState { thread_id, provider_session_id: None, active_provider_turn_id: None, + ambiguous_turn_start_pending: false, + completed_turn_authoritative: false, + provider_process_generation: 0, + completed_turn_process_generation: None, + completed_provider_turn_id: None, last_agent_message: None, pending_events: VecDeque::new(), next_provider_event_seq: initial_provider_event_seq(), @@ -213,6 +227,10 @@ impl CodexProviderState { .active_provider_turn_id .as_ref() .is_some_and(|value| value.is_empty() || value.len() > 240) + || self + .completed_provider_turn_id + .as_ref() + .is_some_and(|value| value.is_empty() || value.len() > 240) || self.completion_contract.as_ref().is_some_and(|contract| { contract.revision.is_empty() || contract.revision.len() > 120 @@ -233,6 +251,20 @@ impl CodexProviderState { || self.active_provider_turn_id.is_some() || matches!(self.lifecycle.as_str(), "session_open" | "turn_active"))) || (self.lifecycle == "turn_active" && self.active_provider_turn_id.is_none()) + || (self.ambiguous_turn_start_pending + && (self.thread_id.is_none() + || self.active_provider_turn_id.is_some() + || matches!( + self.lifecycle.as_str(), + "prepared" | "turn_active" | "closed" + ))) + || (self.completed_turn_authoritative && self.active_provider_turn_id.is_some()) + || (!self.completed_turn_authoritative + && (self.completed_turn_process_generation.is_some() + || self.completed_provider_turn_id.is_some())) + || self + .completed_turn_process_generation + .is_some_and(|generation| generation > self.provider_process_generation) || (matches!( self.lifecycle.as_str(), "prepared" | "session_open" | "closed" @@ -279,6 +311,25 @@ impl CodexProviderState { } Ok(()) } + + fn reconcile_active_provider_turn(&mut self, active_provider_turn_id: Option) { + self.active_provider_turn_id = active_provider_turn_id; + if self.active_provider_turn_id.is_some() { + // A newly discovered turn supersedes completion authority from the + // prior turn. Persisting both would make the recovered state + // invalid and could misclassify a later provider exit. + self.completed_turn_authoritative = false; + self.completed_turn_process_generation = None; + self.completed_provider_turn_id = None; + self.ambiguous_turn_start_pending = false; + self.last_agent_message = None; + } + self.lifecycle = if self.active_provider_turn_id.is_some() { + "turn_active".to_owned() + } else { + "session_open".to_owned() + }; + } } pub struct CodexCommandExecutor { @@ -354,22 +405,56 @@ impl CodexCommandExecutor { DurableRunnerError::invalid("recoverable Codex state omitted its thread id") })?; let previous_active_turn_id = state.active_provider_turn_id.clone(); - let provider = CodexProvider::start(&state.config, Some(&thread_id)).map_err(|error| { + let process_generation = state + .provider_process_generation + .checked_add(1) + .ok_or_else(|| DurableRunnerError::invalid("Codex process generation exhausted"))?; + let completed_turn_authoritative = state.completed_turn_authoritative; + let completed_turn_process_generation = state.completed_turn_process_generation; + let completed_provider_turn_id = state.completed_provider_turn_id.clone(); + let ambiguous_turn_start_pending = state.ambiguous_turn_start_pending; + let mut provider = CodexProvider::start_for_generation( + &state.config, + Some(&thread_id), + process_generation, + ) + .map_err(|error| { DurableRunnerError::invalid(format!("failed to resume Codex provider: {error}")) })?; let recovered_active_turn_id = provider.active_provider_turn_id().map(str::to_owned); + if ambiguous_turn_start_pending { + let recovered_turn_id = recovered_active_turn_id.as_deref().ok_or_else(|| { + DurableRunnerError::invalid( + "cannot safely recover an ambiguous Codex turn start without an active replacement turn", + ) + })?; + if completed_provider_turn_id.as_deref() == Some(recovered_turn_id) { + return Err(DurableRunnerError::invalid( + "ambiguous Codex turn recovery reused the previously completed turn identity", + )); + } + } + provider.restore_completed_turn_authority( + completed_turn_authoritative + && recovered_active_turn_id.is_none() + && !ambiguous_turn_start_pending, + completed_turn_process_generation, + completed_provider_turn_id.as_deref(), + ); self.provider = Some(provider); - if provider_had_exited || recovered_active_turn_id != previous_active_turn_id { + self.state + .as_mut() + .expect("Codex state remains available during recovery") + .provider_process_generation = process_generation; + if provider_had_exited + || ambiguous_turn_start_pending + || recovered_active_turn_id != previous_active_turn_id + { let state = self .state .as_mut() .expect("Codex state remains available during recovery"); - state.active_provider_turn_id = recovered_active_turn_id.clone(); - state.lifecycle = if recovered_active_turn_id.is_some() { - "turn_active".to_owned() - } else { - "session_open".to_owned() - }; + state.reconcile_active_provider_turn(recovered_active_turn_id.clone()); state.push_event(NormalizedProviderEvent { event_type: "session.reconciled".to_owned(), priority: EventPriority::P0, @@ -380,8 +465,8 @@ impl CodexCommandExecutor { "activeProviderTurnId": recovered_active_turn_id, }), })?; - self.save_state()?; } + self.save_state()?; Ok(()) } @@ -490,11 +575,29 @@ impl CodexCommandExecutor { "Codex provider session is closed", )); } - let provider = CodexProvider::start(&state.config, state.thread_id.as_deref()) - .map_err(|error| { - DurableRunnerError::invalid(format!("failed to start Codex provider: {error}")) - })?; + let process_generation = state + .provider_process_generation + .checked_add(1) + .ok_or_else(|| DurableRunnerError::invalid("Codex process generation exhausted"))?; + let mut provider = CodexProvider::start_for_generation( + &state.config, + state.thread_id.as_deref(), + process_generation, + ) + .map_err(|error| { + DurableRunnerError::invalid(format!("failed to start Codex provider: {error}")) + })?; + provider.restore_completed_turn_authority( + state.completed_turn_authoritative && provider.active_provider_turn_id().is_none(), + state.completed_turn_process_generation, + state.completed_provider_turn_id.as_deref(), + ); self.provider = Some(provider); + self.state + .as_mut() + .expect("Codex state remains available after provider start") + .provider_process_generation = process_generation; + self.save_state()?; } self.provider .as_mut() @@ -586,11 +689,46 @@ impl CodexCommandExecutor { .config .cwd .clone(); - let (provider_turn_id, thread_id) = { + self.ensure_provider()?; + { + let state = self + .state + .as_mut() + .expect("Codex state remains available before turn/start dispatch"); + state.ambiguous_turn_start_pending = true; + } + self.save_state()?; + let (start_result, completion_authority_retained, ambiguous_turn_start_pending) = { let provider = self.ensure_provider()?; - provider.start_turn(text, &cwd).map_err(|error| { - DurableRunnerError::invalid(format!("Codex turn/start failed: {error}")) - })?; + let result = provider.start_turn(text, &cwd); + ( + result, + provider.completed_turn_authority().is_some(), + provider.ambiguous_turn_start_pending(), + ) + }; + if let Err(error) = start_result { + let state = self + .state + .as_mut() + .expect("Codex state remains available after turn/start failure"); + state.ambiguous_turn_start_pending = ambiguous_turn_start_pending; + if !completion_authority_retained && !ambiguous_turn_start_pending { + state.completed_turn_authoritative = false; + state.completed_turn_process_generation = None; + state.completed_provider_turn_id = None; + state.last_agent_message = None; + } + self.save_state()?; + return Err(DurableRunnerError::invalid(format!( + "Codex turn/start failed: {error}" + ))); + } + let (provider_turn_id, thread_id) = { + let provider = self + .provider + .as_ref() + .expect("Codex provider remains available after turn/start acceptance"); ( provider .active_provider_turn_id() @@ -606,6 +744,10 @@ impl CodexCommandExecutor { .as_mut() .expect("Codex state exists after turn start"); state.active_provider_turn_id = Some(provider_turn_id.clone()); + state.ambiguous_turn_start_pending = false; + state.completed_turn_authoritative = false; + state.completed_turn_process_generation = None; + state.completed_provider_turn_id = None; state.last_agent_message = None; state.lifecycle = "turn_active".to_owned(); self.save_state()?; @@ -687,6 +829,7 @@ impl CodexCommandExecutor { .as_mut() .ok_or_else(|| DurableRunnerError::invalid("Codex provider is not prepared"))?; state.active_provider_turn_id = None; + state.ambiguous_turn_start_pending = false; state.lifecycle = "closed".to_owned(); let thread_id = state.thread_id.clone(); self.save_state()?; @@ -717,6 +860,11 @@ impl CodexCommandExecutor { fn poll_provider(&mut self) -> Result<(), DurableRunnerError> { self.restore()?; + // `restore_checked` records that the durable file was loaded even when + // provider recovery failed. Retry the provider reconciliation here so + // an ambiguous-start failure cannot degrade into an empty successful + // poll on the same executor. + self.restore_provider_if_needed()?; if self .state .as_ref() @@ -738,7 +886,32 @@ impl CodexCommandExecutor { })?; let Some(event) = event else { break }; match event { + CodexProviderEvent::ToolCall { + call_id, + operation_id, + .. + } => { + return Err(DurableRunnerError::invalid(format!( + "Codex emitted semantic tool call {call_id} for {operation_id} before the durable tool bridge was attached" + ))); + } CodexProviderEvent::Notification { method, params } => { + let active_provider_turn_id = if method == "turn/started" { + self.provider + .as_ref() + .and_then(CodexProvider::active_provider_turn_id) + .map(str::to_owned) + } else { + None + }; + let completed_turn_authority = if method == "turn/completed" { + self.provider + .as_ref() + .and_then(CodexProvider::completed_turn_authority) + .map(|(generation, turn_id)| (generation, turn_id.to_owned())) + } else { + None + }; let normalized = normalize_codex_notification(&method, ¶ms); let terminal_event_type = normalized .iter() @@ -767,8 +940,26 @@ impl CodexCommandExecutor { .map(|text| text.chars().take(1_000_000).collect()); } } + if method == "turn/started" { + let provider_turn_id = active_provider_turn_id.ok_or_else(|| { + DurableRunnerError::invalid( + "Codex turn start notification omitted active turn authority", + ) + })?; + state.reconcile_active_provider_turn(Some(provider_turn_id)); + } if method == "turn/completed" { + let (process_generation, provider_turn_id) = completed_turn_authority + .ok_or_else(|| { + DurableRunnerError::invalid( + "Codex completion omitted process and turn authority", + ) + })?; state.active_provider_turn_id = None; + state.completed_turn_authoritative = true; + state.completed_turn_process_generation = Some(process_generation); + state.completed_provider_turn_id = Some(provider_turn_id); + state.ambiguous_turn_start_pending = false; state.lifecycle = "session_open".to_owned(); } state.extend_events(normalized)?; @@ -811,18 +1002,49 @@ impl CodexCommandExecutor { })?; self.save_state()?; } - CodexProviderEvent::Exited { exit_code, success } => { + CodexProviderEvent::Exited { + exit_code, + success, + completed_turn_authoritative, + completed_turn_observed_by_process, + completion_reconciles_exit, + process_generation, + completed_turn_process_generation, + } => { self.provider = None; - if let Some(state) = self.state.as_mut() { + if !success { + let state = self + .state + .as_mut() + .expect("Codex state remains available while polling"); + // The durable terminal remains the run outcome. Use the + // provider's generation correlation only to decide + // whether this separate session exit belongs to that + // completion or is a later idle-provider failure. state.lifecycle = "provider_exited".to_owned(); state.push_event(NormalizedProviderEvent { - event_type: "session.failed".to_owned(), + // A completed turn remains authoritative, while the + // reusable provider session independently becomes + // unavailable. Avoid emitting session.failed for + // already successful work, but never leave the + // durable lifecycle open after a nonzero exit. + event_type: if completion_reconciles_exit { + "session.reconciled" + } else { + "session.failed" + } + .to_owned(), priority: EventPriority::P0, payload: json!({ "provider": "codex", "code": "provider_exited", "exitCode": exit_code, "expected": success, + "previousTurnCompleted": completed_turn_authoritative, + "completedByExitedProcess": completed_turn_observed_by_process, + "processGeneration": process_generation, + "completedTurnProcessGeneration": completed_turn_process_generation, + "activeProviderTurnId": Value::Null, }), })?; } @@ -941,6 +1163,11 @@ mod tests { thread_id: Some("thread-1".to_owned()), provider_session_id: None, active_provider_turn_id: None, + ambiguous_turn_start_pending: false, + completed_turn_authoritative: false, + provider_process_generation: 0, + completed_turn_process_generation: None, + completed_provider_turn_id: None, last_agent_message: None, pending_events: VecDeque::new(), next_provider_event_seq: initial_provider_event_seq(), @@ -948,6 +1175,45 @@ mod tests { assert!(state.validate().is_err()); } + #[test] + fn recovered_active_turn_revokes_prior_completion_authority() { + let mut state = CodexProviderState::new( + CodexProviderConfig { + provider: "codex".to_owned(), + driver: "codex_app_server".to_owned(), + provider_version: "test".to_owned(), + command: PathBuf::from("codex"), + args: vec!["app-server".to_owned()], + cwd: std::env::current_dir() + .unwrap() + .to_string_lossy() + .into_owned(), + model: None, + provider_session_id: None, + instructions: String::new(), + approval_policy: "never".to_owned(), + }, + None, + ); + state.thread_id = Some("thread-1".to_owned()); + state.lifecycle = "session_open".to_owned(); + state.completed_turn_authoritative = true; + state.provider_process_generation = 1; + state.completed_turn_process_generation = Some(1); + state.completed_provider_turn_id = Some("turn-1".to_owned()); + state.last_agent_message = Some("old turn output".to_owned()); + + state.reconcile_active_provider_turn(Some("turn-2".to_owned())); + + assert_eq!(state.lifecycle, "turn_active"); + assert_eq!(state.active_provider_turn_id.as_deref(), Some("turn-2")); + assert!(!state.completed_turn_authoritative); + assert!(state.completed_turn_process_generation.is_none()); + assert!(state.completed_provider_turn_id.is_none()); + assert!(state.last_agent_message.is_none()); + assert!(state.validate().is_ok()); + } + #[test] fn emits_a_structured_result_before_the_terminal_event() { let mut state = CodexProviderState::new( diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index 5fcd219971..08c3380360 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -7,6 +7,7 @@ use paperclip_runner_core::codex_provider::{ }; use paperclip_runner_core::durable::{Command, CommandExecutor, DurableRunnerError, PolledEvent}; use paperclip_runner_core::provider_backend::CodexCommandExecutor; +use paperclip_runner_core::provider_bridge::{AuthorizedTool, ToolResult}; use paperclip_runner_core::provider_events::normalize_codex_notification; use serde_json::{json, Value}; @@ -51,6 +52,16 @@ fn provider_config(directory: &Path, switches: &[&str]) -> CodexProviderConfig { } } +fn task_context_tool() -> AuthorizedTool { + AuthorizedTool { + operation_id: "get_task_context".to_owned(), + version: 1, + description: "Read task context.".to_owned(), + input_schema: json!({"type": "object"}), + response_schema: json!({"type": "object"}), + } +} + fn command(id: &str, sequence: u64, command_type: &str, payload: Value) -> Command { Command { schema: "paperclip.prp.command.v1".to_owned(), @@ -122,6 +133,1363 @@ fn codex_transport_buffers_notifications_while_waiting_for_responses() { fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); } +#[test] +fn codex_dynamic_tool_round_trips_through_the_provider_boundary() { + let directory = temporary_directory("dynamic-tool"); + let config = provider_config(&directory, &["--require-dynamic-tool", "--emit-tool-call"]); + let mut provider = CodexProvider::start_with_tools(&config, [task_context_tool()], None) + .expect("start Codex with an authorized tool"); + provider + .start_turn("Inspect the fake task.", &config.cwd) + .expect("start provider turn"); + + let mut delivered = false; + let mut completed = false; + for _ in 0..32 { + match provider.poll().expect("poll semantic tool event") { + Some(CodexProviderEvent::ToolCall { + call_id, + operation_id, + input, + }) => { + assert_eq!(call_id, "semantic-call-1"); + assert_eq!(operation_id, "get_task_context"); + assert_eq!(input, json!({})); + assert!(provider + .deliver_tool_result(&ToolResult { + call_id: call_id.clone(), + operation_id: "another_operation".to_owned(), + result: json!({"ok": true}), + is_error: false, + }) + .is_err()); + assert!(provider + .deliver_tool_result(&ToolResult { + call_id: call_id.clone(), + operation_id: operation_id.clone(), + result: json!({"value": "x".repeat(1024 * 1024)}), + is_error: false, + }) + .is_err()); + provider + .deliver_tool_result(&ToolResult { + call_id, + operation_id, + result: json!({"ok": true, "task": {"id": "task-1"}}), + is_error: false, + }) + .expect("deliver correlated semantic result"); + delivered = true; + } + Some(CodexProviderEvent::Notification { method, .. }) if method == "turn/completed" => { + completed = true; + break; + } + _ => {} + } + } + assert!(delivered, "Codex emitted its authorized tool call"); + assert!(completed, "Codex completed after the semantic result"); + provider.shutdown().expect("stop provider"); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn codex_rejects_replay_of_a_completed_tool_call_id_in_the_same_turn() { + let directory = temporary_directory("completed-tool-call-replay"); + let config = provider_config( + &directory, + &[ + "--require-dynamic-tool", + "--emit-tool-call", + "--replay-completed-tool-call", + ], + ); + let mut provider = CodexProvider::start_with_tools(&config, [task_context_tool()], None) + .expect("start Codex with an authorized tool"); + provider + .start_turn("Inspect the fake task once.", &config.cwd) + .expect("start provider turn"); + + let first_call = (0..32) + .find_map(|_| match provider.poll().expect("poll first tool call") { + Some(CodexProviderEvent::ToolCall { + call_id, + operation_id, + .. + }) => Some((call_id, operation_id)), + _ => None, + }) + .expect("observe the first semantic tool call"); + provider + .deliver_tool_result(&ToolResult { + call_id: first_call.0, + operation_id: first_call.1, + result: json!({"ok": true, "task": {"id": "task-1"}}), + is_error: false, + }) + .expect("deliver the first semantic result"); + + let replay_error = (0..32) + .find_map(|_| provider.poll().err()) + .expect("same-turn replay of the completed call id is rejected"); + assert!( + replay_error + .to_string() + .contains("reused a completed tool call id"), + "unexpected replay error: {replay_error}" + ); + + let _ = provider.shutdown(); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn codex_completion_cancels_pending_tool_request_before_releasing_capacity() { + let directory = temporary_directory("completed-tool-call"); + let config = provider_config( + &directory, + &[ + "--require-dynamic-tool", + "--emit-tool-call", + "--complete-after-tool-call", + ], + ); + let mut provider = CodexProvider::start_with_tools(&config, [task_context_tool()], None) + .expect("start Codex with an authorized tool"); + provider + .start_turn("Complete without waiting for the tool result.", &config.cwd) + .expect("start provider turn"); + + let first_call = (0..32) + .find_map( + |_| match provider.poll().expect("poll first provider turn") { + Some(CodexProviderEvent::ToolCall { + call_id, + operation_id, + .. + }) => Some((call_id, operation_id)), + _ => None, + }, + ) + .expect("observe the first semantic tool call"); + let completed = (0..32).any(|_| { + matches!( + provider.poll().expect("poll first completion"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!(completed, "Codex completed with a tool call still pending"); + for _ in 0..100 { + if call_count(&directory, "tool-response:failure") == 1 { + break; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + assert_eq!( + call_count(&directory, "tool-response:failure"), + 1, + "Paperclip explicitly resolves the provider RPC as cancelled", + ); + assert!(provider + .deliver_tool_result(&ToolResult { + call_id: first_call.0, + operation_id: first_call.1, + result: json!({"ok": true}), + is_error: false, + }) + .is_err()); + + provider + .start_turn("Reuse the released provider identities.", &config.cwd) + .expect("start another provider turn"); + let second_call = (0..32).any(|_| { + matches!( + provider.poll().expect("poll second provider turn"), + Some(CodexProviderEvent::ToolCall { call_id, .. }) + if call_id == "semantic-call-1" + ) + }); + assert!(second_call, "the next turn can reuse the released call id"); + + provider.shutdown().expect("stop provider"); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn codex_completion_survives_failed_pending_request_cancellation() { + let directory = temporary_directory("completed-tool-call-provider-exit"); + let config = provider_config( + &directory, + &[ + "--require-dynamic-tool", + "--emit-tool-call", + "--complete-after-tool-call", + "--exit-after-tool-call-completion", + ], + ); + let mut provider = CodexProvider::start_with_tools(&config, [task_context_tool()], None) + .expect("start Codex with an authorized tool"); + provider + .start_turn("Complete and exit with a tool call pending.", &config.cwd) + .expect("start provider turn"); + + let call = (0..32) + .find_map(|_| match provider.poll().expect("poll pending tool call") { + Some(CodexProviderEvent::ToolCall { + call_id, + operation_id, + .. + }) => Some((call_id, operation_id)), + _ => None, + }) + .expect("observe the pending semantic tool call"); + std::thread::sleep(std::time::Duration::from_millis(50)); + + let completed = (0..32).any(|_| { + matches!( + provider + .poll() + .expect("the received completion survives closed provider stdin"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!(completed, "the terminal notification remains authoritative"); + assert!(provider + .deliver_tool_result(&ToolResult { + call_id: call.0, + operation_id: call.1, + result: json!({"ok": true}), + is_error: false, + }) + .is_err()); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn clean_idle_provider_exit_preserves_completed_turn_success() { + let directory = temporary_directory("completion-output-clean-provider-exit"); + let config = provider_config( + &directory, + &[ + "--emit-post-completion-warning", + "--exit-after-turn-completion", + ], + ); + let mut provider = CodexProvider::start(&config, None).expect("start Codex provider"); + provider + .start_turn("Complete, produce idle output, then exit.", &config.cwd) + .expect("start provider turn"); + + let mut completion_seen = false; + let mut post_completion_output_seen = false; + let mut clean_exit = None; + let exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while std::time::Instant::now() < exit_deadline { + match provider.poll().expect("poll completion and clean exit") { + Some(CodexProviderEvent::Notification { method, .. }) => { + completion_seen |= method == "turn/completed"; + post_completion_output_seen |= completion_seen && method == "warning"; + } + Some(CodexProviderEvent::Exited { + success, + completed_turn_authoritative, + completion_reconciles_exit, + .. + }) => { + clean_exit = Some(( + success, + completed_turn_authoritative, + completion_reconciles_exit, + )); + break; + } + Some(_) | None => {} + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + + assert!(completion_seen); + assert!(post_completion_output_seen); + assert_eq!(clean_exit, Some((true, true, false))); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn clean_provider_exit_does_not_refail_a_completed_turn() { + let directory = temporary_directory("completion-then-clean-exit"); + let config = provider_config( + &directory, + &[ + "--emit-post-completion-warning", + "--exit-after-turn-completion", + "--exit-after-thread-read", + ], + ); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "turn", + 3, + "turn.start", + json!({"text": "Complete before exiting cleanly."}), + )) + .expect("start provider turn"); + + let mut event_types = Vec::new(); + let exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while std::time::Instant::now() < exit_deadline { + event_types.extend( + poll_and_ack(&mut executor) + .expect("poll completion and clean exit") + .into_iter() + .map(|event| event.event_type), + ); + if event_types.iter().any(|event| event == "turn.completed") + && event_types + .iter() + .any(|event| event == "provider.notice.recorded") + { + break; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + + // The warning is ordered after the terminal and before the clean exit, so + // seeing it proves the completed process entered the idle/output path that + // previously cleared expected shutdown authority. + assert!(event_types.iter().any(|event| event == "turn.completed")); + assert!(event_types + .iter() + .any(|event| event == "provider.notice.recorded")); + for _ in 0..32 { + event_types.extend( + poll_and_ack(&mut executor) + .expect("poll after provider exit") + .into_iter() + .map(|event| event.event_type), + ); + } + + assert!(!event_types.iter().any(|event| event == "session.failed")); + let persisted: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read provider state after clean exit"), + ) + .expect("parse provider state after clean exit"); + assert_eq!(persisted["lifecycle"], "session_open"); + assert!(persisted["activeProviderTurnId"].is_null()); + + drop(executor); + let mut recovered = CodexCommandExecutor::new(&directory); + let recovered_events = poll_and_ack(&mut recovered) + .expect("poll clean exit from a freshly resumed completed thread"); + assert!(!recovered_events + .iter() + .any(|event| event.event_type == "session.failed")); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn post_completion_observation_does_not_hide_same_or_resumed_process_failure() { + let directory = temporary_directory("completion-then-nonzero-exit"); + let config = provider_config( + &directory, + &[ + "--emit-post-completion-warning", + "--fail-after-turn-completion", + "--fail-after-turn-completion-delay-ms", + "250", + "--fail-after-thread-read", + ], + ); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "turn", + 3, + "turn.start", + json!({"text": "Complete before exiting with an error."}), + )) + .expect("start provider turn"); + + let mut event_types = Vec::new(); + let first_exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while std::time::Instant::now() < first_exit_deadline { + event_types.extend( + poll_and_ack(&mut executor) + .expect("poll completion and nonzero exit") + .into_iter() + .map(|event| event.event_type), + ); + if event_types.iter().any(|event| event == "turn.completed") + && event_types + .iter() + .any(|event| event == "provider.notice.recorded") + && event_types.iter().any(|event| event == "session.failed") + { + break; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + + assert!(event_types.iter().any(|event| event == "turn.completed")); + assert!(event_types + .iter() + .any(|event| event == "provider.notice.recorded")); + assert!(!event_types + .iter() + .any(|event| event == "session.reconciled")); + assert!(event_types.iter().any(|event| event == "session.failed")); + assert!(!event_types.iter().any(|event| event == "turn.failed")); + let persisted: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read provider state after nonzero exit"), + ) + .expect("parse provider state after nonzero exit"); + assert_eq!(persisted["lifecycle"], "provider_exited"); + assert_eq!(persisted["completedTurnAuthoritative"], true); + assert_eq!(persisted["providerProcessGeneration"], 1); + assert_eq!(persisted["completedTurnProcessGeneration"], 1); + + // A fresh process restores the durable completed turn, probes it with + // thread/read, and then exits nonzero. Recovery preserves the completed + // run outcome, while the later idle provider failure remains visible. + let mut recovered = CodexCommandExecutor::new(&directory); + let mut recovered_event_types = Vec::new(); + let recovered_exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while std::time::Instant::now() < recovered_exit_deadline { + recovered_event_types.extend( + poll_and_ack(&mut recovered) + .expect("poll restored provider after idle crash") + .into_iter() + .map(|event| event.event_type), + ); + if recovered_event_types + .iter() + .any(|event| event == "session.failed") + { + break; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + assert!(recovered_event_types + .iter() + .any(|event| event == "session.failed")); + assert!(!recovered_event_types + .iter() + .any(|event| event == "turn.failed")); + assert_eq!( + recovered_event_types + .iter() + .filter(|event| event.as_str() == "session.reconciled") + .count(), + 1, + "recovery is reconciled once, but the resumed provider exit fails its session" + ); + let recovered_persisted: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read provider state after resumed exit"), + ) + .expect("parse provider state after resumed exit"); + assert_eq!(recovered_persisted["lifecycle"], "provider_exited"); + assert_eq!(recovered_persisted["providerProcessGeneration"], 2); + assert_eq!(recovered_persisted["completedTurnProcessGeneration"], 1); + assert_eq!(call_count(&directory, "thread/read"), 1); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn rejected_replacement_turn_start_preserves_result_and_exit_authority() { + let directory = temporary_directory("completion-then-rejected-turn-start"); + let config = provider_config(&directory, &["--reject-second-turn-start"]); + let mut provider = CodexProvider::start(&config, None).expect("start Codex provider"); + provider + .start_turn("Complete the first turn.", &config.cwd) + .expect("start first provider turn"); + let first_completed = (0..32).any(|_| { + matches!( + provider.poll().expect("poll first turn"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!( + first_completed, + "observe the authoritative first completion" + ); + + provider + .start_turn("Reject replacement work.", &config.cwd) + .expect_err("the replacement turn/start returns a definite rejection"); + let mut buffered_notification_seen = false; + let rejected_start_exit = (0..64).find_map(|_| { + match provider + .poll() + .expect("poll exit after rejected replacement start") + { + Some(CodexProviderEvent::Notification { method, params }) + if method == "warning" + && params.get("message").and_then(Value::as_str) + == Some("buffered before replacement rejection") => + { + buffered_notification_seen = true; + None + } + Some(CodexProviderEvent::Exited { + success, + completed_turn_authoritative, + completion_reconciles_exit, + .. + }) => Some(( + success, + completed_turn_authoritative, + completion_reconciles_exit, + )), + _ => None, + } + }); + assert!(buffered_notification_seen); + assert_eq!(rejected_start_exit, Some((false, true, false))); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn rejected_replacement_turn_start_does_not_hide_contradictory_turn_evidence() { + let directory = temporary_directory("completion-then-contradictory-rejection"); + let config = provider_config( + &directory, + &[ + "--reject-second-turn-start", + "--emit-turn-before-rejected-second-start", + ], + ); + let mut provider = CodexProvider::start(&config, None).expect("start Codex provider"); + provider + .start_turn("Complete the first turn.", &config.cwd) + .expect("start first provider turn"); + let first_completed = (0..32).any(|_| { + matches!( + provider.poll().expect("poll first turn"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!( + first_completed, + "observe the authoritative first completion" + ); + + provider + .start_turn( + "Reject replacement work after contradictory evidence.", + &config.cwd, + ) + .expect_err("the replacement turn/start returns a definite rejection"); + let duplicate_error = provider + .start_turn( + "Do not duplicate contradictory replacement work.", + &config.cwd, + ) + .expect_err("provider-work evidence makes the rejected response ambiguous"); + assert!( + duplicate_error + .to_string() + .contains("unresolved ambiguous provider turn start"), + "unexpected duplicate-start error: {duplicate_error}" + ); + let mut contradictory_turn_seen = false; + let rejected_start_exit = (0..64).find_map(|_| { + match provider + .poll() + .expect("poll exit after contradictory replacement rejection") + { + Some(CodexProviderEvent::Notification { method, params }) + if method == "turn/started" + && params.pointer("/turn/id").and_then(Value::as_str) + == Some("provider-turn-contradiction") => + { + contradictory_turn_seen = true; + None + } + Some(CodexProviderEvent::Exited { + success, + completed_turn_authoritative, + completion_reconciles_exit, + .. + }) => Some(( + success, + completed_turn_authoritative, + completion_reconciles_exit, + )), + _ => None, + } + }); + assert!(contradictory_turn_seen); + assert_eq!( + provider.active_provider_turn_id(), + Some("provider-turn-contradiction") + ); + assert_eq!(rejected_start_exit, Some((false, false, false))); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn ambiguous_or_dead_replacement_start_preserves_result_not_exit_authority() { + for (label, switch) in [ + ( + "accepted-before-response", + "--fail-after-accepting-second-turn-before-response", + ), + ("malformed-error", "--malformed-error-second-turn-start"), + ("missing-turn-id", "--missing-id-second-turn-start"), + ] { + let directory = temporary_directory(label); + let config = provider_config(&directory, &[switch]); + let mut provider = CodexProvider::start(&config, None).expect("start Codex provider"); + provider + .start_turn("Complete the first turn.", &config.cwd) + .expect("start first provider turn"); + let first_completed = (0..32).any(|_| { + matches!( + provider.poll().expect("poll first turn"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!( + first_completed, + "observe the authoritative first completion for {label}" + ); + + provider + .start_turn("Accept replacement work before failing.", &config.cwd) + .expect_err("the accepted replacement turn has no valid response"); + let ambiguous_start_exit = (0..64).find_map(|_| { + match provider + .poll() + .expect("poll exit after ambiguous replacement start") + { + Some(CodexProviderEvent::Exited { + success, + completed_turn_authoritative, + completion_reconciles_exit, + .. + }) => Some(( + success, + completed_turn_authoritative, + completion_reconciles_exit, + )), + _ => None, + } + }); + assert_eq!( + ambiguous_start_exit, + Some((false, true, false)), + "{label} must retain the completed result without hiding the provider failure" + ); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); + } +} + +#[test] +fn ambiguous_replacement_turn_adopts_one_later_completion_identity() { + for (label, switch, omit_started) in [ + ( + "accepted-before-response-with-completion", + "--fail-after-accepting-second-turn-before-response", + false, + ), + ( + "malformed-error-with-completion", + "--malformed-error-second-turn-start", + false, + ), + ( + "missing-turn-id-with-completion", + "--missing-id-second-turn-start", + true, + ), + ] { + let directory = temporary_directory(label); + let mut switches = vec![switch, "--complete-ambiguous-second-turn"]; + if omit_started { + switches.push("--omit-ambiguous-turn-started"); + } + let config = provider_config(&directory, &switches); + let mut provider = CodexProvider::start(&config, None).expect("start Codex provider"); + provider + .start_turn("Complete the first turn.", &config.cwd) + .expect("start first provider turn"); + let first_completed = (0..32).any(|_| { + matches!( + provider.poll().expect("poll first turn"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!( + first_completed, + "observe the authoritative first completion for {label}" + ); + + provider + .start_turn("Complete accepted replacement work.", &config.cwd) + .expect_err("the accepted replacement turn has no valid response"); + let unresolved_error = provider + .start_turn("Do not start duplicate replacement work.", &config.cwd) + .expect_err("an unresolved ambiguous start bounds replacement work to one turn"); + assert!( + unresolved_error + .to_string() + .contains("unresolved ambiguous provider turn start"), + "unexpected unresolved-start error for {label}: {unresolved_error}" + ); + + let mut replacement_started = false; + let mut replacement_completed = false; + let replacement_exit = (0..128).find_map(|_| { + match provider + .poll() + .expect("poll evidence for accepted replacement turn") + { + Some(CodexProviderEvent::Notification { method, params }) + if method == "turn/started" => + { + assert_eq!( + params.pointer("/turn/id").and_then(Value::as_str), + Some("provider-turn-2") + ); + replacement_started = true; + None + } + Some(CodexProviderEvent::Notification { method, params }) + if method == "turn/completed" => + { + assert_eq!( + params.pointer("/turn/id").and_then(Value::as_str), + Some("provider-turn-2") + ); + replacement_completed = true; + None + } + Some(CodexProviderEvent::Exited { + success, + completed_turn_authoritative, + completion_reconciles_exit, + .. + }) => Some(( + success, + completed_turn_authoritative, + completion_reconciles_exit, + )), + _ => None, + } + }); + assert!( + replacement_started, + "the replacement identity should be established before replaying its output for {label}" + ); + assert!( + replacement_completed, + "observe replacement completion for {label}" + ); + assert_eq!( + replacement_exit, + Some((false, true, true)), + "the replacement completion, not the old result, reconciles the provider exit for {label}" + ); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); + } +} + +#[test] +fn ambiguous_replacement_turn_rejects_conflicting_later_identity() { + let directory = temporary_directory("conflicting-ambiguous-turn-identities"); + let config = provider_config( + &directory, + &[ + "--missing-id-second-turn-start", + "--conflicting-ambiguous-second-turn", + ], + ); + let mut provider = CodexProvider::start(&config, None).expect("start Codex provider"); + provider + .start_turn("Complete the first turn.", &config.cwd) + .expect("start first provider turn"); + let first_completed = (0..32).any(|_| { + matches!( + provider.poll().expect("poll first turn"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!( + first_completed, + "observe the authoritative first completion" + ); + + provider + .start_turn("Accept replacement work ambiguously.", &config.cwd) + .expect_err("the replacement response omits its turn identity"); + let replacement_started = provider + .poll() + .expect("poll replacement start") + .expect("replacement start is available"); + assert!(matches!( + replacement_started, + CodexProviderEvent::Notification { method, params } + if method == "turn/started" + && params.pointer("/turn/id").and_then(Value::as_str) + == Some("provider-turn-2") + )); + assert_eq!(provider.active_provider_turn_id(), Some("provider-turn-2")); + + let conflicting_completion = provider + .poll() + .expect_err("a second replacement identity must fail closed"); + assert!( + conflicting_completion + .to_string() + .contains("another active turn"), + "unexpected conflicting-identity error: {conflicting_completion}" + ); + assert_eq!(provider.active_provider_turn_id(), Some("provider-turn-2")); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn clean_exit_after_ambiguous_replacement_start_fails_the_durable_session() { + let directory = temporary_directory("durable-clean-exit-after-ambiguous-turn-start"); + let config = provider_config( + &directory, + &["--exit-after-accepting-second-turn-before-response"], + ); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "first-turn", + 3, + "turn.start", + json!({"text": "Complete the first turn."}), + )) + .expect("start first provider turn"); + + let mut first_events = Vec::new(); + for _ in 0..32 { + first_events.extend( + poll_and_ack(&mut executor) + .expect("poll first turn") + .into_iter() + .map(|event| event.event_type), + ); + if first_events.iter().any(|event| event == "turn.completed") { + break; + } + } + assert!(first_events.iter().any(|event| event == "turn.completed")); + + executor + .execute(&command( + "ambiguous-turn", + 4, + "turn.start", + json!({"text": "Accept replacement work before exiting cleanly."}), + )) + .expect_err("accepted replacement start loses its response"); + let persisted_after_start: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read provider state after ambiguous start"), + ) + .expect("parse provider state after ambiguous start"); + assert_eq!(persisted_after_start["completedTurnAuthoritative"], true); + assert_eq!(persisted_after_start["completedTurnProcessGeneration"], 1); + assert_eq!( + persisted_after_start["completedProviderTurnId"], + "provider-turn-1" + ); + + let mut exit_events = Vec::new(); + for _ in 0..64 { + exit_events.extend( + poll_and_ack(&mut executor) + .expect("poll provider after ambiguous start") + .into_iter() + .map(|event| event.event_type), + ); + if exit_events.iter().any(|event| event == "session.failed") { + break; + } + } + assert!(exit_events.iter().any(|event| event == "session.failed")); + assert!(!exit_events + .iter() + .any(|event| event == "session.reconciled")); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn durable_ambiguous_start_recovers_a_distinct_active_replacement_after_process_loss() { + let directory = temporary_directory("durable-ambiguous-active-recovery"); + let config = provider_config( + &directory, + &[ + "--fail-after-accepting-second-turn-before-response", + "--retain-ambiguous-second-turn-active", + ], + ); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "first-turn", + 3, + "turn.start", + json!({"text": "Complete the first turn."}), + )) + .expect("start first provider turn"); + for _ in 0..32 { + if poll_and_ack(&mut executor) + .expect("poll first turn") + .iter() + .any(|event| event.event_type == "turn.completed") + { + break; + } + } + + executor + .execute(&command( + "ambiguous-turn", + 4, + "turn.start", + json!({"text": "Accept replacement work without returning its identity."}), + )) + .expect_err("replacement acceptance loses its response"); + let persisted_ambiguous: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read provider state after ambiguous start"), + ) + .expect("parse provider state after ambiguous start"); + assert_eq!(persisted_ambiguous["ambiguousTurnStartPending"], true); + assert_eq!( + persisted_ambiguous["completedProviderTurnId"], + "provider-turn-1" + ); + + executor.shutdown().expect("stop first provider process"); + drop(executor); + + let mut recovered = CodexCommandExecutor::new(&directory); + let snapshot = recovered + .execute(&command("snapshot", 5, "session.snapshot", json!({}))) + .expect("reconcile active replacement turn"); + assert_eq!(snapshot.result["status"], "turn_active"); + assert_eq!(snapshot.result["activeProviderTurnId"], "provider-turn-2"); + let persisted_recovered: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read recovered provider state"), + ) + .expect("parse recovered provider state"); + assert_eq!(persisted_recovered["ambiguousTurnStartPending"], false); + assert_eq!(persisted_recovered["completedTurnAuthoritative"], false); + assert!(persisted_recovered["completedProviderTurnId"].is_null()); + + recovered + .shutdown() + .expect("stop recovered provider process"); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn replacement_item_is_not_persisted_before_ambiguous_turn_identity() { + let directory = temporary_directory("durable-ambiguous-item-recovery"); + let config = provider_config( + &directory, + &[ + "--missing-id-second-turn-start", + "--hold-ambiguous-second-turn-after-item", + ], + ); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "first-turn", + 3, + "turn.start", + json!({"text": "Complete the first turn."}), + )) + .expect("start first provider turn"); + for _ in 0..32 { + if poll_and_ack(&mut executor) + .expect("poll first turn") + .iter() + .any(|event| event.event_type == "turn.completed") + { + break; + } + } + + executor + .execute(&command( + "ambiguous-turn", + 4, + "turn.start", + json!({"text": "Emit replacement output before terminal authority."}), + )) + .expect_err("replacement response omits its identity"); + assert!( + poll_and_ack(&mut executor) + .expect("defer identity-less replacement output") + .is_empty(), + "replacement output must not escape before its turn identity" + ); + let persisted_before_loss: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read provider state before process loss"), + ) + .expect("parse provider state before process loss"); + assert_eq!(persisted_before_loss["ambiguousTurnStartPending"], true); + assert_ne!( + persisted_before_loss["lastAgentMessage"], + "Replacement output before terminal authority." + ); + + executor.shutdown().expect("stop first provider process"); + drop(executor); + + let mut recovered = CodexCommandExecutor::new(&directory); + let snapshot = recovered + .execute(&command("snapshot", 5, "session.snapshot", json!({}))) + .expect("reconcile the still-active replacement turn"); + assert_eq!(snapshot.result["status"], "turn_active"); + assert_eq!(snapshot.result["activeProviderTurnId"], "provider-turn-2"); + let persisted_recovered: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read recovered provider state"), + ) + .expect("parse recovered provider state"); + assert_eq!(persisted_recovered["ambiguousTurnStartPending"], false); + assert_ne!( + persisted_recovered["lastAgentMessage"], + "Replacement output before terminal authority." + ); + + recovered + .shutdown() + .expect("stop recovered provider process"); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn completed_ambiguous_replacement_fails_closed_after_process_loss() { + let directory = temporary_directory("durable-ambiguous-completed-recovery"); + let config = provider_config( + &directory, + &[ + "--missing-id-second-turn-start", + "--complete-ambiguous-second-turn-before-response", + "--omit-ambiguous-turn-started", + ], + ); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "first-turn", + 3, + "turn.start", + json!({"text": "Complete the first turn."}), + )) + .expect("start first provider turn"); + for _ in 0..32 { + if poll_and_ack(&mut executor) + .expect("poll first turn") + .iter() + .any(|event| event.event_type == "turn.completed") + { + break; + } + } + + executor + .execute(&command( + "ambiguous-turn", + 4, + "turn.start", + json!({"text": "Complete replacement work before returning an invalid response."}), + )) + .expect_err("replacement response omits its identity"); + executor.shutdown().expect("stop first provider process"); + drop(executor); + + let mut recovered = CodexCommandExecutor::new(&directory); + let recovery_error = recovered + .execute(&command("snapshot", 5, "session.snapshot", json!({}))) + .expect_err("completed ambiguous work without durable identity must fail closed"); + assert!( + recovery_error + .to_string() + .contains("cannot safely recover an ambiguous Codex turn start"), + "unexpected ambiguous recovery error: {recovery_error}" + ); + let repeated_poll_error = recovered + .poll_events() + .expect_err("a repeated poll must retain the fail-closed recovery state"); + assert!( + repeated_poll_error + .to_string() + .contains("cannot safely recover an ambiguous Codex turn start"), + "unexpected repeated ambiguous recovery error: {repeated_poll_error}" + ); + assert_eq!(call_count(&directory, "turn/start"), 2); + let persisted: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read fail-closed provider state"), + ) + .expect("parse fail-closed provider state"); + assert_eq!(persisted["ambiguousTurnStartPending"], true); + assert_eq!(persisted["completedProviderTurnId"], "provider-turn-1"); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn ambiguous_replacement_completion_replaces_durable_turn_authority() { + let directory = temporary_directory("durable-ambiguous-turn-completion"); + let config = provider_config( + &directory, + &[ + "--missing-id-second-turn-start", + "--complete-ambiguous-second-turn", + ], + ); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "first-turn", + 3, + "turn.start", + json!({"text": "Complete the first turn."}), + )) + .expect("start first provider turn"); + + let mut first_events = Vec::new(); + for _ in 0..32 { + first_events.extend( + poll_and_ack(&mut executor) + .expect("poll first turn") + .into_iter() + .map(|event| event.event_type), + ); + if first_events.iter().any(|event| event == "turn.completed") { + break; + } + } + assert!(first_events.iter().any(|event| event == "turn.completed")); + + executor + .execute(&command( + "ambiguous-turn", + 4, + "turn.start", + json!({"text": "Complete replacement work after the malformed response."}), + )) + .expect_err("accepted replacement response omits its turn identity"); + + let mut replacement_events = Vec::new(); + for _ in 0..64 { + replacement_events.extend( + poll_and_ack(&mut executor) + .expect("poll accepted replacement evidence") + .into_iter() + .map(|event| event.event_type), + ); + if replacement_events + .iter() + .any(|event| event == "session.reconciled") + { + break; + } + } + assert!(replacement_events + .iter() + .any(|event| event == "turn.started")); + assert!(replacement_events + .iter() + .any(|event| event == "turn.completed")); + assert!(replacement_events + .iter() + .any(|event| event == "session.reconciled")); + + let persisted: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")) + .expect("read provider state after replacement completion"), + ) + .expect("parse provider state after replacement completion"); + assert_eq!(persisted["activeProviderTurnId"], Value::Null); + assert_eq!(persisted["completedTurnAuthoritative"], true); + assert_eq!(persisted["completedTurnProcessGeneration"], 1); + assert_eq!(persisted["completedProviderTurnId"], "provider-turn-2"); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn accepted_replacement_turn_revokes_prior_authority_before_idle_crash() { + let directory = temporary_directory("completion-then-new-turn-failure"); + let config = provider_config(&directory, &["--fail-after-second-turn-start"]); + let mut provider = CodexProvider::start(&config, None).expect("start Codex provider"); + provider + .start_turn("Complete the first turn.", &config.cwd) + .expect("start first provider turn"); + let first_completed = (0..32).any(|_| { + matches!( + provider.poll().expect("poll first turn"), + Some(CodexProviderEvent::Notification { method, .. }) + if method == "turn/completed" + ) + }); + assert!( + first_completed, + "observe the authoritative first completion" + ); + + provider + .start_turn("Start genuinely new provider work.", &config.cwd) + .expect("start second provider turn"); + let second_exit = (0..64).find_map(|_| match provider.poll().expect("poll second turn") { + Some(CodexProviderEvent::Exited { + success, + completed_turn_authoritative, + completion_reconciles_exit, + .. + }) => Some(( + success, + completed_turn_authoritative, + completion_reconciles_exit, + )), + _ => None, + }); + assert_eq!(second_exit, Some((false, false, false))); + + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn codex_rejects_a_tool_call_that_was_not_advertised() { + let directory = temporary_directory("unauthorized-tool"); + let config = provider_config(&directory, &["--emit-tool-call"]); + let mut provider = CodexProvider::start(&config, None).expect("start Codex without tools"); + provider + .start_turn("Attempt an unavailable tool.", &config.cwd) + .expect("start provider turn"); + let error = (0..32) + .find_map(|_| provider.poll().err()) + .expect("unauthorized provider tool call is rejected"); + assert!(error.to_string().contains("unauthorized tool")); + let _ = provider.shutdown(); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn codex_resume_advertises_the_same_authorized_tools() { + let directory = temporary_directory("dynamic-tool-resume"); + let config = provider_config(&directory, &["--require-dynamic-tool"]); + let mut provider = + CodexProvider::start_with_tools(&config, [task_context_tool()], Some("codex-thread-1")) + .expect("resume Codex with the run-scoped tool set"); + assert_eq!(provider.thread_id(), "codex-thread-1"); + provider.shutdown().expect("stop provider"); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + #[test] fn durable_backend_resumes_the_active_thread_without_restarting_the_turn() { let directory = temporary_directory("resume");