supercode_harness/

runtime.rs

//! Primitive live-runtime contracts and the Codex app-server reference adapter.
//!
//! These APIs control harness-native sessions; they do not emulate terminal
//! keystrokes and do not claim to attach to an arbitrary already-running TUI.

use std::collections::{BTreeMap, HashMap};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStdin, Command};
use tokio::sync::{mpsc, oneshot, Mutex};

use crate::{Error, HarnessId, Result};

mod adapters;
mod hosted;
#[cfg(feature = "adapter-api")]
mod supercode_http;
pub(crate) use adapters::generated_session_id;
pub use adapters::{
    AcpRuntimeBackend, ClaudeCodeRuntimeBackend, OpenCodeRuntimeBackend, PiRuntimeBackend,
};
pub use hosted::{HostedHarnessConnection, HostedHarnessRuntime};
#[cfg(feature = "adapter-api")]
pub use supercode_http::SupercodeHttpRuntimeBackend;

/// Mechanical facts an adapter can guarantee.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeCapabilities {
    /// Can create a fresh harness-native session.
    pub start_session: bool,
    /// Can resume a harness-native persisted session by id.
    pub resume_session: bool,
    /// Can join an arbitrary already-running harness process.
    pub attach_existing_process: bool,
    /// Can send user input through a structured protocol.
    pub send_input: bool,
    /// Can receive structured live events.
    pub stream_events: bool,
    /// Can interrupt an in-flight turn.
    pub interrupt: bool,
    /// Can redirect an in-flight turn without interrupting it.
    #[serde(default)]
    pub steer: bool,
    /// Can answer protocol requests such as approvals or elicitation.
    pub respond_to_requests: bool,
}

/// Executable configuration used to launch one adapter endpoint.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeLaunch {
    /// Executable name or path.
    pub program: String,
    /// Arguments passed before adapter-generated protocol arguments.
    pub arguments: Vec<String>,
    /// Extra environment variables.
    pub env: BTreeMap<String, String>,
}

/// Request to create a fresh runtime session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeStartRequest {
    /// Project working directory.
    pub cwd: PathBuf,
    /// Optional executable override, primarily for alternate installs/tests.
    pub launch: Option<RuntimeLaunch>,
}

/// Request to resume or attach through a new adapter connection.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeAttachRequest {
    /// Harness-native session/thread id.
    pub runtime_id: String,
    /// Optional cwd override accepted by the harness protocol.
    pub cwd: Option<PathBuf>,
    /// Optional executable override.
    pub launch: Option<RuntimeLaunch>,
}

/// Observable endpoint backing a runtime connection.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RuntimeEndpoint {
    /// Child process owned by this connection.
    LocalProcess {
        /// Process id when available.
        pid: Option<u32>,
        /// Executable plus arguments.
        command: Vec<String>,
        /// Native protocol spoken over stdio.
        protocol: String,
    },
    /// Existing HTTP service.
    Http {
        /// Service base URL.
        base_url: String,
        /// Native protocol name.
        protocol: String,
    },
}

/// Identity returned after a live session is started or resumed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHandle {
    /// Runtime adapter/harness.
    pub harness: HarnessId,
    /// Harness-native live session identity.
    pub runtime_id: String,
    /// Concrete endpoint used by this connection.
    pub endpoint: RuntimeEndpoint,
}

/// User input accepted by a live runtime.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeInput {
    /// Plain text prompt or steering instruction.
    pub text: String,
    /// Runtime-resolved image URLs or `data:image/...;base64,...` payloads.
    ///
    /// Adapters must either preserve these as native multimodal input or
    /// reject the turn explicitly; they must never flatten image bytes into
    /// the text prompt.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub image_urls: Vec<String>,
}

/// Protocol-neutral envelope around a native live event.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct HarnessEvent {
    /// Canonical SDK sequence when the event originated from an SDK runtime.
    /// Native harness adapters leave this absent and the service sequences
    /// their transport stream locally.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sequence: Option<u64>,
    /// Native method/type name, or `request` for a server-initiated request.
    pub kind: String,
    /// Lossless native event/request value.
    pub payload: Value,
}

