supercode_opencode_frontend/

server.rs

//! Authenticated numeric-loopback HTTP host for the stock OpenCode adapter.

use std::fmt;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use base64::Engine as _;
use serde_json::Value;
use supercode::SdkRuntime;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;
use tokio::task::JoinSet;
use zeroize::Zeroize;

use crate::{OpenCodeAdapter, OpenCodeRequest, ResponseBody};

const CREDENTIAL_BYTES: usize = 32;
const MAX_HEADER_BYTES: usize = 64 * 1024;
const MAX_BODY_BYTES: usize = 1024 * 1024;
const MAX_EVENT_BYTES: usize = 16 * 1024 * 1024;
const MAX_CONCURRENT_CONNECTIONS: usize = 32;
const READ_TIMEOUT: Duration = Duration::from_secs(5);
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);

#[derive(Debug, thiserror::Error)]
pub enum CredentialError {
    #[error("operating system random source failed")]
    RandomSource,
}

/// Opaque stock-client password. It is non-cloneable, redacted, and zeroized;
/// the only production disclosure path installs it in one child environment.
pub struct OpenCodeClientCredential {
    password: [u8; CREDENTIAL_BYTES],
}

impl OpenCodeClientCredential {
    pub fn spawn_tokio_child(
        &self,
        command: &mut tokio::process::Command,
    ) -> std::io::Result<tokio::process::Child> {
        let mut password = encode_hex(&self.password);
        command.env("OPENCODE_SERVER_PASSWORD", &password);
        command.env("OPENCODE_SERVER_USERNAME", "opencode");
        let child = command.spawn();
        command.env_remove("OPENCODE_SERVER_PASSWORD");
        command.env_remove("OPENCODE_SERVER_USERNAME");
        password.zeroize();
        child
    }

    #[cfg(test)]
    fn authorization_header(&self) -> String {
        let mut password = encode_hex(&self.password);
        let encoded =
            base64::engine::general_purpose::STANDARD.encode(format!("opencode:{password}"));
        password.zeroize();
        format!("Basic {encoded}")
    }
}

impl fmt::Debug for OpenCodeClientCredential {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("OpenCodeClientCredential([REDACTED])")
    }
}

impl Drop for OpenCodeClientCredential {
    fn drop(&mut self) {
        self.password.zeroize();
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OpenCodeEndpointHealth {
    Ready,
    ShuttingDown,
    Stopped,
}

/// Removable HTTP endpoint around one already-authorized SDK runtime client.
pub struct OpenCodeEndpoint {
    adapter: Arc<OpenCodeAdapter>,
    credential_digest: [u8; 32],
}

impl OpenCodeEndpoint {
    pub fn new(
        runtime: Arc<dyn SdkRuntime>,
        runtime_id: impl Into<String>,
        workspace: impl Into<PathBuf>,
    ) -> Result<(Arc<Self>, OpenCodeClientCredential), CredentialError> {
        let mut password = [0_u8; CREDENTIAL_BYTES];
        getrandom::getrandom(&mut password).map_err(|_| CredentialError::RandomSource)?;
        let mut basic = format!("opencode:{}", encode_hex(&password));
        let credential_digest = *blake3::hash(basic.as_bytes()).as_bytes();
        basic.zeroize();
        Ok((
            Arc::new(Self {
                adapter: OpenCodeAdapter::new(runtime, runtime_id, workspace),
                credential_digest,
            }),
            OpenCodeClientCredential { password },
        ))
    }

    pub fn session_id(&self) -> &str {
        self.adapter.session_id()
    }

    /// Bind only numeric IPv4 loopback. Remote use is an operator-owned
    /// authenticated tunnel; there is no wildcard/arbitrary-address API.
    pub async fn bind(self: &Arc<Self>, port: u16) -> std::io::Result<OpenCodeServerHandle> {
        let listener =
            TcpListener::bind(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port)).await?;
        let address = listener.local_addr()?;
        let (shutdown, shutdown_receiver) = watch::channel(false);
        let shutting_down = Arc::new(AtomicBool::new(false));
        let stopped = Arc::new(AtomicBool::new(false));
        let task = tokio::spawn(run_accept_loop(
            listener,
            self.clone(),
            address,
            shutdown_receiver,
            shutting_down.clone(),
            stopped.clone(),
        ));
        Ok(OpenCodeServerHandle {
            address,
            shutdown,
            shutting_down,
            stopped,
            task: Some(task),
        })
    }

    fn authenticate(&self, authorization: &str) -> bool {
        let Some(encoded) = authorization.strip_prefix("Basic ") else {
            return false;
        };
        let Ok(mut decoded) = base64::engine::general_purpose::STANDARD.decode(encoded) else {
            return false;
        };
        let digest = *blake3::hash(&decoded).as_bytes();
        decoded.zeroize();
        constant_time_eq(&digest, &self.credential_digest)
    }
}

pub struct OpenCodeServerHandle {
    address: SocketAddr,
    shutdown: watch::Sender<bool>,
    shutting_down: Arc<AtomicBool>,
    stopped: Arc<AtomicBool>,
    task: Option<tokio::task::JoinHandle<()>>,
}

impl OpenCodeServerHandle {
    pub fn address(&self) -> SocketAddr {
        self.address
    }

