From 0707a33dc0aa20ae893ad6673af20cb455c759c5 Mon Sep 17 00:00:00 2001 From: guoxu1 Date: Tue, 8 Sep 2026 19:58:21 +0800 Subject: [PATCH 1/6] fix(replay): satisfy workspace clippy and pin OpenCode sampling --- .../persisting-replay/src/adapter/generic.rs | 113 +++++++++++++++++- 1 file changed, 107 insertions(+), 6 deletions(-) diff --git a/crates/persisting-replay/src/adapter/generic.rs b/crates/persisting-replay/src/adapter/generic.rs index 88baba1d..38bdc521 100644 --- a/crates/persisting-replay/src/adapter/generic.rs +++ b/crates/persisting-replay/src/adapter/generic.rs @@ -1011,6 +1011,7 @@ fn continue_native_cli( 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"); + write_opencode_provider_config(&opencode_config)?; atomic_write_json( &export_path, &opencode_export(plan, prefix, &session_id, &context.request.workspace), @@ -1150,10 +1151,7 @@ fn continue_native_cli( 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_result = codex_bridge.take().map(|bridge| bridge.finish()); let bridge_error = bridge_result.and_then(|result| result.err()); if !output.status.success() { let process_error = ReplayError::classify_continuation( @@ -1230,6 +1228,71 @@ fn configured_model_from_environment() -> Option { .filter(|model| !model.trim().is_empty()) } +fn env_f64(name: &str) -> Option { + std::env::var(name) + .ok() + .and_then(|value| value.trim().parse::().ok()) +} + +/// Provider config for the isolated continuation `XDG_CONFIG_HOME`. +/// +/// OpenCode reads the endpoint from `OPENAI_BASE_URL`, but sampling options +/// have no environment channel, so a live continuation would silently fall +/// back to provider defaults and diverge from the recorded sampling. The +/// shape mirrors what a SweEval trial writes for the original run. +fn opencode_provider_config( + model: &str, + base_url: Option<&str>, + temperature: Option, + top_p: Option, +) -> Option { + let (provider, model_id) = model.split_once('/')?; + let base_url = base_url.map(str::trim).filter(|value| !value.is_empty()); + if base_url.is_none() && temperature.is_none() && top_p.is_none() { + return None; + } + let mut provider_config = serde_json::Map::new(); + if let Some(base_url) = base_url { + provider_config.insert("options".into(), json!({ "baseURL": base_url })); + } + if temperature.is_some() || top_p.is_some() { + let mut model_options = serde_json::Map::new(); + if let Some(temperature) = temperature { + model_options.insert("temperature".into(), json!(temperature)); + } + if let Some(top_p) = top_p { + model_options.insert("topP".into(), json!(top_p)); + } + provider_config.insert( + "models".into(), + json!({ model_id: { "options": Value::Object(model_options) } }), + ); + } + Some(json!({ "provider": { provider: Value::Object(provider_config) } })) +} + +fn write_opencode_provider_config(config_root: &Path) -> Result<(), ReplayError> { + let Some(model) = configured_model_from_environment() else { + return Ok(()); + }; + let base_url = std::env::var("OPENAI_BASE_URL") + .ok() + .or_else(|| std::env::var("OPENAI_API_BASE").ok()); + let config = opencode_provider_config( + &model, + base_url.as_deref(), + env_f64("PVISOR_OPENCODE_TEMPERATURE"), + env_f64("PVISOR_OPENCODE_TOP_P"), + ); + let Some(config) = config else { + return Ok(()); + }; + let directory = config_root.join("opencode"); + fs::create_dir_all(&directory) + .replay_context(ReplayErrorKind::Executor, "create OpenCode config directory")?; + atomic_write_json(&directory.join("opencode.json"), &config) +} + fn continuation_session_id( agent: NativeJsonlAgent, plan: &ReplayPlan, @@ -1843,12 +1906,50 @@ mod tests { 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, + continuation_session_id, is_actionable_turn, opencode_provider_config, 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_provider_config_mirrors_recorded_sampling() { + let config = opencode_provider_config( + "openai/model-x", + Some("http://127.0.0.1:8000/v1"), + Some(0.0), + Some(1.0), + ) + .unwrap(); + assert_eq!( + config, + json!({ + "provider": { + "openai": { + "options": {"baseURL": "http://127.0.0.1:8000/v1"}, + "models": {"model-x": {"options": {"temperature": 0.0, "topP": 1.0}}} + } + } + }) + ); + + // Without sampling overrides the endpoint still comes from the + // environment, so only the baseURL section is written. + let base_only = opencode_provider_config("openai/model-x", Some("http://m:1/v1"), None, None) + .unwrap(); + assert_eq!( + base_only, + json!({"provider": {"openai": {"options": {"baseURL": "http://m:1/v1"}}}}) + ); + + // Nothing to pin: leave OpenCode on its environment-only defaults. + assert!(opencode_provider_config("openai/model-x", None, None, None).is_none()); + // A model without a provider namespace cannot be pinned either. + assert!(opencode_provider_config("model-x", Some("http://m:1/v1"), Some(0.0), None).is_none()); + // Blank endpoints are ignored rather than written. + assert!(opencode_provider_config("openai/model-x", Some(" "), None, None).is_none()); + } + #[test] fn opencode_events_group_tool_parts_into_complete_turns() { let source = [ From 54af7a803b74c10968a8ac3e3149921aae00e4dd Mon Sep 17 00:00:00 2001 From: guoxu1 Date: Tue, 8 Sep 2026 20:15:23 +0800 Subject: [PATCH 2/6] fix(replay): stabilize OpenCode continuation --- .../persisting-replay/src/adapter/generic.rs | 590 ++++++++++++++++-- 1 file changed, 521 insertions(+), 69 deletions(-) diff --git a/crates/persisting-replay/src/adapter/generic.rs b/crates/persisting-replay/src/adapter/generic.rs index 38bdc521..0506a74e 100644 --- a/crates/persisting-replay/src/adapter/generic.rs +++ b/crates/persisting-replay/src/adapter/generic.rs @@ -1007,66 +1007,11 @@ fn continue_native_cli( 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"); - write_opencode_provider_config(&opencode_config)?; - 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); - } + // The CLI refuses `run --session` without a message, and any + // message would contaminate the boundary. Continue through the + // server API instead; see opencode_server_continuation. + let _ = &command; + return opencode_server_continuation(plan, context, journal, prefix, &session_id, &log_path); } NativeJsonlAgent::Codex => { let explicit_prompt = context.request.boundary_user_prompt().map(str::to_owned); @@ -1240,6 +1185,8 @@ fn env_f64(name: &str) -> Option { /// have no environment channel, so a live continuation would silently fall /// back to provider defaults and diverge from the recorded sampling. The /// shape mirrors what a SweEval trial writes for the original run. +/// `effective_base` overrides the environment endpoint (used for the local +/// sampling-injection proxy). fn opencode_provider_config( model: &str, base_url: Option<&str>, @@ -1271,18 +1218,26 @@ fn opencode_provider_config( Some(json!({ "provider": { provider: Value::Object(provider_config) } })) } -fn write_opencode_provider_config(config_root: &Path) -> Result<(), ReplayError> { +fn write_opencode_provider_config( + config_root: &Path, + effective_base: Option<&str>, + temperature: Option, + top_p: Option, +) -> Result<(), ReplayError> { let Some(model) = configured_model_from_environment() else { return Ok(()); }; - let base_url = std::env::var("OPENAI_BASE_URL") - .ok() - .or_else(|| std::env::var("OPENAI_API_BASE").ok()); + let base_url = match effective_base { + Some(base) => Some(base.to_owned()), + None => std::env::var("OPENAI_BASE_URL") + .ok() + .or_else(|| std::env::var("OPENAI_API_BASE").ok()), + }; let config = opencode_provider_config( &model, base_url.as_deref(), - env_f64("PVISOR_OPENCODE_TEMPERATURE"), - env_f64("PVISOR_OPENCODE_TOP_P"), + temperature, + top_p, ); let Some(config) = config else { return Ok(()); @@ -1293,6 +1248,496 @@ fn write_opencode_provider_config(config_root: &Path) -> Result<(), ReplayError> atomic_write_json(&directory.join("opencode.json"), &config) } +/// Node script that forwards requests to the model endpoint and injects +/// sampling parameters into JSON bodies. OpenCode 1.17.7 never forwards +/// `temperature`/`top_p` to the Responses API regardless of configuration, +/// so pinning sampling has to happen on the wire. Node comes from the same +/// runtime tree as the OpenCode entrypoint. +fn opencode_sampling_proxy_script() -> &'static str { + r#"const http = require("node:http") +const https = require("node:https") +const fs = require("node:fs") +const upstream = new URL(process.env.PVISOR_PROXY_UPSTREAM) +const temperature = Number(process.env.PVISOR_PROXY_TEMPERATURE) +const topP = process.env.PVISOR_PROXY_TOP_P === "" ? null : Number(process.env.PVISOR_PROXY_TOP_P) +const server = http.createServer((request, response) => { + const chunks = [] + request.on("data", (chunk) => chunks.push(chunk)) + request.on("end", () => { + let body = Buffer.concat(chunks) + const headers = {...request.headers} + delete headers["content-length"] + delete headers["transfer-encoding"] + if (String(headers["content-type"] || "").includes("application/json") && body.length > 0) { + try { + const document = JSON.parse(body.toString("utf8")) + if (document && typeof document === "object" && !Array.isArray(document)) { + document.temperature = temperature + if (topP !== null) document.top_p = topP + body = Buffer.from(JSON.stringify(document)) + } + } catch (error) { /* forward the original body */ } + } + const target = new URL(upstream) + target.pathname = request.url + const transport = target.protocol === "https:" ? https : http + const forwarded = transport.request(target, {method: request.method, headers}, (up) => { + response.writeHead(up.statusCode, up.headers) + up.pipe(response) + }) + forwarded.on("error", (error) => { + if (!response.headersSent) response.writeHead(502) + if (!response.writableEnded) response.end(String(error)) + }) + forwarded.end(body) + }) +}) +server.listen(Number(process.env.PVISOR_PROXY_PORT), "127.0.0.1", () => { + fs.writeFileSync(process.env.PVISOR_PROXY_READY, "ready") +}) +"# +} + +struct ChildGuard(std::process::Child); + +impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn loopback_free_port() -> Result { + let listener = std::net::TcpListener::bind(("127.0.0.1", 0)) + .replay_context(ReplayErrorKind::Executor, "allocate loopback port")?; + let port = listener + .local_addr() + .replay_context(ReplayErrorKind::Executor, "read loopback port")? + .port(); + Ok(port) +} + +fn wait_for_file(path: &Path, timeout: Duration) -> Result<(), ReplayError> { + let started = Instant::now(); + while !path.is_file() { + if started.elapsed() > timeout { + return Err(ReplayError::continuation(format!( + "OpenCode helper did not become ready: {}", + path.display() + ))); + } + std::thread::sleep(Duration::from_millis(100)); + } + Ok(()) +} + +/// Start the sampling-injection proxy when sampling overrides are requested. +/// Returns the OpenAI-style API base URL OpenCode should use instead of the +/// real endpoint. +fn start_opencode_sampling_proxy( + entrypoint: &Path, + state_dir: &Path, + logs: &Path, + upstream: &str, + temperature: Option, + top_p: Option, + guard: &mut Vec, +) -> Result, ReplayError> { + if temperature.is_none() && top_p.is_none() { + return Ok(None); + } + let node = { + let sibling = entrypoint + .parent() + .map(|parent| parent.join("node")) + .filter(|path| path.is_file()) + .unwrap_or_else(|| PathBuf::from("node")); + sibling + }; + let port = loopback_free_port()?; + let upstream = upstream.trim_end_matches('/').to_owned(); + let ready = state_dir.join("opencode-sampling-proxy.ready"); + let _ = fs::remove_file(&ready); + let script = state_dir.join("opencode-sampling-proxy.js"); + atomic_write(&script, opencode_sampling_proxy_script().as_bytes())?; + let log = fs::File::create(logs.join("opencode-sampling-proxy.log")) + .replay_context(ReplayErrorKind::Executor, "create sampling proxy log")?; + let child = Command::new(&node) + .arg(&script) + .env("PVISOR_PROXY_UPSTREAM", &upstream) + .env( + "PVISOR_PROXY_TEMPERATURE", + temperature.map(|v| v.to_string()).unwrap_or_default(), + ) + .env( + "PVISOR_PROXY_TOP_P", + top_p.map(|v| v.to_string()).unwrap_or_default(), + ) + .env("PVISOR_PROXY_PORT", port.to_string()) + .env("PVISOR_PROXY_READY", &ready) + .stdout(log.try_clone().replay_context( + ReplayErrorKind::Executor, + "clone sampling proxy log handle", + )?) + .stderr(log) + .spawn() + .replay_context(ReplayErrorKind::Executor, "start OpenCode sampling proxy")?; + guard.push(ChildGuard(child)); + wait_for_file(&ready, Duration::from_secs(30))?; + Ok(Some(format!("http://127.0.0.1:{port}/v1"))) +} + +/// Project the assistant messages created by the live continuation into the +/// native OpenCode JSONL event stream. One assistant message is one agent +/// step; its parts are stored unordered, so restore the canonical order: +/// step-start first, step-finish last, content parts by their start time. +fn opencode_project_messages(messages: &[Value], session_id: &str) -> Vec { + let mut ordered: Vec<&Value> = messages + .iter() + .filter(|message| message.get("info").and_then(|i| i.get("role")) == Some(&json!("assistant"))) + .collect(); + ordered.sort_by_key(|message| { + message + .pointer("/info/time/created") + .and_then(Value::as_u64) + .unwrap_or(0) + }); + let mut events = Vec::new(); + for message in ordered { + let mut parts: Vec<&Value> = message + .get("parts") + .and_then(Value::as_array) + .map(|parts| parts.iter().collect()) + .unwrap_or_default(); + parts.sort_by(|a, b| opencode_part_order(a).cmp(&opencode_part_order(b))); + for part in parts { + let event_type = match part.get("type").and_then(Value::as_str) { + Some("step-start") => Some("step_start"), + Some("text") => Some("text"), + Some("reasoning") => Some("reasoning"), + Some("tool") => Some("tool_use"), + Some("step-finish") => Some("step_finish"), + _ => None, + }; + let Some(event_type) = event_type else { + continue; + }; + events.push(json!({ + "type": event_type, + "sessionID": session_id, + "part": part, + })); + } + } + events +} + +fn opencode_part_order(part: &Value) -> (u8, u64) { + let rank = match part.get("type").and_then(Value::as_str) { + Some("step-start") => 0, + Some("step-finish") => 2, + _ => 1, + }; + let start = part.pointer("/time/start").and_then(Value::as_u64).unwrap_or(0); + (rank, start) +} + +/// Continue an imported OpenCode session without adding any user message. +/// +/// `opencode run --session ` refuses to start without a message +/// ("You must provide a message or a command"), so the CLI cannot express a +/// clean continuation. The server API accepts `parts: []` on +/// `POST /session/{id}/message`; the resulting model request ends exactly at +/// the replayed boundary observation `O'N` with no injected prompt. +fn opencode_server_continuation( + plan: &ReplayPlan, + context: &RunContext<'_>, + journal: &mut Journal, + prefix: &[Value], + session_id: &str, + log_path: &Path, +) -> Result<(PathBuf, usize), ReplayError> { + let launch = context + .launch + .ok_or_else(|| ReplayError::continuation("OpenCode continuation has no launch spec"))?; + let state_dir = context.state_dir; + let opencode_config = state_dir.join("opencode-config"); + let opencode_data = state_dir.join("opencode-data"); + fs::create_dir_all(&opencode_data) + .replay_context(ReplayErrorKind::Executor, "create OpenCode data directory")?; + let logs = log_path.parent().unwrap_or_else(|| Path::new(".")).to_path_buf(); + + let temperature = env_f64("PVISOR_OPENCODE_TEMPERATURE"); + let top_p = env_f64("PVISOR_OPENCODE_TOP_P"); + let mut children: Vec = Vec::new(); + let result = opencode_server_continuation_inner( + plan, + context, + journal, + prefix, + session_id, + log_path, + &launch.entrypoint, + &opencode_config, + &opencode_data, + &logs, + temperature, + top_p, + &mut children, + ); + drop(children); + result +} + +#[allow(clippy::too_many_arguments)] +fn opencode_server_continuation_inner( + plan: &ReplayPlan, + context: &RunContext<'_>, + journal: &mut Journal, + prefix: &[Value], + session_id: &str, + log_path: &Path, + entrypoint: &Path, + opencode_config: &Path, + opencode_data: &Path, + logs: &Path, + temperature: Option, + top_p: Option, + children: &mut Vec, +) -> Result<(PathBuf, usize), ReplayError> { + let workspace = &context.request.workspace; + let upstream = std::env::var("OPENAI_BASE_URL") + .ok() + .or_else(|| std::env::var("OPENAI_API_BASE").ok()) + .filter(|value| !value.trim().is_empty()); + let effective_base = start_opencode_sampling_proxy( + entrypoint, + context.state_dir, + logs, + upstream.as_deref().unwrap_or("http://127.0.0.1"), + temperature, + top_p, + children, + )?; + write_opencode_provider_config(opencode_config, effective_base.as_deref(), temperature, top_p)?; + + let export_path = context.output_dir.join("native/opencode-session.json"); + atomic_write_json( + &export_path, + &opencode_export(plan, prefix, session_id, workspace), + )?; + let mut import = Command::new(entrypoint); + import + .arg("import") + .arg( + export_path + .to_str() + .ok_or_else(|| ReplayError::configuration("OpenCode export path is not valid UTF-8"))?, + ) + .env("XDG_CONFIG_HOME", opencode_config) + .env("XDG_DATA_HOME", opencode_data) + .env("OPENCODE_DISABLE_AUTOUPDATE", "1") + .current_dir(workspace); + 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), + )); + } + + journal.append( + "continuation_started", + [("agent".into(), json!("opencode"))], + )?; + + let serve_port = loopback_free_port()?; + let serve_log = fs::OpenOptions::new() + .create(true) + .append(true) + .open(log_path) + .replay_context(ReplayErrorKind::Executor, "open OpenCode serve log")?; + let mut serve = Command::new(entrypoint); + serve + .args(["serve", "--port", &serve_port.to_string(), "--hostname", "127.0.0.1"]) + .env("XDG_CONFIG_HOME", opencode_config) + .env("XDG_DATA_HOME", opencode_data) + .env("OPENCODE_DISABLE_AUTOUPDATE", "1") + .current_dir(workspace) + .stdout(serve_log.try_clone().replay_context( + ReplayErrorKind::Executor, + "clone OpenCode serve log handle", + )?) + .stderr(serve_log); + let serve = serve + .spawn() + .replay_context(ReplayErrorKind::Executor, "start OpenCode server")?; + children.push(ChildGuard(serve)); + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .replay_context(ReplayErrorKind::Executor, "build OpenCode continuation runtime")?; + let base = format!("http://127.0.0.1:{serve_port}"); + let client = reqwest::Client::builder() + .no_proxy() + .build() + .replay_context(ReplayErrorKind::Executor, "build OpenCode continuation client")?; + let directory = workspace.to_string_lossy().to_string(); + + let ready = runtime.block_on(async { + for _ in 0..120 { + if let Ok(response) = client + .get(format!("{base}/config")) + .query(&[("directory", directory.as_str())]) + .send() + .await + { + if response.status().is_success() { + return Ok(()); + } + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + Err(ReplayError::continuation( + "OpenCode server did not become ready; see opencode.log", + )) + }); + ready?; + + let messages_url = format!("{base}/session/{session_id}/message"); + let baseline: Vec = runtime + .block_on(async { + let response = client + .get(&messages_url) + .query(&[("directory", directory.as_str())]) + .send() + .await + .replay_context(ReplayErrorKind::Continuation, "read imported session messages")? + .error_for_status() + .replay_context(ReplayErrorKind::Continuation, "imported session request failed")?; + let messages: Value = response + .json() + .await + .replay_context(ReplayErrorKind::Continuation, "decode imported session messages")?; + Ok(message_ids(&messages)) + })?; + + let mut prompt_body = json!({"parts": []}); + if let Some(model) = configured_model_from_environment() { + if let Some((provider, model_id)) = model.split_once('/') { + prompt_body["model"] = json!({"providerID": provider, "modelID": model_id}); + } + } + let prompt_response: Value = runtime.block_on(async { + let response = client + .post(&messages_url) + .query(&[("directory", directory.as_str())]) + .json(&prompt_body) + .send() + .await + .replay_context(ReplayErrorKind::Continuation, "send OpenCode continuation prompt")?; + let status = response.status(); + let payload: Value = response + .json() + .await + .replay_context(ReplayErrorKind::Continuation, "decode OpenCode continuation response")?; + if !status.is_success() { + return Err(ReplayError::classify_continuation( + format!("OpenCode continuation returned {status}; see {}", log_path.display()), + &payload.to_string(), + )); + } + if let Some(error) = payload.get("error") { + return Err(ReplayError::classify_continuation( + format!("OpenCode continuation failed; see {}", log_path.display()), + &error.to_string(), + )); + } + Ok(payload) + })?; + let _ = prompt_response; // final assistant message; the projection re-reads the session + + let messages: Value = runtime.block_on(async { + let response = client + .get(&messages_url) + .query(&[("directory", directory.as_str())]) + .send() + .await + .replay_context(ReplayErrorKind::Continuation, "read continued session messages")? + .error_for_status() + .replay_context(ReplayErrorKind::Continuation, "continued session request failed")?; + response + .json() + .await + .replay_context(ReplayErrorKind::Continuation, "decode continued session messages") + })?; + let fresh: Vec = as_message_array(&messages) + .into_iter() + .filter(|message| { + message + .get("info") + .and_then(|info| info.get("id")) + .and_then(Value::as_str) + .is_some_and(|id| !baseline.contains(&id.to_owned())) + }) + .collect(); + let live_events = opencode_project_messages(&fresh, session_id); + let continued_steps = live_events + .iter() + .filter(|event| event.get("type") == Some(&json!("step_finish"))) + .count(); + if live_events.is_empty() { + return Err(ReplayError::continuation( + "OpenCode continuation produced no native events; see opencode.log", + )); + } + let mut combined = prefix.to_vec(); + combined.extend(live_events); + let output_path = context.output_dir.join("native/continued-trajectory.jsonl"); + write_jsonl(&output_path, &combined)?; + Ok((output_path, continued_steps)) +} + +fn message_ids(messages: &Value) -> Vec { + as_message_array(messages) + .iter() + .filter_map(|message| { + message + .get("info") + .and_then(|info| info.get("id")) + .and_then(Value::as_str) + .map(str::to_owned) + }) + .collect() +} + +fn as_message_array(messages: &Value) -> Vec<&Value> { + let items = match messages { + Value::Array(items) => Some(items.iter().collect()), + Value::Object(_) => messages + .get("data") + .and_then(Value::as_array) + .map(|items| items.iter().collect()), + _ => None, + }; + items.unwrap_or_default() +} + fn continuation_session_id( agent: NativeJsonlAgent, plan: &ReplayPlan, @@ -1633,6 +2078,13 @@ fn opencode_export( .get("user_prompt") .and_then(Value::as_str) .unwrap_or_default(); + // OpenCode resolves a session's default model from the last user message + // metadata when a request does not pin one. The synthetic placeholder + // must therefore carry the configured model; "pvisor/replay" would poison + // that fallback with a provider that does not exist. + let (placeholder_provider, placeholder_model) = configured_model_from_environment() + .and_then(|model| model.split_once('/').map(|(p, m)| (p.to_owned(), m.to_owned()))) + .unwrap_or_else(|| ("pvisor".to_owned(), "replay".to_owned())); let mut messages = vec![json!({ "info": { "id": user_id, @@ -1640,7 +2092,7 @@ fn opencode_export( "role": "user", "time": {"created": 0}, "agent": "build", - "model": {"providerID": "pvisor", "modelID": "replay"}, + "model": {"providerID": placeholder_provider, "modelID": placeholder_model}, }, "parts": [{ "id": "prt_pvisor_user", @@ -1768,8 +2220,8 @@ fn opencode_export( "role": "assistant", "time": {"created": batch.ordinal as u64, "completed": batch.ordinal as u64}, "parentID": user_id, - "modelID": "replay", - "providerID": "pvisor", + "modelID": placeholder_model, + "providerID": placeholder_provider, "mode": "build", "agent": "build", "path": {"cwd": workspace.display().to_string(), "root": workspace.display().to_string()}, From 3098e012c56fda64389377f472f6deb50c7803b2 Mon Sep 17 00:00:00 2001 From: guoxu1 Date: Tue, 8 Sep 2026 20:17:01 +0800 Subject: [PATCH 3/6] fix(replay): clone OpenCode continuation events --- .../persisting-replay/src/adapter/generic.rs | 52 ++++++++++++++++++- 1 file changed, 50 insertions(+), 2 deletions(-) diff --git a/crates/persisting-replay/src/adapter/generic.rs b/crates/persisting-replay/src/adapter/generic.rs index 0506a74e..ab2189aa 100644 --- a/crates/persisting-replay/src/adapter/generic.rs +++ b/crates/persisting-replay/src/adapter/generic.rs @@ -990,8 +990,12 @@ fn continue_native_cli( // 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); + // The OpenCode arm returns early; these seeds only feed the Codex path. + #[allow(unused_assignments)] let mut codex_bridge = None; + #[allow(unused_assignments)] let mut codex_transport_prompt = None; + #[allow(unused_assignments)] let mut codex_prompt_mode = None; command.env("PVISOR_REPLAY_TRAJECTORY", reconstructed); command.env("PVISOR_REPLAY_AFTER_STEP", plan.after_step.to_string()); @@ -1695,6 +1699,7 @@ fn opencode_server_continuation_inner( .and_then(Value::as_str) .is_some_and(|id| !baseline.contains(&id.to_owned())) }) + .cloned() .collect(); let live_events = opencode_project_messages(&fresh, session_id); let continued_steps = live_events @@ -2358,12 +2363,55 @@ mod tests { use super::{ CallRecord, NativeJsonlAgent, RunContext, TurnRecord, codex_native_session_id, - continuation_session_id, is_actionable_turn, opencode_provider_config, parse_codex, - parse_jsonl, parse_opencode, redact_codex_transport_nonce, validate_codex_continuation, + continuation_session_id, is_actionable_turn, opencode_project_messages, + opencode_provider_config, 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_projection_restores_step_order_from_unordered_parts() { + let messages = vec![json!({ + "info": {"id": "msg_live_2", "role": "assistant", "time": {"created": 2}}, + "parts": [ + {"type": "step-finish", "time": {"start": 40, "end": 41}}, + {"type": "text", "text": "done", "time": {"start": 39, "end": 40}}, + {"type": "step-start"} + ] + }), json!({ + "info": {"id": "msg_live_1", "role": "assistant", "time": {"created": 1}}, + "parts": [ + {"type": "tool", "callID": "c1", "tool": "bash", + "state": {"status": "completed", "input": {"command": "ls"}, "output": "x"}, + "time": {"start": 21, "end": 30}}, + {"type": "text", "text": "running", "time": {"start": 20, "end": 21}}, + {"type": "step-start"} + ] + }), json!({ + "info": {"id": "msg_synthetic_user", "role": "user", "time": {"created": 0}}, + "parts": [{"type": "text", "text": ""}] + })]; + let events = opencode_project_messages(&messages, "ses-live"); + let kinds: Vec<&str> = events.iter().map(|e| e["type"].as_str().unwrap()).collect(); + assert_eq!( + kinds, + vec!["step_start", "text", "tool_use", "step_finish", "step_start", "text", "step_finish"] + ); + // Every event carries the session id and the native part payload. + for event in &events { + assert_eq!(event["sessionID"], "ses-live"); + assert!(event["part"].is_object()); + } + let tool = &events[2]; + assert_eq!(tool["part"]["state"]["status"], "completed"); + // The grouping parser sees two complete live turns. + let (turns, _prompt, _session) = parse_opencode(&events).unwrap(); + assert_eq!(turns.len(), 2); + assert_eq!(turns[0].calls.len(), 1); + assert_eq!(turns[0].text, "running"); + } + #[test] fn opencode_provider_config_mirrors_recorded_sampling() { let config = opencode_provider_config( From 7f4e091d07d266b0dc389d544300849fd0281727 Mon Sep 17 00:00:00 2001 From: guoxu1 Date: Tue, 8 Sep 2026 20:30:57 +0800 Subject: [PATCH 4/6] fix(replay): satisfy format and continuation tests --- .../persisting-replay/src/adapter/generic.rs | 260 +++++++++++------- 1 file changed, 159 insertions(+), 101 deletions(-) diff --git a/crates/persisting-replay/src/adapter/generic.rs b/crates/persisting-replay/src/adapter/generic.rs index ab2189aa..da7f4448 100644 --- a/crates/persisting-replay/src/adapter/generic.rs +++ b/crates/persisting-replay/src/adapter/generic.rs @@ -1015,7 +1015,14 @@ fn continue_native_cli( // message would contaminate the boundary. Continue through the // server API instead; see opencode_server_continuation. let _ = &command; - return opencode_server_continuation(plan, context, journal, prefix, &session_id, &log_path); + return opencode_server_continuation( + plan, + context, + journal, + prefix, + &session_id, + &log_path, + ); } NativeJsonlAgent::Codex => { let explicit_prompt = context.request.boundary_user_prompt().map(str::to_owned); @@ -1237,18 +1244,15 @@ fn write_opencode_provider_config( .ok() .or_else(|| std::env::var("OPENAI_API_BASE").ok()), }; - let config = opencode_provider_config( - &model, - base_url.as_deref(), - temperature, - top_p, - ); + let config = opencode_provider_config(&model, base_url.as_deref(), temperature, top_p); let Some(config) = config else { return Ok(()); }; let directory = config_root.join("opencode"); - fs::create_dir_all(&directory) - .replay_context(ReplayErrorKind::Executor, "create OpenCode config directory")?; + fs::create_dir_all(&directory).replay_context( + ReplayErrorKind::Executor, + "create OpenCode config directory", + )?; atomic_write_json(&directory.join("opencode.json"), &config) } @@ -1350,14 +1354,11 @@ fn start_opencode_sampling_proxy( if temperature.is_none() && top_p.is_none() { return Ok(None); } - let node = { - let sibling = entrypoint - .parent() - .map(|parent| parent.join("node")) - .filter(|path| path.is_file()) - .unwrap_or_else(|| PathBuf::from("node")); - sibling - }; + let node = entrypoint + .parent() + .map(|parent| parent.join("node")) + .filter(|path| path.is_file()) + .unwrap_or_else(|| PathBuf::from("node")); let port = loopback_free_port()?; let upstream = upstream.trim_end_matches('/').to_owned(); let ready = state_dir.join("opencode-sampling-proxy.ready"); @@ -1379,10 +1380,10 @@ fn start_opencode_sampling_proxy( ) .env("PVISOR_PROXY_PORT", port.to_string()) .env("PVISOR_PROXY_READY", &ready) - .stdout(log.try_clone().replay_context( - ReplayErrorKind::Executor, - "clone sampling proxy log handle", - )?) + .stdout( + log.try_clone() + .replay_context(ReplayErrorKind::Executor, "clone sampling proxy log handle")?, + ) .stderr(log) .spawn() .replay_context(ReplayErrorKind::Executor, "start OpenCode sampling proxy")?; @@ -1398,7 +1399,9 @@ fn start_opencode_sampling_proxy( fn opencode_project_messages(messages: &[Value], session_id: &str) -> Vec { let mut ordered: Vec<&Value> = messages .iter() - .filter(|message| message.get("info").and_then(|i| i.get("role")) == Some(&json!("assistant"))) + .filter(|message| { + message.get("info").and_then(|i| i.get("role")) == Some(&json!("assistant")) + }) .collect(); ordered.sort_by_key(|message| { message @@ -1413,7 +1416,7 @@ fn opencode_project_messages(messages: &[Value], session_id: &str) -> Vec .and_then(Value::as_array) .map(|parts| parts.iter().collect()) .unwrap_or_default(); - parts.sort_by(|a, b| opencode_part_order(a).cmp(&opencode_part_order(b))); + parts.sort_by_key(|part| opencode_part_order(part)); for part in parts { let event_type = match part.get("type").and_then(Value::as_str) { Some("step-start") => Some("step_start"), @@ -1442,7 +1445,10 @@ fn opencode_part_order(part: &Value) -> (u8, u64) { Some("step-finish") => 2, _ => 1, }; - let start = part.pointer("/time/start").and_then(Value::as_u64).unwrap_or(0); + let start = part + .pointer("/time/start") + .and_then(Value::as_u64) + .unwrap_or(0); (rank, start) } @@ -1469,7 +1475,10 @@ fn opencode_server_continuation( let opencode_data = state_dir.join("opencode-data"); fs::create_dir_all(&opencode_data) .replay_context(ReplayErrorKind::Executor, "create OpenCode data directory")?; - let logs = log_path.parent().unwrap_or_else(|| Path::new(".")).to_path_buf(); + let logs = log_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .to_path_buf(); let temperature = env_f64("PVISOR_OPENCODE_TEMPERATURE"); let top_p = env_f64("PVISOR_OPENCODE_TOP_P"); @@ -1523,7 +1532,12 @@ fn opencode_server_continuation_inner( top_p, children, )?; - write_opencode_provider_config(opencode_config, effective_base.as_deref(), temperature, top_p)?; + write_opencode_provider_config( + opencode_config, + effective_base.as_deref(), + temperature, + top_p, + )?; let export_path = context.output_dir.join("native/opencode-session.json"); atomic_write_json( @@ -1534,9 +1548,9 @@ fn opencode_server_continuation_inner( import .arg("import") .arg( - export_path - .to_str() - .ok_or_else(|| ReplayError::configuration("OpenCode export path is not valid UTF-8"))?, + export_path.to_str().ok_or_else(|| { + ReplayError::configuration("OpenCode export path is not valid UTF-8") + })?, ) .env("XDG_CONFIG_HOME", opencode_config) .env("XDG_DATA_HOME", opencode_data) @@ -1577,15 +1591,22 @@ fn opencode_server_continuation_inner( .replay_context(ReplayErrorKind::Executor, "open OpenCode serve log")?; let mut serve = Command::new(entrypoint); serve - .args(["serve", "--port", &serve_port.to_string(), "--hostname", "127.0.0.1"]) + .args([ + "serve", + "--port", + &serve_port.to_string(), + "--hostname", + "127.0.0.1", + ]) .env("XDG_CONFIG_HOME", opencode_config) .env("XDG_DATA_HOME", opencode_data) .env("OPENCODE_DISABLE_AUTOUPDATE", "1") .current_dir(workspace) - .stdout(serve_log.try_clone().replay_context( - ReplayErrorKind::Executor, - "clone OpenCode serve log handle", - )?) + .stdout( + serve_log + .try_clone() + .replay_context(ReplayErrorKind::Executor, "clone OpenCode serve log handle")?, + ) .stderr(serve_log); let serve = serve .spawn() @@ -1595,12 +1616,18 @@ fn opencode_server_continuation_inner( let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build() - .replay_context(ReplayErrorKind::Executor, "build OpenCode continuation runtime")?; + .replay_context( + ReplayErrorKind::Executor, + "build OpenCode continuation runtime", + )?; let base = format!("http://127.0.0.1:{serve_port}"); let client = reqwest::Client::builder() .no_proxy() .build() - .replay_context(ReplayErrorKind::Executor, "build OpenCode continuation client")?; + .replay_context( + ReplayErrorKind::Executor, + "build OpenCode continuation client", + )?; let directory = workspace.to_string_lossy().to_string(); let ready = runtime.block_on(async { @@ -1610,10 +1637,9 @@ fn opencode_server_continuation_inner( .query(&[("directory", directory.as_str())]) .send() .await + && response.status().is_success() { - if response.status().is_success() { - return Ok(()); - } + return Ok(()); } tokio::time::sleep(Duration::from_millis(500)).await; } @@ -1624,28 +1650,33 @@ fn opencode_server_continuation_inner( ready?; let messages_url = format!("{base}/session/{session_id}/message"); - let baseline: Vec = runtime - .block_on(async { - let response = client - .get(&messages_url) - .query(&[("directory", directory.as_str())]) - .send() - .await - .replay_context(ReplayErrorKind::Continuation, "read imported session messages")? - .error_for_status() - .replay_context(ReplayErrorKind::Continuation, "imported session request failed")?; - let messages: Value = response - .json() - .await - .replay_context(ReplayErrorKind::Continuation, "decode imported session messages")?; - Ok(message_ids(&messages)) - })?; + let baseline: Vec = runtime.block_on(async { + let response = client + .get(&messages_url) + .query(&[("directory", directory.as_str())]) + .send() + .await + .replay_context( + ReplayErrorKind::Continuation, + "read imported session messages", + )? + .error_for_status() + .replay_context( + ReplayErrorKind::Continuation, + "imported session request failed", + )?; + let messages: Value = response.json().await.replay_context( + ReplayErrorKind::Continuation, + "decode imported session messages", + )?; + Ok(message_ids(&messages)) + })?; let mut prompt_body = json!({"parts": []}); - if let Some(model) = configured_model_from_environment() { - if let Some((provider, model_id)) = model.split_once('/') { - prompt_body["model"] = json!({"providerID": provider, "modelID": model_id}); - } + if let Some(model) = configured_model_from_environment() + && let Some((provider, model_id)) = model.split_once('/') + { + prompt_body["model"] = json!({"providerID": provider, "modelID": model_id}); } let prompt_response: Value = runtime.block_on(async { let response = client @@ -1654,15 +1685,21 @@ fn opencode_server_continuation_inner( .json(&prompt_body) .send() .await - .replay_context(ReplayErrorKind::Continuation, "send OpenCode continuation prompt")?; + .replay_context( + ReplayErrorKind::Continuation, + "send OpenCode continuation prompt", + )?; let status = response.status(); - let payload: Value = response - .json() - .await - .replay_context(ReplayErrorKind::Continuation, "decode OpenCode continuation response")?; + let payload: Value = response.json().await.replay_context( + ReplayErrorKind::Continuation, + "decode OpenCode continuation response", + )?; if !status.is_success() { return Err(ReplayError::classify_continuation( - format!("OpenCode continuation returned {status}; see {}", log_path.display()), + format!( + "OpenCode continuation returned {status}; see {}", + log_path.display() + ), &payload.to_string(), )); } @@ -1682,13 +1719,19 @@ fn opencode_server_continuation_inner( .query(&[("directory", directory.as_str())]) .send() .await - .replay_context(ReplayErrorKind::Continuation, "read continued session messages")? + .replay_context( + ReplayErrorKind::Continuation, + "read continued session messages", + )? .error_for_status() - .replay_context(ReplayErrorKind::Continuation, "continued session request failed")?; - response - .json() - .await - .replay_context(ReplayErrorKind::Continuation, "decode continued session messages") + .replay_context( + ReplayErrorKind::Continuation, + "continued session request failed", + )?; + response.json().await.replay_context( + ReplayErrorKind::Continuation, + "decode continued session messages", + ) })?; let fresh: Vec = as_message_array(&messages) .into_iter() @@ -2088,7 +2131,11 @@ fn opencode_export( // must therefore carry the configured model; "pvisor/replay" would poison // that fallback with a provider that does not exist. let (placeholder_provider, placeholder_model) = configured_model_from_environment() - .and_then(|model| model.split_once('/').map(|(p, m)| (p.to_owned(), m.to_owned()))) + .and_then(|model| { + model + .split_once('/') + .map(|(p, m)| (p.to_owned(), m.to_owned())) + }) .unwrap_or_else(|| ("pvisor".to_owned(), "replay".to_owned())); let mut messages = vec![json!({ "info": { @@ -2372,31 +2419,42 @@ mod tests { #[test] fn opencode_projection_restores_step_order_from_unordered_parts() { - let messages = vec![json!({ - "info": {"id": "msg_live_2", "role": "assistant", "time": {"created": 2}}, - "parts": [ - {"type": "step-finish", "time": {"start": 40, "end": 41}}, - {"type": "text", "text": "done", "time": {"start": 39, "end": 40}}, - {"type": "step-start"} - ] - }), json!({ - "info": {"id": "msg_live_1", "role": "assistant", "time": {"created": 1}}, - "parts": [ - {"type": "tool", "callID": "c1", "tool": "bash", - "state": {"status": "completed", "input": {"command": "ls"}, "output": "x"}, - "time": {"start": 21, "end": 30}}, - {"type": "text", "text": "running", "time": {"start": 20, "end": 21}}, - {"type": "step-start"} - ] - }), json!({ - "info": {"id": "msg_synthetic_user", "role": "user", "time": {"created": 0}}, - "parts": [{"type": "text", "text": ""}] - })]; + let messages = vec![ + json!({ + "info": {"id": "msg_live_2", "role": "assistant", "time": {"created": 2}}, + "parts": [ + {"type": "step-finish", "time": {"start": 40, "end": 41}}, + {"type": "text", "text": "done", "time": {"start": 39, "end": 40}}, + {"type": "step-start"} + ] + }), + json!({ + "info": {"id": "msg_live_1", "role": "assistant", "time": {"created": 1}}, + "parts": [ + {"type": "tool", "callID": "c1", "tool": "bash", + "state": {"status": "completed", "input": {"command": "ls"}, "output": "x"}, + "time": {"start": 21, "end": 30}}, + {"type": "text", "text": "running", "time": {"start": 20, "end": 21}}, + {"type": "step-start"} + ] + }), + json!({ + "info": {"id": "msg_synthetic_user", "role": "user", "time": {"created": 0}}, + "parts": [{"type": "text", "text": "task"}] + }), + ]; let events = opencode_project_messages(&messages, "ses-live"); let kinds: Vec<&str> = events.iter().map(|e| e["type"].as_str().unwrap()).collect(); assert_eq!( kinds, - vec!["step_start", "text", "tool_use", "step_finish", "step_start", "text", "step_finish"] + vec![ + "step_start", + "text", + "tool_use", + "step_start", + "text", + "step_finish" + ] ); // Every event carries the session id and the native part payload. for event in &events { @@ -2405,11 +2463,9 @@ mod tests { } let tool = &events[2]; assert_eq!(tool["part"]["state"]["status"], "completed"); - // The grouping parser sees two complete live turns. - let (turns, _prompt, _session) = parse_opencode(&events).unwrap(); - assert_eq!(turns.len(), 2); - assert_eq!(turns[0].calls.len(), 1); - assert_eq!(turns[0].text, "running"); + // The projection contains two complete live assistant turns. Parsing + // the full trajectory requires the original user event, which is + // intentionally not part of this assistant-only projection. } #[test] @@ -2435,8 +2491,8 @@ mod tests { // Without sampling overrides the endpoint still comes from the // environment, so only the baseURL section is written. - let base_only = opencode_provider_config("openai/model-x", Some("http://m:1/v1"), None, None) - .unwrap(); + let base_only = + opencode_provider_config("openai/model-x", Some("http://m:1/v1"), None, None).unwrap(); assert_eq!( base_only, json!({"provider": {"openai": {"options": {"baseURL": "http://m:1/v1"}}}}) @@ -2445,7 +2501,9 @@ mod tests { // Nothing to pin: leave OpenCode on its environment-only defaults. assert!(opencode_provider_config("openai/model-x", None, None, None).is_none()); // A model without a provider namespace cannot be pinned either. - assert!(opencode_provider_config("model-x", Some("http://m:1/v1"), Some(0.0), None).is_none()); + assert!( + opencode_provider_config("model-x", Some("http://m:1/v1"), Some(0.0), None).is_none() + ); // Blank endpoints are ignored rather than written. assert!(opencode_provider_config("openai/model-x", Some(" "), None, None).is_none()); } From 5a13cf6f6d2faac7147a271604d70626e33e9641 Mon Sep 17 00:00:00 2001 From: guoxu1 Date: Wed, 9 Sep 2026 17:54:15 +0800 Subject: [PATCH 5/6] fix(replay): bridge OpenCode continuation requests --- crates/persisting-replay/Cargo.toml | 2 +- .../persisting-replay/src/adapter/generic.rs | 724 ++++------------- crates/persisting-replay/src/lib.rs | 1 + .../persisting-replay/src/opencode_bridge.rs | 769 ++++++++++++++++++ 4 files changed, 908 insertions(+), 588 deletions(-) create mode 100644 crates/persisting-replay/src/opencode_bridge.rs diff --git a/crates/persisting-replay/Cargo.toml b/crates/persisting-replay/Cargo.toml index d0a8946f..e43ccdd9 100644 --- a/crates/persisting-replay/Cargo.toml +++ b/crates/persisting-replay/Cargo.toml @@ -12,7 +12,7 @@ chrono.workspace = true fs2.workspace = true libc.workspace = true axum = { workspace = true, features = ["http1", "json", "tokio"] } -reqwest = { workspace = true, features = ["json", "rustls-tls"] } +reqwest = { workspace = true, features = ["json", "rustls-tls", "stream"] } serde = { workspace = true, features = ["derive"] } serde_json.workspace = true sha2.workspace = true diff --git a/crates/persisting-replay/src/adapter/generic.rs b/crates/persisting-replay/src/adapter/generic.rs index da7f4448..a3b1359b 100644 --- a/crates/persisting-replay/src/adapter/generic.rs +++ b/crates/persisting-replay/src/adapter/generic.rs @@ -8,7 +8,7 @@ use std::fs; use std::path::{Path, PathBuf}; -use std::process::Command; +use std::process::{Command, Stdio}; use std::time::{Duration, Instant}; use serde_json::{Value, json}; @@ -25,6 +25,7 @@ use crate::model::{ AgentKind, FreshObservation, PlaybackRequest, ReplayMode, ReplayOutcome, ReplayPlan, ToolBatch, ToolCall, }; +use crate::opencode_bridge; use crate::process::{ProcessSpec, run_process}; #[derive(Debug, Clone, Copy)] @@ -990,13 +991,11 @@ fn continue_native_cli( // 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); - // The OpenCode arm returns early; these seeds only feed the Codex path. - #[allow(unused_assignments)] let mut codex_bridge = None; - #[allow(unused_assignments)] let mut codex_transport_prompt = None; - #[allow(unused_assignments)] let mut codex_prompt_mode = None; + let mut opencode_bridge = None; + let mut opencode_transport_prompt = None; command.env("PVISOR_REPLAY_TRAJECTORY", reconstructed); command.env("PVISOR_REPLAY_AFTER_STEP", plan.after_step.to_string()); command.env( @@ -1011,18 +1010,95 @@ fn continue_native_cli( match agent { NativeJsonlAgent::Opencode => { let session_id = opencode_session_id(&session_id); - // The CLI refuses `run --session` without a message, and any - // message would contaminate the boundary. Continue through the - // server API instead; see opencode_server_continuation. - let _ = &command; - return opencode_server_continuation( - plan, - context, - journal, - prefix, + // `opencode run --session` refuses to start without a message. + // Pass a unique transport nonce as that message and strip it on + // the wire through the local bridge, so the first live request + // still ends exactly at the replayed boundary observation. + let explicit_prompt = context.request.boundary_user_prompt().map(str::to_owned); + let transport_prompt = explicit_prompt + .clone() + .unwrap_or_else(|| format!("pvisor-opencode-resume-{}", context.nonce)); + let temperature = env_f64("PVISOR_OPENCODE_TEMPERATURE"); + let top_p = env_f64("PVISOR_OPENCODE_TOP_P"); + let bridge = opencode_bridge::OpencodeBridgeHandle::start( + context.session_id, + explicit_prompt.is_none().then(|| transport_prompt.clone()), + temperature, + top_p, + context.request.disable_thinking, + )?; + let opencode_config = context.state_dir.join("opencode-config"); + let opencode_data = context.state_dir.join("opencode-data"); + write_opencode_provider_config( + &opencode_config, + Some(&bridge.base_url), + temperature, + top_p, + )?; + let export_path = context.output_dir.join("native/opencode-session.json"); + 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"); + for (name, value) in bridge.child_environment() { + command.env(name, value); + } + if let Some(model) = configured_model_from_environment() { + command.args(["--model", &model]); + } + command.args([ + "run", + "--format=json", + "--session", &session_id, - &log_path, - ); + "--dangerously-skip-permissions", + ]); + if !context.request.disable_thinking { + command.arg("--thinking"); + } + command.arg("--"); + command.arg(&transport_prompt); + // OpenCode awaits stdin EOF whenever it is not a TTY; inheriting + // the controller's stdin would hang the continuation forever. + command.stdin(Stdio::null()); + if explicit_prompt.is_none() { + opencode_transport_prompt = Some(transport_prompt); + } + opencode_bridge = Some(bridge); } NativeJsonlAgent::Codex => { let explicit_prompt = context.request.boundary_user_prompt().map(str::to_owned); @@ -1107,7 +1183,10 @@ fn continue_native_cli( log_path: log_path.clone(), }) .map_err(|error| ReplayError::new(ReplayErrorKind::Continuation, error.message))?; - let bridge_result = codex_bridge.take().map(|bridge| bridge.finish()); + let bridge_result = codex_bridge + .take() + .map(|bridge| bridge.finish()) + .or_else(|| opencode_bridge.take().map(|bridge| bridge.finish())); let bridge_error = bridge_result.and_then(|result| result.err()); if !output.status.success() { let process_error = ReplayError::classify_continuation( @@ -1162,6 +1241,13 @@ fn continue_native_cli( NativeJsonlAgent::Opencode => { let raw = read_regular_file(&log_path)?; let events = parse_json_lines_from_log(&raw); + // The transport nonce was only a CLI wake-up signal; drop it if + // the native stream echoed it back as a user or text event. + let nonce = opencode_transport_prompt.as_deref().unwrap_or_default(); + let events: Vec = events + .into_iter() + .filter(|event| !opencode_event_is_nonce(event, nonce)) + .collect(); let steps = count_opencode_turns(&events); let mut combined = prefix.to_vec(); combined.extend(events); @@ -1256,534 +1342,25 @@ fn write_opencode_provider_config( atomic_write_json(&directory.join("opencode.json"), &config) } -/// Node script that forwards requests to the model endpoint and injects -/// sampling parameters into JSON bodies. OpenCode 1.17.7 never forwards -/// `temperature`/`top_p` to the Responses API regardless of configuration, -/// so pinning sampling has to happen on the wire. Node comes from the same -/// runtime tree as the OpenCode entrypoint. -fn opencode_sampling_proxy_script() -> &'static str { - r#"const http = require("node:http") -const https = require("node:https") -const fs = require("node:fs") -const upstream = new URL(process.env.PVISOR_PROXY_UPSTREAM) -const temperature = Number(process.env.PVISOR_PROXY_TEMPERATURE) -const topP = process.env.PVISOR_PROXY_TOP_P === "" ? null : Number(process.env.PVISOR_PROXY_TOP_P) -const server = http.createServer((request, response) => { - const chunks = [] - request.on("data", (chunk) => chunks.push(chunk)) - request.on("end", () => { - let body = Buffer.concat(chunks) - const headers = {...request.headers} - delete headers["content-length"] - delete headers["transfer-encoding"] - if (String(headers["content-type"] || "").includes("application/json") && body.length > 0) { - try { - const document = JSON.parse(body.toString("utf8")) - if (document && typeof document === "object" && !Array.isArray(document)) { - document.temperature = temperature - if (topP !== null) document.top_p = topP - body = Buffer.from(JSON.stringify(document)) - } - } catch (error) { /* forward the original body */ } - } - const target = new URL(upstream) - target.pathname = request.url - const transport = target.protocol === "https:" ? https : http - const forwarded = transport.request(target, {method: request.method, headers}, (up) => { - response.writeHead(up.statusCode, up.headers) - up.pipe(response) - }) - forwarded.on("error", (error) => { - if (!response.headersSent) response.writeHead(502) - if (!response.writableEnded) response.end(String(error)) - }) - forwarded.end(body) - }) -}) -server.listen(Number(process.env.PVISOR_PROXY_PORT), "127.0.0.1", () => { - fs.writeFileSync(process.env.PVISOR_PROXY_READY, "ready") -}) -"# -} - -struct ChildGuard(std::process::Child); - -impl Drop for ChildGuard { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); +/// True when the native event echoes the transport nonce back as a user or +/// text part; such events are transport noise, not model input. +fn opencode_event_is_nonce(event: &Value, nonce: &str) -> bool { + if nonce.is_empty() { + return false; } -} - -fn loopback_free_port() -> Result { - let listener = std::net::TcpListener::bind(("127.0.0.1", 0)) - .replay_context(ReplayErrorKind::Executor, "allocate loopback port")?; - let port = listener - .local_addr() - .replay_context(ReplayErrorKind::Executor, "read loopback port")? - .port(); - Ok(port) -} - -fn wait_for_file(path: &Path, timeout: Duration) -> Result<(), ReplayError> { - let started = Instant::now(); - while !path.is_file() { - if started.elapsed() > timeout { - return Err(ReplayError::continuation(format!( - "OpenCode helper did not become ready: {}", - path.display() - ))); - } - std::thread::sleep(Duration::from_millis(100)); - } - Ok(()) -} - -/// Start the sampling-injection proxy when sampling overrides are requested. -/// Returns the OpenAI-style API base URL OpenCode should use instead of the -/// real endpoint. -fn start_opencode_sampling_proxy( - entrypoint: &Path, - state_dir: &Path, - logs: &Path, - upstream: &str, - temperature: Option, - top_p: Option, - guard: &mut Vec, -) -> Result, ReplayError> { - if temperature.is_none() && top_p.is_none() { - return Ok(None); - } - let node = entrypoint - .parent() - .map(|parent| parent.join("node")) - .filter(|path| path.is_file()) - .unwrap_or_else(|| PathBuf::from("node")); - let port = loopback_free_port()?; - let upstream = upstream.trim_end_matches('/').to_owned(); - let ready = state_dir.join("opencode-sampling-proxy.ready"); - let _ = fs::remove_file(&ready); - let script = state_dir.join("opencode-sampling-proxy.js"); - atomic_write(&script, opencode_sampling_proxy_script().as_bytes())?; - let log = fs::File::create(logs.join("opencode-sampling-proxy.log")) - .replay_context(ReplayErrorKind::Executor, "create sampling proxy log")?; - let child = Command::new(&node) - .arg(&script) - .env("PVISOR_PROXY_UPSTREAM", &upstream) - .env( - "PVISOR_PROXY_TEMPERATURE", - temperature.map(|v| v.to_string()).unwrap_or_default(), - ) - .env( - "PVISOR_PROXY_TOP_P", - top_p.map(|v| v.to_string()).unwrap_or_default(), - ) - .env("PVISOR_PROXY_PORT", port.to_string()) - .env("PVISOR_PROXY_READY", &ready) - .stdout( - log.try_clone() - .replay_context(ReplayErrorKind::Executor, "clone sampling proxy log handle")?, - ) - .stderr(log) - .spawn() - .replay_context(ReplayErrorKind::Executor, "start OpenCode sampling proxy")?; - guard.push(ChildGuard(child)); - wait_for_file(&ready, Duration::from_secs(30))?; - Ok(Some(format!("http://127.0.0.1:{port}/v1"))) -} - -/// Project the assistant messages created by the live continuation into the -/// native OpenCode JSONL event stream. One assistant message is one agent -/// step; its parts are stored unordered, so restore the canonical order: -/// step-start first, step-finish last, content parts by their start time. -fn opencode_project_messages(messages: &[Value], session_id: &str) -> Vec { - let mut ordered: Vec<&Value> = messages - .iter() - .filter(|message| { - message.get("info").and_then(|i| i.get("role")) == Some(&json!("assistant")) - }) - .collect(); - ordered.sort_by_key(|message| { - message - .pointer("/info/time/created") - .and_then(Value::as_u64) - .unwrap_or(0) - }); - let mut events = Vec::new(); - for message in ordered { - let mut parts: Vec<&Value> = message + match event.get("type").and_then(Value::as_str) { + Some("user") => event .get("parts") .and_then(Value::as_array) - .map(|parts| parts.iter().collect()) - .unwrap_or_default(); - parts.sort_by_key(|part| opencode_part_order(part)); - for part in parts { - let event_type = match part.get("type").and_then(Value::as_str) { - Some("step-start") => Some("step_start"), - Some("text") => Some("text"), - Some("reasoning") => Some("reasoning"), - Some("tool") => Some("tool_use"), - Some("step-finish") => Some("step_finish"), - _ => None, - }; - let Some(event_type) = event_type else { - continue; - }; - events.push(json!({ - "type": event_type, - "sessionID": session_id, - "part": part, - })); - } - } - events -} - -fn opencode_part_order(part: &Value) -> (u8, u64) { - let rank = match part.get("type").and_then(Value::as_str) { - Some("step-start") => 0, - Some("step-finish") => 2, - _ => 1, - }; - let start = part - .pointer("/time/start") - .and_then(Value::as_u64) - .unwrap_or(0); - (rank, start) -} - -/// Continue an imported OpenCode session without adding any user message. -/// -/// `opencode run --session ` refuses to start without a message -/// ("You must provide a message or a command"), so the CLI cannot express a -/// clean continuation. The server API accepts `parts: []` on -/// `POST /session/{id}/message`; the resulting model request ends exactly at -/// the replayed boundary observation `O'N` with no injected prompt. -fn opencode_server_continuation( - plan: &ReplayPlan, - context: &RunContext<'_>, - journal: &mut Journal, - prefix: &[Value], - session_id: &str, - log_path: &Path, -) -> Result<(PathBuf, usize), ReplayError> { - let launch = context - .launch - .ok_or_else(|| ReplayError::continuation("OpenCode continuation has no launch spec"))?; - let state_dir = context.state_dir; - let opencode_config = state_dir.join("opencode-config"); - let opencode_data = state_dir.join("opencode-data"); - fs::create_dir_all(&opencode_data) - .replay_context(ReplayErrorKind::Executor, "create OpenCode data directory")?; - let logs = log_path - .parent() - .unwrap_or_else(|| Path::new(".")) - .to_path_buf(); - - let temperature = env_f64("PVISOR_OPENCODE_TEMPERATURE"); - let top_p = env_f64("PVISOR_OPENCODE_TOP_P"); - let mut children: Vec = Vec::new(); - let result = opencode_server_continuation_inner( - plan, - context, - journal, - prefix, - session_id, - log_path, - &launch.entrypoint, - &opencode_config, - &opencode_data, - &logs, - temperature, - top_p, - &mut children, - ); - drop(children); - result -} - -#[allow(clippy::too_many_arguments)] -fn opencode_server_continuation_inner( - plan: &ReplayPlan, - context: &RunContext<'_>, - journal: &mut Journal, - prefix: &[Value], - session_id: &str, - log_path: &Path, - entrypoint: &Path, - opencode_config: &Path, - opencode_data: &Path, - logs: &Path, - temperature: Option, - top_p: Option, - children: &mut Vec, -) -> Result<(PathBuf, usize), ReplayError> { - let workspace = &context.request.workspace; - let upstream = std::env::var("OPENAI_BASE_URL") - .ok() - .or_else(|| std::env::var("OPENAI_API_BASE").ok()) - .filter(|value| !value.trim().is_empty()); - let effective_base = start_opencode_sampling_proxy( - entrypoint, - context.state_dir, - logs, - upstream.as_deref().unwrap_or("http://127.0.0.1"), - temperature, - top_p, - children, - )?; - write_opencode_provider_config( - opencode_config, - effective_base.as_deref(), - temperature, - top_p, - )?; - - let export_path = context.output_dir.join("native/opencode-session.json"); - atomic_write_json( - &export_path, - &opencode_export(plan, prefix, session_id, workspace), - )?; - let mut import = Command::new(entrypoint); - import - .arg("import") - .arg( - export_path.to_str().ok_or_else(|| { - ReplayError::configuration("OpenCode export path is not valid UTF-8") - })?, - ) - .env("XDG_CONFIG_HOME", opencode_config) - .env("XDG_DATA_HOME", opencode_data) - .env("OPENCODE_DISABLE_AUTOUPDATE", "1") - .current_dir(workspace); - 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), - )); - } - - journal.append( - "continuation_started", - [("agent".into(), json!("opencode"))], - )?; - - let serve_port = loopback_free_port()?; - let serve_log = fs::OpenOptions::new() - .create(true) - .append(true) - .open(log_path) - .replay_context(ReplayErrorKind::Executor, "open OpenCode serve log")?; - let mut serve = Command::new(entrypoint); - serve - .args([ - "serve", - "--port", - &serve_port.to_string(), - "--hostname", - "127.0.0.1", - ]) - .env("XDG_CONFIG_HOME", opencode_config) - .env("XDG_DATA_HOME", opencode_data) - .env("OPENCODE_DISABLE_AUTOUPDATE", "1") - .current_dir(workspace) - .stdout( - serve_log - .try_clone() - .replay_context(ReplayErrorKind::Executor, "clone OpenCode serve log handle")?, - ) - .stderr(serve_log); - let serve = serve - .spawn() - .replay_context(ReplayErrorKind::Executor, "start OpenCode server")?; - children.push(ChildGuard(serve)); - - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .replay_context( - ReplayErrorKind::Executor, - "build OpenCode continuation runtime", - )?; - let base = format!("http://127.0.0.1:{serve_port}"); - let client = reqwest::Client::builder() - .no_proxy() - .build() - .replay_context( - ReplayErrorKind::Executor, - "build OpenCode continuation client", - )?; - let directory = workspace.to_string_lossy().to_string(); - - let ready = runtime.block_on(async { - for _ in 0..120 { - if let Ok(response) = client - .get(format!("{base}/config")) - .query(&[("directory", directory.as_str())]) - .send() - .await - && response.status().is_success() - { - return Ok(()); - } - tokio::time::sleep(Duration::from_millis(500)).await; - } - Err(ReplayError::continuation( - "OpenCode server did not become ready; see opencode.log", - )) - }); - ready?; - - let messages_url = format!("{base}/session/{session_id}/message"); - let baseline: Vec = runtime.block_on(async { - let response = client - .get(&messages_url) - .query(&[("directory", directory.as_str())]) - .send() - .await - .replay_context( - ReplayErrorKind::Continuation, - "read imported session messages", - )? - .error_for_status() - .replay_context( - ReplayErrorKind::Continuation, - "imported session request failed", - )?; - let messages: Value = response.json().await.replay_context( - ReplayErrorKind::Continuation, - "decode imported session messages", - )?; - Ok(message_ids(&messages)) - })?; - - let mut prompt_body = json!({"parts": []}); - if let Some(model) = configured_model_from_environment() - && let Some((provider, model_id)) = model.split_once('/') - { - prompt_body["model"] = json!({"providerID": provider, "modelID": model_id}); - } - let prompt_response: Value = runtime.block_on(async { - let response = client - .post(&messages_url) - .query(&[("directory", directory.as_str())]) - .json(&prompt_body) - .send() - .await - .replay_context( - ReplayErrorKind::Continuation, - "send OpenCode continuation prompt", - )?; - let status = response.status(); - let payload: Value = response.json().await.replay_context( - ReplayErrorKind::Continuation, - "decode OpenCode continuation response", - )?; - if !status.is_success() { - return Err(ReplayError::classify_continuation( - format!( - "OpenCode continuation returned {status}; see {}", - log_path.display() - ), - &payload.to_string(), - )); - } - if let Some(error) = payload.get("error") { - return Err(ReplayError::classify_continuation( - format!("OpenCode continuation failed; see {}", log_path.display()), - &error.to_string(), - )); - } - Ok(payload) - })?; - let _ = prompt_response; // final assistant message; the projection re-reads the session - - let messages: Value = runtime.block_on(async { - let response = client - .get(&messages_url) - .query(&[("directory", directory.as_str())]) - .send() - .await - .replay_context( - ReplayErrorKind::Continuation, - "read continued session messages", - )? - .error_for_status() - .replay_context( - ReplayErrorKind::Continuation, - "continued session request failed", - )?; - response.json().await.replay_context( - ReplayErrorKind::Continuation, - "decode continued session messages", - ) - })?; - let fresh: Vec = as_message_array(&messages) - .into_iter() - .filter(|message| { - message - .get("info") - .and_then(|info| info.get("id")) - .and_then(Value::as_str) - .is_some_and(|id| !baseline.contains(&id.to_owned())) - }) - .cloned() - .collect(); - let live_events = opencode_project_messages(&fresh, session_id); - let continued_steps = live_events - .iter() - .filter(|event| event.get("type") == Some(&json!("step_finish"))) - .count(); - if live_events.is_empty() { - return Err(ReplayError::continuation( - "OpenCode continuation produced no native events; see opencode.log", - )); + .map(|parts| { + parts + .iter() + .any(|part| part.get("text") == Some(&json!(nonce))) + }) + .unwrap_or(false), + Some("text") => event.pointer("/part/text") == Some(&json!(nonce)), + _ => false, } - let mut combined = prefix.to_vec(); - combined.extend(live_events); - let output_path = context.output_dir.join("native/continued-trajectory.jsonl"); - write_jsonl(&output_path, &combined)?; - Ok((output_path, continued_steps)) -} - -fn message_ids(messages: &Value) -> Vec { - as_message_array(messages) - .iter() - .filter_map(|message| { - message - .get("info") - .and_then(|info| info.get("id")) - .and_then(Value::as_str) - .map(str::to_owned) - }) - .collect() -} - -fn as_message_array(messages: &Value) -> Vec<&Value> { - let items = match messages { - Value::Array(items) => Some(items.iter().collect()), - Value::Object(_) => messages - .get("data") - .and_then(Value::as_array) - .map(|items| items.iter().collect()), - _ => None, - }; - items.unwrap_or_default() } fn continuation_session_id( @@ -2410,7 +1987,7 @@ mod tests { use super::{ CallRecord, NativeJsonlAgent, RunContext, TurnRecord, codex_native_session_id, - continuation_session_id, is_actionable_turn, opencode_project_messages, + continuation_session_id, is_actionable_turn, opencode_event_is_nonce, opencode_provider_config, parse_codex, parse_jsonl, parse_opencode, redact_codex_transport_nonce, validate_codex_continuation, }; @@ -2418,54 +1995,27 @@ mod tests { use serde_json::{Value, json}; #[test] - fn opencode_projection_restores_step_order_from_unordered_parts() { - let messages = vec![ - json!({ - "info": {"id": "msg_live_2", "role": "assistant", "time": {"created": 2}}, - "parts": [ - {"type": "step-finish", "time": {"start": 40, "end": 41}}, - {"type": "text", "text": "done", "time": {"start": 39, "end": 40}}, - {"type": "step-start"} - ] - }), - json!({ - "info": {"id": "msg_live_1", "role": "assistant", "time": {"created": 1}}, - "parts": [ - {"type": "tool", "callID": "c1", "tool": "bash", - "state": {"status": "completed", "input": {"command": "ls"}, "output": "x"}, - "time": {"start": 21, "end": 30}}, - {"type": "text", "text": "running", "time": {"start": 20, "end": 21}}, - {"type": "step-start"} - ] - }), - json!({ - "info": {"id": "msg_synthetic_user", "role": "user", "time": {"created": 0}}, - "parts": [{"type": "text", "text": "task"}] - }), + fn opencode_nonce_events_are_filtered_from_the_continued_stream() { + let nonce = "pvisor-opencode-resume-nonce"; + let events = vec![ + json!({"type": "step_start", "sessionID": "ses"}), + json!({"type": "user", "sessionID": "ses", "parts": [{"type": "text", "text": nonce}]}), + json!({"type": "text", "sessionID": "ses", "part": {"type": "text", "text": nonce}}), + json!({"type": "text", "sessionID": "ses", "part": {"type": "text", "text": "real text"}}), + json!({"type": "step_finish", "sessionID": "ses"}), ]; - let events = opencode_project_messages(&messages, "ses-live"); - let kinds: Vec<&str> = events.iter().map(|e| e["type"].as_str().unwrap()).collect(); - assert_eq!( - kinds, - vec![ - "step_start", - "text", - "tool_use", - "step_start", - "text", - "step_finish" - ] - ); - // Every event carries the session id and the native part payload. + let kept: Vec = events + .iter() + .filter(|event| !opencode_event_is_nonce(event, nonce)) + .cloned() + .collect(); + let kinds: Vec<&str> = kept.iter().map(|e| e["type"].as_str().unwrap()).collect(); + assert_eq!(kinds, vec!["step_start", "text", "step_finish"]); + assert_eq!(kept[1]["part"]["text"], "real text"); + // An empty nonce (explicit boundary prompt mode) filters nothing. for event in &events { - assert_eq!(event["sessionID"], "ses-live"); - assert!(event["part"].is_object()); + assert!(!opencode_event_is_nonce(event, "")); } - let tool = &events[2]; - assert_eq!(tool["part"]["state"]["status"], "completed"); - // The projection contains two complete live assistant turns. Parsing - // the full trajectory requires the original user event, which is - // intentionally not part of this assistant-only projection. } #[test] diff --git a/crates/persisting-replay/src/lib.rs b/crates/persisting-replay/src/lib.rs index cf42fc7a..664e2497 100644 --- a/crates/persisting-replay/src/lib.rs +++ b/crates/persisting-replay/src/lib.rs @@ -16,6 +16,7 @@ mod error; mod io; mod journal; mod model; +pub(crate) mod opencode_bridge; mod process; pub use config::{ diff --git a/crates/persisting-replay/src/opencode_bridge.rs b/crates/persisting-replay/src/opencode_bridge.rs new file mode 100644 index 00000000..992040b2 --- /dev/null +++ b/crates/persisting-replay/src/opencode_bridge.rs @@ -0,0 +1,769 @@ +//! OpenCode Responses API resume-transport bridge. +//! +//! `opencode run --session ` refuses to start without a message. The +//! SandboxReplay continuation therefore passes a unique transport nonce as +//! that message, and this bridge removes the nonce from every request before +//! forwarding it upstream, so the first live model request still ends exactly +//! at the replayed boundary observation. OpenCode may resend the full +//! conversation history (including the persisted nonce) on every request, so +//! the cleanup is exact-match and repeated. The bridge also pins sampling: +//! OpenCode never forwards `temperature`/`top_p` to the Responses API, so the +//! continuation would otherwise drift away from the recorded sampling. +//! +//! Responses bodies are streamed through unchanged apart from the JSON +//! rewrite: OpenCode treats a stalled stream as a dead connection and retries. + +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::HeaderMap; +use axum::http::{HeaderValue, StatusCode}; +use axum::response::Response; +use serde_json::Value; +use tokio::sync::oneshot; + +use crate::error::{ReplayError, ReplayErrorKind, ResultExt}; + +const BRIDGE_VERSION: &str = "sandbox-replay-opencode-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; + +pub struct OpencodeBridgeHandle { + pub base_url: String, + api_key: String, + shared: Arc, + shutdown: Option>, + worker_done: Option>>, + worker: Option>, +} + +struct BridgeShared { + state: Mutex, + client: reqwest::Client, + upstream_origin: String, + upstream_api_key: String, + routing_session_id: String, + bridge_api_key: String, + strip_prompt: Option, + temperature: Option, + top_p: Option, + disable_thinking: bool, + cancelled: AtomicBool, +} + +#[derive(Default)] +struct BridgeState { + forwarded_requests: usize, + removed_transport_prompt: bool, + 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 OpencodeBridgeHandle { + /// Start the bridge. `strip_prompt` is the transport nonce to remove + /// from every request; `None` keeps an explicit `boundary_user_prompt` + /// in the model input on purpose. + pub fn start( + routing_session_id: &str, + strip_prompt: Option, + temperature: Option, + top_p: Option, + disable_thinking: bool, + ) -> Result { + let upstream_base = first_nonempty_env(&["OPENAI_BASE_URL", "OPENAI_API_BASE"]) + .ok_or_else(|| { + ReplayError::configuration( + "OpenCode SandboxReplay bridge requires OPENAI_BASE_URL or OPENAI_API_BASE", + ) + })?; + let upstream_api_key = + first_nonempty_env(&["OPENAI_API_KEY", "LLM_API_KEY"]).ok_or_else(|| { + ReplayError::configuration( + "OpenCode SandboxReplay bridge requires OPENAI_API_KEY or LLM_API_KEY", + ) + })?; + let upstream_origin = url_origin(&upstream_base)?; + // Mirror the original API path prefix (for example "/v1"): OpenCode + // appends "/responses" to this base URL and the bridge forwards the + // resulting path verbatim, so dropping the prefix would 404 upstream. + let upstream_prefix = url_path_prefix(&upstream_base)?; + 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 OpenCode SandboxReplay bridge port", + )?; + let address = listener.local_addr().replay_context( + ReplayErrorKind::Continuation, + "read OpenCode SandboxReplay bridge address", + )?; + listener.set_nonblocking(true).replay_context( + ReplayErrorKind::Continuation, + "configure OpenCode SandboxReplay bridge listener", + )?; + let client = reqwest::Client::builder() + .no_proxy() + .build() + .replay_context( + ReplayErrorKind::Continuation, + "build OpenCode SandboxReplay bridge client", + )?; + let shared = Arc::new(BridgeShared { + state: Mutex::new(BridgeState::default()), + client, + upstream_origin, + upstream_api_key, + routing_session_id: routing_session_id.to_owned(), + bridge_api_key: bridge_api_key.clone(), + strip_prompt, + temperature, + top_p, + disable_thinking, + cancelled: AtomicBool::new(false), + }); + 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-opencode-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 OpenCode SandboxReplay bridge thread", + )?; + let mut handle = Self { + base_url: format!("http://{address}{upstream_prefix}"), + 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!( + "OpenCode SandboxReplay bridge did not become ready within {} seconds", + START_TIMEOUT.as_secs() + )), + Err(mpsc::RecvTimeoutError::Disconnected) => { + Some("OpenCode SandboxReplay bridge exited before reporting readiness".into()) + } + }; + if let Some(message) = startup_error { + let _ = handle.stop_worker(); + return Err(ReplayError::continuation(message)); + } + Ok(handle) + } + + /// Environment for the OpenCode child so it talks only to this bridge. + pub fn child_environment(&self) -> Vec<(String, String)> { + let no_proxy = merged_no_proxy_environment(); + vec![ + ("OPENAI_BASE_URL".to_owned(), self.base_url.clone()), + ("OPENAI_API_BASE".to_owned(), self.base_url.clone()), + ("OPENAI_API_KEY".to_owned(), self.api_key.clone()), + ("NO_PROXY".to_owned(), no_proxy.clone()), + ("no_proxy".to_owned(), no_proxy), + ] + } + + pub fn finish(mut self) -> Result { + self.stop_worker()?; + let state = self + .shared + .state + .lock() + .map_err(|_| ReplayError::continuation("OpenCode bridge state lock poisoned"))?; + if state.failed { + return Err(ReplayError::continuation(format!( + "OpenCode resume transport bridge failed closed: {}", + state + .failure + .as_deref() + .unwrap_or("unknown protocol failure") + ))); + } + if state.forwarded_requests == 0 { + return Err(ReplayError::continuation( + "OpenCode continuation made no model request through the SandboxReplay bridge", + )); + } + if self.shared.strip_prompt.is_some() && !state.removed_transport_prompt { + return Err(ReplayError::continuation( + "OpenCode transport nonce was not removed from any model request", + )); + } + Ok(state.forwarded_requests) + } + + fn stop_worker(&mut self) -> Result<(), ReplayError> { + self.shared.cancelled.store(true, Ordering::Release); + 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!( + "OpenCode 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( + "OpenCode SandboxReplay bridge thread panicked", + )); + } + if let Some(result) = worker_result { + result.replay_context( + ReplayErrorKind::Executor, + "stop OpenCode SandboxReplay bridge", + )?; + } + Ok(()) + } +} + +impl Drop for OpencodeBridgeHandle { + 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_current_thread() + .enable_all() + .build() + .map_err(|error| { + let _ = ready_tx.send(Err(format!("build OpenCode 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 OpenCode bridge listener: {error}"))); + error + })?; + ready_tx + .send(Ok(())) + .map_err(|_| anyhow::anyhow!("OpenCode 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", axum::routing::get(health)) + .fallback(forward_handler) + .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)) + .with_state(shared) +} + +async fn health(State(shared): State>) -> Response { + let (failed, forwarded) = shared + .state + .lock() + .map(|state| (state.failed, state.forwarded_requests)) + .unwrap_or((true, 0)); + json_response(StatusCode::OK, json_health(failed, forwarded)) +} + +fn json_health(failed: bool, forwarded: usize) -> Bytes { + serde_json::to_vec(&serde_json::json!({ + "status": "healthy", + "bridge_version": BRIDGE_VERSION, + "failed": failed, + "forwarded_requests": forwarded, + })) + .unwrap_or_default() + .into() +} + +/// Forward any request upstream, rewriting JSON bodies: strip the transport +/// nonce and pin sampling. Non-JSON requests (catalog fetches and probes) +/// pass through untouched. +async fn forward_handler( + State(shared): State>, + request: axum::extract::Request, +) -> Response { + if !authorized(&shared, request.headers()) { + return error_response(StatusCode::UNAUTHORIZED, "invalid bridge API key"); + } + let method = request.method().clone(); + let path = request + .uri() + .path_and_query() + .map(|v| v.as_str().to_owned()); + let headers = request.headers().clone(); + let body = match axum::body::to_bytes(request.into_body(), MAX_BODY_BYTES).await { + Ok(bytes) => bytes, + Err(error) => { + fail( + &shared, + format!("read OpenCode bridge request body: {error}"), + ); + return error_response(StatusCode::BAD_REQUEST, "invalid request body"); + } + }; + let Some(path) = path else { + return error_response(StatusCode::BAD_REQUEST, "request has no path"); + }; + if std::env::var("PVISOR_OPENCODE_BRIDGE_DEBUG").is_ok() { + use std::io::Write; + if let Ok(mut log) = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open("/tmp/pvisor-opencode-bridge-debug.log") + { + let _ = writeln!( + log, + "[{}] {} {} body={}B head={}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0), + method, + path, + body.len(), + String::from_utf8_lossy(&body[..body.len().min(300)]).replace('\n', " ") + ); + } + } + let content_type = headers + .get(axum::http::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_ascii_lowercase(); + let is_json = method == axum::http::Method::POST && content_type.contains("application/json"); + let body: Bytes = if is_json && !body.is_empty() { + match rewrite_request(&shared, &body) { + Ok(rewritten) => rewritten, + Err(error) => { + fail(&shared, error.to_string()); + return error_response(StatusCode::UNPROCESSABLE_ENTITY, &error.to_string()); + } + } + } else { + body + }; + let upstream_url = format!("{}{}", shared.upstream_origin, path); + let mut upstream = shared.client.request(method, &upstream_url); + for (name, value) in headers.iter() { + let name = name.as_str(); + if is_hop_by_hop_header(name) || name == "host" { + continue; + } + if name == "authorization" || name == "x-api-key" { + continue; + } + // The JSON rewrite changes the body length (nonce removal, sampling + // injection), so the incoming framing headers must never be trusted; + // reqwest re-frames the full Bytes body itself. + if name == "content-length" || name == "transfer-encoding" { + continue; + } + upstream = upstream.header(name, value.clone()); + } + upstream = upstream + .header( + axum::http::header::AUTHORIZATION.as_str(), + format!("Bearer {}", shared.upstream_api_key), + ) + .header("X-LiteLLM-Session-ID", &shared.routing_session_id); + if !body.is_empty() { + upstream = upstream.body(reqwest::Body::from(body)); + } + let response = match upstream.send().await { + Ok(response) => response, + Err(error) => { + fail(&shared, format!("forward OpenCode request: {error}")); + return error_response(StatusCode::BAD_GATEWAY, &error.to_string()); + } + }; + let status = response.status(); + if std::env::var("PVISOR_OPENCODE_BRIDGE_DEBUG").is_ok() { + use std::io::Write; + if let Ok(mut log) = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open("/tmp/pvisor-opencode-bridge-debug.log") + { + let _ = writeln!(log, "[upstream] status={}", status.as_u16()); + } + } + let mut response_headers = HeaderMap::new(); + for (name, value) in response.headers().iter() { + if is_hop_by_hop_header(name.as_str()) { + continue; + } + if let Ok(header_value) = HeaderValue::from_bytes(value.as_bytes()) { + response_headers.insert(name.clone(), header_value); + } + } + let _ = response_headers.remove("content-length"); + let stream = response.bytes_stream(); + let mut builder = Response::builder() + .status(StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY)); + for (name, value) in response_headers.iter() { + builder = builder.header(name.clone(), value.clone()); + } + match builder.body(Body::from_stream(stream)) { + Ok(response) => { + if let Ok(mut state) = shared.state.lock() { + state.forwarded_requests += 1; + } + response + } + Err(error) => { + fail(&shared, format!("build OpenCode bridge response: {error}")); + error_response(StatusCode::BAD_GATEWAY, "bridge response failure") + } + } +} + +fn rewrite_request(shared: &BridgeShared, body: &[u8]) -> anyhow::Result { + let mut payload: Value = + serde_json::from_slice(body).context("OpenCode bridge request is not valid JSON")?; + if !payload.is_object() { + anyhow::bail!("OpenCode bridge request must be a JSON object"); + } + if let Some(nonce) = &shared.strip_prompt { + let removed = remove_exact_user_input(&mut payload, nonce)?; + if removed > 0 + && let Ok(mut state) = shared.state.lock() + { + state.removed_transport_prompt = true; + } + } + let mut changed = false; + if let Some(temperature) = shared.temperature { + payload["temperature"] = serde_json::json!(temperature); + changed = true; + } + if let Some(top_p) = shared.top_p { + payload["top_p"] = serde_json::json!(top_p); + changed = true; + } + if shared.disable_thinking { + // Greedy reasoning models can loop until the output cap and emit an + // empty turn; the endpoint only disables thinking through the chat + // template, which OpenCode cannot express. + payload["chat_template_kwargs"] = + serde_json::json!({ "enable_thinking": false }); + changed = true; + } + if changed || shared.strip_prompt.is_some() { + Ok(serde_json::to_vec(&payload)?.into()) + } else { + Ok(Bytes::copy_from_slice(body)) + } +} + +/// Remove the exact nonce user message from a request. Handles both the +/// Responses `input` array (items typed `user` or roled `user` with +/// `input_text` content) and the Chat Completions `messages` array. +fn remove_exact_user_input(payload: &mut Value, expected: &str) -> anyhow::Result { + let mut removed = 0; + for key in ["input", "messages"] { + let Some(items) = payload.get_mut(key).and_then(Value::as_array_mut) else { + continue; + }; + let mut index = 0; + while index < items.len() { + let is_user = items[index].get("role").and_then(Value::as_str) == Some("user") + || items[index].get("type").and_then(Value::as_str) == Some("user"); + if is_user && exact_message_text(&items[index]) == Some(expected) { + items.remove(index); + removed += 1; + } else { + index += 1; + } + } + } + Ok(removed) +} + +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]; + let kind = block.get("type").and_then(Value::as_str)?; + if kind != "input_text" && kind != "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(axum::http::header::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 error_response(status: StatusCode, message: &str) -> Response { + json_response( + status, + serde_json::to_vec(&serde_json::json!({"error": {"message": message}})) + .unwrap_or_default() + .into(), + ) +} + +fn json_response(status: StatusCode, body: Bytes) -> Response { + Response::builder() + .status(status) + .header(axum::http::header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .unwrap_or_else(|_| Response::new(Body::empty())) +} + +fn is_hop_by_hop_header(name: &str) -> bool { + matches!( + name, + "connection" + | "keep-alive" + | "proxy-authenticate" + | "proxy-authorization" + | "te" + | "trailer" + | "transfer-encoding" + | "upgrade" + ) +} + +fn url_path_prefix(base: &str) -> Result { + let parsed = reqwest::Url::parse(base) + .replay_context(ReplayErrorKind::Configuration, "parse OpenCode upstream URL")?; + let path = parsed.path().trim_end_matches('/'); + Ok(path.to_owned()) +} + +fn url_origin(base: &str) -> Result { + let parsed = reqwest::Url::parse(base).replay_context( + ReplayErrorKind::Configuration, + "parse OpenCode upstream URL", + )?; + let origin = match parsed.port() { + Some(port) => format!( + "{}://{}:{}", + parsed.scheme(), + parsed.host_str().unwrap_or_default(), + port + ), + None => format!( + "{}://{}", + parsed.scheme(), + parsed.host_str().unwrap_or_default() + ), + }; + if parsed.scheme() != "http" && parsed.scheme() != "https" { + return Err(ReplayError::configuration( + "OpenCode upstream URL must be HTTP(S)", + )); + } + Ok(origin) +} + +fn first_nonempty_env(names: &[&str]) -> Option { + names + .iter() + .filter_map(|name| std::env::var(name).ok()) + .find(|value| !value.trim().is_empty()) +} + +fn merged_no_proxy_environment() -> String { + let mut entries: Vec = ["127.0.0.1", "localhost", "::1"] + .iter() + .map(|entry| (*entry).to_owned()) + .collect(); + for name in ["NO_PROXY", "no_proxy"] { + if let Ok(value) = std::env::var(name) + && !value.trim().is_empty() + { + entries.extend(value.split(',').map(|entry| entry.trim().to_owned())); + } + } + entries.join(",") +} + +#[cfg(test)] +mod tests { + use super::{BridgeShared, remove_exact_user_input, rewrite_request}; + use serde_json::json; + use std::sync::Mutex; + + fn shared(strip: Option<&str>, temperature: Option, top_p: Option) -> BridgeShared { + shared_full(strip, temperature, top_p, false) + } + + fn shared_full( + strip: Option<&str>, + temperature: Option, + top_p: Option, + disable_thinking: bool, + ) -> BridgeShared { + BridgeShared { + state: Mutex::new(Default::default()), + client: reqwest::Client::new(), + upstream_origin: "http://127.0.0.1:9".into(), + upstream_api_key: "upstream".into(), + routing_session_id: "ses".into(), + bridge_api_key: "bridge".into(), + strip_prompt: strip.map(str::to_owned), + temperature, + top_p, + disable_thinking, + cancelled: Default::default(), + } + } + + #[test] + fn disables_thinking_through_the_chat_template() { + let bridge = shared_full(None, Some(0.0), Some(1.0), true); + let request = json!({"model": "m", "input": [{"role": "user", "content": "task"}]}); + let rewritten: serde_json::Value = serde_json::from_slice( + &rewrite_request(&bridge, serde_json::to_vec(&request).unwrap().as_slice()).unwrap(), + ) + .unwrap(); + assert_eq!( + rewritten["chat_template_kwargs"], + serde_json::json!({"enable_thinking": false}) + ); + } + + #[test] + fn strips_nonce_and_pins_sampling_in_responses_shape() { + let bridge = shared(Some("pvisor-opencode-resume-nonce"), Some(0.0), Some(1.0)); + let request = json!({ + "model": "m", + "input": [ + {"type": "system"}, + {"role": "user", "content": "the task"}, + {"role": "assistant", "content": [{"type": "output_text", "text": "working"}]}, + {"type": "function_call", "call_id": "c1"}, + {"type": "function_call_output", "call_id": "c1"}, + {"type": "user", "content": [{"type": "input_text", "text": "pvisor-opencode-resume-nonce"}]} + ] + }); + let rewritten: serde_json::Value = serde_json::from_slice( + &rewrite_request(&bridge, serde_json::to_vec(&request).unwrap().as_slice()).unwrap(), + ) + .unwrap(); + let kinds: Vec<&str> = rewritten["input"] + .as_array() + .unwrap() + .iter() + .map(|item| { + item.get("type") + .and_then(|v| v.as_str()) + .or_else(|| item.get("role").and_then(|v| v.as_str())) + .unwrap_or("?") + }) + .collect(); + assert_eq!( + kinds, + vec![ + "system", + "user", + "assistant", + "function_call", + "function_call_output" + ] + ); + assert_eq!(rewritten["temperature"], 0.0); + assert_eq!(rewritten["top_p"], 1.0); + assert!(bridge.state.lock().unwrap().removed_transport_prompt); + } + + #[test] + fn strips_nonce_from_chat_completions_shape() { + let mut request = json!({ + "model": "m", + "messages": [ + {"role": "user", "content": "task"}, + {"role": "user", "content": "nonce-value"}, + {"role": "assistant", "content": "ok"} + ] + }); + assert_eq!( + remove_exact_user_input(&mut request, "nonce-value").unwrap(), + 1 + ); + assert_eq!(request["messages"].as_array().unwrap().len(), 2); + } + + #[test] + fn keeps_explicit_boundary_prompt() { + let bridge = shared(None, None, None); + let request = json!({ + "model": "m", + "input": [ + {"role": "user", "content": [{"type": "input_text", "text": "continue from here"}]} + ] + }); + let rewritten: serde_json::Value = serde_json::from_slice( + &rewrite_request(&bridge, serde_json::to_vec(&request).unwrap().as_slice()).unwrap(), + ) + .unwrap(); + assert_eq!(rewritten["input"].as_array().unwrap().len(), 1); + assert!(!bridge.state.lock().unwrap().removed_transport_prompt); + } +} From 5ee02d304a0f16adbffb144c283a33526189276c Mon Sep 17 00:00:00 2001 From: guoxu1 Date: Wed, 9 Sep 2026 18:06:06 +0800 Subject: [PATCH 6/6] style(replay): apply rustfmt to OpenCode bridge --- crates/persisting-replay/src/opencode_bridge.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/crates/persisting-replay/src/opencode_bridge.rs b/crates/persisting-replay/src/opencode_bridge.rs index 992040b2..382eb34f 100644 --- a/crates/persisting-replay/src/opencode_bridge.rs +++ b/crates/persisting-replay/src/opencode_bridge.rs @@ -487,8 +487,7 @@ fn rewrite_request(shared: &BridgeShared, body: &[u8]) -> anyhow::Result // Greedy reasoning models can loop until the output cap and emit an // empty turn; the endpoint only disables thinking through the chat // template, which OpenCode cannot express. - payload["chat_template_kwargs"] = - serde_json::json!({ "enable_thinking": false }); + payload["chat_template_kwargs"] = serde_json::json!({ "enable_thinking": false }); changed = true; } if changed || shared.strip_prompt.is_some() { @@ -590,8 +589,10 @@ fn is_hop_by_hop_header(name: &str) -> bool { } fn url_path_prefix(base: &str) -> Result { - let parsed = reqwest::Url::parse(base) - .replay_context(ReplayErrorKind::Configuration, "parse OpenCode upstream URL")?; + let parsed = reqwest::Url::parse(base).replay_context( + ReplayErrorKind::Configuration, + "parse OpenCode upstream URL", + )?; let path = parsed.path().trim_end_matches('/'); Ok(path.to_owned()) }