/// One connected harness-native runtime session.
#[async_trait]
pub trait RuntimeConnection: Send {
    /// Identity and endpoint of this connection.
    fn handle(&self) -> &RuntimeHandle;
    /// Submit structured user input and return the harness-native turn id when
    /// one is allocated.
    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>>;
    /// Wait for the next native live event.
    async fn next_event(&mut self) -> Result<Option<HarnessEvent>>;
    /// Interrupt the current turn, when supported.
    async fn interrupt(&mut self) -> Result<()>;
    /// Redirect the current turn, when supported.
    async fn steer(&mut self, _text: String) -> Result<()> {
        Err(Error::Other(
            "this runtime cannot steer an active turn".into(),
        ))
    }
    /// Answer a server-initiated protocol request by its native JSON id.
    async fn respond(&mut self, request_id: Value, response: Value) -> Result<()>;
    /// Close the adapter-owned transport/process.
    async fn close(&mut self) -> Result<()>;
}

/// Factory for starting, resuming, and (where the native protocol permits it)
/// joining one harness's already-running runtime endpoint.
#[async_trait]
pub trait RuntimeBackend: Send + Sync {
    /// Harness implemented by this backend.
    fn harness(&self) -> HarnessId;
    /// Honest mechanical capability report.
    fn capabilities(&self) -> RuntimeCapabilities;
    /// Create a fresh harness-native session.
    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>>;
    /// Resume a persisted harness-native session through a new protocol
    /// connection. This does not imply joining the process that originally
    /// wrote the session.
    async fn attach(&self, request: RuntimeAttachRequest) -> Result<Box<dyn RuntimeConnection>>;
    /// Join an already-running harness process or server. Most stock harnesses
    /// cannot do this; adapters must opt in rather than silently treating a
    /// persisted resume as a live attach.
    async fn attach_existing(
        &self,
        _request: RuntimeAttachRequest,
    ) -> Result<Box<dyn RuntimeConnection>> {
        Err(Error::Other(format!(
            "{} cannot attach to an already-running process",
            self.harness().as_str()
        )))
    }
}

/// Codex live-runtime backend using the official `codex app-server` JSONL
/// protocol (`initialize`, `thread/start|resume`, `turn/start|interrupt`).
#[derive(Debug, Clone)]
pub struct CodexRuntimeBackend {
    launch: RuntimeLaunch,
}

const CODEX_STARTUP_TIMEOUT: Duration = Duration::from_secs(10);

/// A stock Codex app-server eagerly indexes everything below `CODEX_HOME`
/// before answering `initialize`. That turns a runtime open into an unbounded
/// corpus scan for long-time Codex users. Give each connection a private state
/// database and project only the one native rollout it needs into that home.
/// The rollout itself is hard-linked, so Codex continues the original inode
/// rather than a copy that would need lossy reconciliation later.
#[derive(Debug)]
struct CodexRuntimeHome {
    root: PathBuf,
    native_home: PathBuf,
}

impl CodexRuntimeHome {
    fn prepare(launch: &mut RuntimeLaunch, runtime_id: Option<&str>) -> Result<Self> {
        let native_home = codex_native_home(launch)?;
        let root = supercode_runtime_root()
            .join("codex")
            .join(generated_session_id());
        std::fs::create_dir_all(&root).map_err(|error| {
            Error::Other(format!(
                "could not create isolated Codex runtime home {}: {error}",
                root.display()
            ))
        })?;
        set_private_directory(&root)?;
        let root = std::fs::canonicalize(&root)?;

        for entry in [
            "auth.json",
            "config.toml",
            "hooks.json",
            "models_cache.json",
            "installation_id",
            ".personality_migration",
            ".sandbox_migration",
            "cache",
            "generated_images",
            "mcp-oauth-locks",
            "memories",
            "plugins",
            "rules",
            "shell_snapshots",
            "skills",
            "thread-writer-locks",
        ] {
            link_runtime_resource(&native_home.join(entry), &root.join(entry))?;
        }

        if let Some(runtime_id) = runtime_id {
            let source = find_codex_rollout(&native_home.join("sessions"), runtime_id)?
                .ok_or_else(|| {
                    Error::Other(format!(
                        "could not find Codex rollout `{runtime_id}` below {}",
                        native_home.join("sessions").display()
                    ))
                })?;
            let relative = source.strip_prefix(&native_home).map_err(|_| {
                Error::Other(format!(
                    "Codex rollout {} is outside native home {}",
                    source.display(),
                    native_home.display()
                ))
            })?;
            let projected = root.join(relative);
            if let Some(parent) = projected.parent() {
                std::fs::create_dir_all(parent)?;
            }
            std::fs::hard_link(&source, &projected).map_err(|error| {
                Error::Other(format!(
                    "could not project Codex rollout {} into isolated runtime home: {error}",
                    source.display()
                ))
            })?;
        }

        launch
            .env
            .insert("CODEX_HOME".into(), root.to_string_lossy().into_owned());
        Ok(Self { root, native_home })
    }

