supercode_harness/runtime/

adapters.rs

//! Harness-specific live-runtime adapters built on the primitive contracts.

use std::collections::{BTreeMap, VecDeque};
use std::net::TcpListener;
use std::path::Path;
use std::process::Stdio;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use futures::StreamExt;
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStdin, Command};
use tokio::sync::{mpsc, Mutex};

use super::{
    HarnessEvent, JsonLineClient, RuntimeAttachRequest, RuntimeBackend, RuntimeCapabilities,
    RuntimeConnection, RuntimeEndpoint, RuntimeHandle, RuntimeInput, RuntimeLaunch,
    RuntimeStartRequest,
};
use crate::{Error, HarnessId, Result};

/// Pi live-runtime backend using `pi --mode rpc` JSONL.
#[derive(Debug, Clone)]
pub struct PiRuntimeBackend {
    launch: RuntimeLaunch,
}

impl Default for PiRuntimeBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl PiRuntimeBackend {
    /// Use `pi --mode rpc` from `PATH`.
    pub fn new() -> Self {
        Self {
            launch: RuntimeLaunch {
                program: "pi".into(),
                arguments: vec!["--mode".into(), "rpc".into()],
                env: BTreeMap::new(),
            },
        }
    }

    /// Use an explicit Pi RPC command prefix.
    pub fn with_launch(launch: RuntimeLaunch) -> Self {
        Self { launch }
    }

    async fn open(
        &self,
        cwd: &Path,
        runtime_id: String,
        launch: Option<RuntimeLaunch>,
        resume: bool,
    ) -> Result<Box<dyn RuntimeConnection>> {
        let mut launch = launch.unwrap_or_else(|| self.launch.clone());
        if resume {
            launch
                .arguments
                .extend(["--session".into(), runtime_id.clone()]);
        } else {
            launch
                .arguments
                .extend(["--session-id".into(), runtime_id.clone()]);
        }
        let transport = RawLineTransport::spawn(&launch, Some(cwd), "pi-rpc-jsonl").await?;
        let handle = RuntimeHandle {
            harness: HarnessId::from(HarnessId::PI),
            runtime_id,
            endpoint: transport.endpoint.clone(),
        };
        Ok(Box::new(PiRuntimeConnection {
            handle,
            transport,
            next_request: 1,
        }))
    }
}

#[async_trait]
impl RuntimeBackend for PiRuntimeBackend {
    fn harness(&self) -> HarnessId {
        HarnessId::from(HarnessId::PI)
    }

    fn capabilities(&self) -> RuntimeCapabilities {
        RuntimeCapabilities {
            start_session: true,
            resume_session: true,
            attach_existing_process: false,
            send_input: true,
            stream_events: true,
            interrupt: true,
            respond_to_requests: true,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        self.open(&request.cwd, generated_session_id(), request.launch, false)
            .await
    }

    async fn attach(&self, request: RuntimeAttachRequest) -> Result<Box<dyn RuntimeConnection>> {
        let cwd = request.cwd.unwrap_or(std::env::current_dir()?);
        self.open(&cwd, request.runtime_id, request.launch, true)
            .await
    }
}

struct PiRuntimeConnection {
    handle: RuntimeHandle,
    transport: RawLineTransport,
    next_request: u64,
}

#[async_trait]
impl RuntimeConnection for PiRuntimeConnection {
    fn handle(&self) -> &RuntimeHandle {
        &self.handle
    }

    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>> {
        let id = format!("supercode-{}", self.next_request);
        self.next_request += 1;
        self.transport
            .write(json!({"id": id, "type": "prompt", "message": input.text}))
            .await?;
        Ok(Some(id))
    }

    async fn next_event(&mut self) -> Result<Option<HarnessEvent>> {
        raw_next_event(&mut self.transport.receiver).await
    }

    async fn interrupt(&mut self) -> Result<()> {
        self.transport.write(json!({"type": "abort"})).await
    }

    async fn respond(&mut self, request_id: Value, mut response: Value) -> Result<()> {
        if let Value::Object(object) = &mut response {
            object.entry("id").or_insert(request_id);
            self.transport.write(response).await
        } else {
            self.transport
                .write(json!({"id": request_id, "response": response}))
                .await
        }
    }

    async fn close(&mut self) -> Result<()> {
        self.transport.close().await
    }
}

/// Claude Code live-runtime backend using bidirectional stream-json print
/// mode. It can create/resume sessions and cancel the running turn through the
/// stream-json control channel; the print-mode protocol still exposes no
/// permission-response primitive to this adapter.
#[derive(Debug, Clone)]
pub struct ClaudeCodeRuntimeBackend {
    launch: RuntimeLaunch,
}

impl Default for ClaudeCodeRuntimeBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl ClaudeCodeRuntimeBackend {
    /// Use `claude` from `PATH` in bidirectional stream-json mode.
    pub fn new() -> Self {
        Self {
            launch: RuntimeLaunch {
                program: "claude".into(),
                arguments: vec![
                    "--print".into(),
                    "--input-format".into(),
                    "stream-json".into(),
                    "--output-format".into(),
                    "stream-json".into(),
                    "--verbose".into(),
                ],
                env: BTreeMap::new(),
            },
        }
    }

    /// Use an explicit Claude Code stream-json command prefix.
    pub fn with_launch(launch: RuntimeLaunch) -> Self {
        Self { launch }
    }

    async fn open(
        &self,
        cwd: &Path,
        runtime_id: String,
        launch: Option<RuntimeLaunch>,
        resume: bool,
    ) -> Result<Box<dyn RuntimeConnection>> {
        let mut launch = launch.unwrap_or_else(|| self.launch.clone());
        launch.arguments.extend(if resume {
            vec!["--resume".into(), runtime_id.clone()]
        } else {
            vec!["--session-id".into(), runtime_id.clone()]
        });
        let transport = RawLineTransport::spawn(&launch, Some(cwd), "claude-stream-json").await?;
        Ok(Box::new(ClaudeRuntimeConnection {
            handle: RuntimeHandle {
                harness: HarnessId::from(HarnessId::CLAUDE_CODE),
                runtime_id,
                endpoint: transport.endpoint.clone(),
            },
            transport,
            buffered_events: VecDeque::new(),
            next_control_request: 1,
            control_timeout: CLAUDE_CONTROL_RESPONSE_TIMEOUT,
        }))
    }
}

/// How long `interrupt` waits for the CLI's matching `control_response` before
/// returning a structured error instead of hanging the caller.
///
/// Measured against claude 2.1.224: an interrupt issued while a turn is in
/// flight is acknowledged in ~1 ms, but one issued during process startup —
/// before the CLI has emitted `system/init` — is queued behind session-start
/// hooks and took 1.15 s to acknowledge on a warm box. The bound is set well
/// above the slow case so a legitimately busy startup is never reported as a
/// protocol failure.
const CLAUDE_CONTROL_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);

