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>,
}

/// Connect to an already-running harness endpoint instead of spawning one.
///
/// The registry stores where the endpoint and its credential live — the
/// harness's own config file — never the values themselves. The service
/// resolves them when it opens the connection, so a rotated token or a moved
/// gateway is picked up on the next open without a registry change.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeConnectLaunch {
    /// Harness config file holding the endpoint; a leading `~/` expands to the
    /// caller's home directory at resolve time.
    pub config_path: String,
    /// JSON pointer to the endpoint address inside the config file.
    pub address_pointer: String,
    /// Optional JSON pointer to a PORT number in the config file, consulted
    /// when `address_pointer` names nothing: the address becomes that port on
    /// loopback under `default_address`'s scheme. Harnesses like openclaw
    /// configure a bare `gateway.port`, never a full URL.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub port_pointer: Option<String>,
    /// Optional fallback endpoint when neither pointer resolves — the
    /// harness's documented out-of-the-box endpoint. With this set, a missing
    /// or pointer-less config is the harness "running on defaults", not an
    /// error.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub default_address: Option<String>,
    /// Optional JSON pointer to the bearer credential inside the config file.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub auth_pointer: Option<String>,
    /// Protocol spoken at the endpoint.
    pub protocol: String,
}

/// Bearer credential whose `Debug` output never contains the secret.
#[derive(Clone, PartialEq, Eq)]
pub struct BearerToken(String);

impl BearerToken {
    /// Wrap a resolved credential.
    pub fn new(secret: impl Into<String>) -> Self {
        Self(secret.into())
    }

    /// The secret itself, for constructing an Authorization header.
    pub fn secret(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Debug for BearerToken {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("BearerToken(<redacted>)")
    }
}

/// Endpoint and credential resolved from a [`RuntimeConnectLaunch`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResolvedRuntimeConnection {
    /// Concrete endpoint address.
    pub address: String,
    /// Bearer credential when the launch declares one.
    pub auth: Option<BearerToken>,
}

impl RuntimeConnectLaunch {
    /// Resolve the endpoint address and credential from the harness's config
    /// file. Fails closed: a declared pointer that does not resolve to a
    /// non-empty string is an error, and diagnostics name the path and the
    /// pointer without echoing config contents.
    pub fn resolve(&self, home: &Path) -> Result<ResolvedRuntimeConnection> {
        let path = match self.config_path.strip_prefix("~/") {
            Some(rest) => home.join(rest),
            None => PathBuf::from(&self.config_path),
        };
        // A missing config file is the harness on documented defaults when
        // the descriptor declares them; otherwise it stays an error.
        let config: Value = match std::fs::read_to_string(&path) {
            Ok(raw) => serde_json::from_str(&raw).map_err(|_| {
                Error::Other(format!(
                    "connect-mode config {} is not valid JSON",
                    path.display()
                ))
            })?,
            Err(error) => {
                if self.default_address.is_some() {
                    Value::Object(Default::default())
                } else {
                    return Err(Error::Other(format!(
                        "connect-mode config {} is unreadable: {error}",
                        path.display()
                    )));
                }
            }
        };
        let field = |pointer: &str, name: &str| -> Result<String> {
            match config.pointer(pointer).and_then(Value::as_str) {
                Some(value) if !value.trim().is_empty() => Ok(value.trim().to_string()),
                _ => Err(Error::Other(format!(
                    "connect-mode {name} pointer `{pointer}` does not name a non-empty string in {}",
                    path.display()
                ))),
            }
        };
        // Address chain: explicit URL pointer → configured port on loopback →
        // the descriptor's documented default endpoint.
        let address = match config
            .pointer(&self.address_pointer)
            .and_then(Value::as_str)
        {
            Some(value) if !value.trim().is_empty() => value.trim().to_string(),
            _ => {
                let from_port = self
                    .port_pointer
                    .as_deref()
                    .and_then(|pointer| config.pointer(pointer))
                    .and_then(Value::as_u64)
                    .map(|port| {
                        let scheme = self
                            .default_address
                            .as_deref()
                            .and_then(|address| address.split_once("://"))
                            .map(|(scheme, _)| scheme)
                            .unwrap_or("ws");
                        format!("{scheme}://127.0.0.1:{port}")
                    });
                match from_port.or_else(|| self.default_address.clone()) {
                    Some(address) => address,
                    None => {
                        return Err(Error::Other(format!(
                            "connect-mode address pointer `{}` does not name a non-empty string in {}",
                            self.address_pointer,
                            path.display()
                        )));
                    }
                }
            }
        };
        let mut address = address.trim_end_matches('/').to_string();
        // Normalize a bare host:port to the endpoint's scheme — configs
        // routinely omit it and a scheme-less URL makes gateway clients fall
        // back to their compiled-in default endpoint instead.
        if !address.contains("://") {
            let scheme = self
                .default_address
                .as_deref()
                .and_then(|default| default.split_once("://"))
                .map(|(scheme, _)| scheme)
                .unwrap_or("ws");
            address = format!("{scheme}://{address}");
        }
        // Auth is optional exactly when the endpoint can run without it: a
        // declared pointer that resolves to nothing is only an error when no
        // default endpoint is declared (the original fail-closed contract).
        let auth = match &self.auth_pointer {
            Some(pointer) => match config.pointer(pointer).and_then(Value::as_str) {
                Some(value) if !value.trim().is_empty() => {
                    Some(BearerToken::new(value.trim().to_string()))
                }
                _ if self.default_address.is_some() => None,
                _ => Some(BearerToken::new(field(pointer, "auth")?)),
            },
            None => None,
        };
        Ok(ResolvedRuntimeConnection { address, auth })
    }
}

