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 88baba1d..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)] @@ -993,6 +994,8 @@ fn continue_native_cli( let mut codex_bridge = None; let mut codex_transport_prompt = None; 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( @@ -1007,10 +1010,32 @@ 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"); + // `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), @@ -1049,6 +1074,9 @@ fn continue_native_cli( 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]); } @@ -1062,10 +1090,15 @@ fn continue_native_cli( if !context.request.disable_thinking { command.arg("--thinking"); } - if let Some(prompt) = context.request.boundary_user_prompt() { - command.arg("--"); - command.arg(prompt); + 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); @@ -1150,10 +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| { - let result = bridge.finish(); - result - }); + 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( @@ -1208,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); @@ -1230,6 +1270,99 @@ 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. +/// `effective_base` overrides the environment endpoint (used for the local +/// sampling-injection proxy). +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, + effective_base: Option<&str>, + temperature: Option, + top_p: Option, +) -> Result<(), ReplayError> { + let Some(model) = configured_model_from_environment() else { + return 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(), 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", + )?; + atomic_write_json(&directory.join("opencode.json"), &config) +} + +/// 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; + } + match event.get("type").and_then(Value::as_str) { + Some("user") => event + .get("parts") + .and_then(Value::as_array) + .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, + } +} + fn continuation_session_id( agent: NativeJsonlAgent, plan: &ReplayPlan, @@ -1570,6 +1703,17 @@ 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, @@ -1577,7 +1721,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", @@ -1705,8 +1849,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()}, @@ -1843,12 +1987,77 @@ mod tests { use super::{ CallRecord, NativeJsonlAgent, RunContext, TurnRecord, codex_native_session_id, - continuation_session_id, is_actionable_turn, parse_codex, parse_jsonl, parse_opencode, + 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, }; use crate::model::{AgentKind, PlaybackRequest, ReplayMode, ReplayPlan, ToolBatch, ToolCall}; use serde_json::{Value, json}; + #[test] + 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 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!(!opencode_event_is_nonce(event, "")); + } + } + + #[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 = [ 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..382eb34f --- /dev/null +++ b/crates/persisting-replay/src/opencode_bridge.rs @@ -0,0 +1,770 @@ +//! 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); + } +}