    pub fn url(&self) -> String {
        format!("http://{}", self.address)
    }

    pub fn health(&self) -> OpenCodeEndpointHealth {
        if self.stopped.load(Ordering::SeqCst) {
            OpenCodeEndpointHealth::Stopped
        } else if self.shutting_down.load(Ordering::SeqCst) {
            OpenCodeEndpointHealth::ShuttingDown
        } else {
            OpenCodeEndpointHealth::Ready
        }
    }

    pub async fn shutdown(mut self) {
        self.shutting_down.store(true, Ordering::SeqCst);
        self.shutdown.send_replace(true);
        if let Some(task) = self.task.take() {
            let _ = task.await;
        }
    }
}

impl Drop for OpenCodeServerHandle {
    fn drop(&mut self) {
        self.shutting_down.store(true, Ordering::SeqCst);
        self.shutdown.send_replace(true);
        if let Some(task) = self.task.take() {
            task.abort();
        }
    }
}

async fn run_accept_loop(
    listener: TcpListener,
    endpoint: Arc<OpenCodeEndpoint>,
    address: SocketAddr,
    mut shutdown: watch::Receiver<bool>,
    shutting_down: Arc<AtomicBool>,
    stopped: Arc<AtomicBool>,
) {
    let budget = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_CONNECTIONS));
    let mut connections = JoinSet::new();
    loop {
        tokio::select! {
            changed = shutdown.changed() => {
                if changed.is_err() || *shutdown.borrow() {
                    break;
                }
            }
            _ = connections.join_next(), if !connections.is_empty() => {}
            accepted = listener.accept() => match accepted {
                Ok((stream, _)) => {
                    let Ok(permit) = budget.clone().try_acquire_owned() else {
                        drop(stream);
                        continue;
                    };
                    let endpoint = endpoint.clone();
                    let mut connection_shutdown = shutdown.clone();
                    connections.spawn(async move {
                        let _permit = permit;
                        tokio::select! {
                            _ = wait_for_shutdown(&mut connection_shutdown) => {}
                            _ = serve_connection(stream, endpoint, address) => {}
                        }
                    });
                }
                Err(_) => break,
            }
        }
    }
    connections.shutdown().await;
    shutting_down.store(true, Ordering::SeqCst);
    stopped.store(true, Ordering::SeqCst);
}

async fn wait_for_shutdown(shutdown: &mut watch::Receiver<bool>) {
    while !*shutdown.borrow() {
        if shutdown.changed().await.is_err() {
            break;
        }
    }
}

