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, PathBuf};
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::{
    BearerToken, HarnessEvent, JsonLineClient, McpServerLaunch, 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>,
        mcp_servers: &[McpServerLaunch],
        resume: bool,
    ) -> Result<Box<dyn RuntimeConnection>> {
        let mut launch = launch.unwrap_or_else(|| self.launch.clone());
        // Pi has no MCP client of its own; `pi-mcp-adapter` is the extension
        // that gives it one, loaded for this process from npm by Pi's own
        // package door and pointed at the servers written for this runtime,
        // as Claude Code's backend points `--mcp-config` at the same file shape.
        if !mcp_servers.is_empty() {
            let file = mcp_config_file("pi", &runtime_id, mcp_servers).await?;
            launch.arguments.extend([
                "--extension".into(),
                std::env::var(PI_MCP_ADAPTER_ENV)
                    .ok()
                    .filter(|value| !value.is_empty())
                    .unwrap_or_else(|| PI_MCP_ADAPTER.into()),
                "--mcp-config".into(),
                file.to_string_lossy().into_owned(),
            ]);
        }
        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,
            steer: false,
            respond_to_requests: true,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        self.open(
            &request.cwd,
            generated_session_id(),
            request.launch,
            &request.mcp_servers,
            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,
            &request.mcp_servers,
            true,
        )
        .await
    }
}

/// The Pi extension that mounts MCP servers, pinned: Pi fetches an `npm:`
/// extension into its own package directory on first use.
const PI_MCP_ADAPTER: &str = "npm:pi-mcp-adapter@2.37.0";
/// Where the adapter already is, for a Pi that cannot fetch one: a path or a
/// Pi package source in place of `PI_MCP_ADAPTER` (a browser tab's Pi pack
/// carries it, and the tab's npm resolves no `npm:` source).
const PI_MCP_ADAPTER_ENV: &str = "SUPERCODE_PI_MCP_ADAPTER";

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>> {
        if !input.image_urls.is_empty() {
            return Err(Error::Other(
                "Pi RPC image input is not verified by the installed protocol contract".into(),
            ));
        }
        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 with supercode registered as the CLI's permission handler.
///
/// It can create/resume sessions, cancel the running turn through the
/// stream-json control channel, and answer the permission requests the CLI
/// raises while a tool call is blocked.
#[derive(Debug, Clone)]
pub struct ClaudeCodeRuntimeBackend {
    launch: RuntimeLaunch,
    permission_timeout: Duration,
}

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

impl ClaudeCodeRuntimeBackend {
    /// Use `claude` from `PATH` in bidirectional stream-json mode, with
    /// supercode registered as the permission handler.
    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(),
                    // Register this adapter as the permission handler. Measured
                    // against claude 2.1.258: without it a tool call that needs
                    // an answer is refused outright with
                    // `{"type":"system","subtype":"permission_denied"}`; with
                    // it the CLI raises a `can_use_tool` control request on
                    // stdout and blocks the turn until a `control_response`
                    // arrives. See
                    // `docs/interop/research/orc2-claude-respond-receipt-2026-09-04.json`.
                    "--permission-prompt-tool".into(),
                    "stdio".into(),
                ],
                env: BTreeMap::new(),
            },
            permission_timeout: CLAUDE_PERMISSION_RESPONSE_TIMEOUT,
        }
    }

    /// The command prefix this backend launches: what the registry publishes
    /// as `default_launch`, so a caller that hands the published launch back
    /// (the orchestrator does) gets stream-json mode and not the interactive
    /// TUI blocked on a pipe.
    pub fn launch(&self) -> &RuntimeLaunch {
        &self.launch
    }

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

    /// Override how long an unanswered permission request is held before this
    /// adapter denies it on the caller's behalf.
    ///
    /// See [`CLAUDE_PERMISSION_RESPONSE_TIMEOUT`] for the default and what it
    /// was measured against.
    pub fn with_permission_timeout(mut self, timeout: Duration) -> Self {
        self.permission_timeout = timeout;
        self
    }

    async fn open(
        &self,
        cwd: &Path,
        runtime_id: String,
        launch: Option<RuntimeLaunch>,
        mcp_servers: &[McpServerLaunch],
        resume: bool,
    ) -> Result<Box<dyn RuntimeConnection>> {
        // The prefix WITHOUT session arguments is kept: it is what reopens
        // this same session later, and appending `--session-id` to a launch
        // that already carries one is what the CLI refuses.
        let mut prefix = launch.unwrap_or_else(|| self.launch.clone());
        // MCP servers ride the prefix, not the session: Claude Code mounts them
        // from `--mcp-config <file>` on every process, so the file written for
        // this runtime is what every reopen of the session hands back too.
        if !mcp_servers.is_empty() {
            let file = mcp_config_file("claude", &runtime_id, mcp_servers).await?;
            prefix
                .arguments
                .extend(["--mcp-config".into(), file.to_string_lossy().into_owned()]);
        }
        let mut launch = prefix.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,
            prefix,
            cwd: cwd.to_path_buf(),
            spoke: false,
            buffered_events: VecDeque::new(),
            next_control_request: 1,
            control_timeout: CLAUDE_CONTROL_RESPONSE_TIMEOUT,
            pending_permissions: Vec::new(),
            permission_timeout: self.permission_timeout,
            mounted_mcp_servers: mcp_servers
                .iter()
                .map(|server| server.name.clone())
                .collect(),
        }))
    }
}

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

