From 5139ba5d1e550b581ffede3c70387edaa2f76f0e Mon Sep 17 00:00:00 2001 From: David Viejo Date: Mon, 21 Sep 2026 18:32:46 +0200 Subject: [PATCH 1/6] feat(adapter): let a provider carry a turn on its own protocol streams MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Most provider CLIs speak their protocol over stdout and stdin, so the turn loop reads frames from the child and writes them back to it. `opencode serve` does not: it exposes HTTP and Server-Sent Events on a loopback port and leaves its own stdio empty, so a turn driven that way had no way to join the existing loop. Add `AgentAdapter::attach`, which may return `ProtocolStreams` — a reader and a writer the runtime uses in place of the child's stdout and stdin. The frame contract is unchanged on purpose: newline-delimited frames in, newline-terminated frames out, so `parse_line` stays one synchronous, fully testable state machine and cancellation, cooperative interrupts, interaction timeouts and line bounding keep working for both kinds of provider instead of growing a second turn loop. `prepare_turn` now runs before the command is built, and `command_for_turn` exposes the state it seeded. An adapter that must agree with itself about a per-turn value — the loopback port `opencode serve` is told to bind and that `attach` later connects to — decides it once rather than twice. Both additions are defaulted, so existing adapters are unaffected. --- src/adapter.rs | 76 +++++++++++++++++++++++++++++++++++++++++++++++--- src/lib.rs | 2 +- src/runtime.rs | 26 +++++++++++++++-- 3 files changed, 97 insertions(+), 7 deletions(-) diff --git a/src/adapter.rs b/src/adapter.rs index 516c603..622088a 100644 --- a/src/adapter.rs +++ b/src/adapter.rs @@ -215,6 +215,39 @@ pub struct AdapterOutput { pub terminal: bool, } +/// Protocol carrier an adapter supplies in place of the provider's own stdio. +/// +/// Most provider CLIs speak their protocol over stdout and stdin, so the +/// runtime reads frames from the child and writes [`AdapterOutput::writes`] +/// back to it. A provider whose protocol is *not* carried by its own stdio — +/// `opencode serve`, which exposes HTTP and Server-Sent Events on a loopback +/// port — returns these streams from [`AgentAdapter::attach`] instead. +/// +/// The frame contract is deliberately unchanged: the runtime still reads +/// newline-delimited frames from [`Self::reader`] and still writes +/// newline-terminated frames to [`Self::writer`]. [`AgentAdapter::parse_line`] +/// therefore stays one synchronous, fully testable state machine no matter +/// what actually moves the bytes, and cancellation, interrupts, interaction +/// timeouts and line bounding keep working without a second code path. +/// +/// The child process is still spawned, supervised and torn down by the +/// runtime. An adapter that returns streams here must keep the turn's +/// liveness tied to that child: the runtime fails the turn when the process +/// exits before a terminal frame arrives, so a server that dies mid-turn +/// surfaces immediately instead of hanging until the turn deadline. +pub struct ProtocolStreams { + /// Newline-delimited frames parsed by [`AgentAdapter::parse_line`]. + pub reader: crate::TransportReader, + /// Sink for [`AdapterOutput::writes`] and encoded interaction responses. + pub writer: crate::TransportWriter, +} + +impl fmt::Debug for ProtocolStreams { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.debug_struct("ProtocolStreams").finish() + } +} + /// Adapter between a provider-native CLI protocol and normalized events. #[async_trait] pub trait AgentAdapter: Send + Sync { @@ -327,17 +360,52 @@ pub trait AgentAdapter: Send + Sync { /// Build the provider process for one validated request. fn command(&self, request: &TurnRequest) -> Result; + /// Build the provider process using state seeded by [`Self::prepare_turn`]. + /// + /// The default ignores the state and defers to [`Self::command`], which is + /// what a provider that encodes its whole turn in argv and stdin needs. + /// An adapter that must agree with itself about a value chosen per turn — + /// the loopback port `opencode serve` is told to bind and that + /// [`Self::attach`] then connects to — overrides this instead, so the + /// value is decided once in `prepare_turn` and read back here. + fn command_for_turn(&self, request: &TurnRequest, state: &AdapterState) -> Result { + let _ = state; + self.command(request) + } + /// Seed per-turn parser state from the validated request. /// - /// The runtime calls this once, after [`Self::command`] and before the - /// first output line. Adapters whose protocol issues requests of its own - /// (rather than encoding the whole turn in argv and stdin) use it to - /// retain the turn parameters that [`Self::parse_line`] later needs. + /// The runtime calls this once, before [`Self::command_for_turn`] and + /// before the first output line. Adapters whose protocol issues requests + /// of its own (rather than encoding the whole turn in argv and stdin) use + /// it to retain the turn parameters that [`Self::parse_line`] later needs. fn prepare_turn(&self, request: &TurnRequest, state: &mut AdapterState) -> Result<()> { let _ = (request, state); Ok(()) } + /// Supply a protocol carrier to use instead of the child's stdout and stdin. + /// + /// Called once, after the provider process is spawned and before the first + /// frame is read. Returning `None` — the default — keeps the ordinary + /// stdio contract. Returning [`ProtocolStreams`] tells the runtime to read + /// frames from, and write frames to, those streams instead; the child is + /// still spawned, supervised, stderr-drained and terminated by the runtime + /// exactly as before. + /// + /// This is how a provider whose protocol lives somewhere other than its + /// own stdio joins the normal turn loop rather than growing a parallel + /// one. The implementation typically spawns a task that translates the + /// provider's native transport into newline-delimited frames. + async fn attach( + &self, + request: &TurnRequest, + state: &AdapterState, + ) -> Result> { + let _ = (request, state); + Ok(None) + } + /// Translate one stdout line and update accumulated state. fn parse_line(&self, line: &str, state: &mut AdapterState) -> Result; diff --git a/src/lib.rs b/src/lib.rs index 3ff8a03..67b53f7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -50,7 +50,7 @@ pub mod tailnet; pub use adapter::{ AccountUsageProbeSpec, AdapterOutput, AdapterState, AgentAdapter, AuthenticationProbeSpec, - CatalogProbeSpec, CommandSpec, InteractionRequest, ProviderTerminalFailure, + CatalogProbeSpec, CommandSpec, InteractionRequest, ProtocolStreams, ProviderTerminalFailure, }; pub use chat::{ Chat, ChatApproval, ChatAttachment, ChatCommit, ChatEvent, ChatEventData, ChatMessage, diff --git a/src/runtime.rs b/src/runtime.rs index fc084c7..f432290 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -2265,14 +2265,18 @@ impl AgentRuntime { interactions: &dyn InteractionHandler, ) -> Result { let provider = request.provider; - let mut spec = adapter.command(request)?; let mut state = AdapterState::default(); // A resumed provider process commonly repeats its native session ID in // the startup handshake. Seed the parser with the ID the caller is // already attached to so adapters do not project that handshake as a // second `SessionStarted` lifecycle event. state.result.session_id.clone_from(&request.session_id); + // Seeded before the command is built so an adapter that must agree + // with itself about a per-turn value — the loopback port an + // `opencode serve` child is told to bind, which `attach` later + // connects to — decides it once, here. adapter.prepare_turn(request, &mut state)?; + let mut spec = adapter.command_for_turn(request, &state)?; for (name, value) in &request.environment { spec.environment.insert(name.into(), value.expose().into()); } @@ -2380,7 +2384,25 @@ impl AgentRuntime { source: std::io::Error::new(std::io::ErrorKind::BrokenPipe, "stderr was not piped"), })?; let stderr_task = tokio::spawn(crate::process::bounded_stderr(stderr, STDERR_TAIL_BYTES)); - let mut lines = BufReader::new(stdout).lines(); + // A provider whose protocol is not carried by its own stdio replaces + // both halves here. The child stays spawned, supervised and + // stderr-drained exactly as before; only the frame carrier differs, + // so everything below this point — cancellation, interrupts, + // interaction timeouts, line bounding — is shared by both kinds of + // provider instead of growing a second turn loop. + let reader: crate::TransportReader = match adapter.attach(request, &state).await { + Ok(Some(streams)) => { + stdin = Some(streams.writer); + streams.reader + } + Ok(None) => stdout, + Err(error) => { + let _ = process.terminate().await; + stderr_task.abort(); + return Err(error); + } + }; + let mut lines = BufReader::new(reader).lines(); loop { let line = tokio::select! { _ = request.cancellation.cancelled() => { From dc240cd734a601560eee37b4c5f8e475a6f20c3a Mon Sep 17 00:00:00 2001 From: David Viejo Date: Mon, 21 Sep 2026 18:36:11 +0200 Subject: [PATCH 2/6] feat(opencode): add a loopback HTTP and SSE carrier for opencode serve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `opencode serve` speaks HTTP and Server-Sent Events on a loopback port and writes nothing useful to its own stdio, so a turn driven that way needs a carrier before it can use `AgentAdapter::attach`. Add one, deliberately dumb: it performs the requests it is told to perform and reports what came back, knowing nothing about sessions, permissions or prompts. All protocol sequencing stays in the synchronous state machine that lands next, so that logic remains unit-testable against scripted lines exactly like the Codex app-server transport, and this module remains testable against a scripted HTTP server. No HTTP crate is used. `reqwest` is optional and wired only to the `temps-sandbox` feature; pulling a TLS-capable client stack into the default feature set to talk to 127.0.0.1 would be a poor trade, and `tailnet::proxy` already speaks HTTP/1.1 over `tokio::net::TcpStream` here for the same reason. Chunked transfer framing is decoded properly rather than scanned for `data:` prefixes: SSE is always chunked, so an event that straddles a chunk boundary would otherwise be silently truncated. A test covers exactly that case, and another covers a server that drops its connection mid-stream — the shape a crashing `opencode serve` takes — which is reported as a frame instead of being awaited until the turn deadline. --- src/providers/mod.rs | 2 + src/providers/opencode_http.rs | 717 +++++++++++++++++++++++++++++++++ 2 files changed, 719 insertions(+) create mode 100644 src/providers/opencode_http.rs diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 239b17e..adfad75 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -8,6 +8,8 @@ mod codex; mod codex_app_server; #[cfg(feature = "opencode")] mod opencode; +#[cfg(feature = "opencode")] +mod opencode_http; #[cfg(feature = "claude")] pub use claude::Claude; diff --git a/src/providers/opencode_http.rs b/src/providers/opencode_http.rs new file mode 100644 index 0000000..5f57a18 --- /dev/null +++ b/src/providers/opencode_http.rs @@ -0,0 +1,717 @@ +//! Loopback HTTP/1.1 and Server-Sent Events carrier for `opencode serve`. +//! +//! `opencode serve` speaks HTTP and SSE on a loopback port and leaves its own +//! stdio empty, so the turn loop cannot read it directly. This module bridges +//! the two: it exposes [`crate::ProtocolStreams`] whose reader yields +//! newline-delimited JSON frames and whose writer accepts them, and it runs a +//! task that translates those frames into real HTTP requests and a real SSE +//! subscription. +//! +//! The bridge is deliberately *dumb*. It performs requests it is told to +//! perform and reports what came back; it knows nothing about sessions, +//! permissions or prompts. All protocol sequencing lives in +//! [`super::opencode_serve`]'s synchronous state machine, which keeps that +//! logic unit-testable against scripted lines exactly like the Codex +//! app-server transport, and keeps this module testable against a scripted +//! HTTP server. +//! +//! No HTTP crate is used. `reqwest` is an optional dependency wired only to +//! the `temps-sandbox` feature, and pulling a TLS-capable client stack into +//! the default feature set to talk to `127.0.0.1` would be a poor trade. The +//! surface needed here is small and entirely plaintext loopback: one request +//! per connection with `Connection: close`, plus one long-lived connection +//! for the event stream. `crate::tailnet::proxy` already speaks HTTP/1.1 over +//! `tokio::net::TcpStream` in this crate for the same reason. + +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; +use std::time::Duration; + +use serde_json::{json, Value}; +use tokio::io::{ + AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, DuplexStream, +}; +use tokio::net::TcpStream; + +use crate::ProtocolStreams; + +/// Buffer shared by the bridge and the turn loop, in bytes. +/// +/// Large enough that a burst of SSE events does not stall the reader task, +/// small enough to apply real backpressure if the turn loop stops consuming. +const BRIDGE_BUFFER_BYTES: usize = 256 * 1024; + +/// How long to keep polling before declaring that the server never came up. +/// +/// `opencode serve` signals readiness on no pipe this process can observe — +/// the runtime owns the child's stdio and the server writes nothing useful to +/// it — so readiness is polled against a real endpoint, matching the +/// reference driver. +const READINESS_ATTEMPTS: u32 = 50; +const READINESS_INTERVAL: Duration = Duration::from_millis(200); + +/// Cap on a single HTTP response body, so a wedged or hostile server cannot +/// exhaust memory through the bridge. +const MAX_RESPONSE_BYTES: usize = 8 * 1024 * 1024; + +/// Cap on one SSE payload line. +const MAX_EVENT_BYTES: usize = 4 * 1024 * 1024; + +/// One request the state machine asked the bridge to perform. +#[derive(Debug, serde::Deserialize)] +pub(super) struct BridgeCommand { + /// Correlation identifier echoed on the response frame. + #[serde(default)] + pub id: Option, + /// `GET`, `POST`, or the bridge-private `SUBSCRIBE`. + pub method: String, + /// Request target, including any query string. + pub path: String, + /// Optional JSON request body. + #[serde(default)] + pub body: Option, +} + +/// Frame kinds the bridge emits. Prefixed so they cannot collide with a +/// native OpenCode event type. +pub(super) const FRAME_READY: &str = "@ready"; +pub(super) const FRAME_RESPONSE: &str = "@response"; +pub(super) const FRAME_EVENT: &str = "@event"; +pub(super) const FRAME_ERROR: &str = "@error"; + +/// Start the bridge for a turn and hand the runtime its protocol streams. +/// +/// The returned streams are live immediately; the task behind them polls the +/// server for readiness first and emits [`FRAME_READY`] once it answers. +pub(super) fn connect(port: u16) -> ProtocolStreams { + let (runtime_side, bridge_side) = tokio::io::duplex(BRIDGE_BUFFER_BYTES); + let (reader, writer) = tokio::io::split(runtime_side); + tokio::spawn(run_bridge(port, bridge_side)); + ProtocolStreams { + reader: Box::new(reader), + writer: Box::new(writer), + } +} + +fn address(port: u16) -> SocketAddr { + SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, port)) +} + +/// Drive one turn's HTTP traffic until the state machine stops writing or the +/// server stops answering. +async fn run_bridge(port: u16, bridge_side: DuplexStream) { + let (incoming, outgoing) = tokio::io::split(bridge_side); + let outgoing = std::sync::Arc::new(tokio::sync::Mutex::new(outgoing)); + + if let Err(error) = wait_until_ready(port).await { + emit_error(&outgoing, &error).await; + return; + } + if emit(&outgoing, &json!({"type": FRAME_READY})) + .await + .is_err() + { + return; + } + + let mut commands = BufReader::new(incoming).lines(); + let mut subscription: Option> = None; + loop { + // The turn loop closing its writer is the normal end of a turn. + let Ok(Some(line)) = commands.next_line().await else { + break; + }; + if line.trim().is_empty() { + continue; + } + let Ok(command) = serde_json::from_str::(&line) else { + emit_error(&outgoing, "the OpenCode bridge received an unreadable command").await; + break; + }; + if command.method == "SUBSCRIBE" { + if subscription.is_none() { + subscription = Some(tokio::spawn(stream_events( + port, + command.path, + std::sync::Arc::clone(&outgoing), + ))); + } + continue; + } + let outcome = perform(port, &command.method, &command.path, command.body.as_ref()).await; + let frame = match outcome { + Ok((status, body)) => json!({ + "type": FRAME_RESPONSE, + "id": command.id, + "status": status, + "body": body, + }), + Err(error) => json!({ + "type": FRAME_RESPONSE, + "id": command.id, + "status": 0, + "error": error.to_string(), + }), + }; + if emit(&outgoing, &frame).await.is_err() { + break; + } + } + if let Some(subscription) = subscription { + subscription.abort(); + } +} + +/// Poll a cheap, always-safe endpoint until the server answers. +async fn wait_until_ready(port: u16) -> std::result::Result<(), String> { + for attempt in 0..READINESS_ATTEMPTS { + match perform(port, "GET", "/app", None).await { + Ok((status, _)) if status < 500 => return Ok(()), + // Connection refused while the server is still binding its port is + // expected for the first attempts. + _ => {} + } + if attempt + 1 < READINESS_ATTEMPTS { + tokio::time::sleep(READINESS_INTERVAL).await; + } + } + Err("OpenCode's server never became reachable on its loopback port.".to_string()) +} + +/// Perform one request on its own connection and read the whole response. +async fn perform( + port: u16, + method: &str, + path: &str, + body: Option<&Value>, +) -> std::io::Result<(u16, Value)> { + let mut stream = TcpStream::connect(address(port)).await?; + stream.set_nodelay(true).ok(); + let encoded = body.map(ToString::to_string).unwrap_or_default(); + let mut head = format!( + "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nAccept: application/json\r\nConnection: close\r\n" + ); + if body.is_some() { + head.push_str("Content-Type: application/json\r\n"); + } + head.push_str(&format!("Content-Length: {}\r\n\r\n", encoded.len())); + stream.write_all(head.as_bytes()).await?; + if !encoded.is_empty() { + stream.write_all(encoded.as_bytes()).await?; + } + stream.flush().await?; + + let mut reader = BufReader::new(stream); + let (status, framing) = read_head(&mut reader).await?; + let body = read_body(&mut reader, framing).await?; + let body = if body.trim().is_empty() { + Value::Null + } else { + serde_json::from_str(&body).unwrap_or(Value::Null) + }; + Ok((status, body)) +} + +/// Subscribe to the event stream and forward every payload as one frame. +/// +/// The connection dying is how a `opencode serve` process that exited +/// mid-turn becomes visible: the kernel closes its sockets, this read fails, +/// and the turn is failed with a real diagnostic instead of hanging until the +/// turn deadline. +async fn stream_events( + port: u16, + path: String, + outgoing: std::sync::Arc>>, +) { + let stream = match TcpStream::connect(address(port)).await { + Ok(stream) => stream, + Err(error) => { + emit_error( + &outgoing, + &format!("Couldn't open OpenCode's event stream: {error}"), + ) + .await; + return; + } + }; + stream.set_nodelay(true).ok(); + let mut reader = BufReader::new(stream); + let head = format!( + "GET {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nAccept: text/event-stream\r\nCache-Control: no-cache\r\n\r\n" + ); + { + let inner = reader.get_mut(); + if let Err(error) = inner.write_all(head.as_bytes()).await { + emit_error( + &outgoing, + &format!("Couldn't request OpenCode's event stream: {error}"), + ) + .await; + return; + } + let _ = inner.flush().await; + } + let framing = match read_head(&mut reader).await { + Ok((status, framing)) if status == 200 => framing, + Ok((status, _)) => { + emit_error( + &outgoing, + &format!("OpenCode's event stream returned HTTP {status}."), + ) + .await; + return; + } + Err(error) => { + emit_error( + &outgoing, + &format!("Couldn't read OpenCode's event stream: {error}"), + ) + .await; + return; + } + }; + + let mut body = BodyReader::new(&mut reader, framing); + let mut payload = String::new(); + loop { + match body.read_line().await { + Ok(Some(line)) => { + let line = line.trim_end_matches(['\r', '\n']); + if let Some(data) = line.strip_prefix("data:") { + if payload.len() + data.len() > MAX_EVENT_BYTES { + emit_error(&outgoing, "An OpenCode event exceeded the size limit.").await; + return; + } + payload.push_str(data.trim_start()); + } else if line.is_empty() && !payload.is_empty() { + // A blank line terminates one SSE event. + let event = serde_json::from_str::(&payload).unwrap_or(Value::Null); + payload.clear(); + if !event.is_null() + && emit(&outgoing, &json!({"type": FRAME_EVENT, "event": event})) + .await + .is_err() + { + return; + } + } + } + Ok(None) => { + emit_error( + &outgoing, + "OpenCode's event stream ended before the turn finished.", + ) + .await; + return; + } + Err(error) => { + emit_error( + &outgoing, + &format!("OpenCode's event stream failed: {error}"), + ) + .await; + return; + } + } + } +} + +/// How a response body is delimited. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Framing { + /// Exactly this many bytes follow. + Length(usize), + /// `Transfer-Encoding: chunked`. + Chunked, + /// Read until the connection closes. + ToClose, +} + +/// Read a status line and headers, returning the status and body framing. +async fn read_head(reader: &mut BufReader) -> std::io::Result<(u16, Framing)> +where + R: AsyncRead + Unpin, +{ + let mut status_line = String::new(); + if reader.read_line(&mut status_line).await? == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "the server closed the connection before responding", + )); + } + let status = status_line + .split_whitespace() + .nth(1) + .and_then(|code| code.parse::().ok()) + .ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "the server sent an unreadable status line", + ) + })?; + + let mut framing = Framing::ToClose; + loop { + let mut header = String::new(); + if reader.read_line(&mut header).await? == 0 { + break; + } + let header = header.trim_end_matches(['\r', '\n']); + if header.is_empty() { + break; + } + let Some((name, value)) = header.split_once(':') else { + continue; + }; + let (name, value) = (name.trim().to_ascii_lowercase(), value.trim()); + if name == "content-length" { + if let Ok(length) = value.parse::() { + framing = Framing::Length(length.min(MAX_RESPONSE_BYTES)); + } + } else if name == "transfer-encoding" && value.to_ascii_lowercase().contains("chunked") { + framing = Framing::Chunked; + } + } + Ok((status, framing)) +} + +/// Read a complete, bounded response body. +async fn read_body(reader: &mut BufReader, framing: Framing) -> std::io::Result +where + R: AsyncRead + Unpin, +{ + match framing { + Framing::Length(length) => { + let mut buffer = vec![0_u8; length]; + reader.read_exact(&mut buffer).await?; + Ok(String::from_utf8_lossy(&buffer).into_owned()) + } + Framing::ToClose => { + let mut buffer = Vec::new(); + reader + .take(MAX_RESPONSE_BYTES as u64) + .read_to_end(&mut buffer) + .await?; + Ok(String::from_utf8_lossy(&buffer).into_owned()) + } + Framing::Chunked => { + let mut body = BodyReader::new(reader, framing); + let mut collected = String::new(); + while let Some(line) = body.read_line().await? { + if collected.len() + line.len() > MAX_RESPONSE_BYTES { + break; + } + collected.push_str(&line); + } + Ok(collected) + } + } +} + +/// Line reader that transparently removes chunked-transfer framing. +/// +/// Server-Sent Events are always delivered chunked, so the interleaved +/// hex-length prefixes and their trailing `CRLF`s have to be stripped before +/// anything can look for `data:` lines. Scanning the raw socket for that +/// prefix instead would work right up until a chunk boundary split an event. +struct BodyReader<'a, R> { + reader: &'a mut BufReader, + framing: Framing, + /// Bytes still owed by the chunk currently being read. + remaining: usize, + finished: bool, +} + +impl<'a, R> BodyReader<'a, R> +where + R: AsyncRead + Unpin, +{ + fn new(reader: &'a mut BufReader, framing: Framing) -> Self { + let remaining = match framing { + Framing::Length(length) => length, + _ => 0, + }; + Self { + reader, + framing, + remaining, + finished: false, + } + } + + /// Next logical body line, including its terminator, or `None` at the end. + async fn read_line(&mut self) -> std::io::Result> { + if self.finished { + return Ok(None); + } + match self.framing { + Framing::Chunked => self.read_chunked_line().await, + _ => { + let mut line = String::new(); + if self.reader.read_line(&mut line).await? == 0 { + self.finished = true; + return Ok(None); + } + if let Framing::Length(_) = self.framing { + self.remaining = self.remaining.saturating_sub(line.len()); + if self.remaining == 0 { + self.finished = true; + } + } + Ok(Some(line)) + } + } + } + + /// Assemble one line from however many chunks it spans. + async fn read_chunked_line(&mut self) -> std::io::Result> { + let mut line = Vec::new(); + loop { + if self.remaining == 0 { + if !self.next_chunk_header().await? { + self.finished = true; + return Ok((!line.is_empty()) + .then(|| String::from_utf8_lossy(&line).into_owned())); + } + } + let mut byte = [0_u8; 1]; + self.reader.read_exact(&mut byte).await?; + self.remaining -= 1; + line.push(byte[0]); + if self.remaining == 0 { + // Consume the CRLF that terminates the chunk itself. + let mut terminator = [0_u8; 2]; + self.reader.read_exact(&mut terminator).await?; + } + if byte[0] == b'\n' { + return Ok(Some(String::from_utf8_lossy(&line).into_owned())); + } + if line.len() > MAX_EVENT_BYTES { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "a response line exceeded the size limit", + )); + } + } + } + + /// Read the next chunk size line. Returns `false` on the terminating chunk. + async fn next_chunk_header(&mut self) -> std::io::Result { + let mut header = String::new(); + if self.reader.read_line(&mut header).await? == 0 { + return Ok(false); + } + let header = header.trim(); + if header.is_empty() { + // Tolerate the stray CRLF some servers emit between chunks. + return Box::pin(self.next_chunk_header()).await; + } + let size = header.split(';').next().unwrap_or("0"); + let size = usize::from_str_radix(size.trim(), 16).map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "the server sent an unreadable chunk size", + ) + })?; + if size == 0 { + return Ok(false); + } + self.remaining = size; + Ok(true) + } +} + +async fn emit( + outgoing: &std::sync::Arc>, + frame: &Value, +) -> std::io::Result<()> +where + W: AsyncWrite + Unpin, +{ + let mut bytes = serde_json::to_vec(frame).unwrap_or_else(|_| b"{}".to_vec()); + bytes.push(b'\n'); + let mut guard = outgoing.lock().await; + guard.write_all(&bytes).await?; + guard.flush().await +} + +async fn emit_error(outgoing: &std::sync::Arc>, message: &str) +where + W: AsyncWrite + Unpin, +{ + let _ = emit(outgoing, &json!({"type": FRAME_ERROR, "message": message})).await; +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::AsyncBufReadExt; + + /// Feed a canned HTTP response through the head/body readers. + async fn parse(response: &[u8]) -> (u16, String) { + let mut reader = BufReader::new(std::io::Cursor::new(response.to_vec())); + let (status, framing) = read_head(&mut reader).await.unwrap(); + let body = read_body(&mut reader, framing).await.unwrap(); + (status, body) + } + + #[tokio::test] + async fn reads_a_content_length_response() { + let (status, body) = + parse(b"HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\n{\"id\":\"abc\"}\r\n").await; + assert_eq!(status, 200); + assert!(body.starts_with("{\"id\":\"abc\"}")); + } + + #[tokio::test] + async fn reads_a_chunked_response() { + let (status, body) = parse( + b"HTTP/1.1 201 Created\r\nTransfer-Encoding: chunked\r\n\r\n4\r\n{\"a\"\r\n4\r\n:1}\n\r\n0\r\n\r\n", + ) + .await; + assert_eq!(status, 201); + assert_eq!(body.trim(), "{\"a\":1}"); + } + + /// The case a naive scan for `data:` would get wrong: one SSE event split + /// across two chunks. + #[tokio::test] + async fn reassembles_an_event_split_across_chunk_boundaries() { + let raw = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\ + 8\r\ndata: {\"\r\n\ + 13\r\ntype\":\"session.idle\r\n\ + 4\r\n\"}\n\n\r\n\ + 0\r\n\r\n"; + let mut reader = BufReader::new(std::io::Cursor::new(raw.to_vec())); + let (_, framing) = read_head(&mut reader).await.unwrap(); + let mut body = BodyReader::new(&mut reader, framing); + + let first = body.read_line().await.unwrap().unwrap(); + assert_eq!(first.trim_end(), "data: {\"type\":\"session.idle\"}"); + } + + #[tokio::test] + async fn an_unreadable_chunk_size_is_an_error_not_a_silent_truncation() { + let mut reader = BufReader::new(std::io::Cursor::new( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\nzz\r\n".to_vec(), + )); + let (_, framing) = read_head(&mut reader).await.unwrap(); + let mut body = BodyReader::new(&mut reader, framing); + assert!(body.read_line().await.is_err()); + } + + /// End-to-end against a real socket: the bridge must report readiness, + /// perform a correlated request, and forward an SSE payload. + #[tokio::test] + async fn drives_a_real_server_through_the_frame_contract() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let mut reader = BufReader::new(&mut socket); + let mut request = String::new(); + reader.read_line(&mut request).await.unwrap(); + // Drain headers. + loop { + let mut header = String::new(); + if reader.read_line(&mut header).await.unwrap() == 0 + || header.trim().is_empty() + { + break; + } + } + let response: &[u8] = if request.contains("/event") { + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\ + 1E\r\ndata: {\"type\":\"session.idle\"}\n\r\n\ + 1\r\n\n\r\n" + } else if request.starts_with("POST") { + b"HTTP/1.1 200 OK\r\nContent-Length: 14\r\n\r\n{\"ok\":\"yes\"}\r\n" + } else { + b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}" + }; + let _ = socket.write_all(response).await; + let _ = socket.flush().await; + }); + } + }); + + let streams = connect(port); + let mut writer = streams.writer; + let mut lines = BufReader::new(streams.reader).lines(); + + let ready: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + assert_eq!(ready["type"], json!(FRAME_READY)); + + writer + .write_all(b"{\"id\":7,\"method\":\"POST\",\"path\":\"/session\",\"body\":{}}\n") + .await + .unwrap(); + writer.flush().await.unwrap(); + let response: Value = + serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + assert_eq!(response["type"], json!(FRAME_RESPONSE)); + assert_eq!(response["id"], json!(7)); + assert_eq!(response["body"]["ok"], json!("yes")); + + writer + .write_all(b"{\"method\":\"SUBSCRIBE\",\"path\":\"/event\"}\n") + .await + .unwrap(); + writer.flush().await.unwrap(); + let event: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + assert_eq!(event["type"], json!(FRAME_EVENT)); + assert_eq!(event["event"]["type"], json!("session.idle")); + } + + /// A server that dies mid-turn must surface as a frame, not a hang. + #[tokio::test] + async fn a_server_that_disappears_is_reported_rather_than_awaited() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut reader = BufReader::new(&mut socket); + let mut request = String::new(); + let _ = reader.read_line(&mut request).await; + loop { + let mut header = String::new(); + if reader.read_line(&mut header).await.unwrap_or(0) == 0 + || header.trim().is_empty() + { + break; + } + } + if request.contains("/event") { + // Accept the subscription, then drop the connection the + // way a crashing process would. + let _ = socket + .write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n") + .await; + let _ = socket.flush().await; + drop(socket); + return; + } + let _ = socket + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}") + .await; + } + }); + + let streams = connect(port); + let mut writer = streams.writer; + let mut lines = BufReader::new(streams.reader).lines(); + assert!(lines.next_line().await.unwrap().is_some(), "ready frame"); + + writer + .write_all(b"{\"method\":\"SUBSCRIBE\",\"path\":\"/event\"}\n") + .await + .unwrap(); + writer.flush().await.unwrap(); + + let frame: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + assert_eq!(frame["type"], json!(FRAME_ERROR)); + } +} From 64c67c487c3b4ec9b2e53606f92394a023af855b Mon Sep 17 00:00:00 2001 From: David Viejo Date: Mon, 21 Sep 2026 18:45:18 +0200 Subject: [PATCH 3/6] feat(opencode): drive turns through opencode serve with enforced permissions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `opencode run --format json` has no permission enforcement an application can rely on. The only flags it accepts are `--auto`, which approves everything, and `--agent plan`; every other policy comes from whatever `opencode` configuration happens to exist on the machine. A caller could not ask for "prompt before running a shell command", could not be told that a tool call was refused, and — worst of all — a silently auto-refused edit looked exactly like a successful turn that produced no text. Add `OpenCode::serve()`, a second turn mode that starts `opencode serve` on a reserved loopback port and drives it over HTTP and SSE. The turn's policy is supplied through `OPENCODE_CONFIG_CONTENT`, which the server reads instead of the ambient configuration, so the policy an application asked for is the one the harness runs under. Anything marked `ask` arrives as a live approval and is answered on the native permission endpoint. Mapping onto OpenCode's two permission axes: Default/Custom edit: ask bash: ask AcceptEdits edit: allow bash: ask FullAccess edit: allow bash: allow Plan edit: deny bash: deny Reviewing an edit you were never asked about is not a review, hence AcceptEdits is "write freely, ask before a shell command". Plan denies both categories rather than selecting the planning agent: read-only tools are gated by neither, so a plan turn can still look around but can never have a side effect. An explicitly empty tool allowlist becomes a `{"*": "deny"}` wildcard, because an empty tool set has to be an enforcement boundary rather than a suggestion in the prompt. A plan turn and an empty allowlist also refuse any permission that reaches the adapter without consulting the application. Both are already denied by the configuration the server started with, so arriving there means the policy did not hold — and the turn promised the user no such choice would exist. `Run` mode is untouched and still reports `live_approvals: false`. Also carried over from the reference driver: subscribing before prompting so early events cannot be missed, resolving a resumed session by id rather than by listing and filtering (which fails closed on a symlinked workspace), failing fast on a subagent session instead of hanging until the deadline, treating an idle turn that produced nothing as a failure, and per-turn stdio and HTTP MCP servers whose credentials are referenced by variable name and never serialized into the configuration. --- src/providers/mod.rs | 4 +- src/providers/opencode.rs | 177 ++++- src/providers/opencode_http.rs | 93 ++- src/providers/opencode_serve.rs | 1316 +++++++++++++++++++++++++++++++ 4 files changed, 1546 insertions(+), 44 deletions(-) create mode 100644 src/providers/opencode_serve.rs diff --git a/src/providers/mod.rs b/src/providers/mod.rs index adfad75..cecd75a 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -10,13 +10,15 @@ mod codex_app_server; mod opencode; #[cfg(feature = "opencode")] mod opencode_http; +#[cfg(feature = "opencode")] +mod opencode_serve; #[cfg(feature = "claude")] pub use claude::Claude; #[cfg(feature = "codex")] pub use codex::{Codex, CodexTurnMode}; #[cfg(feature = "opencode")] -pub use opencode::OpenCode; +pub use opencode::{OpenCode, OpenCodeTurnMode}; #[cfg(any(feature = "claude", feature = "codex"))] use serde_json::Value; diff --git a/src/providers/opencode.rs b/src/providers/opencode.rs index e14fe58..2648173 100644 --- a/src/providers/opencode.rs +++ b/src/providers/opencode.rs @@ -14,10 +14,43 @@ use crate::{ RuntimeError, ToolCallStatus, TurnEvent, TurnRequest, }; -/// OpenCode CLI adapter using `opencode run --format json`. +/// Transport used to run one OpenCode turn. +/// +/// Both modes produce the same normalized [`TurnEvent`] stream. They differ in +/// whether the turn's permission policy is something the application controls. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[non_exhaustive] +pub enum OpenCodeTurnMode { + /// One-shot `opencode run --format json`. + /// + /// Output only, and with no permission enforcement an application can + /// rely on: the CLI accepts `--auto` (approve everything) and + /// `--agent plan`, so every other policy comes from whatever `opencode` + /// configuration happens to exist on the machine. A caller cannot request + /// "ask before running a shell command" here, nor learn that a tool call + /// was refused. + #[default] + Run, + /// `opencode serve`, driven over HTTP and Server-Sent Events. + /// + /// The permission policy is supplied per turn through + /// `OPENCODE_CONFIG_CONTENT`, which the server reads instead of the + /// ambient configuration, so the policy an application asked for is the + /// one the harness actually runs under. Anything the policy marks `ask` + /// arrives as a live approval. + /// + /// The server is reached on the SDK host's loopback interface, so this + /// mode needs a transport that runs the provider on that host. Using it + /// with a remote transport would require the port to be forwarded back, + /// which the SDK does not arrange. + Serve, +} + +/// OpenCode CLI adapter using `opencode run --format json` or `opencode serve`. #[derive(Debug, Clone, Default)] pub struct OpenCode { executable: Option, + turn_mode: OpenCodeTurnMode, } impl OpenCode { @@ -25,9 +58,54 @@ impl OpenCode { pub fn with_executable(path: impl Into) -> Self { Self { executable: Some(path.into()), + turn_mode: OpenCodeTurnMode::default(), } } + /// Drive turns through `opencode serve` instead of `opencode run`. + /// + /// This is the mode to use when the application — rather than whatever + /// configuration exists on the machine — must decide what the agent is + /// allowed to do. + pub fn serve() -> Self { + Self::default().with_turn_mode(OpenCodeTurnMode::Serve) + } + + /// Select the transport used for turns. + pub fn with_turn_mode(mut self, mode: OpenCodeTurnMode) -> Self { + self.turn_mode = mode; + self + } + + /// Transport this adapter uses for turns. + pub fn turn_mode(&self) -> OpenCodeTurnMode { + self.turn_mode + } + + fn serve_mode(&self) -> bool { + self.turn_mode == OpenCodeTurnMode::Serve + } + + /// Build the `opencode serve` invocation for one turn. + fn serve_command(&self, request: &TurnRequest, port: u16) -> Result { + let mut spec = CommandSpec::new(self.configured_executable()); + spec.args.extend([ + "serve".into(), + "--hostname".into(), + "127.0.0.1".into(), + "--port".into(), + port.to_string().into(), + ]); + // The policy travels in the environment rather than argv because it is + // this turn's entire enforcement boundary, and `clear_environment` + // means nothing reaches the child that was not put here deliberately. + spec.environment.insert( + "OPENCODE_CONFIG_CONTENT".into(), + super::opencode_serve::permission_config(request)?.into(), + ); + Ok(spec) + } + fn resolved(&self) -> Option { resolve_executable(self.executable.as_ref(), "opencode") } @@ -52,15 +130,41 @@ impl AgentAdapter for OpenCode { fn permission_support(&self) -> PermissionSupport { PermissionSupport { default: true, - accept_edits: false, - plan: false, + // `opencode run` has no flag that approves only edits; the served + // policy expresses it as `edit: allow, bash: ask`. + accept_edits: self.serve_mode(), + // `--agent plan` selects a planning agent but does not guarantee + // the absence of side effects. The served policy denies both + // permission categories outright, which does. + plan: self.serve_mode(), full_access: true, custom: true, - live_approvals: false, + // `opencode run` resolves permissions itself from whatever + // configuration it finds and has no channel back into a running + // turn; the served transport answers `permission.asked` over HTTP. + live_approvals: self.serve_mode(), + // OpenCode has no question channel in either mode. live_questions: false, } } + fn launch_context_capabilities(&self) -> crate::LaunchContextCapabilities { + if !self.serve_mode() { + return crate::LaunchContextCapabilities::default(); + } + crate::LaunchContextCapabilities { + // Neither is a native field: OpenCode has no system-prompt or + // tool-restriction input on its prompt body, so both are carried + // as a prompt prefix. An *empty* allowlist is different — it is + // enforced by a wildcard deny rule in the served policy. + system_prompt_append: true, + allowed_tools: true, + stdio_mcp: true, + http_mcp: true, + ..crate::LaunchContextCapabilities::default() + } + } + fn control_groups(&self) -> Vec { vec![ HarnessControlGroup { @@ -170,6 +274,47 @@ impl AgentAdapter for OpenCode { inspect_executable(Provider::OpenCode, self.resolved()).await } + fn prepare_turn(&self, request: &TurnRequest, state: &mut AdapterState) -> Result<()> { + if !self.serve_mode() { + return Ok(()); + } + // Bind a throwaway listener so the OS picks a free port, then drop it + // so the server can bind the same one. `opencode serve --port 0` does + // not do this: it falls back to its fixed default port instead. + let port = std::net::TcpListener::bind(("127.0.0.1", 0)) + .and_then(|listener| listener.local_addr()) + .map(|address| address.port()) + .map_err(|error| RuntimeError::Protocol { + provider: Provider::OpenCode, + message: format!("could not reserve a loopback port for OpenCode: {error}"), + })?; + super::opencode_serve::prepare_turn(request, state, port); + Ok(()) + } + + fn command_for_turn(&self, request: &TurnRequest, state: &AdapterState) -> Result { + match super::opencode_serve::turn_port(state) { + Some(port) if self.serve_mode() => self.serve_command(request, port), + _ => self.command(request), + } + } + + async fn attach( + &self, + _request: &TurnRequest, + state: &AdapterState, + ) -> Result> { + Ok(super::opencode_serve::turn_port(state) + .filter(|_| self.serve_mode()) + .map(super::opencode_http::connect)) + } + + fn interrupt_request(&self, state: &AdapterState) -> Option> { + self.serve_mode() + .then(|| super::opencode_serve::interrupt(state)) + .flatten() + } + fn command(&self, request: &TurnRequest) -> Result { let mut spec = CommandSpec::new(self.configured_executable()); spec.args.push("run".into()); @@ -229,6 +374,9 @@ impl AgentAdapter for OpenCode { } fn parse_line(&self, line: &str, state: &mut AdapterState) -> Result { + if self.serve_mode() { + return super::opencode_serve::parse_line(line, state); + } let value: Value = serde_json::from_str(line).map_err(|error| RuntimeError::Protocol { provider: Provider::OpenCode, message: format!("invalid JSON event: {error}"), @@ -387,18 +535,25 @@ impl AgentAdapter for OpenCode { fn approval_response( &self, _request: &ApprovalRequest, - _original: &Value, - _decision: ApprovalDecision, + original: &Value, + decision: ApprovalDecision, ) -> Result>> { - Ok(None) + if !self.serve_mode() { + return Ok(None); + } + Ok(Some(super::opencode_serve::approval_response( + original, decision, + )?)) } fn question_response( &self, - _request: &QuestionRequest, - _original: &Value, - _answer: Option, + request: &QuestionRequest, + original: &Value, + answer: Option, ) -> Result>> { - Ok(None) + Ok(super::opencode_serve::question_response( + request, original, answer, + )) } } diff --git a/src/providers/opencode_http.rs b/src/providers/opencode_http.rs index 5f57a18..7d748d5 100644 --- a/src/providers/opencode_http.rs +++ b/src/providers/opencode_http.rs @@ -77,6 +77,12 @@ pub(super) const FRAME_READY: &str = "@ready"; pub(super) const FRAME_RESPONSE: &str = "@response"; pub(super) const FRAME_EVENT: &str = "@event"; pub(super) const FRAME_ERROR: &str = "@error"; +/// Emitted once the event stream is accepted and before any payload. +/// +/// The prompt must not be sent until this arrives: the reference driver +/// subscribes first precisely so nothing emitted in a turn's first moments is +/// missed, and only the bridge can know when the stream is actually open. +pub(super) const FRAME_SUBSCRIBED: &str = "@subscribed"; /// Start the bridge for a turn and hand the runtime its protocol streams. /// @@ -124,7 +130,11 @@ async fn run_bridge(port: u16, bridge_side: DuplexStream) { continue; } let Ok(command) = serde_json::from_str::(&line) else { - emit_error(&outgoing, "the OpenCode bridge received an unreadable command").await; + emit_error( + &outgoing, + "the OpenCode bridge received an unreadable command", + ) + .await; break; }; if command.method == "SUBSCRIBE" { @@ -187,13 +197,17 @@ async fn perform( let mut stream = TcpStream::connect(address(port)).await?; stream.set_nodelay(true).ok(); let encoded = body.map(ToString::to_string).unwrap_or_default(); - let mut head = format!( - "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nAccept: application/json\r\nConnection: close\r\n" + let content_type = if body.is_some() { + "Content-Type: application/json\r\n" + } else { + "" + }; + let head = format!( + "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n\ + Accept: application/json\r\nConnection: close\r\n{content_type}\ + Content-Length: {}\r\n\r\n", + encoded.len() ); - if body.is_some() { - head.push_str("Content-Type: application/json\r\n"); - } - head.push_str(&format!("Content-Length: {}\r\n\r\n", encoded.len())); stream.write_all(head.as_bytes()).await?; if !encoded.is_empty() { stream.write_all(encoded.as_bytes()).await?; @@ -251,7 +265,7 @@ async fn stream_events( let _ = inner.flush().await; } let framing = match read_head(&mut reader).await { - Ok((status, framing)) if status == 200 => framing, + Ok((200, framing)) => framing, Ok((status, _)) => { emit_error( &outgoing, @@ -270,6 +284,13 @@ async fn stream_events( } }; + if emit(&outgoing, &json!({"type": FRAME_SUBSCRIBED})) + .await + .is_err() + { + return; + } + let mut body = BodyReader::new(&mut reader, framing); let mut payload = String::new(); loop { @@ -443,35 +464,30 @@ where if self.finished { return Ok(None); } - match self.framing { - Framing::Chunked => self.read_chunked_line().await, - _ => { - let mut line = String::new(); - if self.reader.read_line(&mut line).await? == 0 { - self.finished = true; - return Ok(None); - } - if let Framing::Length(_) = self.framing { - self.remaining = self.remaining.saturating_sub(line.len()); - if self.remaining == 0 { - self.finished = true; - } - } - Ok(Some(line)) + if self.framing == Framing::Chunked { + return self.read_chunked_line().await; + } + let mut line = String::new(); + if self.reader.read_line(&mut line).await? == 0 { + self.finished = true; + return Ok(None); + } + if let Framing::Length(_) = self.framing { + self.remaining = self.remaining.saturating_sub(line.len()); + if self.remaining == 0 { + self.finished = true; } } + Ok(Some(line)) } /// Assemble one line from however many chunks it spans. async fn read_chunked_line(&mut self) -> std::io::Result> { let mut line = Vec::new(); loop { - if self.remaining == 0 { - if !self.next_chunk_header().await? { - self.finished = true; - return Ok((!line.is_empty()) - .then(|| String::from_utf8_lossy(&line).into_owned())); - } + if self.remaining == 0 && !self.next_chunk_header().await? { + self.finished = true; + return Ok((!line.is_empty()).then(|| String::from_utf8_lossy(&line).into_owned())); } let mut byte = [0_u8; 1]; self.reader.read_exact(&mut byte).await?; @@ -642,7 +658,8 @@ mod tests { let mut writer = streams.writer; let mut lines = BufReader::new(streams.reader).lines(); - let ready: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + let ready: Value = + serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); assert_eq!(ready["type"], json!(FRAME_READY)); writer @@ -661,7 +678,15 @@ mod tests { .await .unwrap(); writer.flush().await.unwrap(); - let event: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + let subscribed: Value = + serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + assert_eq!( + subscribed["type"], + json!(FRAME_SUBSCRIBED), + "the stream must be confirmed open before the prompt is sent" + ); + let event: Value = + serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); assert_eq!(event["type"], json!(FRAME_EVENT)); assert_eq!(event["event"]["type"], json!("session.idle")); } @@ -711,7 +736,11 @@ mod tests { .unwrap(); writer.flush().await.unwrap(); - let frame: Value = serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + let subscribed: Value = + serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + assert_eq!(subscribed["type"], json!(FRAME_SUBSCRIBED)); + let frame: Value = + serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); assert_eq!(frame["type"], json!(FRAME_ERROR)); } } diff --git a/src/providers/opencode_serve.rs b/src/providers/opencode_serve.rs new file mode 100644 index 0000000..4df5c80 --- /dev/null +++ b/src/providers/opencode_serve.rs @@ -0,0 +1,1316 @@ +//! Bidirectional OpenCode transport built on `opencode serve`. +//! +//! `opencode run --format json` streams a turn one way and, crucially, has no +//! permission enforcement the SDK can rely on: the flags it accepts are +//! `--auto` (approve everything) and `--agent plan`, so every other policy is +//! whatever the user's own `opencode` configuration happens to say. An +//! application cannot ask for "prompt before running a shell command" and be +//! told what actually happened. +//! +//! `opencode serve` fixes both halves. Policy is supplied per turn through +//! `OPENCODE_CONFIG_CONTENT`, which the server reads instead of the ambient +//! configuration, and anything the policy marks `ask` is surfaced as a +//! `permission.asked` event answered over HTTP. That makes permission +//! enforcement a property of the turn rather than of the machine it runs on. +//! +//! This module is the synchronous half of that transport. It sequences the +//! whole protocol — session, subscription, prompt, events, permission answers +//! — as one state machine over newline-delimited frames, exactly like +//! [`super::codex_app_server`]. [`super::opencode_http`] moves the bytes. +//! +//! The wire shapes are those of the reference driver, which read them off a +//! live `opencode 1.4.3 serve` process rather than from the published types: +//! notably the permission event really is `permission.asked`, while the +//! shipped SDK types still declare `permission.updated`. Events are therefore +//! read as loosely-typed JSON, which survives that class of drift by +//! construction. + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use crate::adapter::{AdapterOutput, AdapterState, InteractionRequest}; +use crate::error::classify_provider_failure; +use crate::lifecycle::DeliveryState; +use crate::{ + ApprovalDecision, ApprovalRequest, PermissionMode, Provider, ProviderTerminalFailure, + QuestionAnswer, QuestionRequest, Result, RunStatus, RuntimeError, ToolCallStatus, TurnEvent, + TurnRequest, +}; + +use super::opencode_http::{ + FRAME_ERROR, FRAME_EVENT, FRAME_READY, FRAME_RESPONSE, FRAME_SUBSCRIBED, +}; + +/// Key under which this transport keeps its per-turn protocol state. +const STATE_KEY: &str = "opencode.serve"; + +/// Correlation identifiers for the bridge requests this module issues. +const ID_SESSION: u64 = 1; +const ID_PROMPT: u64 = 2; + +/// Tool output is bounded to the same budget the other adapters use. +const MAX_TOOL_OUTPUT_CHARS: usize = 4096; + +/// Per-turn protocol state retained between frames. +#[derive(Debug, Default, Serialize, Deserialize)] +#[serde(default)] +pub(super) struct TurnState { + /// Loopback port the `opencode serve` child was told to bind. + pub port: u16, + /// Workspace passed as the `directory` query parameter. + directory: String, + /// Session this turn resumes, if any. + resume: Option, + /// Message parts sent as the prompt. + parts: Vec, + /// Native `{providerID, modelID}` selection, when the caller pinned one. + model: Option, + /// Session being prompted. + session_id: Option, + /// Whether every permission must be refused without consulting the + /// application, because the turn asked for a read-only plan. + plan_mode: bool, + /// Whether an explicitly empty tool allowlist denied every tool. + tools_denied: bool, + /// Message id to role, so only assistant output is streamed. + roles: std::collections::BTreeMap, + /// Part id to the text already emitted for it. + emitted: std::collections::BTreeMap, + /// Part id to `text` or `reasoning`. + part_kinds: std::collections::BTreeMap, + /// Last error reported by the session. + error_message: Option, + /// Whether any assistant text or tool call was observed. + saw_activity: bool, +} + +fn load(state: &AdapterState) -> TurnState { + state + .extensions + .get(STATE_KEY) + .cloned() + .and_then(|value| serde_json::from_value(value).ok()) + .unwrap_or_default() +} + +fn store(state: &mut AdapterState, turn: &TurnState) { + if let Ok(value) = serde_json::to_value(turn) { + state.extensions.insert(STATE_KEY.to_string(), value); + } +} + +/// Read back the port chosen for this turn, for `command_for_turn`. +pub(super) fn turn_port(state: &AdapterState) -> Option { + let port = load(state).port; + (port != 0).then_some(port) +} + +fn protocol(message: impl Into) -> RuntimeError { + RuntimeError::Protocol { + provider: Provider::OpenCode, + message: message.into(), + } +} + +fn encode(value: &Value) -> Result> { + serde_json::to_vec(value) + .map_err(|error| protocol(format!("could not encode an OpenCode request: {error}"))) +} + +/// Percent-encode a query-parameter value. +/// +/// A workspace path can contain spaces, `#`, `&` or `?`, any of which would +/// otherwise change which directory the server is told to use. +fn query_escape(value: &str) -> String { + value + .bytes() + .map(|byte| match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + (byte as char).to_string() + } + _ => format!("%{byte:02X}"), + }) + .collect() +} + +fn get(id: u64, path: String) -> Value { + json!({"id": id, "method": "GET", "path": path}) +} + +fn post(id: Option, path: String, body: Value) -> Value { + json!({"id": id, "method": "POST", "path": path, "body": body}) +} + +/// Permission policy for one turn, as `OPENCODE_CONFIG_CONTENT` expects it. +/// +/// OpenCode's only permission levers are the blanket `edit` and `bash` +/// categories, so every mode is expressed on those two axes. Plan mode denies +/// both outright: OpenCode's read-only tools are gated by neither category, so +/// a plan turn can still look around but can never write or run a shell +/// command — the same "no side effects, ever" guarantee Codex gets from a +/// read-only sandbox. +/// +/// This is the whole reason `Serve` mode exists. It is supplied per turn and +/// replaces the user's ambient configuration, so the policy an application +/// asked for is the policy the harness runs under. +pub(super) fn permission_config(request: &TurnRequest) -> Result { + let plan = is_plan_mode(request); + let (edit, bash) = if plan { + ("deny", "deny") + } else { + match &request.permission_mode { + // Reviewing an edit you were never asked about is not a review, + // so this is "write freely, ask before running a shell command". + PermissionMode::AcceptEdits => ("allow", "ask"), + PermissionMode::FullAccess => ("allow", "allow"), + PermissionMode::Default | PermissionMode::Plan | PermissionMode::Custom(_) => { + ("ask", "ask") + } + } + }; + let native = request + .harness_options + .get("permission_mode") + .map(String::as_str); + let (edit, bash) = match native { + Some("auto") => ("allow", "allow"), + Some("ask") => ("ask", "ask"), + Some("default") | None => (edit, bash), + Some(other) => { + return Err(RuntimeError::InvalidRequest { + field: "harness_options.permission_mode", + message: format!("unsupported OpenCode permission mode `{other}`"), + }) + } + }; + + let mut config = json!({"permission": {"edit": edit, "bash": bash}}); + if request + .launch_context + .allowed_tools + .as_deref() + .is_some_and(<[String]>::is_empty) + { + // An explicitly empty tool set must be an enforcement boundary, not a + // prompt-level suggestion. OpenCode's wildcard rule is the only thing + // that expresses it. + config["permission"] = json!({"*": "deny"}); + } + if let Some(servers) = mcp_config(request)? { + config["mcp"] = servers; + } + Ok(config.to_string()) +} + +/// Whether this turn is a read-only plan turn. +fn is_plan_mode(request: &TurnRequest) -> bool { + matches!(request.permission_mode, PermissionMode::Plan) + || request + .harness_options + .get("agent") + .is_some_and(|agent| agent == "plan") +} + +/// Translate turn-scoped MCP servers into OpenCode's `mcp` configuration. +/// +/// OpenCode distinguishes `local` (a subprocess speaking MCP over stdio) from +/// `remote` (an HTTP endpoint), and takes the command as an argv array, so no +/// shell quoting is involved. Credentials are never serialized here: stdio +/// servers name the harness variables to forward, and remote servers name the +/// variable each header is read from. +fn mcp_config(request: &TurnRequest) -> Result> { + let servers = &request.launch_context.mcp_servers; + if servers.is_empty() { + return Ok(None); + } + let mut configured = serde_json::Map::new(); + for (name, server) in servers { + let entry = match server { + crate::McpServerConfig::Stdio { + command, + args, + environment_from, + } => { + let Some(command) = command.to_str() else { + return Err(RuntimeError::InvalidRequest { + field: "launch_context.mcp_servers.command", + message: "stdio commands must be valid UTF-8".into(), + }); + }; + let mut argv = vec![json!(command)]; + argv.extend(args.iter().map(|argument| json!(argument))); + let mut entry = json!({"type": "local", "command": argv, "enabled": true}); + if !environment_from.is_empty() { + // OpenCode takes an environment map rather than a list of + // names to forward, so a reference is expanded by the + // server from its own environment. The value never + // appears in this configuration. + let environment = environment_from + .iter() + .map(|(target, source)| { + (target.clone(), json!(format!("{{env:{source}}}"))) + }) + .collect::>(); + entry["environment"] = Value::Object(environment); + } + entry + } + crate::McpServerConfig::Http { url, headers_from } => { + let mut entry = json!({"type": "remote", "url": url, "enabled": true}); + if !headers_from.is_empty() { + let headers = headers_from + .iter() + .map(|(header, source)| { + (header.clone(), json!(format!("{{env:{source}}}"))) + }) + .collect::>(); + entry["headers"] = Value::Object(headers); + } + entry + } + }; + configured.insert(name.clone(), entry); + } + Ok(Some(Value::Object(configured))) +} + +/// Seed the state machine from the validated request. +pub(super) fn prepare_turn(request: &TurnRequest, state: &mut AdapterState, port: u16) { + let mut parts = Vec::new(); + // The persona/system-prompt prefix is prompt-level for OpenCode: + // `OPENCODE_CONFIG_CONTENT` has no system-prompt field and the prompt body + // has no injection field, so there is nowhere else to put it. + let prompt = match system_prompt_prefix(request) { + Some(prefix) if !request.prompt.is_empty() => { + format!("{prefix}\n\n---\n\n{}", request.prompt) + } + Some(prefix) => prefix, + None => request.prompt.clone(), + }; + if !prompt.is_empty() { + parts.push(json!({"type": "text", "text": prompt})); + } + for attachment in &request.attachments { + parts.push(json!({ + "type": "file", + "mime": attachment.media_type, + "filename": attachment.display_name.clone().or_else(|| attachment + .path + .file_name() + .map(|name| name.to_string_lossy().into_owned())), + "url": format!("file://{}", attachment.path.display()), + })); + } + + store( + state, + &TurnState { + port, + directory: request.working_directory.to_string_lossy().into_owned(), + resume: request.session_id.clone(), + parts, + model: model_selection(request.model.as_deref()), + plan_mode: is_plan_mode(request), + tools_denied: request + .launch_context + .allowed_tools + .as_deref() + .is_some_and(<[String]>::is_empty), + ..TurnState::default() + }, + ); +} + +/// Render the launch context's system-prompt and tool policy as a prompt +/// prefix, which is the only channel OpenCode offers for either. +fn system_prompt_prefix(request: &TurnRequest) -> Option { + let context = &request.launch_context; + let mut sections = Vec::new(); + if let Some(instructions) = context + .system_prompt_append + .as_deref() + .map(str::trim) + .filter(|instructions| !instructions.is_empty()) + { + sections.push(instructions.to_string()); + } + // A non-empty allowlist is advisory here; an empty one is enforced by the + // wildcard deny rule in `permission_config` instead. + if let Some(tools) = context + .allowed_tools + .as_deref() + .filter(|tools| !tools.is_empty()) + { + sections.push(format!( + "Use only these tools: {}. Do not use any other tool.", + tools.join(", ") + )); + } + (!sections.is_empty()).then(|| sections.join("\n\n")) +} + +/// OpenCode takes a split `{providerID, modelID}`, not one identifier. +fn model_selection(model: Option<&str>) -> Option { + let model = model?; + let (provider, id) = model.split_once('/')?; + (!provider.is_empty() && !id.is_empty()).then(|| json!({"providerID": provider, "modelID": id})) +} + +/// Encode a cooperative abort for a turn whose session is known. +pub(super) fn interrupt(state: &AdapterState) -> Option> { + let turn = load(state); + let session = turn.session_id?; + encode(&post( + None, + format!( + "/session/{}/abort?directory={}", + query_escape(&session), + query_escape(&turn.directory) + ), + json!({}), + )) + .ok() +} + +/// Translate one bridge frame. +pub(super) fn parse_line(line: &str, state: &mut AdapterState) -> Result { + let value: Value = serde_json::from_str(line) + .map_err(|error| protocol(format!("invalid bridge frame: {error}")))?; + let mut turn = load(state); + let mut output = AdapterOutput::default(); + match value + .get("type") + .and_then(Value::as_str) + .unwrap_or_default() + { + FRAME_READY => ready(&turn, &mut output)?, + FRAME_SUBSCRIBED => prompt(&turn, &mut output)?, + FRAME_RESPONSE => response(&value, &mut turn, state, &mut output)?, + FRAME_EVENT => { + if let Some(event) = value.get("event") { + self::event(event, &mut turn, state, &mut output); + } + } + FRAME_ERROR => { + let message = value + .get("message") + .and_then(Value::as_str) + .unwrap_or("The OpenCode server became unreachable."); + fail(&mut turn, state, &mut output, message, None); + } + _ => {} + } + store(state, &turn); + Ok(output) +} + +/// Resolve the session as soon as the server answers. +fn ready(turn: &TurnState, output: &mut AdapterOutput) -> Result<()> { + let directory = query_escape(&turn.directory); + let frame = match turn.resume.as_deref() { + // `session.get` finds a session by id without needing the directory to + // match, so a workspace reached through a symlink cannot fail closed + // on a spurious "not found" the way listing and filtering would. + Some(resume) => get( + ID_SESSION, + format!("/session/{}?directory={directory}", query_escape(resume)), + ), + None => post( + Some(ID_SESSION), + format!("/session?directory={directory}"), + json!({}), + ), + }; + output.writes.push(encode(&frame)?); + Ok(()) +} + +/// Send the prompt once the event stream is confirmed open. +fn prompt(turn: &TurnState, output: &mut AdapterOutput) -> Result<()> { + let Some(session) = turn.session_id.as_deref() else { + return Ok(()); + }; + let mut body = json!({"parts": turn.parts}); + if let Some(model) = &turn.model { + body["model"] = model.clone(); + } + output.writes.push(encode(&post( + Some(ID_PROMPT), + format!( + "/session/{}/message?directory={}", + query_escape(session), + query_escape(&turn.directory) + ), + body, + ))?); + Ok(()) +} + +/// Advance the handshake with the answer to one of our own requests. +fn response( + value: &Value, + turn: &mut TurnState, + state: &mut AdapterState, + output: &mut AdapterOutput, +) -> Result<()> { + let status = value.get("status").and_then(Value::as_u64).unwrap_or(0); + let body = value.get("body").cloned().unwrap_or(Value::Null); + match value.get("id").and_then(Value::as_u64) { + Some(ID_SESSION) => { + if status == 0 || status >= 400 { + let message = turn.resume.as_deref().map_or_else( + || "OpenCode could not start a session for this turn.".to_string(), + |_| { + "This OpenCode session can't be resumed — it may no longer exist." + .to_string() + }, + ); + fail(turn, state, output, &message, None); + return Ok(()); + } + let session = body.get("id").and_then(Value::as_str); + // A session with a parent is a subagent session. Prompting it + // returns success but produces no event activity at all, so the + // turn would hang until its deadline instead of failing here. + let parented = body.get("parentID").is_some_and(|parent| !parent.is_null()); + let Some(session) = session.filter(|_| !parented) else { + fail( + turn, + state, + output, + "This OpenCode session can't be resumed — it may be a subagent session or no longer exist.", + None, + ); + return Ok(()); + }; + turn.session_id = Some(session.to_string()); + if let Some(title) = body + .get("title") + .and_then(Value::as_str) + .map(str::trim) + .filter(|title| !title.is_empty()) + { + state.result.session_title = Some(title.to_string()); + } + if state.result.session_id.as_deref() != Some(session) { + state.result.session_id = Some(session.to_string()); + output.events.push(TurnEvent::SessionStarted { + session_id: session.to_string(), + title: state.result.session_title.clone(), + }); + } + // Subscribe before prompting so nothing emitted in the turn's + // first moments is missed; the prompt follows `@subscribed`. + output.writes.push(encode(&json!({ + "method": "SUBSCRIBE", + "path": "/event", + }))?); + } + Some(ID_PROMPT) if status == 0 || status >= 400 => { + let message = body + .pointer("/data/message") + .or_else(|| body.get("message")) + .and_then(Value::as_str) + .map_or_else( + || "OpenCode rejected this turn's prompt.".to_string(), + str::to_owned, + ); + fail(turn, state, output, &message, None); + } + _ => {} + } + Ok(()) +} + +/// Map one OpenCode SSE event onto the normalized stream. +fn event( + event: &Value, + turn: &mut TurnState, + state: &mut AdapterState, + output: &mut AdapterOutput, +) { + let kind = event + .get("type") + .and_then(Value::as_str) + .unwrap_or_default(); + let properties = event.get("properties").unwrap_or(&Value::Null); + // Another session's activity must never be charged to this turn. + if let (Some(reported), Some(current)) = ( + properties.get("sessionID").and_then(Value::as_str), + turn.session_id.as_deref(), + ) { + if reported != current { + return; + } + } + match kind { + "message.updated" => { + let info = properties.get("info").unwrap_or(&Value::Null); + if let (Some(id), Some(role)) = ( + info.get("id").and_then(Value::as_str), + info.get("role").and_then(Value::as_str), + ) { + turn.roles.insert(id.to_string(), role.to_string()); + } + if info.get("role").and_then(Value::as_str) == Some("assistant") { + if let Some(cost) = info.get("cost").and_then(Value::as_f64) { + state.result.usage.cost_usd = Some(cost); + } + } + } + "message.part.updated" => { + let part = properties.get("part").unwrap_or(&Value::Null); + let message = part.get("messageID").and_then(Value::as_str); + if message.is_none_or(|id| turn.roles.get(id).map(String::as_str) != Some("assistant")) + { + return; + } + let Some(part_id) = part.get("id").and_then(Value::as_str) else { + return; + }; + match part.get("type").and_then(Value::as_str) { + Some(kind @ ("text" | "reasoning")) => { + let full = part.get("text").and_then(Value::as_str).unwrap_or_default(); + if let Some(delta) = observe_full(turn, part_id, kind, full) { + emit_delta(kind, &delta, state, output); + } + } + Some("tool") => { + if let Some(event) = tool_event(part) { + turn.saw_activity = true; + output.events.push(event); + } + } + _ => {} + } + } + "message.part.delta" => { + let message = properties.get("messageID").and_then(Value::as_str); + if message.is_none_or(|id| turn.roles.get(id).map(String::as_str) != Some("assistant")) + { + return; + } + let (Some(part_id), Some(delta)) = ( + properties.get("partID").and_then(Value::as_str), + properties.get("delta").and_then(Value::as_str), + ) else { + return; + }; + if delta.is_empty() { + return; + } + let kind = observe_delta(turn, part_id, delta); + emit_delta(&kind, delta, state, output); + } + "permission.asked" => permission(properties, turn, output), + "session.error" => { + turn.error_message = properties + .pointer("/error/data/message") + .or_else(|| properties.pointer("/error/message")) + .and_then(Value::as_str) + .map(str::to_owned) + .or(turn.error_message.take()); + } + "session.idle" => { + output.terminal = true; + if let Some(message) = turn.error_message.clone() { + fail(turn, state, output, &message, None); + } else if state.result.text.trim().is_empty() && !turn.saw_activity { + // A turn that produced no text and no tool call is a real + // failure, not an empty success: it is what a silently + // auto-refused edit looked like before permissions were + // enforced, and reporting it as success hid exactly the + // problem this transport exists to solve. + fail( + turn, + state, + output, + "OpenCode finished without producing a reply.", + None, + ); + } + } + _ => {} + } +} + +/// Surface a permission request, or refuse it outright when the turn's policy +/// says no decision is available to make. +fn permission(properties: &Value, turn: &TurnState, output: &mut AdapterOutput) { + let Some(id) = properties.get("id").and_then(Value::as_str) else { + return; + }; + let kind = properties + .get("permission") + .and_then(Value::as_str) + .unwrap_or("tool"); + let patterns = properties + .get("patterns") + .and_then(Value::as_array) + .map(|patterns| { + patterns + .iter() + .filter_map(Value::as_str) + .collect::>() + .join(", ") + }) + .filter(|patterns| !patterns.is_empty()); + let (tool_name, description) = describe(kind, patterns.as_deref()); + + let request = ApprovalRequest { + id: id.to_string(), + tool_name, + description: Some(description), + input: properties.clone(), + }; + // Defence in depth. A plan turn and an empty tool allowlist are already + // denied by the configuration the server was started with, so reaching + // here means the policy did not hold. Refusing without asking keeps the + // guarantee even then, and never presents the user a choice that the turn + // promised would not exist. + if turn.plan_mode || turn.tools_denied { + output + .events + .push(TurnEvent::ApprovalRequested(request.clone())); + if let Ok(frame) = answer_frame( + turn.session_id.as_deref().unwrap_or_default(), + &turn.directory, + id, + ApprovalDecision::Deny { reason: None }, + ) { + output.writes.push(frame); + } + return; + } + output + .events + .push(TurnEvent::ApprovalRequested(request.clone())); + // The reply is addressed to a session and a workspace, and + // `approval_response` sees only this value, so both travel with it. + output.interaction = Some(InteractionRequest::Approval { + request, + original: json!({ + "id": id, + "session": turn.session_id, + "directory": turn.directory, + }), + }); +} + +/// Name and describe a permission the way the reference driver does. +fn describe(kind: &str, patterns: Option<&str>) -> (String, String) { + match (kind, patterns) { + ("bash", Some(patterns)) => ("Bash".into(), format!("run `{patterns}`")), + ("bash", None) => ("Bash".into(), "run a shell command".into()), + ("edit", Some(patterns)) => ("Edit".into(), format!("edit {patterns}")), + ("edit", None) => ("Edit".into(), "edit files".into()), + (kind, Some(patterns)) => (kind.into(), format!("use {kind}: {patterns}")), + (kind, None) => (kind.into(), format!("use {kind}")), + } +} + +/// Encode the HTTP request that answers one permission request. +fn answer_frame( + session: &str, + directory: &str, + permission_id: &str, + decision: ApprovalDecision, +) -> Result> { + let response = match decision { + ApprovalDecision::Allow => "once", + ApprovalDecision::AllowForSession => "always", + ApprovalDecision::Deny { .. } => "reject", + }; + encode(&post( + None, + format!( + "/session/{}/permissions/{}?directory={}", + query_escape(session), + query_escape(permission_id), + query_escape(directory) + ), + json!({"response": response}), + )) +} + +/// Encode an application's decision for the runtime to write. +/// +/// Everything needed to address the reply travels on `original`, which is the +/// only value this is given. +pub(super) fn approval_response(original: &Value, decision: ApprovalDecision) -> Result> { + let text = |field: &str| { + original + .get(field) + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() + }; + answer_frame(&text("session"), &text("directory"), &text("id"), decision) +} + +/// OpenCode has no question channel, so nothing is ever encoded for one. +pub(super) fn question_response( + _request: &QuestionRequest, + _original: &Value, + _answer: Option, +) -> Option> { + None +} + +/// A `message.part.updated` carries a part's whole current text; return only +/// the genuinely new suffix so a later resend cannot double-count what a +/// `message.part.delta` already streamed. +fn observe_full(turn: &mut TurnState, part_id: &str, kind: &str, full: &str) -> Option { + turn.part_kinds + .insert(part_id.to_string(), kind.to_string()); + let previous = turn.emitted.get(part_id).cloned().unwrap_or_default(); + if !full.starts_with(&previous) { + return None; + } + let delta = full[previous.len()..].to_string(); + if delta.is_empty() { + return None; + } + turn.emitted.insert(part_id.to_string(), full.to_string()); + Some(delta) +} + +/// Record an incremental delta against the same tracker, so a later full-value +/// resend does not re-emit it. +fn observe_delta(turn: &mut TurnState, part_id: &str, delta: &str) -> String { + let kind = turn + .part_kinds + .get(part_id) + .cloned() + .unwrap_or_else(|| "text".to_string()); + turn.emitted + .entry(part_id.to_string()) + .or_default() + .push_str(delta); + kind +} + +fn emit_delta(kind: &str, delta: &str, state: &mut AdapterState, output: &mut AdapterOutput) { + if kind == "reasoning" { + state + .result + .reasoning + .get_or_insert_with(String::new) + .push_str(delta); + output.events.push(TurnEvent::ReasoningDelta { + text: delta.to_string(), + }); + } else { + state.result.text.push_str(delta); + state.saw_text_delta = true; + output.events.push(TurnEvent::TextDelta { + text: delta.to_string(), + }); + } +} + +fn truncate(text: &str) -> String { + if text.chars().count() <= MAX_TOOL_OUTPUT_CHARS { + return text.to_string(); + } + let mut bounded: String = text.chars().take(MAX_TOOL_OUTPUT_CHARS).collect(); + bounded.push_str("… [truncated]"); + bounded +} + +fn tool_event(part: &Value) -> Option { + let id = part.get("callID").and_then(Value::as_str)?; + let name = part.get("tool").and_then(Value::as_str).unwrap_or("tool"); + let state = part.get("state").unwrap_or(&Value::Null); + let status = state + .get("status") + .and_then(Value::as_str) + .unwrap_or("running"); + let text = state.get("output").and_then(Value::as_str).map(truncate); + let (status, out, error) = match status { + "completed" => (ToolCallStatus::Succeeded, text, None), + "error" => ( + ToolCallStatus::Failed, + None, + Some(text.unwrap_or_else(|| "Tool call failed.".to_string())), + ), + _ => (ToolCallStatus::Started, None, None), + }; + Some(TurnEvent::ToolCall { + id: Some(id.to_string()), + name: name.to_string(), + status, + input: state.get("input").cloned(), + output: out, + error, + task_id: None, + }) +} + +/// Record a provider-native terminal failure. +fn fail( + turn: &mut TurnState, + state: &mut AdapterState, + output: &mut AdapterOutput, + message: &str, + code: Option<&str>, +) { + let kind = classify_provider_failure(&format!("{} {message}", code.unwrap_or_default())); + let mut failure = ProviderTerminalFailure::new(kind, message, DeliveryState::Accepted); + if let Some(code) = code { + failure = failure.with_provider_code(format!("opencode::{code}")); + } + turn.error_message = Some(message.to_string()); + state.terminal_failure = Some(failure); + state.result.status = RunStatus::Failed; + output.terminal = true; +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + + fn decode(bytes: &[u8]) -> Value { + serde_json::from_slice(bytes).unwrap() + } + + fn request(mode: PermissionMode) -> TurnRequest { + let mut request = TurnRequest::new(Provider::OpenCode, "/tmp/work", "do the thing"); + request.permission_mode = mode; + request + } + + /// Drive the state machine to the point where a turn is running. + fn running_turn(mode: PermissionMode) -> AdapterState { + let request = request(mode); + let mut state = AdapterState::default(); + prepare_turn(&request, &mut state, 4242); + parse_line(&json!({"type": FRAME_READY}).to_string(), &mut state).unwrap(); + parse_line( + &json!({"type": FRAME_RESPONSE, "id": ID_SESSION, "status": 200, + "body": {"id": "session-1"}}) + .to_string(), + &mut state, + ) + .unwrap(); + parse_line(&json!({"type": FRAME_SUBSCRIBED}).to_string(), &mut state).unwrap(); + state + } + + fn permission_event(kind: &str) -> String { + json!({"type": FRAME_EVENT, "event": { + "type": "permission.asked", + "properties": { + "id": "permission-9", "sessionID": "session-1", + "permission": kind, "patterns": ["rm -rf /"] + } + }}) + .to_string() + } + + #[test] + fn each_permission_mode_maps_onto_opencodes_two_axes() { + for (mode, edit, bash) in [ + (PermissionMode::Default, "ask", "ask"), + (PermissionMode::AcceptEdits, "allow", "ask"), + (PermissionMode::FullAccess, "allow", "allow"), + (PermissionMode::Plan, "deny", "deny"), + ] { + let config: Value = + serde_json::from_str(&permission_config(&request(mode.clone())).unwrap()).unwrap(); + assert_eq!( + config["permission"]["edit"], + json!(edit), + "edit for {mode:?}" + ); + assert_eq!( + config["permission"]["bash"], + json!(bash), + "bash for {mode:?}" + ); + } + } + + #[test] + fn an_empty_tool_allowlist_becomes_a_wildcard_denial() { + let mut request = request(PermissionMode::FullAccess); + request.launch_context.allowed_tools = Some(Vec::new()); + + let config: Value = serde_json::from_str(&permission_config(&request).unwrap()).unwrap(); + + assert_eq!( + config["permission"], + json!({"*": "deny"}), + "an explicitly empty tool set must be enforced, not merely suggested" + ); + } + + #[test] + fn a_permission_ask_reaches_the_application_and_its_answer_is_posted() { + let mut state = running_turn(PermissionMode::Default); + + let output = parse_line(&permission_event("bash"), &mut state).unwrap(); + + let Some(InteractionRequest::Approval { request, original }) = output.interaction else { + panic!("expected an approval interaction, got {:?}", output.events); + }; + assert_eq!(request.id, "permission-9"); + assert_eq!(request.tool_name, "Bash"); + assert!(request.description.unwrap().contains("rm -rf /")); + + let allow = decode(&approval_response(&original, ApprovalDecision::Allow).unwrap()); + assert_eq!(allow["method"], json!("POST")); + assert_eq!(allow["body"]["response"], json!("once")); + assert!(allow["path"] + .as_str() + .unwrap() + .starts_with("/session/session-1/permissions/permission-9")); + } + + #[test] + fn a_denied_permission_is_refused_on_the_native_wire_vocabulary() { + let original = + json!({"id": "permission-9", "session": "session-1", "directory": "/tmp/work"}); + + for (decision, expected) in [ + (ApprovalDecision::Allow, "once"), + (ApprovalDecision::AllowForSession, "always"), + (ApprovalDecision::Deny { reason: None }, "reject"), + ] { + let frame = decode(&approval_response(&original, decision).unwrap()); + assert_eq!(frame["body"]["response"], json!(expected)); + } + } + + #[test] + fn a_plan_turn_refuses_a_permission_without_consulting_the_application() { + let mut state = running_turn(PermissionMode::Plan); + + let output = parse_line(&permission_event("edit"), &mut state).unwrap(); + + assert!( + output.interaction.is_none(), + "a plan turn promised no side effects, so there is no decision to offer" + ); + let refusal = decode(&output.writes[0]); + assert_eq!(refusal["body"]["response"], json!("reject")); + assert!(matches!( + output.events.as_slice(), + [TurnEvent::ApprovalRequested(request)] if request.id == "permission-9" + )); + } + + #[test] + fn an_empty_tool_allowlist_also_refuses_without_asking() { + let mut request = request(PermissionMode::FullAccess); + request.launch_context.allowed_tools = Some(Vec::new()); + let mut state = AdapterState::default(); + prepare_turn(&request, &mut state, 4242); + parse_line(&json!({"type": FRAME_READY}).to_string(), &mut state).unwrap(); + parse_line( + &json!({"type": FRAME_RESPONSE, "id": ID_SESSION, "status": 200, + "body": {"id": "session-1"}}) + .to_string(), + &mut state, + ) + .unwrap(); + + let output = parse_line(&permission_event("bash"), &mut state).unwrap(); + + assert!(output.interaction.is_none()); + assert_eq!( + decode(&output.writes[0])["body"]["response"], + json!("reject") + ); + } + + #[test] + fn the_prompt_waits_for_the_event_stream_to_be_open() { + let request = request(PermissionMode::Default); + let mut state = AdapterState::default(); + prepare_turn(&request, &mut state, 4242); + + parse_line(&json!({"type": FRAME_READY}).to_string(), &mut state).unwrap(); + let opened = parse_line( + &json!({"type": FRAME_RESPONSE, "id": ID_SESSION, "status": 200, + "body": {"id": "session-1"}}) + .to_string(), + &mut state, + ) + .unwrap(); + + assert_eq!( + decode(&opened.writes[0])["method"], + json!("SUBSCRIBE"), + "subscribing first is what keeps early events from being missed" + ); + assert!( + opened.writes.len() == 1, + "the prompt must not be sent before the stream is open" + ); + + let prompted = + parse_line(&json!({"type": FRAME_SUBSCRIBED}).to_string(), &mut state).unwrap(); + let frame = decode(&prompted.writes[0]); + assert_eq!(frame["method"], json!("POST")); + assert!(frame["path"].as_str().unwrap().contains("/message")); + assert_eq!(frame["body"]["parts"][0]["text"], json!("do the thing")); + } + + #[test] + fn a_subagent_session_fails_fast_instead_of_hanging() { + let mut request = request(PermissionMode::Default); + request.session_id = Some("child-1".into()); + let mut state = AdapterState::default(); + prepare_turn(&request, &mut state, 4242); + parse_line(&json!({"type": FRAME_READY}).to_string(), &mut state).unwrap(); + + let output = parse_line( + &json!({"type": FRAME_RESPONSE, "id": ID_SESSION, "status": 200, + "body": {"id": "child-1", "parentID": "parent-1"}}) + .to_string(), + &mut state, + ) + .unwrap(); + + assert!(output.terminal); + assert_eq!(state.result.status, RunStatus::Failed); + assert!(state + .terminal_failure + .unwrap() + .diagnostic + .contains("subagent")); + } + + #[test] + fn a_server_that_dies_mid_turn_fails_the_turn_with_its_diagnostic() { + let mut state = running_turn(PermissionMode::Default); + + let output = parse_line( + &json!({"type": FRAME_ERROR, + "message": "OpenCode's event stream ended before the turn finished."}) + .to_string(), + &mut state, + ) + .unwrap(); + + assert!(output.terminal); + assert_eq!(state.result.status, RunStatus::Failed); + assert!(state + .terminal_failure + .unwrap() + .diagnostic + .contains("ended before the turn finished")); + } + + #[test] + fn only_assistant_text_is_streamed_and_never_counted_twice() { + let mut state = running_turn(PermissionMode::Default); + let role = |id: &str, role: &str| { + json!({"type": FRAME_EVENT, "event": {"type": "message.updated", "properties": { + "sessionID": "session-1", "info": {"id": id, "role": role} + }}}) + .to_string() + }; + parse_line(&role("m-user", "user"), &mut state).unwrap(); + parse_line(&role("m-1", "assistant"), &mut state).unwrap(); + + let user = parse_line( + &json!({"type": FRAME_EVENT, "event": {"type": "message.part.updated", "properties": { + "sessionID": "session-1", + "part": {"id": "p-0", "messageID": "m-user", "type": "text", "text": "echo"} + }}}) + .to_string(), + &mut state, + ) + .unwrap(); + assert!( + user.events.is_empty(), + "the user's own message is not output" + ); + + let first = parse_line( + &json!({"type": FRAME_EVENT, "event": {"type": "message.part.updated", "properties": { + "sessionID": "session-1", + "part": {"id": "p-1", "messageID": "m-1", "type": "text", "text": "Hello"} + }}}) + .to_string(), + &mut state, + ) + .unwrap(); + assert_eq!( + first.events, + vec![TurnEvent::TextDelta { + text: "Hello".into() + }] + ); + + // The same part resent in full must only yield the new suffix. + let second = parse_line( + &json!({"type": FRAME_EVENT, "event": {"type": "message.part.updated", "properties": { + "sessionID": "session-1", + "part": {"id": "p-1", "messageID": "m-1", "type": "text", "text": "Hello there"} + }}}) + .to_string(), + &mut state, + ) + .unwrap(); + assert_eq!( + second.events, + vec![TurnEvent::TextDelta { + text: " there".into() + }] + ); + assert_eq!(state.result.text, "Hello there"); + } + + #[test] + fn another_sessions_events_are_not_charged_to_this_turn() { + let mut state = running_turn(PermissionMode::Default); + + let output = parse_line( + &json!({"type": FRAME_EVENT, "event": {"type": "message.part.delta", "properties": { + "sessionID": "session-other", "messageID": "m-1", + "partID": "p-1", "delta": "leak" + }}}) + .to_string(), + &mut state, + ) + .unwrap(); + + assert!(output.events.is_empty()); + assert_eq!(state.result.text, ""); + } + + #[test] + fn an_idle_turn_that_produced_nothing_is_a_failure_not_an_empty_success() { + let mut state = running_turn(PermissionMode::Default); + + let output = parse_line( + &json!({"type": FRAME_EVENT, "event": {"type": "session.idle", + "properties": {"sessionID": "session-1"}}}) + .to_string(), + &mut state, + ) + .unwrap(); + + assert!(output.terminal); + assert_eq!(state.result.status, RunStatus::Failed); + } + + #[test] + fn stdio_and_remote_mcp_servers_reach_the_config_without_their_secrets() { + let mut request = request(PermissionMode::Default); + request.launch_context.mcp_servers.insert( + "temps_fleet".into(), + crate::McpServerConfig::Stdio { + command: "/opt/tools/temps fleet".into(), + args: vec!["mcp".into(), "serve".into()], + environment_from: BTreeMap::from([("TOKEN".into(), "FLEET_TOKEN".into())]), + }, + ); + request.launch_context.mcp_servers.insert( + "platform".into(), + crate::McpServerConfig::Http { + url: "https://relay.example.test/mcp".into(), + headers_from: BTreeMap::from([("Authorization".into(), "RELAY_TOKEN".into())]), + }, + ); + + let config: Value = serde_json::from_str(&permission_config(&request).unwrap()).unwrap(); + + let fleet = &config["mcp"]["temps_fleet"]; + assert_eq!(fleet["type"], json!("local")); + assert_eq!( + fleet["command"], + json!(["/opt/tools/temps fleet", "mcp", "serve"]), + "an argv array needs no quoting, so a spaced path stays one argument" + ); + assert_eq!(fleet["environment"]["TOKEN"], json!("{env:FLEET_TOKEN}")); + assert_eq!(fleet["enabled"], json!(true)); + + let platform = &config["mcp"]["platform"]; + assert_eq!(platform["type"], json!("remote")); + assert_eq!( + platform["headers"]["Authorization"], + json!("{env:RELAY_TOKEN}") + ); + // The permission policy still merges alongside the MCP block. + assert_eq!(config["permission"]["edit"], json!("ask")); + } + + #[test] + fn the_launch_context_is_carried_as_a_prompt_prefix() { + let mut request = request(PermissionMode::Default); + request.launch_context.system_prompt_append = Some("Answer in French.".into()); + request.launch_context.allowed_tools = Some(vec!["read".into(), "grep".into()]); + let mut state = AdapterState::default(); + prepare_turn(&request, &mut state, 4242); + + let turn = load(&state); + let text = turn.parts[0]["text"].as_str().unwrap(); + assert!(text.starts_with("Answer in French.")); + assert!(text.contains("Use only these tools: read, grep")); + assert!(text.ends_with("do the thing")); + } + + #[test] + fn a_workspace_path_with_separators_cannot_rewrite_the_query() { + let mut request = TurnRequest::new(Provider::OpenCode, "/tmp/a b&c?d", "hi"); + request.permission_mode = PermissionMode::Default; + let mut state = AdapterState::default(); + prepare_turn(&request, &mut state, 4242); + + let output = parse_line(&json!({"type": FRAME_READY}).to_string(), &mut state).unwrap(); + + let path = decode(&output.writes[0])["path"] + .as_str() + .unwrap() + .to_string(); + assert!(path.contains("%26"), "`&` must be escaped in {path}"); + assert!(path.contains("%3F"), "`?` must be escaped in {path}"); + assert_eq!( + path.matches('?').count(), + 1, + "only the query starts a query" + ); + } + + #[test] + fn a_running_turn_can_be_interrupted_on_its_own_session() { + let state = running_turn(PermissionMode::Default); + + let frame = decode(&interrupt(&state).unwrap()); + + assert_eq!(frame["method"], json!("POST")); + assert!(frame["path"] + .as_str() + .unwrap() + .starts_with("/session/session-1/abort")); + } + + #[test] + fn a_model_identifier_is_split_into_opencodes_native_selection() { + let mut request = request(PermissionMode::Default); + request.model = Some("anthropic/claude-sonnet-4".into()); + let mut state = AdapterState::default(); + prepare_turn(&request, &mut state, 4242); + parse_line(&json!({"type": FRAME_READY}).to_string(), &mut state).unwrap(); + parse_line( + &json!({"type": FRAME_RESPONSE, "id": ID_SESSION, "status": 200, + "body": {"id": "session-1"}}) + .to_string(), + &mut state, + ) + .unwrap(); + + let prompted = + parse_line(&json!({"type": FRAME_SUBSCRIBED}).to_string(), &mut state).unwrap(); + + assert_eq!( + decode(&prompted.writes[0])["body"]["model"], + json!({"providerID": "anthropic", "modelID": "claude-sonnet-4"}) + ); + } +} From a5af90e45e19e3f877d61d0766b67343aa6bb060 Mon Sep 17 00:00:00 2001 From: David Viejo Date: Mon, 21 Sep 2026 18:54:37 +0200 Subject: [PATCH 4/6] test(opencode): cover served turns against a scripted opencode serve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fixture is a real HTTP server bound to the very port the adapter reserved, speaking the real protocol over a real socket: readiness, session, chunked SSE, prompt, and the permission endpoint. Nothing stubs the carrier, so a served turn is covered end to end rather than only the state machine above it. Covered: a permission allowed and answered `once`, a permission denied and answered `reject`, a plan turn whose policy denies both categories and which refuses without ever consulting the application, the requested policy and MCP servers arriving in the child's environment with the credential referenced by name rather than serialized, a server that dies mid-turn, and a cancelled turn that aborts the session cooperatively before the process is killed. Two real lifecycle defects surfaced while writing them. `opencode serve` is a server: nothing makes it exit because a turn ended, so waiting for a natural exit hung until the turn deadline. A turn carried on adapter-supplied streams now terminates the child as its normal shutdown, with a bounded wait to reap it. The bridge also deadlocked. Its reader and writer were the two halves of one duplex stream, which stays alive until both halves drop — so the runtime closing its writer never reached the bridge as end-of-input, while the bridge held the reader open waiting for exactly that. Each direction is now its own pipe. The turn loop additionally stops reading at a terminal frame when it is attached, so a third-party carrier that never closes its reader cannot hang a turn either. --- src/providers/opencode_http.rs | 21 +- src/runtime.rs | 49 ++- tests/opencode_serve.rs | 672 +++++++++++++++++++++++++++++++++ 3 files changed, 729 insertions(+), 13 deletions(-) create mode 100644 tests/opencode_serve.rs diff --git a/src/providers/opencode_http.rs b/src/providers/opencode_http.rs index 7d748d5..473dc70 100644 --- a/src/providers/opencode_http.rs +++ b/src/providers/opencode_http.rs @@ -89,12 +89,18 @@ pub(super) const FRAME_SUBSCRIBED: &str = "@subscribed"; /// The returned streams are live immediately; the task behind them polls the /// server for readiness first and emits [`FRAME_READY`] once it answers. pub(super) fn connect(port: u16) -> ProtocolStreams { - let (runtime_side, bridge_side) = tokio::io::duplex(BRIDGE_BUFFER_BYTES); - let (reader, writer) = tokio::io::split(runtime_side); - tokio::spawn(run_bridge(port, bridge_side)); + // Two independent pipes rather than the two halves of one. Splitting a + // single duplex stream would keep it alive until *both* halves drop, so + // the runtime closing its writer at the end of a turn would never reach + // the bridge as end-of-input — and the bridge would keep the reader open + // while the runtime waited for it to close. Separate pipes make each + // direction close on its own. + let (runtime_reader, bridge_writer) = tokio::io::duplex(BRIDGE_BUFFER_BYTES); + let (bridge_reader, runtime_writer) = tokio::io::duplex(BRIDGE_BUFFER_BYTES); + tokio::spawn(run_bridge(port, bridge_reader, bridge_writer)); ProtocolStreams { - reader: Box::new(reader), - writer: Box::new(writer), + reader: Box::new(runtime_reader), + writer: Box::new(runtime_writer), } } @@ -104,8 +110,7 @@ fn address(port: u16) -> SocketAddr { /// Drive one turn's HTTP traffic until the state machine stops writing or the /// server stops answering. -async fn run_bridge(port: u16, bridge_side: DuplexStream) { - let (incoming, outgoing) = tokio::io::split(bridge_side); +async fn run_bridge(port: u16, incoming: DuplexStream, outgoing: DuplexStream) { let outgoing = std::sync::Arc::new(tokio::sync::Mutex::new(outgoing)); if let Err(error) = wait_until_ready(port).await { @@ -234,7 +239,7 @@ async fn perform( async fn stream_events( port: u16, path: String, - outgoing: std::sync::Arc>>, + outgoing: std::sync::Arc>, ) { let stream = match TcpStream::connect(address(port)).await { Ok(stream) => stream, diff --git a/src/runtime.rs b/src/runtime.rs index f432290..5353849 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -52,6 +52,10 @@ const SANDBOX_RETRY_PROMPT: &str = "The sandbox profile was updated with the app /// How long a cancelled provider may keep running after acknowledging an /// adapter-encoded interrupt, before the process tree is terminated anyway. const INTERRUPT_GRACE: Duration = Duration::from_secs(5); +/// How long to wait for a terminated provider to be reaped when the turn ran +/// over adapter-supplied protocol streams. The process is already being +/// stopped; this only bounds how long the turn waits to observe it. +const ATTACHED_SHUTDOWN_GRACE: Duration = Duration::from_secs(5); /// Write newline-terminated provider frames to an interactive stdin. async fn write_provider_frames( @@ -2390,8 +2394,10 @@ impl AgentRuntime { // so everything below this point — cancellation, interrupts, // interaction timeouts, line bounding — is shared by both kinds of // provider instead of growing a second turn loop. + let mut attached = false; let reader: crate::TransportReader = match adapter.attach(request, &state).await { Ok(Some(streams)) => { + attached = true; stdin = Some(streams.writer); streams.reader } @@ -2505,21 +2511,54 @@ impl AgentRuntime { } if output.terminal { stdin.take(); + if attached { + // A stdio provider is read to end-of-output because + // closing its stdin is what makes it exit, and trailing + // lines can still arrive. An adapter-supplied carrier has + // no such contract: the protocol is over, and waiting for + // the adapter to close its own reader would hand a third + // party the ability to hang the turn. + break; + } } } drop(stdin); - let status = process - .wait() - .await - .map_err(|source| RuntimeError::Transport { provider, source })?; + let status = if attached { + // A provider driven over adapter-supplied streams has no reason to + // exit when the protocol ends: `opencode serve` is a server, and + // nothing closes it because a turn finished. Waiting for a natural + // exit would hang until the turn deadline, so stopping it *is* the + // normal shutdown here. A turn that failed has already recorded + // why, so the exit status this synthesizes is never what decides + // the outcome. + process + .terminate() + .await + .map_err(|source| RuntimeError::Transport { provider, source })?; + match tokio::time::timeout(ATTACHED_SHUTDOWN_GRACE, process.wait()).await { + Ok(status) => { + status.map_err(|source| RuntimeError::Transport { provider, source })? + } + Err(_) => TransportExitStatus { + success: true, + code: None, + }, + } + } else { + process + .wait() + .await + .map_err(|source| RuntimeError::Transport { provider, source })? + }; match request.tool_process_policy { crate::ToolProcessPolicy::PreserveOnCompletion => process.disarm(), - crate::ToolProcessPolicy::TerminateOnCompletion => { + crate::ToolProcessPolicy::TerminateOnCompletion if !attached => { process .terminate() .await .map_err(|source| RuntimeError::Transport { provider, source })?; } + crate::ToolProcessPolicy::TerminateOnCompletion => {} } let stderr = stderr_task .await diff --git a/tests/opencode_serve.rs b/tests/opencode_serve.rs new file mode 100644 index 0000000..d1def42 --- /dev/null +++ b/tests/opencode_serve.rs @@ -0,0 +1,672 @@ +//! Served-OpenCode coverage against a scripted `opencode serve`. +//! +//! The fixture is a real HTTP server bound to the very port the adapter +//! reserved for the turn, speaking the real protocol: `/app` for readiness, +//! `/session` to open one, `/event` as a chunked Server-Sent Events stream, +//! `/session/{id}/message` to prompt, and +//! `/session/{id}/permissions/{id}` to answer a permission. Nothing here +//! shells out to an OpenCode binary, and nothing stubs the HTTP or SSE +//! carrier — the bytes really cross a socket, so the transport is covered +//! end to end rather than only the state machine above it. + +#![cfg(feature = "opencode")] + +use std::collections::BTreeMap; +use std::path::Path; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use serde_json::{json, Value}; +use temps_agent_runtime::providers::{OpenCode, OpenCodeTurnMode}; +use temps_agent_runtime::{ + AgentRuntime, ApprovalDecision, ApprovalRequest, EventSink, ExecutionTransport, + InteractionHandler, McpServerConfig, PermissionMode, Provider, ProviderReadiness, + QuestionAnswer, QuestionRequest, Result, RuntimeError, SandboxCapabilities, + TransportCapabilities, TransportError, TransportErrorKind, TransportExitStatus, + TransportProcess, TransportProcessControl, TransportProcessHandle, TransportReader, + TransportReadinessRequest, TransportResult, TransportSpawnRequest, TransportWriter, TurnEvent, + TurnRequest, +}; +use tokio::io::{duplex, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_util::sync::CancellationToken; + +/// Turn shape the fixture server plays out once the prompt arrives. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Script { + /// Ask one `bash` permission, then reply according to the decision. + Permission, + /// Close the event stream mid-turn, the way a crashing server does. + Crash, + /// Stream one delta and then go quiet, so the turn must be cancelled. + Interrupt, +} + +/// One request the SDK made, recorded for assertions. +#[derive(Debug, Clone)] +struct Recorded { + path: String, + body: Value, +} + +#[derive(Clone)] +struct Server { + script: Script, + requests: Arc>>, + /// Argument vector the SDK asked the transport to spawn. + arguments: Arc>>, + /// The per-turn policy the SDK put in the child's environment. + config: Arc>>, +} + +impl Server { + fn new(script: Script) -> Self { + Self { + script, + requests: Arc::new(Mutex::new(Vec::new())), + arguments: Arc::new(Mutex::new(Vec::new())), + config: Arc::new(Mutex::new(None)), + } + } + + fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + + /// The `OPENCODE_CONFIG_CONTENT` the served turn was started with. + fn config(&self) -> Value { + let raw = self.config.lock().unwrap().clone().expect("policy was set"); + serde_json::from_str(&raw).expect("the policy is JSON") + } + + fn arguments(&self) -> Vec { + self.arguments.lock().unwrap().clone() + } + + fn first_matching(&self, needle: &str) -> Option { + self.requests() + .into_iter() + .find(|recorded| recorded.path.contains(needle)) + } + + /// Wait until the SDK issued a request whose path contains `needle`. + async fn wait_for(&self, needle: &str) -> Option { + for _ in 0..400 { + if let Some(recorded) = self.first_matching(needle) { + return Some(recorded); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + None + } +} + +#[async_trait] +impl ExecutionTransport for Server { + fn name(&self) -> &'static str { + "fixture-opencode-serve" + } + + fn capabilities(&self) -> TransportCapabilities { + TransportCapabilities { + remote: false, + interactive_stdin: true, + reconnect: false, + managed_processes: false, + process_tree_termination: true, + sandbox: SandboxCapabilities::NONE, + } + } + + async fn readiness( + &self, + request: TransportReadinessRequest, + ) -> TransportResult { + Ok(ProviderReadiness { + provider: request.provider, + installed: true, + executable: Some(request.program), + version: Some("opencode 1.4.3".to_string()), + detail: "fixture".to_string(), + }) + } + + async fn validate_working_directory(&self, _working_directory: &Path) -> TransportResult<()> { + Ok(()) + } + + async fn spawn(&self, request: TransportSpawnRequest) -> TransportResult { + let arguments = request + .command + .args + .iter() + .map(|argument| argument.to_string_lossy().into_owned()) + .collect::>(); + assert_eq!( + arguments.first().map(String::as_str), + Some("serve"), + "the served turn mode must not launch `opencode run`" + ); + // The policy is the whole enforcement boundary, so it must arrive in + // the child's environment rather than being left to the machine. + let config = request + .command + .environment + .get(std::ffi::OsStr::new("OPENCODE_CONFIG_CONTENT")) + .map(|value| value.to_string_lossy().into_owned()) + .expect("the served turn must supply a permission policy"); + *self.config.lock().unwrap() = Some(config); + + let port = arguments + .iter() + .position(|argument| argument == "--port") + .and_then(|index| arguments.get(index + 1)) + .and_then(|port| port.parse::().ok()) + .expect("the served turn must pin a port"); + *self.arguments.lock().unwrap() = arguments; + + // Bind the port the adapter reserved, exactly as the real child does. + let listener = TcpListener::bind(("127.0.0.1", port)) + .await + .expect("the reserved port is free for the child to bind"); + tokio::spawn(accept_loop( + listener, + self.script, + Arc::clone(&self.requests), + )); + + let (sdk_stdin, child_input) = duplex(1024); + let (sdk_stdout, child_output) = duplex(1024); + let (sdk_stderr, child_stderr) = duplex(1024); + // `opencode serve` says nothing on its own stdio; the protocol lives + // entirely on the socket above. + drop(child_input); + drop(child_output); + drop(child_stderr); + Ok(TransportProcess::new( + TransportProcessHandle { + transport: "fixture-opencode-serve".to_string(), + native_id: "1".to_string(), + }, + None, + Some(Box::new(sdk_stdin) as TransportWriter), + Box::new(sdk_stdout) as TransportReader, + Box::new(sdk_stderr) as TransportReader, + Control::default(), + )) + } + + async fn attach( + &self, + _handle: &TransportProcessHandle, + _cursor: Option, + ) -> TransportResult { + Err(TransportError::new( + TransportErrorKind::Unsupported, + self.name(), + "attach", + "the fixture server cannot be reattached", + false, + )) + } +} + +/// Mirrors a real server's lifetime: it runs until something stops it. +#[derive(Default)] +struct Control { + stopped: bool, +} + +#[async_trait] +impl TransportProcessControl for Control { + async fn wait(&mut self) -> TransportResult { + if !self.stopped { + // `opencode serve` never exits because a turn ended, so a runtime + // that waited for it here would hang until the turn deadline. + std::future::pending::<()>().await; + } + Ok(TransportExitStatus { + success: true, + code: Some(0), + }) + } + + async fn terminate(&mut self) -> TransportResult<()> { + self.stopped = true; + Ok(()) + } +} + +async fn accept_loop(listener: TcpListener, script: Script, requests: Arc>>) { + loop { + let Ok((socket, _)) = listener.accept().await else { + return; + }; + tokio::spawn(handle(socket, script, Arc::clone(&requests))); + } +} + +/// Read one request, record it, and answer it. +async fn handle(mut socket: TcpStream, script: Script, requests: Arc>>) { + let mut reader = BufReader::new(&mut socket); + let mut request_line = String::new(); + if reader.read_line(&mut request_line).await.unwrap_or(0) == 0 { + return; + } + let mut length = 0_usize; + loop { + let mut header = String::new(); + if reader.read_line(&mut header).await.unwrap_or(0) == 0 || header.trim().is_empty() { + break; + } + if let Some(value) = header.to_ascii_lowercase().strip_prefix("content-length:") { + length = value.trim().parse().unwrap_or(0); + } + } + let body = if length > 0 { + let mut buffer = vec![0_u8; length]; + reader.read_exact(&mut buffer).await.ok(); + serde_json::from_slice(&buffer).unwrap_or(Value::Null) + } else { + Value::Null + }; + + let path = request_line + .split_whitespace() + .nth(1) + .unwrap_or_default() + .to_string(); + requests.lock().unwrap().push(Recorded { + path: path.clone(), + body, + }); + + if path.starts_with("/event") { + stream_events(socket, script, requests).await; + return; + } + let payload = if path.starts_with("/session?") || path == "/session" { + json!({"id": "session-fixture", "title": "Fixture session"}).to_string() + } else { + json!({}).to_string() + }; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{payload}", + payload.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + let _ = socket.flush().await; +} + +/// Write one SSE payload as its own chunk, as a real server does. +async fn send(socket: &mut TcpStream, event: &Value) -> bool { + let payload = format!("data: {event}\n\n"); + let chunk = format!("{:X}\r\n{payload}\r\n", payload.len()); + socket.write_all(chunk.as_bytes()).await.is_ok() && socket.flush().await.is_ok() +} + +/// Play the scripted turn out over a chunked SSE stream. +async fn stream_events(mut socket: TcpStream, script: Script, requests: Arc>>) { + let _ = socket + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n", + ) + .await; + let _ = socket.flush().await; + + // Establish the assistant message every part below belongs to. + if !send( + &mut socket, + &json!({"type": "message.updated", "properties": { + "sessionID": "session-fixture", + "info": {"id": "message-1", "role": "assistant"} + }}), + ) + .await + { + return; + } + + if script == Script::Crash { + // Exactly what a dying server looks like from the SDK's side. + drop(socket); + return; + } + + if script == Script::Permission { + if !send( + &mut socket, + &json!({"type": "permission.asked", "properties": { + "sessionID": "session-fixture", + "id": "permission-1", + "permission": "bash", + "patterns": ["rm -rf /"] + }}), + ) + .await + { + return; + } + // Wait for the decision to come back on the permission endpoint. + let mut answer = None; + for _ in 0..400 { + let found = requests + .lock() + .unwrap() + .iter() + .find(|recorded| recorded.path.contains("/permissions/")) + .cloned(); + if let Some(found) = found { + answer = found + .body + .get("response") + .and_then(Value::as_str) + .map(str::to_owned); + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + let text = match answer.as_deref() { + Some("reject") => "I was not allowed to run that.", + Some(_) => "Removed everything, as instructed.", + None => "No decision ever arrived.", + }; + if !send( + &mut socket, + &json!({"type": "message.part.updated", "properties": { + "sessionID": "session-fixture", + "part": {"id": "part-1", "messageID": "message-1", "type": "text", "text": text} + }}), + ) + .await + { + return; + } + } + + if script == Script::Interrupt { + if !send( + &mut socket, + &json!({"type": "message.part.updated", "properties": { + "sessionID": "session-fixture", + "part": {"id": "part-1", "messageID": "message-1", + "type": "text", "text": "Working on it"} + }}), + ) + .await + { + return; + } + // Go quiet: only cancellation can end this turn. + std::future::pending::<()>().await; + } + + let _ = send( + &mut socket, + &json!({"type": "session.idle", "properties": {"sessionID": "session-fixture"}}), + ) + .await; +} + +/// Answers every approval with one fixed decision. +struct Decide(ApprovalDecision); + +#[async_trait] +impl InteractionHandler for Decide { + async fn approve(&self, _request: ApprovalRequest) -> ApprovalDecision { + self.0.clone() + } + + async fn answer(&self, _request: QuestionRequest) -> Option { + None + } +} + +/// Refuses to answer, so a turn that consults it would stall. +struct NeverAsked; + +#[async_trait] +impl InteractionHandler for NeverAsked { + async fn approve(&self, request: ApprovalRequest) -> ApprovalDecision { + panic!( + "the application must not be asked to approve `{}`", + request.tool_name + ); + } + + async fn answer(&self, _request: QuestionRequest) -> Option { + None + } +} + +#[derive(Clone, Default)] +struct Collected(Arc>>); + +#[async_trait] +impl EventSink for Collected { + async fn emit(&self, event: TurnEvent) -> Result<()> { + self.0.lock().unwrap().push(event); + Ok(()) + } +} + +impl Collected { + fn events(&self) -> Vec { + self.0.lock().unwrap().clone() + } +} + +fn runtime(server: &Server) -> AgentRuntime { + let mut builder = AgentRuntime::builder().transport(server.clone()); + builder.register(OpenCode::serve()); + builder.build().expect("runtime builds") +} + +fn turn(mode: PermissionMode) -> TurnRequest { + let mut request = TurnRequest::new( + Provider::OpenCode, + std::env::temp_dir(), + "delete everything", + ); + request.permission_mode = mode; + request.timeout = Duration::from_secs(30); + request.interaction_timeout = Duration::from_secs(10); + request +} + +#[tokio::test] +async fn an_allowed_permission_is_answered_once_and_the_turn_continues() { + let server = Server::new(Script::Permission); + let events = Collected::default(); + + let result = runtime(&server) + .run( + turn(PermissionMode::Default), + &events, + Some(&Decide(ApprovalDecision::Allow)), + ) + .await + .expect("the turn completes"); + + let answer = server + .first_matching("/permissions/") + .expect("the decision reached OpenCode"); + assert_eq!(answer.body["response"], json!("once")); + assert!(answer + .path + .contains("/session/session-fixture/permissions/permission-1")); + assert_eq!(result.text, "Removed everything, as instructed."); + assert!(events.events().iter().any( + |event| matches!(event, TurnEvent::ApprovalRequested(request) + if request.tool_name == "Bash" + && request.description.as_deref().is_some_and(|text| text.contains("rm -rf /"))) + )); +} + +#[tokio::test] +async fn a_denied_permission_is_rejected_on_the_wire() { + let server = Server::new(Script::Permission); + let events = Collected::default(); + + let result = runtime(&server) + .run( + turn(PermissionMode::Default), + &events, + Some(&Decide(ApprovalDecision::Deny { + reason: Some("not on my machine".into()), + })), + ) + .await + .expect("a refused tool call is still a completed turn"); + + assert_eq!( + server + .first_matching("/permissions/") + .expect("the refusal reached OpenCode") + .body["response"], + json!("reject") + ); + assert_eq!(result.text, "I was not allowed to run that."); +} + +#[tokio::test] +async fn a_plan_turn_denies_both_permission_categories_and_never_asks() { + let server = Server::new(Script::Permission); + let events = Collected::default(); + + let result = runtime(&server) + .run(turn(PermissionMode::Plan), &events, Some(&NeverAsked)) + .await + .expect("the turn completes"); + + // The policy the server was started with is the real boundary. + let config = server.config(); + assert_eq!(config["permission"]["edit"], json!("deny")); + assert_eq!(config["permission"]["bash"], json!("deny")); + // And a permission that arrives anyway is refused without the + // application ever being consulted — `NeverAsked` would panic. + assert_eq!( + server + .first_matching("/permissions/") + .expect("the refusal reached OpenCode") + .body["response"], + json!("reject") + ); + assert_eq!(result.text, "I was not allowed to run that."); +} + +#[tokio::test] +async fn the_requested_policy_and_mcp_servers_reach_the_child_environment() { + let server = Server::new(Script::Permission); + let events = Collected::default(); + let mut request = turn(PermissionMode::AcceptEdits); + request.environment.insert( + "FLEET_TOKEN".into(), + temps_agent_runtime::SecretString::new("super-secret-value"), + ); + request.launch_context.mcp_servers.insert( + "temps_fleet".into(), + McpServerConfig::Stdio { + command: "/opt/tools/temps fleet".into(), + args: vec!["mcp".into(), "serve".into()], + environment_from: BTreeMap::from([("TOKEN".into(), "FLEET_TOKEN".into())]), + }, + ); + + runtime(&server) + .run(request, &events, Some(&Decide(ApprovalDecision::Allow))) + .await + .expect("the turn completes"); + + let config = server.config(); + assert_eq!(config["permission"]["edit"], json!("allow")); + assert_eq!(config["permission"]["bash"], json!("ask")); + let fleet = &config["mcp"]["temps_fleet"]; + assert_eq!(fleet["type"], json!("local")); + assert_eq!( + fleet["command"], + json!(["/opt/tools/temps fleet", "mcp", "serve"]) + ); + assert_eq!(fleet["environment"]["TOKEN"], json!("{env:FLEET_TOKEN}")); + assert!( + !server + .config + .lock() + .unwrap() + .as_ref() + .unwrap() + .contains("FLEET_TOKEN=") + && !format!("{:?}", server.arguments()).contains("FLEET_TOKEN"), + "a credential must be referenced by name, never serialized" + ); +} + +#[tokio::test] +async fn a_server_that_dies_mid_turn_fails_instead_of_hanging() { + let server = Server::new(Script::Crash); + let events = Collected::default(); + + let error = runtime(&server) + .run(turn(PermissionMode::Default), &events, Some(&NeverAsked)) + .await + .expect_err("a turn whose server disappeared cannot succeed"); + + match error { + RuntimeError::ProcessFailed { stderr, .. } => { + assert!( + stderr.contains("event stream"), + "the diagnostic should say the stream died, got `{stderr}`" + ); + } + other => panic!("expected a process failure, got {other:?}"), + } +} + +#[tokio::test] +async fn cancelling_a_turn_aborts_the_session_before_the_process_is_killed() { + let server = Server::new(Script::Interrupt); + let events = Collected::default(); + let cancellation = CancellationToken::new(); + let mut request = turn(PermissionMode::Default); + request.cancellation = cancellation.clone(); + + let running = { + let runtime = runtime(&server); + let events = events.clone(); + tokio::spawn(async move { runtime.run(request, &events, Some(&NeverAsked)).await }) + }; + + // Cancel only once the turn is genuinely under way. + server + .wait_for("/message") + .await + .expect("the prompt was delivered"); + tokio::time::sleep(Duration::from_millis(100)).await; + cancellation.cancel(); + + let error = running.await.unwrap().expect_err("a cancelled turn fails"); + assert!(matches!( + error, + RuntimeError::Cancelled { + provider: Provider::OpenCode + } + )); + assert!( + server.wait_for("/abort").await.is_some(), + "the session must be asked to stop cooperatively before the process is killed" + ); +} + +#[tokio::test] +async fn the_run_turn_mode_keeps_its_one_shot_behaviour() { + use temps_agent_runtime::AgentAdapter; + + let adapter = OpenCode::default(); + assert_eq!(adapter.turn_mode(), OpenCodeTurnMode::Run); + assert!( + !adapter.permission_support().live_approvals, + "`opencode run` cannot answer a permission mid-turn and must not claim to" + ); + assert!(OpenCode::serve().permission_support().live_approvals); +} From a73e1d32dd39568405df39bab681c48e09ca982e Mon Sep 17 00:00:00 2001 From: David Viejo Date: Mon, 21 Sep 2026 18:56:34 +0200 Subject: [PATCH 5/6] docs(opencode): describe the serve turn mode and its permission policy Records what `Serve` mode changes and, importantly, why it exists: `Run` mode's only permission levers are `--auto` and `--agent plan`, so the policy a turn runs under is whatever configuration the machine happens to have. The capability matrix now reports both modes separately rather than attributing `Run`'s limits to the provider as a whole. Documents the `PermissionMode` mapping onto OpenCode's `edit`/`bash` axes, why plan denies both categories instead of selecting the planning agent, why an empty tool allowlist becomes a wildcard deny, why a non-empty allowlist and the system prompt are carried as a prompt prefix, and that `Serve` needs a transport running the provider on the SDK host. Also states plainly that reading transcripts from OpenCode's local database is not implemented, so the gap is visible rather than inferred. --- CHANGELOG.md | 39 +++++++++++++++++++ README.md | 7 +++- docs/reference/api.md | 91 ++++++++++++++++++++++++++++++++++++------- 3 files changed, 121 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aa80e7d..cad1b32 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,40 @@ Versioning and Keep a Changelog conventions. ### Added +- Bidirectional OpenCode support through `opencode serve`. `OpenCodeTurnMode` + selects the transport; `OpenCode::serve()` starts the server on a reserved + loopback port and drives it over HTTP and Server-Sent Events, adding live + approvals (`once`/`always`/`reject`), incremental text and reasoning deltas, + tool lifecycle events, and a cooperative `session/abort` on cancellation. + `PermissionSupport` reports `live_approvals` in that mode. The default + `opencode run --format json` transport is unchanged and still reports + `live_approvals: false`. +- Enforced per-turn permissions for OpenCode. `Serve` mode supplies the policy + through `OPENCODE_CONFIG_CONTENT`, which the server reads instead of the + ambient configuration, so the requested policy is the one the harness runs + under. `PermissionMode` maps onto OpenCode's `edit`/`bash` axes as + `Default`/`Custom` = ask/ask, `AcceptEdits` = allow/ask, `FullAccess` = + allow/allow and `Plan` = deny/deny. An empty + `LaunchContext::allowed_tools` becomes a `{"*": "deny"}` wildcard. A plan + turn and an empty allowlist additionally refuse any permission that reaches + the adapter without consulting the application. `Run` mode had no + enforcement an application could rely on: it accepts only `--auto` and + `--agent plan`, leaving every other policy to the machine's own + configuration. +- Turn-scoped stdio and HTTP MCP servers for OpenCode in `Serve` mode, + translated into native `local` and `remote` `mcp` entries with credentials + referenced as `{env:NAME}` rather than serialized. + `LaunchContextCapabilities` now advertises `stdio_mcp`, `http_mcp`, + `system_prompt_append` and `allowed_tools` for that mode; the latter two are + carried as a prompt prefix, which is the only channel OpenCode offers. +- `AgentAdapter::attach`, returning optional `ProtocolStreams`, lets an adapter + carry a turn on streams of its own instead of the child's stdout and stdin, + for a provider whose protocol is not on its own stdio. The frame contract is + unchanged, so `parse_line` stays one synchronous state machine and + cancellation, interrupts, interaction timeouts and line bounding are shared + by both kinds of provider. `AgentAdapter::command_for_turn` exposes the state + `prepare_turn` seeded, which now runs before the command is built. + - Bidirectional Codex support through `codex app-server`. `CodexTurnMode` selects the transport; `Codex::app_server()` drives JSON-RPC over stdio with live approvals (`accept`/`acceptForSession`/`decline`), @@ -131,6 +165,11 @@ Versioning and Keep a Changelog conventions. - Turn cancellation now interrupts pending approval and question handlers and terminates the supervised provider instead of waiting for the interaction timeout. +- A turn carried on adapter-supplied protocol streams now terminates its child + as the normal shutdown instead of waiting for an exit that never comes: + `opencode serve` is a server and does not stop because a turn ended. The turn + loop also stops reading at a terminal frame in that mode, so a carrier that + never closes its reader cannot hang a turn. ## [0.1.0] - 2026-08-31 diff --git a/README.md b/README.md index 97fd303..cbd3d75 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,8 @@ APIs, own credentials, or silently fall back to an unsandboxed process. - Typed requests, streaming events, terminal results, and errors - Claude Code, Codex, and OpenCode adapters behind independent Cargo features +- Live approvals with enforced per-turn permission policy: Codex through + `codex app-server`, OpenCode through `opencode serve` - Bounded concurrent turns and bounded provider output - Deadlines, cooperative cancellation, and process-tree cleanup - Long-running tool processes preserved after natural turn completion by @@ -167,8 +169,9 @@ For context meters and automatic/manual compaction in a durable host, read transport and bounded host-extension helpers use fixed POSIX shell scripts; untrusted values remain positional arguments and managed paths are derived from validated names. -- Prompts are not logged. Claude and Codex prompts use stdin. OpenCode's current - headless CLI accepts the message as an argument, so it may be visible to local +- Prompts are not logged. Claude and Codex prompts use stdin, and OpenCode's + `Serve` turn mode posts the prompt in an HTTP body. OpenCode's headless + `Run` mode accepts the message as an argument, so it may be visible to local process inspection; see the capability matrix. - The runtime keeps no database and emits no telemetry. The embedding application owns both. diff --git a/docs/reference/api.md b/docs/reference/api.md index d5bd034..d0315aa 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -9,7 +9,7 @@ pin and test the CLI versions used in production. | --- | --- | --- | | `claude` | yes | `providers::Claude` stream-JSON adapter | | `codex` | yes | `providers::Codex` `exec --json` and `app-server` adapter | -| `opencode` | yes | `providers::OpenCode` JSON adapter | +| `opencode` | yes | `providers::OpenCode` `run --format json` and `serve` adapter | | `nono` | yes | profile management and per-turn Nono execution | | `tailnet` | yes | per-agent userspace Tailscale daemons and split proxy | | `ssh` | yes | OpenSSH execution transport | @@ -32,24 +32,24 @@ features. | Context-window occupancy | yes, estimated from native usage components and direct after compaction | no | no | | Configurable automatic compaction | yes | no | no | | Provider-native manual compaction | yes, retained runtime with an existing session | no | no | -| Session identifier | yes | yes | when reported | +| Session identifier | yes | yes | yes in `Serve` mode; when reported in `Run` | | Resume by session identifier | yes | yes | yes | -| `Default` | yes | yes, static workspace sandbox | yes, asks auto-reject in headless mode | -| `AcceptEdits` | yes | yes, static workspace sandbox | no; rejected rather than broadening access | -| `Plan` | yes | yes, read-only sandbox | yes, built-in `plan` agent | -| `FullAccess` | yes | yes | yes, `--auto`; explicit configured denies remain | +| `Default` | yes | yes, static workspace sandbox | `Serve`: `edit: ask`, `bash: ask`; `Run`: asks auto-reject in headless mode | +| `AcceptEdits` | yes | yes, static workspace sandbox | `Serve`: `edit: allow`, `bash: ask`; `Run`: rejected rather than broadening access | +| `Plan` | yes | yes, read-only sandbox | `Serve`: both categories denied; `Run`: built-in `plan` agent | +| `FullAccess` | yes | yes | `Serve`: `edit: allow`, `bash: allow`; `Run`: `--auto`, explicit configured denies remain | | Custom mode | yes | yes, native approval policy | yes, configured agent | -| Live approvals | yes | app-server mode only (`exec` is configured non-interactively) | no (`run` is non-interactive) | -| Live user questions | yes | app-server mode only, blocking and async | no | -| Cooperative interrupt on cancellation | no | app-server mode only (`turn/interrupt`) | no | -| Structured launch context | system prompt, exact tools, stdio/HTTP MCP, strict MCP | additive stdio/HTTP MCP | rejected | +| Live approvals | yes | app-server mode only (`exec` is configured non-interactively) | `Serve` mode only (`run` is non-interactive) | +| Live user questions | yes | app-server mode only, blocking and async | no; OpenCode has no question channel | +| Cooperative interrupt on cancellation | no | app-server mode only (`turn/interrupt`) | `Serve` mode only (`session/abort`) | +| Structured launch context | system prompt, exact tools, stdio/HTTP MCP, strict MCP | additive stdio/HTTP MCP | `Serve`: system prompt and tools as a prompt prefix, stdio/HTTP MCP; `Run`: rejected | | Native image attachments | no; described as prompt paths | yes (`--image`, `localImage` input) | no; described as prompt paths | | Prompt kept out of argv | yes | yes | no; current `run` CLI uses message args | -| Current backend | CLI stream JSON | `codex exec --json` (default) or `codex app-server` | `opencode run --format json` | +| Current backend | CLI stream JSON | `codex exec --json` (default) or `codex app-server` | `opencode run --format json` (default) or `opencode serve` | -The public adapter trait is the extension point for OpenCode server or -SDK-backed adapters. Such adapters should preserve the normalized contract and -establish compatibility coverage before replacing a CLI adapter. +The public adapter trait is the extension point for further provider +backends. Such adapters should preserve the normalized contract and establish +compatibility coverage before replacing a CLI adapter. ### Codex turn modes @@ -78,6 +78,69 @@ resumed thread identifier travel in `thread/start`/`thread/resume` and `turn/start` instead of argv; turn-scoped stdio and HTTP MCP servers and the model relay still use `--config` overrides. +### OpenCode turn modes + +`providers::OpenCodeTurnMode` selects how an OpenCode turn runs. `Run` (the +default) keeps the one-way `opencode run --format json` behavior. + +`Run` mode has no permission enforcement an application can rely on. The only +flags the CLI accepts are `--auto`, which approves everything, and +`--agent plan`; every other policy comes from whatever `opencode` +configuration exists on the machine. A caller cannot request "ask before +running a shell command", and a silently auto-refused tool call is +indistinguishable from a turn that simply produced no text. + +`Serve`, selected with `OpenCode::serve()` or +`OpenCode::default().with_turn_mode(OpenCodeTurnMode::Serve)`, starts +`opencode serve` on a reserved loopback port and drives it over HTTP and +Server-Sent Events. It adds: + +- a per-turn permission policy supplied through `OPENCODE_CONFIG_CONTENT`, + which the server reads *instead of* the ambient configuration, so the + requested policy is the one the harness runs under; +- live approvals for `permission.asked`, answered through + `InteractionHandler::approve` and posted to + `/session/{id}/permissions/{id}`. `ApprovalDecision::Allow`, + `ApprovalDecision::AllowForSession` and `ApprovalDecision::Deny` map to the + native `once`, `always` and `reject` responses; +- turn-scoped stdio and HTTP MCP servers, translated into OpenCode's `local` + and `remote` `mcp` entries. Credentials are referenced as `{env:NAME}` and + never serialized into the configuration; +- incremental text and reasoning deltas, tool lifecycle events, and a + cooperative `session/abort` when the turn's `CancellationToken` fires. + +`PermissionMode` maps onto OpenCode's two permission axes: + +| `PermissionMode` | `edit` | `bash` | +| --- | --- | --- | +| `Default`, `Custom` | `ask` | `ask` | +| `AcceptEdits` | `allow` | `ask` | +| `FullAccess` | `allow` | `allow` | +| `Plan` | `deny` | `deny` | + +`Plan` denies both categories rather than selecting the planning agent. +OpenCode's read-only tools are gated by neither category, so a plan turn can +still inspect the workspace but can never have a side effect. + +An empty `LaunchContext::allowed_tools` list becomes a `{"*": "deny"}` +wildcard rule, because an explicitly empty tool set has to be an enforcement +boundary rather than a suggestion in the prompt. A *non-empty* list, and +`LaunchContext::system_prompt_append`, are carried as a prompt prefix: +OpenCode has no system-prompt or tool-restriction field on its prompt body. + +A plan turn and an empty tool allowlist also refuse any permission that +reaches the adapter without consulting the application. Both are already +denied by the policy the server started with, so arriving there means the +policy did not hold — and the turn promised the user that no such choice +would exist. + +`Serve` mode reaches the server on the SDK host's loopback interface, so it +requires a transport that runs the provider on that host. Session resume is +validated with `session.get` rather than by listing and filtering, which +cannot fail closed on a workspace reached through a symlink; a session with a +parent is a subagent session and is rejected immediately rather than left to +hang. Reading transcripts from OpenCode's local database is not implemented. + ## Private-network providers `NetworkProviderRegistry` stores heterogeneous, trusted in-process From ead8f3ba49a93c613bbc67af8b736e2556034f53 Mon Sep 17 00:00:00 2001 From: David Viejo Date: Mon, 21 Sep 2026 18:57:49 +0200 Subject: [PATCH 6/6] docs(opencode): record why attachment parts stay additive and unadvertised --- src/providers/opencode_serve.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/providers/opencode_serve.rs b/src/providers/opencode_serve.rs index 4df5c80..2da5abc 100644 --- a/src/providers/opencode_serve.rs +++ b/src/providers/opencode_serve.rs @@ -290,6 +290,13 @@ pub(super) fn prepare_turn(request: &TurnRequest, state: &mut AdapterState, port if !prompt.is_empty() { parts.push(json!({"type": "text", "text": prompt})); } + // Best-effort, and deliberately additive. The reference driver sends a + // `file` part built from a URL it already has; the SDK only has an + // execution-host path, and that `file://` spelling has not been confirmed + // against a live server. So this adapter does *not* advertise + // `TurnCapabilities::native_image_attachments`: the caller still describes + // the files in the prompt, and a part OpenCode ignores costs nothing, + // whereas dropping the attachment silently would lose it. for attachment in &request.attachments { parts.push(json!({ "type": "file",