async fn serve_connection(
    mut stream: TcpStream,
    endpoint: Arc<OpenCodeEndpoint>,
    expected_address: SocketAddr,
) -> std::io::Result<()> {
    let request = match read_request(&mut stream, expected_address).await {
        Ok(request) => request,
        Err(error) => return write_error(&mut stream, error.status, error.message).await,
    };
    if !endpoint.authenticate(&request.authorization) {
        return write_error(&mut stream, 401, "authentication failed").await;
    }
    let body = if request.body.is_empty() {
        Value::Null
    } else {
        match serde_json::from_slice(&request.body) {
            Ok(body) => body,
            Err(_) => return write_error(&mut stream, 400, "invalid JSON body").await,
        }
    };
    // Runtime mutations outlive a frontend connection. If the endpoint is
    // shut down while a model turn is running, dropping this connection task
    // must not cancel or partially unwind the SDK-owned turn.
    let adapter = endpoint.adapter.clone();
    let operation = tokio::spawn(async move {
        adapter
            .handle(OpenCodeRequest {
                method: request.method,
                target: request.target,
                body,
            })
            .await
    });
    let response = operation.await.map_err(std::io::Error::other)?;
    match response.body {
        ResponseBody::Json(body) => write_json(&mut stream, response.status, &body).await,
        ResponseBody::EventStream(attachment) => {
            write_event_stream(&mut stream, endpoint.adapter.clone(), *attachment).await
        }
    }
}

struct WireRequest {
    method: String,
    target: String,
    authorization: String,
    body: Vec<u8>,
}

struct WireError {
    status: u16,
    message: &'static str,
}

async fn read_request(
    stream: &mut TcpStream,
    expected_address: SocketAddr,
) -> Result<WireRequest, WireError> {
    read_request_before(
        stream,
        expected_address,
        tokio::time::Instant::now() + READ_TIMEOUT,
    )
    .await
}

async fn read_request_before(
    stream: &mut TcpStream,
    expected_address: SocketAddr,
    deadline: tokio::time::Instant,
) -> Result<WireRequest, WireError> {
    let mut bytes = Vec::with_capacity(4096);
    let header_end = loop {
        if bytes.len() >= MAX_HEADER_BYTES {
            return Err(WireError {
                status: 431,
                message: "request headers too large",
            });
        }
        let mut chunk = [0_u8; 4096];
        let read = tokio::time::timeout_at(deadline, stream.read(&mut chunk))
            .await
            .map_err(|_| WireError {
                status: 408,
                message: "request read timeout",
            })?
            .map_err(|_| WireError {
                status: 400,
                message: "request read failed",
            })?;
        if read == 0 {
            return Err(WireError {
                status: 400,
                message: "incomplete request",
            });
        }
        bytes.extend_from_slice(&chunk[..read]);
        if let Some(index) = find_subslice(&bytes, b"\r\n\r\n") {
            if index + 4 > MAX_HEADER_BYTES {
                return Err(WireError {
                    status: 431,
                    message: "request headers too large",
                });
            }
            break index + 4;
        }
    };
    let header = std::str::from_utf8(&bytes[..header_end]).map_err(|_| WireError {
        status: 400,
        message: "request headers are not UTF-8",
    })?;
    let mut lines = header[..header.len() - 4].split("\r\n");
    let request_line = lines.next().ok_or(WireError {
        status: 400,
        message: "missing request line",
    })?;
    let fields = request_line.split(' ').collect::<Vec<_>>();
    if fields.len() != 3 || fields[2] != "HTTP/1.1" {
        return Err(WireError {
            status: 400,
            message: "invalid HTTP/1.1 request line",
        });
    }
    if !fields[1].starts_with('/') || fields[1].starts_with("//") {
        return Err(WireError {
            status: 400,
            message: "absolute-form target rejected",
        });
    }
    let method = fields[0].to_string();
    let target = fields[1].to_string();
    let mut host = None;
    let mut authorization = None;
    let mut content_length = None;
    for line in lines {
        let (name, value) = line.split_once(':').ok_or(WireError {
            status: 400,
            message: "invalid request header",
        })?;
        let name = name.to_ascii_lowercase();
        let value = value.trim();
        match name.as_str() {
            "host" if host.replace(value).is_some() => return Err(duplicate_security_header()),
            "authorization" if authorization.replace(value).is_some() => {
                return Err(duplicate_security_header())
            }
            "content-length" if content_length.replace(value).is_some() => {
                return Err(duplicate_security_header())
            }
            "transfer-encoding" | "origin" => {
                return Err(WireError {
                    status: 400,
                    message: "forbidden request header",
                })
            }
            _ => {}
        }
    }
    if host != Some(expected_address.to_string().as_str()) {
        return Err(WireError {
            status: 400,
            message: "invalid loopback Host",
        });
    }
    let authorization = authorization
        .ok_or(WireError {
            status: 401,
            message: "authentication failed",
        })?
        .to_string();
    let content_length = content_length
        .unwrap_or("0")
        .parse::<usize>()
        .map_err(|_| WireError {
            status: 400,
            message: "invalid content length",
        })?;
    if content_length > MAX_BODY_BYTES {
        return Err(WireError {
            status: 413,
            message: "request body too large",
        });
    }
    while bytes.len() - header_end < content_length {
        let remaining = content_length - (bytes.len() - header_end);
        let mut chunk = vec![0_u8; remaining.min(4096)];
        let read = tokio::time::timeout_at(deadline, stream.read(&mut chunk))
            .await
            .map_err(|_| WireError {
                status: 408,
                message: "request body timeout",
            })?
            .map_err(|_| WireError {
                status: 400,
                message: "request body read failed",
            })?;
        if read == 0 {
            return Err(WireError {
                status: 400,
                message: "incomplete request body",
            });
        }
        bytes.extend_from_slice(&chunk[..read]);
    }
    if bytes.len() - header_end != content_length {
        return Err(WireError {
            status: 400,
            message: "pipelined request bytes rejected",
        });
    }
    Ok(WireRequest {
        method,
        target,
        authorization,
        body: bytes[header_end..].to_vec(),
    })
}