    fn started_rollout_path(&self, response: &Value) -> Result<PathBuf> {
        let path = response
            .pointer("/thread/path")
            .and_then(Value::as_str)
            .map(PathBuf::from)
            .ok_or_else(|| {
                Error::Other("Codex thread/start response omitted thread.path".into())
            })?;
        let relative = path.strip_prefix(&self.root).map_err(|_| {
            Error::Other(format!(
                "Codex created rollout {} outside isolated runtime home {}",
                path.display(),
                self.root.display()
            ))
        })?;
        if !relative.starts_with("sessions") {
            return Err(Error::Other(format!(
                "Codex created non-session rollout {}",
                path.display()
            )));
        }
        Ok(path)
    }

    async fn publish_rollout(&self, path: &Path) -> Result<()> {
        let relative = path.strip_prefix(&self.root).map_err(|_| {
            Error::Other(format!(
                "Codex created rollout {} outside isolated runtime home {}",
                path.display(),
                self.root.display()
            ))
        })?;
        let publish_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        while !path.is_file() {
            if tokio::time::Instant::now() >= publish_deadline {
                return Err(Error::Other(format!(
                    "Codex did not create promised rollout {} within 2s",
                    path.display()
                )));
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let native = self.native_home.join(relative);
        if let Some(parent) = native.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::hard_link(path, &native).map_err(|error| {
            Error::Other(format!(
                "could not publish Codex rollout {} to native home: {error}",
                path.display()
            ))
        })
    }

    fn cleanup(&self) -> Result<()> {
        match std::fs::remove_dir_all(&self.root) {
            Ok(()) => Ok(()),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(error) => Err(Error::Other(format!(
                "could not clean isolated Codex runtime home {}: {error}",
                self.root.display()
            ))),
        }
    }
}

impl Drop for CodexRuntimeHome {
    fn drop(&mut self) {
        let _ = self.cleanup();
    }
}

fn is_stock_codex_launch(launch: &RuntimeLaunch) -> bool {
    launch
        .arguments
        .iter()
        .any(|argument| argument == "app-server")
        && Path::new(&launch.program)
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| name == "codex" || name == "codex.exe")
}

fn codex_native_home(launch: &RuntimeLaunch) -> Result<PathBuf> {
    launch
        .env
        .get("CODEX_HOME")
        .map(PathBuf::from)
        .or_else(|| std::env::var_os("CODEX_HOME").map(PathBuf::from))
        .or_else(|| {
            std::env::var_os("HOME")
                .map(PathBuf::from)
                .map(|home| home.join(".codex"))
        })
        .ok_or_else(|| Error::Other("Codex runtime requires CODEX_HOME or HOME".into()))
}

fn supercode_runtime_root() -> PathBuf {
    std::env::var_os("SUPERCODE_HOME")
        .map(PathBuf::from)
        .or_else(|| {
            std::env::var_os("HOME")
                .map(PathBuf::from)
                .map(|home| home.join(".supercode"))
        })
        .unwrap_or_else(|| std::env::temp_dir().join("supercode"))
        .join("runtime-homes")
}

fn find_codex_rollout(root: &Path, runtime_id: &str) -> Result<Option<PathBuf>> {
    let entries = match std::fs::read_dir(root) {
        Ok(entries) => entries,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };
    let expected_suffix = format!("-{runtime_id}.jsonl");
    for entry in entries {
        let entry = entry?;
        let kind = entry.file_type()?;
        if kind.is_dir() {
            if let Some(path) = find_codex_rollout(&entry.path(), runtime_id)? {
                return Ok(Some(path));
            }
        } else if kind.is_file()
            && entry
                .file_name()
                .to_str()
                .is_some_and(|name| name.ends_with(&expected_suffix))
        {
            return Ok(Some(entry.path()));
        }
    }
    Ok(None)
}

#[cfg(unix)]
fn link_runtime_resource(source: &Path, target: &Path) -> Result<()> {
    use std::os::unix::fs::symlink;

    if source.exists() {
        symlink(source, target)?;
    }
    Ok(())
}

#[cfg(not(unix))]
fn link_runtime_resource(source: &Path, target: &Path) -> Result<()> {
    if source.is_file() {
        std::fs::copy(source, target)?;
    }
    Ok(())
}

#[cfg(unix)]
fn set_private_directory(path: &Path) -> Result<()> {
    use std::os::unix::fs::PermissionsExt;

    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))?;
    Ok(())
}

