use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{anyhow, Result};
use async_trait::async_trait;
use reqwest::Method;
use serde_json::{json, Value};
use tokio::process::{Child, Command};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use crate::chat::types::{ConversationEvent, Lifecycle};
use crate::engine::agent::AgentConfig;
use crate::engine::config::parse_agent;
use crate::harness::common::{spawn_stderr_logger, TurnInProgressGuard};
use crate::harness::{
opencode_mapping, opencode_runtime, ApprovalPolicy, Capabilities, Harness, HarnessError,
RawProviderEvent,
};
const OPENCODE_DISCONNECTED_CODE: &str = "opencode_disconnected";
pub struct OpenCodeHarness {
events: mpsc::UnboundedSender<ConversationEvent>,
raw_provider: Option<mpsc::UnboundedSender<RawProviderEvent>>,
client: reqwest::Client,
approval: ApprovalPolicy,
config: Option<AgentConfig>,
should_seed_prompt: bool,
turn_in_progress: Arc<AtomicBool>,
shutdown_requested: Arc<AtomicBool>,
interrupt_requested: Arc<AtomicBool>,
child: Option<Child>,
stderr_task: Option<JoinHandle<()>>,
sse_task: Option<JoinHandle<()>>,
server_base_url: Option<String>,
provider_session_id: Option<String>,
}
impl std::fmt::Debug for OpenCodeHarness {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OpenCodeHarness").finish()
}
}
impl OpenCodeHarness {
pub fn new(events: mpsc::UnboundedSender<ConversationEvent>, approval: ApprovalPolicy) -> Self {
Self {
events,
raw_provider: None,
client: reqwest::Client::new(),
approval,
config: None,
should_seed_prompt: true,
turn_in_progress: Arc::new(AtomicBool::new(false)),
shutdown_requested: Arc::new(AtomicBool::new(false)),
interrupt_requested: Arc::new(AtomicBool::new(false)),
child: None,
stderr_task: None,
sse_task: None,
server_base_url: None,
provider_session_id: None,
}
}
async fn start_inner(&mut self, config: &AgentConfig) -> Result<()> {
let port = allocate_port()?;
let mut command = Command::new("opencode");
command
.arg("serve")
.arg("--port")
.arg(port.to_string())
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::piped());
if let Some(cwd) = &config.cwd {
command.current_dir(cwd);
}
super::configure_vendor_tokio_env(&mut command)?;
let mut child = command
.spawn()
.map_err(|err| anyhow!("failed to spawn opencode serve: {err}"))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| anyhow!("missing opencode stderr"))?;
let base_url = format!("http://127.0.0.1:{port}");
if let Err(err) = wait_for_server(&self.client, &base_url, &mut child).await {
shutdown_child(&mut child).await;
return Err(err);
}
let provider_session_id = match create_provider_session(&self.client, &base_url).await {
Ok(provider_session_id) => provider_session_id,
Err(err) => {
shutdown_child(&mut child).await;
return Err(err);
}
};
let event_tx = self.events.clone();
let raw_provider = self.raw_provider.clone();
let client = self.client.clone();
let shutdown_requested = self.shutdown_requested.clone();
let turn_in_progress = self.turn_in_progress.clone();
let interrupt_requested = self.interrupt_requested.clone();
let approval = self.approval;
let reader_base_url = base_url.clone();
let reader_session_id = provider_session_id.clone();
let sse_task = tokio::spawn(async move {
let stream_url = format!("{reader_base_url}/event");
let request = client
.get(&stream_url)
.header(reqwest::header::ACCEPT, "text/event-stream");
let mut response = match request.send().await {
Ok(response) => response,
Err(err) => {
send_disconnect_error(
&event_tx,
&shutdown_requested,
format!("failed to connect to OpenCode SSE stream: {err}"),
);
return;
}
};
if let Err(err) = response.error_for_status_ref() {
send_disconnect_error(
&event_tx,
&shutdown_requested,
format!("OpenCode SSE stream failed: {err}"),
);
return;
}
let mut parser = SseParser::default();
let mut state = opencode_mapping::ReaderState::new(reader_session_id.clone());
loop {
if shutdown_requested.load(Ordering::Relaxed) {
break;
}
let chunk = match response.chunk().await {
Ok(chunk) => chunk,
Err(err) => {
tracing::warn!(error = %err, "opencode SSE chunk read failed");
break;
}
};
let Some(chunk) = chunk else {
break;
};
for payload in parser.push(&chunk) {
if payload.trim().is_empty() || payload.trim() == "[DONE]" {
continue;
}
if let Some(raw_provider) = &raw_provider {
let _ = raw_provider.send(RawProviderEvent {
stream: "sse",
line: payload.clone(),
});
}
let parsed = serde_json::from_str::<Value>(&payload);
let raw = match parsed {
Ok(raw) => raw,
Err(err) => {
tracing::debug!(error = %err, payload = %payload, "invalid SSE data");
continue;
}
};
let mapped = opencode_mapping::map_event(&raw, &mut state);
for event in mapped.events {
let event = match event {
ConversationEvent::TurnStarted { turn_id } => {
turn_in_progress.store(true, Ordering::SeqCst);
interrupt_requested.store(false, Ordering::SeqCst);
ConversationEvent::TurnStarted { turn_id }
}
ConversationEvent::TurnCompleted { turn_id, status } => {
turn_in_progress.store(false, Ordering::SeqCst);
let status = if interrupt_requested.swap(false, Ordering::SeqCst) {
Lifecycle::Interrupted
} else {
status
};
ConversationEvent::TurnCompleted { turn_id, status }
}
other => other,
};
let _ = event_tx.send(event);
}
for request_id in mapped.permission_requests {
if let Err(err) = answer_permission(
&client,
&reader_base_url,
&reader_session_id,
&request_id,
approval,
)
.await
{
tracing::warn!(
request_id = %request_id,
error = %err,
"failed to answer OpenCode permission request"
);
}
}
}
}
turn_in_progress.store(false, Ordering::SeqCst);
send_disconnect_error(
&event_tx,
&shutdown_requested,
"OpenCode event stream disconnected",
);
});
let stderr_task = spawn_stderr_logger(stderr, "harness::opencode");
let opencode_pid = child.id();
if let Some(pid) = opencode_pid {
if let Err(err) = opencode_runtime::register_opencode_server(pid) {
tracing::warn!(
opencode_pid = pid,
error = %err,
"failed to register OpenCode server runtime metadata"
);
}
}
self.child = Some(child);
self.stderr_task = Some(stderr_task);
self.sse_task = Some(sse_task);
self.server_base_url = Some(base_url);
self.provider_session_id = Some(provider_session_id);
Ok(())
}
}
#[async_trait]
impl Harness for OpenCodeHarness {
fn set_raw_provider_sender(
&mut self,
raw_provider: Option<mpsc::UnboundedSender<RawProviderEvent>>,
) {
self.raw_provider = raw_provider;
}
async fn start(&mut self, config: &AgentConfig) -> Result<()> {
if self.child.is_some() {
return Ok(());
}
self.shutdown_requested.store(false, Ordering::SeqCst);
self.config = Some(config.clone());
self.should_seed_prompt = true;
let start_result = self.start_inner(config).await;
if let Err(err) = start_result {
let _ = self.stop().await;
return Err(err);
}
Ok(())
}
async fn send_input(&mut self, content: &str) -> Result<()> {
let config = self
.config
.as_ref()
.ok_or_else(|| anyhow!("opencode harness not started"))?;
let first_turn = self.should_seed_prompt;
let turn_content = build_turn_content(content, config, first_turn);
let Some(turn_content) = turn_content else {
return Ok(());
};
self.interrupt_requested.store(false, Ordering::SeqCst);
if self
.turn_in_progress
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_err()
{
return Err(HarnessError::TurnAlreadyInProgress.into());
}
let mut turn_guard = TurnInProgressGuard::new(self.turn_in_progress.clone());
let base_url = self
.server_base_url
.clone()
.ok_or_else(|| anyhow!("opencode server not started"))?;
let provider_session_id = self
.provider_session_id
.clone()
.ok_or_else(|| anyhow!("opencode provider session id is not available"))?;
let payload = build_turn_payload(&turn_content, config, first_turn);
let message_url = format!("{base_url}/session/{provider_session_id}/message");
send_request_with_retry(&self.client, Method::POST, &message_url, Some(payload)).await?;
self.should_seed_prompt = false;
turn_guard.disarm();
Ok(())
}
async fn interrupt(&mut self) -> Result<()> {
if !self.turn_in_progress.load(Ordering::SeqCst) {
return Ok(());
}
let base_url = self
.server_base_url
.clone()
.ok_or_else(|| anyhow!("opencode server not started"))?;
let provider_session_id = self
.provider_session_id
.clone()
.ok_or_else(|| anyhow!("opencode provider session id is not available"))?;
self.interrupt_requested.store(true, Ordering::SeqCst);
let abort_url = format!("{base_url}/session/{provider_session_id}/abort");
send_request_with_retry(&self.client, Method::POST, &abort_url, Some(json!({}))).await?;
Ok(())
}
async fn stop(&mut self) -> Result<()> {
self.shutdown_requested.store(true, Ordering::SeqCst);
if let (Some(base_url), Some(provider_session_id)) =
(&self.server_base_url, &self.provider_session_id)
{
let abort_url = format!("{base_url}/session/{provider_session_id}/abort");
let _ =
send_request_with_retry(&self.client, Method::POST, &abort_url, Some(json!({})))
.await;
let delete_url = format!("{base_url}/session/{provider_session_id}");
let _ = send_request_with_retry(&self.client, Method::DELETE, &delete_url, None).await;
}
let opencode_pid = self.child.as_ref().and_then(|child| child.id());
if let Some(child) = self.child.as_mut() {
shutdown_child(child).await;
}
self.child = None;
if let Some(pid) = opencode_pid {
if let Err(err) = opencode_runtime::unregister_opencode_server(pid) {
tracing::warn!(
opencode_pid = pid,
error = %err,
"failed to unregister OpenCode server runtime metadata"
);
}
}
if let Some(task) = self.sse_task.take() {
task.abort();
let _ = task.await;
}
if let Some(task) = self.stderr_task.take() {
task.abort();
let _ = task.await;
}
self.turn_in_progress.store(false, Ordering::SeqCst);
self.provider_session_id = None;
self.server_base_url = None;
Ok(())
}
fn capabilities(&self) -> Capabilities {
Capabilities {
supports_steer: false,
}
}
fn provider_session_id(&self) -> Option<String> {
self.provider_session_id.clone()
}
fn set_provider_session_id(&mut self, provider_session_id: Option<String>) {
self.provider_session_id = provider_session_id;
}
}
async fn shutdown_child(child: &mut Child) {
let _ = child.start_kill();
let _ = child.wait().await;
}
async fn create_provider_session(client: &reqwest::Client, base_url: &str) -> Result<String> {
let session_url = format!("{base_url}/session");
let response =
send_request_with_retry(client, Method::POST, &session_url, Some(json!({}))).await?;
let body: Value = response
.json()
.await
.map_err(|err| anyhow!("failed to parse opencode session response: {err}"))?;
parse_session_id(&body).ok_or_else(|| {
anyhow!(
"opencode session response did not include session id: {}",
body
)
})
}
fn send_disconnect_error(
event_tx: &mpsc::UnboundedSender<ConversationEvent>,
shutdown_requested: &AtomicBool,
message: impl Into<String>,
) {
if shutdown_requested.load(Ordering::Relaxed) {
return;
}
let _ = event_tx.send(ConversationEvent::Error {
code: OPENCODE_DISCONNECTED_CODE.to_string(),
message: message.into(),
});
}
fn allocate_port() -> Result<u16> {
let listener = std::net::TcpListener::bind("127.0.0.1:0")
.map_err(|err| anyhow!("failed to allocate port for OpenCode: {err}"))?;
let port = listener
.local_addr()
.map_err(|err| anyhow!("failed to read allocated OpenCode port: {err}"))?
.port();
Ok(port)
}
async fn wait_for_server(
client: &reqwest::Client,
base_url: &str,
child: &mut Child,
) -> Result<()> {
let deadline = Instant::now() + Duration::from_secs(15);
let mut delay = Duration::from_millis(100);
loop {
if let Some(exit_status) = child
.try_wait()
.map_err(|err| anyhow!("failed to poll opencode serve process: {err}"))?
{
return Err(anyhow!(
"opencode serve exited before becoming ready: {exit_status}"
));
}
if client.get(base_url).send().await.is_ok() {
return Ok(());
}
if Instant::now() >= deadline {
return Err(anyhow!(
"timed out waiting for opencode serve health check at {base_url}"
));
}
tokio::time::sleep(delay).await;
delay = std::cmp::min(delay * 2, Duration::from_secs(1));
}
}
async fn send_request_with_retry(
client: &reqwest::Client,
method: Method,
url: &str,
payload: Option<Value>,
) -> Result<reqwest::Response> {
let mut attempt = 0;
loop {
let mut request = client.request(method.clone(), url);
if let Some(body) = payload.clone() {
request = request.json(&body);
}
match request.send().await {
Ok(response) if response.status().is_server_error() && attempt == 0 => {
attempt += 1;
tokio::time::sleep(Duration::from_millis(200)).await;
}
Ok(response) => {
return response
.error_for_status()
.map_err(|err| anyhow!("OpenCode request failed ({method} {url}): {err}"));
}
Err(err) if attempt == 0 && (err.is_timeout() || err.is_connect()) => {
attempt += 1;
tokio::time::sleep(Duration::from_millis(200)).await;
}
Err(err) => {
return Err(anyhow!("OpenCode request failed ({method} {url}): {err}"));
}
}
}
}
async fn answer_permission(
client: &reqwest::Client,
base_url: &str,
session_id: &str,
request_id: &str,
approval: ApprovalPolicy,
) -> Result<()> {
let url = format!("{base_url}/session/{session_id}/permissions/{request_id}");
let response = match approval {
ApprovalPolicy::AutoApprove => "always",
};
let payload = json!({ "response": response });
let _ = send_request_with_retry(client, Method::POST, &url, Some(payload)).await?;
Ok(())
}
fn parse_session_id(value: &Value) -> Option<String> {
value
.get("id")
.and_then(Value::as_str)
.map(ToString::to_string)
}
fn build_turn_content(content: &str, config: &AgentConfig, first_turn: bool) -> Option<String> {
if first_turn {
let mut parts = Vec::new();
if !config.task_prompt.trim().is_empty() {
parts.push(config.task_prompt.trim().to_string());
}
if !content.trim().is_empty() {
parts.push(content.trim().to_string());
}
if parts.is_empty() {
None
} else {
Some(parts.join("\n\n"))
}
} else {
let text = content.trim();
if text.is_empty() {
None
} else {
Some(text.to_string())
}
}
}
fn build_turn_payload(content: &str, config: &AgentConfig, first_turn: bool) -> Value {
let mut payload = json!({
"parts": [
{ "type": "text", "text": content }
]
});
if first_turn && !config.system_prompt.trim().is_empty() {
payload["system"] = Value::String(config.system_prompt.trim().to_string());
}
if let Some((provider_id, model_id)) = opencode_model(config) {
payload["model"] = json!({
"providerID": provider_id,
"modelID": model_id,
});
}
payload
}
fn opencode_model(config: &AgentConfig) -> Option<(String, String)> {
let agent_str = config.agent.as_deref()?;
let (harness, variant) = parse_agent(agent_str);
if harness != "opencode" {
return None;
}
let variant = variant?;
let (provider_id, model_id) = variant.split_once('/')?;
if provider_id.is_empty() || model_id.is_empty() {
return None;
}
Some((provider_id.to_string(), model_id.to_string()))
}
#[derive(Debug, Default)]
struct SseParser {
buffer: String,
}
impl SseParser {
fn push(&mut self, chunk: &[u8]) -> Vec<String> {
self.buffer.push_str(&String::from_utf8_lossy(chunk));
if self.buffer.contains('\r') {
self.buffer = self.buffer.replace("\r\n", "\n").replace('\r', "\n");
}
let mut events = Vec::new();
while let Some(separator) = self.buffer.find("\n\n") {
let frame = self.buffer[..separator].to_string();
self.buffer.drain(..separator + 2);
if let Some(data) = parse_data_frame(&frame) {
events.push(data);
}
}
events
}
}
fn parse_data_frame(frame: &str) -> Option<String> {
let mut data_lines = Vec::new();
for line in frame.lines() {
if let Some(data) = line.strip_prefix("data:") {
data_lines.push(data.trim_start().to_string());
}
}
if data_lines.is_empty() {
None
} else {
Some(data_lines.join("\n"))
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn build_turn_content_includes_task_prompt_on_first_turn() {
let content = build_turn_content(
"",
&AgentConfig {
task_prompt: "task".to_string(),
..Default::default()
},
true,
);
assert_eq!(content.as_deref(), Some("task"));
}
#[test]
fn sse_parser_collects_data_lines() {
let mut parser = SseParser::default();
let events = parser.push(b"event: message\ndata: {\"a\":1}\n\n");
assert_eq!(events, vec!["{\"a\":1}".to_string()]);
}
#[test]
fn sse_parser_handles_split_crlf_frames() {
let mut parser = SseParser::default();
assert!(parser.push(b"data: {\"a\":").is_empty());
let events = parser.push(b"1}\r\n\r\n");
assert_eq!(events, vec!["{\"a\":1}".to_string()]);
}
#[test]
fn parse_session_id_requires_canonical_top_level_id() {
assert_eq!(
parse_session_id(&json!({"id": "session_1"})),
Some("session_1".to_string())
);
assert_eq!(
parse_session_id(&json!({"session": {"id": "session_2"}})),
None
);
assert_eq!(parse_session_id(&json!({"sessionID": "session_3"})), None);
}
#[test]
fn build_turn_payload_includes_explicit_opencode_model() {
let payload = build_turn_payload(
"hello",
&AgentConfig {
agent: Some("opencode:moonshotai/kimi-k2".to_string()),
..Default::default()
},
false,
);
assert_eq!(
payload.get("model"),
Some(&json!({
"providerID": "moonshotai",
"modelID": "kimi-k2"
}))
);
}
#[test]
fn build_turn_payload_omits_model_for_non_opencode_agent_model() {
let payload = build_turn_payload(
"hello",
&AgentConfig {
agent: Some("claude:sonnet".to_string()),
..Default::default()
},
false,
);
assert!(payload.get("model").is_none());
}
#[test]
fn build_turn_payload_uses_loopflow_default_for_bare_opencode() {
let payload = build_turn_payload(
"hello",
&AgentConfig {
agent: Some("opencode".to_string()),
..Default::default()
},
false,
);
assert_eq!(
payload.get("model"),
Some(&json!({
"providerID": "opencode",
"modelID": "glm-5.2"
}))
);
}
}