/// How long an unanswered `can_use_tool` request is held before this adapter
/// denies it on the caller's behalf.
///
/// Measured against claude 2.1.258: a `can_use_tool` control request the host
/// never answers blocks the turn indefinitely — the probe watched one sit for
/// 20 s with no further frame and no tool result, and the CLI has no bound of
/// its own (receipt step `unanswered_request_blocks_the_turn` in
/// `docs/interop/research/orc2-claude-respond-receipt-2026-09-04.json`). The
/// bound therefore has to live here. Five minutes is long enough for an
/// operator to read `supercode approvals list` and answer, and short enough
/// that a forgotten prompt does not wedge a driven session forever. Override
/// per backend with [`ClaudeCodeRuntimeBackend::with_permission_timeout`].
pub const CLAUDE_PERMISSION_RESPONSE_TIMEOUT: Duration = Duration::from_secs(300);

/// The `behavior` values Claude Code's permission handler protocol accepts.
///
/// Measured against claude 2.1.258: anything else — including a `deny` with no
/// `message` — is refused by the CLI with "The canUseTool callback returned an
/// invalid permission result. Expected {behavior: 'allow', updatedInput?:
/// object} or {behavior: 'deny', message: string}."
const CLAUDE_PERMISSION_BEHAVIORS: [&str; 2] = ["allow", "deny"];

/// What this adapter says when it denies a request nobody answered in time.
const CLAUDE_PERMISSION_TIMEOUT_MESSAGE: &str =
    "Volter Harness denied this permission request: no answer arrived before the adapter's \
     permission timeout elapsed";

#[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,
            steer: true,
            respond_to_requests: true,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        let session_id = generated_session_id();
        // The agent the launch named is recorded in the session's config home before its transcript exists, so
        // whoever discovers the session finds its agent with it (launch_agent.rs).
        let env = request
            .launch
            .as_ref()
            .map(|launch| launch.env.clone())
            .unwrap_or_default();
        let home = env.get("CLAUDE_CONFIG_DIR").map(PathBuf::from).or_else(|| {
            crate::HarnessHomes::default()
                .claude_code
                .parent()
                .map(Path::to_path_buf)
        });
        if let Some(home) = home {
            if let Err(error) = crate::launch_agent::record(&home, &session_id, &env) {
                tracing::warn!(
                    "could not record the launch agent of Claude Code session {session_id}: {error}"
                );
            }
        }
        self.open(
            &request.cwd,
            session_id,
            request.launch,
            &request.mcp_servers,
            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,
            &request.mcp_servers,
            true,
        )
        .await
    }
}