#[cfg(not(unix))]
fn set_private_directory(_path: &Path) -> Result<()> {
    Ok(())
}

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

impl CodexRuntimeBackend {
    /// Use `codex app-server` from `PATH`.
    pub fn new() -> Self {
        Self {
            launch: RuntimeLaunch {
                program: "codex".into(),
                arguments: vec!["app-server".into()],
                env: BTreeMap::new(),
            },
        }
    }

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

    async fn connect(
        &self,
        launch: Option<RuntimeLaunch>,
        runtime_id: Option<&str>,
    ) -> Result<(
        Arc<JsonLineClient>,
        mpsc::UnboundedReceiver<Value>,
        RuntimeEndpoint,
        Option<CodexRuntimeHome>,
    )> {
        let mut launch = launch.unwrap_or_else(|| self.launch.clone());
        let runtime_home = if is_stock_codex_launch(&launch) {
            Some(CodexRuntimeHome::prepare(&mut launch, runtime_id)?)
        } else {
            None
        };
        let (client, receiver, endpoint) =
            JsonLineClient::spawn(&launch, None, false, "codex-app-server-jsonl").await?;
        tokio::time::timeout(
            CODEX_STARTUP_TIMEOUT,
            client.request(
                "initialize",
                json!({
                    "clientInfo": {
                        "name": "supercode",
                        "title": "Supercode",
                        "version": env!("CARGO_PKG_VERSION"),
                    }
                }),
            ),
        )
        .await
        .map_err(|_| Error::Other("Codex app-server initialize timed out after 10s".into()))??;
        client.notify("initialized", json!({})).await?;
        Ok((client, receiver, endpoint, runtime_home))
    }

    async fn open_thread(
        &self,
        method: &str,
        params: Value,
        launch: Option<RuntimeLaunch>,
        runtime_id: Option<&str>,
    ) -> Result<Box<dyn RuntimeConnection>> {
        let (client, receiver, endpoint, runtime_home) = self.connect(launch, runtime_id).await?;
        let response = tokio::time::timeout(CODEX_STARTUP_TIMEOUT, client.request(method, params))
            .await
            .map_err(|_| Error::Other(format!("Codex {method} timed out after 10s")))??;
        let thread_id = response
            .pointer("/thread/id")
            .and_then(Value::as_str)
            .ok_or_else(|| Error::Other(format!("Codex {method} response omitted thread.id")))?
            .to_string();
        let unpublished_rollout = if method == "thread/start" {
            runtime_home
                .as_ref()
                .map(|home| home.started_rollout_path(&response))
                .transpose()?
        } else {
            None
        };
        Ok(Box::new(CodexRuntimeConnection {
            handle: RuntimeHandle {
                harness: HarnessId::from(HarnessId::CODEX),
                runtime_id: thread_id,
                endpoint,
            },
            client,
            receiver,
            active_turn: None,
            runtime_home,
            unpublished_rollout,
        }))
    }
}

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

    fn capabilities(&self) -> RuntimeCapabilities {
        RuntimeCapabilities {
            start_session: true,
            resume_session: true,
            // A new app-server can resume the same stored thread, but stock
            // Codex does not let it join an arbitrary already-running TUI's
            // transport/event fanout.
            attach_existing_process: false,
            send_input: true,
            stream_events: true,
            interrupt: true,
            steer: true,
            respond_to_requests: true,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        self.open_thread(
            "thread/start",
            json!({"cwd": request.cwd}),
            request.launch,
            None,
        )
        .await
    }

    async fn attach(&self, request: RuntimeAttachRequest) -> Result<Box<dyn RuntimeConnection>> {
        let mut params = json!({"threadId": request.runtime_id});
        if let Some(cwd) = request.cwd {
            params["cwd"] = json!(cwd);
        }
        let runtime_id = request.runtime_id.clone();
        self.open_thread("thread/resume", params, request.launch, Some(&runtime_id))
            .await
    }
}