#[async_trait]
impl RuntimeBackend for ClaudeCodeRuntimeBackend {
    fn harness(&self) -> HarnessId {
        HarnessId::from(HarnessId::CLAUDE_CODE)
    }

    fn capabilities(&self) -> RuntimeCapabilities {
        RuntimeCapabilities {
            start_session: true,
            resume_session: true,
            attach_existing_process: false,
            send_input: true,
            stream_events: true,
            interrupt: true,
            respond_to_requests: false,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        self.open(&request.cwd, generated_session_id(), request.launch, false)
            .await
    }

    async fn attach(&self, request: RuntimeAttachRequest) -> Result<Box<dyn RuntimeConnection>> {
        let cwd = request.cwd.unwrap_or(std::env::current_dir()?);
        self.open(&cwd, request.runtime_id, request.launch, true)
            .await
    }
}

struct ClaudeRuntimeConnection {
    handle: RuntimeHandle,
    transport: RawLineTransport,
    /// Native events read off the transport while `interrupt` was waiting for
    /// its `control_response`. They are handed to `next_event` in arrival order
    /// so cancelling a turn never costs the consumer an event.
    buffered_events: VecDeque<Value>,
    next_control_request: u64,
    control_timeout: Duration,
}

impl ClaudeRuntimeConnection {
    /// The control channel is adapter-private plumbing: a `control_response` is
    /// the reply to a frame this adapter sent, never a harness event, so it is
    /// dropped rather than forwarded to event consumers. A late reply that
    /// arrives after `interrupt` gave up is dropped here too.
    fn is_control_response(value: &Value) -> bool {
        value.get("type").and_then(Value::as_str) == Some("control_response")
    }

    /// Match one `control_response` envelope against an outstanding request id.
    ///
    /// Ground truth (claude 2.1.224, verified live): the CLI answers
    /// `{"type":"control_request","request_id":ID,"request":{"subtype":"interrupt"}}`
    /// with
    /// `{"type":"control_response","response":{"subtype":"success","request_id":ID,"response":{"still_queued":[]}}}`,
    /// or with `{"subtype":"error","request_id":ID,"error":"…"}` on failure.
    fn control_result(value: &Value, request_id: &str) -> Option<Result<()>> {
        let response = value.get("response")?;
        if response.get("request_id").and_then(Value::as_str) != Some(request_id) {
            return None;
        }
        match response.get("subtype").and_then(Value::as_str) {
            Some("success") => Some(Ok(())),
            other => Some(Err(Error::Other(format!(
                "Claude Code rejected the interrupt control request: {}",
                response
                    .get("error")
                    .and_then(Value::as_str)
                    .map(str::to_string)
                    .unwrap_or_else(|| format!(
                        "control_response subtype {}",
                        other.unwrap_or("(missing)")
                    ))
            )))),
        }
    }
}

#[async_trait]
impl RuntimeConnection for ClaudeRuntimeConnection {
    fn handle(&self) -> &RuntimeHandle {
        &self.handle
    }

    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>> {
        self.transport
            .write(json!({
                "type": "user",
                "session_id": self.handle.runtime_id,
                "message": {"role": "user", "content": input.text},
            }))
            .await?;
        Ok(None)
    }

    async fn next_event(&mut self) -> Result<Option<HarnessEvent>> {
        if let Some(payload) = self.buffered_events.pop_front() {
            return Ok(Some(harness_event(payload)));
        }
        loop {
            let Some(payload) = self.transport.receiver.recv().await else {
                return Ok(None);
            };
            if Self::is_control_response(&payload) {
                continue;
            }
            return Ok(Some(harness_event(payload)));
        }
    }

    /// Cancel the running turn through the stream-json control channel and wait
    /// for the CLI's acknowledgement.
    ///
    /// Interrupting with no turn in flight is safe and succeeds: claude 2.1.224
    /// acknowledges the request with `subtype: "success"` and an empty
    /// `still_queued` list rather than erroring, and the session keeps
    /// accepting input. The adapter reports what the harness reports instead of
    /// inventing a turn-state gate of its own.
    async fn interrupt(&mut self) -> Result<()> {
        let request_id = format!(
            "supercode-{}-interrupt-{}",
            self.handle.runtime_id, self.next_control_request
        );
        self.next_control_request += 1;
        self.transport
            .write(json!({
                "type": "control_request",
                "request_id": request_id,
                "request": {"subtype": "interrupt"},
            }))
            .await?;

        let deadline = tokio::time::Instant::now() + self.control_timeout;
        loop {
            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
            if remaining.is_zero() {
                return Err(claude_interrupt_timeout(self.control_timeout));
            }
            match tokio::time::timeout(remaining, self.transport.receiver.recv()).await {
                Err(_) => return Err(claude_interrupt_timeout(self.control_timeout)),
                Ok(None) => return Err(Error::Other(
                    "Claude Code stream-json transport closed before acknowledging the interrupt"
                        .into(),
                )),
                Ok(Some(payload)) => {
                    if Self::is_control_response(&payload) {
                        if let Some(result) = Self::control_result(&payload, &request_id) {
                            return result;
                        }
                        continue;
                    }
                    self.buffered_events.push_back(payload);
                }
            }
        }
    }

    async fn respond(&mut self, _request_id: Value, _response: Value) -> Result<()> {
        Err(unsupported(
            "Claude Code stream-json",
            "respond to protocol requests",
        ))
    }

    async fn close(&mut self) -> Result<()> {
        self.transport.close().await
    }
}

/// Generic ACP v1 client backend for any ACP agent command.
#[derive(Debug, Clone)]
pub struct AcpRuntimeBackend {
    harness: HarnessId,
    launch: RuntimeLaunch,
    resume_session: bool,
}

impl AcpRuntimeBackend {
    /// Construct an ACP adapter for a named harness/agent command.
    pub fn new(harness: HarnessId, launch: RuntimeLaunch) -> Self {
        Self {
            harness,
            launch,
            resume_session: false,
        }
    }

    /// Declare that this known ACP agent advertises `session/load` or
    /// `session/resume`. Attach still validates the capability negotiated by
    /// `initialize`, so a changed or incompatible agent fails honestly.
    pub fn with_resume_support(mut self, supported: bool) -> Self {
        self.resume_session = supported;
        self
    }