/// One stdio MCP server the caller wants mounted into the session it is
/// starting. Uniform shape; each backend translates it into whatever its own
/// harness accepts (the ACP backend into `session/new`'s `mcpServers`).
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct McpServerLaunch {
    /// Server name the harness registers the tools under.
    pub name: String,
    /// Executable to spawn.
    pub command: String,
    /// Arguments passed to it.
    #[serde(default)]
    pub arguments: Vec<String>,
    /// Extra environment for the spawned server.
    #[serde(default)]
    pub env: BTreeMap<String, String>,
}

/// Request to create a fresh runtime session.
#[derive(Debug, Clone, Default, 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>,
    /// MCP servers to mount into the new session, where the harness's own
    /// start door carries them. Backends that have no such door ignore it —
    /// their caller mounts through a config file instead.
    #[serde(default)]
    pub mcp_servers: Vec<McpServerLaunch>,
}

/// 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>,
    /// MCP servers to mount into the resumed session, exactly as a start
    /// request mounts them: a session's tools do not survive its process, so
    /// the caller that resumes it names them again.
    #[serde(default)]
    pub mcp_servers: Vec<McpServerLaunch>,
}

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

/// Keeps a runtime's closing diagnostics short enough to read in an error.
const STDERR_TAIL_LINES: usize = 20;
const STDERR_TAIL_CHARACTERS: usize = 2_000;

/// Reports a closed protocol together with whatever the runtime last said.
fn closed_reason(recent_stderr: &std::collections::VecDeque<String>) -> String {
    if recent_stderr.is_empty() {
        return "runtime protocol closed".into();
    }
    let mut tail = recent_stderr
        .iter()
        .map(String::as_str)
        .collect::<Vec<_>>()
        .join(" | ");
    if tail.chars().count() > STDERR_TAIL_CHARACTERS {
        tail = tail
            .chars()
            .take(STDERR_TAIL_CHARACTERS)
            .collect::<String>()
            + "…";
    }
    format!("runtime protocol closed: {tail}")
}

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>>>;