struct CodexRuntimeConnection {
    handle: RuntimeHandle,
    client: Arc<JsonLineClient>,
    receiver: mpsc::UnboundedReceiver<Value>,
    active_turn: Option<String>,
    runtime_home: Option<CodexRuntimeHome>,
    unpublished_rollout: Option<PathBuf>,
}

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

    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>> {
        let mut parts = Vec::new();
        if !input.text.is_empty() {
            parts.push(json!({"type": "text", "text": input.text}));
        }
        parts.extend(
            input
                .image_urls
                .into_iter()
                .map(|url| json!({"type": "image", "url": url})),
        );
        let response = self
            .client
            .request(
                "turn/start",
                json!({
                    "threadId": self.handle.runtime_id,
                    "input": parts,
                }),
            )
            .await?;
        let turn_id = response
            .pointer("/turn/id")
            .and_then(Value::as_str)
            .map(str::to_owned);
        if let (Some(home), Some(path)) = (
            self.runtime_home.as_ref(),
            self.unpublished_rollout.as_ref(),
        ) {
            home.publish_rollout(path).await?;
            self.unpublished_rollout = None;
        }
        self.active_turn = turn_id.clone();
        Ok(turn_id)
    }

    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)
            .map(str::to_owned)
            .unwrap_or_else(|| "protocol".into());
        if kind == "turn/completed" {
            self.active_turn = None;
        }
        Ok(Some(HarnessEvent {
            sequence: None,
            kind,
            payload,
        }))
    }

    async fn interrupt(&mut self) -> Result<()> {
        let Some(turn_id) = self.active_turn.as_ref() else {
            return Err(Error::Other("Codex has no active turn to interrupt".into()));
        };
        self.client
            .request(
                "turn/interrupt",
                json!({"threadId": self.handle.runtime_id, "turnId": turn_id}),
            )
            .await?;
        Ok(())
    }

    async fn steer(&mut self, text: String) -> Result<()> {
        let Some(turn_id) = self.active_turn.as_ref() else {
            return Err(Error::Other("Codex has no active turn to steer".into()));
        };
        self.client
            .request(
                "turn/steer",
                json!({
                    "threadId": self.handle.runtime_id,
                    "expectedTurnId": turn_id,
                    "input": [{"type":"text", "text":text}],
                }),
            )
            .await?;
        Ok(())
    }

    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?;
        if let Some(home) = self.runtime_home.take() {
            home.cleanup()?;
        }
        Ok(())
    }
}

type PendingResponse = oneshot::Sender<std::result::Result<Value, String>>;
type PendingResponses = Arc<Mutex<HashMap<u64, PendingResponse>>>;

pub(super) struct JsonLineClient {
    stdin: Mutex<ChildStdin>,
    child: Mutex<Child>,
    pending: PendingResponses,
    next_id: Mutex<u64>,
    include_jsonrpc: bool,
    events: mpsc::UnboundedSender<Value>,
    process_group: Option<u32>,
}