async fn write_json(stream: &mut TcpStream, status: u16, body: &Value) -> std::io::Result<()> {
    let body = serde_json::to_vec(body).expect("JSON Value serializes");
    let reason = status_reason(status);
    let header = format!(
        "HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\nX-Content-Type-Options: nosniff\r\n\r\n",
        body.len()
    );
    stream.write_all(header.as_bytes()).await?;
    stream.write_all(&body).await?;
    stream.shutdown().await
}

async fn write_error(
    stream: &mut TcpStream,
    status: u16,
    message: &'static str,
) -> std::io::Result<()> {
    write_json(
        stream,
        status,
        &serde_json::json!({"name": "transport_error", "message": message}),
    )
    .await
}

async fn write_event_stream(
    stream: &mut TcpStream,
    adapter: Arc<OpenCodeAdapter>,
    mut attachment: supercode::FrontendAttachment,
) -> std::io::Result<()> {
    stream
        .write_all(
            b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: close\r\nX-Accel-Buffering: no\r\n\r\n",
        )
        .await?;
    for event in adapter.initial_events(&attachment) {
        write_sse(stream, &event).await?;
    }
    let mut projection = adapter.event_projection(&attachment);
    loop {
        tokio::select! {
            event = attachment.next_event() => match event {
                Ok(event) => {
                    for projected in projection.project(&event) {
                        write_sse(stream, &projected).await?;
                    }
                }
                Err(_) => break,
            },
            _ = tokio::time::sleep(HEARTBEAT_INTERVAL) => {
                write_sse(stream, &serde_json::json!({"type": "server.heartbeat", "properties": {}})).await?;
            }
        }
    }
    stream.shutdown().await
}