    async fn connect(
        &self,
        cwd: &Path,
        launch: Option<RuntimeLaunch>,
    ) -> Result<(
        Arc<JsonLineClient>,
        mpsc::UnboundedReceiver<Value>,
        RuntimeEndpoint,
        Value,
    )> {
        let launch = launch.unwrap_or_else(|| self.launch.clone());
        let (client, receiver, endpoint) =
            JsonLineClient::spawn(&launch, Some(cwd), true, "acp-v1-jsonrpc").await?;
        let initialized = client
            .request(
                "initialize",
                json!({
                    "protocolVersion": 1,
                    "clientCapabilities": {},
                    "clientInfo": {
                        "name": "supercode",
                        "title": "Supercode",
                        "version": env!("CARGO_PKG_VERSION"),
                    },
                }),
            )
            .await?;
        if initialized.get("protocolVersion").and_then(Value::as_u64) != Some(1) {
            return Err(Error::Other(format!(
                "ACP agent negotiated unsupported protocol version: {}",
                initialized
                    .get("protocolVersion")
                    .cloned()
                    .unwrap_or(Value::Null)
            )));
        }
        Ok((client, receiver, endpoint, initialized))
    }

    async fn session_request(
        &self,
        client: &JsonLineClient,
        initialized: &Value,
        method: &str,
        params: Value,
    ) -> Result<Value> {
        match client.request(method, params.clone()).await {
            Ok(response) => Ok(response),
            Err(error) if acp_auth_required(&error.to_string()) => {
                let cached = initialized
                    .get("authMethods")
                    .and_then(Value::as_array)
                    .and_then(|methods| {
                        methods.iter().find_map(|candidate| {
                            (candidate.get("id").and_then(Value::as_str) == Some("cached_token"))
                                .then_some("cached_token")
                        })
                    });
                let Some(method_id) = cached else {
                    return Err(Error::Other(
                        "ACP agent requires authentication but did not advertise the non-interactive `cached_token` method"
                            .into(),
                    ));
                };
                client
                    .request(
                        "authenticate",
                        json!({"methodId": method_id, "_meta": {"headless": true}}),
                    )
                    .await?;
                client.request(method, params).await
            }
            Err(error) => Err(error),
        }
    }

    async fn connection(
        &self,
        cwd: &Path,
        runtime_id: Option<String>,
        launch: Option<RuntimeLaunch>,
    ) -> Result<Box<dyn RuntimeConnection>> {
        let (client, mut receiver, endpoint, initialized) = self.connect(cwd, launch).await?;
        let session_id = if let Some(session_id) = runtime_id {
            let resume = initialized
                .pointer("/agentCapabilities/sessionCapabilities/resume")
                .is_some();
            let load = initialized
                .pointer("/agentCapabilities/loadSession")
                .and_then(Value::as_bool)
                .unwrap_or(false);
            let method = if resume {
                "session/resume"
            } else if load {
                "session/load"
            } else {
                return Err(Error::Other(
                    "ACP agent did not advertise session resume or load".into(),
                ));
            };
            self.session_request(
                client.as_ref(),
                &initialized,
                method,
                json!({"sessionId": session_id, "cwd": cwd, "mcpServers": []}),
            )
            .await?;
            session_id
        } else {
            self.session_request(
                client.as_ref(),
                &initialized,
                "session/new",
                json!({"cwd": cwd, "mcpServers": []}),
            )
            .await?
            .get("sessionId")
            .and_then(Value::as_str)
            .ok_or_else(|| Error::Other("ACP session/new omitted sessionId".into()))?
            .to_string()
        };
        // `session/load` is allowed to replay the persisted conversation as
        // `session/update` notifications before returning its response. Those
        // are bootstrap data, not output from a newly submitted prompt. If
        // they escape through the live runtime stream, clients fabricate an
        // assistant delta and a turn that can never complete because no
        // `session/prompt` request exists. The persisted transcript already
        // supplies this history, so discard every notification queued by the
        // completed new/load handshake before exposing the connection.
        while receiver.try_recv().is_ok() {}
        Ok(Box::new(AcpRuntimeConnection {
            handle: RuntimeHandle {
                harness: self.harness.clone(),
                runtime_id: session_id,
                endpoint,
            },
            client,
            receiver,
            active_prompt: None,
        }))
    }
}

fn acp_auth_required(message: &str) -> bool {
    let message = message.to_ascii_lowercase();
    [
        "auth",
        "login",
        "sign in",
        "sign-in",
        "unauthorized",
        "forbidden",
        "credential",
    ]
    .iter()
    .any(|needle| message.contains(needle))
}

#[async_trait]
impl RuntimeBackend for AcpRuntimeBackend {
    fn harness(&self) -> HarnessId {
        self.harness.clone()
    }

    fn capabilities(&self) -> RuntimeCapabilities {
        RuntimeCapabilities {
            start_session: true,
            // Optional in ACP v1. Known agents may declare it here; attach
            // still checks the actual initialize response before use.
            resume_session: self.resume_session,
            attach_existing_process: false,
            send_input: true,
            stream_events: true,
            interrupt: true,
            respond_to_requests: true,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        self.connection(&request.cwd, None, request.launch).await
    }

    async fn attach(&self, request: RuntimeAttachRequest) -> Result<Box<dyn RuntimeConnection>> {
        let cwd = request.cwd.unwrap_or(std::env::current_dir()?);
        self.connection(&cwd, Some(request.runtime_id), request.launch)
            .await
    }
}

struct AcpRuntimeConnection {
    handle: RuntimeHandle,
    client: Arc<JsonLineClient>,
    receiver: mpsc::UnboundedReceiver<Value>,
    active_prompt: Option<u64>,
}

#[async_trait]
impl RuntimeConnection for AcpRuntimeConnection {
    fn handle(&self) -> &RuntimeHandle {
        &self.handle
    }

    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>> {
        let (id, response) = self
            .client
            .begin_request(
                "session/prompt",
                json!({
                    "sessionId": self.handle.runtime_id,
                    "prompt": [{"type": "text", "text": input.text}],
                }),
            )
            .await?;
        self.active_prompt = Some(id);
        let client = self.client.clone();
        tokio::spawn(async move {
            let result = match response.await {
                Ok(Ok(result)) => json!({"id": id, "result": result}),
                Ok(Err(error)) => json!({"id": id, "error": error}),
                Err(_) => json!({"id": id, "error": "response channel closed"}),
            };
            client.emit(json!({
                "jsonrpc": "2.0",
                "method": "supercode/acp_request_completed",
                "params": result,
            }));
        });
        Ok(Some(id.to_string()))
    }

    async fn next_event(&mut self) -> Result<Option<HarnessEvent>> {
        let Some(payload) = self.receiver.recv().await else {
            return Ok(None);
        };
        let kind = payload
            .get("method")
            .and_then(Value::as_str)
            .or_else(|| payload.get("type").and_then(Value::as_str))
            .unwrap_or("protocol")
            .to_string();
        if kind == "supercode/acp_request_completed" {
            self.active_prompt = None;
        }
        Ok(Some(HarnessEvent {
            sequence: None,
            kind,
            payload,
        }))
    }