impl JsonLineClient {
    pub(super) async fn spawn(
        launch: &RuntimeLaunch,
        cwd: Option<&std::path::Path>,
        include_jsonrpc: bool,
        protocol: &str,
    ) -> Result<(Arc<Self>, mpsc::UnboundedReceiver<Value>, RuntimeEndpoint)> {
        let mut command = Command::new(&launch.program);
        command
            .args(&launch.arguments)
            .envs(&launch.env)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        // Package-manager shims commonly spawn a native worker. Isolate the
        // complete adapter tree so close can reap it instead of orphaning the
        // worker with inherited protocol handles.
        #[cfg(unix)]
        command.process_group(0);
        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 stderr = child
            .stderr
            .take()
            .ok_or_else(|| Error::Other("runtime child has no stderr".into()))?;
        let pending: PendingResponses = Arc::new(Mutex::new(HashMap::new()));
        let (events_tx, events_rx) = mpsc::unbounded_channel();
        let reader_events = events_tx.clone();
        let reader_pending = pending.clone();
        tokio::spawn(async move {
            let mut stdout_lines = BufReader::new(stdout).lines();
            let mut stderr_lines = BufReader::new(stderr).lines();
            let mut stdout_open = true;
            let mut stderr_open = true;
            while stdout_open || stderr_open {
                tokio::select! {
                    line = stdout_lines.next_line(), if stdout_open => match line {
                        Ok(Some(line)) => {
                            let Ok(value) = serde_json::from_str::<Value>(&line) else {
                                let _ = reader_events.send(json!({"type": "malformed_output", "line": line}));
                                continue;
                            };
                            let response_id = value.get("id").and_then(Value::as_u64);
                            let is_response = value.get("result").is_some() || value.get("error").is_some();
                            if let Some(id) = response_id.filter(|_| is_response) {
                                if let Some(sender) = reader_pending.lock().await.remove(&id) {
                                    let result = if let Some(error) = value.get("error") {
                                        Err(error.to_string())
                                    } else {
                                        Ok(value.get("result").cloned().unwrap_or(Value::Null))
                                    };
                                    let _ = sender.send(result);
                                    continue;
                                }
                            }
                            let _ = reader_events.send(value);
                        }
                        Ok(None) => stdout_open = false,
                        Err(error) => {
                            let _ = reader_events.send(json!({"type": "transport_error", "message": error.to_string()}));
                            stdout_open = false;
                        }
                    },
                    line = stderr_lines.next_line(), if stderr_open => match line {
                        Ok(Some(line)) => {
                            let _ = reader_events.send(json!({"type": "transport_stderr", "line": line}));
                        }
                        Ok(None) => stderr_open = false,
                        Err(error) => {
                            let _ = reader_events.send(json!({"type": "transport_error", "message": error.to_string()}));
                            stderr_open = false;
                        }
                    }
                }
            }
            let _ = reader_events.send(json!({"type": "transport_closed"}));
            let mut pending = reader_pending.lock().await;
            for (_, sender) in pending.drain() {
                let _ = sender.send(Err("runtime protocol closed".into()));
            }
        });
        let endpoint = RuntimeEndpoint::LocalProcess {
            pid,
            command: std::iter::once(launch.program.clone())
                .chain(launch.arguments.iter().cloned())
                .collect(),
            protocol: protocol.into(),
        };
        Ok((
            Arc::new(Self {
                stdin: Mutex::new(stdin),
                child: Mutex::new(child),
                pending,
                next_id: Mutex::new(1),
                include_jsonrpc,
                events: events_tx,
                process_group: pid,
            }),
            events_rx,
            endpoint,
        ))
    }

    pub(super) async fn request(&self, method: &str, params: Value) -> Result<Value> {
        let (_id, rx) = self.begin_request(method, params).await?;
        rx.await
            .map_err(|_| Error::Other("runtime response channel closed".into()))?
            .map_err(|message| {
                Error::Other(format!("runtime request `{method}` failed: {message}"))
            })
    }