async fn write_sse(stream: &mut TcpStream, event: &Value) -> std::io::Result<()> {
    let encoded = serde_json::to_vec(event).expect("JSON Value serializes");
    if encoded.len() > MAX_EVENT_BYTES {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "projected event exceeded byte limit",
        ));
    }
    stream.write_all(b"data: ").await?;
    stream.write_all(&encoded).await?;
    stream.write_all(b"\n\n").await?;
    stream.flush().await
}

fn duplicate_security_header() -> WireError {
    WireError {
        status: 400,
        message: "duplicate security header",
    }
}

fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack
        .windows(needle.len())
        .position(|window| window == needle)
}

fn constant_time_eq(left: &[u8; 32], right: &[u8; 32]) -> bool {
    left.iter()
        .zip(right)
        .fold(0_u8, |difference, (left, right)| {
            difference | (left ^ right)
        })
        == 0
}

fn encode_hex(secret: &[u8; CREDENTIAL_BYTES]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut encoded = String::with_capacity(CREDENTIAL_BYTES * 2);
    for byte in secret {
        encoded.push(HEX[(byte >> 4) as usize] as char);
        encoded.push(HEX[(byte & 0x0f) as usize] as char);
    }
    encoded
}

fn status_reason(status: u16) -> &'static str {
    match status {
        200 => "OK",
        400 => "Bad Request",
        401 => "Unauthorized",
        403 => "Forbidden",
        404 => "Not Found",
        408 => "Request Timeout",
        409 => "Conflict",
        413 => "Payload Too Large",
        431 => "Request Header Fields Too Large",
        500 => "Internal Server Error",
        _ => "Error",
    }
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, VecDeque};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex;

    use async_trait::async_trait;
    use supercode::{
        ChatMessage, CoordinatedRuntime, FrontendActions, FrontendAttachSnapshot,
        FrontendAttachment, FrontendConnectionState, FrontendDisplayCapabilities, FrontendResponse,
        FrontendRuntimeDescriptor, FrontendTurnState, RuntimeAuthorization, RuntimeClientId,
        SdkError, SdkEvent,
    };
    use tokio::sync::broadcast;

    use super::*;

    struct FixtureRuntime {
        history: Mutex<Vec<ChatMessage>>,
        submissions: AtomicUsize,
        responses: Mutex<Vec<FrontendResponse>>,
        events: broadcast::Sender<SdkEvent>,
    }

    impl FixtureRuntime {
        fn new() -> Arc<Self> {
            let (events, _) = broadcast::channel(64);
            Arc::new(Self {
                history: Mutex::new(vec![
                    ChatMessage::system("Claude context"),
                    ChatMessage::user("before"),
                    ChatMessage::assistant("ready"),
                ]),
                submissions: AtomicUsize::new(0),
                responses: Mutex::new(Vec::new()),
                events,
            })
        }

        fn descriptor() -> FrontendRuntimeDescriptor {
            FrontendRuntimeDescriptor {
                schema_version: 2,
                session_id: "runtime-1".into(),
                source_harness: Some("claude-code".into()),
                emulation_profile: Some("claude-code".into()),
                active_modules: Vec::new(),
                commands: Vec::new(),
                operations: Vec::new(),
                actions: FrontendActions {
                    submit: true,
                    interrupt: true,
                    steer: true,
                    respond: true,
                    detach: true,
                    close: false,
                },
                display: FrontendDisplayCapabilities {
                    event_kinds: vec!["text_delta".into(), "request".into()],
                    opaque_fallback: true,
                },
                model: "openrouter/glm-5.2".into(),
                turn_state: FrontendTurnState::Idle,
                connection_state: FrontendConnectionState::Connected,
                extensions: BTreeMap::new(),
            }
        }
    }

    #[async_trait]
    impl SdkRuntime for FixtureRuntime {
        async fn describe(&self) -> Result<FrontendRuntimeDescriptor, SdkError> {
            Ok(Self::descriptor())
        }

        async fn attach(&self, history_limit: usize) -> Result<FrontendAttachment, SdkError> {
            let history = self
                .history
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .clone();
            let start = history.len().saturating_sub(history_limit);
            Ok(FrontendAttachment::from_snapshot(
                FrontendAttachSnapshot {
                    descriptor: Self::descriptor(),
                    history: history[start..].to_vec(),
                    history_cursor: history.len() as u64,
                    replay: VecDeque::new(),
                },
                self.events.subscribe(),
            ))
        }

        async fn send_input(self: Arc<Self>, prompt: String) -> Result<(), SdkError> {
            self.submit(prompt).await.map(|_| ())
        }

        async fn submit(&self, prompt: String) -> Result<String, SdkError> {
            self.submissions.fetch_add(1, Ordering::SeqCst);
            let reply = format!("reply:{prompt}");
            self.history
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .extend([
                    ChatMessage::user(prompt),
                    ChatMessage::assistant(reply.clone()),
                ]);
            Ok(reply)
        }

        async fn interrupt(&self) -> Result<bool, SdkError> {
            Ok(true)
        }

        async fn steer(&self, _prompt: String) -> Result<(), SdkError> {
            Ok(())
        }

        async fn respond(&self, response: FrontendResponse) -> Result<(), SdkError> {
            self.responses
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .push(response);
            Ok(())
        }
    }

    async fn endpoint_for(
        runtime: Arc<dyn SdkRuntime>,
    ) -> (OpenCodeServerHandle, OpenCodeClientCredential) {
        let (endpoint, credential) =
            OpenCodeEndpoint::new(runtime, "runtime-1", "/runtime").expect("create endpoint");
        let handle = endpoint.bind(0).await.expect("bind endpoint");
        (handle, credential)
    }

    async fn raw_request(
        address: SocketAddr,
        authorization: Option<&str>,
        method: &str,
        target: &str,
        body: &str,
    ) -> String {
        let mut stream = TcpStream::connect(address).await.unwrap();
        let auth = authorization
            .map(|authorization| format!("Authorization: {authorization}\r\n"))
            .unwrap_or_default();
        let request = format!(
            "{method} {target} HTTP/1.1\r\nHost: {address}\r\n{auth}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
            body.len()
        );
        stream.write_all(request.as_bytes()).await.unwrap();
        let mut response = Vec::new();
        stream.read_to_end(&mut response).await.unwrap();
        String::from_utf8(response).unwrap()
    }

    async fn raw_wire(address: SocketAddr, request: String) -> String {
        let mut stream = TcpStream::connect(address).await.unwrap();
        stream.write_all(request.as_bytes()).await.unwrap();
        let mut response = Vec::new();
        stream.read_to_end(&mut response).await.unwrap();
        String::from_utf8(response).unwrap()
    }

    async fn read_until(stream: &mut TcpStream, needle: &str) -> String {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        let mut bytes = Vec::new();
        loop {
            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
            assert!(
                !remaining.is_zero(),
                "did not receive {needle}: {}",
                String::from_utf8_lossy(&bytes)
            );
            let mut chunk = [0_u8; 4096];
            let read = tokio::time::timeout(remaining, stream.read(&mut chunk))
                .await
                .expect("stream read deadline")
                .expect("stream read");
            assert!(read > 0, "stream closed before {needle}");
            bytes.extend_from_slice(&chunk[..read]);
            let text = String::from_utf8_lossy(&bytes);
            if text.contains(needle) {
                return text.into_owned();
            }
        }
    }

    fn message_body(prompt: &str) -> String {
        serde_json::json!({
            "messageID": "msg_stock", "agent": "build",
            "model": {"providerID": "openrouter", "modelID": "glm-5.2"},
            "parts": [{"id": "prt_stock", "type": "text", "text": prompt}]
        })
        .to_string()
    }

    #[tokio::test]
    async fn loopback_host_requires_exact_basic_credential_and_host() {
        let runtime = FixtureRuntime::new();
        let (handle, credential) = endpoint_for(runtime).await;
        let missing = raw_request(handle.address(), None, "GET", "/agent", "").await;
        assert!(missing.starts_with("HTTP/1.1 401"));
        let wrong = raw_request(
            handle.address(),
            Some("Basic b3BlbmNvZGU6d3Jvbmc="),
            "GET",
            "/agent",
            "",
        )
        .await;
        assert!(wrong.starts_with("HTTP/1.1 401"));
        let authorized = raw_request(
            handle.address(),
            Some(&credential.authorization_header()),
            "GET",
            "/agent",
            "",
        )
        .await;
        assert!(authorized.starts_with("HTTP/1.1 200"));
        assert!(!format!("{credential:?}").contains(&credential.authorization_header()));
        handle.shutdown().await;
    }

    #[tokio::test]
    async fn authenticated_owner_submits_but_observer_cannot_duplicate_work() {
        let runtime = FixtureRuntime::new();
        let coordinator = CoordinatedRuntime::new(runtime.clone());
        let owner = coordinator.client(
            RuntimeClientId::parse("opencode-owner").unwrap(),
            RuntimeAuthorization::interactive(),
        );
        let (owner_handle, owner_credential) = endpoint_for(owner).await;
        let session = OpenCodeAdapter::new(runtime.clone(), "runtime-1", "/runtime")
            .session_id()
            .to_string();
        let response = raw_request(
            owner_handle.address(),
            Some(&owner_credential.authorization_header()),
            "POST",
            &format!("/session/{session}/message"),
            &message_body("continue once"),
        )
        .await;
        assert!(response.starts_with("HTTP/1.1 200"), "{response}");
        assert_eq!(runtime.submissions.load(Ordering::SeqCst), 1);
        owner_handle.shutdown().await;

        let observer = coordinator.client(
            RuntimeClientId::parse("opencode-observer").unwrap(),
            RuntimeAuthorization::observer(),
        );
        let (observer_handle, observer_credential) = endpoint_for(observer).await;
        let denied = raw_request(
            observer_handle.address(),
            Some(&observer_credential.authorization_header()),
            "POST",
            &format!("/session/{session}/message"),
            &message_body("must not run"),
        )
        .await;
        assert!(denied.starts_with("HTTP/1.1 409"), "{denied}");
        assert_eq!(runtime.submissions.load(Ordering::SeqCst), 1);
        observer_handle.shutdown().await;
    }

    #[tokio::test]
    async fn transport_rejects_duplicate_security_headers_wrong_host_and_oversized_body() {
        let runtime = FixtureRuntime::new();
        let (handle, credential) = endpoint_for(runtime).await;
        let auth = credential.authorization_header();
        let duplicate = raw_wire(
            handle.address(),
            format!(
                "GET /agent HTTP/1.1\r\nHost: {}\r\nAuthorization: {auth}\r\nAuthorization: {auth}\r\n\r\n",
                handle.address()
            ),
        )
        .await;
        assert!(duplicate.starts_with("HTTP/1.1 400"));
        let wrong_host = raw_wire(
            handle.address(),
            format!(
                "GET /agent HTTP/1.1\r\nHost: localhost:{}\r\nAuthorization: {auth}\r\n\r\n",
                handle.address().port()
            ),
        )
        .await;
        assert!(wrong_host.starts_with("HTTP/1.1 400"));
        let oversized = raw_wire(
            handle.address(),
            format!(
                "POST /session HTTP/1.1\r\nHost: {}\r\nAuthorization: {auth}\r\nContent-Length: {}\r\n\r\n",
                handle.address(),
                MAX_BODY_BYTES + 1
            ),
        )
        .await;
        assert!(oversized.starts_with("HTTP/1.1 413"));
        handle.shutdown().await;
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn credential_enters_only_one_child_environment_and_leaves_reusable_command_clean() {
        use std::process::Stdio;

        let runtime = FixtureRuntime::new();
        let (_endpoint, credential) =
            OpenCodeEndpoint::new(runtime, "runtime-1", "/runtime").unwrap();
        let mut command = tokio::process::Command::new("sh");
        command
            .args([
                "-c",
                "printf '%s:%s' \"$OPENCODE_SERVER_USERNAME\" \"$OPENCODE_SERVER_PASSWORD\"",
            ])
            .stdout(Stdio::piped());
        let output = credential
            .spawn_tokio_child(&mut command)
            .unwrap()
            .wait_with_output()
            .await
            .unwrap();
        let first = String::from_utf8(output.stdout).unwrap();
        assert!(first.starts_with("opencode:"));
        assert_eq!(first.len(), "opencode:".len() + CREDENTIAL_BYTES * 2);

        let reused = command.output().await.unwrap();
        assert_eq!(reused.stdout, b":");
    }

    #[tokio::test]
    async fn event_stream_projects_live_sdk_delta_and_permission_without_second_runtime() {
        let runtime = FixtureRuntime::new();
        let (handle, credential) = endpoint_for(runtime.clone()).await;
        let mut stream = TcpStream::connect(handle.address()).await.unwrap();
        let request = format!(
            "GET /event HTTP/1.1\r\nHost: {}\r\nAuthorization: {}\r\nContent-Length: 0\r\n\r\n",
            handle.address(),
            credential.authorization_header()
        );
        stream.write_all(request.as_bytes()).await.unwrap();
        let initial = read_until(&mut stream, "session.status").await;
        assert!(initial.contains("text/event-stream"));

        runtime
            .events
            .send(SdkEvent {
                sequence: 10,
                kind: "text_delta".into(),
                payload: serde_json::json!({"type": "text_delta", "text": "live GLM output"}),
            })
            .unwrap();
        let projected = read_until(&mut stream, "live GLM output").await;
        assert!(projected.contains("message.part.delta"), "{projected}");
        assert!(projected.contains("live GLM output"), "{projected}");
        drop(stream);
        handle.shutdown().await;
    }

    #[tokio::test]
    async fn shutdown_is_durable_and_closes_an_active_event_stream() {
        for _ in 0..16 {
            let (handle, _credential) = endpoint_for(FixtureRuntime::new()).await;
            tokio::time::timeout(Duration::from_secs(1), handle.shutdown())
                .await
                .expect("shutdown sent immediately after bind must not be lost");
        }

        let (handle, credential) = endpoint_for(FixtureRuntime::new()).await;
        let mut stream = TcpStream::connect(handle.address()).await.unwrap();
        let request = format!(
            "GET /event HTTP/1.1\r\nHost: {}\r\nAuthorization: {}\r\nContent-Length: 0\r\n\r\n",
            handle.address(),
            credential.authorization_header()
        );
        stream.write_all(request.as_bytes()).await.unwrap();
        let initial = read_until(&mut stream, "session.updated").await;
        assert!(initial.contains("text/event-stream"));
        tokio::time::timeout(Duration::from_secs(1), handle.shutdown())
            .await
            .expect("active SSE connection must be joined during shutdown");
        let mut tail = Vec::new();
        tokio::time::timeout(Duration::from_secs(1), stream.read_to_end(&mut tail))
            .await
            .expect("SSE peer must observe endpoint shutdown")
            .expect("read closed SSE stream");
    }

    #[tokio::test]
    async fn request_deadline_is_total_not_reset_by_trickled_bytes() {
        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
        let address = listener.local_addr().unwrap();
        let mut client = TcpStream::connect(address).await.unwrap();
        let (mut server, _) = listener.accept().await.unwrap();
        let writer = tokio::spawn(async move {
            for byte in b"GET /agent HTTP/1.1\r\n" {
                if client.write_all(&[*byte]).await.is_err() {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        });
        let started = tokio::time::Instant::now();
        let error =
            match read_request_before(&mut server, address, started + Duration::from_millis(35))
                .await
            {
                Ok(_) => panic!("trickled request must exceed one total deadline"),
                Err(error) => error,
            };
        assert_eq!(error.status, 408);
        assert!(started.elapsed() < Duration::from_millis(100));
        writer.abort();
    }
}