    async fn interrupt(&mut self) -> Result<()> {
        self.client
            .notify(
                "session/cancel",
                json!({"sessionId": self.handle.runtime_id}),
            )
            .await
    }

    async fn respond(&mut self, request_id: Value, response: Value) -> Result<()> {
        self.client.respond(request_id, response).await
    }

    async fn close(&mut self) -> Result<()> {
        self.client.close().await
    }
}

/// OpenCode live-runtime backend using its official HTTP API and SSE event
/// stream. [`OpenCodeRuntimeBackend::connect`] can join the server embedded in
/// an already-running TUI when that TUI was launched with a known host/port.
#[derive(Debug, Clone)]
pub struct OpenCodeRuntimeBackend {
    launch: RuntimeLaunch,
    base_url: Option<String>,
}

impl Default for OpenCodeRuntimeBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl OpenCodeRuntimeBackend {
    /// Launch a fresh `opencode serve` process for each connection.
    pub fn new() -> Self {
        Self {
            launch: RuntimeLaunch {
                program: "opencode".into(),
                arguments: vec!["serve".into()],
                env: BTreeMap::new(),
            },
            base_url: None,
        }
    }

    /// Connect to an existing OpenCode server, including a TUI's server when
    /// it was launched on a known address.
    pub fn connect(base_url: impl Into<String>) -> Self {
        Self {
            base_url: Some(base_url.into().trim_end_matches('/').to_string()),
            ..Self::new()
        }
    }

    /// Override the command used when launching a new OpenCode server.
    pub fn with_launch(mut self, launch: RuntimeLaunch) -> Self {
        self.launch = launch;
        self
    }

    async fn service(&self, launch: Option<RuntimeLaunch>) -> Result<(String, Option<Child>)> {
        if let Some(base_url) = &self.base_url {
            wait_for_health(base_url).await?;
            return Ok((base_url.clone(), None));
        }
        let port = TcpListener::bind(("127.0.0.1", 0))?.local_addr()?.port();
        let mut launch = launch.unwrap_or_else(|| self.launch.clone());
        launch.arguments.extend([
            "--hostname".into(),
            "127.0.0.1".into(),
            "--port".into(),
            port.to_string(),
        ]);
        let mut command = Command::new(&launch.program);
        command
            .args(&launch.arguments)
            .envs(&launch.env)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::inherit())
            .kill_on_drop(true);
        // OpenCode's launcher may replace itself with or spawn a native
        // worker. Give the runtime its own group so close can reap the whole
        // server tree instead of orphaning the worker and its inherited FDs.
        #[cfg(unix)]
        command.process_group(0);
        let mut child = command.spawn().map_err(|error| {
            Error::Other(format!("could not launch {}: {error}", launch.program))
        })?;
        let base_url = format!("http://127.0.0.1:{port}");
        if let Err(error) = wait_for_health(&base_url).await {
            // `kill_on_drop` only targets the launcher. Explicitly close its
            // isolated group so a slow or failed startup cannot orphan the
            // native OpenCode worker.
            let _ = terminate_opencode_server(&mut child).await;
            return Err(error);
        }
        Ok((base_url, Some(child)))
    }

    async fn open(
        &self,
        cwd: &Path,
        runtime_id: Option<String>,
        launch: Option<RuntimeLaunch>,
    ) -> Result<Box<dyn RuntimeConnection>> {
        let (base_url, child) = self.service(launch).await?;
        let client = reqwest::Client::new();
        let cwd_string = cwd.to_string_lossy().to_string();
        let runtime_id = match runtime_id {
            Some(id) => {
                http_ok(
                    client
                        .get(format!("{base_url}/session/{id}"))
                        .query(&[("directory", &cwd_string)])
                        .send()
                        .await,
                )
                .await?;
                id
            }
            None => {
                let response = http_ok(
                    client
                        .post(format!("{base_url}/session"))
                        .query(&[("directory", &cwd_string)])
                        .json(&json!({}))
                        .send()
                        .await,
                )
                .await?;
                response
                    .json::<Value>()
                    .await
                    .map_err(http_error)?
                    .get("id")
                    .and_then(Value::as_str)
                    .ok_or_else(|| Error::Other("OpenCode create session omitted id".into()))?
                    .to_string()
            }
        };
        let receiver = spawn_sse(
            client.clone(),
            format!("{base_url}/event"),
            cwd_string.clone(),
        );
        Ok(Box::new(OpenCodeRuntimeConnection {
            handle: RuntimeHandle {
                harness: HarnessId::from(HarnessId::OPENCODE),
                runtime_id,
                endpoint: RuntimeEndpoint::Http {
                    base_url: base_url.clone(),
                    protocol: "opencode-http-sse".into(),
                },
            },
            base_url,
            cwd: cwd_string,
            client,
            receiver,
            child,
        }))
    }
}

#[async_trait]
impl RuntimeBackend for OpenCodeRuntimeBackend {
    fn harness(&self) -> HarnessId {
        HarnessId::from(HarnessId::OPENCODE)
    }

    fn capabilities(&self) -> RuntimeCapabilities {
        RuntimeCapabilities {
            start_session: true,
            resume_session: true,
            attach_existing_process: self.base_url.is_some(),
            send_input: true,
            stream_events: true,
            interrupt: true,
            respond_to_requests: true,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        self.open(&request.cwd, None, request.launch).await
    }

    async fn attach(&self, request: RuntimeAttachRequest) -> Result<Box<dyn RuntimeConnection>> {
        let cwd = request.cwd.unwrap_or(std::env::current_dir()?);
        self.open(&cwd, Some(request.runtime_id), request.launch)
            .await
    }

    async fn attach_existing(
        &self,
        request: RuntimeAttachRequest,
    ) -> Result<Box<dyn RuntimeConnection>> {
        if self.base_url.is_none() {
            return Err(Error::Other(
                "OpenCode live attach requires the existing server's `base_url`".into(),
            ));
        }
        let cwd = request.cwd.unwrap_or(std::env::current_dir()?);
        self.open(&cwd, Some(request.runtime_id), request.launch)
            .await
    }
}

struct OpenCodeRuntimeConnection {
    handle: RuntimeHandle,
    base_url: String,
    cwd: String,
    client: reqwest::Client,
    receiver: mpsc::UnboundedReceiver<Value>,
    child: Option<Child>,
}

#[async_trait]
impl RuntimeConnection for OpenCodeRuntimeConnection {
    fn handle(&self) -> &RuntimeHandle {
        &self.handle
    }

    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>> {
        http_ok(
            self.client
                .post(format!(
                    "{}/session/{}/prompt_async",
                    self.base_url, self.handle.runtime_id
                ))
                .query(&[("directory", &self.cwd)])
                .json(&json!({"parts": [{"type": "text", "text": input.text}]}))
                .send()
                .await,
        )
        .await?;
        Ok(None)
    }