    pub(super) async fn begin_request(
        &self,
        method: &str,
        params: Value,
    ) -> Result<(u64, oneshot::Receiver<std::result::Result<Value, String>>)> {
        let id = {
            let mut next = self.next_id.lock().await;
            let id = *next;
            *next += 1;
            id
        };
        let (tx, rx) = oneshot::channel();
        self.pending.lock().await.insert(id, tx);
        let mut request = json!({"id": id, "method": method, "params": params});
        if self.include_jsonrpc {
            request["jsonrpc"] = json!("2.0");
        }
        if let Err(error) = self.write(&request).await {
            self.pending.lock().await.remove(&id);
            return Err(error);
        }
        Ok((id, rx))
    }

    pub(super) async fn notify(&self, method: &str, params: Value) -> Result<()> {
        let mut notification = json!({"method": method, "params": params});
        if self.include_jsonrpc {
            notification["jsonrpc"] = json!("2.0");
        }
        self.write(&notification).await
    }

    pub(super) async fn respond(&self, id: Value, result: Value) -> Result<()> {
        let mut response = json!({"id": id, "result": result});
        if self.include_jsonrpc {
            response["jsonrpc"] = json!("2.0");
        }
        self.write(&response).await
    }

    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(())
    }

    pub(super) fn emit(&self, value: Value) {
        let _ = self.events.send(value);
    }

    pub(super) async fn close(&self) -> Result<()> {
        let mut child = self.child.lock().await;
        #[cfg(unix)]
        if let Some(pid) = self.process_group {
            crate::lsp::kill_process_group(pid);
            tokio::time::timeout(Duration::from_secs(3), child.wait())
                .await
                .map_err(|_| Error::Other("timed out reaping runtime process group".into()))??;
            return Ok(());
        }
        #[cfg(not(unix))]
        if child.try_wait()?.is_none() {
            child.kill().await?;
        }
        Ok(())
    }
}

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

    #[test]
    fn codex_capabilities_do_not_claim_arbitrary_process_attach() {
        let capabilities = CodexRuntimeBackend::new().capabilities();
        assert!(capabilities.start_session);
        assert!(capabilities.resume_session);
        assert!(!capabilities.attach_existing_process);
        assert!(capabilities.send_input);
        assert!(capabilities.stream_events);
        assert!(capabilities.interrupt);
        assert!(capabilities.steer);
    }

    #[test]
    fn runtime_handle_is_language_neutral_json() {
        let handle = RuntimeHandle {
            harness: HarnessId::from(HarnessId::CODEX),
            runtime_id: "thread-1".into(),
            endpoint: RuntimeEndpoint::LocalProcess {
                pid: Some(42),
                command: vec!["codex".into(), "app-server".into()],
                protocol: "codex-app-server-jsonl".into(),
            },
        };
        let encoded = serde_json::to_string(&handle).unwrap();
        assert_eq!(
            serde_json::from_str::<RuntimeHandle>(&encoded).unwrap(),
            handle
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn codex_adapter_performs_handshake_start_and_turn() {
        let script = r#"
            i=0
            while IFS= read -r line; do
              i=$((i + 1))
              case "$i" in
                1) printf '%s\n' '{"id":1,"result":{"userAgent":"mock"}}' ;;
                2) ;;
                3) printf '%s\n' '{"id":2,"result":{"thread":{"id":"thr_mock"}}}' ;;
                4)
                  printf '%s\n' '{"id":3,"result":{"turn":{"id":"turn_mock"}}}'
                  printf '%s\n' '{"method":"turn/started","params":{"turn":{"id":"turn_mock"}}}'
                  ;;
                5) printf '%s\n' '{"id":4,"result":{"turnId":"turn_mock"}}' ;;
              esac
            done
        "#;
        let backend = CodexRuntimeBackend::with_launch(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, "thr_mock");
        assert_eq!(
            connection
                .send_input(RuntimeInput {
                    text: "hi".into(),
                    image_urls: Vec::new(),
                })
                .await
                .unwrap()
                .as_deref(),
            Some("turn_mock")
        );
        connection.steer("focus on tests".into()).await.unwrap();
        assert_eq!(
            connection.next_event().await.unwrap().unwrap().kind,
            "turn/started"
        );
        connection.close().await.unwrap();
    }
}