/// Claude Code's own MCP configuration shape for the servers a start request
/// mounts, written once per runtime under the OS temp directory (mode 0600 on
/// Unix: a server's env may carry a token meant for that server alone).
async fn mcp_config_file(
    harness: &str,
    runtime_id: &str,
    servers: &[McpServerLaunch],
) -> Result<PathBuf> {
    let mut entries = serde_json::Map::new();
    for server in servers {
        entries.insert(
            server.name.clone(),
            json!({
                "type": "stdio",
                "command": server.command,
                "args": server.arguments,
                "env": server.env,
            }),
        );
    }
    let safe: String = runtime_id
        .chars()
        .filter(|c| c.is_ascii_alphanumeric() || *c == '-' || *c == '_')
        .collect();
    let path = std::env::temp_dir().join(format!("supercode-{harness}-mcp-{safe}.json"));
    let body = serde_json::to_vec_pretty(&json!({ "mcpServers": entries })).map_err(|error| {
        Error::Other(format!(
            "{harness} mcp config could not be encoded: {error}"
        ))
    })?;
    tokio::fs::write(&path, body).await.map_err(|error| {
        Error::Other(format!(
            "{harness} mcp config could not be written: {error}"
        ))
    })?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        tokio::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))
            .await
            .map_err(|error| {
                Error::Other(format!(
                    "{harness} mcp config could not be protected: {error}"
                ))
            })?;
    }
    Ok(path)
}

struct ClaudeRuntimeConnection {
    handle: RuntimeHandle,
    transport: RawLineTransport,
    /// The launch prefix this session was opened with, carrying no session
    /// argument of its own: half of what reopens the session after its
    /// process has exited.
    prefix: RuntimeLaunch,
    /// The workspace the session belongs to. `--resume` resolves a session id
    /// against the project the process runs in, so a reopen has to run in the
    /// same directory the first process did.
    cwd: PathBuf,
    /// Whether the process now behind this connection has produced a frame.
    /// A process that exited without ever speaking never opened a session, so
    /// its close is a launch failure and is reported; one that spoke left a
    /// session `--resume` reopens.
    spoke: bool,
    /// 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,
    /// Permission requests the CLI raised and is still blocked on, oldest
    /// first. A request is recorded when this adapter hands it to the event
    /// consumer and dropped when it is answered or denied on timeout.
    pending_permissions: Vec<PendingPermission>,
    permission_timeout: Duration,
    /// Names of the MCP servers the START REQUEST mounted. A caller that mounts
    /// a server intends its tools: a `can_use_tool` for `mcp__<name>__…` is
    /// allowed here at once, never held for the caller to answer.
    mounted_mcp_servers: Vec<String>,
}

/// One `can_use_tool` control request Claude Code is blocked on.
struct PendingPermission {
    /// The CLI's own `request_id`, echoed verbatim in the answer.
    request_id: String,
    /// When this adapter denies it if nobody has answered.
    deadline: tokio::time::Instant,
}

impl ClaudeRuntimeConnection {
    fn is_relay(&self) -> bool {
        self.prefix
            .env
            .get("SUPERCODE_CLAUDE_RELAY")
            .is_some_and(|value| value == "1")
    }

    /// Whether the `claude` process behind this connection has exited.
    ///
    /// A stream-json `claude --print` process is the session's CURRENT
    /// speaker, not the session: it ends when it has nothing left to do (its
    /// own exit, a crash, a reaped process group), while the session it wrote
    /// stays on disk under the same id.
    async fn process_ended(&self) -> bool {
        matches!(self.transport.child.lock().await.try_wait(), Ok(Some(_)))
    }