/// A spawned child that LEADS its own process group (`process_group(0)`), so
/// dropping it signals the whole group rather than just the leader.
///
/// `kill_on_drop(true)` reaches the direct child only. Harness launchers are
/// commonly package-manager shims that spawn the real worker — the worker
/// holding the protocol pipes — so a dropped launcher leaves that worker
/// running with nothing attached to it. Dropping is not a rare path: it is
/// what a blown deadline does to a launch or a control call still in flight.
///
/// The group is signalled only while `id()` still answers, i.e. while this
/// process has not been reaped here. A reaped leader's pid can be reused by
/// an unrelated group, and killing that group would be someone else's
/// outage; a graceful `close()` that reaped the group therefore makes this
/// drop a no-op.
pub(super) struct GroupLeader(Child);

impl std::ops::Deref for GroupLeader {
    type Target = Child;

    fn deref(&self) -> &Child {
        &self.0
    }
}

impl std::ops::DerefMut for GroupLeader {
    fn deref_mut(&mut self) -> &mut Child {
        &mut self.0
    }
}

impl Drop for GroupLeader {
    fn drop(&mut self) {
        #[cfg(unix)]
        if let Some(pid) = self.0.id() {
            crate::lsp::kill_process_group(pid);
        }
    }
}

pub(super) struct JsonLineClient {
    stdin: Mutex<ChildStdin>,
    child: Mutex<GroupLeader>,
    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;
            // A runtime that dies mid-handshake explains itself on stderr and
            // nowhere else. Events reach only an already-started runtime, so
            // without this the caller is told the protocol closed and never
            // told why.
            let mut recent_stderr: std::collections::VecDeque<String> =
                std::collections::VecDeque::new();
            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)) => {
                            if !line.trim().is_empty() {
                                if recent_stderr.len() == STDERR_TAIL_LINES {
                                    recent_stderr.pop_front();
                                }
                                recent_stderr.push_back(line.clone());
                            }
                            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 reason = closed_reason(&recent_stderr);
            let mut pending = reader_pending.lock().await;
            for (_, sender) in pending.drain() {
                let _ = sender.send(Err(reason.clone()));
            }
        });
        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(GroupLeader(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;
        // `process_group` is a pid COPY taken at spawn, and a reaped pid
        // belongs to whoever the OS hands it to next. Close is called more
        // than once — a hosted runtime closes on shutdown, on transport end,
        // and again when its host task exits — so the second call must find
        // this child still unreaped here before signalling anything, exactly
        // as the raw-line transport does.
        if child.try_wait()?.is_some() {
            return Ok(());
        }
        #[cfg(unix)]
        {
            match self.process_group {
                Some(pid) => crate::lsp::kill_process_group(pid),
                None => child.kill().await?,
            }
            tokio::time::timeout(Duration::from_secs(3), child.wait())
                .await
                .map_err(|_| Error::Other("timed out reaping runtime process group".into()))??;
        }
        #[cfg(not(unix))]
        child.kill().await?;
        Ok(())
    }
}

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

    #[test]
    fn closed_reason_reports_the_runtime_last_words() {
        let mut stderr = std::collections::VecDeque::new();
        stderr.push_back("grok: unsupported syscall SYS_execve".to_string());
        assert_eq!(
            closed_reason(&stderr),
            "runtime protocol closed: grok: unsupported syscall SYS_execve",
        );
    }

    #[test]
    fn closed_reason_stays_bare_without_stderr() {
        assert_eq!(
            closed_reason(&std::collections::VecDeque::new()),
            "runtime protocol closed",
        );
    }

    #[test]
    fn closed_reason_truncates_a_long_tail() {
        let mut stderr = std::collections::VecDeque::new();
        stderr.push_back("x".repeat(STDERR_TAIL_CHARACTERS + 500));
        let reason = closed_reason(&stderr);
        assert!(reason.ends_with('…'), "{reason}");
        assert_eq!(
            reason.chars().count(),
            "runtime protocol closed: ".chars().count() + STDERR_TAIL_CHARACTERS + 1,
        );
    }

    fn scratch_home(tag: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!(
            "supercode-connect-launch-{tag}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    #[test]
    fn connect_launch_resolves_address_and_auth_from_the_harness_config() {
        let home = scratch_home("resolve");
        std::fs::create_dir_all(home.join(".gateway")).unwrap();
        std::fs::write(
            home.join(".gateway/config.json"),
            r#"{"gateway": {"url": "ws://127.0.0.1:18789/", "auth": {"token": "secret-credential"}}}"#,
        )
        .unwrap();
        let launch = RuntimeConnectLaunch {
            config_path: "~/.gateway/config.json".into(),
            address_pointer: "/gateway/url".into(),
            port_pointer: None,
            default_address: None,
            auth_pointer: Some("/gateway/auth/token".into()),
            protocol: "acp-v1-jsonrpc".into(),
        };
        let resolved = launch.resolve(&home).unwrap();
        assert_eq!(resolved.address, "ws://127.0.0.1:18789");
        assert_eq!(
            resolved.auth.as_ref().unwrap().secret(),
            "secret-credential"
        );
        let debugged = format!("{resolved:?}");
        assert!(!debugged.contains("secret-credential"));
        assert!(debugged.contains("<redacted>"));
    }

    #[test]
    fn connect_launch_resolution_fails_closed_without_echoing_config_contents() {
        let home = scratch_home("fail-closed");
        let launch = RuntimeConnectLaunch {
            config_path: "~/missing.json".into(),
            address_pointer: "/url".into(),
            port_pointer: None,
            default_address: None,
            auth_pointer: None,
            protocol: "acp-v1-jsonrpc".into(),
        };
        assert!(launch.resolve(&home).is_err());

        std::fs::write(
            home.join("present.json"),
            r#"{"url": "", "auth": {"token": "secret-credential"}}"#,
        )
        .unwrap();
        let empty_address = RuntimeConnectLaunch {
            config_path: "~/present.json".into(),
            address_pointer: "/url".into(),
            port_pointer: None,
            default_address: None,
            auth_pointer: None,
            protocol: "acp-v1-jsonrpc".into(),
        };
        let error = empty_address.resolve(&home).unwrap_err();
        assert!(error.to_string().contains("/url"));
        assert!(!error.to_string().contains("secret-credential"));

        let missing_auth = RuntimeConnectLaunch {
            config_path: "~/present.json".into(),
            address_pointer: "/auth/token".into(),
            port_pointer: None,
            default_address: None,
            auth_pointer: Some("/absent".into()),
            protocol: "acp-v1-jsonrpc".into(),
        };
        let error = missing_auth.resolve(&home).unwrap_err();
        assert!(error.to_string().contains("/absent"));
        assert!(!error.to_string().contains("secret-credential"));
    }

    #[test]
    fn connect_launch_round_trips_through_json() {
        let launch = RuntimeConnectLaunch {
            config_path: "~/.openclaw/openclaw.json".into(),
            address_pointer: "/gateway/url".into(),
            port_pointer: None,
            default_address: None,
            auth_pointer: Some("/gateway/token".into()),
            protocol: "acp-v1-jsonrpc".into(),
        };
        let encoded = serde_json::to_value(&launch).unwrap();
        let decoded: RuntimeConnectLaunch = serde_json::from_value(encoded).unwrap();
        assert_eq!(decoded, launch);
        let minimal: RuntimeConnectLaunch = serde_json::from_value(json!({
            "config_path": "~/.gateway.json",
            "address_pointer": "/url",
            "protocol": "http",
        }))
        .unwrap();
        assert_eq!(minimal.auth_pointer, None);
    }

    #[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,
                mcp_servers: Vec::new(),
            })
            .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();
    }
}