    async fn next_event(&mut self) -> Result<Option<HarnessEvent>> {
        loop {
            let Some(payload) = self.receiver.recv().await else {
                return Ok(None);
            };
            if opencode_event_session_id(&payload)
                .is_some_and(|session_id| session_id != self.handle.runtime_id)
            {
                continue;
            }
            let kind = payload
                .get("type")
                .and_then(Value::as_str)
                .unwrap_or("event")
                .to_string();
            return Ok(Some(HarnessEvent {
                sequence: None,
                kind,
                payload,
            }));
        }
    }

    async fn interrupt(&mut self) -> Result<()> {
        http_ok(
            self.client
                .post(format!(
                    "{}/session/{}/abort",
                    self.base_url, self.handle.runtime_id
                ))
                .query(&[("directory", &self.cwd)])
                .send()
                .await,
        )
        .await?;
        Ok(())
    }

    async fn respond(&mut self, request_id: Value, response: Value) -> Result<()> {
        let permission = request_id.as_str().ok_or_else(|| {
            Error::Other("OpenCode permission request id must be a string".into())
        })?;
        http_ok(
            self.client
                .post(format!(
                    "{}/session/{}/permissions/{permission}",
                    self.base_url, self.handle.runtime_id
                ))
                .query(&[("directory", &self.cwd)])
                .json(&response)
                .send()
                .await,
        )
        .await?;
        Ok(())
    }

    async fn close(&mut self) -> Result<()> {
        if let Some(child) = &mut self.child {
            terminate_opencode_server(child).await?;
        }
        Ok(())
    }
}

fn opencode_event_session_id(payload: &Value) -> Option<&str> {
    let properties = payload.get("properties").unwrap_or(payload);
    properties
        .get("sessionID")
        .and_then(Value::as_str)
        .or_else(|| {
            properties
                .get("part")
                .and_then(|part| part.get("sessionID"))
                .and_then(Value::as_str)
        })
        .or_else(|| {
            properties
                .get("info")
                .and_then(|info| info.get("sessionID"))
                .and_then(Value::as_str)
        })
}

async fn terminate_opencode_server(child: &mut Child) -> Result<()> {
    #[cfg(unix)]
    let process_group = child.id();
    let leader_exited = child.try_wait()?.is_some();
    if leader_exited {
        #[cfg(unix)]
        if let Some(pid) = process_group.filter(|pid| process_group_exists(*pid)) {
            crate::lsp::kill_process_group(pid);
            wait_for_process_group_exit(pid, Duration::from_secs(3)).await?;
        }
        return Ok(());
    }
    // `Child::kill().await` waits for process reaping and can block forever
    // when a launcher leaves its native worker and inherited handles alive.
    // Terminate the isolated group while its leader can still reap workers;
    // killing leader and workers simultaneously can leave transient orphan
    // zombies and made close observably race process cleanup on Linux.
    #[cfg(unix)]
    if let Some(pid) = process_group {
        unsafe {
            libc::kill(-(pid as libc::pid_t), libc::SIGTERM);
        }
        let mut leader_reaped = false;
        if let Ok(status) = tokio::time::timeout(Duration::from_millis(500), child.wait()).await {
            status?;
            leader_reaped = true;
            if !process_group_exists(pid) {
                return Ok(());
            }
        }
        // The leader may exit while a detached worker ignores SIGTERM. Do
        // not mistake a reaped launcher for a stopped server tree.
        crate::lsp::kill_process_group(pid);
        if leader_reaped {
            return wait_for_process_group_exit(pid, Duration::from_secs(3)).await;
        }
    }
    #[cfg(not(unix))]
    child.start_kill()?;
    tokio::time::timeout(Duration::from_secs(3), child.wait())
        .await
        .map_err(|_| Error::Other("timed out reaping the OpenCode server".into()))??;
    #[cfg(unix)]
    if let Some(pid) = process_group {
        wait_for_process_group_exit(pid, Duration::from_secs(3)).await?;
    }
    Ok(())
}

#[cfg(unix)]
fn process_group_exists(pid: u32) -> bool {
    let result = unsafe { libc::kill(-(pid as libc::pid_t), 0) };
    result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
}

#[cfg(unix)]
async fn wait_for_process_group_exit(pid: u32, timeout: Duration) -> Result<()> {
    let deadline = tokio::time::Instant::now() + timeout;
    while process_group_exists(pid) {
        if tokio::time::Instant::now() >= deadline {
            return Err(Error::Other(format!(
                "timed out stopping OpenCode process group {pid}"
            )));
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    Ok(())
}

struct RawLineTransport {
    stdin: Mutex<ChildStdin>,
    child: Mutex<Child>,
    receiver: mpsc::UnboundedReceiver<Value>,
    endpoint: RuntimeEndpoint,
}

impl RawLineTransport {
    async fn spawn(launch: &RuntimeLaunch, cwd: Option<&Path>, protocol: &str) -> Result<Self> {
        let mut command = Command::new(&launch.program);
        command
            .args(&launch.arguments)
            .envs(&launch.env)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::inherit())
            .kill_on_drop(true);
        if let Some(cwd) = cwd {
            command.current_dir(cwd);
        }
        let mut child = command.spawn().map_err(|error| {
            Error::Other(format!("could not launch {}: {error}", launch.program))
        })?;
        let pid = child.id();
        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| Error::Other("runtime child has no stdin".into()))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| Error::Other("runtime child has no stdout".into()))?;
        let (sender, receiver) = mpsc::unbounded_channel();
        tokio::spawn(async move {
            let mut lines = BufReader::new(stdout).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                let value = serde_json::from_str(&line)
                    .unwrap_or_else(|_| json!({"type": "malformed_output", "line": line}));
                let _ = sender.send(value);
            }
        });
        Ok(Self {
            stdin: Mutex::new(stdin),
            child: Mutex::new(child),
            receiver,
            endpoint: RuntimeEndpoint::LocalProcess {
                pid,
                command: std::iter::once(launch.program.clone())
                    .chain(launch.arguments.iter().cloned())
                    .collect(),
                protocol: protocol.into(),
            },
        })
    }

    async fn write(&self, value: Value) -> Result<()> {
        let mut stdin = self.stdin.lock().await;
        stdin.write_all(value.to_string().as_bytes()).await?;
        stdin.write_all(b"\n").await?;
        stdin.flush().await?;
        Ok(())
    }

    async fn close(&self) -> Result<()> {
        let mut child = self.child.lock().await;
        if child.try_wait()?.is_none() {
            child.kill().await?;
        }
        Ok(())
    }
}

async fn raw_next_event(
    receiver: &mut mpsc::UnboundedReceiver<Value>,
) -> Result<Option<HarnessEvent>> {
    let Some(payload) = receiver.recv().await else {
        return Ok(None);
    };
    Ok(Some(harness_event(payload)))
}