    /// Put a live `claude` process back behind this connection by resuming the
    /// session it already owns.
    ///
    /// The runtime id, the connection and the endpoint's protocol are
    /// unchanged — only the process is new — so a caller of `harness.v1`
    /// never learns that one ended. Verified against claude 2.1.258:
    /// `--resume <id>` in bidirectional stream-json print mode reopens the
    /// SAME session id (its `system/init` and every `result` echo it) with the
    /// conversation intact, and goes on reading turns from stdin.
    async fn reopen(&mut self) -> Result<()> {
        let mut launch = self.prefix.clone();
        launch
            .arguments
            .extend(["--resume".into(), self.handle.runtime_id.clone()]);
        let transport = RawLineTransport::spawn(&launch, Some(&self.cwd), "claude-stream-json")
            .await
            .map_err(|error| {
                Error::Other(format!(
                    "could not resume Claude Code session `{}` after its process exited: {error}",
                    self.handle.runtime_id
                ))
            })?;
        self.handle.endpoint = transport.endpoint.clone();
        self.transport = transport;
        // Whatever the dead process was blocked on died with it: those ids
        // name nothing the new process can be told about.
        self.pending_permissions.clear();
        self.spoke = false;
        Ok(())
    }

    /// Write one frame to the CLI, resuming the session first when the process
    /// that was reading stdin is gone.
    ///
    /// Both orders are covered because both happen: the exit can be visible
    /// before the write (the child is reaped) or only in the write itself (a
    /// grandchild — a hook, an MCP server — still holds the pipe, so nothing
    /// upstream has noticed the exit and only `EPIPE` reveals it).
    async fn write_turn(&mut self, frame: Value) -> Result<()> {
        if self.process_ended().await {
            if self.is_relay() {
                return Err(Error::Other("Claude relay process exited".into()));
            }
            self.reopen().await?;
        }
        match self.transport.write(frame.clone()).await {
            Ok(()) => Ok(()),
            Err(error) if broken_pipe(&error) && !self.is_relay() => {
                self.reopen().await?;
                self.transport.write(frame).await
            }
            Err(error) => Err(error),
        }
    }

    /// 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)")
                    ))
            )))),
        }
    }

    /// The CLI's own `request_id` when this frame is a permission request.
    ///
    /// Ground truth (claude 2.1.258, recorded live): a tool call that needs an
    /// answer from the registered permission handler arrives as
    /// `{"type":"control_request","request_id":"<uuid>","request":{"subtype":"can_use_tool","tool_name":…,"input":{…},"permission_suggestions":[…],"tool_use_id":…}}`.
    /// No other inbound `control_request` subtype is answered here.
    fn permission_request_id(value: &Value) -> Option<&str> {
        if value.get("type").and_then(Value::as_str)? != "control_request" {
            return None;
        }
        let request = value.get("request")?;
        if request.get("subtype").and_then(Value::as_str)? != "can_use_tool" {
            return None;
        }
        value.get("request_id").and_then(Value::as_str)
    }

    /// The request id of a `can_use_tool` for a tool of an MCP server this
    /// connection's start request mounted (`mcp__<server>__<tool>`), if that
    /// is what the frame is.
    fn mounted_tool_permission_request(&self, payload: &Value) -> Option<String> {
        let request_id = Self::permission_request_id(payload)?;
        let tool = payload
            .get("request")?
            .get("tool_name")
            .and_then(Value::as_str)?;
        let mounted = self.mounted_mcp_servers.iter().any(|name| {
            tool.strip_prefix("mcp__")
                .and_then(|rest| rest.strip_prefix(name.as_str()))
                .is_some_and(|rest| rest.starts_with("__"))
        });
        mounted.then(|| request_id.to_string())
    }

    /// Start the clock on a permission request about to reach the consumer.
    fn note_permission_request(&mut self, payload: &Value) {
        let Some(request_id) = Self::permission_request_id(payload) else {
            return;
        };
        if self
            .pending_permissions
            .iter()
            .any(|pending| pending.request_id == request_id)
        {
            return;
        }
        self.pending_permissions.push(PendingPermission {
            request_id: request_id.to_string(),
            deadline: tokio::time::Instant::now() + self.permission_timeout,
        });
    }

    /// Write one `control_response` for a request the CLI is blocked on.
    async fn write_permission_response(&mut self, request_id: &str, body: Value) -> Result<()> {
        self.transport
            .write(json!({
                "type": "control_response",
                "response": {
                    "subtype": "success",
                    "request_id": request_id,
                    "response": body,
                },
            }))
            .await
    }

    /// Deny every held request whose deadline has passed.
    ///
    /// Claude Code blocks the turn forever on an unanswered request, so an
    /// unanswered request is denied rather than left to wedge the session.
    async fn deny_expired_permissions(&mut self) -> Result<()> {
        let now = tokio::time::Instant::now();
        let expired = self
            .pending_permissions
            .iter()
            .filter(|pending| pending.deadline <= now)
            .map(|pending| pending.request_id.clone())
            .collect::<Vec<_>>();
        self.pending_permissions
            .retain(|pending| pending.deadline > now);
        for request_id in expired {
            self.write_permission_response(
                &request_id,
                json!({"behavior": "deny", "message": CLAUDE_PERMISSION_TIMEOUT_MESSAGE}),
            )
            .await?;
        }
        Ok(())
    }

    /// What `next_event` reports when the CLI's stdout ended.
    ///
    /// For every other harness a closed transport is a closed runtime, and
    /// reporting it is right. Claude Code is the one whose process is not the
    /// session: `claude --print` exits when it has answered, and the session
    /// it wrote is reopened by the next `send_input`. Reporting that exit as
    /// `transport_closed` is what made this harness the only one whose
    /// connection died between turns — the service drops a closed runtime, so
    /// the caller's next input is refused on a connection the harness itself
    /// considers perfectly resumable.
    ///
    /// So a session that has spoken PARKS here instead: the connection stays,
    /// and it has nothing to say until someone speaks to it again — which is
    /// exactly the state an idle runtime of any other harness is in. A process
    /// that exited without ever speaking opened no session to park on, and its
    /// close is reported as before.
    async fn transport_ended(&mut self) -> Result<Option<HarnessEvent>> {
        // A relay is a persistent mailbox speaker. Its death must close its host;
        // parking it would leave a busy host steering messages into a dead pipe.
        if self.is_relay() || !self.spoke {
            return Ok(None);
        }
        std::future::pending().await
    }

    /// How long until the oldest held request must be denied.
    fn next_permission_deadline(&self) -> Option<Duration> {
        let now = tokio::time::Instant::now();
        self.pending_permissions
            .iter()
            .map(|pending| pending.deadline.saturating_duration_since(now))
            .min()
    }
}

