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 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/providers/mod.rs b/src/providers/mod.rs index 239b17e..cecd75a 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -8,13 +8,17 @@ mod codex; mod codex_app_server; #[cfg(feature = "opencode")] 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 new file mode 100644 index 0000000..473dc70 --- /dev/null +++ b/src/providers/opencode_http.rs @@ -0,0 +1,751 @@ +//! 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"; +/// 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. +/// +/// 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 { + // 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(runtime_reader), + writer: Box::new(runtime_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, incoming: DuplexStream, outgoing: DuplexStream) { + 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 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() + ); + 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((200, framing)) => 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; + } + }; + + 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 { + 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); + } + 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 && !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 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")); + } + + /// 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 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..2da5abc --- /dev/null +++ b/src/providers/opencode_serve.rs @@ -0,0 +1,1323 @@ +//! 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})); + } + // 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", + "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"}) + ); + } +} diff --git a/src/runtime.rs b/src/runtime.rs index fc084c7..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( @@ -2265,14 +2269,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 +2388,27 @@ 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 mut attached = false; + let reader: crate::TransportReader = match adapter.attach(request, &state).await { + Ok(Some(streams)) => { + attached = true; + 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() => { @@ -2483,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); +}