From 2b40e9d771d70cd37521d49e6ceb0fc9655c093b Mon Sep 17 00:00:00 2001 From: guoxu1 Date: Mon, 7 Sep 2026 17:15:17 +0800 Subject: [PATCH] feat(replay): add Codex resume bridge and replay adapters --- .../persisting-overlaynet/src/interception.rs | 18 +- crates/persisting-pvisor/src/cli/replay.rs | 2 + .../assets/pi_agent_runner.mjs | 15 +- .../persisting-replay/src/adapter/generic.rs | 2063 +++++++++++++++++ crates/persisting-replay/src/adapter/mod.rs | 13 + .../persisting-replay/src/adapter/runtime.rs | 36 +- crates/persisting-replay/src/codex_bridge.rs | 761 ++++++ crates/persisting-replay/src/engine.rs | 36 + crates/persisting-replay/src/lib.rs | 1 + crates/persisting-replay/src/model.rs | 18 +- docs/src/en/pvisor/guides/sandbox-replay.md | 52 +- docs/src/en/pvisor/reference/cli.md | 22 +- docs/src/zh/pvisor/guides/sandbox-replay.md | 133 ++ docs/src/zh/pvisor/reference/cli.md | 20 +- 14 files changed, 3164 insertions(+), 26 deletions(-) create mode 100644 crates/persisting-replay/src/adapter/generic.rs create mode 100644 crates/persisting-replay/src/codex_bridge.rs diff --git a/crates/persisting-overlaynet/src/interception.rs b/crates/persisting-overlaynet/src/interception.rs index 3225a1d0..60166d6e 100644 --- a/crates/persisting-overlaynet/src/interception.rs +++ b/crates/persisting-overlaynet/src/interception.rs @@ -198,11 +198,19 @@ impl InterceptionMetrics { } pub(crate) fn tcp_flow_closed(&self) { - let _ = self.counters.active_tcp_flows.try_update( - Ordering::Relaxed, - Ordering::Relaxed, - |active| active.checked_sub(1), - ); + let active_tcp_flows = &self.counters.active_tcp_flows; + let mut current = active_tcp_flows.load(Ordering::Relaxed); + while current > 0 { + match active_tcp_flows.compare_exchange_weak( + current, + current - 1, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(next) => current = next, + } + } } pub(crate) fn tcp_flow_denied(&self) { diff --git a/crates/persisting-pvisor/src/cli/replay.rs b/crates/persisting-pvisor/src/cli/replay.rs index 566d1a17..d777ecea 100644 --- a/crates/persisting-pvisor/src/cli/replay.rs +++ b/crates/persisting-pvisor/src/cli/replay.rs @@ -69,6 +69,8 @@ pub struct ReplayArgs { #[arg(long, value_name = "DIR")] output_dir: Option, + /// Model-router/run session key. Codex native continuation identity is + /// derived from the trajectory and is never taken from this field. #[arg(long)] session_id: Option, diff --git a/crates/persisting-replay/assets/pi_agent_runner.mjs b/crates/persisting-replay/assets/pi_agent_runner.mjs index 250a95d8..f353ab11 100644 --- a/crates/persisting-replay/assets/pi_agent_runner.mjs +++ b/crates/persisting-replay/assets/pi_agent_runner.mjs @@ -4,6 +4,11 @@ import fs from "node:fs"; import path from "node:path"; import { pathToFileURL } from "node:url"; +// Pi runtimes may be launched with Node 16/18 in older sandboxes where the +// global structuredClone helper is unavailable. JSON is sufficient for the +// native event objects we copy here and keeps replay portable across runtimes. +const clone = globalThis.structuredClone ?? ((value) => JSON.parse(JSON.stringify(value))); + function load(filename) { return JSON.parse(fs.readFileSync(filename, "utf8")); } @@ -181,15 +186,15 @@ async function run(request) { for (const event of events) { if (event?.type === "message_end" && event?.message?.role === "user") { - sessionManager.appendMessage(structuredClone(event.message)); - reconstructedEvents.push(structuredClone(event)); + sessionManager.appendMessage(clone(event.message)); + reconstructedEvents.push(clone(event)); continue; } if (event?.type !== "turn_end" || event?.message?.role !== "assistant") continue; const calls = toolCalls(event.message); if (calls.length > 0 && replayedBatches >= request.after_step) break; prefixTurns += 1; - const assistant = structuredClone(event.message); + const assistant = clone(event.message); sessionManager.appendMessage(assistant); const freshResults = []; for (const call of calls) { @@ -198,7 +203,7 @@ async function run(request) { observations.push(fresh.observation); sessionManager.appendMessage(fresh.message); } - reconstructedEvents.push({ ...structuredClone(event), toolResults: freshResults }); + reconstructedEvents.push({ ...clone(event), toolResults: freshResults }); if (calls.length > 0) { replayedBatches += 1; if (replayedBatches === request.after_step) break; @@ -259,7 +264,7 @@ async function run(request) { let terminalError = null; const remaining = request.max_steps == null ? null : request.max_steps - prefixTurns; const unsubscribe = session.subscribe((event) => { - liveEvents.push(structuredClone(event)); + liveEvents.push(clone(event)); if (event.type === "turn_end") { continuedSteps += 1; if (event.message?.stopReason === "error") { diff --git a/crates/persisting-replay/src/adapter/generic.rs b/crates/persisting-replay/src/adapter/generic.rs new file mode 100644 index 00000000..88baba1d --- /dev/null +++ b/crates/persisting-replay/src/adapter/generic.rs @@ -0,0 +1,2063 @@ +//! Replay support for agents whose native transcript is a JSONL event stream. +//! +//! OpenCode prints `run --format=json` events, while Codex persists +//! `response_item` events in its rollout JSONL. Both formats carry the +//! assistant tool call and its observation in the transcript, so the replay +//! prefix can be rebuilt without an Agent SDK. The live phase is delegated to +//! the native CLI after the reconstructed transcript has been staged. + +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::time::{Duration, Instant}; + +use serde_json::{Value, json}; + +use super::{ + MAX_TOOL_OUTPUT_BYTES, RunContext, agent_command, check_boundary, prepared_outcome, + with_boundary_user_prompt_metadata, +}; +use crate::codex_bridge::{CodexBridgeHandle, PromptMode}; +use crate::error::{ReplayError, ReplayErrorKind, ResultExt}; +use crate::io::{atomic_write, atomic_write_json, canonicalize, read_regular_file, sha256}; +use crate::journal::Journal; +use crate::model::{ + AgentKind, FreshObservation, PlaybackRequest, ReplayMode, ReplayOutcome, ReplayPlan, ToolBatch, + ToolCall, +}; +use crate::process::{ProcessSpec, run_process}; + +#[derive(Debug, Clone, Copy)] +pub(super) enum NativeJsonlAgent { + Opencode, + Codex, +} + +impl NativeJsonlAgent { + fn kind(self) -> AgentKind { + match self { + Self::Opencode => AgentKind::Opencode, + Self::Codex => AgentKind::Codex, + } + } + + fn label(self) -> &'static str { + self.kind().as_str() + } +} + +#[derive(Debug, Clone)] +struct CallRecord { + call_event: usize, + output_event: usize, + call_id: String, + name: String, + arguments: Value, + observation: Value, + is_error: bool, + complete: bool, +} + +#[derive(Debug, Clone)] +struct TurnRecord { + start_event: usize, + end_event: usize, + text: String, + reasoning: String, + calls: Vec, +} + +type ParsedNative = (Vec, Option, Option); + +pub(super) fn build( + request: &PlaybackRequest, + agent: NativeJsonlAgent, +) -> Result { + let raw = read_regular_file(&request.trajectory)?; + let events = parse_jsonl(&raw, agent.label())?; + let (turns, user_prompt, session_id) = match agent { + NativeJsonlAgent::Opencode => parse_opencode(&events)?, + NativeJsonlAgent::Codex => parse_codex(&events)?, + }; + let complete_turns = turns + .iter() + .filter(|turn| !turn.calls.is_empty() && turn.calls.iter().all(|call| call.complete)) + .collect::>(); + check_boundary(request.after_step, complete_turns.len())?; + let selected = &complete_turns[..request.after_step]; + let boundary_end = selected + .last() + .map(|turn| turn.end_event) + .ok_or_else(|| ReplayError::trajectory("native JSONL replay boundary has no turn"))?; + let prefix_model_turns = turns + .iter() + .filter(|turn| turn.end_event <= boundary_end) + .count(); + let original_next_action = turns + .iter() + .filter(|turn| turn.start_event > boundary_end) + .find(|turn| is_actionable_turn(turn)) + .map(turn_signature); + let batches = selected + .iter() + .enumerate() + .map(|(index, turn)| ToolBatch { + ordinal: index + 1, + native_locator: format!("events:{}-{}", turn.start_event, turn.end_event), + tool_calls: turn + .calls + .iter() + .enumerate() + .map(|(ordinal, call)| ToolCall { + ordinal: ordinal + 1, + call_id: call.call_id.clone(), + name: call.name.clone(), + arguments: call.arguments.clone(), + original_observation: call.observation.clone(), + original_is_error: call.is_error, + native: json!({ + "call_event": call.call_event, + "output_event": call.output_event, + }), + }) + .collect(), + assistant_text: turn.text.clone(), + native: json!({ + "start_event": turn.start_event, + "end_event": turn.end_event, + }), + }) + .collect(); + Ok(ReplayPlan { + agent: request.agent, + source_path: canonicalize( + &request.trajectory, + ReplayErrorKind::Trajectory, + "trajectory", + )?, + source_sha256: sha256(&raw), + after_step: request.after_step, + prefix_model_turns, + batches, + native: json!({ + "format": agent.label(), + "events": events, + "user_prompt": user_prompt, + "session_id": session_id, + }), + original_next_action, + }) +} + +pub(super) fn execute( + plan: &ReplayPlan, + context: &RunContext<'_>, + journal: &mut Journal, + agent: NativeJsonlAgent, +) -> Result { + let events = plan + .native + .get("events") + .and_then(Value::as_array) + .ok_or_else(|| ReplayError::trajectory("native JSONL plan has no events"))?; + let boundary_end = plan + .batches + .last() + .and_then(|batch| batch.native.get("end_event")) + .and_then(Value::as_u64) + .ok_or_else(|| ReplayError::trajectory("native JSONL batch has no end event"))? + as usize; + if boundary_end >= events.len() { + return Err(ReplayError::trajectory( + "native JSONL boundary is out of bounds", + )); + } + let mut reconstructed_events = events[..=boundary_end].to_vec(); + let mut observations = Vec::new(); + for call in plan.calls() { + let fresh = execute_call(call, context)?; + let output_event = call + .native + .get("output_event") + .and_then(Value::as_u64) + .ok_or_else(|| ReplayError::trajectory("native JSONL call has no output event"))? + as usize; + match agent { + NativeJsonlAgent::Opencode => replace_opencode_observation( + reconstructed_events.get_mut(output_event).ok_or_else(|| { + ReplayError::trajectory("OpenCode call event is out of bounds") + })?, + &fresh, + )?, + NativeJsonlAgent::Codex => replace_codex_observation( + reconstructed_events.get_mut(output_event).ok_or_else(|| { + ReplayError::trajectory("Codex output event is out of bounds") + })?, + &fresh, + )?, + } + observations.push(fresh); + } + let prepared = context.output_dir.join("native/prepared-prefix.jsonl"); + write_jsonl(&prepared, &reconstructed_events)?; + journal.append( + "session_rebuilt", + [( + "prepared_only".into(), + json!(context.request.mode == ReplayMode::PrepareOnly), + )], + )?; + if context.request.mode == ReplayMode::PrepareOnly { + return Ok(prepared_outcome(prepared, context.request)); + } + + let replayed = context + .output_dir + .join("native/reconstructed-trajectory.jsonl"); + write_jsonl(&replayed, &reconstructed_events)?; + if context.request.mode == ReplayMode::ReplayOnly { + write_comparison(context, plan, &observations)?; + return Ok(ReplayOutcome { + status: "replayed".into(), + reconstructed_path: Some(replayed), + continued_path: None, + observations, + continued_steps: 0, + metadata: with_boundary_user_prompt_metadata( + json!({"native_cli": agent.label()}), + context.request, + false, + ), + }); + } + + let (continued, continued_steps) = continue_native_cli( + plan, + context, + journal, + agent, + &reconstructed_events, + &replayed, + )?; + write_comparison(context, plan, &observations)?; + if continued_steps == 0 { + return Err(ReplayError::continuation(format!( + "{} produced no continuation turns; see logs", + agent.label() + ))); + } + let mut metadata = json!({"native_cli": agent.label()}); + if matches!(agent, NativeJsonlAgent::Codex) { + let prompt_mode = if context.request.boundary_user_prompt().is_some() { + PromptMode::ExplicitUserPrompt + } else { + PromptMode::TransportNonce + }; + metadata["codex_resume_transport"] = json!({ + "prompt_mode": prompt_mode.as_str(), + "removed_before_model_request": prompt_mode == PromptMode::TransportNonce, + "removed_from_native_trajectory": prompt_mode == PromptMode::TransportNonce, + }); + } + Ok(ReplayOutcome { + status: "completed".into(), + reconstructed_path: None, + continued_path: Some(continued), + observations, + continued_steps, + metadata: with_boundary_user_prompt_metadata( + metadata, + context.request, + context.request.boundary_user_prompt().is_some(), + ), + }) +} + +fn parse_jsonl(raw: &[u8], label: &str) -> Result, ReplayError> { + let source = std::str::from_utf8(raw).map_err(|error| { + ReplayError::trajectory(format!("{label} trajectory is not UTF-8: {error}")) + })?; + let physical = source.split('\n').collect::>(); + let mut events = Vec::new(); + for (index, line) in physical.iter().enumerate() { + if line.trim().is_empty() { + continue; + } + match serde_json::from_str::(line) { + Ok(Value::Object(event)) => events.push(Value::Object(event)), + Ok(_) => { + return Err(ReplayError::trajectory(format!( + "{label} JSONL line {} must be an object", + index + 1 + ))); + } + Err(_) if index + 1 == physical.len() && !source.ends_with('\n') => break, + Err(error) => { + return Err(ReplayError::trajectory(format!( + "invalid {label} JSONL line {}: {error}", + index + 1 + ))); + } + } + } + if events.is_empty() { + return Err(ReplayError::trajectory(format!( + "{label} trajectory has no JSON events" + ))); + } + Ok(events) +} + +fn parse_opencode(events: &[Value]) -> Result { + let mut user_prompt = None; + let mut session_id = None; + let mut turns = Vec::new(); + let mut current: Option = None; + for (index, event) in events.iter().enumerate() { + session_id = session_id.or_else(|| { + event + .get("sessionID") + .and_then(Value::as_str) + .map(str::to_owned) + }); + match event.get("type").and_then(Value::as_str) { + Some("user") => { + if user_prompt.is_none() { + user_prompt = opencode_text(event); + } + } + Some("step_start") => { + if let Some(previous) = current.take() { + turns.push(previous); + } + current = Some(TurnRecord { + start_event: index, + end_event: index, + text: String::new(), + reasoning: String::new(), + calls: Vec::new(), + }); + } + Some("text") | Some("reasoning") | Some("tool_use") => { + let Some(turn) = current.as_mut() else { + continue; + }; + turn.end_event = index; + let part = event.get("part").cloned().unwrap_or_else(|| json!({})); + let part_type = part + .get("type") + .and_then(Value::as_str) + .or_else(|| event.get("type").and_then(Value::as_str)); + match part_type { + Some("text") => { + append_text(&mut turn.text, part.get("text").and_then(Value::as_str)) + } + Some("reasoning") => append_text( + &mut turn.reasoning, + part.get("text").and_then(Value::as_str), + ), + Some("tool") | Some("tool_use") => { + let id = part + .get("callID") + .or_else(|| part.get("id")) + .and_then(Value::as_str) + .map(str::to_owned) + .unwrap_or_else(|| format!("opencode-{index}")); + let name = part + .get("tool") + .or_else(|| part.get("name")) + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(); + let state = part.get("state").cloned().unwrap_or_else(|| json!({})); + let arguments = state.get("input").cloned().unwrap_or_else(|| json!({})); + let is_error = state.get("status").and_then(Value::as_str) == Some("error") + || state.get("error").is_some_and(|value| !value.is_null()); + let observation = state.get("output").cloned().unwrap_or(Value::Null); + let complete = !observation.is_null() + || is_error + || state.get("status").and_then(Value::as_str) == Some("completed"); + // OpenCode can emit more than one `tool_use` event for + // a call while its state transitions from pending to + // completed. Keep one call record and retain the final + // observation/event location. + if let Some(call) = turn.calls.iter_mut().find(|call| call.call_id == id) { + call.output_event = index; + if !name.is_empty() { + call.name = name; + } + if arguments != json!({}) { + call.arguments = arguments; + } + if !observation.is_null() { + call.observation = observation; + } + call.is_error |= is_error; + call.complete |= complete; + } else { + turn.calls.push(CallRecord { + call_event: index, + output_event: index, + call_id: id, + name, + arguments, + observation, + is_error, + complete, + }); + } + } + _ => {} + } + } + Some("step_finish") => { + if let Some(mut turn) = current.take() { + turn.end_event = index; + turns.push(turn); + } + } + _ => {} + } + } + if let Some(turn) = current { + turns.push(turn); + } + if user_prompt.is_none() { + return Err(ReplayError::trajectory( + "OpenCode trajectory has no user prompt", + )); + } + Ok((turns, user_prompt, session_id)) +} + +fn parse_codex(events: &[Value]) -> Result { + let mut user_prompt = None; + let mut session_id = None; + let mut turns = Vec::new(); + let mut current: Option = None; + let mut pending_outputs: Vec<(String, usize, Value, bool)> = Vec::new(); + for (index, event) in events.iter().enumerate() { + let event_type = event + .get("type") + .and_then(Value::as_str) + .unwrap_or_default(); + let Some(payload) = event.get("payload") else { + continue; + }; + if event_type == "session_meta" { + session_id = session_id.or_else(|| { + payload + .get("id") + .or_else(|| payload.get("session_id")) + .or_else(|| event.get("id")) + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + .map(str::to_owned) + }); + continue; + } + if event_type != "response_item" { + continue; + } + match payload.get("type").and_then(Value::as_str) { + Some("message") => { + let role = payload + .get("role") + .and_then(Value::as_str) + .unwrap_or_default(); + if role == "user" { + if user_prompt.is_none() { + user_prompt = codex_message_text(payload, "input_text"); + } + if let Some(turn) = current.take() { + turns.push(turn); + } + } else if role == "assistant" { + if let Some(turn) = current.take() + && (!turn.calls.is_empty() + || !turn.text.is_empty() + || !turn.reasoning.is_empty()) + { + turns.push(turn); + } + current = Some(TurnRecord { + start_event: index, + end_event: index, + text: codex_message_text(payload, "output_text").unwrap_or_default(), + reasoning: String::new(), + calls: Vec::new(), + }); + } + } + Some("reasoning") => { + let turn = current.get_or_insert_with(|| TurnRecord { + start_event: index, + end_event: index, + text: String::new(), + reasoning: String::new(), + calls: Vec::new(), + }); + turn.end_event = index; + if let Some(summary) = payload.get("summary").and_then(Value::as_array) { + for item in summary { + append_text( + &mut turn.reasoning, + item.get("text").and_then(Value::as_str), + ); + } + } + } + Some("function_call") | Some("custom_tool_call") => { + let turn = current.get_or_insert_with(|| TurnRecord { + start_event: index, + end_event: index, + text: String::new(), + reasoning: String::new(), + calls: Vec::new(), + }); + turn.end_event = index; + let call_id = payload + .get("call_id") + .or_else(|| payload.get("id")) + .and_then(Value::as_str) + .map(str::to_owned) + .ok_or_else(|| { + ReplayError::trajectory(format!( + "Codex function_call at event {index} has no call_id" + )) + })?; + let arguments = match payload.get("arguments").or_else(|| payload.get("input")) { + Some(Value::String(raw)) => { + serde_json::from_str(raw).unwrap_or_else(|_| json!(raw)) + } + Some(value) => value.clone(), + None => json!({}), + }; + let name = payload + .get("name") + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(); + turn.calls.push(CallRecord { + call_event: index, + output_event: usize::MAX, + call_id, + name, + arguments, + observation: Value::Null, + is_error: false, + complete: false, + }); + } + Some("function_call_output") | Some("custom_tool_call_output") => { + let call_id = payload + .get("call_id") + .and_then(Value::as_str) + .unwrap_or_default(); + if let Some(turn) = current.as_mut() { + turn.end_event = index; + if let Some(call) = turn.calls.iter_mut().find(|call| call.call_id == call_id) { + call.output_event = index; + call.observation = payload.get("output").cloned().unwrap_or(Value::Null); + call.complete = true; + call.is_error = payload + .get("status") + .and_then(Value::as_str) + .is_some_and(|status| status != "completed") + || payload.get("error").is_some_and(|value| !value.is_null()); + } else { + pending_outputs.push(( + call_id.to_owned(), + index, + payload.get("output").cloned().unwrap_or(Value::Null), + false, + )); + } + } + } + _ => {} + } + if !pending_outputs.is_empty() { + for (call_id, output_event, observation, is_error) in pending_outputs.drain(..) { + if let Some(turn) = current.as_mut() + && let Some(call) = turn.calls.iter_mut().find(|call| call.call_id == call_id) + { + call.output_event = output_event; + call.observation = observation; + call.is_error = is_error; + call.complete = true; + } + } + } + } + if let Some(turn) = current { + turns.push(turn); + } + if user_prompt.is_none() { + return Err(ReplayError::trajectory( + "Codex trajectory has no user message", + )); + } + // A response_item session_meta carries the ID in its payload, but older + // rollouts put it directly in the event. Accept both forms. + if session_id.is_none() { + session_id = events.iter().find_map(|event| { + (event.get("type").and_then(Value::as_str) == Some("session_meta")) + .then(|| { + event + .get("payload") + .and_then(|payload| payload.get("id")) + .or_else(|| event.get("id")) + .and_then(Value::as_str) + .map(str::to_owned) + }) + .flatten() + }); + } + Ok((turns, user_prompt, session_id)) +} + +fn turn_signature(turn: &TurnRecord) -> Value { + json!({ + "text": turn.text, + "reasoning": turn.reasoning, + "tools": turn.calls.iter().map(|call| json!({"name": call.name, "arguments": call.arguments})).collect::>(), + }) +} + +fn is_actionable_turn(turn: &TurnRecord) -> bool { + !turn.text.trim().is_empty() || !turn.calls.is_empty() +} + +fn opencode_text(event: &Value) -> Option { + event + .get("parts") + .and_then(Value::as_array) + .or_else(|| event.get("part").and_then(Value::as_array)) + .map(|parts| { + parts + .iter() + .filter_map(|part| part.get("text").and_then(Value::as_str)) + .collect::>() + .join("\n") + }) + .filter(|text| !text.is_empty()) +} + +fn codex_message_text(payload: &Value, wanted_type: &str) -> Option { + payload + .get("content") + .and_then(Value::as_array) + .map(|content| { + content + .iter() + .filter_map(|part| { + (part.get("type").and_then(Value::as_str) == Some(wanted_type)) + .then(|| part.get("text").and_then(Value::as_str)) + .flatten() + }) + .collect::>() + .join("\n") + }) + .filter(|text| !text.is_empty()) +} + +fn append_text(target: &mut String, value: Option<&str>) { + let Some(value) = value.filter(|value| !value.is_empty()) else { + return; + }; + if !target.is_empty() { + target.push('\n'); + } + target.push_str(value); +} + +fn execute_call( + call: &ToolCall, + context: &RunContext<'_>, +) -> Result { + let started = Instant::now(); + let mut is_error = false; + let mut return_code = None; + let content = match execute_tool_value(&call.name, &call.arguments, context, call.ordinal) { + Ok((content, code)) => { + return_code = code; + content + } + Err(error) => { + is_error = true; + Value::String(error.to_string()) + } + }; + if return_code.is_some_and(|code| code != 0) { + is_error = true; + } + Ok(FreshObservation { + call_id: call.call_id.clone(), + content, + is_error, + return_code, + duration_ms: started.elapsed().as_millis(), + truncated: false, + metadata: Default::default(), + }) +} + +fn execute_tool_value( + name: &str, + arguments: &Value, + context: &RunContext<'_>, + ordinal: usize, +) -> Result<(Value, Option), ReplayError> { + let mut arguments = arguments.clone(); + if let Value::String(raw) = &arguments { + arguments = match serde_json::from_str(raw) { + Ok(parsed) => parsed, + Err(_) if normalized_name(name) == "apply_patch" => { + return Err(ReplayError::new( + ReplayErrorKind::UnsupportedVersion, + "native apply_patch calls must provide structured JSON arguments", + )); + } + Err(_) => json!({"command": raw}), + }; + } + let normalized = normalized_name(name); + if matches!( + normalized.as_str(), + "bash" | "shell" | "exec" | "execute" | "terminal" + ) || arguments.get("command").is_some() + || arguments.get("cmd").is_some() + { + let command = arguments + .get("command") + .or_else(|| arguments.get("cmd")) + .or_else(|| arguments.get("script")) + .and_then(Value::as_str) + .ok_or_else(|| { + ReplayError::trajectory(format!( + "{} tool {name:?} has no command", + context.request.agent.as_str() + )) + })?; + let mut process = Command::new("/bin/sh"); + process + .args(["-c", command]) + .current_dir(&context.request.workspace); + let output = run_process(ProcessSpec { + command: process, + stdin: None, + timeout: Duration::from_secs(30 * 60), + termination_grace: Duration::from_secs(2), + pipe_grace: Duration::from_millis(250), + retained_bytes: MAX_TOOL_OUTPUT_BYTES, + log_path: context.state_dir.join(format!("native-tool-{ordinal}.log")), + }) + .map_err(|error| ReplayError::new(ReplayErrorKind::Executor, error.message))?; + let mut rendered = String::from_utf8_lossy(&output.stdout_tail).into_owned(); + if !output.stderr_tail.is_empty() { + if !rendered.is_empty() { + rendered.push('\n'); + } + rendered.push_str(&String::from_utf8_lossy(&output.stderr_tail)); + } + return Ok((Value::String(rendered), output.status.code())); + } + match normalized.as_str() { + "read" | "cat" => { + let path = tool_path( + arguments + .get("path") + .or_else(|| arguments.get("filePath")) + .or_else(|| arguments.get("file_path")), + context, + )?; + Ok(( + Value::String(String::from_utf8_lossy(&read_regular_file(&path)?).into_owned()), + Some(0), + )) + } + "write" => { + let path = tool_path( + arguments + .get("path") + .or_else(|| arguments.get("filePath")) + .or_else(|| arguments.get("file_path")), + context, + )?; + let content = arguments + .get("content") + .or_else(|| arguments.get("file_text")) + .and_then(Value::as_str) + .unwrap_or_default(); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent) + .replay_context(ReplayErrorKind::Executor, "create native write parent")?; + } + fs::write(path, content) + .replay_context(ReplayErrorKind::Executor, "write native tool file")?; + Ok((Value::String(String::new()), Some(0))) + } + "edit" => { + let path = tool_path( + arguments + .get("path") + .or_else(|| arguments.get("filePath")) + .or_else(|| arguments.get("file_path")), + context, + )?; + let old = arguments + .get("oldString") + .or_else(|| arguments.get("old_str")) + .and_then(Value::as_str) + .unwrap_or_default(); + let new = arguments + .get("newString") + .or_else(|| arguments.get("new_str")) + .and_then(Value::as_str) + .unwrap_or_default(); + let original = String::from_utf8_lossy(&read_regular_file(&path)?).into_owned(); + if !original.contains(old) { + return Err(ReplayError::trajectory(format!( + "edit target does not contain old text: {}", + path.display() + ))); + } + fs::write(&path, original.replacen(old, new, 1)) + .replay_context(ReplayErrorKind::Executor, "write native edit")?; + Ok((Value::String(String::new()), Some(0))) + } + _ => Err(ReplayError::new( + ReplayErrorKind::UnsupportedVersion, + format!( + "{} replay does not support tool {name:?}; use a command-shaped tool or add an adapter", + context.request.agent.as_str() + ), + )), + } +} + +fn normalized_name(name: &str) -> String { + name.to_ascii_lowercase() +} + +fn tool_path(value: Option<&Value>, context: &RunContext<'_>) -> Result { + let rendered = value + .and_then(Value::as_str) + .ok_or_else(|| ReplayError::trajectory("native file tool has no path"))?; + let path = Path::new(rendered); + let workspace = canonicalize( + &context.request.workspace, + ReplayErrorKind::Workspace, + "workspace", + )?; + let candidate = if path.is_absolute() { + path.to_path_buf() + } else { + workspace.join(path) + }; + // Resolve the nearest existing ancestor when the target is new. This + // catches a symlinked directory that would otherwise let a write escape + // the workspace before the file itself exists. + let check = if fs::symlink_metadata(&candidate).is_err() { + let mut existing = candidate.as_path(); + let mut missing = Vec::new(); + while fs::symlink_metadata(existing).is_err() { + missing.push( + existing + .file_name() + .ok_or_else(|| ReplayError::trajectory("native file tool path has no name"))? + .to_os_string(), + ); + existing = existing.parent().ok_or_else(|| { + ReplayError::trajectory("native file tool path has no existing parent") + })?; + } + let mut resolved = canonicalize( + existing, + ReplayErrorKind::Executor, + "native file tool path parent", + )?; + for component in missing.iter().rev() { + resolved.push(component); + } + resolved + } else { + canonicalize( + &candidate, + ReplayErrorKind::Executor, + "native file tool path", + )? + }; + if !check.starts_with(&workspace) { + return Err(ReplayError::trajectory(format!( + "native file tool path escapes workspace: {rendered:?}" + ))); + } + Ok(candidate) +} + +fn replace_opencode_observation( + event: &mut Value, + fresh: &FreshObservation, +) -> Result<(), ReplayError> { + let part = event + .get_mut("part") + .ok_or_else(|| ReplayError::trajectory("OpenCode tool event has no part"))?; + let state = part + .as_object_mut() + .and_then(|part| part.get_mut("state")) + .and_then(Value::as_object_mut) + .ok_or_else(|| ReplayError::trajectory("OpenCode tool event has no state"))?; + state.insert( + "status".into(), + json!(if fresh.is_error { "error" } else { "completed" }), + ); + state.insert("output".into(), fresh.content.clone()); + if fresh.is_error { + state.insert("error".into(), fresh.content.clone()); + } else { + state.remove("error"); + } + Ok(()) +} + +fn replace_codex_observation( + event: &mut Value, + fresh: &FreshObservation, +) -> Result<(), ReplayError> { + let payload = event + .get_mut("payload") + .and_then(Value::as_object_mut) + .ok_or_else(|| ReplayError::trajectory("Codex output event has no payload"))?; + payload.insert("output".into(), fresh.content.clone()); + if fresh.is_error { + payload.insert("status".into(), json!("failed")); + } + Ok(()) +} + +fn write_jsonl(path: &Path, values: &[Value]) -> Result<(), ReplayError> { + let mut bytes = Vec::new(); + for value in values { + serde_json::to_writer(&mut bytes, value) + .replay_context(ReplayErrorKind::Executor, "serialize native JSONL")?; + bytes.push(b'\n'); + } + atomic_write(path, &bytes) +} + +fn write_comparison( + context: &RunContext<'_>, + plan: &ReplayPlan, + observations: &[FreshObservation], +) -> Result<(), ReplayError> { + let comparisons = plan.calls().zip(observations).map(|(call, fresh)| json!({ + "call_id": call.call_id, + "tool": call.name, + "exact": call.original_observation == fresh.content && call.original_is_error == fresh.is_error, + "original_is_error": call.original_is_error, + "replayed_is_error": fresh.is_error, + })).collect::>(); + atomic_write_json( + &context.output_dir.join("observation-comparison.json"), + &comparisons, + ) +} + +fn continue_native_cli( + plan: &ReplayPlan, + context: &RunContext<'_>, + journal: &mut Journal, + agent: NativeJsonlAgent, + prefix: &[Value], + reconstructed: &Path, +) -> Result<(PathBuf, usize), ReplayError> { + let launch = context + .launch + .ok_or_else(|| ReplayError::continuation("native CLI replay has no launch spec"))?; + let logs = context.output_dir.join("logs"); + fs::create_dir_all(&logs) + .replay_context(ReplayErrorKind::Executor, "create native CLI log directory")?; + let log_path = logs.join(format!("{}.log", agent.label())); + // `PlaybackRequest::session_id` is the pVisor/model-router session. It + // must not be used as a native Codex session identity: SweEval (and other + // callers) deliberately set it to a routing key such as + // `sweeval-`. Codex resume resolves the rollout from the native + // session id stored in the source trajectory. Mixing the two makes + // `codex exec resume` silently start a fresh conversation, which is much + // worse than failing the replay because the resulting patch can still + // pass a verifier while A(N+1) is no longer comparable with A'(N+1). + let session_id = continuation_session_id(agent, plan, context)?; + let mut command = agent_command(&launch.entrypoint, context); + let mut codex_bridge = None; + let mut codex_transport_prompt = None; + let mut codex_prompt_mode = None; + command.env("PVISOR_REPLAY_TRAJECTORY", reconstructed); + command.env("PVISOR_REPLAY_AFTER_STEP", plan.after_step.to_string()); + command.env( + "PVISOR_REPLAY_MAX_STEPS", + context + .request + .max_steps + .map(|value| value.to_string()) + .unwrap_or_default(), + ); + command.env("PVISOR_REPLAY_SESSION_ID", &session_id); + match agent { + NativeJsonlAgent::Opencode => { + let session_id = opencode_session_id(&session_id); + command.env("PVISOR_REPLAY_SESSION_ID", &session_id); + let export_path = context.output_dir.join("native/opencode-session.json"); + let opencode_config = context.state_dir.join("opencode-config"); + let opencode_data = context.state_dir.join("opencode-data"); + atomic_write_json( + &export_path, + &opencode_export(plan, prefix, &session_id, &context.request.workspace), + )?; + let mut import = agent_command(&launch.entrypoint, context); + import.args([ + "import", + export_path.to_str().ok_or_else(|| { + ReplayError::configuration("OpenCode export path is not valid UTF-8") + })?, + ]); + import.env("XDG_CONFIG_HOME", &opencode_config); + import.env("XDG_DATA_HOME", &opencode_data); + import.env("OPENCODE_DISABLE_AUTOUPDATE", "1"); + let import_log = logs.join("opencode-import.log"); + let imported = run_process(ProcessSpec { + command: import, + stdin: None, + timeout: Duration::from_secs(5 * 60), + termination_grace: Duration::from_secs(2), + pipe_grace: Duration::from_millis(250), + retained_bytes: MAX_TOOL_OUTPUT_BYTES / 4, + log_path: import_log.clone(), + }) + .map_err(|error| ReplayError::new(ReplayErrorKind::Continuation, error.message))?; + if !imported.status.success() { + return Err(ReplayError::classify_continuation( + format!( + "OpenCode session import exited {}; see {}", + imported.status, + import_log.display() + ), + &String::from_utf8_lossy(&imported.stderr_tail), + )); + } + command.env("XDG_CONFIG_HOME", &opencode_config); + command.env("XDG_DATA_HOME", &opencode_data); + command.env("OPENCODE_DISABLE_AUTOUPDATE", "1"); + if let Some(model) = configured_model_from_environment() { + command.args(["--model", &model]); + } + command.args([ + "run", + "--format=json", + "--session", + &session_id, + "--dangerously-skip-permissions", + ]); + if !context.request.disable_thinking { + command.arg("--thinking"); + } + if let Some(prompt) = context.request.boundary_user_prompt() { + command.arg("--"); + command.arg(prompt); + } + } + NativeJsonlAgent::Codex => { + let explicit_prompt = context.request.boundary_user_prompt().map(str::to_owned); + let transport_prompt = explicit_prompt + .clone() + .unwrap_or_else(|| format!("pvisor-codex-resume-{}", context.nonce)); + let bridge = CodexBridgeHandle::start( + context.session_id, + transport_prompt.clone(), + explicit_prompt.as_deref(), + )?; + let codex_home = context.state_dir.join("codex-home"); + let session_path = codex_session_path(&codex_home, &session_id, &plan.native)?; + fs::create_dir_all( + session_path + .parent() + .expect("Codex session path has parent"), + ) + .replay_context(ReplayErrorKind::Executor, "create Codex session directory")?; + let staged = codex_staged_events(prefix, &session_id, &context.request.workspace); + write_jsonl(&session_path, &staged)?; + // Recent Codex releases resolve the Responses endpoint from a + // model-provider profile. Point the isolated profile at the + // local SandboxReplay bridge; the bridge removes the transport + // nonce before forwarding the request upstream. + let encoded = serde_json::to_string(bridge.base_url.as_str()).map_err(|error| { + ReplayError::configuration(format!("cannot encode Codex bridge URL: {error}")) + })?; + let config = format!( + "model_provider = \"pvisor-replay\"\n\n[model_providers.pvisor-replay]\nname = \"pvisor-replay\"\nbase_url = {encoded}\nenv_key = \"OPENAI_API_KEY\"\nwire_api = \"responses\"\n" + ); + atomic_write(&codex_home.join("config.toml"), config.as_bytes())?; + for (name, value) in bridge.child_environment() { + command.env(name, value); + } + command.env("CODEX_HOME", &codex_home); + command.args([ + "exec", + "resume", + &session_id, + "--json", + "--skip-git-repo-check", + "--dangerously-bypass-approvals-and-sandbox", + ]); + // ``exec resume`` otherwise falls back to Codex's default model. + // The native SweEval launch pins the configured model explicitly; + // carry the same value into the continuation so local + // OpenAI-compatible endpoints do not receive an unsupported + // default model (for example ``gpt-5``). + if let Some(model) = configured_model_from_environment() { + command.args(["--model", model.rsplit('/').next().unwrap_or(&model)]); + } + if let Some(reasoning_effort) = std::env::var("PVISOR_REPLAY_REASONING_EFFORT") + .ok() + .filter(|value| !value.trim().is_empty()) + { + command.args(["-c", &format!("model_reasoning_effort={reasoning_effort}")]); + } + if let Some(max_steps) = context.request.max_steps { + command.args(["-c", &format!("agent_max_steps={max_steps}")]); + } + // Older Codex releases require a prompt argument for resume. In + // the default mode this is an opaque nonce removed by the local + // Responses bridge before the request reaches the model. + command.arg(&transport_prompt); + codex_transport_prompt = Some(transport_prompt); + codex_prompt_mode = Some(bridge.prompt_mode()); + codex_bridge = Some(bridge); + } + } + journal.append( + "continuation_started", + [("agent".into(), json!(agent.label()))], + )?; + let output = run_process(ProcessSpec { + command, + stdin: None, + timeout: Duration::from_secs(24 * 60 * 60), + termination_grace: Duration::from_secs(2), + pipe_grace: Duration::from_millis(250), + retained_bytes: MAX_TOOL_OUTPUT_BYTES / 2, + log_path: log_path.clone(), + }) + .map_err(|error| ReplayError::new(ReplayErrorKind::Continuation, error.message))?; + let bridge_result = codex_bridge.take().map(|bridge| { + let result = bridge.finish(); + result + }); + let bridge_error = bridge_result.and_then(|result| result.err()); + if !output.status.success() { + let process_error = ReplayError::classify_continuation( + format!( + "{} replay/continuation exited {}; see {}", + agent.label(), + output.status, + log_path.display() + ), + &String::from_utf8_lossy(&output.stderr_tail), + ); + if let Some(error) = bridge_error { + return Err(ReplayError::continuation(format!( + "{process_error}; Codex bridge validation also failed: {error}" + ))); + } + return Err(process_error); + } + if let Some(error) = bridge_error { + return Err(error); + } + let output_path = context.output_dir.join("native/continued-trajectory.jsonl"); + let (continued_events, continued_steps) = match agent { + NativeJsonlAgent::Codex => { + let codex_home = context.state_dir.join("codex-home"); + let staged_path = codex_session_path(&codex_home, &session_id, &plan.native)?; + // `exec resume` may rotate the rollout into a new timestamped + // file instead of appending to the staged path. Prefer the + // newest file from this isolated CODEX_HOME so the continuation + // is not reported as zero-step merely because we read the stale + // prefix file. + let session_path = + latest_codex_session_path(&codex_home, session_id.as_str()).unwrap_or(staged_path); + let raw_events = if session_path.is_file() { + parse_jsonl(&read_regular_file(&session_path)?, agent.label())? + } else { + Vec::new() + }; + validate_codex_continuation(&raw_events, plan, &session_id, &session_path)?; + let events = clean_codex_transport_events( + raw_events, + plan, + &session_id, + codex_prompt_mode.unwrap_or(PromptMode::TransportNonce), + codex_transport_prompt.as_deref(), + &session_path, + )?; + validate_codex_continuation(&events, plan, &session_id, &session_path)?; + let steps = count_codex_turns_after(&events, plan); + (events, steps) + } + NativeJsonlAgent::Opencode => { + let raw = read_regular_file(&log_path)?; + let events = parse_json_lines_from_log(&raw); + let steps = count_opencode_turns(&events); + let mut combined = prefix.to_vec(); + combined.extend(events); + (combined, steps) + } + }; + if continued_events.is_empty() { + return Err(ReplayError::continuation( + "native CLI produced no JSONL trajectory", + )); + } + write_jsonl(&output_path, &continued_events)?; + Ok((output_path, continued_steps)) +} + +fn configured_model_from_environment() -> Option { + std::env::var("MODEL_NAME") + .ok() + .or_else(|| std::env::var("OPENAI_MODEL").ok()) + .filter(|model| !model.trim().is_empty()) +} + +fn continuation_session_id( + agent: NativeJsonlAgent, + plan: &ReplayPlan, + context: &RunContext<'_>, +) -> Result { + let native = plan + .native + .get("session_id") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()); + match agent { + NativeJsonlAgent::Codex => codex_native_session_id(&plan.native), + NativeJsonlAgent::Opencode => Ok(context + .request + .session_id + .as_deref() + .or(native) + .unwrap_or(context.session_id) + .to_owned()), + } +} + +fn codex_native_session_id(native: &Value) -> Result { + native + .get("session_id") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + .ok_or_else(|| { + ReplayError::continuation( + "Codex trajectory has no native session_meta id; refusing to use the pVisor/router session_id for resume", + ) + }) +} + +fn latest_codex_session_path(root: &Path, session_id: &str) -> Option { + fn visit( + directory: &Path, + session_id: &str, + newest: &mut Option<(std::time::SystemTime, PathBuf)>, + ) { + let Ok(entries) = fs::read_dir(directory) else { + return; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + visit(&path, session_id, newest); + continue; + } + if path.extension().and_then(|value| value.to_str()) != Some("jsonl") { + continue; + } + // Never pick an unrelated/stale rollout from the isolated + // CODEX_HOME. The native session id is also checked from the + // event stream below; matching the filename avoids selecting a + // rotated file belonging to another replay attempt. + let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else { + continue; + }; + if !file_name.ends_with(&format!("-{session_id}.jsonl")) { + continue; + } + let Ok(modified) = entry.metadata().and_then(|metadata| metadata.modified()) else { + continue; + }; + if newest + .as_ref() + .is_none_or(|(current, _)| modified > *current) + { + *newest = Some((modified, path)); + } + } + } + let mut newest = None; + visit(root, session_id, &mut newest); + newest.map(|(_, path)| path) +} + +fn validate_codex_continuation( + events: &[Value], + plan: &ReplayPlan, + session_id: &str, + path: &Path, +) -> Result<(), ReplayError> { + if events.is_empty() { + return Err(ReplayError::continuation(format!( + "Codex resume produced no events in {}", + path.display() + ))); + } + let observed_session_id = events.iter().find_map(|event| { + (event.get("type").and_then(Value::as_str) == Some("session_meta")).then(|| { + event + .pointer("/payload/id") + .or_else(|| event.pointer("/payload/session_id")) + .and_then(Value::as_str) + }) + }); + if observed_session_id.flatten() != Some(session_id) { + return Err(ReplayError::continuation(format!( + "Codex resume did not return native session {session_id:?} (observed {:?}); refusing an unverified continuation", + observed_session_id.flatten() + ))); + } + + let assistant_turns = events + .iter() + .filter(|event| { + event.get("type").and_then(Value::as_str) == Some("response_item") + && event.pointer("/payload/type").and_then(Value::as_str) == Some("message") + && event.pointer("/payload/role").and_then(Value::as_str) == Some("assistant") + }) + .count(); + if assistant_turns < plan.batches.len() { + return Err(ReplayError::continuation(format!( + "Codex resume returned only {assistant_turns} assistant turns, but the staged boundary contains {} tool batches; refusing a fresh-session continuation", + plan.batches.len() + ))); + } + + if let Some(expected_prompt) = plan + .native + .get("user_prompt") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + { + let has_original_prompt = events.iter().any(|event| { + event.get("type").and_then(Value::as_str) == Some("response_item") + && event.pointer("/payload/type").and_then(Value::as_str) == Some("message") + && event.pointer("/payload/role").and_then(Value::as_str) == Some("user") + && event + .get("payload") + .and_then(|payload| codex_message_text(payload, "input_text")) + .as_deref() + == Some(expected_prompt) + }); + if !has_original_prompt { + return Err(ReplayError::continuation( + "Codex resume did not contain the original user task; refusing a fresh-session continuation", + )); + } + } + + if let Some(last_call_id) = plan + .batches + .last() + .and_then(|batch| batch.tool_calls.last()) + .map(|call| call.call_id.as_str()) + { + let has_boundary_call = events.iter().any(|event| { + event.pointer("/payload/call_id").and_then(Value::as_str) == Some(last_call_id) + }); + if !has_boundary_call { + return Err(ReplayError::continuation(format!( + "Codex resume did not contain the boundary call {last_call_id:?}; refusing an unverified continuation" + ))); + } + } + Ok(()) +} + +fn clean_codex_transport_events( + mut events: Vec, + plan: &ReplayPlan, + _session_id: &str, + prompt_mode: PromptMode, + transport_prompt: Option<&str>, + path: &Path, +) -> Result, ReplayError> { + if prompt_mode == PromptMode::ExplicitUserPrompt { + return Ok(events); + } + let expected = transport_prompt + .filter(|prompt| !prompt.is_empty()) + .ok_or_else(|| ReplayError::continuation("Codex transport nonce is missing"))?; + let boundary_call_id = plan + .batches + .last() + .and_then(|batch| batch.tool_calls.last()) + .map(|call| call.call_id.as_str()) + .ok_or_else(|| ReplayError::trajectory("Codex plan has no boundary call"))?; + let boundary_index = events + .iter() + .rposition(|event| { + event.pointer("/payload/call_id").and_then(Value::as_str) == Some(boundary_call_id) + }) + .ok_or_else(|| { + ReplayError::continuation(format!( + "Codex continuation has no boundary call in {}", + path.display() + )) + })?; + let matches = events + .iter() + .enumerate() + .filter_map(|(index, event)| { + (index > boundary_index + && event.get("type").and_then(Value::as_str) == Some("response_item") + && event.pointer("/payload/type").and_then(Value::as_str) == Some("message") + && event.pointer("/payload/role").and_then(Value::as_str) == Some("user") + && event + .get("payload") + .and_then(|payload| codex_message_text(payload, "input_text")) + .as_deref() + == Some(expected)) + .then_some(index) + }) + .collect::>(); + if matches.len() != 1 { + return Err(ReplayError::continuation(format!( + "expected exactly one Codex transport nonce in the resumed trajectory, found {}", + matches.len() + ))); + } + events.remove(matches[0]); + for event in &mut events { + redact_codex_transport_nonce(event, expected); + } + let legacy_prompt = "Continue from the replay boundary."; + if events.iter().any(|event| { + event.get("type").and_then(Value::as_str) == Some("response_item") + && event.pointer("/payload/type").and_then(Value::as_str) == Some("message") + && event.pointer("/payload/role").and_then(Value::as_str) == Some("user") + && event + .get("payload") + .and_then(|payload| codex_message_text(payload, "input_text")) + .as_deref() + == Some(legacy_prompt) + }) { + return Err(ReplayError::continuation( + "legacy Codex resume prompt remains in the cleaned trajectory", + )); + } + if events + .iter() + .any(|event| event.to_string().contains(expected)) + { + return Err(ReplayError::continuation( + "Codex transport nonce remains in the cleaned trajectory", + )); + } + Ok(events) +} + +fn redact_codex_transport_nonce(value: &mut Value, expected: &str) { + match value { + Value::String(text) => { + if text.contains(expected) { + *text = text.replace(expected, ""); + } + } + Value::Array(values) => { + for value in values { + redact_codex_transport_nonce(value, expected); + } + } + Value::Object(fields) => { + for value in fields.values_mut() { + redact_codex_transport_nonce(value, expected); + } + } + Value::Null | Value::Bool(_) | Value::Number(_) => {} + } +} + +fn codex_session_path( + codex_home: &Path, + session_id: &str, + native: &Value, +) -> Result { + if session_id.is_empty() + || !session_id + .chars() + .all(|character| character.is_ascii_alphanumeric() || matches!(character, '-' | '_')) + { + return Err(ReplayError::configuration( + "Codex session_id must contain only ASCII letters, digits, '-' and '_'", + )); + } + // Codex discovers sessions below CODEX_HOME/sessions by the rollout file + // name. Keep the directory and timestamp shape used by the native CLI + // while placing the staged transcript in the replay state directory. A + // fixed 1970 path is accepted by some versions but is not a native rollout + // identity and can make `exec resume` ignore the staged file. + let timestamp = native + .get("events") + .and_then(Value::as_array) + .and_then(|events| { + events.iter().find_map(|event| { + (event.get("type").and_then(Value::as_str) == Some("session_meta")).then(|| { + event + .pointer("/payload/timestamp") + .or_else(|| event.pointer("/timestamp")) + .and_then(Value::as_str) + }) + }) + }) + .flatten() + .and_then(|value| chrono::DateTime::parse_from_rfc3339(value).ok()); + let (directory, filename_timestamp) = timestamp + .map(|value| { + ( + value.format("%Y/%m/%d").to_string(), + value.format("%Y-%m-%dT%H-%M-%S").to_string(), + ) + }) + .unwrap_or_else(|| ("1970/01/01".into(), "1970-01-01T00-00-00".into())); + Ok(codex_home.join(format!( + "sessions/{directory}/rollout-{filename_timestamp}-{session_id}.jsonl" + ))) +} + +fn opencode_session_id(raw: &str) -> String { + let suffix = raw + .chars() + .filter(|character| character.is_ascii_alphanumeric() || matches!(character, '_' | '-')) + .collect::(); + if raw.starts_with("ses_") && raw == suffix && !suffix.is_empty() { + raw.to_owned() + } else { + format!( + "ses_pvisor_{}", + if suffix.is_empty() { "replay" } else { &suffix } + ) + } +} + +fn opencode_export( + plan: &ReplayPlan, + prefix: &[Value], + session_id: &str, + workspace: &Path, +) -> Value { + let user_id = "msg_pvisor_user"; + let prompt = plan + .native + .get("user_prompt") + .and_then(Value::as_str) + .unwrap_or_default(); + let mut messages = vec![json!({ + "info": { + "id": user_id, + "sessionID": session_id, + "role": "user", + "time": {"created": 0}, + "agent": "build", + "model": {"providerID": "pvisor", "modelID": "replay"}, + }, + "parts": [{ + "id": "prt_pvisor_user", + "sessionID": session_id, + "messageID": user_id, + "type": "text", + "text": prompt, + }], + })]; + for batch in &plan.batches { + let message_id = format!("msg_pvisor_{:04}", batch.ordinal); + let mut parts = vec![json!({ + "id": format!("prt_pvisor_step_start_{:04}", batch.ordinal), + "sessionID": session_id, + "messageID": message_id, + "type": "step-start", + })]; + let start_event = batch + .native + .get("start_event") + .and_then(Value::as_u64) + .unwrap_or_default() as usize; + let end_event = batch + .native + .get("end_event") + .and_then(Value::as_u64) + .unwrap_or(start_event as u64) as usize; + for (reasoning_ordinal, event) in prefix + .get(start_event..=end_event.min(prefix.len().saturating_sub(1))) + .into_iter() + .flatten() + .filter(|event| event.get("type").and_then(Value::as_str) == Some("reasoning")) + .enumerate() + { + let Some(text) = event + .get("part") + .and_then(|part| part.get("text")) + .and_then(Value::as_str) + .filter(|text| !text.is_empty()) + else { + continue; + }; + parts.push(json!({ + "id": format!("prt_pvisor_reasoning_{:04}_{:04}", batch.ordinal, reasoning_ordinal + 1), + "sessionID": session_id, + "messageID": message_id, + "type": "reasoning", + "text": text, + "time": {"start": 0, "end": 0}, + })); + } + if !batch.assistant_text.is_empty() { + parts.push(json!({ + "id": format!("prt_pvisor_text_{:04}", batch.ordinal), + "sessionID": session_id, + "messageID": message_id, + "type": "text", + "text": batch.assistant_text, + })); + } + for (ordinal, call) in batch.tool_calls.iter().enumerate() { + let output_event = call + .native + .get("output_event") + .and_then(Value::as_u64) + .unwrap_or_default() as usize; + let state = prefix + .get(output_event) + .and_then(|event| event.get("part")) + .and_then(|part| part.get("state")); + let fresh_output = state + .and_then(|state| state.get("output")) + .map(render_opencode_output) + .unwrap_or_default(); + let fresh_error = state + .and_then(|state| state.get("status")) + .and_then(Value::as_str) + == Some("error"); + let input = if call.arguments.is_object() { + call.arguments.clone() + } else { + json!({"value": call.arguments}) + }; + let state = if fresh_error { + json!({ + "status": "error", + "input": input, + "error": fresh_output, + "metadata": {}, + "time": {"start": 0, "end": 0}, + }) + } else { + json!({ + "status": "completed", + "input": input, + "output": fresh_output, + "title": call.name, + "metadata": {}, + "time": {"start": 0, "end": 0}, + }) + }; + parts.push(json!({ + "id": format!("prt_pvisor_tool_{:04}_{:04}", batch.ordinal, ordinal + 1), + "sessionID": session_id, + "messageID": message_id, + "type": "tool", + "callID": call.call_id, + "tool": call.name, + "state": state, + })); + } + parts.push(json!({ + "id": format!("prt_pvisor_step_finish_{:04}", batch.ordinal), + "sessionID": session_id, + "messageID": message_id, + "type": "step-finish", + "reason": "tool-calls", + "cost": 0, + "tokens": {"input": 0, "output": 0, "reasoning": 0, "cache": {"read": 0, "write": 0}}, + })); + messages.push(json!({ + "info": { + "id": message_id, + "sessionID": session_id, + "role": "assistant", + "time": {"created": batch.ordinal as u64, "completed": batch.ordinal as u64}, + "parentID": user_id, + "modelID": "replay", + "providerID": "pvisor", + "mode": "build", + "agent": "build", + "path": {"cwd": workspace.display().to_string(), "root": workspace.display().to_string()}, + "cost": 0, + "tokens": {"input": 0, "output": 0, "reasoning": 0, "cache": {"read": 0, "write": 0}}, + }, + "parts": parts, + })); + } + json!({ + "info": { + "id": session_id, + "slug": "pvisor-replay", + "projectID": "global", + "directory": workspace.display().to_string(), + "path": "", + "title": if prompt.is_empty() { "pVisor replay" } else { prompt }, + "version": "1", + "time": {"created": 0, "updated": plan.batches.len() as u64}, + }, + "messages": messages, + }) +} + +fn render_opencode_output(value: &Value) -> String { + value + .as_str() + .map(str::to_owned) + .unwrap_or_else(|| value.to_string()) +} + +fn codex_staged_events(prefix: &[Value], session_id: &str, workspace: &Path) -> Vec { + let mut events = prefix.to_vec(); + let mut found_meta = false; + for event in &mut events { + if event.get("type").and_then(Value::as_str) != Some("session_meta") { + continue; + } + found_meta = true; + if let Some(payload) = event.get_mut("payload").and_then(Value::as_object_mut) { + payload.insert("id".into(), json!(session_id)); + if payload.contains_key("session_id") { + payload.insert("session_id".into(), json!(session_id)); + } + payload + .entry("cwd") + .or_insert_with(|| json!(workspace.display().to_string())); + payload + .entry("cli_version") + .or_insert_with(|| json!(AgentKind::Codex.supported_version())); + } + } + if !found_meta { + events.insert( + 0, + json!({ + "timestamp": "1970-01-01T00:00:00Z", + "type": "session_meta", + "payload": { + "id": session_id, + "cwd": workspace.display().to_string(), + "cli_version": AgentKind::Codex.supported_version(), + }, + }), + ); + } + events +} + +fn parse_json_lines_from_log(raw: &[u8]) -> Vec { + String::from_utf8_lossy(raw) + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter(|event| { + matches!( + event.get("type").and_then(Value::as_str), + Some("user" | "step_start" | "text" | "reasoning" | "tool_use" | "step_finish") + ) + }) + .collect() +} + +fn count_opencode_turns(events: &[Value]) -> usize { + events + .iter() + .filter(|event| event.get("type").and_then(Value::as_str) == Some("step_finish")) + .count() +} + +fn count_codex_turns_after(events: &[Value], plan: &ReplayPlan) -> usize { + let boundary_call = plan + .batches + .last() + .and_then(|batch| batch.tool_calls.last()) + .map(|call| call.call_id.as_str()); + let boundary_index = boundary_call.and_then(|call_id| { + events.iter().rposition(|event| { + event.pointer("/payload/call_id").and_then(Value::as_str) == Some(call_id) + }) + }); + let continuation_events = boundary_index + .and_then(|index| events.get(index.saturating_add(1)..)) + .unwrap_or(events); + let turns = continuation_events + .iter() + .filter(|event| { + event.get("type").and_then(Value::as_str) == Some("response_item") + && event.pointer("/payload/role").and_then(Value::as_str) == Some("assistant") + }) + .count(); + if turns > 0 { + return turns; + } + // Some Codex releases emit tool calls without an assistant message for a + // short continuation. Such a stream is still a live turn; use the + // number of post-prefix tool calls as a conservative lower bound rather + // than incorrectly classifying a successful run as zero-step. + let continuation_calls = continuation_events + .iter() + .filter(|event| { + event.get("type").and_then(Value::as_str) == Some("response_item") + && matches!( + event.pointer("/payload/type").and_then(Value::as_str), + Some("function_call" | "custom_tool_call") + ) + }) + .count(); + usize::from(continuation_calls > 0) +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use super::{ + CallRecord, NativeJsonlAgent, RunContext, TurnRecord, codex_native_session_id, + continuation_session_id, is_actionable_turn, parse_codex, parse_jsonl, parse_opencode, + redact_codex_transport_nonce, validate_codex_continuation, + }; + use crate::model::{AgentKind, PlaybackRequest, ReplayMode, ReplayPlan, ToolBatch, ToolCall}; + use serde_json::{Value, json}; + + #[test] + fn opencode_events_group_tool_parts_into_complete_turns() { + let source = [ + json!({"type":"user","sessionID":"ses-test","parts":[{"type":"text","text":"fix it"}]}), + json!({"type":"step_start","sessionID":"ses-test"}), + json!({"type":"text","sessionID":"ses-test","part":{"type":"text","text":"Inspecting"}}), + json!({"type":"tool_use","sessionID":"ses-test","part":{"type":"tool","callID":"call-1","tool":"bash","state":{"status":"completed","input":{"command":"pwd"},"output":"/workspace"}}}), + json!({"type":"step_finish","sessionID":"ses-test","part":{"reason":"tool-calls"}}), + json!({"type":"step_start","sessionID":"ses-test"}), + json!({"type":"text","sessionID":"ses-test","part":{"type":"text","text":"Done"}}), + json!({"type":"step_finish","sessionID":"ses-test","part":{"reason":"stop"}}), + ]; + let (turns, prompt, session) = parse_opencode(&source).unwrap(); + assert_eq!(prompt.as_deref(), Some("fix it")); + assert_eq!(session.as_deref(), Some("ses-test")); + assert_eq!(turns.len(), 2); + assert_eq!(turns[0].calls[0].name, "bash"); + assert_eq!(turns[0].calls[0].observation, "/workspace"); + assert_eq!(turns[1].text, "Done"); + } + + #[test] + fn reasoning_only_turn_is_not_an_actionable_next_step() { + let reasoning_only = TurnRecord { + start_event: 1, + end_event: 1, + text: " ".into(), + reasoning: "internal planning".into(), + calls: Vec::new(), + }; + assert!(!is_actionable_turn(&reasoning_only)); + + let visible_text = TurnRecord { + text: "continue".into(), + ..reasoning_only.clone() + }; + assert!(is_actionable_turn(&visible_text)); + + let tool_call = TurnRecord { + calls: vec![CallRecord { + call_event: 2, + output_event: 3, + call_id: "call-1".into(), + name: "exec_command".into(), + arguments: json!({"cmd": "pwd"}), + observation: Value::Null, + is_error: false, + complete: false, + }], + ..reasoning_only + }; + assert!(is_actionable_turn(&tool_call)); + } + + #[test] + fn codex_rollouts_accept_responses_and_custom_tool_calls() { + let source = [ + json!({"type":"session_meta","payload":{"id":"sess-test"}}), + json!({"type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"fix it"}]}}), + json!({"type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Inspecting"}]}}), + json!({"type":"response_item","payload":{"type":"custom_tool_call","call_id":"call-1","name":"exec","input":"{\"command\":\"pwd\"}"}}), + json!({"type":"response_item","payload":{"type":"custom_tool_call_output","call_id":"call-1","output":[{"type":"input_text","text":"/workspace"}]}}), + json!({"type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Done"}]}}), + ]; + let (turns, prompt, session) = parse_codex(&source).unwrap(); + assert_eq!(prompt.as_deref(), Some("fix it")); + assert_eq!(session.as_deref(), Some("sess-test")); + assert_eq!(turns.len(), 2); + assert_eq!(turns[0].calls[0].name, "exec"); + assert_eq!(turns[0].calls[0].arguments["command"], "pwd"); + assert_eq!(turns[0].calls[0].output_event, 4); + assert_eq!(turns[1].text, "Done"); + } + + #[test] + fn codex_transport_nonce_is_redacted_from_model_echoes() { + let nonce = "pvisor-codex-resume-abc123"; + let mut event = json!({ + "payload": { + "summary": [{"text": format!("I received {nonce}")}], + "content": [{"text": format!("{nonce} should not persist")}] + } + }); + redact_codex_transport_nonce(&mut event, nonce); + assert!(!event.to_string().contains(nonce)); + assert_eq!(event["payload"]["summary"][0]["text"], "I received "); + } + + #[test] + fn codex_session_meta_accepts_legacy_top_level_id() { + let source = [ + json!({"type":"session_meta","id":"legacy-session"}), + json!({"type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"fix it"}]}}), + ]; + let (_, _, session) = parse_codex(&source).unwrap(); + assert_eq!(session.as_deref(), Some("legacy-session")); + } + + #[test] + fn parser_tolerates_only_a_truncated_final_jsonl_line() { + let raw = b"{\"type\":\"user\",\"parts\":[{\"type\":\"text\",\"text\":\"hi\"}]}\n{"; + let events = parse_jsonl(raw, NativeJsonlAgent::Opencode.label()).unwrap(); + assert_eq!(events.len(), 1); + } + + #[test] + fn codex_native_identity_is_taken_from_the_trajectory() { + let native = json!({"session_id": "native-session"}); + assert_eq!(codex_native_session_id(&native).unwrap(), "native-session"); + assert!(codex_native_session_id(&json!({})).is_err()); + } + + #[test] + fn codex_native_identity_cannot_be_overridden_by_router_session() { + let plan = ReplayPlan { + agent: AgentKind::Codex, + source_path: PathBuf::from("/trajectory.jsonl"), + source_sha256: "sha".into(), + after_step: 1, + prefix_model_turns: 1, + native: json!({"session_id": "native-session", "user_prompt": "Original task"}), + original_next_action: None, + batches: Vec::new(), + }; + let request = PlaybackRequest { + agent: AgentKind::Codex, + trajectory: PathBuf::from("/trajectory.jsonl"), + after_step: 1, + workspace: PathBuf::from("/workspace"), + state_dir: PathBuf::from("/state"), + output_dir: PathBuf::from("/output"), + agent_entrypoint: None, + agent_runtime: None, + disallowed_tools: Vec::new(), + trajectory_assets: None, + session_id: Some("sweeval-router-key".into()), + max_steps: None, + mode: ReplayMode::ReplayAndContinue, + allow_stale_observations: false, + run_id: None, + disable_thinking: false, + boundary_user_prompt: None, + }; + let context = RunContext { + request: &request, + state_dir: std::path::Path::new("/state"), + output_dir: std::path::Path::new("/output"), + launch: None, + session_id: "sweeval-router-key", + nonce: "nonce", + }; + assert_eq!( + continuation_session_id(NativeJsonlAgent::Codex, &plan, &context).unwrap(), + "native-session" + ); + } + + #[test] + fn codex_continuation_rejects_a_fresh_session_without_the_staged_prefix() { + let plan = ReplayPlan { + agent: AgentKind::Codex, + source_path: PathBuf::from("/trajectory.jsonl"), + source_sha256: "sha".into(), + after_step: 1, + prefix_model_turns: 1, + native: json!({"session_id": "native-session", "user_prompt": "Original task"}), + original_next_action: None, + batches: vec![ToolBatch { + ordinal: 1, + native_locator: "events:0-3".into(), + assistant_text: "before".into(), + native: json!({"start_event": 0, "end_event": 3}), + tool_calls: vec![ToolCall { + ordinal: 1, + call_id: "boundary-call".into(), + name: "exec_command".into(), + arguments: json!({"cmd": "true"}), + original_observation: json!("old"), + original_is_error: false, + native: json!({"call_event": 2, "output_event": 3}), + }], + }], + }; + let fresh = vec![ + json!({"type":"session_meta","payload":{"id":"native-session"}}), + json!({"type":"response_item","payload":{"type":"message","role":"assistant","content":[]}}), + ]; + let error = validate_codex_continuation( + &fresh, + &plan, + "native-session", + std::path::Path::new("/rollout.jsonl"), + ) + .unwrap_err(); + assert!(error.message.contains("user task") || error.message.contains("boundary")); + + let resumed = vec![ + json!({"type":"session_meta","payload":{"id":"native-session"}}), + json!({"type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"Original task"}]}}), + json!({"type":"response_item","payload":{"type":"message","role":"assistant","content":[]}}), + json!({"type":"response_item","payload":{"type":"function_call","call_id":"boundary-call"}}), + json!({"type":"response_item","payload":{"type":"message","role":"assistant","content":[]}}), + ]; + validate_codex_continuation( + &resumed, + &plan, + "native-session", + std::path::Path::new("/rollout.jsonl"), + ) + .unwrap(); + } +} diff --git a/crates/persisting-replay/src/adapter/mod.rs b/crates/persisting-replay/src/adapter/mod.rs index 8e03f513..b9f4f551 100644 --- a/crates/persisting-replay/src/adapter/mod.rs +++ b/crates/persisting-replay/src/adapter/mod.rs @@ -5,6 +5,7 @@ use std::process::Command; use std::time::Duration; mod claude_code; +mod generic; mod mini_swe_agent; mod openhands; mod pi_agent; @@ -41,8 +42,14 @@ pub struct RunContext<'a> { pub fn build_plan(request: &PlaybackRequest) -> Result { match request.agent { AgentKind::ClaudeCode => claude_code::build(request), + AgentKind::Codex => { + generic::build(request, generic::NativeJsonlAgent::Codex).map(AdapterPlan::Codex) + } AgentKind::MiniSweAgent => mini_swe_agent::build(request), AgentKind::Openhands => openhands::build(request), + AgentKind::Opencode => { + generic::build(request, generic::NativeJsonlAgent::Opencode).map(AdapterPlan::Opencode) + } AgentKind::PiAgent => pi_agent::build(request), AgentKind::SweAgent => swe_agent::build(request), } @@ -55,8 +62,14 @@ pub fn run( ) -> Result { match plan { AdapterPlan::ClaudeCode(plan) => claude_code::execute(plan, context, journal), + AdapterPlan::Codex(plan) => { + generic::execute(plan, context, journal, generic::NativeJsonlAgent::Codex) + } AdapterPlan::MiniSweAgent(plan) => mini_swe_agent::execute(plan, context, journal), AdapterPlan::Openhands(plan) => openhands::execute(plan, context, journal), + AdapterPlan::Opencode(plan) => { + generic::execute(plan, context, journal, generic::NativeJsonlAgent::Opencode) + } AdapterPlan::PiAgent(plan) => pi_agent::execute(plan, context, journal), AdapterPlan::SweAgent(plan) => swe_agent::execute(plan, context, journal), } diff --git a/crates/persisting-replay/src/adapter/runtime.rs b/crates/persisting-replay/src/adapter/runtime.rs index de350157..e216dedb 100644 --- a/crates/persisting-replay/src/adapter/runtime.rs +++ b/crates/persisting-replay/src/adapter/runtime.rs @@ -162,7 +162,11 @@ fn probe_version(agent: AgentKind, entrypoint: &Path) -> Result { + AgentKind::ClaudeCode + | AgentKind::Codex + | AgentKind::MiniSweAgent + | AgentKind::Opencode + | AgentKind::PiAgent => { command.arg("--version"); } AgentKind::Openhands => { @@ -240,6 +244,25 @@ fn parse_version(agent: AgentKind, rendered: &str) -> Option<&'static str> { AgentKind::Openhands | AgentKind::PiAgent | AgentKind::SweAgent => { (rendered.trim() == expected).then_some(expected) } + AgentKind::Codex => rendered + .split_whitespace() + .filter_map(|token| token.strip_prefix('v').or(Some(token))) + .find(|version| *version == expected) + .map(|_| expected), + AgentKind::Opencode => rendered + .trim() + .strip_prefix('v') + .unwrap_or_else(|| rendered.trim()) + .strip_suffix("-baseline") + .unwrap_or_else(|| { + rendered + .trim() + .strip_prefix('v') + .unwrap_or_else(|| rendered.trim()) + }) + .trim() + .eq(expected) + .then_some(expected), } } @@ -516,6 +539,17 @@ Loading global config from '/root/.config/mini-swe-agent/.env'"; ); assert_eq!(parse_version(AgentKind::SweAgent, "1.1.0"), Some("1.1.0")); assert_eq!(parse_version(AgentKind::SweAgent, "swe-agent 1.1.0"), None); + assert_eq!( + parse_version(AgentKind::Codex, "codex-cli 0.149.0"), + Some("0.149.0") + ); + assert_eq!(parse_version(AgentKind::Codex, "codex-cli 0.148.0"), None); + assert_eq!(parse_version(AgentKind::Opencode, "1.17.7"), Some("1.17.7")); + assert_eq!( + parse_version(AgentKind::Opencode, "v1.17.7"), + Some("1.17.7") + ); + assert_eq!(parse_version(AgentKind::Opencode, "1.17.6"), None); } #[cfg(unix)] diff --git a/crates/persisting-replay/src/codex_bridge.rs b/crates/persisting-replay/src/codex_bridge.rs new file mode 100644 index 00000000..4a171347 --- /dev/null +++ b/crates/persisting-replay/src/codex_bridge.rs @@ -0,0 +1,761 @@ +//! Codex Responses API resume-transport bridge. +//! +//! Older Codex releases require a prompt argument for `exec resume`. The +//! prompt is useful as a CLI wake-up signal, but it must not become part of +//! the model input for an unmodified replay. This bridge removes the unique +//! nonce from every request before forwarding it upstream; Codex may resend +//! the full conversation history on subsequent requests. + +use std::collections::BTreeMap; +use std::net::TcpListener; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +use anyhow::Context; +use axum::Router; +use axum::body::{Body, Bytes}; +use axum::extract::{DefaultBodyLimit, State}; +use axum::http::header::{AUTHORIZATION, CACHE_CONTROL, CONTENT_TYPE}; +use axum::http::{HeaderMap, HeaderValue, StatusCode}; +use axum::response::Response; +use axum::routing::{get, post}; +use serde_json::{Value, json}; +use tokio::sync::{Notify, oneshot}; + +use crate::error::{ReplayError, ReplayErrorKind, ResultExt}; + +const BRIDGE_VERSION: &str = "sandbox-replay-codex-responses-bridge/1"; +const START_TIMEOUT: Duration = Duration::from_secs(10); +const STOP_TIMEOUT: Duration = Duration::from_secs(5); +const MAX_BODY_BYTES: usize = 64 * 1024 * 1024; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PromptMode { + TransportNonce, + ExplicitUserPrompt, +} + +impl PromptMode { + pub fn as_str(self) -> &'static str { + match self { + Self::TransportNonce => "transport_nonce", + Self::ExplicitUserPrompt => "explicit_user_prompt", + } + } +} + +pub struct CodexBridgeHandle { + pub base_url: String, + api_key: String, + shared: Arc, + shutdown: Option>, + worker_done: Option>>, + worker: Option>, +} + +struct BridgeShared { + state: Mutex, + client: reqwest::Client, + upstream_url: String, + upstream_api_key: String, + routing_session_id: String, + bridge_api_key: String, + prompt_mode: PromptMode, + transport_prompt: String, + explicit_prompt: Option, + cancelled: AtomicBool, + cancel_notify: Notify, +} + +struct BridgeState { + request_sequence: usize, + forwarded_requests: usize, + removed_transport_prompt: bool, + pending_forward_sequence: Option, + failed: bool, + failure: Option, +} + +impl BridgeState { + fn fail(&mut self, message: impl Into) { + self.failed = true; + if self.failure.is_none() { + self.failure = Some(message.into()); + } + } +} + +impl CodexBridgeHandle { + pub fn start( + routing_session_id: &str, + transport_prompt: String, + explicit_prompt: Option<&str>, + ) -> Result { + let upstream_base = first_nonempty_env(&[ + "OPENAI_BASE_URL", + "OPENAI_API_BASE", + "LLM_BASE_URL", + ]) + .ok_or_else(|| { + ReplayError::configuration( + "Codex SandboxReplay bridge requires OPENAI_BASE_URL, OPENAI_API_BASE, or LLM_BASE_URL", + ) + })?; + let upstream_api_key = + first_nonempty_env(&["OPENAI_API_KEY", "LLM_API_KEY"]).ok_or_else(|| { + ReplayError::configuration( + "Codex SandboxReplay bridge requires OPENAI_API_KEY or LLM_API_KEY", + ) + })?; + let upstream_url = responses_url(&upstream_base)?; + let prompt_mode = if explicit_prompt.is_some() { + PromptMode::ExplicitUserPrompt + } else { + PromptMode::TransportNonce + }; + let bridge_api_key = format!("pvisor-sandbox-replay-{}", uuid::Uuid::new_v4().simple()); + let listener = TcpListener::bind(("127.0.0.1", 0)).replay_context( + ReplayErrorKind::Continuation, + "allocate Codex SandboxReplay bridge port", + )?; + let address = listener.local_addr().replay_context( + ReplayErrorKind::Continuation, + "read Codex SandboxReplay bridge address", + )?; + listener.set_nonblocking(true).replay_context( + ReplayErrorKind::Continuation, + "configure Codex SandboxReplay bridge listener", + )?; + let client = reqwest::Client::builder() + .no_proxy() + .build() + .replay_context( + ReplayErrorKind::Continuation, + "build Codex SandboxReplay bridge client", + )?; + let shared = Arc::new(BridgeShared { + state: Mutex::new(BridgeState { + request_sequence: 0, + forwarded_requests: 0, + removed_transport_prompt: false, + pending_forward_sequence: None, + failed: false, + failure: None, + }), + client, + upstream_url, + upstream_api_key, + routing_session_id: routing_session_id.to_owned(), + bridge_api_key: bridge_api_key.clone(), + prompt_mode, + transport_prompt, + explicit_prompt: explicit_prompt.map(str::to_owned), + cancelled: AtomicBool::new(false), + cancel_notify: Notify::new(), + }); + let router = router(Arc::clone(&shared)); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let (ready_tx, ready_rx) = mpsc::sync_channel(1); + let (done_tx, done_rx) = mpsc::sync_channel(1); + let worker = thread::Builder::new() + .name("pvisor-codex-replay-bridge".into()) + .spawn(move || { + let result = run_worker(listener, router, shutdown_rx, ready_tx); + let _ = done_tx.send(result); + }) + .replay_context( + ReplayErrorKind::Continuation, + "start Codex SandboxReplay bridge thread", + )?; + let mut handle = Self { + base_url: format!("http://{address}/v1"), + api_key: bridge_api_key, + shared, + shutdown: Some(shutdown_tx), + worker_done: Some(done_rx), + worker: Some(worker), + }; + let startup_error = match ready_rx.recv_timeout(START_TIMEOUT) { + Ok(Ok(())) => None, + Ok(Err(message)) => Some(message), + Err(mpsc::RecvTimeoutError::Timeout) => Some(format!( + "Codex SandboxReplay bridge did not become ready within {} seconds", + START_TIMEOUT.as_secs() + )), + Err(mpsc::RecvTimeoutError::Disconnected) => { + Some("Codex SandboxReplay bridge exited before reporting readiness".into()) + } + }; + if let Some(message) = startup_error { + let _ = handle.stop_worker(); + return Err(ReplayError::continuation(message)); + } + Ok(handle) + } + + pub fn child_environment(&self) -> BTreeMap { + let no_proxy = merged_no_proxy_environment(); + BTreeMap::from([ + ("OPENAI_BASE_URL".into(), self.base_url.clone()), + ("OPENAI_API_BASE".into(), self.base_url.clone()), + ("OPENAI_API_KEY".into(), self.api_key.clone()), + ("NO_PROXY".into(), no_proxy.clone()), + ("no_proxy".into(), no_proxy), + ]) + } + + pub fn prompt_mode(&self) -> PromptMode { + self.shared.prompt_mode + } + + pub fn finish(mut self) -> Result { + self.stop_worker()?; + let state = self + .shared + .state + .lock() + .map_err(|_| ReplayError::continuation("Codex bridge state lock poisoned"))?; + if state.failed { + return Err(ReplayError::continuation(format!( + "Codex resume transport bridge failed closed: {}", + state + .failure + .as_deref() + .unwrap_or("unknown protocol failure") + ))); + } + if state.pending_forward_sequence.is_some() { + return Err(ReplayError::continuation( + "Codex bridge has a validated request that was not forwarded", + )); + } + if state.forwarded_requests == 0 { + return Err(ReplayError::continuation( + "Codex continuation made no validated model request through the SandboxReplay bridge", + )); + } + if self.shared.prompt_mode == PromptMode::TransportNonce && !state.removed_transport_prompt + { + return Err(ReplayError::continuation( + "Codex transport nonce was not removed from the first model request", + )); + } + Ok(state.forwarded_requests) + } + + fn stop_worker(&mut self) -> Result<(), ReplayError> { + self.shared.cancelled.store(true, Ordering::Release); + self.shared.cancel_notify.notify_waiters(); + if let Some(shutdown) = self.shutdown.take() { + let _ = shutdown.send(()); + } + let worker_result = match self.worker_done.take() { + Some(done) => match done.recv_timeout(STOP_TIMEOUT) { + Ok(result) => Some(result), + Err(mpsc::RecvTimeoutError::Disconnected) => None, + Err(mpsc::RecvTimeoutError::Timeout) => { + self.worker.take(); + return Err(ReplayError::continuation(format!( + "Codex SandboxReplay bridge did not stop within {} seconds", + STOP_TIMEOUT.as_secs() + ))); + } + }, + None => None, + }; + if let Some(worker) = self.worker.take() + && worker.join().is_err() + { + return Err(ReplayError::continuation( + "Codex SandboxReplay bridge thread panicked", + )); + } + if let Some(result) = worker_result { + result.replay_context(ReplayErrorKind::Executor, "stop Codex SandboxReplay bridge")?; + } + Ok(()) + } +} + +impl Drop for CodexBridgeHandle { + fn drop(&mut self) { + let _ = self.stop_worker(); + } +} + +fn run_worker( + listener: TcpListener, + router: Router, + shutdown_rx: oneshot::Receiver<()>, + ready_tx: mpsc::SyncSender>, +) -> anyhow::Result<()> { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .map_err(|error| { + let _ = ready_tx.send(Err(format!("build Codex bridge runtime: {error}"))); + error + })?; + runtime.block_on(async move { + let listener = tokio::net::TcpListener::from_std(listener).map_err(|error| { + let _ = ready_tx.send(Err(format!("adopt Codex bridge listener: {error}"))); + error + })?; + ready_tx + .send(Ok(())) + .map_err(|_| anyhow::anyhow!("Codex bridge startup receiver was dropped"))?; + axum::serve(listener, router) + .with_graceful_shutdown(async move { + let _ = shutdown_rx.await; + }) + .await?; + anyhow::Ok(()) + }) +} + +fn router(shared: Arc) -> Router { + Router::new() + .route("/health", get(health)) + .route("/responses", post(responses)) + .route("/v1/responses", post(responses)) + .fallback(not_found) + .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)) + .with_state(shared) +} + +async fn health(State(shared): State>) -> Response { + let (failed, request_sequence) = shared + .state + .lock() + .map(|state| (state.failed, state.request_sequence)) + .unwrap_or((true, 0)); + response( + StatusCode::OK.as_u16(), + "application/json", + json!({ + "status": "healthy", + "bridge_version": BRIDGE_VERSION, + "resume_mode": true, + "resume_failed": failed, + "request_sequence": request_sequence, + }) + .to_string() + .into_bytes(), + ) +} + +async fn responses( + State(shared): State>, + headers: HeaderMap, + body: Bytes, +) -> Response { + if !authorized(&shared, &headers) { + return error_response(StatusCode::UNAUTHORIZED, "invalid bridge API key"); + } + let payload: Value = match serde_json::from_slice(&body) { + Ok(Value::Object(payload)) => Value::Object(payload), + Ok(_) => return error_response(StatusCode::BAD_REQUEST, "request must be a JSON object"), + Err(error) => { + return error_response(StatusCode::BAD_REQUEST, &format!("invalid JSON: {error}")); + } + }; + let (cleaned, sequence) = match clean_request(&shared, payload) { + Ok(result) => result, + Err(error) => { + fail(&shared, error.to_string()); + return error_response(StatusCode::UNPROCESSABLE_ENTITY, &error.to_string()); + } + }; + let serialized = match serde_json::to_vec(&cleaned) { + Ok(serialized) => serialized, + Err(error) => { + fail(&shared, format!("serialize cleaned Codex request: {error}")); + return error_response(StatusCode::UNPROCESSABLE_ENTITY, &error.to_string()); + } + }; + let upstream = match forward(&shared, serialized).await { + Ok(response) => response, + Err(error) => { + fail(&shared, error.to_string()); + return error_response(StatusCode::BAD_GATEWAY, &error.to_string()); + } + }; + { + let mut state = match shared.state.lock() { + Ok(state) => state, + Err(_) => { + return error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "bridge state lock poisoned", + ); + } + }; + if state.pending_forward_sequence != Some(sequence) { + state.fail("forwarded Codex request sequence did not match the validated request"); + return error_response( + StatusCode::UNPROCESSABLE_ENTITY, + "bridge request sequence mismatch", + ); + } + state.pending_forward_sequence = None; + state.forwarded_requests += 1; + } + response_with_headers(upstream.status, upstream.headers, upstream.body) +} + +struct UpstreamResponse { + status: u16, + headers: HeaderMap, + body: Vec, +} + +async fn forward(shared: &BridgeShared, body: Vec) -> anyhow::Result { + if shared.cancelled.load(Ordering::Acquire) { + anyhow::bail!("Codex SandboxReplay bridge was cancelled"); + } + let request = shared + .client + .post(&shared.upstream_url) + .header( + AUTHORIZATION.as_str(), + format!("Bearer {}", shared.upstream_api_key), + ) + .header(CONTENT_TYPE.as_str(), "application/json") + .header("X-LiteLLM-Session-ID", &shared.routing_session_id) + .body(body) + .send() + .await?; + let status = request.status().as_u16(); + let headers: HeaderMap = request + .headers() + .iter() + .filter(|(name, _)| !is_hop_by_hop_header(name.as_str())) + .map(|(name, value)| (name.clone(), value.clone())) + .collect(); + Ok(UpstreamResponse { + status, + headers, + body: request.bytes().await?.to_vec(), + }) +} + +fn clean_request(shared: &Arc, mut payload: Value) -> anyhow::Result<(Value, usize)> { + let mut state = shared + .state + .lock() + .map_err(|_| anyhow::anyhow!("Codex bridge state lock poisoned"))?; + if state.failed { + anyhow::bail!( + "Codex bridge is failed closed: {}", + state.failure.as_deref().unwrap_or("unknown failure") + ); + } + if state.pending_forward_sequence.is_some() { + state.fail("another Codex request arrived before the previous request was forwarded"); + anyhow::bail!("another request arrived before the previous request was forwarded"); + } + state.request_sequence += 1; + let sequence = state.request_sequence; + let input = payload + .get_mut("input") + .context("Codex Responses request has no input")?; + match shared.prompt_mode { + PromptMode::TransportNonce => { + let removed = remove_exact_user_input(input, &shared.transport_prompt)?; + if sequence == 1 && removed != 1 { + state.fail(format!( + "expected exactly one Codex transport nonce in the first request, found {removed}" + )); + anyhow::bail!( + "expected exactly one Codex transport nonce in the first request, found {removed}" + ); + } + if sequence == 1 { + state.removed_transport_prompt = true; + } + } + PromptMode::ExplicitUserPrompt => { + if sequence == 1 { + let prompt = shared + .explicit_prompt + .as_deref() + .context("explicit Codex prompt mode has no prompt")?; + let count = count_exact_user_input(input, prompt)?; + if count != 1 { + state.fail(format!( + "expected exactly one explicit Codex boundary prompt in the first request, found {count}" + )); + anyhow::bail!( + "expected exactly one explicit Codex boundary prompt in the first request, found {count}" + ); + } + } + } + } + state.pending_forward_sequence = Some(sequence); + Ok((payload, sequence)) +} + +fn remove_exact_user_input(input: &mut Value, expected: &str) -> anyhow::Result { + let items = input + .as_array_mut() + .context("Codex Responses input must be an array for resume transport cleanup")?; + let mut matches = Vec::new(); + for (index, item) in items.iter().enumerate() { + if item.get("role").and_then(Value::as_str) == Some("user") + && exact_message_text(item) == Some(expected) + { + matches.push(index); + } + } + if matches.len() > 1 { + anyhow::bail!("Codex transport nonce occurred more than once in input"); + } + if let Some(index) = matches.first().copied() { + items.remove(index); + Ok(1) + } else { + Ok(0) + } +} + +fn count_exact_user_input(input: &Value, expected: &str) -> anyhow::Result { + let items = input + .as_array() + .context("Codex Responses input must be an array for resume transport validation")?; + Ok(items + .iter() + .filter(|item| { + item.get("role").and_then(Value::as_str) == Some("user") + && exact_message_text(item) == Some(expected) + }) + .count()) +} + +fn exact_message_text(message: &Value) -> Option<&str> { + let content = message.get("content")?; + if let Some(text) = content.as_str() { + return Some(text); + } + let blocks = content.as_array()?; + if blocks.len() != 1 { + return None; + } + let block = &blocks[0]; + if block.get("type").and_then(Value::as_str) != Some("input_text") { + return None; + } + block.get("text").and_then(Value::as_str) +} + +fn authorized(shared: &BridgeShared, headers: &HeaderMap) -> bool { + let supplied = headers + .get("x-api-key") + .and_then(|value| value.to_str().ok()) + .or_else(|| { + headers + .get(AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + }); + supplied == Some(shared.bridge_api_key.as_str()) +} + +fn fail(shared: &BridgeShared, message: String) { + if let Ok(mut state) = shared.state.lock() { + state.fail(message); + } +} + +fn response(status: u16, content_type: &str, body: Vec) -> Response { + let mut response = Response::new(Body::from(body)); + *response.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::BAD_GATEWAY); + if let Ok(value) = HeaderValue::from_str(content_type) { + response.headers_mut().insert(CONTENT_TYPE, value); + } + response + .headers_mut() + .insert(CACHE_CONTROL, HeaderValue::from_static("no-cache")); + response +} + +fn response_with_headers(status: u16, headers: HeaderMap, body: Vec) -> Response { + let mut response = Response::new(Body::from(body)); + *response.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::BAD_GATEWAY); + *response.headers_mut() = headers; + response +} + +fn is_hop_by_hop_header(name: &str) -> bool { + matches!( + name.to_ascii_lowercase().as_str(), + "connection" + | "keep-alive" + | "proxy-authenticate" + | "proxy-authorization" + | "te" + | "trailer" + | "transfer-encoding" + | "upgrade" + | "content-length" + ) +} + +fn error_response(status: StatusCode, message: &str) -> Response { + response( + status.as_u16(), + "application/json", + json!({"error": {"message": message}}) + .to_string() + .into_bytes(), + ) +} + +async fn not_found() -> Response { + error_response(StatusCode::NOT_FOUND, "not found") +} + +fn responses_url(base: &str) -> Result { + let mut url = reqwest::Url::parse(base).map_err(|error| { + ReplayError::configuration(format!("invalid OpenAI base URL {base:?}: {error}")) + })?; + let path = url.path().trim_end_matches('/'); + let path = if path.ends_with("/responses") { + path.to_owned() + } else if path.ends_with("/v1") { + format!("{path}/responses") + } else if path.is_empty() { + "/v1/responses".into() + } else { + format!("{path}/v1/responses") + }; + url.set_path(&path); + url.set_query(None); + url.set_fragment(None); + Ok(url.to_string()) +} + +fn merged_no_proxy_environment() -> String { + let configured = ["NO_PROXY", "no_proxy"] + .iter() + .filter_map(|name| std::env::var(name).ok()) + .collect::>(); + let mut entries = Vec::new(); + for value in configured { + for entry in value + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + { + if !entries.iter().any(|existing| existing == entry) { + entries.push(entry.to_owned()); + } + } + } + for required in ["127.0.0.1", "localhost", "::1"] { + if !entries.iter().any(|existing| existing == required) { + entries.push(required.into()); + } + } + entries.join(",") +} + +fn first_nonempty_env(names: &[&str]) -> Option { + names.iter().find_map(|name| { + std::env::var(name) + .ok() + .filter(|value| !value.trim().is_empty()) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn responses_url_appends_responses_endpoint() { + assert_eq!( + responses_url("http://model/v1").unwrap(), + "http://model/v1/responses" + ); + assert_eq!( + responses_url("http://model/v1/").unwrap(), + "http://model/v1/responses" + ); + assert_eq!( + responses_url("http://model").unwrap(), + "http://model/v1/responses" + ); + } + + #[test] + fn hop_by_hop_response_headers_are_not_forwarded() { + assert!(is_hop_by_hop_header("connection")); + assert!(is_hop_by_hop_header("Transfer-Encoding")); + assert!(is_hop_by_hop_header("content-length")); + assert!(!is_hop_by_hop_header("content-type")); + assert!(!is_hop_by_hop_header("date")); + } + + #[test] + fn removes_exact_nonce_from_responses_input() { + let mut input = json!([ + {"role":"user","content":[{"type":"input_text","text":"task"}]}, + {"role":"user","content":[{"type":"input_text","text":"nonce-1234567890"}]}, + {"role":"assistant","content":[]} + ]); + assert_eq!( + remove_exact_user_input(&mut input, "nonce-1234567890").unwrap(), + 1 + ); + assert_eq!(input.as_array().unwrap().len(), 2); + assert_eq!( + count_exact_user_input(&input, "nonce-1234567890").unwrap(), + 0 + ); + } + + #[test] + fn nonce_must_be_unique() { + let mut input = json!([ + {"role":"user","content":"nonce"}, + {"role":"user","content":"nonce"} + ]); + assert!(remove_exact_user_input(&mut input, "nonce").is_err()); + } + + #[test] + fn historical_nonce_is_removed_from_later_requests() { + let nonce = "pvisor-codex-resume-test"; + let mut first = json!({ + "input": [ + {"role": "user", "content": [{"type": "input_text", "text": nonce}]}, + {"role": "assistant", "content": [{"type": "output_text", "text": "ok"}]} + ] + }); + let mut second = first.clone(); + let removed_first = + remove_exact_user_input(first.get_mut("input").unwrap(), nonce).unwrap(); + let removed_second = + remove_exact_user_input(second.get_mut("input").unwrap(), nonce).unwrap(); + assert_eq!(removed_first, 1); + assert_eq!(removed_second, 1); + assert!(!first.to_string().contains(nonce)); + assert!(!second.to_string().contains(nonce)); + } + + #[test] + fn explicit_prompt_is_not_removed() { + let input = json!([ + {"role":"user","content":"task"}, + {"role":"user","content":"review O-prime N"} + ]); + assert_eq!( + count_exact_user_input(&input, "review O-prime N").unwrap(), + 1 + ); + } +} diff --git a/crates/persisting-replay/src/engine.rs b/crates/persisting-replay/src/engine.rs index 88b53bc8..52759d34 100644 --- a/crates/persisting-replay/src/engine.rs +++ b/crates/persisting-replay/src/engine.rs @@ -508,8 +508,10 @@ fn artifacts( ]; let native_format = match request.agent { AgentKind::ClaudeCode => "claude-code/native-jsonl-2.1.220", + AgentKind::Codex => "codex/native-jsonl-0.149.0", AgentKind::MiniSweAgent => "mini-swe-agent/native-json-2.4.6", AgentKind::Openhands => "openhands/native-json-0.53.0", + AgentKind::Opencode => "opencode/native-events-jsonl-1.17.7", AgentKind::PiAgent => "pi-agent/native-events-jsonl-0.83.0", AgentKind::SweAgent => "swe-agent/native-traj-1.1.0", }; @@ -521,6 +523,16 @@ fn artifacts( prepared_path.clone(), )); } + if request.agent == AgentKind::Opencode { + let path = output_dir.join("native/opencode-session.json"); + if path.is_file() { + artifacts.push(artifact( + "native_session_export", + "opencode/session-export-v1", + path, + )); + } + } if let Some(path) = &outcome.reconstructed_path && path != &prepared_path { @@ -600,8 +612,10 @@ fn existing_artifacts(request: &PlaybackRequest, output_dir: &Path) -> Vec "claude-code/native-jsonl-2.1.220", + AgentKind::Codex => "codex/native-jsonl-0.149.0", AgentKind::MiniSweAgent => "mini-swe-agent/native-json-2.4.6", AgentKind::Openhands => "openhands/native-json-0.53.0", + AgentKind::Opencode => "opencode/native-events-jsonl-1.17.7", AgentKind::PiAgent => "pi-agent/native-events-jsonl-0.83.0", AgentKind::SweAgent => "swe-agent/native-traj-1.1.0", }; @@ -617,6 +631,17 @@ fn existing_artifacts(request: &PlaybackRequest, output_dir: &Path) -> Vec &[ + ("prepared_native_prefix", "native/prepared-prefix.jsonl"), + ( + "reconstructed_native_trajectory", + "native/reconstructed-trajectory.jsonl", + ), + ( + "continued_native_trajectory", + "native/continued-trajectory.jsonl", + ), + ], AgentKind::MiniSweAgent => &[ ("prepared_native_prefix", "native/prepared-prefix.json"), ( @@ -671,6 +696,16 @@ fn existing_artifacts(request: &PlaybackRequest, output_dir: &Path) -> Vec Vec std::path::PathBuf { output_dir.join(match agent { AgentKind::ClaudeCode => "native/prepared-prefix.jsonl", + AgentKind::Codex | AgentKind::Opencode => "native/prepared-prefix.jsonl", AgentKind::MiniSweAgent => "native/prepared-prefix.json", AgentKind::Openhands => "native/prepared-replay-events.json", AgentKind::PiAgent => "native/prepared-prefix.jsonl", diff --git a/crates/persisting-replay/src/lib.rs b/crates/persisting-replay/src/lib.rs index a92804fb..cf42fc7a 100644 --- a/crates/persisting-replay/src/lib.rs +++ b/crates/persisting-replay/src/lib.rs @@ -8,6 +8,7 @@ mod adapter; mod claude_bridge; mod claude_resume; +mod codex_bridge; mod comparison; mod config; mod engine; diff --git a/crates/persisting-replay/src/model.rs b/crates/persisting-replay/src/model.rs index 268b2d78..b06f2668 100644 --- a/crates/persisting-replay/src/model.rs +++ b/crates/persisting-replay/src/model.rs @@ -13,8 +13,10 @@ pub const RESULT_SCHEMA_VERSION: &str = "sandbox-playback.result/v3"; #[serde(rename_all = "kebab-case")] pub enum AgentKind { ClaudeCode, + Codex, MiniSweAgent, Openhands, + Opencode, PiAgent, SweAgent, } @@ -23,8 +25,10 @@ impl AgentKind { pub fn as_str(self) -> &'static str { match self { Self::ClaudeCode => "claude-code", + Self::Codex => "codex", Self::MiniSweAgent => "mini-swe-agent", Self::Openhands => "openhands", + Self::Opencode => "opencode", Self::PiAgent => "pi-agent", Self::SweAgent => "swe-agent", } @@ -33,8 +37,10 @@ impl AgentKind { pub fn supported_version(self) -> &'static str { match self { Self::ClaudeCode => "2.1.220", + Self::Codex => "0.149.0", Self::MiniSweAgent => "2.4.6", Self::Openhands => "0.53.0", + Self::Opencode => "1.17.7", Self::PiAgent => "0.83.0", Self::SweAgent => "1.1.0", } @@ -43,8 +49,10 @@ impl AgentKind { pub fn profile(self) -> &'static str { match self { Self::ClaudeCode => "claude-code/2.1.220/native-resume-v1", + Self::Codex => "codex/0.149.0/native-responses-jsonl-v1", Self::MiniSweAgent => "mini-swe-agent/2.4.6/native-messages-v1", Self::Openhands => "openhands/0.53.0/native-replay-v1", + Self::Opencode => "opencode/1.17.7/native-events-jsonl-v1", Self::PiAgent => "pi-agent/0.83.0/native-rpc-events-v1", Self::SweAgent => "swe-agent/1.1.0/replay-then-live-v1", } @@ -57,12 +65,14 @@ impl FromStr for AgentKind { fn from_str(value: &str) -> Result { match value { "claude-code" => Ok(Self::ClaudeCode), + "codex" => Ok(Self::Codex), "mini-swe-agent" => Ok(Self::MiniSweAgent), "openhands" => Ok(Self::Openhands), + "opencode" => Ok(Self::Opencode), "pi-agent" => Ok(Self::PiAgent), "swe-agent" => Ok(Self::SweAgent), other => Err(format!( - "unsupported agent {other:?}; expected claude-code, mini-swe-agent, openhands, pi-agent, or swe-agent" + "unsupported agent {other:?}; expected claude-code, codex, mini-swe-agent, openhands, opencode, pi-agent, or swe-agent" )), } } @@ -175,8 +185,10 @@ impl ReplayPlan { #[derive(Debug, Clone)] pub(crate) enum AdapterPlan { ClaudeCode(ReplayPlan), + Codex(ReplayPlan), MiniSweAgent(ReplayPlan), Openhands(ReplayPlan), + Opencode(ReplayPlan), PiAgent(ReplayPlan), SweAgent(ReplayPlan), } @@ -213,8 +225,10 @@ impl AdapterPlan { fn plan(&self) -> &ReplayPlan { match self { Self::ClaudeCode(plan) + | Self::Codex(plan) | Self::MiniSweAgent(plan) | Self::Openhands(plan) + | Self::Opencode(plan) | Self::PiAgent(plan) | Self::SweAgent(plan) => plan, } @@ -343,8 +357,10 @@ mod tests { fn adapter_plan_exposes_only_common_dispatch_fields() { let plans = [ AdapterPlan::ClaudeCode(replay_plan(AgentKind::ClaudeCode, "claude")), + AdapterPlan::Codex(replay_plan(AgentKind::Codex, "codex")), AdapterPlan::MiniSweAgent(replay_plan(AgentKind::MiniSweAgent, "mini")), AdapterPlan::Openhands(replay_plan(AgentKind::Openhands, "openhands")), + AdapterPlan::Opencode(replay_plan(AgentKind::Opencode, "opencode")), AdapterPlan::PiAgent(replay_plan(AgentKind::PiAgent, "pi")), AdapterPlan::SweAgent(replay_plan(AgentKind::SweAgent, "swe")), ]; diff --git a/docs/src/en/pvisor/guides/sandbox-replay.md b/docs/src/en/pvisor/guides/sandbox-replay.md index 6fa9052b..a34ad716 100644 --- a/docs/src/en/pvisor/guides/sandbox-replay.md +++ b/docs/src/en/pvisor/guides/sandbox-replay.md @@ -44,6 +44,17 @@ session with new observations, then continues through Pi's SDK. Its initial tool surface is intentionally limited to Pi's `read`, `bash`, `edit`, and `write` tools; trajectories containing another tool fail validation. +OpenCode `1.17.7` and Codex CLI `0.149.0` are also supported. OpenCode consumes +the native `opencode run --format=json` event stream and groups +`step_start`/`tool_use`/`step_finish` events into complete replay steps. Codex +consumes rollout JSONL (`session_meta` plus `response_item` messages, reasoning, +function calls/custom tool calls, and outputs). Both adapters re-execute the +selected command/file tools in the fresh workspace, replace native observations, +write a native JSONL prefix, and then invoke the native continuation command. +OpenCode uses `run --format=json --session`; Codex stages the prefix below an +isolated `CODEX_HOME` and uses `exec resume --json`. Unknown tool shapes fail +explicitly instead of being silently skipped. + ## Run replay ```bash @@ -70,7 +81,7 @@ disable_thinking = true boundary_user_prompt = "Review the fresh observation before continuing." ``` -For a Pi runtime installed at the SweEval default location, the CLI form is: +When the Pi runtime is installed at `/opt/pi-agent`, the CLI form is: ```bash pvisor replay \ @@ -80,6 +91,45 @@ pvisor replay \ --agent-entrypoint /opt/pi-agent/bin/pi ``` +OpenCode consumes its native event stream: + +```bash +pvisor replay \ + --agent opencode \ + --trajectory /input/opencode.jsonl \ + --after-step 30 \ + --agent-entrypoint /usr/bin/opencode +``` + +Codex consumes its native rollout JSONL. SandboxReplay derives the Codex native +session ID from the trajectory's `session_meta`; the request `session_id` remains +a model-router/run key and cannot override the native Codex identity. If the +trajectory has no native session ID, continuation fails closed instead of +starting a fresh conversation: + +```bash +pvisor replay \ + --agent codex \ + --trajectory /input/rollout.jsonl \ + --after-step 30 \ + --agent-entrypoint /usr/bin/codex +``` + +For Codex CLI versions that require a prompt on `exec resume`, SandboxReplay +generates a per-run transport nonce and starts a loopback-only Responses bridge. +The CLI receives the nonce, but the bridge removes it before every upstream +request (Codex may resend the full history on later requests) and fails closed +on malformed or ambiguous requests. The continued native JSONL is also scrubbed +of the nonce and the legacy `Continue from the replay boundary.` message. The +default metadata records `prompt_mode = "transport_nonce"` and +`input_condition = "replayed_boundary_only"`. With an explicit +`boundary_user_prompt`, the user message is retained and metadata records +`prompt_mode = "explicit_user_prompt"` and +`input_condition = "boundary_user_prompt_appended"`. + +The same TOML surface is used for both; change `[replay].agent`, +`trajectory`, and `agent_entrypoint` to the selected runtime. + ### Execution modes and results - The default mode executes the selected prefix and continues with the live Agent. diff --git a/docs/src/en/pvisor/reference/cli.md b/docs/src/en/pvisor/reference/cli.md index 7c5208fb..6e02e512 100644 --- a/docs/src/en/pvisor/reference/cli.md +++ b/docs/src/en/pvisor/reference/cli.md @@ -129,7 +129,7 @@ pvisor replay \ --boundary-user-prompt 'Review the fresh observation before continuing.' ``` -OpenHands, mini-swe-agent, Pi agent, and SWE-agent use the model endpoint and +OpenHands, mini-swe-agent, Pi agent, OpenCode, Codex, and SWE-agent use the model endpoint and credentials already present in their environment. Pi requires its exact `0.83.0` runtime and accepts native RPC event JSONL containing the core `read`, `bash`, `edit`, and `write` tools. Claude Code uses a temporary bridge owned @@ -138,6 +138,14 @@ The bridge validates and removes that exact Resume Transport envelope before forwarding the model request. It does not enable pVisor Gateway, capture model traffic, or persist a bridge audit. +OpenCode requires the exact `1.17.7` runtime and its native +`opencode run --format=json` event JSONL. Codex requires the exact `0.149.0` +runtime and Codex rollout `response_item` JSONL. Both rebuild the native prefix +in the fresh sandbox and invoke their native resume command for continuation. +Codex derives its native session ID from the trajectory's `session_meta`; the +request `session_id` is only a model-router/Run key and cannot override it. +Continuation fails closed when the native session is missing. + The equivalent strict replay TOML is: ```toml @@ -153,8 +161,8 @@ disable_thinking = true boundary_user_prompt = "Review the fresh observation before continuing." ``` -Pi uses the same CLI/TOML surface. Its default SweEval entrypoint is -`/opt/pi-agent/bin/pi`, for example: +Pi uses the same CLI/TOML surface. When the runtime is installed at +`/opt/pi-agent`, for example: ```bash pvisor replay --agent pi-agent \ @@ -184,9 +192,11 @@ structured failure. Existing non-Claude callers that used `replay_only = true` only to construct a prefix must migrate to `prepare_only = true`. `disable_thinking` belongs to `[replay]` and is also exposed as -`--disable-thinking`; it is applied by the Claude protocol bridge without -turning on Gateway capture. Optional `[run]`, `[overlayfs]`, and `[overlaynet]` sections create an outer -managed `pvisor run`; they do not change the inner replay model path. +`--disable-thinking`. Claude Code's protocol bridge applies it to the upstream +request; OpenCode omits its `--thinking` flag when it is set. It does not turn +on Gateway capture. Optional `[run]`, `[overlayfs]`, and `[overlaynet]` sections +create an outer managed `pvisor run`; they do not change the inner replay model +path. By default, replay's internal state, WAL, manifest, fresh-observation comparisons, and native working files remain under diff --git a/docs/src/zh/pvisor/guides/sandbox-replay.md b/docs/src/zh/pvisor/guides/sandbox-replay.md index e0f8572d..f22a2219 100644 --- a/docs/src/zh/pvisor/guides/sandbox-replay.md +++ b/docs/src/zh/pvisor/guides/sandbox-replay.md @@ -76,6 +76,31 @@ SandboxReplay 使用 Pi 自身的工具实现重新执行 `read`、`bash`、`edi 这四种工具之外的调用时会拒绝执行，避免静默改变工具语义。未配置边界提示词时调用 Pi 的原生 `continue()`；配置后则在 `O′N` 后通过 `prompt()` 追加一次用户消息。 +### 3.5 OpenCode + +OpenCode 适配固定支持 `1.17.7`，输入为 `opencode run --format=json` 产生的原生 +事件 JSONL。`user`、`step_start`、`text`、`reasoning`、`tool_use` 和 +`step_finish` 事件会按 step 分组；一个 replay step 对应一个完整的工具调用批次。 +SandboxReplay 在新沙箱中重新执行命令型工具以及 `read`、`write`、`edit` 文件工具， +用新的 `state.output` 重建前缀，并通过 `opencode run --format=json --session` 从边界 +继续。工具执行期间的中间 `tool_use` 状态会合并，避免同一调用被重复回放。 + +### 3.6 Codex + +Codex 适配固定支持 CLI `0.149.0`，输入为 Codex 原生 rollout JSONL。一个 replay step +对应一组完整的 `function_call`/`custom_tool_call` 及其 outputs。SandboxReplay 在新沙箱 +中重放前缀工具，将原生 `response_item` 前缀写入隔离的 `CODEX_HOME`，再执行 +`codex exec resume --json`。native session ID 从 `session_meta` 自动提取； +未知工具或缺少 native session ID 时显式失败，不会退化成全新会话。 + +Codex 保留自身的 system prompt、工具定义和任务上下文，SandboxReplay 只替换边界前的 +observation。为兼容需要 resume prompt 的旧版 CLI，SandboxReplay 使用本地 Responses +bridge 删除 transport nonce，再将请求转发到模型服务；续跑结束后也会从 native trajectory +中清理 nonce。默认模式的输入条件为 `replayed_boundary_only`，不会向模型注入额外提示词。 + +配置 `boundary_user_prompt` 时，提示词只在 `O′N` 后注入一次并保留；此时输入条件标记为 +`boundary_user_prompt_appended`。bridge 校验失败会拒绝续跑，不降级为直连。 + ## 4. 使用方式 ### 4.1 安装 pVisor @@ -151,6 +176,30 @@ max_steps = 200 disable_thinking = true ~~~ +OpenCode 使用原生事件 JSONL： + +~~~bash +pvisor replay \ + --agent opencode \ + --trajectory /input/opencode.jsonl \ + --after-step 30 \ + --agent-entrypoint /usr/bin/opencode +~~~ + +Codex 使用原生 rollout JSONL；SandboxReplay 会自动读取轨迹中的 Codex native +session ID，用户无需手工填写 `session_id`： + +~~~bash +pvisor replay \ + --agent codex \ + --trajectory /input/rollout.jsonl \ + --after-step 30 \ + --agent-entrypoint /usr/bin/codex +~~~ + +两者也可使用同一个 TOML 文件，只需将 `[replay].agent`、`trajectory` 和 +`agent_entrypoint` 分别改为 `opencode` 或 `codex`。 + Claude Code 等价的 pVisor TOML 配置为： ~~~toml @@ -223,6 +272,9 @@ replay journal 不记录提示词明文；Agent 原生的 prepared 或 continued | Pi agent | NodeBB（291） | 54 | 111 | 92 | 1 | 1 | 是 | N/A | | Pi agent | Vuls（666） | 38 | 77 | 62 | 1 | 1 | 否 | 1.00 | | Pi agent | qutebrowser（667） | 37 | 71 | 91 | 1 | 1 | 否 | 0.39 | +| Codex | NodeBB（291） | 37 | 74 | 88 | 1 | 1 | 否 | 0.42 | +| Codex | Vuls（666） | 28 | 57 | 69 | 1 | 1 | 否 | 0.47 | +| Codex | qutebrowser（667） | 27 | 54 | 43 | 1 | 1 | 是 | 0.97 | Claude Code / NodeBB 使用 `N=1`，以避开异步子 Agent 完成后的 Resume Transport canonical-prefix 歧义。边界后的原始可见文本非空，且 `A′(N+1)` 成功复现同一个 `TaskOutput` 调用。 @@ -230,6 +282,10 @@ Pi agent 三题分别在新的任务沙箱中并发执行完成，均未发生 verifier Reward 均为 1。NodeBB 边界两侧的下一动作可见文本均为空，因此文本相似度 按当前指标语义记为 N/A，而不是把两个空字符串报告为 1.00。 +Codex 三题使用 `Ornith-1.5-35B-A3B`，原始轨迹来自 r24，续跑轨迹来自最终成功的 +r30；续跑使用新的任务沙箱和默认 `replayed_boundary_only` 输入条件。Codex 轨迹中未 +发现 transport nonce 或 `Continue from the replay boundary.` 消息。 + ### 5.3 A(N+1) 与 A′(N+1) 以下直接展示原始下一步和续跑下一步。reasoning 已移除；较长的整文件替换只保留修改目标和能够区分调用的关键内容。 @@ -568,6 +624,83 @@ read: 区别：原轨迹直接编辑测试文件，续跑先读取测试文件；工具和参数均不同。文本相似度为 0.39。 +#### Codex + +##### NodeBB（291），N=37 + +`A(N+1)`： + +~~~text +Both files parse. Now let me write a reproduction script that exercises the real NodeBB Redis DB through `db.incrObjectFieldByBulk` to verify all the requirements. +~~~ + +工具：`exec_command`，在 `/app` 下创建并执行 `/tmp/repro.js`，验证 bulk increment、 +负数递增、空输入、非法输入、危险字段名、原子性、缓存失效和多次累加。 + +`A′(N+1)`： + +~~~text +Both files parse. Now let me update the plan and write a comprehensive test script to verify the Redis behavior against a real instance. +~~~ + +工具：`update_plan`，前三个实现步骤已完成，当前步骤为“Write reproduction script and +verify behavior”，后续执行 lint 和 hash test suite。 + +区别：工具和参数不一致，文本相似度为 0.42。 + +##### Vuls（666），N=28 + +`A(N+1)`： + +~~~text +The build cache is now warm and builds succeed. Let me set up a plan and start implementing the changes. +~~~ + +工具：`update_plan`，计划从导出 Debian.Supported、修正 Unmarshall 拼写、补充文档注释、 +增加 Oracle Linux 分支开始。 + +`A′(N+1)`： + +~~~text +The build takes longer than the 10s poll window due to the large dependency tree. Let me run it in the background and poll the log. +~~~ + +工具：`exec_command`： + +~~~bash +cd /app && export PATH=$PATH:/usr/local/go/bin && \ +nohup go build ./gost/ ./scan/ ./oval/ ./report/ > /tmp/build.log 2>&1 & +echo "started $!" +~~~ + +区别：工具和参数不一致，文本相似度为 0.47。 + +##### qutebrowser（667），N=27 + +`A(N+1)`： + +~~~text +Now let me run this gold-style verification test. +~~~ + +工具：`exec_command`： + +~~~bash +cd /app && xvfb-run -a env QT_QPA_PLATFORM=offscreen \ +QTWEBENGINE_CHROMIUM_FLAGS="--no-sandbox --disable-gpu --disable-dev-shm-usage" \ +python -m pytest tests/unit/browser/webengine/test_webview_goldcheck.py -q 2>&1 | tail -20 +~~~ + +`A′(N+1)`： + +~~~text +Now let me run the gold-style verification test. +~~~ + +工具：`exec_command`，命令和参数与原始轨迹完全一致。 + +区别：工具完全一致，文本相似度为 0.97。 + 精确参数见 [`pvisor replay` 命令参考](../reference/cli.md#replay-an-agent-trajectory)；执行边界见 [执行指南](execution.md)。 diff --git a/docs/src/zh/pvisor/reference/cli.md b/docs/src/zh/pvisor/reference/cli.md index 3a5d039d..e56f5de7 100644 --- a/docs/src/zh/pvisor/reference/cli.md +++ b/docs/src/zh/pvisor/reference/cli.md @@ -111,7 +111,7 @@ pvisor replay \ --boundary-user-prompt 'Review the fresh observation before continuing.' ``` -OpenHands、mini-swe-agent、Pi agent 和 SWE-agent 使用环境中已有的模型端点 +OpenHands、mini-swe-agent、Pi agent、OpenCode、Codex 和 SWE-agent 使用环境中已有的模型端点 和凭据。Pi 要求精确的 `0.83.0` runtime，并接受包含核心 `read`、`bash`、 `edit` 和 `write` 工具的原生 RPC event JSONL。Claude Code 使用 SandboxReplay 拥有的临时 bridge，因为它的原生 resume 传输会插入 wake-up @@ -119,6 +119,13 @@ SandboxReplay 拥有的临时 bridge，因为它的原生 resume 传输会插入 envelope。它不启用 pVisor Gateway、不捕获模型流量、也不持久化 bridge 审计。 +OpenCode 要求精确的 `1.17.7` runtime，轨迹格式为 +`opencode run --format=json` 的事件 JSONL；Codex 要求精确的 `0.149.0` runtime， +轨迹格式为 Codex rollout `response_item` JSONL。两者均在新沙箱中重建原生前缀， +并调用各自的原生 resume 命令续跑。 +Codex 的 native session ID 从轨迹 `session_meta` 提取；`session_id` 只作为模型 +路由/Run 标识，不能覆盖该 native session。缺少 native session 时会 fail-closed。 + 等价的严格 replay TOML 是： ```toml @@ -134,8 +141,7 @@ disable_thinking = true boundary_user_prompt = "Review the fresh observation before continuing." ``` -Pi 使用同一套 CLI/TOML 面。它的默认 SweEval entrypoint 是 -`/opt/pi-agent/bin/pi`，例如： +Pi 使用同一套 CLI/TOML 面。runtime 安装在 `/opt/pi-agent` 时，例如： ```bash pvisor replay --agent pi-agent \ @@ -161,10 +167,10 @@ Agent 原生的 prepared 或 continued 轨迹可以包含这条用户消息。 原先只用 `replay_only = true` 来构造前缀的非 Claude 调用方必须迁移到 `prepare_only = true`。 -`disable_thinking` 属于 `[replay]`，也暴露为 `--disable-thinking`；它由 -Claude 协议 bridge 应用，且不会打开 Gateway capture。可选的 `[run]`、 -`[overlayfs]` 和 `[overlaynet]` 段会创建外层受管 `pvisor run`；它们不改变 -内部 replay 模型路径。 +`disable_thinking` 属于 `[replay]`，也暴露为 `--disable-thinking`。Claude Code +由协议 bridge 将其应用到上游请求；OpenCode 设置后会省略 `--thinking`。该选项 +不会打开 Gateway capture。可选的 `[run]`、`[overlayfs]` 和 `[overlaynet]` 段会 +创建外层受管 `pvisor run`；它们不改变内部 replay 模型路径。 默认情况下，replay 的内部状态、WAL、manifest、新鲜 observation 比较和原生 工作文件留在 `/tmp/pvisor-sandbox-replay`，并随 sandbox 消失。Replay 不启用