fn harness_event(payload: Value) -> HarnessEvent {
    let kind = payload
        .get("type")
        .and_then(Value::as_str)
        .unwrap_or("event")
        .to_string();
    HarnessEvent {
        sequence: None,
        kind,
        payload,
    }
}

fn claude_interrupt_timeout(bound: Duration) -> Error {
    Error::Other(format!(
        "Claude Code did not acknowledge the interrupt control request within {}s",
        bound.as_secs_f32()
    ))
}

pub(crate) fn generated_session_id() -> String {
    let mut bytes = [0_u8; 16];
    if getrandom::getrandom(&mut bytes).is_err() {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
            .to_le_bytes();
        bytes.copy_from_slice(&nanos);
    }
    bytes[6] = (bytes[6] & 0x0f) | 0x40;
    bytes[8] = (bytes[8] & 0x3f) | 0x80;
    format!(
        "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
        bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
        bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15]
    )
}

fn unsupported(protocol: &str, operation: &str) -> Error {
    Error::Other(format!("{protocol} does not support {operation}"))
}

async fn wait_for_health(base_url: &str) -> Result<()> {
    wait_for_health_for(base_url, Duration::from_secs(10)).await
}

async fn wait_for_health_for(base_url: &str, total_timeout: Duration) -> Result<()> {
    let client = reqwest::Client::new();
    let url = format!("{base_url}/global/health");
    let mut last = None;
    let deadline = tokio::time::Instant::now() + total_timeout;
    // Local package-manager shims can take longer than five seconds to start
    // under build or indexing load. Ten seconds avoids false unavailability
    // without permitting an unbounded launch; inventory handshakes retain
    // their separate 30-second bound around the complete startup.
    loop {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            break;
        }
        let request_timeout = remaining.min(Duration::from_millis(500));
        match tokio::time::timeout(request_timeout, client.get(&url).send()).await {
            Ok(Ok(response)) if response.status().is_success() => return Ok(()),
            Ok(Ok(response)) => last = Some(format!("HTTP {}", response.status())),
            Ok(Err(error)) => last = Some(error.to_string()),
            Err(_) => last = Some("health request timed out".into()),
        }
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if !remaining.is_zero() {
            tokio::time::sleep(remaining.min(Duration::from_millis(100))).await;
        }
    }
    Err(Error::Other(format!(
        "OpenCode server at {base_url} did not become healthy: {}",
        last.unwrap_or_else(|| "no response".into())
    )))
}

async fn http_ok(
    response: std::result::Result<reqwest::Response, reqwest::Error>,
) -> Result<reqwest::Response> {
    response
        .map_err(http_error)?
        .error_for_status()
        .map_err(http_error)
}

fn http_error(error: reqwest::Error) -> Error {
    Error::Other(format!("runtime HTTP request failed: {error}"))
}

fn spawn_sse(
    client: reqwest::Client,
    url: String,
    directory: String,
) -> mpsc::UnboundedReceiver<Value> {
    let (sender, receiver) = mpsc::unbounded_channel();
    tokio::spawn(async move {
        let response = client
            .get(url)
            .query(&[("directory", directory)])
            .send()
            .await;
        let Ok(response) = response.and_then(reqwest::Response::error_for_status) else {
            let _ = sender.send(
                json!({"type": "stream_error", "message": "could not open OpenCode SSE stream"}),
            );
            return;
        };
        let mut stream = response.bytes_stream();
        let mut buffer = String::new();
        while let Some(chunk) = stream.next().await {
            let Ok(chunk) = chunk else {
                break;
            };
            buffer.push_str(&String::from_utf8_lossy(&chunk));
            while let Some(newline) = buffer.find('\n') {
                let line = buffer[..newline].trim_end_matches('\r').to_string();
                buffer.drain(..=newline);
                if let Some(data) = line.strip_prefix("data:") {
                    let data = data.trim();
                    if let Ok(value) = serde_json::from_str(data) {
                        let _ = sender.send(value);
                    }
                }
            }
        }
    });
    receiver
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Fake `claude --print --input-format stream-json` child. It appends every
    /// stdin frame to `$1` so a test can assert the exact bytes this adapter
    /// wrote, and replies with the control envelope the real CLI replies with.
    #[cfg(unix)]
    const FAKE_CLAUDE_ACKS: &str = r#"
cap="$1"
while IFS= read -r line; do
  printf '%s\n' "$line" >> "$cap"
  case "$line" in
    *'"subtype":"interrupt"'*)
      rid=$(printf '%s' "$line" | sed -n 's/.*"request_id":"\([^"]*\)".*/\1/p')
      printf '{"type":"system","subtype":"mid_flight"}\n'
      printf '{"type":"control_response","response":{"subtype":"success","request_id":"someone-elses-request","response":{}}}\n'
      printf '{"type":"control_response","response":{"subtype":"success","request_id":"%s","response":{"still_queued":[]}}}\n' "$rid"
      ;;
    *'"type":"user"'*)
      printf '{"type":"assistant","message":{"role":"assistant","content":"replied"}}\n'
      ;;
  esac
done
"#;

    /// Same, but the control channel never answers — the hang this adapter must
    /// convert into a bounded, structured error.
    #[cfg(unix)]
    const FAKE_CLAUDE_NEVER_ACKS: &str = r#"
cap="$1"
while IFS= read -r line; do
  printf '%s\n' "$line" >> "$cap"
done
"#;

    /// Rejects the interrupt the way the CLI reports a control failure.
    #[cfg(unix)]
    const FAKE_CLAUDE_REJECTS: &str = r#"
cap="$1"
while IFS= read -r line; do
  printf '%s\n' "$line" >> "$cap"
  case "$line" in
    *'"subtype":"interrupt"'*)
      rid=$(printf '%s' "$line" | sed -n 's/.*"request_id":"\([^"]*\)".*/\1/p')
      printf '{"type":"control_response","response":{"subtype":"error","request_id":"%s","error":"no active worker"}}\n' "$rid"
      ;;
  esac