/// Validate one caller-supplied answer against the permission-result shape
/// Claude Code accepts, filling in the parts the CLI requires.
///
/// The uniform door (`harness.v1.approvals.resolve`) sends
/// `{"behavior":"allow"}` or `{"behavior":"deny","message":…}`; a caller using
/// `harness.v1.runtimes.respond` directly may add the CLI's optional fields
/// (`updatedInput`, `updatedPermissions`) and they are passed through
/// untouched. Anything without a recognized `behavior` is refused by name
/// rather than guessed at, because the CLI itself refuses it.
fn claude_permission_result(response: Value) -> Result<Value> {
    let Value::Object(mut body) = response else {
        return Err(claude_permission_shape_error(&response));
    };
    match body.get("behavior").and_then(Value::as_str) {
        Some("allow") => {}
        Some("deny") => {
            // Measured: the CLI rejects a `deny` with no `message`.
            let empty = body
                .get("message")
                .and_then(Value::as_str)
                .is_none_or(str::is_empty);
            if empty {
                body.insert(
                    "message".into(),
                    Value::String("Volter Harness denied this permission request".into()),
                );
            }
        }
        _ => return Err(claude_permission_shape_error(&Value::Object(body))),
    }
    Ok(Value::Object(body))
}

fn claude_permission_shape_error(response: &Value) -> Error {
    Error::Other(format!(
        "Claude Code permission answers must carry a `behavior` of {}; got {response}",
        CLAUDE_PERMISSION_BEHAVIORS
            .map(|behavior| format!("`{behavior}`"))
            .join(" or "),
    ))
}

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

    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>> {
        let content = if input.image_urls.is_empty() {
            Value::String(input.text)
        } else {
            let mut parts = Vec::new();
            if !input.text.is_empty() {
                parts.push(json!({"type":"text", "text":input.text}));
            }
            for url in input.image_urls {
                parts.push(claude_image_part(&url)?);
            }
            Value::Array(parts)
        };
        self.write_turn(json!({
            "type": "user",
            "session_id": self.handle.runtime_id,
            "message": {"role": "user", "content": content},
        }))
        .await?;
        Ok(None)
    }

    async fn next_event(&mut self) -> Result<Option<HarnessEvent>> {
        if let Some(payload) = self.buffered_events.pop_front() {
            self.spoke = true;
            return Ok(Some(harness_event(payload)));
        }
        loop {
            // A held permission request has a deadline, so the wait is bounded
            // by the nearest one: an unanswered request is denied here rather
            // than left blocking the CLI's turn forever.
            self.deny_expired_permissions().await?;
            if self.is_relay() && self.process_ended().await {
                return Ok(None);
            }
            let payload = match self
                .next_permission_deadline()
                .map(|remaining| remaining.min(Duration::from_millis(200)))
            {
                Some(remaining) => {
                    match tokio::time::timeout(remaining, self.transport.receiver.recv()).await {
                        Err(_) => continue,
                        Ok(None) => return self.transport_ended().await,
                        Ok(Some(payload)) => payload,
                    }
                }
                None => match tokio::time::timeout(
                    Duration::from_millis(200),
                    self.transport.receiver.recv(),
                )
                .await
                {
                    Err(_) => continue,
                    Ok(None) => return self.transport_ended().await,
                    Ok(Some(payload)) => payload,
                },
            };
            self.spoke = true;
            if Self::is_control_response(&payload) {
                continue;
            }
            if let Some(request_id) = self.mounted_tool_permission_request(&payload) {
                self.write_permission_response(&request_id, json!({"behavior": "allow"}))
                    .await?;
                continue;
            }
            self.note_permission_request(&payload);
            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<()> {
        // A session whose process has exited has no turn in flight, and the
        // adapter does not start one just to cancel nothing. This is the same
        // answer the CLI gives an interrupt with nothing running.
        if self.process_ended().await {
            return Ok(());
        }
        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;
                    }
                    // A permission request raised while the interrupt was in
                    // flight is still a request the CLI is blocked on: start
                    // its clock now, not when the consumer drains the buffer.
                    if let Some(request_id) = self.mounted_tool_permission_request(&payload) {
                        self.write_permission_response(&request_id, json!({"behavior": "allow"}))
                            .await?;
                        continue;
                    }
                    self.note_permission_request(&payload);
                    self.buffered_events.push_back(payload);
                }
            }
        }
    }

    async fn steer(&mut self, text: String) -> Result<()> {
        self.send_input(RuntimeInput {
            text,
            image_urls: Vec::new(),
        })
        .await
        .map(|_| ())
    }

    /// Answer one `can_use_tool` permission request through the stream-json
    /// control channel.
    ///
    /// Only a request this adapter has surfaced to the event consumer is
    /// answerable: the CLI's `request_id` is the identity, the answer is sent
    /// once, and a second answer for the same id is refused rather than
    /// silently ignored (claude 2.1.258 logs and drops a duplicate
    /// `control_response`).
    async fn respond(&mut self, request_id: Value, response: Value) -> Result<()> {
        let Some(request_id) = request_id.as_str().map(str::to_string) else {
            return Err(Error::Other(format!(
                "Claude Code control requests are identified by a string `request_id`; got \
                 {request_id}"
            )));
        };
        let Some(index) = self
            .pending_permissions
            .iter()
            .position(|pending| pending.request_id == request_id)
        else {
            return Err(Error::Other(format!(
                "no Claude Code permission request `{request_id}` is waiting on this connection — \
                 a `can_use_tool` request is answerable only while its turn is blocked on it, and \
                 only until it is answered or denied on timeout"
            )));
        };
        let body = claude_permission_result(response)?;
        self.pending_permissions.remove(index);
        self.write_permission_response(&request_id, body).await
    }

    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": "Volter Harness",
                        "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>,
        mcp_servers: Vec<McpServerLaunch>,
    ) -> Result<Box<dyn RuntimeConnection>> {
        let mcp_servers = acp_mcp_servers(&mcp_servers);
        // Supercode's ACP server serves exactly the session it was launched
        // with, and a resumed Supercode conversation continues under a new
        // id. So Supercode is resumed by launching the continuation
        // (`supercode resume <id> --acp`, with the default launch's flags) and
        // opening the session that process serves; asking a fresh `acp`
        // process to resume another id is always refused.
        let (launch, runtime_id) = match runtime_id {
            Some(session_id) if self.harness.as_str() == HarnessId::SUPERCODE => {
                let mut continuation = launch.unwrap_or_else(|| self.launch.clone());
                let flags = continuation
                    .arguments
                    .iter()
                    .skip_while(|argument| argument.as_str() != "acp")
                    .skip(1)
                    .cloned()
                    .collect::<Vec<_>>();
                continuation.arguments = ["resume", &session_id, "--harness", "supercode", "--acp"]
                    .into_iter()
                    .map(String::from)
                    .chain(flags)
                    .collect();
                (Some(continuation), None)
            }
            other => (launch, other),
        };
        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": mcp_servers}),
            )
            .await?;
            session_id
        } else {
            self.session_request(
                client.as_ref(),
                &initialized,
                "session/new",
                json!({"cwd": cwd, "mcpServers": mcp_servers}),
            )
            .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,
        }))
    }
}

/// The uniform [`McpServerLaunch`] list in ACP's own `session/new` shape:
/// `{name, command, args, env: [{name, value}]}` (Agent Client Protocol v1
/// `McpServer`, the stdio form). An empty list stays `[]`, which is what this
/// backend has always sent.
fn acp_mcp_servers(servers: &[McpServerLaunch]) -> Value {
    Value::Array(
        servers
            .iter()
            .map(|server| {
                json!({
                    "name": server.name,
                    "command": server.command,
                    "args": server.arguments,
                    "env": server
                        .env
                        .iter()
                        .map(|(name, value)| json!({"name": name, "value": value}))
                        .collect::<Vec<_>>(),
                })
            })
            .collect::<Vec<_>>(),
    )
}

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,
            steer: false,
            respond_to_requests: true,
        }
    }

    async fn start(&self, request: RuntimeStartRequest) -> Result<Box<dyn RuntimeConnection>> {
        self.connection(&request.cwd, None, request.launch, request.mcp_servers)
            .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,
            request.mcp_servers,
        )
        .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 mut prompt = Vec::new();
        if !input.text.is_empty() {
            prompt.push(json!({"type": "text", "text": input.text}));
        }
        for url in input.image_urls {
            let (mime_type, data) = data_image_parts(&url).ok_or_else(|| {
                Error::Other("ACP image prompts require base64 image data URLs".into())
            })?;
            prompt.push(json!({"type":"image", "mimeType":mime_type, "data":data}));
        }
        let (id, response) = self
            .client
            .begin_request(
                "session/prompt",
                json!({
                    "sessionId": self.handle.runtime_id,
                    "prompt": prompt,
                }),
            )
            .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>,
    bearer: Option<BearerToken>,
}

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

    /// Send the resolved connect-mode credential as a bearer Authorization
    /// header on every request to the joined server.
    pub fn with_bearer(mut self, token: BearerToken) -> Self {
        self.bearer = Some(token);
        self
    }

    fn http_client(&self) -> Result<reqwest::Client> {
        let Some(token) = &self.bearer else {
            return Ok(reqwest::Client::new());
        };
        let mut headers = reqwest::header::HeaderMap::new();
        let mut value =
            reqwest::header::HeaderValue::from_str(&format!("Bearer {}", token.secret())).map_err(
                |_| Error::Other("connect-mode bearer token is not a valid header value".into()),
            )?;
        value.set_sensitive(true);
        headers.insert(reqwest::header::AUTHORIZATION, value);
        reqwest::Client::builder()
            .default_headers(headers)
            .build()
            .map_err(|error| Error::Other(format!("could not build HTTP client: {error}")))
    }

    async fn service(
        &self,
        client: &reqwest::Client,
        launch: Option<RuntimeLaunch>,
    ) -> Result<(String, Option<super::GroupLeader>)> {
        if let Some(base_url) = &self.base_url {
            wait_for_health(client, 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);
        // `kill_on_drop` only targets the launcher, and a startup that blows
        // its caller's deadline is DROPPED mid-wait. `GroupLeader` turns that
        // drop into a group signal; the explicit teardown below covers a
        // startup that fails while this future is still running.
        let mut child = super::GroupLeader(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(client, &base_url).await {
            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 client = self.http_client()?;
        let (base_url, child) = self.service(&client, launch).await?;
        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,
            steer: false,
            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<super::GroupLeader>,
}

#[async_trait]
impl RuntimeConnection for OpenCodeRuntimeConnection {
    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}));
        }
        for url in input.image_urls {
            let mime = image_mime_type(&url).ok_or_else(|| {
                Error::Other("OpenCode image prompts require a recognizable image MIME type".into())
            })?;
            parts.push(json!({"type":"file", "mime":mime, "url":url}));
        }
        http_ok(
            self.client
                .post(format!(
                    "{}/session/{}/prompt_async",
                    self.base_url, self.handle.runtime_id
                ))
                .query(&[("directory", &self.cwd)])
                .json(&json!({"parts": parts}))
                .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 data_image_parts(url: &str) -> Option<(&str, &str)> {
    let rest = url.strip_prefix("data:")?;
    let (mime_type, data) = rest.split_once(";base64,")?;
    mime_type.starts_with("image/").then_some((mime_type, data))
}

fn image_mime_type(url: &str) -> Option<&str> {
    if let Some((mime_type, _)) = data_image_parts(url) {
        return Some(mime_type);
    }
    let path = url.split(['?', '#']).next()?.to_ascii_lowercase();
    if path.ends_with(".png") {
        Some("image/png")
    } else if path.ends_with(".jpg") || path.ends_with(".jpeg") {
        Some("image/jpeg")
    } else if path.ends_with(".gif") {
        Some("image/gif")
    } else if path.ends_with(".webp") {
        Some("image/webp")
    } else {
        None
    }
}

fn claude_image_part(url: &str) -> Result<Value> {
    if let Some((media_type, data)) = data_image_parts(url) {
        return Ok(json!({
            "type":"image",
            "source":{"type":"base64", "media_type":media_type, "data":data}
        }));
    }
    if url.starts_with("https://") || url.starts_with("http://") {
        return Ok(json!({"type":"image", "source":{"type":"url", "url":url}}));
    }
    Err(Error::Other(
        "Claude image prompts require image data URLs or HTTP(S) URLs".into(),
    ))
}

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<super::GroupLeader>,
    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);
        // A raw-protocol launcher spawns its worker the same way a JSON-line
        // one does. Give it its own group so close — and a dropped deadline —
        // reap the worker instead of orphaning it holding this stdout.
        #[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 (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(super::GroupLeader(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_some() {
            return Ok(());
        }
        // `Child::kill` reaches the launcher only. Signal the whole group so
        // a shim's worker cannot outlive the connection that owns it.
        #[cfg(unix)]
        if let Some(pid) = child.id() {
            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))]
        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,
    }
}

/// Whether this error is the write end of a pipe whose reader is gone.
///
/// `EPIPE` is how a process exit reaches a writer that had no other way to
/// see it: the child's stdout can still be held open by a grandchild it
/// spawned, so nothing upstream has noticed, and the failed write is the
/// first evidence.
fn broken_pipe(error: &Error) -> bool {
    matches!(error, Error::Io(io) if io.kind() == std::io::ErrorKind::BrokenPipe)
}

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

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

async fn wait_for_health_for(
    client: &reqwest::Client,
    base_url: &str,
    total_timeout: Duration,
) -> Result<()> {
    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
}