done
"#;

    #[cfg(unix)]
    struct FakeClaude {
        connection: ClaudeRuntimeConnection,
        capture: std::path::PathBuf,
        _dir: std::path::PathBuf,
    }

    #[cfg(unix)]
    impl FakeClaude {
        async fn spawn(script: &str, control_timeout: Duration) -> Self {
            let dir = std::env::temp_dir().join(format!(
                "supercode-fake-claude-{}-{}",
                std::process::id(),
                generated_session_id()
            ));
            std::fs::create_dir_all(&dir).unwrap();
            let capture = dir.join("stdin.jsonl");
            let launch = RuntimeLaunch {
                program: "/bin/sh".into(),
                arguments: vec![
                    "-c".into(),
                    script.into(),
                    "fake-claude".into(),
                    capture.display().to_string(),
                ],
                env: BTreeMap::new(),
            };
            let transport = RawLineTransport::spawn(&launch, Some(&dir), "claude-stream-json")
                .await
                .unwrap();
            let connection = ClaudeRuntimeConnection {
                handle: RuntimeHandle {
                    harness: HarnessId::from(HarnessId::CLAUDE_CODE),
                    runtime_id: "fake-session".into(),
                    endpoint: transport.endpoint.clone(),
                },
                transport,
                buffered_events: VecDeque::new(),
                next_control_request: 1,
                control_timeout,
            };
            Self {
                connection,
                capture,
                _dir: dir,
            }
        }

        fn written_frames(&self) -> Vec<Value> {
            std::fs::read_to_string(&self.capture)
                .unwrap_or_default()
                .lines()
                .filter(|line| !line.trim().is_empty())
                .map(|line| serde_json::from_str(line).expect("adapter wrote a non-JSON frame"))
                .collect()
        }
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn claude_interrupt_writes_one_control_request_per_call_with_a_fresh_id() {
        let mut fake = FakeClaude::spawn(FAKE_CLAUDE_ACKS, Duration::from_secs(5)).await;

        fake.connection.interrupt().await.unwrap();
        fake.connection.interrupt().await.unwrap();

        let frames = fake.written_frames();
        assert_eq!(
            frames.len(),
            2,
            "each interrupt must write exactly one control frame: {frames:?}"
        );
        let mut ids = Vec::new();
        for frame in &frames {
            assert_eq!(frame["type"], "control_request");
            assert_eq!(frame["request"]["subtype"], "interrupt");
            let id = frame["request_id"].as_str().expect("frame carries an id");
            assert!(!id.is_empty());
            ids.push(id.to_string());
        }
        assert_ne!(ids[0], ids[1], "request ids must be unique per call");
    }

    /// The harness acknowledges an interrupt sent with no turn in flight —
    /// measured against claude 2.1.224, which replies `success` with an empty
    /// `still_queued` list and keeps taking input. The adapter reports that
    /// rather than inventing a turn-state gate, and the session stays usable.
    #[cfg(unix)]
    #[tokio::test]
    async fn claude_interrupt_with_no_turn_in_flight_is_acknowledged_and_the_session_survives() {
        let mut fake = FakeClaude::spawn(FAKE_CLAUDE_ACKS, Duration::from_secs(5)).await;

        fake.connection.interrupt().await.unwrap();
        fake.connection
            .send_input(RuntimeInput {
                text: "redirect".into(),
            })
            .await
            .unwrap();

        // Events observed while the interrupt was pending are replayed first,
        // and the control channel's own frames never reach the consumer. Read
        // through the assistant event before inspecting the fake child's
        // capture so the child has necessarily consumed the user frame.
        let mut kinds = Vec::new();
        while kinds.len() < 2 {
            let event = fake.connection.next_event().await.unwrap().unwrap();
            assert_ne!(event.kind, "control_response");
            kinds.push(event.kind);
        }
        assert_eq!(kinds, vec!["system".to_string(), "assistant".to_string()]);

        let frames = fake.written_frames();
        assert_eq!(frames[0]["type"], "control_request");
        assert_eq!(
            frames[1]["type"], "user",
            "a send issued after an interrupt must reach the harness, in order"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn claude_interrupt_times_out_with_a_structured_error_instead_of_hanging() {
        let mut fake = FakeClaude::spawn(FAKE_CLAUDE_NEVER_ACKS, Duration::from_millis(250)).await;

        let started = std::time::Instant::now();
        let error = fake.connection.interrupt().await.unwrap_err();

        assert!(
            started.elapsed() < Duration::from_secs(5),
            "interrupt must return on its own bound, not hang"
        );
        assert!(
            error
                .to_string()
                .contains("did not acknowledge the interrupt"),
            "unexpected error: {error}"
        );
        assert_eq!(fake.written_frames().len(), 1);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn claude_interrupt_surfaces_a_rejecting_control_response_as_an_error() {
        let mut fake = FakeClaude::spawn(FAKE_CLAUDE_REJECTS, Duration::from_secs(5)).await;

        let error = fake.connection.interrupt().await.unwrap_err();

        assert!(
            error.to_string().contains("no active worker"),
            "unexpected error: {error}"
        );
    }

    #[test]
    fn claude_code_runtime_advertises_interrupt() {
        assert!(ClaudeCodeRuntimeBackend::new().capabilities().interrupt);
    }

    #[test]
    fn capability_reports_distinguish_resume_from_process_attach() {
        assert!(
            !PiRuntimeBackend::new()
                .capabilities()
                .attach_existing_process
        );
        assert!(
            !ClaudeCodeRuntimeBackend::new()
                .capabilities()
                .attach_existing_process
        );
        assert!(
            !OpenCodeRuntimeBackend::new()
                .capabilities()
                .attach_existing_process
        );
        assert!(
            OpenCodeRuntimeBackend::connect("http://127.0.0.1:4096")
                .capabilities()
                .attach_existing_process
        );
    }

    #[test]
    fn generated_ids_are_uuid_shaped_and_unique() {
        let first = generated_session_id();
        let second = generated_session_id();
        assert_eq!(first.len(), 36);
        assert_ne!(first, second);
    }

    #[test]
    fn opencode_event_session_id_covers_current_event_shapes() {
        assert_eq!(
            opencode_event_session_id(&json!({
                "type": "session.status",
                "properties": {"sessionID": "session-direct", "status": {"type": "busy"}}
            })),
            Some("session-direct")
        );
        assert_eq!(
            opencode_event_session_id(&json!({
                "type": "message.part.updated",
                "properties": {"part": {"sessionID": "session-part", "type": "text"}}
            })),
            Some("session-part")
        );
        assert_eq!(
            opencode_event_session_id(&json!({
                "type": "message.updated",
                "properties": {"info": {"sessionID": "session-info", "role": "assistant"}}
            })),
            Some("session-info")
        );
        assert_eq!(
            opencode_event_session_id(&json!({"type": "server.connected"})),
            None
        );
    }

    #[tokio::test]
    async fn opencode_runtime_skips_events_for_other_sessions() {
        let (sender, receiver) = mpsc::unbounded_channel();
        sender
            .send(json!({
                "type": "session.idle",
                "properties": {"sessionID": "foreign-session"}
            }))
            .unwrap();
        sender
            .send(json!({
                "type": "message.part.delta",
                "properties": {"sessionID": "local-session", "delta": "hello"}
            }))
            .unwrap();
        let mut connection = OpenCodeRuntimeConnection {
            handle: RuntimeHandle {
                harness: HarnessId::from(HarnessId::OPENCODE),
                runtime_id: "local-session".into(),
                endpoint: RuntimeEndpoint::Http {
                    base_url: "http://127.0.0.1:1".into(),
                    protocol: "opencode-http".into(),
                },
            },
            base_url: "http://127.0.0.1:1".into(),
            cwd: "/tmp".into(),
            client: reqwest::Client::new(),
            receiver,
            child: None,
        };

        let event = connection.next_event().await.unwrap().unwrap();

        assert_eq!(event.kind, "message.part.delta");
        assert_eq!(event.payload["properties"]["sessionID"], "local-session");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn opencode_shutdown_reaps_a_launcher_process_group() {
        let mut command = Command::new("/bin/sh");
        command
            .args(["-c", "sleep 30 & wait"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let mut child = command.spawn().unwrap();
        let pid = child.id().unwrap();

        terminate_opencode_server(&mut child).await.unwrap();

        assert!(child.try_wait().unwrap().is_some());
        let group_still_exists = unsafe { libc::kill(-(pid as libc::pid_t), 0) } == 0;
        assert!(
            !group_still_exists,
            "OpenCode worker process group survived close"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn opencode_shutdown_reaps_workers_after_launcher_exit() {
        let mut command = Command::new("/bin/sh");
        command
            .args(["-c", "sleep 30 & exit 0"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let mut child = command.spawn().unwrap();
        let pid = child.id().unwrap();
        tokio::time::sleep(Duration::from_millis(200)).await;

        terminate_opencode_server(&mut child).await.unwrap();

        assert!(child.try_wait().unwrap().is_some());
        assert!(
            !process_group_exists(pid),
            "OpenCode worker process group survived its exited launcher"
        );
    }

    #[tokio::test]
    async fn opencode_health_probe_is_bounded_when_a_socket_never_responds() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (_socket, _) = listener.accept().await.unwrap();
            tokio::time::sleep(Duration::from_secs(30)).await;
        });
        let started = tokio::time::Instant::now();

        let error = wait_for_health_for(&format!("http://{address}"), Duration::from_millis(200))
            .await
            .unwrap_err();

        assert!(error.to_string().contains("health request timed out"));
        assert!(started.elapsed() < Duration::from_secs(1));
        server.abort();
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn acp_adapter_negotiates_starts_and_streams_without_blocking_prompt() {
        let script = r#"
            i=0
            while IFS= read -r line; do
              i=$((i + 1))
              case "$i" in
                1) printf '%s\n' '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1,"agentCapabilities":{},"authMethods":[]}}' ;;
                2) printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"sessionId":"acp_mock"}}' ;;
                3)
                  printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"acp_mock","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"hello"}}}}'
                  printf '%s\n' '{"jsonrpc":"2.0","id":3,"result":{"stopReason":"end_turn"}}'
                  ;;
              esac
            done
        "#;
        let backend = AcpRuntimeBackend::new(
            HarnessId::from("mock-acp"),
            RuntimeLaunch {
                program: "/bin/sh".into(),
                arguments: vec!["-c".into(), script.into()],
                env: BTreeMap::new(),
            },
        );
        let mut connection = backend
            .start(RuntimeStartRequest {
                cwd: std::env::current_dir().unwrap(),
                launch: None,
            })
            .await
            .unwrap();
        assert_eq!(connection.handle().runtime_id, "acp_mock");
        assert_eq!(
            connection
                .send_input(RuntimeInput { text: "hi".into() })
                .await
                .unwrap()
                .as_deref(),
            Some("3")
        );
        assert_eq!(
            connection.next_event().await.unwrap().unwrap().kind,
            "session/update"
        );
        assert_eq!(
            connection.next_event().await.unwrap().unwrap().kind,
            "supercode/acp_request_completed"
        );
        connection.close().await.unwrap();
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn acp_uses_an_existing_login_before_trying_an_advertised_auth_method() {
        let script = r#"
            i=0
            while IFS= read -r line; do
              i=$((i + 1))
              if [ "$i" -eq 1 ]; then
                printf '%s\n' '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1,"agentCapabilities":{},"authMethods":[{"id":"cached_token"}]}}'
              elif printf '%s' "$line" | grep -q 'session/new'; then
                printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"sessionId":"existing_login"}}'
              else
                exit 9
              fi
            done
        "#;
        let backend = AcpRuntimeBackend::new(
            HarnessId::from("mock-acp"),
            RuntimeLaunch {
                program: "/bin/sh".into(),
                arguments: vec!["-c".into(), script.into()],
                env: BTreeMap::new(),
            },
        );
        let mut connection = backend
            .start(RuntimeStartRequest {
                cwd: std::env::current_dir().unwrap(),
                launch: None,
            })
            .await
            .unwrap();
        assert_eq!(connection.handle().runtime_id, "existing_login");
        connection.close().await.unwrap();
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn known_acp_agent_reports_and_uses_load_session_for_resume() {
        let script = r#"
            i=0
            while IFS= read -r line; do
              i=$((i + 1))
              case "$i" in
                1) printf '%s\n' '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true},"authMethods":[]}}' ;;
                2)
                  case "$line" in
                    *'"method":"session/load"'*'"sessionId":"existing-session"'*)
                      printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"existing-session","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"historical replay"}}}}'
                      printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{}}'
                      ;;
                    *) exit 42 ;;
                  esac
                  ;;
                3)
                  printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"existing-session","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"fresh output"}}}}'
                  printf '%s\n' '{"jsonrpc":"2.0","id":3,"result":{"stopReason":"end_turn"}}'
                  ;;
              esac
            done
        "#;
        let backend = AcpRuntimeBackend::new(
            HarnessId::from("known-acp"),
            RuntimeLaunch {
                program: "/bin/sh".into(),
                arguments: vec!["-c".into(), script.into()],
                env: BTreeMap::new(),
            },
        )
        .with_resume_support(true);
        assert!(backend.capabilities().resume_session);
        let mut connection = backend
            .attach(RuntimeAttachRequest {
                runtime_id: "existing-session".into(),
                cwd: Some(std::env::current_dir().unwrap()),
                launch: None,
            })
            .await
            .unwrap();
        assert_eq!(connection.handle().runtime_id, "existing-session");
        assert_eq!(
            connection
                .send_input(RuntimeInput {
                    text: "continue".into(),
                })
                .await
                .unwrap()
                .as_deref(),
            Some("3")
        );
        let event = connection.next_event().await.unwrap().unwrap();
        assert_eq!(event.kind, "session/update");
        assert_eq!(
            event
                .payload
                .pointer("/params/update/content/text")
                .and_then(Value::as_str),
            Some("fresh output")
        );
        assert_eq!(
            connection.next_event().await.unwrap().unwrap().kind,
            "supercode/acp_request_completed"
        );
        connection.close().await.unwrap();
    }
}